On this tutorial, we implement a dual-agent governance system that applies Constitutional AI ideas to monetary operations. We exhibit 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 instantly into a proper structure and mixing rule-based checks with AI-assisted reasoning, we are able to construct programs which might be self-reflective, auditable, and resilient to dangerous or non-compliant habits 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 Checklist, Dict, Any, Elective, Literal
from pydantic import BaseModel, Subject, 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 information fashions, enums, and validation, whereas customary Python utilities deal with timestamps, parsing, and setting configuration. Try the FULL CODES right here.
class PolicyViolationType(str, Enum):
"""Forms 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"""
title: str
description: str
severity: Literal["low", "medium", "high", "critical"]
check_function: str
class Structure(BaseModel):
"""The 'Structure' - A algorithm that govern agent habits"""
insurance policies: Checklist[SafetyPolicy]
max_transaction_amount: float = 10000.0
require_approval_above: float = 5000.0
allowed_pii_fields: Checklist[str] = ["name", "account_id"]
def get_policy_by_name(self, title: str) -> Elective[SafetyPolicy]:
return subsequent((p for p in self.insurance policies if p.title == title), 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 habits by formalizing coverage sorts, severities, and enforcement guidelines. We encode monetary security constraints akin to PII safety, finances 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: Elective[float] = None
recipient: Elective[str] = None
description: str
justification: Elective[str] = None
metadata: Dict[str, Any] = Subject(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 = Subject(default_factory=lambda: datetime.now().isoformat())
class PolicyViolation(BaseModel):
"""Detected coverage violation"""
policy_name: str
violation_type: PolicyViolationType
severity: str
description: str
suggested_fix: Elective[str] = None
class AuditResult(BaseModel):
"""End result from the Auditor Agent"""
permitted: bool
violations: Checklist[PolicyViolation] = Subject(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 information fashions that construction how monetary requests, agent outputs, and audit findings move by way of the system. We use these schemas to make sure each motion, determination, 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: Checklist[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).exchange(',', '')) if amount_match else 0
if 'switch' in msg.decrease():
motion = 'switch'
elif 'cost' in msg.decrease() or 'pay' in msg.decrease():
motion = 'cost'
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",
"sort": "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",
"sort": "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": "Funds Limits",
"sort": "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",
"sort": "missing_justification",
"severity": "medium",
"element": "Excessive-value transaction lacks justification"
})
audit_result = {
"permitted": len(violations) == 0,
"violations": violations,
"risk_score": min(len(violations) * 30, 100),
"suggestions": "Transaction permitted" 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 habits of a big language mannequin by implementing a mock AI consumer that differentiates between employee and auditor roles. We deliberately inject coverage violations akin to PII leakage and finances points to stress-test the governance logic beneath sensible 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.hundreds(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 offered'}
Course of this request and return a JSON response with:
- action_taken
- quantity
- standing
- recipient
- account_id
- timestamp
- Another 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.hundreds(raw_audit)
besides json.JSONDecodeError:
audit_data = {"permitted": False, "violations": violations, "risk_score": 50}
consequence = AuditResult(
permitted=audit_data.get("permitted", 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("permitted", False) or len(violations) > 0
)
self.audit_history.append(consequence)
self._display_audit_result(consequence)
return consequence
def _check_rules(self, output: WorkerOutput) -> Checklist[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 subject"
))
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="Funds 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"Scale 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 subject explaining the transaction"
))
return violations
def _build_auditor_prompt(self, output: WorkerOutput, violations: Checklist[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 further evaluation and return JSON with:
- permitted (boolean)
- risk_score (0-100)
- suggestions (string)
- Any further considerations
Return ONLY legitimate JSON."""
def _display_audit_result(self, consequence: AuditResult):
"""Show audit leads to a readable format"""
print(f"n📊 AUDIT RESULTS:")
print(f"Standing: {'✅ APPROVED' if consequence.permitted else '❌ REJECTED'}")
print(f"Threat 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" Sort: {v.violation_type.worth}")
print(f" Problem: {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 obligations 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 way of 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]:
"""Primary 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.permitted:
print(f"n{'='*60}")
print(f"✅ FINAL RESULT: APPROVED")
print(f"{'='*60}")
return {
"standing": "permitted",
"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 (Important Violations)")
print(f"{'='*60}")
return {
"standing": "rejected",
"motive": "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",
"motive": "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="cost",
quantity=2500.00,
recipient="Vendor Corp",
description="Month-to-month software program license cost",
justification="Common recurring cost 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 cost to government",
justification="This fall efficiency bonus"
)
result2 = system.process_with_governance(request2)
print("nn" + "="*80)
print("EXAMPLE 3: Funds-Exceeding Transaction ($15,000)")
print("="*80)
request3 = FinancialRequest(
motion="switch",
quantity=15000.00,
recipient="Provider",
description="Massive 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('motive', 'N/A')}")
print(f"Instance 3: {result3['status'].higher()} - {result3.get('motive', '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.title} [{policy.severity.upper()}]")
print(f" {coverage.description}")We exhibit the system end-to-end by working sensible monetary eventualities that train each secure and unsafe behaviors. We present how the governance loop responds otherwise to compliant transactions, PII leaks, and finances 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, Funds, Authorization")
print("✓ Iterative Revision Loop - Self-correction mechanism")
print("✓ Threat Scoring - Quantitative security evaluation")
print("nNext Steps:")
print("• Exchange MockAIClient with actual Anthropic API")
print("• Implement precise revision logic in Employee Agent")
print("• Add extra subtle sample detection")
print("• Combine with actual monetary programs")
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 methods to operationalize Constitutional AI past idea and embed it into real-world monetary decision-making pipelines. We illustrated how we detect and reply to PII leakage, finances overruns, and lacking justifications whereas quantifying threat and imposing laborious governance boundaries. By orchestrating iterative evaluate loops between employee and auditor brokers, we demonstrated a sensible blueprint for constructing reliable, compliant, and scalable AI-driven monetary programs the place security and accountability are first-class design targets reasonably than afterthoughts.
Try the FULL CODES right here. Additionally, be happy to comply with us on Twitter and don’t neglect to hitch our 100k+ ML SubReddit and Subscribe to our Publication. Wait! are you on telegram? now you possibly can be a part of us on telegram as nicely.
Elevate your perspective with NextTech Information, the place innovation meets perception.
Uncover the most recent breakthroughs, get unique updates, and join with a worldwide community of future-focused thinkers.
Unlock tomorrow’s traits as we speak: learn extra, subscribe to our publication, and change into a part of the NextTech group at NextTech-news.com
