用 MCP 和多智能体搭建发票审核流程

把邮箱里的发票交给 AI，难点并不是让它读懂金额，而是让它按规则查供应商、识别重复单据，并把异常交给人处理。作者用 MCP 和 LangChain Deep Agents 搭了一个发票分拣示例，核心结论是：模型负责提取与解释，业务系统提供受限工具，人保留审批权。

这套方案值得看的地方是职责划分。它把文档处理拆成可检查的环节，也暴露了教程与实际交付之间的距离：PDF 怎么读取、异常标准怎么定、提取错误怎么重试，都需要继续补齐。

MCP 管数据入口，Agent 管任务分工

作者设定的业务流程很明确：收到供应商发票后，提取供应商、发票号、日期、明细和总额，核对供应商与历史记录，标出异常，再生成供负责人审批的摘要。

架构里有三种角色。总控 Agent 规划任务并汇总结果；extractor 负责读取和提取；reviewer 负责查记录、检查异常。数据库和文件读取统一放在 MCP（Model Context Protocol，模型上下文协议）服务器后面。

MCP 在这里相当于业务系统的工具接口。Agent 不需要知道 SQLite 表怎么查，只需要调用 lookup_vendor 等工具。作者希望以后更换 Agent 框架，或者接入桌面助手、内部应用时，这层接口还能复用。但复用能做到什么程度，原文没有实际验证。

示例使用 Python，安装命令是 pip install "mcp[cli]" langchain-mcp-adapters deepagents langchain-openai。虽然面向小企业，原文也没有给出任何依赖版本号，因此它更适合作为架构样例，不能视为锁定版本后可直接运行的工程。

MCP 服务通过 FastMCP("invoices") 创建，读取 business.db，暴露四个工具：

• read_invoice_text 读取 inbox 中的发票文本。
• lookup_vendor 用参数化查询匹配供应商名称。
• invoice_history 返回指定供应商最近最多 24 条记录。
• is_duplicate 按供应商 ID 和发票号判断重复。

服务器用 mcp.run(transport="stdio") 启动，可以通过 mcp dev invoice_server.py 检查工具。客户端再用 MultiServerMCPClient 启动 python invoice_server.py，通过 await client.get_tools() 加载工具，并按名称建立 by_name 映射。

这套设计最有价值的约束，是业务工具只有读取能力，付款和审批没有交给模型。

不过，开头描述的是 PDF 发票，示例里的 read_invoice_text 实际调用 path.read_text()，最后处理的也是 .txt 文件。PDF 文本提取、扫描件 OCR 都没有实现。Path(filename).name 用于去掉输入路径中的目录部分，也不应被当成完整的文件访问隔离机制。

拆成两个子 Agent，究竟解决了什么

Deep Agents 提供待办列表、用于临时笔记的虚拟文件系统，以及带独立提示词和工具集的子 Agent。作者看重的是上下文隔离：提取器可以处理杂乱的发票文字，总控只接收结构化结果。

下面这段配置最能体现分工，也展示了工具权限如何分配：

extractor = {
    "name": "extractor",
    "description": "Reads one invoice file and returns structured JSON.",
    "system_prompt": (
        "You extract invoice data. Call read_invoice_text, then return ONLY JSON "
        "with keys: vendor_name, invoice_number, date (YYYY-MM-DD), "
        "line_items (list of {description, amount}), total. "
        "If a field is missing, use null. Never guess numbers."
    ),
    "tools": [by_name["read_invoice_text"]],
}

reviewer = {
    "name": "reviewer",
    "description": "Checks extracted invoice JSON against vendor records.",
    "system_prompt": (
        "You review invoices. Given extracted JSON: look up the vendor, check for "
        "duplicates, and compare the total to that vendor's history. Report "
        "flags: UNKNOWN_VENDOR, DUPLICATE, TOTAL_MISMATCH (line items don't sum), "
        "UNUSUAL_AMOUNT (well above typical). Explain each flag in one sentence."
    ),
    "tools": [by_name["lookup_vendor"], by_name["invoice_history"],
              by_name["is_duplicate"]],
}

重点看两处 tools：提取器只能读文件，审核器只能查询供应商和历史。提示词描述职责，工具集合限定可执行的业务操作。

总控通过 create_deep_agent 创建，示例模型是 openai:gpt-4.1，配置 tools=[]、subagents=[extractor, reviewer]。它先安排待办，再委派提取，把 JSON 交给审核器，最后输出供应商、发票号、总额、异常，以及 APPROVE 或 NEEDS_HUMAN_REVIEW 建议。

这里的 APPROVE 是建议标签，没有对应付款或批准操作。总控通过 agent.ainvoke 接收任务，示例输入是 acme-1043.txt 和 newco-0007.txt。作者还提醒，实际代码应把 MCP 客户端和子 Agent 配置放进 main，让 await client.get_tools() 在同一个事件循环里执行。

这种分工有助于组织上下文，但原文没有提供单 Agent 与多 Agent 的准确率、延迟或成本对照。是否需要两个子 Agent，仍是架构取舍。

可靠性不能只靠提示词保证

作者定义了四类异常：UNKNOWN_VENDOR、DUPLICATE、TOTAL_MISMATCH 和 UNUSUAL_AMOUNT。前两项有查询工具支持；后两项暴露了规则落地的问题。

明细加总属于确定性计算，作者给出了 Pydantic 校验模型：

from pydantic import BaseModel, model_validator

class LineItem(BaseModel):
    description: str
    amount: float

class Invoice(BaseModel):
    vendor_name: str
    invoice_number: str
    date: str
    line_items: list[LineItem]
    total: float

    @model_validator(mode="after")
    def totals_match(self):
        if abs(sum(i.amount for i in self.line_items) - self.total) > 0.01:
            raise ValueError("line items do not sum to total")
        return self

关键是 > 0.01 这一行：允许的金额差异被写进代码，超出便校验失败。作者建议解析失败后让提取器重试，但没有展示把校验、错误反馈和重试接入 Agent 流程的完整实现。

这段代码也有契约不一致：提取提示词允许缺失字段返回 null，模型字段却没有声明可空；date: str 也没有落实 YYYY-MM-DD 的日期校验。实际使用前必须统一这些约定。

UNUSUAL_AMOUNT 的问题更明显。提示词只说金额「远高于典型水平」，没有说明用均值、中位数还是其他基准，也没有阈值。虽然作者强调业务规则由人定义，这条规则仍然给模型留下了较大的判断空间。

此外，作者建议记录每次工具调用及结果，收集 20–30 张已知正确答案的匿名历史发票，每次修改提示词或模型后重新运行。这是评估建议，原文没有公开样本、通过率或回归结果。

发票还应被视为不可信输入。文档可能包含「忽略此前指令」等提示注入内容，受限工具可以降低影响范围，却不能保证摘要和判断不受干扰。以后如果加入 mark_approved 等写工具，作者建议利用 Deep Agents 在敏感调用前中断，由人确认；示例没有实现这一环。

我的判断

我认同作者把重心放在业务规则、权限和评估集上。一份能解释异常来源的待审摘要，比一句看似肯定的审批结论更有用。 查重交给数据库、加总交给代码、审批交给人，这些边界比选择哪个模型更值得先确定。

但「演示能工作」还不是效果证据。原文没有完整运行结果，也没有证明多 Agent 优于固定流水线。对于输入固定、处理顺序明确的任务，普通 Python 或 Node 工作流同样可以串起提取、校验、查询和汇总，执行路径也更容易追踪。

当文档类型增加、任务需要动态拆分时，Deep Agents 的规划和委派能力才更值得评估。当前示例说明了这种组织方式，没有证明它在成本或准确率上更好。尤其是模糊供应商匹配、异常金额阈值、校验失败后的处理，仍然是团队需要补写的业务逻辑。

能拿来做什么

• 做内部单据审核页面时，可以先围绕「待人工处理」组织界面，展示提取字段、异常原因及对应查询结果，让负责人有依据地处理单据。
• 用 Node 搭类似流程时，可以沿用提取、校验、查记录、汇总的顺序，把金额计算和查重写成确定性代码，再判断是否需要 Agent 动态调度。
• 给日常 AI 编程助手开放业务数据时，可以借鉴按职责分配工具的方式，先提供范围明确的读取接口，并保存调用记录，方便追查错误。
• 试运行前先整理作者建议的 20–30 张匿名历史单据，标注正确字段和异常结果；修改模型、提示词或规则后，用同一批样本检查退化情况。