On this tutorial, we implement a dual-agent governance system that applies Constitutional AI rules to monetary operations. We show how we separate execution and oversight by pairing a Employee Agent that performs monetary actions with an Auditor Agent that enforces coverage, security, and compliance. By encoding governance guidelines immediately into a proper structure and mixing rule-based checks with AI-assisted reasoning, we will construct methods which are self-reflective, auditable, and resilient to dangerous or non-compliant conduct in high-stakes monetary workflows. Try the FULL CODES right here.
!pip set up -q pydantic anthropic python-dotenv
import json
import re
from typing import Record, Dict, Any, Non-compulsory, Literal
from pydantic import BaseModel, Area, validator
from enum import Enum
from datetime import datetime
import osWe set up and import the core libraries required to construction, validate, and govern our agent-based system. We depend on Pydantic for strongly typed knowledge fashions, enums, and validation, whereas normal Python utilities deal with timestamps, parsing, and surroundings configuration. Try the FULL CODES right here.
class PolicyViolationType(str, Enum):
"""Sorts of coverage violations"""
PII_EXPOSURE = "pii_exposure"
BUDGET_EXCEEDED = "budget_exceeded"
UNAUTHORIZED_ACTION = "unauthorized_action"
MISSING_JUSTIFICATION = "missing_justification"
SUSPICIOUS_PATTERN = "suspicious_pattern"
class SafetyPolicy(BaseModel):
"""Particular person security coverage rule"""
identify: str
description: str
severity: Literal["low", "medium", "high", "critical"]
check_function: str
class Structure(BaseModel):
"""The 'Structure' - A algorithm that govern agent conduct"""
insurance policies: Record[SafetyPolicy]
max_transaction_amount: float = 10000.0
require_approval_above: float = 5000.0
allowed_pii_fields: Record[str] = ["name", "account_id"]
def get_policy_by_name(self, identify: str) -> Non-compulsory[SafetyPolicy]:
return subsequent((p for p in self.insurance policies if p.identify == identify), None)
FINANCIAL_CONSTITUTION = Structure(
insurance policies=[
SafetyPolicy(
name="PII Protection",
description="Must not expose sensitive PII (SSN, full credit card, passwords)",
severity="critical",
check_function="Scan for SSN patterns, credit card numbers, passwords"
),
SafetyPolicy(
name="Budget Limits",
description="Transactions must not exceed predefined budget limits",
severity="high",
check_function="Check if transaction amount exceeds max_transaction_amount"
),
SafetyPolicy(
name="Action Authorization",
description="Only pre-approved action types are allowed",
severity="high",
check_function="Verify action type is in approved list"
),
SafetyPolicy(
name="Justification Required",
description="All transactions above threshold must have justification",
severity="medium",
check_function="Check for justification field in high-value transactions"
),
SafetyPolicy(
name="Pattern Detection",
description="Detect suspicious patterns (multiple rapid transactions, round numbers)",
severity="medium",
check_function="Analyze transaction patterns for anomalies"
)
],
max_transaction_amount=10000.0,
require_approval_above=5000.0
)
We outline the core constitutional framework that governs agent conduct by formalizing coverage sorts, severities, and enforcement guidelines. We encode monetary security constraints corresponding to PII safety, funds limits, authorization checks, and justification necessities as first-class, machine-readable insurance policies. Try the FULL CODES right here.
class FinancialRequest(BaseModel):
"""Enter request to the Employee Agent"""
motion: str
quantity: Non-compulsory[float] = None
recipient: Non-compulsory[str] = None
description: str
justification: Non-compulsory[str] = None
metadata: Dict[str, Any] = Area(default_factory=dict)
class WorkerOutput(BaseModel):
"""Output from the Employee Agent"""
request_id: str
action_taken: str
particulars: Dict[str, Any]
raw_response: str
timestamp: str = Area(default_factory=lambda: datetime.now().isoformat())
class PolicyViolation(BaseModel):
"""Detected coverage violation"""
policy_name: str
violation_type: PolicyViolationType
severity: str
description: str
suggested_fix: Non-compulsory[str] = None
class AuditResult(BaseModel):
"""End result from the Auditor Agent"""
authorized: bool
violations: Record[PolicyViolation] = Area(default_factory=checklist)
risk_score: float # 0-100
suggestions: str
revision_needed: bool
@classmethod
def validate_risk_score(cls, v):
if isinstance(v, (int, float)):
return max(0.0, min(100.0, v))
return vWe outline strongly typed knowledge fashions that construction how monetary requests, agent outputs, and audit findings move by the system. We use these schemas to make sure each motion, choice, and violation is captured in a constant, machine-validated format with full traceability. Try the FULL CODES right here.
class MockAIClient:
"""Simulates the Anthropic API for this tutorial"""
def __init__(self):
self.call_count = 0
def messages_create(self, mannequin: str, max_tokens: int, messages: Record[Dict]) -> Any:
"""Simulate API name"""
self.call_count += 1
user_msg = messages[-1]["content"]
if "WORKER AGENT" in user_msg or "monetary request" in user_msg.decrease():
return self._worker_response(user_msg)
elif "AUDITOR AGENT" in user_msg or "audit" in user_msg.decrease():
return self._auditor_response(user_msg)
return self._default_response()
def _worker_response(self, msg: str) -> Any:
"""Simulate employee agent processing a request"""
amount_match = re.search(r'$?(d+(?:,d{3})*(?:.d{2})?)', msg)
quantity = float(amount_match.group(1).change(',', '')) if amount_match else 0
if 'switch' in msg.decrease():
motion = 'switch'
elif 'fee' in msg.decrease() or 'pay' in msg.decrease():
motion = 'fee'
elif 'report' in msg.decrease():
motion = 'report'
else:
motion = 'general_query'
response = {
"action_taken": motion,
"quantity": quantity,
"standing": "accomplished",
"recipient": "John Doe" if quantity > 0 else None,
"account_id": "ACC-12345",
"timestamp": datetime.now().isoformat()
}
if quantity > 5000:
response["ssn"] = "123-45-6789"
if quantity > 8000:
response["credit_card"] = "4532-1234-5678-9010"
class MockResponse:
def __init__(self, content material):
self.content material = [type('obj', (object,), {
'type': 'text',
'text': json.dumps(content, indent=2)
})]
return MockResponse(response)
def _auditor_response(self, msg: str) -> Any:
"""Simulate auditor agent checking insurance policies"""
violations = []
if 'ssn' in msg.decrease() or re.search(r'd{3}-d{2}-d{4}', msg):
violations.append({
"coverage": "PII Safety",
"kind": "pii_exposure",
"severity": "important",
"element": "SSN detected in output"
})
if 'credit_card' in msg.decrease() or re.search(r'd{4}-d{4}-d{4}-d{4}', msg):
violations.append({
"coverage": "PII Safety",
"kind": "pii_exposure",
"severity": "important",
"element": "Bank card quantity detected"
})
amount_match = re.search(r'"quantity":s*(d+(?:.d+)?)', msg)
if amount_match:
quantity = float(amount_match.group(1))
if quantity > 10000:
violations.append({
"coverage": "Finances Limits",
"kind": "budget_exceeded",
"severity": "excessive",
"element": f"Quantity ${quantity} exceeds restrict of $10,000"
})
elif quantity > 5000 and 'justification' not in msg.decrease():
violations.append({
"coverage": "Justification Required",
"kind": "missing_justification",
"severity": "medium",
"element": "Excessive-value transaction lacks justification"
})
audit_result = {
"authorized": len(violations) == 0,
"violations": violations,
"risk_score": min(len(violations) * 30, 100),
"suggestions": "Transaction authorized" if len(violations) == 0 else "Violations detected - revision required"
}
class MockResponse:
def __init__(self, content material):
self.content material = [type('obj', (object,), {
'type': 'text',
'text': json.dumps(content, indent=2)
})]
return MockResponse(audit_result)
def _default_response(self) -> Any:
class MockResponse:
def __init__(self):
self.content material = [type('obj', (object,), {
'type': 'text',
'text': '{"status": "acknowledged"}'
})]
return MockResponse()We simulate the conduct of a giant language mannequin by implementing a mock AI consumer that differentiates between employee and auditor roles. We deliberately inject coverage violations corresponding to PII leakage and funds points to stress-test the governance logic underneath real looking failure situations. Try the FULL CODES right here.
class WorkerAgent:
"""Agent A - The Employee that processes monetary requests"""
def __init__(self, consumer: MockAIClient):
self.consumer = consumer
self.function = "Monetary Operations Employee"
self.processed_requests = []
def process_request(self, request: FinancialRequest) -> WorkerOutput:
"""Course of a monetary request"""
print(f"n{'='*60}")
print(f"🔧 WORKER AGENT: Processing request...")
print(f"{'='*60}")
print(f"Motion: {request.motion}")
if request.quantity:
print(f"Quantity: ${request.quantity:,.2f}")
else:
print("Quantity: N/A")
print(f"Description: {request.description}")
immediate = self._build_worker_prompt(request)
response = self.consumer.messages_create(
mannequin="claude-sonnet-4-20250514",
max_tokens=1000,
messages=[{"role": "user", "content": prompt}]
)
raw_response = response.content material[0].textual content
strive:
particulars = json.masses(raw_response)
besides json.JSONDecodeError:
particulars = {"uncooked": raw_response}
output = WorkerOutput(
request_id=f"REQ-{len(self.processed_requests)+1:04d}",
action_taken=request.motion,
particulars=particulars,
raw_response=raw_response
)
self.processed_requests.append(output)
print(f"n✅ Employee accomplished processing (ID: {output.request_id})")
return output
def _build_worker_prompt(self, request: FinancialRequest) -> str:
"""Construct immediate for employee agent"""
amount_str = f"${request.quantity:,.2f}" if request.quantity else "$0.00"
return f"""You're a WORKER AGENT processing a monetary request.
Request Particulars:
- Motion: {request.motion}
- Quantity: {amount_str}
- Recipient: {request.recipient or 'N/A'}
- Description: {request.description}
- Justification: {request.justification or 'None supplied'}
Course of this request and return a JSON response with:
- action_taken
- quantity
- standing
- recipient
- account_id
- timestamp
- Some other related particulars
Return ONLY legitimate JSON."""
class AuditorAgent:
"""Agent B - The Auditor that validates employee output"""
def __init__(self, consumer: MockAIClient, structure: Structure):
self.consumer = consumer
self.structure = structure
self.function = "Governance Auditor"
self.audit_history = []
def audit(self, worker_output: WorkerOutput) -> AuditResult:
"""Audit the employee's output towards the structure"""
print(f"n{'='*60}")
print(f"🔍 AUDITOR AGENT: Auditing output...")
print(f"{'='*60}")
violations = self._check_rules(worker_output)
immediate = self._build_auditor_prompt(worker_output, violations)
response = self.consumer.messages_create(
mannequin="claude-sonnet-4-20250514",
max_tokens=1000,
messages=[{"role": "user", "content": prompt}]
)
raw_audit = response.content material[0].textual content
strive:
audit_data = json.masses(raw_audit)
besides json.JSONDecodeError:
audit_data = {"authorized": False, "violations": violations, "risk_score": 50}
consequence = AuditResult(
authorized=audit_data.get("authorized", False) and len(violations) == 0,
violations=violations,
risk_score=audit_data.get("risk_score", len(violations) * 25),
suggestions=audit_data.get("suggestions", "Audit accomplished"),
revision_needed=not audit_data.get("authorized", False) or len(violations) > 0
)
self.audit_history.append(consequence)
self._display_audit_result(consequence)
return consequence
def _check_rules(self, output: WorkerOutput) -> Record[PolicyViolation]:
"""Carry out rule-based constitutional checks"""
violations = []
details_str = json.dumps(output.particulars)
if re.search(r'd{3}-d{2}-d{4}', details_str):
violations.append(PolicyViolation(
policy_name="PII Safety",
violation_type=PolicyViolationType.PII_EXPOSURE,
severity="important",
description="Social Safety Quantity detected in output",
suggested_fix="Take away or masks SSN area"
))
if re.search(r'd{4}[-s]?d{4}[-s]?d{4}[-s]?d{4}', details_str):
violations.append(PolicyViolation(
policy_name="PII Safety",
violation_type=PolicyViolationType.PII_EXPOSURE,
severity="important",
description="Bank card quantity detected in output",
suggested_fix="Take away or tokenize bank card quantity"
))
quantity = output.particulars.get("quantity", 0)
if quantity > self.structure.max_transaction_amount:
violations.append(PolicyViolation(
policy_name="Finances Limits",
violation_type=PolicyViolationType.BUDGET_EXCEEDED,
severity="excessive",
description=f"Quantity ${quantity:,.2f} exceeds restrict of ${self.structure.max_transaction_amount:,.2f}",
suggested_fix=f"Cut back quantity to ${self.structure.max_transaction_amount:,.2f} or request approval"
))
if quantity > self.structure.require_approval_above:
if "justification" not in details_str.decrease():
violations.append(PolicyViolation(
policy_name="Justification Required",
violation_type=PolicyViolationType.MISSING_JUSTIFICATION,
severity="medium",
description=f"Transaction of ${quantity:,.2f} requires justification",
suggested_fix="Add justification area explaining the transaction"
))
return violations
def _build_auditor_prompt(self, output: WorkerOutput, violations: Record[PolicyViolation]) -> str:
"""Construct immediate for auditor agent"""
return f"""You're an AUDITOR AGENT validating monetary operations towards a Structure.
Structure Insurance policies:
{json.dumps([p.dict() for p in self.constitution.policies], indent=2)}
Employee Output to Audit:
{output.raw_response}
Already Detected Violations:
{json.dumps([v.dict() for v in violations], indent=2)}
Carry out extra evaluation and return JSON with:
- authorized (boolean)
- risk_score (0-100)
- suggestions (string)
- Any extra considerations
Return ONLY legitimate JSON."""
def _display_audit_result(self, consequence: AuditResult):
"""Show audit ends in a readable format"""
print(f"n📊 AUDIT RESULTS:")
print(f"Standing: {'✅ APPROVED' if consequence.authorized else '❌ REJECTED'}")
print(f"Danger Rating: {consequence.risk_score:.1f}/100")
print(f"Violations Discovered: {len(consequence.violations)}")
if consequence.violations:
print(f"n⚠️ POLICY VIOLATIONS:")
for i, v in enumerate(consequence.violations, 1):
print(f"n {i}. {v.policy_name} [{v.severity.upper()}]")
print(f" Kind: {v.violation_type.worth}")
print(f" Concern: {v.description}")
if v.suggested_fix:
print(f" Repair: {v.suggested_fix}")
print(f"n💬 Suggestions: {consequence.suggestions}")
print(f"Revision Wanted: {'Sure' if consequence.revision_needed else 'No'}")We implement the core dual-agent logic by separating execution and governance tasks between a Employee Agent and an Auditor Agent. We permit the employee to focus purely on fulfilling monetary requests, whereas we implement constitutional guidelines by deterministic checks and AI-assisted auditing. By combining structured prompts, rule-based validation, and clear audit suggestions, we create a self-reflective management loop that prioritizes security, accountability, and compliance. Try the FULL CODES right here.
class GovernanceSystem:
"""Orchestrates the dual-agent governance workflow"""
def __init__(self, structure: Structure):
self.consumer = MockAIClient()
self.employee = WorkerAgent(self.consumer)
self.auditor = AuditorAgent(self.consumer, structure)
self.structure = structure
self.max_revision_attempts = 3
def process_with_governance(self, request: FinancialRequest) -> Dict[str, Any]:
"""Major workflow: Employee processes, Auditor validates, loop if wanted"""
print(f"n{'#'*60}")
print(f"# GOVERNANCE SYSTEM: New Request")
print(f"{'#'*60}")
try = 0
whereas try < self.max_revision_attempts:
try += 1
print(f"n🔄 Try {try}/{self.max_revision_attempts}")
worker_output = self.employee.process_request(request)
audit_result = self.auditor.audit(worker_output)
if audit_result.authorized:
print(f"n{'='*60}")
print(f"✅ FINAL RESULT: APPROVED")
print(f"{'='*60}")
return {
"standing": "authorized",
"output": worker_output.dict(),
"audit": audit_result.dict(),
"makes an attempt": try
}
critical_violations = [v for v in audit_result.violations if v.severity == "critical"]
if critical_violations:
print(f"n{'='*60}")
print(f"🛑 FINAL RESULT: REJECTED (Essential Violations)")
print(f"{'='*60}")
return {
"standing": "rejected",
"cause": "critical_violations",
"audit": audit_result.dict(),
"makes an attempt": try
}
if try >= self.max_revision_attempts:
print(f"n{'='*60}")
print(f"🛑 FINAL RESULT: REJECTED (Max Makes an attempt)")
print(f"{'='*60}")
return {
"standing": "rejected",
"cause": "max_attempts_exceeded",
"audit": audit_result.dict(),
"makes an attempt": try
}
return {"standing": "error", "message": "Sudden exit from loop"}We orchestrate the entire governance workflow by coordinating the employee and auditor brokers inside a managed revision loop. We consider every try towards constitutional guidelines and instantly halt execution when important violations are detected. Try the FULL CODES right here.
def run_examples():
"""Run demonstration examples"""
print("="*80)
print(" DUAL-AGENT GOVERNANCE SYSTEM WITH CONSTITUTIONAL AI")
print(" Tutorial: Self-Reflective Monetary Operations Brokers")
print("="*80)
system = GovernanceSystem(FINANCIAL_CONSTITUTION)
print("nn" + "="*80)
print("EXAMPLE 1: Secure Transaction ($2,500)")
print("="*80)
request1 = FinancialRequest(
motion="fee",
quantity=2500.00,
recipient="Vendor Corp",
description="Month-to-month software program license fee",
justification="Common recurring fee for important companies"
)
result1 = system.process_with_governance(request1)
print("nn" + "="*80)
print("EXAMPLE 2: Excessive-Worth Transaction with PII Leak ($7,500)")
print("="*80)
request2 = FinancialRequest(
motion="switch",
quantity=7500.00,
recipient="Government",
description="Bonus fee to government",
justification="This fall efficiency bonus"
)
result2 = system.process_with_governance(request2)
print("nn" + "="*80)
print("EXAMPLE 3: Finances-Exceeding Transaction ($15,000)")
print("="*80)
request3 = FinancialRequest(
motion="switch",
quantity=15000.00,
recipient="Provider",
description="Giant tools buy",
justification="New manufacturing tools for manufacturing line"
)
result3 = system.process_with_governance(request3)
print("nn" + "="*80)
print(" SUMMARY OF RESULTS")
print("="*80)
print(f"nExample 1: {result1['status'].higher()}")
print(f"Instance 2: {result2['status'].higher()} - {result2.get('cause', 'N/A')}")
print(f"Instance 3: {result3['status'].higher()} - {result3.get('cause', 'N/A')}")
print(f"nnTotal API Calls: {system.consumer.call_count}")
print(f"Employee Processed: {len(system.employee.processed_requests)} requests")
print(f"Auditor Carried out: {len(system.auditor.audit_history)} audits")
print("nn" + "="*80)
print(" ACTIVE CONSTITUTION")
print("="*80)
for coverage in FINANCIAL_CONSTITUTION.insurance policies:
print(f"n📜 {coverage.identify} [{policy.severity.upper()}]")
print(f" {coverage.description}")We show the system end-to-end by working real looking monetary situations that train each secure and unsafe behaviors. We present how the governance loop responds in a different way to compliant transactions, PII leaks, and funds violations whereas producing clear audit outcomes. Try the FULL CODES right here.
if __name__ == "__main__":
run_examples()
print("nn" + "="*80)
print(" 🎓 TUTORIAL COMPLETE!")
print("="*80)
print("nKey Ideas Demonstrated:")
print("✓ Constitutional AI - Rule-based governance")
print("✓ Twin-Agent System - Employee + Auditor sample")
print("✓ Coverage Violation Detection - PII, Finances, Authorization")
print("✓ Iterative Revision Loop - Self-correction mechanism")
print("✓ Danger Scoring - Quantitative security evaluation")
print("nNext Steps:")
print("• Substitute MockAIClient with actual Anthropic API")
print("• Implement precise revision logic in Employee Agent")
print("• Add extra refined sample detection")
print("• Combine with actual monetary methods")
print("• Construct logging and monitoring dashboard")
print("="*80)We conclude the tutorial by executing all examples and clearly surfacing the core ideas demonstrated by the system. We recap how constitutional guidelines, dual-agent governance, violation detection, and threat scoring work collectively in apply.
In conclusion, we demonstrated the way to operationalize Constitutional AI past concept and embed it into real-world monetary decision-making pipelines. We illustrated how we detect and reply to PII leakage, funds overruns, and lacking justifications whereas quantifying threat and imposing arduous governance boundaries. By orchestrating iterative overview loops between employee and auditor brokers, we demonstrated a sensible blueprint for constructing reliable, compliant, and scalable AI-driven monetary methods the place security and accountability are first-class design objectives somewhat than afterthoughts.
Try the FULL CODES right here. Additionally, be happy to observe us on Twitter and don’t neglect to affix our 100k+ ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you possibly can be part of us on telegram as properly.
