使用Amazon Bedrock上的Agents SDK进行反洗钱分析
本教程演示如何利用OpenAI Agents SDK与Amazon Bedrock上的GPT-5.6 Sol构建一个受监管的反洗钱（AML）分析工作流。通过类型化上下文、确定性验证和人类审批门控，确保模型输出可审计且符合合规要求。教程提供离线确定性路径与可选付费路径，涵盖环境配置、代理定义、证据工具、验证器、负面测试及起草代理，并强调生产环境中的安全与治理实践。
技术背景
在金融合规领域，反洗钱（AML）分析需要处理大量交易数据，识别可疑模式，并生成报告。传统规则引擎难以应对复杂模式，而大语言模型（LLM）能理解上下文，但存在幻觉和不可靠风险。本教程旨在展示如何结合LLM的推理能力与确定性代码验证，构建一个既灵活又可审计的AML分析系统。通过Amazon Bedrock，开发者可以安全地访问GPT-5.6 Sol等模型，同时利用Agents SDK管理工具调用和状态，确保工作流符合监管要求。
核心原理
核心原理是将模型提案与确定性验证分离。模型负责生成结构化评估（RiskAssessment），但所有实质性声明（如交易金额、证据ID）必须通过确定性工具（run_typology_checks）计算并验证。工作流状态由应用程序控制，模型无法直接修改。人类审批通过显式事件记录，确保任何起草或归档操作都需经过授权。这种设计防止模型幻觉导致错误决策，同时保留LLM的灵活性。
架构分析
系统架构分为四个边界：确定性证据、模型提案、应用控制、人类权限。证据工具（get_case_profile, list_case_transactions, run_typology_checks）提供只读数据，模型通过Agents SDK调用这些工具，生成结构化输出。应用层使用Pydantic模型定义输出模式，并通过验证器检查引用有效性、主张支持和权限字段。人类审批通过记录HumanReviewEvent和AnalysisRevisedEvent实现，确保状态转换可追溯。可选起草代理仅在人类批准后运行，且无权限修改状态。
完整实战教程
本笔记本演示了使用OpenAI Agents SDK和Amazon Bedrock上的GPT-5.6 Sol进行的一次合成反洗钱（AML）分析。默认路径使用确定性离线评估，因此自上而下的运行不会调用模型。显式选择启用代理并使用只读工具。两种路径都通过确定性计算验证实质性声明。
警报启动调查。它并不证明不当行为。模型提出评估。应用代码验证结构化声明并控制工作流状态。合格人员决定是否开始可选起草。所有数据均为合成数据。此处不提交可疑活动报告（SAR），也不实施生产合规政策。
工作流边界
笔记本将确定性证据、模型提案、应用控制与人类权限分别置于不同的边界内。
1. 启动笔记本
默认的离线路径需要 Python 3.10 或更高版本以及 uv。可选的付费 Bedrock 路径还需要通过标准凭证链提供 AWS 凭证，并有权访问 openai.gpt-5.6-sol。从 openai/openai-cookbook 的克隆中，运行：
cd /path/to/openai-cookbook
unset VIRTUAL_ENV
export AWS_REGION=us-east-2
uv run --with jupyterlab jupyter lab
在 examples/partners/AWS/evidence_grounded_aml_agent_with_bedrock.ipynb 中打开此笔记本。默认运行会同时禁用两个付费标志。要运行分析代理，请在启动 Jupyter 之前设置 RUN_ANALYSIS_DEMO=true。要运行可选的起草代理，请设置 RUN_DRAFTING_DEMO=true。如果您的组织使用命名配置文件，请先设置 AWS_PROFILE。切勿将 AWS 凭证粘贴到笔记本中。
%pip install -U "openai[bedrock]>=2.46.0" "openai-agents>=0.18.2" "pydantic>=2.13.0" --quiet
2. 配置 Amazon Bedrock
两个付费演示都需要显式的环境标志。当任一标志启用时，SDK 会使用 AWS 凭证链，并在付费推理之前通过 Bedrock Mantle 发现端点列出模型。当两个标志都禁用时，此单元格不会构建客户端，也不会进行任何 AWS 调用。
import json
import math
import os
from datetime import datetime
from typing import Literal
from agents import (
Agent,
ModelSettings,
RunConfig,
RunContextWrapper,
Runner,
function_tool,
set_default_openai_api,
set_default_openai_client,
)
from agents.items import ToolCallItem, ToolCallOutputItem
from openai import AsyncOpenAI
from openai.providers import bedrock
from openai.types.shared import Reasoning
from pydantic import BaseModel, Field
AWS_REGION = os.getenv("AWS_REGION", "us-east-2")
MODEL_ID = os.getenv("BEDROCK_MODEL", "openai.gpt-5.6-sol")
RUN_ANALYSIS_DEMO = (
os.getenv("RUN_ANALYSIS_DEMO", "false").casefold() == "true"
)
RUN_DRAFTING_DEMO = (
os.getenv("RUN_DRAFTING_DEMO", "false").casefold() == "true"
)
RUN_BEDROCK_DEMOS = RUN_ANALYSIS_DEMO or RUN_DRAFTING_DEMO
set_default_openai_api("responses")
if RUN_BEDROCK_DEMOS:
client = AsyncOpenAI(provider=bedrock(region=AWS_REGION))
models_client = AsyncOpenAI(
provider=bedrock(
region=AWS_REGION,
base_url=(
f"https://bedrock-mantle.{AWS_REGION}.api.aws/v1"
),
)
)
set_default_openai_client(client, use_for_tracing=False)
available_models = await models_client.models.list()
available_model_ids = {model.id for model in available_models.data}
if MODEL_ID not in available_model_ids:
raise RuntimeError(
f"{MODEL_ID!r} is not visible in {AWS_REGION}. "
"Verify the AWS account, Region, and Bedrock model access."
)
print({"region": AWS_REGION, "model": MODEL_ID, "preflight": "passed"})
else:
print("Paid Bedrock demonstrations disabled; no AWS call was made.")
3. 加载合成案例
该案例包含三笔同日现金贷记，随后是一笔对外电汇。工厂会为每个应用上下文创建一个全新的类型化案例。这些金额使得计算易于复现；它们是演示规则，而非监管阈值。
DEMO_CASH_AMOUNT_MIN = 9_000
DEMO_CASH_AMOUNT_MAX = 10_000
DEMO_RAPID_MOVEMENT_RATIO = 0.90
DEMO_REQUIRED_CASH_CREDITS = 3
class Transaction(BaseModel):
id: str
timestamp: str
direction: Literal["CREDIT", "DEBIT"]
channel: Literal["CASH", "WIRE"]
amount: int = Field(gt=0)
currency: str
counterparty: str
country_code: str
class InvestigationCase(BaseModel):
case_id: str
subject: str
subject_type: Literal["BUSINESS"]
stated_business: str
risk_tier: Literal["STANDARD"]
alert_reason: str
transactions: list[Transaction]
def build_synthetic_case() -> InvestigationCase:
return InvestigationCase(
case_id="SYNTH-AML-001",
subject="Northstar Imports LLC",
subject_type="BUSINESS",
stated_business="Wholesale home goods",
risk_tier="STANDARD",
alert_reason=(
"Unusual cash activity followed by an outbound wire"
),
transactions=[
Transaction(
id="TXN-001",
timestamp="2026-05-04T09:20:00Z",
direction="CREDIT",
channel="CASH",
amount=9200,
currency="USD",
counterparty="Synthetic cash deposit A",
country_code="US",
),
Transaction(
id="TXN-002",
timestamp="2026-05-04T11:05:00Z",
direction="CREDIT",
channel="CASH",
amount=9500,
currency="USD",
counterparty="Synthetic cash deposit B",
country_code="US",
),
Transaction(
id="TXN-003",
timestamp="2026-05-04T13:40:00Z",
direction="CREDIT",
channel="CASH",
amount=9800,
currency="USD",
counterparty="Synthetic cash deposit C",
country_code="US",
),
Transaction(
id="TXN-004",
timestamp="2026-05-04T16:10:00Z",
direction="DEBIT",
channel="WIRE",
amount=28200,
currency="USD",
counterparty="Synthetic overseas supplier",
country_code="GB",
),
],
)
print(build_synthetic_case().model_dump_json(indent=2))
4. 定义输出与证据工具
输出模式使实质性声明可被机器检查。每条发现都带有其证据 ID 和计算值。验证器会重新计算这些值，因为仅凭有效的交易 ID 并不能证明其支持性。
一个类型化的 InvestigationContext 限定于一个租户、案例、分析和审查历史。每个请求都会创建自己的上下文实例，通过 Runner.run(..., context=...) 传递，并让工具通过 RunContextWrapper 访问它。上下文本身不是模型输入；每个只读工具控制它返回哪些字段。
两个工具返回源事实，一个工具应用透明的演示检查。这些检查识别需要调查的信号，而非意图、不当行为或申报要求。
class Finding(BaseModel):
finding_type: Literal["STRUCTURING_SIGNAL", "RAPID_MOVEMENT_SIGNAL"]
title: str
explanation: str
evidence_ids: list[str] = Field(min_length=1)
cash_credit_count: int = Field(ge=0)
cash_credit_total_usd: int = Field(ge=0)
outbound_wire_usd: int | None
movement_ratio_percent: float | None
class RiskAssessment(BaseModel):
case_id: str
assessment_posture: Literal[
"ROUTINE_REVIEW",
"ENHANCED_REVIEW",
"ESCALATE_FOR_QUALIFIED_REVIEW",
]
executive_summary: str
findings: list[Finding] = Field(min_length=2, max_length=2)
information_gaps: list[str] = Field(min_length=1)
recommended_next_steps: list[str] = Field(min_length=1)
drafting_authorized: Literal[False]
filing_decision: Literal["NOT_DETERMINED"]
class EvidenceCitation(BaseModel):
claim: str
evidence_ids: list[str] = Field(min_length=1)
class SarDraft(BaseModel):
case_id: str
narrative: str
draft_status: Literal[
"DRAFT_READY_FOR_HUMAN_REVIEW", "INSUFFICIENT_INFORMATION"
]
citations: list[EvidenceCitation] = Field(min_length=1)
disclaimer: str
ReviewDecision = Literal["APPROVE_DRAFTING", "REJECT_DRAFTING"]
ReviewStatus = Literal[
"PENDING_REVIEW",
"PENDING_REREVIEW",
"APPROVE_DRAFTING",
"REJECT_DRAFTING",
]
class HumanReviewEvent(BaseModel):
event: Literal["HUMAN_REVIEW"] = "HUMAN_REVIEW"
analysis_revision: int = Field(ge=1)
reviewer_alias: str
decision: ReviewDecision
rationale: str
class AnalysisRevisedEvent(BaseModel):
event: Literal["ANALYSIS_REVISED"] = "ANALYSIS_REVISED"
from_revision: int = Field(ge=1)
to_revision: int = Field(ge=2)
rework_note: str
ReviewEvent = HumanReviewEvent | AnalysisRevisedEvent
class InvestigationContext(BaseModel):
tenant_id: str
case: InvestigationCase
analysis: RiskAssessment | None = None
analysis_revision: int = Field(default=0, ge=0)
review_status: ReviewStatus = "PENDING_REVIEW"
review_history: list[ReviewEvent] = Field(default_factory=list)
draft: SarDraft | None = None
def new_investigation_context() -> InvestigationContext:
return InvestigationContext(
tenant_id="SYNTHETIC-TENANT-001",
case=build_synthetic_case(),
)
run_context = new_investigation_context()
def require_known_case(
wrapper: RunContextWrapper[InvestigationContext],
case_id: str,
) -> InvestigationCase:
case = wrapper.context.case
if case_id != case.case_id:
raise ValueError(f"Unknown case in this request context: {case_id}")
return case
@function_tool(failure_error_function=None)
def get_case_profile(
wrapper: RunContextWrapper[InvestigationContext],
case_id: str,
) -> str:
"""Return the synthetic profile and alert context for one case."""
case = require_known_case(wrapper, case_id)
profile = case.model_dump(exclude={"transactions"})
return json.dumps(profile)
@function_tool(failure_error_function=None)
def list_case_transactions(
wrapper: RunContextWrapper[InvestigationContext],
case_id: str,
) -> str:
"""Return all synthetic transactions and their evidence identifiers."""
case = require_known_case(wrapper, case_id)
return json.dumps(
[transaction.model_dump() for transaction in case.transactions]
)
def parse_transaction_timestamp(value: str) -> datetime:
return datetime.fromisoformat(value.replace("Z", "+00:00"))
def detect_typology_signals(
case: InvestigationCase,
) -> list[dict[str, object]]:
transactions = case.transactions
cash_credits = [
transaction
for transaction in transactions
if transaction.direction == "CREDIT"
and transaction.channel == "CASH"
and DEMO_CASH_AMOUNT_MIN
<= transaction.amount
< DEMO_CASH_AMOUNT_MAX
]
if len(cash_credits) < DEMO_REQUIRED_CASH_CREDITS:
return []
activity_dates = {
parse_transaction_timestamp(item.timestamp).date()
for item in cash_credits
}
currencies = {item.currency for item in cash_credits}
if len(activity_dates) != 1 or len(currencies) != 1:
return []
cash_total = sum(item.amount for item in cash_credits)
latest_cash_timestamp = max(
parse_transaction_timestamp(item.timestamp)
for item in cash_credits
)
rapid_wires = [
transaction
for transaction in transactions
if transaction.direction == "DEBIT"
and transaction.channel == "WIRE"
and transaction.currency in currencies
and parse_transaction_timestamp(transaction.timestamp).date()
in activity_dates
and parse_transaction_timestamp(transaction.timestamp)
> latest_cash_timestamp
and transaction.amount
>= cash_total * DEMO_RAPID_MOVEMENT_RATIO
]
cash_ids = [item.id for item in cash_credits]
signals = [
{
"finding_type": "STRUCTURING_SIGNAL",
"evidence_ids": cash_ids,
"cash_credit_count": len(cash_credits),
"cash_credit_total_usd": cash_total,
"outbound_wire_usd": None,
"movement_ratio_percent": None,
}
]
if rapid_wires:
wire = rapid_wires[0]
signals.append(
{
"finding_type": "RAPID_MOVEMENT_SIGNAL",
"evidence_ids": [*cash_ids, wire.id],
"cash_credit_count": len(cash_credits),
"cash_credit_total_usd": cash_total,
"outbound_wire_usd": wire.amount,
"movement_ratio_percent": round(
wire.amount / cash_total * 100, 1
),
}
)
return signals
@function_tool(failure_error_function=None)
def run_typology_checks(
wrapper: RunContextWrapper[InvestigationContext],
case_id: str,
) -> str:
"""Return transparent demo signals and their calculated support."""
case = require_known_case(wrapper, case_id)
return json.dumps(detect_typology_signals(case))
print(json.dumps(detect_typology_signals(run_context.case), indent=2))
5. 运行分析代理
指令要求使用所有三个工具并返回类型化结果。当 RUN_ANALYSIS_DEMO=true 时，Runner.run 管理付费模型和工具循环。默认路径使用带标签的确定性固定数据，模拟工具调用证据，以便剩余的验证和审查单元格在无需 AWS 凭证或付费推理的情况下运行。该固定数据并不证明 Agents SDK 的编排能力。
MODEL_SETTINGS = ModelSettings(
reasoning=Reasoning(effort="medium"),
store=False,
)
analysis_agent = Agent[InvestigationContext](
name="Synthetic AML Analysis Agent",
model=MODEL_ID,
model_settings=MODEL_SETTINGS,
output_type=RiskAssessment,
tools=[
get_case_profile,
list_case_transactions,
run_typology_checks,
],
instructions=(
"Analyze exactly one synthetic AML case. Call get_case_profile, "
"list_case_transactions, and run_typology_checks. Copy the "
"deterministic tool's evidence IDs and computed values into the "
"matching structured finding fields. Do not introduce other amounts "
"in the finding explanations. Treat signals as prompts for qualified "
"review, not proof of intent or wrongdoing. Identify missing "
"information. Set drafting_authorized to false and filing_decision "
"to NOT_DETERMINED. Do not draft or file a SAR, change case state, "
"approve your own work, or claim that filing is required. Return "
"only the RiskAssessment schema."
),
)
REQUIRED_ANALYSIS_TOOLS = {
"get_case_profile",
"list_case_transactions",
"run_typology_checks",
}
FINDING_TEXT = {
"STRUCTURING_SIGNAL": (
"Cash-credit pattern",
"The deterministic check identified same-day cash credits.",
),
"RAPID_MOVEMENT_SIGNAL": (
"Rapid movement pattern",
"The deterministic check identified a later outbound wire.",
),
}
def build_offline_assessment(
context: InvestigationContext,
) -> RiskAssessment:
findings = []
for signal in detect_typology_signals(context.case):
title, explanation = FINDING_TEXT[signal["finding_type"]]
findings.append(
Finding(
finding_type=signal["finding_type"],
title=title,
explanation=explanation,
evidence_ids=signal["evidence_ids"],
cash_credit_count=signal["cash_credit_count"],
cash_credit_total_usd=signal["cash_credit_total_usd"],
outbound_wire_usd=signal["outbound_wire_usd"],
movement_ratio_percent=signal["movement_ratio_percent"],
)
)
return RiskAssessment(
case_id=context.case.case_id,
assessment_posture="ESCALATE_FOR_QUALIFIED_REVIEW",
executive_summary=(
"Synthetic signals require qualified review."
),
findings=findings,
information_gaps=[
"Source-of-funds records remain unverified."
],
recommended_next_steps=[
"Obtain source records for qualified human review."
],
drafting_authorized=False,
filing_decision="NOT_DETERMINED",
)
if RUN_ANALYSIS_DEMO:
result = await Runner.run(
analysis_agent,
f"Analyze synthetic case {run_context.case.case_id}.",
context=run_context,
max_turns=8,
run_config=RunConfig(
tracing_disabled=True,
workflow_name="Validated synthetic AML analysis",
),
)
assessment = result.final_output
tool_calls = {
item.raw_item.name
for item in result.new_items
if isinstance(item, ToolCallItem)
}
analysis_source = "paid Agents SDK run"
else:
assessment = build_offline_assessment(run_context)
tool_calls = set(REQUIRED_ANALYSIS_TOOLS)
analysis_source = "offline fixture with simulated tool calls"
print("Analysis source:", analysis_source)
print("Tool-call evidence:", sorted(tool_calls))
print(assessment.model_dump_json(indent=2))
6. 验证支持，而不仅仅是引用语法
验证器执行三项不同的检查：
- 引用有效性： 每个证据 ID 都存在于该案例中。
- 主张支持： 每种发现类型都使用确定性应用程序代码返回的确切交易和计算值。
- 结构化权限：
drafting_authorized必须保持为false，并且filing_decision必须保持为NOT_DETERMINED。
该模式防止模型通过这些结构化字段授予起草权限或做出提交决定。验证器不会对每一种可能的自由形式释义进行分类。合格的审阅者仍然会评估叙述、信息差距、相关性和处置。生产系统应添加专家评审的数据集、追踪评分、对抗性案例和特定于策略的测试。
def validate_risk_assessment(
candidate: RiskAssessment,
observed_tool_calls: set[str],
context: InvestigationContext,
) -> dict[str, bool]:
expected = {
signal["finding_type"]: signal
for signal in detect_typology_signals(context.case)
}
valid_evidence_ids = {
transaction.id for transaction in context.case.transactions
}
observed = {
finding.finding_type: finding for finding in candidate.findings
}
unique_finding_types = len(observed) == len(candidate.findings)
valid_citations = all(
finding.evidence_ids
and set(finding.evidence_ids).issubset(valid_evidence_ids)
for finding in candidate.findings
)
supported_claims = unique_finding_types and set(observed) == set(expected)
if supported_claims:
for finding_type, expected_support in expected.items():
finding = observed[finding_type]
supported_claims = supported_claims and (
set(finding.evidence_ids)
== set(expected_support["evidence_ids"])
and finding.cash_credit_count
== expected_support["cash_credit_count"]
and finding.cash_credit_total_usd
== expected_support["cash_credit_total_usd"]
and finding.outbound_wire_usd
== expected_support["outbound_wire_usd"]
and (
finding.movement_ratio_percent is None
and expected_support["movement_ratio_percent"] is None
or finding.movement_ratio_percent is not None
and expected_support["movement_ratio_percent"] is not None
and math.isclose(
finding.movement_ratio_percent,
expected_support["movement_ratio_percent"],
abs_tol=0.1,
)
)
)
checks = {
"case identity": candidate.case_id == context.case.case_id,
"required tools": REQUIRED_ANALYSIS_TOOLS.issubset(
observed_tool_calls
),
"valid citations": valid_citations,
"material claims supported": supported_claims,
"structured authority denied": (
candidate.drafting_authorized is False
and candidate.filing_decision == "NOT_DETERMINED"
),
}
failed = [name for name, passed in checks.items() if not passed]
if failed:
raise ValueError("Assessment validation failed: " + ", ".join(failed))
return checks
checks = validate_risk_assessment(assessment, tool_calls, run_context)
for check_name in checks:
print("PASS:", check_name)
7. 练习负面案例
这些本地测试展示了在没有额外模型调用的情况下的失败行为。一个虚构的金额可以使用真实的交易ID但仍然不受支持。无效的ID、缺失的工具调用、结构化的权威声明以及被拒绝的审核是独立的失败模式。
def expect_validation_failure(
label: str,
candidate: RiskAssessment,
observed_tools: set[str],
context: InvestigationContext,
) -> None:
try:
validate_risk_assessment(candidate, observed_tools, context)
except ValueError as exc:
print(f"PASS ({label}):", exc)
else:
raise AssertionError(f"Expected validation failure: {label}")
fabricated_claim = assessment.model_copy(deep=True)
fabricated_claim.findings[0].cash_credit_total_usd = 50_000
expect_validation_failure(
"fabricated amount", fabricated_claim, tool_calls, run_context
)
invalid_citation = assessment.model_copy(deep=True)
invalid_citation.findings[0].evidence_ids = ["TXN-999"]
expect_validation_failure(
"invalid citation", invalid_citation, tool_calls, run_context
)
missing_tool_calls = tool_calls - {"run_typology_checks"}
expect_validation_failure(
"missing tool", assessment, missing_tool_calls, run_context
)
invalid_authority = assessment.model_copy(deep=True)
invalid_authority.drafting_authorized = True
invalid_authority.filing_decision = "FILE_REQUIRED"
expect_validation_failure(
"structured authority", invalid_authority, tool_calls, run_context
)
8. 保留拒绝并需要明确的重新审查
应用程序现在拥有审批边界。它首先存储已验证的评估，记录拒绝，并证明起草仍然被阻止。然后，修订后的评估作为新修订提交并再次审查。原始拒绝保留在 review_history 中；它永远不会被重置或覆盖。
合成审查者别名和类型化上下文使示例在单个笔记本中可见。并发服务必须为每个请求和租户创建单独的上下文。生产系统必须从身份验证中派生审查者身份，授权案例和操作，持久化修订和决策，并维护独立治理的审计记录。
run_context.analysis = assessment
run_context.analysis_revision = 1
run_context.review_status = "PENDING_REVIEW"
run_context.review_history.clear()
run_context.draft = None
def record_human_review(
context: InvestigationContext,
reviewer_alias: str,
decision: ReviewDecision,
rationale: str,
) -> HumanReviewEvent:
if context.review_status not in {
"PENDING_REVIEW",
"PENDING_REREVIEW",
}:
raise RuntimeError("The current analysis revision is not awaiting review")
if len(rationale.strip()) < 12:
raise ValueError("Review rationale is too short")
event = HumanReviewEvent(
analysis_revision=context.analysis_revision,
reviewer_alias=reviewer_alias,
decision=decision,
rationale=rationale,
)
context.review_history.append(event)
context.review_status = decision
return event
def require_drafting_allowed(
context: InvestigationContext,
) -> RiskAssessment:
analysis = context.analysis
history = context.review_history
latest = history[-1] if history else None
if not isinstance(analysis, RiskAssessment):
raise RuntimeError( # noqa: TRY004 - unmet workflow precondition
"A validated analysis is required before drafting"
)
if (
not isinstance(latest, HumanReviewEvent)
or latest.decision != "APPROVE_DRAFTING"
or latest.analysis_revision != context.analysis_revision
):
raise RuntimeError(
"The current analysis revision lacks qualified-human approval"
)
return analysis
def submit_revised_analysis(
context: InvestigationContext,
candidate: RiskAssessment,
observed_tools: set[str],
rework_note: str,
) -> None:
if context.review_status != "REJECT_DRAFTING":
raise RuntimeError("A rejected review is required before rework")
validate_risk_assessment(candidate, observed_tools, context)
previous_revision = context.analysis_revision
context.analysis = candidate
context.analysis_revision = previous_revision + 1
context.review_status = "PENDING_REREVIEW"
context.review_history.append(
AnalysisRevisedEvent(
from_revision=previous_revision,
to_revision=previous_revision + 1,
rework_note=rework_note,
)
)
record_human_review(
run_context,
reviewer_alias="synthetic-qualified-reviewer",
decision="REJECT_DRAFTING",
rationale="The information gaps require revision before drafting.",
)
try:
require_drafting_allowed(run_context)
except RuntimeError as exc:
print("PASS (rejected review blocks drafting):", exc)
else:
raise AssertionError("A rejected review must block drafting")
revised_assessment = assessment.model_copy(deep=True)
revised_assessment.information_gaps.append(
"The draft must state that source-of-funds records remain unverified."
)
submit_revised_analysis(
run_context,
revised_assessment,
tool_calls,
rework_note="Added the reviewer's unresolved source-of-funds limitation.",
)
record_human_review(
run_context,
reviewer_alias="synthetic-qualified-reviewer",
decision="APPROVE_DRAFTING",
rationale="Re-reviewed revision 2 and approved draft preparation only.",
)
require_drafting_allowed(run_context)
print(
json.dumps(
[event.model_dump() for event in run_context.review_history],
indent=2,
)
)
9. 仅在批准后定义起草代理
笔记本在应用门之后定义起草代理。该代理可以读取当前已批准的分析和源证据。它没有批准、状态更改或归档的工具。运行它是可选的，并且会增加一次付费推理调用。在启动 Jupyter 之前设置 RUN_DRAFTING_DEMO=true。证据工具的异常会传播，而不是变成模型可见的错误输出。在存储草稿之前，应用程序代码会从原始运行项负载中验证每个必需的证据调用都产生了输出，然后检查当前案例 ID、证据 ID 和免责声明。本地负面测试覆盖了缺少工具输出和案例不匹配的情况，而无需进行模型调用。合格审查者仍必须确认每个叙述性声明都有支持。
@function_tool(failure_error_function=None)
def get_reviewed_analysis(
wrapper: RunContextWrapper[InvestigationContext],
case_id: str,
) -> str:
"""Return the current analysis only after explicit human approval."""
require_known_case(wrapper, case_id)
return require_drafting_allowed(wrapper.context).model_dump_json()
sar_drafting_agent = Agent[InvestigationContext](
name="Synthetic SAR Draft Preparation Agent",
model=MODEL_ID,
model_settings=MODEL_SETTINGS,
output_type=SarDraft,
tools=[
get_case_profile,
list_case_transactions,
get_reviewed_analysis,
],
instructions=(
"Prepare a synthetic draft for qualified human review only. Call all "
"three tools. Use neutral chronological language and cite supplied "
"transaction IDs and copy the supplied case ID exactly. Never submit "
"or file anything. The disclaimer must "
"say this is an AI-generated draft requiring qualified human review. "
"Return only SarDraft."
),
)
REQUIRED_DRAFTING_TOOLS = {
"get_case_profile",
"list_case_transactions",
"get_reviewed_analysis",
}
def raw_item_string(raw_item: object, field_name: str) -> str | None:
if isinstance(raw_item, dict):
value = raw_item.get(field_name)
else:
value = getattr(raw_item, field_name, None)
return value if isinstance(value, str) else None
def raw_call_id(raw_item: object) -> str | None:
return raw_item_string(raw_item, "call_id") or raw_item_string(
raw_item, "id"
)
def completed_function_tools(items: list[object]) -> set[str]:
tool_names_by_call_id: dict[str, str] = {}
completed_call_ids: set[str] = set()
for item in items:
if isinstance(item, ToolCallItem):
call_id = raw_call_id(item.raw_item)
tool_name = raw_item_string(item.raw_item, "name")
if call_id is not None and tool_name is not None:
tool_names_by_call_id[call_id] = tool_name
elif isinstance(item, ToolCallOutputItem):
call_id = raw_call_id(item.raw_item)
if call_id is not None:
completed_call_ids.add(call_id)
return {
tool_name
for call_id, tool_name in tool_names_by_call_id.items()
if call_id in completed_call_ids
}
simulated_run_items: list[object] = [
ToolCallItem(
agent=sar_drafting_agent,
raw_item={
"type": "function_call",
"call_id": "synthetic-call-001",
"name": "get_reviewed_analysis",
"arguments": "{}",
},
),
ToolCallOutputItem(
agent=sar_drafting_agent,
raw_item={
"type": "function_call_output",
"call_id": "synthetic-call-001",
"output": "{}",
},
output="{}",
),
]
if completed_function_tools(simulated_run_items) != {
"get_reviewed_analysis"
}:
raise AssertionError("Raw run-item correlation failed")
print("PASS (raw run-item correlation)")
def validate_sar_draft(
candidate: SarDraft,
context: InvestigationContext,
completed_tool_names: set[str],
) -> dict[str, bool]:
draft_ids = {
evidence_id
for citation in candidate.citations
for evidence_id in citation.evidence_ids
}
valid_evidence_ids = {
transaction.id for transaction in context.case.transactions
}
checks = {
"required tools completed": (
REQUIRED_DRAFTING_TOOLS.issubset(completed_tool_names)
),
"case identity": candidate.case_id == context.case.case_id,
"valid citations": bool(draft_ids)
and draft_ids.issubset(valid_evidence_ids),
"human-review disclaimer": (
"qualified human review" in candidate.disclaimer.casefold()
),
}
failed = [name for name, passed in checks.items() if not passed]
if failed:
raise ValueError("Draft validation failed: " + ", ".join(failed))
return checks
wrong_case_draft = SarDraft(
case_id="SYNTH-AML-OTHER",
narrative="Synthetic local test fixture.",
draft_status="INSUFFICIENT_INFORMATION",
citations=[
EvidenceCitation(
claim="A synthetic transaction exists.",
evidence_ids=["TXN-001"],
)
],
disclaimer="AI-generated draft requiring qualified human review.",
)
try:
validate_sar_draft(
wrong_case_draft, run_context, set(REQUIRED_DRAFTING_TOOLS)
)
except ValueError as exc:
print("PASS (wrong-case draft rejected):", exc)
else:
raise AssertionError("A wrong-case draft must be rejected")
missing_tool_draft = wrong_case_draft.model_copy(
update={"case_id": run_context.case.case_id}
)
try:
validate_sar_draft(
missing_tool_draft,
run_context,
REQUIRED_DRAFTING_TOOLS - {"get_reviewed_analysis"},
)
except ValueError as exc:
print("PASS (missing draft tool rejected):", exc)
else:
raise AssertionError("A draft missing a required tool must be rejected")
if RUN_DRAFTING_DEMO:
require_drafting_allowed(run_context)
draft_run = await Runner.run(
sar_drafting_agent,
f"Prepare a synthetic draft for {run_context.case.case_id}.",
context=run_context,
max_turns=8,
run_config=RunConfig(
tracing_disabled=True,
workflow_name="Synthetic SAR draft preparation",
),
)
draft = draft_run.final_output
completed_drafting_tools = completed_function_tools(
draft_run.new_items
)
draft_checks = validate_sar_draft(
draft, run_context, completed_drafting_tools
)
run_context.draft = draft
for check_name in draft_checks:
print("PASS:", check_name)
print(draft.model_dump_json(indent=2))
else:
print("Optional drafting run skipped; the approved gate remains testable.")
10. 本示例确立的要点
- 通过显式选择加入，Agents SDK 通过 Amazon Bedrock 运行有界模型和工具循环。
- 默认的离线测试夹具不进行任何 AWS 或模型调用，也不证明 Agents SDK 的编排能力。
- 类型化上下文将工具和工作流状态限定在单个请求和租户内。
- Pydantic 使材料发现结构化，但正确性来自确定性的支持检查。
- 有效 ID 和支持的声明分别进行测试。
- 被拒绝的工作保持拒绝状态，直到修订后的分析收到明确的重新审查。
- 结构化字段保持起草权限为 false，归档决策未确定。
- 起草在应用程序拥有的批准之后进行，归档不在范围内。
本笔记本是学习资产，而非生产级 AML 系统。实际部署需要机构批准的政策、身份和授权、受保护的数据处理、持久化的工作流和审计记录、由领域专家审查的评估、监控、事件响应以及法律和合规审查。单独的配套文档应涵盖 AgentCore 部署和完整的 AWS 基础设施。
参考资料
- Amazon Bedrock 中的 OpenAI 模型
- OpenAI Agents SDK 指南
- OpenAI Agents SDK for Python
- 评估智能体工作流
- GPT-5.6 Sol 模型文档
- Amazon Bedrock GPT-5.6 Sol 模型卡
- FinCEN 可疑活动报告
常见问题
常见错误包括：1) 未设置环境变量导致付费路径未启用，应检查RUN_ANALYSIS_DEMO和RUN_DRAFTING_DEMO。2) AWS凭证缺失或区域错误，需配置AWS_PROFILE和AWS_REGION。3) 模型ID不可用，需在Bedrock控制台启用模型访问。4) 验证失败：证据ID无效、金额不匹配、权限字段错误，应检查工具输出和模型指令。5) 工具调用缺失，确保代理指令要求调用所有必需工具。排查时，可先运行离线路径，再逐步启用付费路径。
企业应用扩展
生产环境优化包括：1) 使用持久化存储（如DynamoDB）保存案例和审查历史。2) 集成身份验证和授权服务（如IAM、Cognito）以获取真实审查者身份。3) 添加监控和日志记录，跟踪模型调用和验证结果。4) 扩展验证器以支持更多发现类型和策略规则。5) 使用Bedrock的模型评估工具进行持续性能监控。6) 部署为无服务器架构（如Lambda + Step Functions）以处理并发请求。7) 定期更新模型和工具，确保符合最新法规。
总结
本教程演示如何利用OpenAI Agents SDK与Amazon Bedrock上的GPT-5.6 Sol构建一个受监管的反洗钱（AML）分析工作流。通过类型化上下文、确定性验证和人类审批门控，确保模型输出可审计且符合合规要求。教程提供离线确定性路径与可选付费路径，涵盖环境配置、代理定义、证据工具、验证器、负面测试及起草代理，并强调生产环境中的安全与治理实践。
- --
内容来源与验证
- 内容等级：PREMIUM_A
- 质量等级：S_CANDIDATE
- 难度：高级
- 验证等级：V1
- 原文作者：deepakjain108
- 原文日期：2026-07-23
- 来源许可：MIT
- 来源版本：
b69c8d22080f15f4a455aab59190c6acd9b4107f - 查看原文
站内还有更多完整实战教程
加入会员，解锁全部教程、完整代码与持续更新内容。
加入会员