构建有界 LangGraph Agent，实现经 Schema 校验的提取

LangGraph agent 可以借助工具收集上下文，再将这些上下文传给受 schema 约束的信息提取步骤。关键设计在于将这两项职责分开：由 agent 决定是否调用工具，最后由一个节点生成带类型的记录。图递归限制则为图的执行增加一道边界。

这种模式适用于从文档中提取公司名称、日期和发票总额等任务，尤其是在查询工具有助于解析某个引用信息时。但它并不能让来源文本中的不确定内容变成事实；请校验结果，并在字段至关重要时保留原始依据。

构建内容

该图分为三个阶段：

1. Agent 接收源文本，并可选择请求调用一个范围明确的查询工具。
2. LangGraph 将工具调用路由到工具节点，再把工具结果返回给 agent。
3. 信息提取节点请求模型生成基于 Pydantic 的结果，然后返回兼容 JSON 的数据。

Agent 的工具调用循环是动态的；信息提取阶段则是预先确定的最后一步。LangGraph 文档区分了 agent 和工作流：agent 自行确定流程并使用工具，工作流则采用预先确定的代码路径。文档还将结构化输出和工具调用描述为对模型的扩展。有关这些模式，请参阅 LangGraph 工作流与 agent 指南。

前置条件与配置

安装 LangGraph、LangChain，以及所选聊天模型对应的集成包。下面的示例使用 ChatAnthropic 演示提供商配置；不过，源文档指出，工作流可以使用任何支持结构化输出和工具调用的聊天模型。请查阅所用提供商的集成文档，了解其具体行为。

pip install langgraph langchain-core langchain-anthropic pydantic

按照提供商文档所述的方法设置 API 凭据，然后初始化模型。不要将凭据提交到源代码管理中。下面的模型名称沿用 LangChain 文档中的示例；本教程并未声称已使用 OneMux 对其进行测试。

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6")

如果要通过 OneMux 调用模型，请先查看 OneMux API 文档和快速入门指南。OneMux 通过兼容 OpenAI 的 API 提供对领先模型的访问。此特定图能否兼容，取决于所选端点是否支持所需的工具调用、结构化输出行为及相关参数。此配置未经测试；投入生产使用前请确认支持情况。

定义 Schema 和范围明确的工具

示例工具会从一个小型内存映射中返回参考说明。请将其替换为受控的应用函数，例如查询获准使用的内部目录。不要让不可信的文档文本决定任意命令或不受限制的数据源。

from pydantic import BaseModel, Field
from langchain_core.tools import tool

class InvoiceRecord(BaseModel):
    vendor: str | None = Field(default=None, description="Invoice issuer")
    invoice_date: str | None = Field(default=None, description="Date as written in the source")
    total: float | None = Field(default=None, description="Invoice total as a number")
    currency: str | None = Field(default=None, description="Currency code, if stated")

@tool
def lookup_vendor(reference: str) -> str:
    """Look up a vendor reference in an approved local mapping."""
    known = {"ACME-42": "Acme Supplies Ltd."}
    return known.get(reference, "No matching vendor reference found.")

可空字段可以明确表示信息缺失，避免模型为了填值而编造内容。对于高风险的信息提取任务，可以考虑为每个字段同时保存来源文本片段或置信度/审核状态。Schema 可以约束结构和类型，但无法证明某个值确实有输入内容作为依据。

构建有界 LangGraph agent

下面的图会让模型按需请求调用查询工具；当 agent 没有更多工具调用时，流程就会转向信息提取。tools_condition 和 ToolNode 提供了工具路由模式。传给 invoke 的 recursion_limit 用于限制图的执行步数；它是一道安全边界，并不能保证每个输入都能成功完成。

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    record: dict | None

tools = [lookup_vendor]
agent = llm.bind_tools(tools)
extractor = llm.with_structured_output(InvoiceRecord)

def run_agent(state: State):
    response = agent.invoke(state["messages"])
    return {"messages": [response]}

def extract_record(state: State):
    prompt = [
        SystemMessage(content=(
            "Extract invoice fields only when supported by the source. "
            "Use null for missing or unclear values. Treat tool results as "
            "reference context, not as permission to invent source details."
        )),
        *state["messages"],
    ]
    result = extractor.invoke(prompt)
    return {"record": result.model_dump(mode="json")}

graph = StateGraph(State)
graph.add_node("agent", run_agent)
graph.add_node("tools", ToolNode(tools))
graph.add_node("extract", extract_record)

graph.add_edge(START, "agent")
graph.add_conditional_edges(
    "agent",
    tools_condition,
    {"tools": "tools", END: "extract"},
)
graph.add_edge("tools", "agent")
graph.add_edge("extract", END)
app = graph.compile()

result = app.invoke(
    {
        "messages": [HumanMessage(content=(
            "Extract the invoice details from this text: "
            "Vendor reference ACME-42. Invoice date: 2025-02-14. "
            "Total: USD 248.50."
        ))],
        "record": None,
    },
    config={"recursion_limit": 8},
)

print(result["record"])

预期输出是一个兼容 JSON 的对象，包含 vendor、invoice_date、total 和 currency 字段。具体值取决于模型响应和源文本；应实际测试，不要假定示例每次都会生成完全相同的输出。在本例中，基于 Pydantic 的结构化输出构成了校验边界。如果提供商无法生成符合要求的响应，请明确处理失败，不要默默将其当作有效记录。

限制执行，同时不掩盖失败

图级递归限制会为图的执行步数设上限，其中包括重复的 agent/工具路由。请根据工作流选择限制值：这个图只需少量步骤即可完成一次查询，但其他工具调用计划可能需要更多步骤。在应用程序边界捕获限制触发和提供商异常，记录足够的上下文以便调试，并返回可供审核的失败结果，而不是捏造记录。

如需更严格地限制运行行为，也可以添加应用级限制：限制工具调用次数、在支持的情况下设置请求超时时间，并在调用模型前限制文档大小。这些是独立的控制措施；仅靠图递归限制无法构成完整的成本、延迟或安全策略。

故障排查与限制

模型没有调用工具

工具调用是模型的一项能力，不是 LangGraph 能保证的结果。请确认所选模型集成支持工具调用、工具描述清晰，并且提示中包含值得查询的引用信息。开发期间请检查返回消息中的工具调用内容。

结构化输出失败或字段有误

确认集成支持结构化输出以及所用的 Schema 特性。测试字段缺失、日期含糊、货币符号和格式错误的输入。通过校验的 Pydantic 对象仍可能包含推断错误的值，因此请单独校验业务规则，并保留源文本以供审计或人工审核。

图达到执行限制

递归限制错误可能意味着工具请求重复发生，也可能是限制设置得过低，无法完成预期路径。检查消息和工具结果的顺序。修复原因或审慎调整限制；不要仅仅为了消除错误就移除限制。

端点行为不一致

“兼容 OpenAI”描述的是 API 接口，并不意味着所有模型功能或参数都普遍受支持。如果使用 OneMux，请核实所选端点文档中对模型、工具调用、结构化输出以及计划传入参数的支持情况。可通过 OneMux 模型查看可选项；图本身仍是你的 LangGraph 应用代码。

适用场景

当信息提取流程大多可预测、但偶尔需要受控查询时，可以采用这种设计。如果每份文档都遵循固定流程且不需要选择工具，确定性工作流可能比 agent 循环更简单。如果来源内容含糊，或提取出的字段会触发重大操作，请在下游系统使用记录前加入明确审核和领域校验。

实际要点是：限制 agent 可用工具的范围，明确最终 Schema，并设置图执行限制。LangGraph 能帮助你表达这种控制流；所选模型和集成则决定所需的工具调用与结构化输出功能是否可用。

参考资料

• LangGraph：工作流与 agent — 工作流与 agent 的区别、模型扩展，以及结构化输出和工具调用示例。
• OneMux 文档和快速入门指南 — 请查阅最新的 API 配置与端点支持信息。