{"id":330,"date":"2026-01-29T09:32:47","date_gmt":"2026-01-29T01:32:47","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=330"},"modified":"2026-01-29T09:32:47","modified_gmt":"2026-01-29T01:32:47","slug":"how-to-design-self-reflective-dual-agent-governance-systems-with-constitutional-ai-for-secure-and-compliant-financial-operations","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=330","title":{"rendered":"How to Design Self-Reflective Dual-Agent Governance Systems with Constitutional AI for Secure and Compliant Financial Operations"},"content":{"rendered":"

In this tutorial, we implement a dual-agent governance system that applies Constitutional AI principles to financial operations. We demonstrate how we separate execution and oversight by pairing a Worker Agent that performs financial actions with an Auditor Agent that enforces policy, safety, and compliance. By encoding governance rules directly into a formal constitution and combining rule-based checks with AI-assisted reasoning, we can build systems that are self-reflective, auditable, and resilient to risky or non-compliant behavior in high-stakes financial workflows. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
!pip install -q pydantic anthropic python-dotenv\n\n\nimport json\nimport re\nfrom typing import List, Dict, Any, Optional, Literal\nfrom pydantic import BaseModel, Field, validator\nfrom enum import Enum\nfrom datetime import datetime\nimport os<\/code><\/pre>\n<\/div>\n<\/div>\n
We install and import the core libraries required to structure, validate, and govern our agent-based system. We rely on Pydantic for strongly typed data models, enums, and validation, while standard Python utilities handle timestamps, parsing, and environment configuration. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class PolicyViolationType(str, Enum):\n   \"\"\"Types of policy violations\"\"\"\n   PII_EXPOSURE = \"pii_exposure\"\n   BUDGET_EXCEEDED = \"budget_exceeded\"\n   UNAUTHORIZED_ACTION = \"unauthorized_action\"\n   MISSING_JUSTIFICATION = \"missing_justification\"\n   SUSPICIOUS_PATTERN = \"suspicious_pattern\"\n\n\nclass SafetyPolicy(BaseModel):\n   \"\"\"Individual safety policy rule\"\"\"\n   name: str\n   description: str\n   severity: Literal[\"low\", \"medium\", \"high\", \"critical\"]\n   check_function: str \n\n\nclass Constitution(BaseModel):\n   \"\"\"The 'Constitution' - A set of rules that govern agent behavior\"\"\"\n   policies: List[SafetyPolicy]\n   max_transaction_amount: float = 10000.0\n   require_approval_above: float = 5000.0\n   allowed_pii_fields: List[str] = [\"name\", \"account_id\"]\n  \n   def get_policy_by_name(self, name: str) -> Optional[SafetyPolicy]:\n       return next((p for p in self.policies if p.name == name), None)\n\n\nFINANCIAL_CONSTITUTION = Constitution(\n   policies=[\n       SafetyPolicy(\n           name=\"PII Protection\",\n           description=\"Must not expose sensitive PII (SSN, full credit card, passwords)\",\n           severity=\"critical\",\n           check_function=\"Scan for SSN patterns, credit card numbers, passwords\"\n       ),\n       SafetyPolicy(\n           name=\"Budget Limits\",\n           description=\"Transactions must not exceed predefined budget limits\",\n           severity=\"high\",\n           check_function=\"Check if transaction amount exceeds max_transaction_amount\"\n       ),\n       SafetyPolicy(\n           name=\"Action Authorization\",\n           description=\"Only pre-approved action types are allowed\",\n           severity=\"high\",\n           check_function=\"Verify action type is in approved list\"\n       ),\n       SafetyPolicy(\n           name=\"Justification Required\",\n           description=\"All transactions above threshold must have justification\",\n           severity=\"medium\",\n           check_function=\"Check for justification field in high-value transactions\"\n       ),\n       SafetyPolicy(\n           name=\"Pattern Detection\",\n           description=\"Detect suspicious patterns (multiple rapid transactions, round numbers)\",\n           severity=\"medium\",\n           check_function=\"Analyze transaction patterns for anomalies\"\n       )\n   ],\n   max_transaction_amount=10000.0,\n   require_approval_above=5000.0\n)\n<\/code><\/pre>\n<\/div>\n<\/div>\n
We define the core constitutional framework that governs agent behavior by formalizing policy types, severities, and enforcement rules. We encode financial safety constraints such as PII protection, budget limits, authorization checks, and justification requirements as first-class, machine-readable policies. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class FinancialRequest(BaseModel):\n   \"\"\"Input request to the Worker Agent\"\"\"\n   action: str \n   amount: Optional[float] = None\n   recipient: Optional[str] = None\n   description: str\n   justification: Optional[str] = None\n   metadata: Dict[str, Any] = Field(default_factory=dict)\n\n\nclass WorkerOutput(BaseModel):\n   \"\"\"Output from the Worker Agent\"\"\"\n   request_id: str\n   action_taken: str\n   details: Dict[str, Any]\n   raw_response: str\n   timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())\n\n\nclass PolicyViolation(BaseModel):\n   \"\"\"Detected policy violation\"\"\"\n   policy_name: str\n   violation_type: PolicyViolationType\n   severity: str\n   description: str\n   suggested_fix: Optional[str] = None\n\n\nclass AuditResult(BaseModel):\n   \"\"\"Result from the Auditor Agent\"\"\"\n   approved: bool\n   violations: List[PolicyViolation] = Field(default_factory=list)\n   risk_score: float  # 0-100\n   feedback: str\n   revision_needed: bool\n  \n   @classmethod\n   def validate_risk_score(cls, v):\n       if isinstance(v, (int, float)):\n           return max(0.0, min(100.0, v))\n       return v<\/code><\/pre>\n<\/div>\n<\/div>\n
We define strongly typed data models that structure how financial requests, agent outputs, and audit findings flow through the system. We use these schemas to ensure every action, decision, and violation is captured in a consistent, machine-validated format with full traceability. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class MockAIClient:\n   \"\"\"Simulates the Anthropic API for this tutorial\"\"\"\n  \n   def __init__(self):\n       self.call_count = 0\n  \n   def messages_create(self, model: str, max_tokens: int, messages: List[Dict]) -> Any:\n       \"\"\"Simulate API call\"\"\"\n       self.call_count += 1\n       user_msg = messages[-1][\"content\"]\n      \n       if \"WORKER AGENT\" in user_msg or \"financial request\" in user_msg.lower():\n           return self._worker_response(user_msg)\n      \n       elif \"AUDITOR AGENT\" in user_msg or \"audit\" in user_msg.lower():\n           return self._auditor_response(user_msg)\n      \n       return self._default_response()\n  \n   def _worker_response(self, msg: str) -> Any:\n       \"\"\"Simulate worker agent processing a request\"\"\"\n      \n       amount_match = re.search(r'$?(d+(?:,d{3})*(?:.d{2})?)', msg)\n       amount = float(amount_match.group(1).replace(',', '')) if amount_match else 0\n      \n       if 'transfer' in msg.lower():\n           action = 'transfer'\n       elif 'payment' in msg.lower() or 'pay' in msg.lower():\n           action = 'payment'\n       elif 'report' in msg.lower():\n           action = 'report'\n       else:\n           action = 'general_query'\n      \n       response = {\n           \"action_taken\": action,\n           \"amount\": amount,\n           \"status\": \"completed\",\n           \"recipient\": \"John Doe\" if amount > 0 else None,\n           \"account_id\": \"ACC-12345\",\n           \"timestamp\": datetime.now().isoformat()\n       }\n      \n       if amount > 5000:\n           response[\"ssn\"] = \"123-45-6789\" \n      \n       if amount > 8000:\n           response[\"credit_card\"] = \"4532-1234-5678-9010\" \n      \n       class MockResponse:\n           def __init__(self, content):\n               self.content = [type('obj', (object,), {\n                   'type': 'text',\n                   'text': json.dumps(content, indent=2)\n               })]\n      \n       return MockResponse(response)\n  \n   def _auditor_response(self, msg: str) -> Any:\n       \"\"\"Simulate auditor agent checking policies\"\"\"\n      \n       violations = []\n      \n       if 'ssn' in msg.lower() or re.search(r'd{3}-d{2}-d{4}', msg):\n           violations.append({\n               \"policy\": \"PII Protection\",\n               \"type\": \"pii_exposure\",\n               \"severity\": \"critical\",\n               \"detail\": \"SSN detected in output\"\n           })\n      \n       if 'credit_card' in msg.lower() or re.search(r'd{4}-d{4}-d{4}-d{4}', msg):\n           violations.append({\n               \"policy\": \"PII Protection\",\n               \"type\": \"pii_exposure\",\n               \"severity\": \"critical\",\n               \"detail\": \"Credit card number detected\"\n           })\n      \n       amount_match = re.search(r'\"amount\":s*(d+(?:.d+)?)', msg)\n       if amount_match:\n           amount = float(amount_match.group(1))\n           if amount > 10000:\n               violations.append({\n                   \"policy\": \"Budget Limits\",\n                   \"type\": \"budget_exceeded\",\n                   \"severity\": \"high\",\n                   \"detail\": f\"Amount ${amount} exceeds limit of $10,000\"\n               })\n           elif amount > 5000 and 'justification' not in msg.lower():\n               violations.append({\n                   \"policy\": \"Justification Required\",\n                   \"type\": \"missing_justification\",\n                   \"severity\": \"medium\",\n                   \"detail\": \"High-value transaction lacks justification\"\n               })\n      \n       audit_result = {\n           \"approved\": len(violations) == 0,\n           \"violations\": violations,\n           \"risk_score\": min(len(violations) * 30, 100),\n           \"feedback\": \"Transaction approved\" if len(violations) == 0 else \"Violations detected - revision required\"\n       }\n      \n       class MockResponse:\n           def __init__(self, content):\n               self.content = [type('obj', (object,), {\n                   'type': 'text',\n                   'text': json.dumps(content, indent=2)\n               })]\n      \n       return MockResponse(audit_result)\n  \n   def _default_response(self) -> Any:\n       class MockResponse:\n           def __init__(self):\n               self.content = [type('obj', (object,), {\n                   'type': 'text',\n                   'text': '{\"status\": \"acknowledged\"}'\n               })]\n       return MockResponse()<\/code><\/pre>\n<\/div>\n<\/div>\n
We simulate the behavior of a large language model by implementing a mock AI client that differentiates between worker and auditor roles. We intentionally inject policy violations such as PII leakage and budget issues to stress-test the governance logic under realistic failure conditions. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class WorkerAgent:\n   \"\"\"Agent A - The Worker that processes financial requests\"\"\"\n  \n   def __init__(self, client: MockAIClient):\n       self.client = client\n       self.role = \"Financial Operations Worker\"\n       self.processed_requests = []\n  \n   def process_request(self, request: FinancialRequest) -> WorkerOutput:\n       \"\"\"Process a financial request\"\"\"\n       print(f\"n{'='*60}\")\n       print(f\"\"\ud83d\udd27\" WORKER AGENT: Processing request...\")\n       print(f\"{'='*60}\")\n       print(f\"Action: {request.action}\")\n       if request.amount:\n           print(f\"Amount: ${request.amount:,.2f}\")\n       else:\n           print(\"Amount: N\/A\")\n       print(f\"Description: {request.description}\")\n      \n       prompt = self._build_worker_prompt(request)\n      \n       response = self.client.messages_create(\n           model=\"claude-sonnet-4-20250514\",\n           max_tokens=1000,\n           messages=[{\"role\": \"user\", \"content\": prompt}]\n       )\n      \n       raw_response = response.content[0].text\n      \n       try:\n           details = json.loads(raw_response)\n       except json.JSONDecodeError:\n           details = {\"raw\": raw_response}\n      \n       output = WorkerOutput(\n           request_id=f\"REQ-{len(self.processed_requests)+1:04d}\",\n           action_taken=request.action,\n           details=details,\n           raw_response=raw_response\n       )\n      \n       self.processed_requests.append(output)\n       print(f\"n\"\u2705\" Worker completed processing (ID: {output.request_id})\")\n      \n       return output\n  \n   def _build_worker_prompt(self, request: FinancialRequest) -> str:\n       \"\"\"Build prompt for worker agent\"\"\"\n       amount_str = f\"${request.amount:,.2f}\" if request.amount else \"$0.00\"\n       return f\"\"\"You are a WORKER AGENT processing a financial request.\n\n\nRequest Details:\n- Action: {request.action}\n- Amount: {amount_str}\n- Recipient: {request.recipient or 'N\/A'}\n- Description: {request.description}\n- Justification: {request.justification or 'None provided'}\n\n\nProcess this request and return a JSON response with:\n- action_taken\n- amount\n- status\n- recipient\n- account_id\n- timestamp\n- Any other relevant details\n\n\nReturn ONLY valid JSON.\"\"\"\n\n\nclass AuditorAgent:\n   \"\"\"Agent B - The Auditor that validates worker output\"\"\"\n  \n   def __init__(self, client: MockAIClient, constitution: Constitution):\n       self.client = client\n       self.constitution = constitution\n       self.role = \"Governance Auditor\"\n       self.audit_history = []\n  \n   def audit(self, worker_output: WorkerOutput) -> AuditResult:\n       \"\"\"Audit the worker's output against the constitution\"\"\"\n       print(f\"n{'='*60}\")\n       print(f\"\"\ud83d\udd0d\" AUDITOR AGENT: Auditing output...\")\n       print(f\"{'='*60}\")\n      \n       violations = self._check_rules(worker_output)\n      \n       prompt = self._build_auditor_prompt(worker_output, violations)\n      \n       response = self.client.messages_create(\n           model=\"claude-sonnet-4-20250514\",\n           max_tokens=1000,\n           messages=[{\"role\": \"user\", \"content\": prompt}]\n       )\n      \n       raw_audit = response.content[0].text\n       try:\n           audit_data = json.loads(raw_audit)\n       except json.JSONDecodeError:\n           audit_data = {\"approved\": False, \"violations\": violations, \"risk_score\": 50}\n      \n       result = AuditResult(\n           approved=audit_data.get(\"approved\", False) and len(violations) == 0,\n           violations=violations,\n           risk_score=audit_data.get(\"risk_score\", len(violations) * 25),\n           feedback=audit_data.get(\"feedback\", \"Audit completed\"),\n           revision_needed=not audit_data.get(\"approved\", False) or len(violations) > 0\n       )\n      \n       self.audit_history.append(result)\n      \n       self._display_audit_result(result)\n      \n       return result\n  \n   def _check_rules(self, output: WorkerOutput) -> List[PolicyViolation]:\n       \"\"\"Perform rule-based constitutional checks\"\"\"\n       violations = []\n       details_str = json.dumps(output.details)\n      \n       if re.search(r'd{3}-d{2}-d{4}', details_str):\n           violations.append(PolicyViolation(\n               policy_name=\"PII Protection\",\n               violation_type=PolicyViolationType.PII_EXPOSURE,\n               severity=\"critical\",\n               description=\"Social Security Number detected in output\",\n               suggested_fix=\"Remove or mask SSN field\"\n           ))\n      \n       if re.search(r'd{4}[-s]?d{4}[-s]?d{4}[-s]?d{4}', details_str): \n           violations.append(PolicyViolation(\n               policy_name=\"PII Protection\",\n               violation_type=PolicyViolationType.PII_EXPOSURE,\n               severity=\"critical\",\n               description=\"Credit card number detected in output\",\n               suggested_fix=\"Remove or tokenize credit card number\"\n           ))\n      \n       amount = output.details.get(\"amount\", 0)\n       if amount > self.constitution.max_transaction_amount:\n           violations.append(PolicyViolation(\n               policy_name=\"Budget Limits\",\n               violation_type=PolicyViolationType.BUDGET_EXCEEDED,\n               severity=\"high\",\n               description=f\"Amount ${amount:,.2f} exceeds limit of ${self.constitution.max_transaction_amount:,.2f}\",\n               suggested_fix=f\"Reduce amount to ${self.constitution.max_transaction_amount:,.2f} or request approval\"\n           ))\n      \n       if amount > self.constitution.require_approval_above:\n           if \"justification\" not in details_str.lower():\n               violations.append(PolicyViolation(\n                   policy_name=\"Justification Required\",\n                   violation_type=PolicyViolationType.MISSING_JUSTIFICATION,\n                   severity=\"medium\",\n                   description=f\"Transaction of ${amount:,.2f} requires justification\",\n                   suggested_fix=\"Add justification field explaining the transaction\"\n               ))\n      \n       return violations\n  \n   def _build_auditor_prompt(self, output: WorkerOutput, violations: List[PolicyViolation]) -> str:\n       \"\"\"Build prompt for auditor agent\"\"\"\n       return f\"\"\"You are an AUDITOR AGENT validating financial operations against a Constitution.\n\n\nConstitution Policies:\n{json.dumps([p.dict() for p in self.constitution.policies], indent=2)}\n\n\nWorker Output to Audit:\n{output.raw_response}\n\n\nAlready Detected Violations:\n{json.dumps([v.dict() for v in violations], indent=2)}\n\n\nPerform additional analysis and return JSON with:\n- approved (boolean)\n- risk_score (0-100)\n- feedback (string)\n- Any additional concerns\n\n\nReturn ONLY valid JSON.\"\"\"\n  \n   def _display_audit_result(self, result: AuditResult):\n       \"\"\"Display audit results in a readable format\"\"\"\n       print(f\"n\"\ud83d\udcca\" AUDIT RESULTS:\")\n       print(f\"Status: {'\"\u2705\" APPROVED' if result.approved else '\"\u274c\" REJECTED'}\")\n       print(f\"Risk Score: {result.risk_score:.1f}\/100\")\n       print(f\"Violations Found: {len(result.violations)}\")\n      \n       if result.violations:\n           print(f\"n\"\u26a0\"  POLICY VIOLATIONS:\")\n           for i, v in enumerate(result.violations, 1):\n               print(f\"n  {i}. {v.policy_name} [{v.severity.upper()}]\")\n               print(f\"     Type: {v.violation_type.value}\")\n               print(f\"     Issue: {v.description}\")\n               if v.suggested_fix:\n                   print(f\"     Fix: {v.suggested_fix}\")\n      \n       print(f\"n\"\ud83d\udcac\" Feedback: {result.feedback}\")\n       print(f\"Revision Needed: {'Yes' if result.revision_needed else 'No'}\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We implement the core dual-agent logic by separating execution and governance responsibilities between a Worker Agent and an Auditor Agent. We allow the worker to focus purely on fulfilling financial requests, while we enforce constitutional rules through deterministic checks and AI-assisted auditing. By combining structured prompts, rule-based validation, and clear audit feedback, we create a self-reflective control loop that prioritizes safety, accountability, and compliance. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class GovernanceSystem:\n   \"\"\"Orchestrates the dual-agent governance workflow\"\"\"\n  \n   def __init__(self, constitution: Constitution):\n       self.client = MockAIClient()\n       self.worker = WorkerAgent(self.client)\n       self.auditor = AuditorAgent(self.client, constitution)\n       self.constitution = constitution\n       self.max_revision_attempts = 3\n  \n   def process_with_governance(self, request: FinancialRequest) -> Dict[str, Any]:\n       \"\"\"Main workflow: Worker processes, Auditor validates, loop if needed\"\"\"\n       print(f\"n{'#'*60}\")\n       print(f\"# GOVERNANCE SYSTEM: New Request\")\n       print(f\"{'#'*60}\")\n      \n       attempt = 0\n       while attempt < self.max_revision_attempts:\n           attempt += 1\n           print(f\"n\"\ud83d\udd04\" Attempt {attempt}\/{self.max_revision_attempts}\")\n          \n           worker_output = self.worker.process_request(request)\n          \n           audit_result = self.auditor.audit(worker_output)\n          \n           if audit_result.approved:\n               print(f\"n{'='*60}\")\n               print(f\"\"\u2705\" FINAL RESULT: APPROVED\")\n               print(f\"{'='*60}\")\n               return {\n                   \"status\": \"approved\",\n                   \"output\": worker_output.dict(),\n                   \"audit\": audit_result.dict(),\n                   \"attempts\": attempt\n               }\n          \n           critical_violations = [v for v in audit_result.violations if v.severity == \"critical\"]\n           if critical_violations:\n               print(f\"n{'='*60}\")\n               print(f\"\"\ud83d\uded1\" FINAL RESULT: REJECTED (Critical Violations)\")\n               print(f\"{'='*60}\")\n               return {\n                   \"status\": \"rejected\",\n                   \"reason\": \"critical_violations\",\n                   \"audit\": audit_result.dict(),\n                   \"attempts\": attempt\n               }\n          \n           if attempt >= self.max_revision_attempts:\n               print(f\"n{'='*60}\")\n               print(f\"\"\ud83d\uded1\" FINAL RESULT: REJECTED (Max Attempts)\")\n               print(f\"{'='*60}\")\n               return {\n                   \"status\": \"rejected\",\n                   \"reason\": \"max_attempts_exceeded\",\n                   \"audit\": audit_result.dict(),\n                   \"attempts\": attempt\n               }\n      \n       return {\"status\": \"error\", \"message\": \"Unexpected exit from loop\"}<\/code><\/pre>\n<\/div>\n<\/div>\n
We orchestrate the complete governance workflow by coordinating the worker and auditor agents within a controlled revision loop. We evaluate each attempt against constitutional rules and immediately halt execution when critical violations are detected. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def run_examples():\n   \"\"\"Run demonstration examples\"\"\"\n  \n   print(\"=\"*80)\n   print(\" DUAL-AGENT GOVERNANCE SYSTEM WITH CONSTITUTIONAL AI\")\n   print(\" Tutorial: Self-Reflective Financial Operations Agents\")\n   print(\"=\"*80)\n  \n   system = GovernanceSystem(FINANCIAL_CONSTITUTION)\n  \n   print(\"nn\" + \"=\"*80)\n   print(\"EXAMPLE 1: Safe Transaction ($2,500)\")\n   print(\"=\"*80)\n  \n   request1 = FinancialRequest(\n       action=\"payment\",\n       amount=2500.00,\n       recipient=\"Vendor Corp\",\n       description=\"Monthly software license payment\",\n       justification=\"Regular recurring payment for essential services\"\n   )\n  \n   result1 = system.process_with_governance(request1)\n  \n   print(\"nn\" + \"=\"*80)\n   print(\"EXAMPLE 2: High-Value Transaction with PII Leak ($7,500)\")\n   print(\"=\"*80)\n  \n   request2 = FinancialRequest(\n       action=\"transfer\",\n       amount=7500.00,\n       recipient=\"Executive\",\n       description=\"Bonus payment to executive\",\n       justification=\"Q4 performance bonus\"\n   )\n  \n   result2 = system.process_with_governance(request2)\n  \n   print(\"nn\" + \"=\"*80)\n   print(\"EXAMPLE 3: Budget-Exceeding Transaction ($15,000)\")\n   print(\"=\"*80)\n  \n   request3 = FinancialRequest(\n       action=\"transfer\",\n       amount=15000.00,\n       recipient=\"Supplier\",\n       description=\"Large equipment purchase\",\n       justification=\"New manufacturing equipment for production line\"\n   )\n  \n   result3 = system.process_with_governance(request3)\n  \n   print(\"nn\" + \"=\"*80)\n   print(\" SUMMARY OF RESULTS\")\n   print(\"=\"*80)\n   print(f\"nExample 1: {result1['status'].upper()}\")\n   print(f\"Example 2: {result2['status'].upper()} - {result2.get('reason', 'N\/A')}\")\n   print(f\"Example 3: {result3['status'].upper()} - {result3.get('reason', 'N\/A')}\")\n  \n   print(f\"nnTotal API Calls: {system.client.call_count}\")\n   print(f\"Worker Processed: {len(system.worker.processed_requests)} requests\")\n   print(f\"Auditor Performed: {len(system.auditor.audit_history)} audits\")\n  \n   print(\"nn\" + \"=\"*80)\n   print(\" ACTIVE CONSTITUTION\")\n   print(\"=\"*80)\n   for policy in FINANCIAL_CONSTITUTION.policies:\n       print(f\"n\"\ud83d\udcdc\" {policy.name} [{policy.severity.upper()}]\")\n       print(f\"   {policy.description}\")<\/code><\/pre>\n<\/div>\n<\/div>\n
We demonstrate the system end-to-end by running realistic financial scenarios that exercise both safe and unsafe behaviors. We show how the governance loop responds differently to compliant transactions, PII leaks, and budget violations while producing transparent audit outcomes. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
if __name__ == \"__main__\":\n   run_examples()\n  \n   print(\"nn\" + \"=\"*80)\n   print(\" \"\ud83c\udf93\" TUTORIAL COMPLETE!\")\n   print(\"=\"*80)\n   print(\"nKey Concepts Demonstrated:\")\n   print(\"\u2713 Constitutional AI - Rule-based governance\")\n   print(\"\u2713 Dual-Agent System - Worker + Auditor pattern\")\n   print(\"\u2713 Policy Violation Detection - PII, Budget, Authorization\")\n   print(\"\u2713 Iterative Revision Loop - Self-correction mechanism\")\n   print(\"\u2713 Risk Scoring - Quantitative safety assessment\")\n   print(\"nNext Steps:\")\n   print(\"\u2022 Replace MockAIClient with real Anthropic API\")\n   print(\"\u2022 Implement actual revision logic in Worker Agent\")\n   print(\"\u2022 Add more sophisticated pattern detection\")\n   print(\"\u2022 Integrate with real financial systems\")\n   print(\"\u2022 Build logging and monitoring dashboard\")\n   print(\"=\"*80)<\/code><\/pre>\n<\/div>\n<\/div>\n
We conclude the tutorial by executing all examples and clearly surfacing the core concepts demonstrated by the system. We recap how constitutional rules, dual-agent governance, violation detection, and risk scoring work together in practice.<\/p>\n
In conclusion, we demonstrated how to operationalize Constitutional AI beyond theory and embed it into real-world financial decision-making pipelines. We illustrated how we detect and respond to PII leakage, budget overruns, and missing justifications while quantifying risk and enforcing hard governance boundaries. By orchestrating iterative review loops between worker and auditor agents, we demonstrated a practical blueprint for building trustworthy, compliant, and scalable AI-driven financial systems where safety and accountability are first-class design goals rather than afterthoughts.<\/p>\n
\n
Check out the\u00a0FULL CODES here<\/a><\/strong>.\u00a0Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0100k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
The post How to Design Self-Reflective Dual-Agent Governance Systems with Constitutional AI for Secure and Compliant Financial Operations<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we implement…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-330","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/330","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=330"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/330\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=330"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=330"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=330"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}