{ "cells": [ { "cell_type": "markdown", "id": "intro", "metadata": {}, "source": [ "# AML Analysis with the Agents SDK on Amazon Bedrock\n", "\n", "This notebook demonstrates one synthetic anti-money-laundering (AML) analysis with the OpenAI Agents SDK and GPT-5.6 Sol on Amazon Bedrock. The default path uses a deterministic offline assessment so a top-to-bottom run makes no model call. An explicit opt-in runs the agent with read-only tools. Both paths validate material claims against deterministic calculations.\n", "\n", "An alert starts an investigation. It does not prove wrongdoing. The model proposes an assessment. Application code verifies the structured claims and controls workflow state. A qualified human decides whether optional drafting may begin. All data is synthetic. Nothing here files a Suspicious Activity Report (SAR) or implements production compliance policy.\n" ] }, { "cell_type": "markdown", "id": "workflow-boundaries", "metadata": {}, "source": [ "## Workflow boundaries\n", "\n", "The notebook keeps deterministic evidence, model proposals, application controls, and human authority in separate boundaries.\n", "\n", "![Four boundaries for the regulated-investigation workflow](../../../images/partners/AWS/evidence-grounded-four-boundaries.png)\n" ] }, { "cell_type": "markdown", "id": "start", "metadata": {}, "source": [ "## 1. Start the notebook\n", "\n", "The default offline path requires Python 3.10 or newer and `uv`. The optional paid Bedrock paths also require AWS credentials through the standard credential chain and access to `openai.gpt-5.6-sol`. From a clone of [openai/openai-cookbook](https://github.com/openai/openai-cookbook), run:\n", "\n", "```bash\n", "cd /path/to/openai-cookbook\n", "unset VIRTUAL_ENV\n", "export AWS_REGION=us-east-2\n", "uv run --with jupyterlab jupyter lab\n", "```\n", "\n", "Open this notebook at `examples/partners/AWS/evidence_grounded_aml_agent_with_bedrock.ipynb`. The default run keeps both paid flags disabled. To run the analysis agent, set `RUN_ANALYSIS_DEMO=true` before starting Jupyter. To run the optional drafting agent, set `RUN_DRAFTING_DEMO=true`. Set `AWS_PROFILE` first if your organization uses a named profile. Never paste AWS credentials into the notebook.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "install", "metadata": {}, "outputs": [], "source": [ "%pip install -U \"openai[bedrock]>=2.46.0\" \"openai-agents>=0.18.2\" \"pydantic>=2.13.0\" --quiet\n" ] }, { "cell_type": "markdown", "id": "configure", "metadata": {}, "source": [ "## 2. Configure Amazon Bedrock\n", "\n", "Both paid demonstrations require an explicit environment flag. When either flag is enabled, the SDK uses the AWS credential chain and a separate client lists models through the Bedrock Mantle discovery endpoint before paid inference. With both flags disabled, this cell constructs no client and makes no AWS call.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "bedrock-client", "metadata": {}, "outputs": [], "source": [ "import json\n", "import math\n", "import os\n", "from datetime import datetime\n", "from typing import Literal\n", "\n", "from agents import (\n", " Agent,\n", " ModelSettings,\n", " RunConfig,\n", " RunContextWrapper,\n", " Runner,\n", " function_tool,\n", " set_default_openai_api,\n", " set_default_openai_client,\n", ")\n", "from agents.items import ToolCallItem, ToolCallOutputItem\n", "from openai import AsyncOpenAI\n", "from openai.providers import bedrock\n", "from openai.types.shared import Reasoning\n", "from pydantic import BaseModel, Field\n", "\n", "AWS_REGION = os.getenv(\"AWS_REGION\", \"us-east-2\")\n", "MODEL_ID = os.getenv(\"BEDROCK_MODEL\", \"openai.gpt-5.6-sol\")\n", "RUN_ANALYSIS_DEMO = (\n", " os.getenv(\"RUN_ANALYSIS_DEMO\", \"false\").casefold() == \"true\"\n", ")\n", "RUN_DRAFTING_DEMO = (\n", " os.getenv(\"RUN_DRAFTING_DEMO\", \"false\").casefold() == \"true\"\n", ")\n", "RUN_BEDROCK_DEMOS = RUN_ANALYSIS_DEMO or RUN_DRAFTING_DEMO\n", "\n", "set_default_openai_api(\"responses\")\n", "\n", "if RUN_BEDROCK_DEMOS:\n", " client = AsyncOpenAI(provider=bedrock(region=AWS_REGION))\n", " models_client = AsyncOpenAI(\n", " provider=bedrock(\n", " region=AWS_REGION,\n", " base_url=(\n", " f\"https://bedrock-mantle.{AWS_REGION}.api.aws/v1\"\n", " ),\n", " )\n", " )\n", " set_default_openai_client(client, use_for_tracing=False)\n", " available_models = await models_client.models.list()\n", " available_model_ids = {model.id for model in available_models.data}\n", " if MODEL_ID not in available_model_ids:\n", " raise RuntimeError(\n", " f\"{MODEL_ID!r} is not visible in {AWS_REGION}. \"\n", " \"Verify the AWS account, Region, and Bedrock model access.\"\n", " )\n", " print({\"region\": AWS_REGION, \"model\": MODEL_ID, \"preflight\": \"passed\"})\n", "else:\n", " print(\"Paid Bedrock demonstrations disabled; no AWS call was made.\")\n" ] }, { "cell_type": "markdown", "id": "case-intro", "metadata": {}, "source": [ "## 3. Load the synthetic case\n", "\n", "The case has three same-day cash credits followed by an outbound wire. A factory creates a fresh typed case for each application context. The amounts make the calculations easy to reproduce; they are demonstration rules, not regulatory thresholds.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "synthetic-case", "metadata": {}, "outputs": [], "source": [ "DEMO_CASH_AMOUNT_MIN = 9_000\n", "DEMO_CASH_AMOUNT_MAX = 10_000\n", "DEMO_RAPID_MOVEMENT_RATIO = 0.90\n", "DEMO_REQUIRED_CASH_CREDITS = 3\n", "\n", "class Transaction(BaseModel):\n", " id: str\n", " timestamp: str\n", " direction: Literal[\"CREDIT\", \"DEBIT\"]\n", " channel: Literal[\"CASH\", \"WIRE\"]\n", " amount: int = Field(gt=0)\n", " currency: str\n", " counterparty: str\n", " country_code: str\n", "\n", "\n", "class InvestigationCase(BaseModel):\n", " case_id: str\n", " subject: str\n", " subject_type: Literal[\"BUSINESS\"]\n", " stated_business: str\n", " risk_tier: Literal[\"STANDARD\"]\n", " alert_reason: str\n", " transactions: list[Transaction]\n", "\n", "\n", "def build_synthetic_case() -> InvestigationCase:\n", " return InvestigationCase(\n", " case_id=\"SYNTH-AML-001\",\n", " subject=\"Northstar Imports LLC\",\n", " subject_type=\"BUSINESS\",\n", " stated_business=\"Wholesale home goods\",\n", " risk_tier=\"STANDARD\",\n", " alert_reason=(\n", " \"Unusual cash activity followed by an outbound wire\"\n", " ),\n", " transactions=[\n", " Transaction(\n", " id=\"TXN-001\",\n", " timestamp=\"2026-05-04T09:20:00Z\",\n", " direction=\"CREDIT\",\n", " channel=\"CASH\",\n", " amount=9200,\n", " currency=\"USD\",\n", " counterparty=\"Synthetic cash deposit A\",\n", " country_code=\"US\",\n", " ),\n", " Transaction(\n", " id=\"TXN-002\",\n", " timestamp=\"2026-05-04T11:05:00Z\",\n", " direction=\"CREDIT\",\n", " channel=\"CASH\",\n", " amount=9500,\n", " currency=\"USD\",\n", " counterparty=\"Synthetic cash deposit B\",\n", " country_code=\"US\",\n", " ),\n", " Transaction(\n", " id=\"TXN-003\",\n", " timestamp=\"2026-05-04T13:40:00Z\",\n", " direction=\"CREDIT\",\n", " channel=\"CASH\",\n", " amount=9800,\n", " currency=\"USD\",\n", " counterparty=\"Synthetic cash deposit C\",\n", " country_code=\"US\",\n", " ),\n", " Transaction(\n", " id=\"TXN-004\",\n", " timestamp=\"2026-05-04T16:10:00Z\",\n", " direction=\"DEBIT\",\n", " channel=\"WIRE\",\n", " amount=28200,\n", " currency=\"USD\",\n", " counterparty=\"Synthetic overseas supplier\",\n", " country_code=\"GB\",\n", " ),\n", " ],\n", " )\n", "\n", "\n", "print(build_synthetic_case().model_dump_json(indent=2))\n" ] }, { "cell_type": "markdown", "id": "contract-intro", "metadata": {}, "source": [ "## 4. Define the output and evidence tools\n", "\n", "The output schema makes the material claims machine-checkable. Each finding carries its evidence IDs and computed values. The validator recomputes those values because a valid transaction ID alone does not prove support.\n", "\n", "A typed `InvestigationContext` scopes one tenant, case, analysis, and review history. Each request creates its own context instance, passes it through `Runner.run(..., context=...)`, and lets tools access it through `RunContextWrapper`. The context is not itself model input; each read-only tool controls which fields it returns.\n", "\n", "Two tools return source facts, and one applies transparent demo checks. The checks identify signals for investigation, not intent, wrongdoing, or a filing requirement.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "output-contract", "metadata": {}, "outputs": [], "source": [ "class Finding(BaseModel):\n", " finding_type: Literal[\"STRUCTURING_SIGNAL\", \"RAPID_MOVEMENT_SIGNAL\"]\n", " title: str\n", " explanation: str\n", " evidence_ids: list[str] = Field(min_length=1)\n", " cash_credit_count: int = Field(ge=0)\n", " cash_credit_total_usd: int = Field(ge=0)\n", " outbound_wire_usd: int | None\n", " movement_ratio_percent: float | None\n", "\n", "\n", "class RiskAssessment(BaseModel):\n", " case_id: str\n", " assessment_posture: Literal[\n", " \"ROUTINE_REVIEW\",\n", " \"ENHANCED_REVIEW\",\n", " \"ESCALATE_FOR_QUALIFIED_REVIEW\",\n", " ]\n", " executive_summary: str\n", " findings: list[Finding] = Field(min_length=2, max_length=2)\n", " information_gaps: list[str] = Field(min_length=1)\n", " recommended_next_steps: list[str] = Field(min_length=1)\n", " drafting_authorized: Literal[False]\n", " filing_decision: Literal[\"NOT_DETERMINED\"]\n", "\n", "\n", "class EvidenceCitation(BaseModel):\n", " claim: str\n", " evidence_ids: list[str] = Field(min_length=1)\n", "\n", "\n", "class SarDraft(BaseModel):\n", " case_id: str\n", " narrative: str\n", " draft_status: Literal[\n", " \"DRAFT_READY_FOR_HUMAN_REVIEW\", \"INSUFFICIENT_INFORMATION\"\n", " ]\n", " citations: list[EvidenceCitation] = Field(min_length=1)\n", " disclaimer: str\n", "\n", "\n", "ReviewDecision = Literal[\"APPROVE_DRAFTING\", \"REJECT_DRAFTING\"]\n", "ReviewStatus = Literal[\n", " \"PENDING_REVIEW\",\n", " \"PENDING_REREVIEW\",\n", " \"APPROVE_DRAFTING\",\n", " \"REJECT_DRAFTING\",\n", "]\n", "\n", "\n", "class HumanReviewEvent(BaseModel):\n", " event: Literal[\"HUMAN_REVIEW\"] = \"HUMAN_REVIEW\"\n", " analysis_revision: int = Field(ge=1)\n", " reviewer_alias: str\n", " decision: ReviewDecision\n", " rationale: str\n", "\n", "\n", "class AnalysisRevisedEvent(BaseModel):\n", " event: Literal[\"ANALYSIS_REVISED\"] = \"ANALYSIS_REVISED\"\n", " from_revision: int = Field(ge=1)\n", " to_revision: int = Field(ge=2)\n", " rework_note: str\n", "\n", "\n", "ReviewEvent = HumanReviewEvent | AnalysisRevisedEvent\n", "\n", "\n", "class InvestigationContext(BaseModel):\n", " tenant_id: str\n", " case: InvestigationCase\n", " analysis: RiskAssessment | None = None\n", " analysis_revision: int = Field(default=0, ge=0)\n", " review_status: ReviewStatus = \"PENDING_REVIEW\"\n", " review_history: list[ReviewEvent] = Field(default_factory=list)\n", " draft: SarDraft | None = None\n", "\n", "\n", "def new_investigation_context() -> InvestigationContext:\n", " return InvestigationContext(\n", " tenant_id=\"SYNTHETIC-TENANT-001\",\n", " case=build_synthetic_case(),\n", " )\n", "\n", "\n", "run_context = new_investigation_context()\n" ] }, { "cell_type": "code", "execution_count": null, "id": "evidence-tools", "metadata": {}, "outputs": [], "source": [ "def require_known_case(\n", " wrapper: RunContextWrapper[InvestigationContext],\n", " case_id: str,\n", ") -> InvestigationCase:\n", " case = wrapper.context.case\n", " if case_id != case.case_id:\n", " raise ValueError(f\"Unknown case in this request context: {case_id}\")\n", " return case\n", "\n", "\n", "@function_tool(failure_error_function=None)\n", "def get_case_profile(\n", " wrapper: RunContextWrapper[InvestigationContext],\n", " case_id: str,\n", ") -> str:\n", " \"\"\"Return the synthetic profile and alert context for one case.\"\"\"\n", "\n", " case = require_known_case(wrapper, case_id)\n", " profile = case.model_dump(exclude={\"transactions\"})\n", " return json.dumps(profile)\n", "\n", "\n", "@function_tool(failure_error_function=None)\n", "def list_case_transactions(\n", " wrapper: RunContextWrapper[InvestigationContext],\n", " case_id: str,\n", ") -> str:\n", " \"\"\"Return all synthetic transactions and their evidence identifiers.\"\"\"\n", "\n", " case = require_known_case(wrapper, case_id)\n", " return json.dumps(\n", " [transaction.model_dump() for transaction in case.transactions]\n", " )\n", "\n", "\n", "def parse_transaction_timestamp(value: str) -> datetime:\n", " return datetime.fromisoformat(value.replace(\"Z\", \"+00:00\"))\n", "\n", "\n", "def detect_typology_signals(\n", " case: InvestigationCase,\n", ") -> list[dict[str, object]]:\n", " transactions = case.transactions\n", " cash_credits = [\n", " transaction\n", " for transaction in transactions\n", " if transaction.direction == \"CREDIT\"\n", " and transaction.channel == \"CASH\"\n", " and DEMO_CASH_AMOUNT_MIN\n", " <= transaction.amount\n", " < DEMO_CASH_AMOUNT_MAX\n", " ]\n", " if len(cash_credits) < DEMO_REQUIRED_CASH_CREDITS:\n", " return []\n", "\n", " activity_dates = {\n", " parse_transaction_timestamp(item.timestamp).date()\n", " for item in cash_credits\n", " }\n", " currencies = {item.currency for item in cash_credits}\n", " if len(activity_dates) != 1 or len(currencies) != 1:\n", " return []\n", "\n", " cash_total = sum(item.amount for item in cash_credits)\n", " latest_cash_timestamp = max(\n", " parse_transaction_timestamp(item.timestamp)\n", " for item in cash_credits\n", " )\n", " rapid_wires = [\n", " transaction\n", " for transaction in transactions\n", " if transaction.direction == \"DEBIT\"\n", " and transaction.channel == \"WIRE\"\n", " and transaction.currency in currencies\n", " and parse_transaction_timestamp(transaction.timestamp).date()\n", " in activity_dates\n", " and parse_transaction_timestamp(transaction.timestamp)\n", " > latest_cash_timestamp\n", " and transaction.amount\n", " >= cash_total * DEMO_RAPID_MOVEMENT_RATIO\n", " ]\n", "\n", " cash_ids = [item.id for item in cash_credits]\n", " signals = [\n", " {\n", " \"finding_type\": \"STRUCTURING_SIGNAL\",\n", " \"evidence_ids\": cash_ids,\n", " \"cash_credit_count\": len(cash_credits),\n", " \"cash_credit_total_usd\": cash_total,\n", " \"outbound_wire_usd\": None,\n", " \"movement_ratio_percent\": None,\n", " }\n", " ]\n", " if rapid_wires:\n", " wire = rapid_wires[0]\n", " signals.append(\n", " {\n", " \"finding_type\": \"RAPID_MOVEMENT_SIGNAL\",\n", " \"evidence_ids\": [*cash_ids, wire.id],\n", " \"cash_credit_count\": len(cash_credits),\n", " \"cash_credit_total_usd\": cash_total,\n", " \"outbound_wire_usd\": wire.amount,\n", " \"movement_ratio_percent\": round(\n", " wire.amount / cash_total * 100, 1\n", " ),\n", " }\n", " )\n", " return signals\n", "\n", "\n", "@function_tool(failure_error_function=None)\n", "def run_typology_checks(\n", " wrapper: RunContextWrapper[InvestigationContext],\n", " case_id: str,\n", ") -> str:\n", " \"\"\"Return transparent demo signals and their calculated support.\"\"\"\n", "\n", " case = require_known_case(wrapper, case_id)\n", " return json.dumps(detect_typology_signals(case))\n", "\n", "\n", "print(json.dumps(detect_typology_signals(run_context.case), indent=2))\n" ] }, { "cell_type": "markdown", "id": "run-intro", "metadata": {}, "source": [ "## 5. Run the analysis agent\n", "\n", "The instructions require all three tools and a typed result. With `RUN_ANALYSIS_DEMO=true`, `Runner.run` manages the paid model and tool loop. The default path uses a labeled deterministic fixture with simulated tool-call evidence so the remaining validation and review cells run without AWS credentials or paid inference. The fixture does not prove Agents SDK orchestration.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "analysis-run", "metadata": {}, "outputs": [], "source": [ "MODEL_SETTINGS = ModelSettings(\n", " reasoning=Reasoning(effort=\"medium\"),\n", " store=False,\n", ")\n", "\n", "analysis_agent = Agent[InvestigationContext](\n", " name=\"Synthetic AML Analysis Agent\",\n", " model=MODEL_ID,\n", " model_settings=MODEL_SETTINGS,\n", " output_type=RiskAssessment,\n", " tools=[\n", " get_case_profile,\n", " list_case_transactions,\n", " run_typology_checks,\n", " ],\n", " instructions=(\n", " \"Analyze exactly one synthetic AML case. Call get_case_profile, \"\n", " \"list_case_transactions, and run_typology_checks. Copy the \"\n", " \"deterministic tool's evidence IDs and computed values into the \"\n", " \"matching structured finding fields. Do not introduce other amounts \"\n", " \"in the finding explanations. Treat signals as prompts for qualified \"\n", " \"review, not proof of intent or wrongdoing. Identify missing \"\n", " \"information. Set drafting_authorized to false and filing_decision \"\n", " \"to NOT_DETERMINED. Do not draft or file a SAR, change case state, \"\n", " \"approve your own work, or claim that filing is required. Return \"\n", " \"only the RiskAssessment schema.\"\n", " ),\n", ")\n", "\n", "REQUIRED_ANALYSIS_TOOLS = {\n", " \"get_case_profile\",\n", " \"list_case_transactions\",\n", " \"run_typology_checks\",\n", "}\n", "\n", "FINDING_TEXT = {\n", " \"STRUCTURING_SIGNAL\": (\n", " \"Cash-credit pattern\",\n", " \"The deterministic check identified same-day cash credits.\",\n", " ),\n", " \"RAPID_MOVEMENT_SIGNAL\": (\n", " \"Rapid movement pattern\",\n", " \"The deterministic check identified a later outbound wire.\",\n", " ),\n", "}\n", "\n", "\n", "def build_offline_assessment(\n", " context: InvestigationContext,\n", ") -> RiskAssessment:\n", " findings = []\n", " for signal in detect_typology_signals(context.case):\n", " title, explanation = FINDING_TEXT[signal[\"finding_type\"]]\n", " findings.append(\n", " Finding(\n", " finding_type=signal[\"finding_type\"],\n", " title=title,\n", " explanation=explanation,\n", " evidence_ids=signal[\"evidence_ids\"],\n", " cash_credit_count=signal[\"cash_credit_count\"],\n", " cash_credit_total_usd=signal[\"cash_credit_total_usd\"],\n", " outbound_wire_usd=signal[\"outbound_wire_usd\"],\n", " movement_ratio_percent=signal[\"movement_ratio_percent\"],\n", " )\n", " )\n", " return RiskAssessment(\n", " case_id=context.case.case_id,\n", " assessment_posture=\"ESCALATE_FOR_QUALIFIED_REVIEW\",\n", " executive_summary=(\n", " \"Synthetic signals require qualified review.\"\n", " ),\n", " findings=findings,\n", " information_gaps=[\n", " \"Source-of-funds records remain unverified.\"\n", " ],\n", " recommended_next_steps=[\n", " \"Obtain source records for qualified human review.\"\n", " ],\n", " drafting_authorized=False,\n", " filing_decision=\"NOT_DETERMINED\",\n", " )\n", "\n", "\n", "if RUN_ANALYSIS_DEMO:\n", " result = await Runner.run(\n", " analysis_agent,\n", " f\"Analyze synthetic case {run_context.case.case_id}.\",\n", " context=run_context,\n", " max_turns=8,\n", " run_config=RunConfig(\n", " tracing_disabled=True,\n", " workflow_name=\"Validated synthetic AML analysis\",\n", " ),\n", " )\n", " assessment = result.final_output\n", " tool_calls = {\n", " item.raw_item.name\n", " for item in result.new_items\n", " if isinstance(item, ToolCallItem)\n", " }\n", " analysis_source = \"paid Agents SDK run\"\n", "else:\n", " assessment = build_offline_assessment(run_context)\n", " tool_calls = set(REQUIRED_ANALYSIS_TOOLS)\n", " analysis_source = \"offline fixture with simulated tool calls\"\n", "\n", "print(\"Analysis source:\", analysis_source)\n", "print(\"Tool-call evidence:\", sorted(tool_calls))\n", "print(assessment.model_dump_json(indent=2))\n" ] }, { "cell_type": "markdown", "id": "validation-intro", "metadata": {}, "source": [ "## 6. Validate support, not just citation syntax\n", "\n", "The validator performs three distinct checks:\n", "\n", "1. **Citation validity:** every evidence ID exists in this case.\n", "2. **Claim support:** each finding type uses the exact transactions and computed values returned by deterministic application code.\n", "3. **Structured authority:** `drafting_authorized` must remain `false`, and `filing_decision` must remain `NOT_DETERMINED`.\n", "\n", "The schema prevents the model from granting drafting authority or making a filing decision through those structured fields. The validator does not classify every possible free-form paraphrase. A qualified reviewer still evaluates the narrative, information gaps, relevance, and disposition. Production systems should add expert-reviewed datasets, trace grading, adversarial cases, and policy-specific tests.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "support-validation", "metadata": {}, "outputs": [], "source": [ "def validate_risk_assessment(\n", " candidate: RiskAssessment,\n", " observed_tool_calls: set[str],\n", " context: InvestigationContext,\n", ") -> dict[str, bool]:\n", " expected = {\n", " signal[\"finding_type\"]: signal\n", " for signal in detect_typology_signals(context.case)\n", " }\n", " valid_evidence_ids = {\n", " transaction.id for transaction in context.case.transactions\n", " }\n", " observed = {\n", " finding.finding_type: finding for finding in candidate.findings\n", " }\n", " unique_finding_types = len(observed) == len(candidate.findings)\n", "\n", " valid_citations = all(\n", " finding.evidence_ids\n", " and set(finding.evidence_ids).issubset(valid_evidence_ids)\n", " for finding in candidate.findings\n", " )\n", " supported_claims = unique_finding_types and set(observed) == set(expected)\n", " if supported_claims:\n", " for finding_type, expected_support in expected.items():\n", " finding = observed[finding_type]\n", " supported_claims = supported_claims and (\n", " set(finding.evidence_ids)\n", " == set(expected_support[\"evidence_ids\"])\n", " and finding.cash_credit_count\n", " == expected_support[\"cash_credit_count\"]\n", " and finding.cash_credit_total_usd\n", " == expected_support[\"cash_credit_total_usd\"]\n", " and finding.outbound_wire_usd\n", " == expected_support[\"outbound_wire_usd\"]\n", " and (\n", " finding.movement_ratio_percent is None\n", " and expected_support[\"movement_ratio_percent\"] is None\n", " or finding.movement_ratio_percent is not None\n", " and expected_support[\"movement_ratio_percent\"] is not None\n", " and math.isclose(\n", " finding.movement_ratio_percent,\n", " expected_support[\"movement_ratio_percent\"],\n", " abs_tol=0.1,\n", " )\n", " )\n", " )\n", "\n", " checks = {\n", " \"case identity\": candidate.case_id == context.case.case_id,\n", " \"required tools\": REQUIRED_ANALYSIS_TOOLS.issubset(\n", " observed_tool_calls\n", " ),\n", " \"valid citations\": valid_citations,\n", " \"material claims supported\": supported_claims,\n", " \"structured authority denied\": (\n", " candidate.drafting_authorized is False\n", " and candidate.filing_decision == \"NOT_DETERMINED\"\n", " ),\n", " }\n", " failed = [name for name, passed in checks.items() if not passed]\n", " if failed:\n", " raise ValueError(\"Assessment validation failed: \" + \", \".join(failed))\n", " return checks\n", "\n", "\n", "checks = validate_risk_assessment(assessment, tool_calls, run_context)\n", "for check_name in checks:\n", " print(\"PASS:\", check_name)\n" ] }, { "cell_type": "markdown", "id": "negative-intro", "metadata": {}, "source": [ "## 7. Exercise negative cases\n", "\n", "These local tests demonstrate failure behavior without additional model calls. A fabricated amount can use real transaction IDs and still be unsupported. Invalid IDs, a missing tool call, a structured authority claim, and a rejected review are separate failure modes.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "negative-tests", "metadata": {}, "outputs": [], "source": [ "def expect_validation_failure(\n", " label: str,\n", " candidate: RiskAssessment,\n", " observed_tools: set[str],\n", " context: InvestigationContext,\n", ") -> None:\n", " try:\n", " validate_risk_assessment(candidate, observed_tools, context)\n", " except ValueError as exc:\n", " print(f\"PASS ({label}):\", exc)\n", " else:\n", " raise AssertionError(f\"Expected validation failure: {label}\")\n", "\n", "\n", "fabricated_claim = assessment.model_copy(deep=True)\n", "fabricated_claim.findings[0].cash_credit_total_usd = 50_000\n", "expect_validation_failure(\n", " \"fabricated amount\", fabricated_claim, tool_calls, run_context\n", ")\n", "\n", "invalid_citation = assessment.model_copy(deep=True)\n", "invalid_citation.findings[0].evidence_ids = [\"TXN-999\"]\n", "expect_validation_failure(\n", " \"invalid citation\", invalid_citation, tool_calls, run_context\n", ")\n", "\n", "missing_tool_calls = tool_calls - {\"run_typology_checks\"}\n", "expect_validation_failure(\n", " \"missing tool\", assessment, missing_tool_calls, run_context\n", ")\n", "\n", "invalid_authority = assessment.model_copy(deep=True)\n", "invalid_authority.drafting_authorized = True\n", "invalid_authority.filing_decision = \"FILE_REQUIRED\"\n", "expect_validation_failure(\n", " \"structured authority\", invalid_authority, tool_calls, run_context\n", ")\n" ] }, { "cell_type": "markdown", "id": "review-intro", "metadata": {}, "source": [ "## 8. Preserve rejection and require explicit re-review\n", "\n", "The application now owns the approval boundary. It first stores the validated assessment, records a rejection, and proves drafting remains blocked. A revised assessment is then submitted as a new revision and reviewed again. The original rejection stays in `review_history`; it is never reset or overwritten.\n", "\n", "The synthetic reviewer alias and typed context keep the example visible in one notebook. A concurrent service must create a separate context for every request and tenant. Production systems must derive reviewer identity from authentication, authorize the case and operation, persist revisions and decisions durably, and maintain an independently governed audit record.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "review-workflow", "metadata": {}, "outputs": [], "source": [ "run_context.analysis = assessment\n", "run_context.analysis_revision = 1\n", "run_context.review_status = \"PENDING_REVIEW\"\n", "run_context.review_history.clear()\n", "run_context.draft = None\n", "\n", "\n", "def record_human_review(\n", " context: InvestigationContext,\n", " reviewer_alias: str,\n", " decision: ReviewDecision,\n", " rationale: str,\n", ") -> HumanReviewEvent:\n", " if context.review_status not in {\n", " \"PENDING_REVIEW\",\n", " \"PENDING_REREVIEW\",\n", " }:\n", " raise RuntimeError(\"The current analysis revision is not awaiting review\")\n", " if len(rationale.strip()) < 12:\n", " raise ValueError(\"Review rationale is too short\")\n", "\n", " event = HumanReviewEvent(\n", " analysis_revision=context.analysis_revision,\n", " reviewer_alias=reviewer_alias,\n", " decision=decision,\n", " rationale=rationale,\n", " )\n", " context.review_history.append(event)\n", " context.review_status = decision\n", " return event\n", "\n", "\n", "def require_drafting_allowed(\n", " context: InvestigationContext,\n", ") -> RiskAssessment:\n", " analysis = context.analysis\n", " history = context.review_history\n", " latest = history[-1] if history else None\n", " if not isinstance(analysis, RiskAssessment):\n", " raise RuntimeError( # noqa: TRY004 - unmet workflow precondition\n", " \"A validated analysis is required before drafting\"\n", " )\n", " if (\n", " not isinstance(latest, HumanReviewEvent)\n", " or latest.decision != \"APPROVE_DRAFTING\"\n", " or latest.analysis_revision != context.analysis_revision\n", " ):\n", " raise RuntimeError(\n", " \"The current analysis revision lacks qualified-human approval\"\n", " )\n", " return analysis\n", "\n", "\n", "def submit_revised_analysis(\n", " context: InvestigationContext,\n", " candidate: RiskAssessment,\n", " observed_tools: set[str],\n", " rework_note: str,\n", ") -> None:\n", " if context.review_status != \"REJECT_DRAFTING\":\n", " raise RuntimeError(\"A rejected review is required before rework\")\n", " validate_risk_assessment(candidate, observed_tools, context)\n", " previous_revision = context.analysis_revision\n", " context.analysis = candidate\n", " context.analysis_revision = previous_revision + 1\n", " context.review_status = \"PENDING_REREVIEW\"\n", " context.review_history.append(\n", " AnalysisRevisedEvent(\n", " from_revision=previous_revision,\n", " to_revision=previous_revision + 1,\n", " rework_note=rework_note,\n", " )\n", " )\n", "\n", "\n", "record_human_review(\n", " run_context,\n", " reviewer_alias=\"synthetic-qualified-reviewer\",\n", " decision=\"REJECT_DRAFTING\",\n", " rationale=\"The information gaps require revision before drafting.\",\n", ")\n", "try:\n", " require_drafting_allowed(run_context)\n", "except RuntimeError as exc:\n", " print(\"PASS (rejected review blocks drafting):\", exc)\n", "else:\n", " raise AssertionError(\"A rejected review must block drafting\")\n", "\n", "revised_assessment = assessment.model_copy(deep=True)\n", "revised_assessment.information_gaps.append(\n", " \"The draft must state that source-of-funds records remain unverified.\"\n", ")\n", "submit_revised_analysis(\n", " run_context,\n", " revised_assessment,\n", " tool_calls,\n", " rework_note=\"Added the reviewer's unresolved source-of-funds limitation.\",\n", ")\n", "record_human_review(\n", " run_context,\n", " reviewer_alias=\"synthetic-qualified-reviewer\",\n", " decision=\"APPROVE_DRAFTING\",\n", " rationale=\"Re-reviewed revision 2 and approved draft preparation only.\",\n", ")\n", "require_drafting_allowed(run_context)\n", "print(\n", " json.dumps(\n", " [event.model_dump() for event in run_context.review_history],\n", " indent=2,\n", " )\n", ")\n" ] }, { "cell_type": "markdown", "id": "draft-intro", "metadata": {}, "source": [ "## 9. Define the drafting agent only after approval\n", "\n", "The notebook defines the drafting agent after the application gate. The agent can read the current approved analysis and source evidence. It has no tool for approval, state changes, or filing. Running it is optional and adds one paid inference call. Set `RUN_DRAFTING_DEMO=true` before starting Jupyter. Evidence-tool exceptions propagate instead of becoming model-visible error outputs. Before storing a draft, application code verifies from the raw run-item payloads that every required evidence call produced an output, then checks the current case ID, evidence IDs, and disclaimer. Local negative tests cover a missing tool output and a mismatched case without making a model call. A qualified reviewer must still confirm that each narrative claim has support.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "draft-agent", "metadata": {}, "outputs": [], "source": [ "@function_tool(failure_error_function=None)\n", "def get_reviewed_analysis(\n", " wrapper: RunContextWrapper[InvestigationContext],\n", " case_id: str,\n", ") -> str:\n", " \"\"\"Return the current analysis only after explicit human approval.\"\"\"\n", "\n", " require_known_case(wrapper, case_id)\n", " return require_drafting_allowed(wrapper.context).model_dump_json()\n", "\n", "\n", "sar_drafting_agent = Agent[InvestigationContext](\n", " name=\"Synthetic SAR Draft Preparation Agent\",\n", " model=MODEL_ID,\n", " model_settings=MODEL_SETTINGS,\n", " output_type=SarDraft,\n", " tools=[\n", " get_case_profile,\n", " list_case_transactions,\n", " get_reviewed_analysis,\n", " ],\n", " instructions=(\n", " \"Prepare a synthetic draft for qualified human review only. Call all \"\n", " \"three tools. Use neutral chronological language and cite supplied \"\n", " \"transaction IDs and copy the supplied case ID exactly. Never submit \"\n", " \"or file anything. The disclaimer must \"\n", " \"say this is an AI-generated draft requiring qualified human review. \"\n", " \"Return only SarDraft.\"\n", " ),\n", ")\n", "\n", "REQUIRED_DRAFTING_TOOLS = {\n", " \"get_case_profile\",\n", " \"list_case_transactions\",\n", " \"get_reviewed_analysis\",\n", "}\n", "\n", "\n", "def raw_item_string(raw_item: object, field_name: str) -> str | None:\n", " if isinstance(raw_item, dict):\n", " value = raw_item.get(field_name)\n", " else:\n", " value = getattr(raw_item, field_name, None)\n", " return value if isinstance(value, str) else None\n", "\n", "\n", "def raw_call_id(raw_item: object) -> str | None:\n", " return raw_item_string(raw_item, \"call_id\") or raw_item_string(\n", " raw_item, \"id\"\n", " )\n", "\n", "\n", "def completed_function_tools(items: list[object]) -> set[str]:\n", " tool_names_by_call_id: dict[str, str] = {}\n", " completed_call_ids: set[str] = set()\n", " for item in items:\n", " if isinstance(item, ToolCallItem):\n", " call_id = raw_call_id(item.raw_item)\n", " tool_name = raw_item_string(item.raw_item, \"name\")\n", " if call_id is not None and tool_name is not None:\n", " tool_names_by_call_id[call_id] = tool_name\n", " elif isinstance(item, ToolCallOutputItem):\n", " call_id = raw_call_id(item.raw_item)\n", " if call_id is not None:\n", " completed_call_ids.add(call_id)\n", " return {\n", " tool_name\n", " for call_id, tool_name in tool_names_by_call_id.items()\n", " if call_id in completed_call_ids\n", " }\n", "\n", "\n", "simulated_run_items: list[object] = [\n", " ToolCallItem(\n", " agent=sar_drafting_agent,\n", " raw_item={\n", " \"type\": \"function_call\",\n", " \"call_id\": \"synthetic-call-001\",\n", " \"name\": \"get_reviewed_analysis\",\n", " \"arguments\": \"{}\",\n", " },\n", " ),\n", " ToolCallOutputItem(\n", " agent=sar_drafting_agent,\n", " raw_item={\n", " \"type\": \"function_call_output\",\n", " \"call_id\": \"synthetic-call-001\",\n", " \"output\": \"{}\",\n", " },\n", " output=\"{}\",\n", " ),\n", "]\n", "if completed_function_tools(simulated_run_items) != {\n", " \"get_reviewed_analysis\"\n", "}:\n", " raise AssertionError(\"Raw run-item correlation failed\")\n", "print(\"PASS (raw run-item correlation)\")\n", "\n", "\n", "def validate_sar_draft(\n", " candidate: SarDraft,\n", " context: InvestigationContext,\n", " completed_tool_names: set[str],\n", ") -> dict[str, bool]:\n", " draft_ids = {\n", " evidence_id\n", " for citation in candidate.citations\n", " for evidence_id in citation.evidence_ids\n", " }\n", " valid_evidence_ids = {\n", " transaction.id for transaction in context.case.transactions\n", " }\n", " checks = {\n", " \"required tools completed\": (\n", " REQUIRED_DRAFTING_TOOLS.issubset(completed_tool_names)\n", " ),\n", " \"case identity\": candidate.case_id == context.case.case_id,\n", " \"valid citations\": bool(draft_ids)\n", " and draft_ids.issubset(valid_evidence_ids),\n", " \"human-review disclaimer\": (\n", " \"qualified human review\" in candidate.disclaimer.casefold()\n", " ),\n", " }\n", " failed = [name for name, passed in checks.items() if not passed]\n", " if failed:\n", " raise ValueError(\"Draft validation failed: \" + \", \".join(failed))\n", " return checks\n", "\n", "\n", "wrong_case_draft = SarDraft(\n", " case_id=\"SYNTH-AML-OTHER\",\n", " narrative=\"Synthetic local test fixture.\",\n", " draft_status=\"INSUFFICIENT_INFORMATION\",\n", " citations=[\n", " EvidenceCitation(\n", " claim=\"A synthetic transaction exists.\",\n", " evidence_ids=[\"TXN-001\"],\n", " )\n", " ],\n", " disclaimer=\"AI-generated draft requiring qualified human review.\",\n", ")\n", "try:\n", " validate_sar_draft(\n", " wrong_case_draft, run_context, set(REQUIRED_DRAFTING_TOOLS)\n", " )\n", "except ValueError as exc:\n", " print(\"PASS (wrong-case draft rejected):\", exc)\n", "else:\n", " raise AssertionError(\"A wrong-case draft must be rejected\")\n", "\n", "missing_tool_draft = wrong_case_draft.model_copy(\n", " update={\"case_id\": run_context.case.case_id}\n", ")\n", "try:\n", " validate_sar_draft(\n", " missing_tool_draft,\n", " run_context,\n", " REQUIRED_DRAFTING_TOOLS - {\"get_reviewed_analysis\"},\n", " )\n", "except ValueError as exc:\n", " print(\"PASS (missing draft tool rejected):\", exc)\n", "else:\n", " raise AssertionError(\"A draft missing a required tool must be rejected\")\n", "\n", "\n", "if RUN_DRAFTING_DEMO:\n", " require_drafting_allowed(run_context)\n", " draft_run = await Runner.run(\n", " sar_drafting_agent,\n", " f\"Prepare a synthetic draft for {run_context.case.case_id}.\",\n", " context=run_context,\n", " max_turns=8,\n", " run_config=RunConfig(\n", " tracing_disabled=True,\n", " workflow_name=\"Synthetic SAR draft preparation\",\n", " ),\n", " )\n", " draft = draft_run.final_output\n", " completed_drafting_tools = completed_function_tools(\n", " draft_run.new_items\n", " )\n", " draft_checks = validate_sar_draft(\n", " draft, run_context, completed_drafting_tools\n", " )\n", " run_context.draft = draft\n", " for check_name in draft_checks:\n", " print(\"PASS:\", check_name)\n", " print(draft.model_dump_json(indent=2))\n", "else:\n", " print(\"Optional drafting run skipped; the approved gate remains testable.\")\n" ] }, { "cell_type": "markdown", "id": "takeaways", "metadata": {}, "source": [ "## 10. What this example establishes\n", "\n", "- With explicit opt-in, the Agents SDK runs a bounded model and tool loop through Amazon Bedrock.\n", "- The default offline fixture makes no AWS or model call and does not prove Agents SDK orchestration.\n", "- A typed context scopes tools and workflow state to one request and tenant.\n", "- Pydantic makes material findings structured, but correctness comes from deterministic support checks.\n", "- Valid IDs and supported claims are tested separately.\n", "- Rejected work remains rejected until a revised analysis receives an explicit re-review.\n", "- Structured fields keep drafting authority false and the filing decision undetermined.\n", "- Drafting is downstream of application-owned approval, and filing stays out of scope.\n", "\n", "This notebook is a learning asset, not a production AML system. Real deployments need institution-approved policy, identity and authorization, protected data handling, durable workflow and audit records, evals reviewed by domain experts, monitoring, incident response, and legal and compliance review. A separate companion document should cover AgentCore deployment and full AWS infrastructure.\n" ] }, { "cell_type": "markdown", "id": "references", "metadata": {}, "source": [ "## References\n", "\n", "- [OpenAI models in Amazon Bedrock](https://developers.openai.com/api/docs/guides/amazon-bedrock)\n", "- [OpenAI Agents SDK guide](https://developers.openai.com/api/docs/guides/agents)\n", "- [OpenAI Agents SDK for Python](https://github.com/openai/openai-agents-python)\n", "- [Evaluate agent workflows](https://developers.openai.com/api/docs/guides/agent-evals)\n", "- [GPT-5.6 Sol model documentation](https://developers.openai.com/api/docs/models/gpt-5.6-sol)\n", "- [Amazon Bedrock GPT-5.6 Sol model card](https://docs.aws.amazon.com/bedrock/latest/userguide/model-card-openai-gpt-56-sol.html)\n", "- [FinCEN Suspicious Activity Reports](https://www.fincen.gov/suspicious-activity-reports-sars)\n" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (ipykernel)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.13" } }, "nbformat": 4, "nbformat_minor": 5 }