利用LangChain、LangGraph和LangSmith构建多智能体金融犯罪图谱

实际面临的问题

金融犯罪审查并非单一的提示词操作。一个案件可能需要进行客户身份识别信息补充、制裁名单筛查、负面媒体信息分析、交易模式识别，以及为调查人员提供案情说明——这些任务各自需要不同的工具、服务级别协议和故障处理方式。试图独自完成所有工作的单一链条会变得不透明，难以测试，且一旦某个工具出现故障就会带来风险。

为何多智能体系统优于单一大型链条

专业化的智能体会让提示词和工具的用途更加明确。监督者可以分配或分发任务。故障也会局限在特定部分。人工审核能够对高风险操作进行管控。通过可观测性，可以按不同角色统计成本和错误情况，而非仅依赖一个整体的追踪记录。

三层架构：LangChain、LangGraph、LangSmith

LangChain 提供模型、工具以及消息处理功能。LangGraph 负责有状态的任务编排——包括共享的案例状态、边关系、数据持久化以及中断处理。LangSmith 则负责追踪功能、数据集和评估器。普通的 while 循环代理虽然可以调用工具，但很难实现持久的案例记忆、并行检查、子图复用或一流的 HITL 功能。

共享的案例状态

首先需要明确的状态结构包括：案例编号、消息内容、评分值、标志位、路由决策以及调查员备注。

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from operator import add

class CaseState(TypedDict):
    case_id: str
    customer_id: str
    transactions: list[dict]
    messages: Annotated[list, add_messages]
    kyc_flags: Annotated[list[str], add]
    sanctions_hits: Annotated[list[dict], add]
    risk_score: float
    narrative: str
    requires_human_review: bool

带工具的单一代理节点

将模型与领域特定工具（如制裁查询、监控名单检索等）绑定，由工具节点负责执行相关调用。

from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool

@tool
def screen_ofac(name: str, dob: str | None = None) -> dict:
    """Screen an individual or entity name against the UK list."""
    # In production: call a real screening API (e.g. UK100, Refinitiv)
    return {"name": name, "match": False, "score": 0.02}

@tool
def screen_un_sanctions(name: str) -> dict:
    """Screen a name against the UN Consolidated Sanctions List."""
    return {"name": name, "match": False, "score": 0.0}
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)
sanctions_llm = llm.bind_tools([screen_ofac, screen_un_sanctions])

def sanctions_agent(state: CaseState) -> dict:
    response = sanctions_llm.invoke(state["messages"])
    return {"messages": [response]}

图的连接方式

连接代理节点和工具节点，使用 tools_condition（或类似机制）进行循环处理，直到模型停止调用工具，之后再进入下一阶段。

from langgraph.prebuilt import ToolNode, tools_condition

graph = StateGraph(CaseState)
graph.add_node("kyc_agent", kyc_agent)
graph.add_node("sanctions_agent", sanctions_agent)
graph.add_node("sanctions_tools", ToolNode([screen_ofac, screen_un_sanctions]))
graph.add_node("tx_monitoring_agent", tx_monitoring_agent)
graph.add_node("narrative_agent", narrative_agent)
graph.set_entry_point("kyc_agent")
graph.add_edge("kyc_agent", "sanctions_agent")
# Conditional edge: loop back to tools if the LLM requested a tool call
graph.add_conditional_edges(
    "sanctions_agent",
    tools_condition,
    {"tools": "sanctions_tools", END: "tx_monitoring_agent"},
)
graph.add_edge("sanctions_tools", "sanctions_agent")
graph.add_edge("tx_monitoring_agent", "narrative_agent")
graph.add_edge("narrative_agent", END)
app = graph.compile()

在构建完整图结构之前，先使用虚拟状态测试节点：

def test_sanctions_agent_flags_high_score_match():
    fake_state = {
        "messages": [HumanMessage(content="Screen: Igor Petrov, DOB 1978-03-11")],
    }
    result = sanctions_agent(fake_state)
    assert "tool_calls" in result["messages"][-1].additional_kwargs

监督路由

监督提示会确定下一个处理专家——身份验证、制裁审查或负面媒体分析——或者以总结形式结束。

from langchain_core.messages import SystemMessage

SUPERVISOR_PROMPT = """You are a financial crime case supervisor.
Route to exactly one of: kyc_agent, sanctions_agent, tx_monitoring_agent,
adverse_media_agent, narrative_agent, or FINISH if the case is fully investigated."""

def supervisor(state: CaseState) -> dict:
    routing_llm = llm.with_structured_output(RouteDecision)
    decision = routing_llm.invoke(
        [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    )
    return {"next": decision.next_agent}

def route_next(state: CaseState) -> str:
    return state["next"] if state["next"] != "FINISH" else END
graph.add_node("supervisor", supervisor)
for agent_name in ["kyc_agent", "sanctions_agent", "tx_monitoring_agent", "adverse_media_agent"]:
    graph.add_edge(agent_name, "supervisor")
graph.add_conditional_edges("supervisor", route_next)

并行分发

无需依次处理所有任务，可在身份验证之后独立进行其他检查：

graph.add_edge("kyc_agent", "sanctions_agent")
graph.add_edge("kyc_agent", "adverse_media_agent")
graph.add_edge("kyc_agent", "tx_monitoring_agent")

# Fan-in: all three must complete before scoring
graph.add_edge("sanctions_agent", "risk_scoring_agent")
graph.add_edge("adverse_media_agent", "risk_scoring_agent")
graph.add_edge("tx_monitoring_agent", "risk_scoring_agent")

人工干预机制

设置检查点，当风险超出政策阈值时中断流程，将部分状态同步到案件界面，随后根据调查人员的输入继续处理。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

def compliance_review(state: CaseState) -> dict:
    if state["risk_score"] > 0.7:
        decision = interrupt({
            "narrative": state["narrative"],
            "risk_score": state["risk_score"],
            "action_needed": "approve_or_reject_sar",
        })
        return {"requires_human_review": False, "human_decision": decision}
    return {"requires_human_review": False}

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "case-2026-00931"}}
result = app.invoke({"case_id": "case-2026-00931", ...}, config=config)

# Later, once a human reviews in your case management UI:
final = app.invoke(Command(resume={"approved": True, "reviewer": "j.smith"}), config=config)

for chunk in app.stream(
    {"case_id": "case-2026-00931", ...},
    config=config,
    stream_mode="updates",
):
    for node_name, node_output in chunk.items():
        print(f"[{node_name}] emitted: {node_output}")
        # push node_output to a websocket / case UI here

可复用单元的子图结构

将身份验证处理过程封装为独立的图结构，并作为节点嵌入其中，以便其他类型的案件复用同一处理单元。

kyc_subgraph = StateGraph(CaseState)
kyc_subgraph.add_node("resolve_entity", resolve_entity_node)
kyc_subgraph.add_node("verify_documents", verify_documents_node)
kyc_subgraph.add_node("pep_check", pep_check_node)
kyc_subgraph.set_entry_point("resolve_entity")
kyc_subgraph.add_edge("resolve_entity", "verify_documents")
kyc_subgraph.add_edge("verify_documents", "pep_check")
kyc_subgraph.add_edge("pep_check", END)
compiled_kyc = kyc_subgraph.compile()
parent_graph = StateGraph(CaseState)
parent_graph.add_node("kyc_process", compiled_kyc)  # subgraph as a node

利用LangSmith实现可观测性

启用追踪功能，并为每次运行添加案件编号和处理人员角色标签。

import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls__..."
os.environ["LANGSMITH_PROJECT"] = "financial-crime-agents"

result = app.invoke(
    {"case_id": "case-2026-00931", ...},
    config={
        "configurable": {"thread_id": "case-2026-00931"},
        "metadata": {"case_id": "case-2026-00931", "customer_tier": "high_risk"},
        "tags": ["sar_pipeline", "prod"],
    },
)

基于评估的迭代优化

该数据集可用于检测提示或工具变更时出现的错误回归。以大语言模型作为评估者，可对类似SAR的摘要进行叙事清晰度评分——在关键场景下始终会以人工审核作为最终依据。

from langsmith import Client
from langsmith.evaluation import evaluate

client = Client()
def correctness_evaluator(run, example) -> dict:
    predicted_risk = run.outputs.get("risk_score", 0)
    actual_risk = example.outputs.get("risk_score", 0)
    return {
        "key": "risk_score_accuracy",
        "score": 1.0 if abs(predicted_risk - actual_risk) < 0.15 else 0.0,
    }
results = evaluate(
    lambda inputs: app.invoke(inputs, config={"configurable": {"thread_id": "eval"}}),
    data="aml-labeled-cases-2026-h1",
    evaluators=[correctness_evaluator],
    experiment_prefix="sar-pipeline-v3",
)

from langsmith.evaluation import LangChainStringEvaluator

narrative_judge = LangChainStringEvaluator(
    "criteria",
    config={
        "criteria": {
            "cites_evidence": "Does the narrative cite specific transaction IDs and dates?",
            "regulatory_tone": "Is the language neutral and factual, avoiding speculation?",
        },
        "llm": ChatAnthropic(model="claude-sonnet-4-6", temperature=0),
    },
)
results = evaluate(
    lambda inputs: app.invoke(inputs, config=config),
    data="aml-labeled-cases-2026-h1",
    evaluators=[correctness_evaluator, narrative_judge],
    experiment_prefix="sar-narrative-v2",
)

约束机制与故障处理

对表现不稳定的工具节点进行重试，限制递归深度，并在出现制裁相关故障时直接终止流程，而非强行生成一个看似合理的结果。

from langgraph.pregel import RetryPolicy

graph.add_node(
    "sanctions_tools",
    ToolNode([screen_ofac, screen_un_sanctions]),
    retry=RetryPolicy(max_attempts=3, backoff_factor=2.0),
)

result = app.invoke(initial_state, config={**config, "recursion_limit": 25})

生产环境中的AML流程隐患

存在隐藏的共享可变全局变量、未经监督的自动提交机制、从评分到证据之间的链路缺失，以及不断重复向同一代理发起相同请求的监督节点。需限制循环出现、要求在状态中标注引用来源，并将所有提交操作置于HITL之后处理。

参考结构

一种实用的拓扑结构：中心为监管节点，专业代理配备工具循环模块，可选并行分支、合并/叙事节点、中断门控，最后进行输出——在单元重复出现时封装可重用的子图。

             ┌─────────────┐
             │  Supervisor │◄───────────────-─┐
             └──────┬──────┘                  │
     ┌───────────────┼───────────────┐        │
     ▼               ▼               ▼        │
┌─────────┐   ┌──────────────┐  ┌───────────┐ │
│   KYC   │   │ Sanctions    │  │ TX Monitor│ │
│Subgraph │   │ Agent + Tools│  │  Agent    │ │
└────┬────┘   └──────┬───────┘  └─────┬─────┘ │
     └───────────────┴────────────────┘       │
                       ▼                      │
              ┌─────────────────┐             │
              │ Risk Scoring    │─────────────┘
              └────────┬────────┘
                       ▼
              ┌─────────────────┐        ┌──────────────┐
              │ Narrative Agent │───────►│ Human Review │
              └─────────────────┘        │ (interrupt)  │
                                         └──────┬───────┘
                                                 ▼
                                          SAR Filed / Closed

成本与延迟

信息扩散会消耗代币；缓存稳定的身份验证信息；路由任务使用较小模型，叙事任务则使用较大模型；实时推送用户界面更新，让调查人员在最终报告生成前就能看到进展。

总结

多代理金融犯罪系统只有在状态明确、专业角色可测试、协调过程透明，且对不可逆操作仍需人工干预时才能成功运行。LangGraph负责控制层，LangChain提供工具，LangSmith则构建反馈循环，确保随着类型学的发展图表依然保持准确性。

受监管调查代理的治理机制

反洗钱与欺诈防范团队本就面临着模型风险及审计要求的压力。代理图必须生成可供调查人员和监管机构追踪的证据链：使用了哪些工具、匹配了哪些列表、触发了哪些阈值、谁批准了例外处理，以及是哪个模型版本生成了相关报告。这些信息应存储在案件状态中，并随案件文件一同导出，而不仅仅体现在专有的追踪界面中。

提示语和工具的更改应遵循与规则引擎相同的变更管理流程。需先在历史案件上试用新的监督提示语，比较其处理路径的分布情况，获得批准后方可正式推广。LangSmith数据集能提供帮助，但治理负责人仍需有书面的确认依据。

人为因素至关重要。如果干扰事件发生过于频繁，调查人员会忽略它们；而若发生频率过低，则风险就会逃逸。应通过抽样审查来调整阈值。流式中间状态应突出显示不确定性，而非仅呈现确定的结论，这样审查人员才能知道该关注何处。

对抗性鲁棒性是可靠性的一部分：工具输出的结果可能存在噪声或受到恶意干扰（如不良媒体中的篡改文本）。验证节点应进行净化处理并标注来源。在没有相应的校验机制的情况下，绝不能仅凭一段爬取到的文本就批准提交操作。

在衡量成功标准时，需权衡自动化处理率与质量：自动关闭的案例比例、平均分类时间、抽样审计中的误判率以及每起案件的成本。那些虽能节省分析师时间却会增加误判率的图表并不能视为成功。应设置紧急切断机制，一旦供应商的表现恶化，即可将所有任务转交给人工处理。

这些实践将参考架构从一个巧妙的图表转变为可运行的系统。上述代码示例仅是框架，而真正的运行模型才是让反金融犯罪工具获得信任的关键。

避免重复地构建专用提示词

每个专用智能体应仅接收其所需的案件状态信息。将整个调查内容都放入每个提示词中会浪费资源，还会将不必要的个人敏感信息泄露到不需要这些信息的工具中。应定义视图函数：sanctions_view(state)、kyc_view(state)，并对它们进行单元测试。监管人员可查看路由汇总信息，而内容编写者则能看到证据包。

共享工具客户端应放在包含重试机制和性能指标的库中。各智能体不应各自以不同方式实现HTTP超时处理功能。应集中管理列表提供者的凭证及断路器设置，这样制裁系统出现故障时，整个系统都能一致地作出响应。

数据溯源与特征存储

用于模式识别任务的交易特征应来自受管控的流程，而非工具内部的临时SQL查询。需为特征定义版本控制。当分析师对某个评分提出异议时，必须重新构建相关输入数据。无法解释该评分的模型图将无法通过模型风险审查。

并行性与一致性

分支输出必须以确定性的方式合并。需为列表型字段定义归约函数，为标量类型定义最后写入规则。应测试并发更新场景。如果不同的数据源同时修改risk_flags字段，应使用带注释的归约函数，使各标志值合并而非被覆盖。

与人工工作流的集成

案例管理系统（类似Jira的，或定制化的CRM）应能够处理中断操作。将中断相关的任务数据以任务形式推送；任务完成后，使用该任务的结果调用恢复API。如果调查人员已经在现有的控制台中工作，就不必再构建专门的审批网站——直接在现有系统中嵌入功能即可。

对图结构进行红队测试

模拟工具污染、客户备注中的命令注入以及实体解析不匹配的情况。确保在证据不足时系统拒绝提交相关文件。将这些测试记录为评估案例。金融犯罪检测工具具有双重用途，攻击者会试图强行清除相关记录。

成本管控

LangSmith（或类似工具）通过case_type、agent和model来运行。为每种案例类型设定预算，并对异常值发出警报。对于明显可处理或需拒绝的情况，优先使用确定性规则，而将大语言模型用于处理表述模糊的任务。混合系统在成本控制和可审计性方面均优于纯大语言模型方案。

人员配置与责任划分

需指定协调负责人、工具/数据负责人以及合规性负责人。若责任划分不清，就会产生没人敢轻易修改的复杂架构图。在新增专业代理时，应安排联合审查——每个新节点都可能成为新的故障点。

通过这些做法，LangGraph多代理模式就能真正成为用于调查工作的操作系统，而不仅仅是在某个理想情况下才能运行的演示工具。

从试点到全面应用

分析师通常只能处理为数不多的合成案例。要实现业务层面的应用，就需要对案例类型进行本地化处理、为描述内容提供语言支持，并与现有的案例处理时间表及服务级别协议相集成。应分阶段推进部署：先从风险较低的产品开始，再逐步处理风险较高的业务场景。需保留一组通过人工审核的案例作为对照组，以衡量改进效果。要明确告知团队，借助模型辅助并不意味着可以无需监督。针对调查人员的培训应包括如何解读图表轨迹、如何进行手动干预以及如何报告工具产生的错误数据。即便架构图毫无瑕疵，若缺乏此类培训，技术系统的表现也会不佳。

需预料到供应商和内部数据团队可能会在背后修改API。必须使用版本锁定、测试用图以及根据需求启用的功能标志。多智能体金融犯罪检测系统是一种持续运行的服务，而非一次性完成的笔记本转换项目。

生产环境上线前的检查清单

架构已通过合规部门审核；工具均配置了授权与超时机制；主管对历史案例的评估结果为合格；已对HITL路径进行端到端测试；LangSmith（或同类工具）的监控面板已正常运行；递归限制已设置；紧急关闭机制已有文档记录；成本预算设置了警报功能；子图版本控制方案已达成一致；并行合并简化器已通过测试；需要提供详细的说明引用；值班操作手册也已发布。多智能体金融犯罪检测系统更依赖于这些看似普通的组件，而非复杂的提示语。请逐一确认这些项，随后谨慎增加处理量，在首个季度内每周检查是否存在误判样本。

以产品用户身份与调查人员合作

调查人员并非事后才考虑的元素。在试点阶段要与他们一同工作，观察他们在哪些情况下不信任汇总结果、需要在何处查看原始数据、以及希望在哪实现一键升级功能。将这些需求转化为状态字段和中断载荷。仅以自动化率为优化目标的图表会被绕过，而这种绕过行为会消除你所宣称的审计优势。

共同设计仪表板，展示为何在某个周二主管会在完成客户身份验证之前就采取制裁措施。透明度有助于推动采用，而采用又会带来反馈，进而优化提示信息和阈值设置。没有这个循环，即便是最完美的参考架构在发布后一周也会陷入停滞。

每月发布一份由分析师协助评估的质量报告。通过共同负责的方式，可使多智能体系统始终以打击金融犯罪这一核心目标为指引，而非仅追求演示指标。只要坚持这样做，LangGraph的多智能体设计就能赢得在金融犯罪工作中至关重要的信任——这种信任源于其在真实案件、实际调查人员及真实审计中的表现，而不仅仅是因为在概念图上看起来美观。先实现那些基础的控制功能；只有在这些控制功能稳定可靠之后，再考虑通过智能路由来提升性能。调查人员会立刻察觉到差异。始终要为这样的严格审查做好准备，否则根本不够用。