"""
Pattern 270: Verification Pattern with LangChain/LangGraph
Purpose: Correctness verification and property checking with formal methods.
Provides intelligent verification orchestration with LLM-powered proof analysis.
"""
from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json
class VerificationType(Enum):
"""Types of verification."""
PROPERTY = "property"
INVARIANT = "invariant"
PRECONDITION = "precondition"
POSTCONDITION = "postcondition"
ASSERTION = "assertion"
CONTRACT = "contract"
class VerificationSeverity(Enum):
"""Severity of verification issues."""
CRITICAL = "critical"
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
@dataclass
class VerificationCondition:
"""Verification condition definition."""
name: str
verification_type: VerificationType
condition: Callable
description: str
severity: VerificationSeverity = VerificationSeverity.CRITICAL
@dataclass
class VerificationResult:
"""Result of verification."""
condition_name: str
verification_type: VerificationType
passed: bool
message: str
severity: VerificationSeverity
evidence: Dict[str, Any] = field(default_factory=dict)
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class VerificationProof:
"""LLM-generated verification proof."""
formal_statement: str
proof_sketch: str
assumptions: List[str]
conclusions: List[str]
confidence: float
class VerificationState(TypedDict):
"""State for verification workflow."""
context: Dict[str, Any]
conditions: List[VerificationCondition]
results: List[VerificationResult]
all_passed: bool
failures: List[VerificationResult]
proof: Optional[VerificationProof]
recommendations: List[str]
verification_summary: Dict[str, Any]
class IntelligentVerifier:
"""
Intelligent Verifier with LangGraph workflow integration.
Features:
- Property verification
- Invariant checking
- Pre/postcondition verification
- Contract checking
- LLM-powered proof generation
- Evidence collection
- Formal reasoning support
- Counterexample detection
"""
def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.5):
self.name = name
self.llm = ChatOpenAI(model=model_name, temperature=temperature)
# Conditions
self.conditions: List[VerificationCondition] = []
# History
self.verification_history: List[VerificationResult] = []
# Build workflow
self.workflow = self._build_workflow()
# Statistics
self.stats = {
"total_verifications": 0,
"passed": 0,
"failed": 0,
"verification_types": {t.value: 0 for t in VerificationType}
}
def _build_workflow(self) -> StateGraph:
"""Build LangGraph workflow."""
workflow = StateGraph(VerificationState)
workflow.add_node("analyze_conditions", self._analyze_conditions_node)
workflow.add_node("verify_properties", self._verify_properties_node)
workflow.add_node("check_invariants", self._check_invariants_node)
workflow.add_node("collect_evidence", self._collect_evidence_node)
workflow.add_node("generate_proof", self._generate_proof_node)
workflow.set_entry_point("analyze_conditions")
workflow.add_edge("analyze_conditions", "verify_properties")
workflow.add_edge("verify_properties", "check_invariants")
workflow.add_edge("check_invariants", "collect_evidence")
workflow.add_edge("collect_evidence", "generate_proof")
workflow.add_edge("generate_proof", END)
return workflow.compile()
def _analyze_conditions_node(self, state: VerificationState) -> VerificationState:
"""Analyze verification conditions."""
state["conditions"] = self.conditions
state["results"] = []
state["failures"] = []
return state
def _verify_properties_node(self, state: VerificationState) -> VerificationState:
"""Verify properties and conditions."""
context = state["context"]
conditions = state["conditions"]
results = []
for condition in conditions:
self.stats["total_verifications"] += 1
self.stats["verification_types"][condition.verification_type.value] += 1
try:
passed = condition.condition(context)
message = (
f"✓ {condition.description}" if passed
else f"✗ {condition.description} FAILED"
)
result = VerificationResult(
condition_name=condition.name,
verification_type=condition.verification_type,
passed=passed,
message=message,
severity=condition.severity,
evidence=context.copy()
)
if passed:
self.stats["passed"] += 1
else:
self.stats["failed"] += 1
state["failures"].append(result)
results.append(result)
self.verification_history.append(result)
except Exception as e:
result = VerificationResult(
condition_name=condition.name,
verification_type=condition.verification_type,
passed=False,
message=f"Verification error: {str(e)}",
severity=VerificationSeverity.CRITICAL,
evidence={"error": str(e)}
)
results.append(result)
state["failures"].append(result)
self.verification_history.append(result)
self.stats["failed"] += 1
state["results"] = results
state["all_passed"] = len(state["failures"]) == 0
return state
def _check_invariants_node(self, state: VerificationState) -> VerificationState:
"""Check system invariants."""
# Additional invariant checks already handled in verify_properties
return state
def _collect_evidence_node(self, state: VerificationState) -> VerificationState:
"""Collect verification evidence."""
# Create verification summary
results = state["results"]
failures = state["failures"]
state["verification_summary"] = {
"total_conditions": len(results),
"passed": sum(1 for r in results if r.passed),
"failed": len(failures),
"critical_failures": sum(1 for f in failures if f.severity == VerificationSeverity.CRITICAL),
"verification_types": {}
}
# Count by type
for vtype in VerificationType:
count = sum(1 for r in results if r.verification_type == vtype)
if count > 0:
state["verification_summary"]["verification_types"][vtype.value] = count
return state
def _generate_proof_node(self, state: VerificationState) -> VerificationState:
"""Generate verification proof using LLM."""
results = state["results"]
failures = state["failures"]
summary = state["verification_summary"]
context = state["context"]
if state["all_passed"]:
# Generate proof for successful verification
prompt = ChatPromptTemplate.from_messages([
("system", "You are an expert in formal verification and mathematical proof."),
("user", """Generate a verification proof for this successful verification:
Context: {context}
Conditions Verified: {total_conditions}
Verification Types: {verification_types}
Provide proof as JSON with:
- formal_statement: Formal statement of what was verified
- proof_sketch: Brief proof sketch
- assumptions: List of assumptions (2-3 items)
- conclusions: List of conclusions (2-3 items)
- confidence: Confidence score 0.0-1.0
Focus on formal correctness.""")
])
chain = prompt | self.llm
response = chain.invoke({
"context": str(context),
"total_conditions": summary["total_conditions"],
"verification_types": str(summary["verification_types"])
})
try:
proof_data = json.loads(response.content)
state["proof"] = VerificationProof(
formal_statement=proof_data.get("formal_statement", "System properties verified"),
proof_sketch=proof_data.get("proof_sketch", "All conditions satisfied"),
assumptions=proof_data.get("assumptions", []),
conclusions=proof_data.get("conclusions", ["System is correct"]),
confidence=proof_data.get("confidence", 0.95)
)
state["recommendations"] = ["System verification passed", "Properties hold"]
except:
state["proof"] = VerificationProof(
formal_statement="System properties verified",
proof_sketch="All verification conditions passed",
assumptions=["Input conditions valid"],
conclusions=["System behaves correctly"],
confidence=0.90
)
state["recommendations"] = ["Verification successful"]
else:
# Generate analysis for failures
prompt = ChatPromptTemplate.from_messages([
("system", "You are an expert in formal verification and debugging."),
("user", """Analyze these verification failures:
Context: {context}
Total Failures: {failure_count}
Critical Failures: {critical_count}
Failed Conditions:
{failures}
Provide analysis as JSON with:
- formal_statement: What failed to verify
- proof_sketch: Why verification failed
- assumptions: Violated assumptions (2-3 items)
- conclusions: Failure implications (2-3 items)
- confidence: Confidence in analysis 0.0-1.0
Also suggest fixes.""")
])
chain = prompt | self.llm
failure_details = "\n".join([f"- {f.message}" for f in failures])
response = chain.invoke({
"context": str(context),
"failure_count": summary["failed"],
"critical_count": summary["critical_failures"],
"failures": failure_details
})
try:
proof_data = json.loads(response.content)
state["proof"] = VerificationProof(
formal_statement=proof_data.get("formal_statement", "Verification failed"),
proof_sketch=proof_data.get("proof_sketch", "Conditions not satisfied"),
assumptions=proof_data.get("assumptions", []),
conclusions=proof_data.get("conclusions", ["System has errors"]),
confidence=proof_data.get("confidence", 0.85)
)
state["recommendations"] = [
"Fix critical verification failures",
"Review violated conditions",
"Update system implementation"
]
except:
state["proof"] = VerificationProof(
formal_statement="Verification failed",
proof_sketch="Some conditions not satisfied",
assumptions=["Conditions were violated"],
conclusions=["System needs correction"],
confidence=0.80
)
state["recommendations"] = ["Fix verification failures"]
return state
def add_condition(
self,
name: str,
verification_type: VerificationType,
condition: Callable,
description: str,
severity: VerificationSeverity = VerificationSeverity.CRITICAL
):
"""Add verification condition."""
vc = VerificationCondition(
name=name,
verification_type=verification_type,
condition=condition,
description=description,
severity=severity
)
self.conditions.append(vc)
def verify(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""Verify system through workflow."""
initial_state: VerificationState = {
"context": context,
"conditions": [],
"results": [],
"all_passed": True,
"failures": [],
"proof": None,
"recommendations": [],
"verification_summary": {}
}
final_state = self.workflow.invoke(initial_state)
return {
"passed": final_state["all_passed"],
"results": final_state["results"],
"failures": final_state["failures"],
"summary": final_state["verification_summary"],
"proof": final_state["proof"],
"recommendations": final_state["recommendations"]
}
def verify_property(
self,
property_name: str,
property_check: Callable,
context: Dict[str, Any]
) -> bool:
"""Verify specific property."""
self.add_condition(
property_name,
VerificationType.PROPERTY,
property_check,
f"Property: {property_name}",
VerificationSeverity.CRITICAL
)
result = self.verify(context)
return result["passed"]
def check_invariant(
self,
invariant_name: str,
invariant_check: Callable,
context: Dict[str, Any]
) -> bool:
"""Check invariant."""
self.add_condition(
invariant_name,
VerificationType.INVARIANT,
invariant_check,
f"Invariant: {invariant_name}",
VerificationSeverity.CRITICAL
)
result = self.verify(context)
return result["passed"]
def get_failures(self) -> List[VerificationResult]:
"""Get all failed verifications."""
return [r for r in self.verification_history if not r.passed]
def get_statistics(self) -> Dict[str, Any]:
"""Get verification statistics."""
return {
"name": self.name,
"conditions_count": len(self.conditions),
**self.stats
}
class VerificationPattern:
"""
Comprehensive Verification Pattern coordinator with LangChain/LangGraph.
"""
def __init__(self, model_name: str = "gpt-4"):
self.verifiers: Dict[str, IntelligentVerifier] = {}
self.model_name = model_name
def create_verifier(self, name: str) -> IntelligentVerifier:
"""Create a verifier."""
verifier = IntelligentVerifier(name, model_name=self.model_name)
self.verifiers[name] = verifier
return verifier
def verify(
self,
verifier_name: str,
context: Dict[str, Any]
) -> Dict[str, Any]:
"""Run verification."""
if verifier_name not in self.verifiers:
return {"error": f"Verifier {verifier_name} not found"}
verifier = self.verifiers[verifier_name]
result = verifier.verify(context)
return {
"verifier": verifier_name,
**result
}
def get_statistics(self) -> Dict[str, Any]:
"""Get overall statistics."""
total_verifications = sum(v.stats["total_verifications"] for v in self.verifiers.values())
total_passed = sum(v.stats["passed"] for v in self.verifiers.values())
total_failed = sum(v.stats["failed"] for v in self.verifiers.values())
return {
"total_verifiers": len(self.verifiers),
"total_verifications": total_verifications,
"total_passed": total_passed,
"total_failed": total_failed
}
if __name__ == "__main__":
print("=== Verification Pattern with LangChain/LangGraph Demo ===\n")
print("Scenario 1: Banking System Verification")
print("-" * 50)
pattern = VerificationPattern(model_name="gpt-4")
banking_verifier = pattern.create_verifier("BankingVerifier")
# Add verification conditions
print("\nAdding verification conditions...")
# Property: Balance non-negative
banking_verifier.add_condition(
"balance_non_negative",
VerificationType.PROPERTY,
lambda ctx: ctx.get("balance", 0) >= 0,
"Account balance must be non-negative",
VerificationSeverity.CRITICAL
)
# Invariant: Total equals sum
banking_verifier.add_condition(
"total_equals_sum",
VerificationType.INVARIANT,
lambda ctx: abs(ctx.get("total", 0) - sum(ctx.get("transactions", []))) 0,
"Order total must be positive",
VerificationSeverity.CRITICAL
)
order_verifier.add_condition(
"items_not_empty",
VerificationType.INVARIANT,
lambda ctx: len(ctx.get("items", [])) > 0,
"Order must contain items",
VerificationSeverity.CRITICAL
)
order_verifier.add_condition(
"payment_authorized",
VerificationType.PRECONDITION,
lambda ctx: ctx.get("payment_status") == "authorized",
"Payment must be authorized",
VerificationSeverity.HIGH
)
order_verifier.add_condition(
"shipping_address_valid",
VerificationType.ASSERTION,
lambda ctx: ctx.get("shipping_address") is not None and len(ctx.get("shipping_address", "")) > 0,
"Shipping address must be provided",
VerificationSeverity.HIGH
)
print("\nVerification 3: Valid order")
result3 = pattern.verify("OrderVerifier", {
"order_total": 150.99,
"items": ["item1", "item2", "item3"],
"payment_status": "authorized",
"shipping_address": "123 Main St, City, State 12345"
})
print(f"Passed: {result3['passed']}")
print(f"Summary: {result3['summary']}")
print("\n" + "=" * 50)
print("\nVerification 4: Order with issues")
result4 = pattern.verify("OrderVerifier", {
"order_total": 0.0, # Invalid: zero total
"items": [], # Invalid: no items
"payment_status": "pending", # Invalid: not authorized
"shipping_address": "" # Invalid: empty address
})
print(f"Passed: {result4['passed']}")
print(f"Summary: {result4['summary']}")
print(f"\nFailures:")
for failure in result4['failures']:
print(f" - {failure.message}")
print("\n" + "=" * 50)
print("\nScenario 3: Data Structure Invariants")
print("-" * 50)
structure_verifier = pattern.create_verifier("StructureVerifier")
# Data structure invariants
structure_verifier.add_condition(
"array_sorted",
VerificationType.INVARIANT,
lambda ctx: all(ctx["array"][i] 1 else True,
"Array must be sorted",
VerificationSeverity.MEDIUM
)
structure_verifier.add_condition(
"no_duplicates",
VerificationType.PROPERTY,
lambda ctx: len(ctx.get("array", [])) == len(set(ctx.get("array", []))),
"Array must not contain duplicates",
VerificationSeverity.LOW
)
structure_verifier.add_condition(
"size_within_bounds",
VerificationType.ASSERTION,
lambda ctx: 0 <= len(ctx.get("array", [])) <= ctx.get("max_size", 1000),
"Array size must be within bounds",
VerificationSeverity.HIGH
)
print("\nVerification 5: Valid sorted array")
result5 = pattern.verify("StructureVerifier", {
"array": [1, 2, 3, 5, 8, 13],
"max_size": 100
})
print(f"Passed: {result5['passed']}")
print(f"Summary: {result5['summary']}")
print("\n" + "=" * 50)
print("\nVerification 6: Unsorted array with duplicates")
result6 = pattern.verify("StructureVerifier", {
"array": [5, 2, 8, 2, 1], # Unsorted and has duplicates
"max_size": 100
})
print(f"Passed: {result6['passed']}")
print(f"Summary: {result6['summary']}")
if result6['failures']:
print(f"\nFailures:")
for failure in result6['failures']:
print(f" - [{failure.severity.value}] {failure.message}")
print("\n" + "=" * 50)
print("\nStatistics")
print("-" * 50)
overall_stats = pattern.get_statistics()
print(f"\nOverall Statistics:")
print(f" Total Verifiers: {overall_stats['total_verifiers']}")
print(f" Total Verifications: {overall_stats['total_verifications']}")
print(f" Total Passed: {overall_stats['total_passed']}")
print(f" Total Failed: {overall_stats['total_failed']}")
print(f"\nBanking Verifier Statistics:")
banking_stats = banking_verifier.get_statistics()
for key, value in banking_stats.items():
print(f" {key}: {value}")
print(f"\nOrder Verifier Statistics:")
order_stats = order_verifier.get_statistics()
for key, value in order_stats.items():
print(f" {key}: {value}")
print("\n" + "=" * 50)
print("Verification pattern demonstration complete!")
print("=" * 50)
Agentic Architecture Pattern: Verification
Pattern 270: Verification Pattern – Line-by-Line Explanation
Overview
This pattern implements formal verification with property checking, invariant validation, and LLM-powered proof generation. It ensures system correctness through mathematical reasoning and formal methods.
Code Explanation
Lines 1-15: Imports
"""
Pattern 270: Verification Pattern with LangChain/LangGraph
Purpose: Formal verification with property checking and proof generation.
Provides intelligent verification orchestration with LLM-powered formal reasoning.
"""
Explanation: Module docstring – pattern for formal correctness verification.
from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json
Explanation: Standard imports for verification with LLM integration.
Lines 17-35: Enum Definitions
class VerificationType(Enum):
"""Types of verification."""
PROPERTY = "property" # General properties
INVARIANT = "invariant" # Always true conditions
PRECONDITION = "precondition" # Must be true before execution
POSTCONDITION = "postcondition" # Must be true after execution
ASSERTION = "assertion" # Point-in-time checks
CONTRACT = "contract" # Interface contracts
Explanation: Six verification types based on formal methods: – PROPERTY: General system properties (e.g., “balance never negative”) – INVARIANT: Conditions that must always hold (e.g., “list is always sorted”) – PRECONDITION: Requirements before function execution (e.g., “input not null”) – POSTCONDITION: Guarantees after function execution (e.g., “result is positive”) – ASSERTION: Runtime checks at specific points (e.g., “index in bounds”) – CONTRACT: Interface guarantees (e.g., “implements all required methods”)
class VerificationResult(Enum):
"""Result of verification."""
VERIFIED = "verified" # Proof successful
VIOLATED = "violated" # Proof failed
UNKNOWN = "unknown" # Cannot determine
Explanation: Three possible verification outcomes.
Lines 37-79: Dataclasses
@dataclass
class VerificationProperty:
"""Property to verify."""
name: str
verification_type: VerificationType
checker: Callable
description: str
expected_result: bool = True
Explanation: Defines a property to verify: – name: Identifier – verification_type: Category – checker: Function that checks property – description: Human-readable explanation – expected_result: What checker should return (usually True)
@dataclass
class VerificationEvidence:
"""Evidence for verification."""
property_name: str
result: VerificationResult
evidence: List[str]
counterexample: Optional[Dict[str, Any]] = None
timestamp: datetime = field(default_factory=datetime.now)
Explanation: Evidence supporting verification: – result: Verified, violated, or unknown – evidence: List of proof steps or test results – counterexample: Example input that violates property (if violated)
Example counterexample: {"input": -5, "output": None, "expected": "positive number"}
@dataclass
class ProofStep:
"""Step in verification proof."""
step_number: int
statement: str
justification: str
proven: bool
Explanation: Individual step in formal proof. Tracks logical progression.
@dataclass
class VerificationInsights:
"""LLM-generated verification insights."""
summary: str
proof_quality: str
gaps: List[str]
suggestions: List[str]
formal_proof: List[ProofStep]
Explanation: AI-generated formal reasoning: – summary: Overall verification status – proof_quality: Assessment of proof completeness – gaps: Missing proof steps or assumptions – suggestions: How to strengthen proof – formal_proof: Step-by-step logical proof
Lines 81-94: TypedDict State
class VerificationState(TypedDict):
"""State for verification workflow."""
properties: List[VerificationProperty]
evidence: Dict[str, VerificationEvidence]
system_state: Dict[str, Any]
verification_results: Dict[str, VerificationResult]
insights: Optional[VerificationInsights]
proof_complete: bool
Explanation: Workflow state tracking properties, evidence, and formal proofs.
Lines 96-142: IntelligentVerifier Initialization
class IntelligentVerifier:
"""
Intelligent Verifier with LangGraph workflow integration.
Features:
- Multiple verification types (property, invariant, contract, etc.)
- LLM-powered proof generation
- Formal reasoning
- Counterexample generation
- Evidence collection
- Symbolic execution
- Theorem proving assistance
"""
Explanation: Main verifier class with formal methods and AI assistance.
def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.3):
self.name = name
self.llm = ChatOpenAI(model=model_name, temperature=temperature) # Lower temp for formal reasoning
# Properties
self.properties: List[VerificationProperty] = []
# Evidence
self.evidence_history: List[VerificationEvidence] = []
# Build workflow
self.workflow = self._build_workflow()
# Statistics
self.stats = {
"total_verifications": 0,
"verified": 0,
"violated": 0,
"unknown": 0
}
Explanation: Initialize with lower temperature (0.3) for more deterministic formal reasoning.
Lines 144-162: Workflow Building
def _build_workflow(self) -> StateGraph:
"""Build LangGraph workflow."""
workflow = StateGraph(VerificationState)
workflow.add_node("check_properties", self._check_properties_node)
workflow.add_node("collect_evidence", self._collect_evidence_node)
workflow.add_node("generate_proof", self._generate_proof_node)
workflow.add_node("validate_proof", self._validate_proof_node)
workflow.add_node("synthesize_insights", self._synthesize_insights_node)
Explanation: Create 5-node workflow: 1. Check properties: Execute checkers 2. Collect evidence: Gather proof steps 3. Generate proof: LLM creates formal proof 4. Validate proof: Verify logical soundness 5. Synthesize insights: LLM analyzes completeness
workflow.set_entry_point("check_properties")
workflow.add_edge("check_properties", "collect_evidence")
workflow.add_edge("collect_evidence", "generate_proof")
workflow.add_edge("generate_proof", "validate_proof")
workflow.add_edge("validate_proof", "synthesize_insights")
workflow.add_edge("synthesize_insights", END)
return workflow.compile()
Explanation: Linear flow through verification stages.
Lines 164-227: Check Properties Node
def _check_properties_node(self, state: VerificationState) -> VerificationState:
"""Check all properties."""
system_state = state["system_state"]
verification_results = {}
evidence = {}
Explanation: Initialize results and evidence dictionaries.
for prop in self.properties:
self.stats["total_verifications"] += 1
try:
# Execute checker
result = prop.checker(system_state)
if result == prop.expected_result:
verification_results[prop.name] = VerificationResult.VERIFIED
self.stats["verified"] += 1
evidence_list = [f"Property '{prop.name}' holds", f"Checker returned {result}"]
else:
verification_results[prop.name] = VerificationResult.VIOLATED
self.stats["violated"] += 1
evidence_list = [
f"Property '{prop.name}' violated",
f"Expected {prop.expected_result}, got {result}"
]
Explanation: Execute each checker: – If returns expected result → VERIFIED – If returns unexpected result → VIOLATED – Record evidence for both cases
# Try to find counterexample if violated
counterexample = None
if verification_results[prop.name] == VerificationResult.VIOLATED:
counterexample = {
"system_state": system_state,
"checker_result": result,
"expected": prop.expected_result
}
Explanation: For violations, capture system state as counterexample. This helps diagnose why property failed.
evidence[prop.name] = VerificationEvidence(
property_name=prop.name,
result=verification_results[prop.name],
evidence=evidence_list,
counterexample=counterexample
)
except Exception as e:
verification_results[prop.name] = VerificationResult.UNKNOWN
self.stats["unknown"] += 1
evidence[prop.name] = VerificationEvidence(
property_name=prop.name,
result=VerificationResult.UNKNOWN,
evidence=[f"Verification error: {str(e)}"]
)
Explanation: Handle exceptions as UNKNOWN result. Cannot prove or disprove.
state["verification_results"] = verification_results
state["evidence"] = evidence
return state
Lines 229-251: Collect Evidence Node
def _collect_evidence_node(self, state: VerificationState) -> VerificationState:
"""Collect additional evidence for verification."""
evidence = state["evidence"]
# Add evidence to history
for prop_name, evidence_obj in evidence.items():
self.evidence_history.append(evidence_obj)
# Collect counterexamples if any
counterexamples = []
for prop_name, evidence_obj in evidence.items():
if evidence_obj.counterexample:
counterexamples.append({
"property": prop_name,
"counterexample": evidence_obj.counterexample
})
if counterexamples:
state["insights"] = VerificationInsights(
summary=f"Found {len(counterexamples)} violations",
proof_quality="Incomplete",
gaps=[f"Property {ce['property']} violated" for ce in counterexamples],
suggestions=["Address counterexamples to complete verification"],
formal_proof=[]
)
return state
Explanation: Collect evidence into history, identify counterexamples for further analysis.
Lines 253-338: Generate Proof Node (LLM-Powered)
def _generate_proof_node(self, state: VerificationState) -> VerificationState:
"""Generate formal proof using LLM."""
properties = state["properties"]
evidence = state["evidence"]
verification_results = state["verification_results"]
# Prepare properties summary
properties_summary = []
for prop in properties:
result = verification_results.get(prop.name, VerificationResult.UNKNOWN)
properties_summary.append(f"- {prop.name}: {prop.description} [{result.value}]")
Explanation: Prepare property list with current verification status.
# Prepare evidence summary
evidence_summary = []
for prop_name, evidence_obj in evidence.items():
evidence_summary.append(f"Property: {prop_name}")
evidence_summary.append(f" Result: {evidence_obj.result.value}")
evidence_summary.extend([f" - {e}" for e in evidence_obj.evidence])
if evidence_obj.counterexample:
evidence_summary.append(f" Counterexample: {evidence_obj.counterexample}")
Explanation: Prepare evidence list with counterexamples if any.
# Use LLM to generate formal proof
prompt = ChatPromptTemplate.from_messages([
("system", "You are an expert in formal verification and mathematical proof generation."),
("user", """Generate a formal verification proof:
System: {name}
Properties to Verify:
{properties}
Evidence Collected:
{evidence}
Provide analysis as JSON with:
- summary: Brief summary of verification status
- proof_quality: Assessment (Complete/Partial/Incomplete)
- gaps: List of gaps in current proof
- suggestions: How to complete verification
- formal_proof: Array of proof steps with:
- step_number: Sequential step number
- statement: Logical statement
- justification: Why this step is valid
- proven: true/false
Focus on logical rigor and formal correctness.""")
])
Explanation: Ask LLM to generate formal proof with step-by-step logical reasoning.
chain = prompt | self.llm
response = chain.invoke({
"name": self.name,
"properties": "\n".join(properties_summary),
"evidence": "\n".join(evidence_summary)
})
Explanation: Invoke LLM with properties and evidence.
try:
insights_data = json.loads(response.content)
# Parse proof steps
formal_proof = []
for step_data in insights_data.get("formal_proof", []):
formal_proof.append(ProofStep(
step_number=step_data.get("step_number", 0),
statement=step_data.get("statement", ""),
justification=step_data.get("justification", ""),
proven=step_data.get("proven", False)
))
state["insights"] = VerificationInsights(
summary=insights_data.get("summary", "Verification analysis completed"),
proof_quality=insights_data.get("proof_quality", "Unknown"),
gaps=insights_data.get("gaps", []),
suggestions=insights_data.get("suggestions", []),
formal_proof=formal_proof
)
except Exception as e:
# Fallback insights
state["insights"] = VerificationInsights(
summary="Verification completed",
proof_quality="Partial",
gaps=["LLM proof generation encountered an error"],
suggestions=["Manual verification may be needed"],
formal_proof=[]
)
Explanation: Parse JSON response into proof steps. Fallback if parsing fails.
return state
Lines 340-368: Validate Proof Node
def _validate_proof_node(self, state: VerificationState) -> VerificationState:
"""Validate logical soundness of proof."""
insights = state["insights"]
if not insights or not insights.formal_proof:
state["proof_complete"] = False
return state
# Check if all proof steps are proven
all_proven = all(step.proven for step in insights.formal_proof)
# Check if there are no critical gaps
critical_gaps = [g for g in insights.gaps if "violated" in g.lower() or "failed" in g.lower()]
proof_complete = all_proven and len(critical_gaps) == 0
state["proof_complete"] = proof_complete
return state
Explanation: Validate proof completeness: – All steps must be proven – No critical gaps (violations/failures) – Sets proof_complete flag
Lines 370-392: Synthesize Insights Node
def _synthesize_insights_node(self, state: VerificationState) -> VerificationState:
"""Synthesize final verification insights."""
insights = state["insights"]
if not insights:
return state
# Add summary based on results
verified_count = sum(1 for r in state["verification_results"].values() if r == VerificationResult.VERIFIED)
violated_count = sum(1 for r in state["verification_results"].values() if r == VerificationResult.VIOLATED)
unknown_count = sum(1 for r in state["verification_results"].values() if r == VerificationResult.UNKNOWN)
insights.summary += f" | Verified: {verified_count}, Violated: {violated_count}, Unknown: {unknown_count}"
return state
Explanation: Add verification statistics to insights summary.
Lines 394-460: Public Methods
def add_property(self, prop: VerificationProperty):
"""Add verification property."""
self.properties.append(prop)
Explanation: Register property to verify.
def add_invariant(self, name: str, checker: Callable, description: str):
"""Add invariant property."""
prop = VerificationProperty(
name=name,
verification_type=VerificationType.INVARIANT,
checker=checker,
description=description
)
self.add_property(prop)
Explanation: Convenience method for invariants. Example: add_invariant("balance_non_negative", lambda s: s["balance"] >= 0, "Balance is always non-negative").
def verify(self, system_state: Dict[str, Any]) -> Dict[str, Any]:
"""Verify system with workflow."""
initial_state: VerificationState = {
"properties": self.properties,
"evidence": {},
"system_state": system_state,
"verification_results": {},
"insights": None,
"proof_complete": False
}
final_state = self.workflow.invoke(initial_state)
return {
"results": final_state["verification_results"],
"evidence": final_state["evidence"],
"insights": final_state["insights"],
"proof_complete": final_state["proof_complete"]
}
Explanation: Main entry point – execute workflow and return verification results.
Lines 485-630: Demo Code
Scenario 1: Banking System Verification
verifier = pattern.create_verifier("BankingSystem")
# Invariant: balance never negative
def balance_non_negative(state: Dict) -> bool:
return state.get("balance", 0) >= 0
verifier.add_invariant("balance_non_negative", balance_non_negative, "Balance must be non-negative")
# Precondition: withdrawal amount valid
def withdrawal_valid(state: Dict) -> bool:
withdrawal = state.get("withdrawal_amount", 0)
balance = state.get("balance", 0)
return withdrawal > 0 and withdrawal <= balance
verifier.add_precondition("withdrawal_valid", withdrawal_valid, "Withdrawal must be positive and not exceed balance")
# Valid state
valid_state = {"balance": 1000, "withdrawal_amount": 100}
result1 = pattern.verify("BankingSystem", valid_state)
Explanation: Define invariants and preconditions for banking system. Valid state passes verification.
Scenario 2: Contract Violation
# Invalid state (negative balance)
invalid_state = {"balance": -500, "withdrawal_amount": 100}
result2 = pattern.verify("BankingSystem", invalid_state)
Explanation: Negative balance violates invariant, triggers proof generation with counterexample.
Scenario 3: E-commerce Order Verification
order_verifier = pattern.create_verifier("OrderProcessing")
def total_matches_items(state: Dict) -> bool:
"""Postcondition: order total matches sum of items."""
items = state.get("items", [])
total = state.get("total", 0)
calculated = sum(item.get("price", 0) * item.get("quantity", 0) for item in items)
return abs(total - calculated) < 0.01
order_verifier.add_postcondition("total_matches", total_matches_items, "Order total must equal sum of line items")
order_state = {
"total": 150.0,
"items": [
{"price": 50.0, "quantity": 2}, # 100
{"price": 25.0, "quantity": 2} # 50
]
}
result3 = pattern.verify("OrderProcessing", order_state)
Explanation: Postcondition verification ensures business logic correctness.
Verification Type Comparison
| Type | When | Example |
|---|---|---|
| PROPERTY | General system properties | “System is responsive” |
| INVARIANT | Always true | “Balance ≥ 0” |
| PRECONDITION | Before function | “Input not null” |
| POSTCONDITION | After function | “Result is sorted” |
| ASSERTION | Point in code | “Index in bounds” |
| CONTRACT | Interface guarantee | “Returns positive” |
Key Design Patterns
- Specification Pattern: Properties specify correctness criteria
- Chain of Responsibility: Sequential proof step validation
- Template Method: Common workflow with type-specific checkers
- Strategy Pattern: Different verification strategies per type
- Observer Pattern: Evidence collection observes all checks
Performance Considerations
- Property Checking: O(properties × checker_complexity) – depends on checker
- Evidence Collection: O(properties) – linear
- Proof Generation: O(LLM) – ~2-4 seconds for complex proofs
- Proof Validation: O(proof_steps) – linear in proof length
- Memory: O(properties + evidence) – stores all results
Best Practices: – Use fast checkers (avoid heavy computation) – Order properties by likelihood of violation (fail fast) – Cache verification results for immutable states – Use symbolic execution for exhaustive checking – Combine runtime checks with static analysis
Extension Points
- Add symbolic execution for path exploration
- Implement SMT solver integration (Z3, CVC4)
- Add model checking for state space exploration
- Support temporal logic (LTL, CTL)
- Add automated test generation from counterexamples
- Implement proof tactics library
- Add visualization of proof trees
Leave a Reply