深入解析 LangChain 源码(四):结构化输出解析——大模型的控制语言

引言

在大部分使用 AI 的场景中，AI 往往是以“对话”的模式呈现在我们的使用体验里。我们向 AI 提出问题，AI 给出回答；我们继续追问，AI 再基于上下文进行补充。这种交互方式非常自然，也非常接近人与人之间的沟通方式，因此成为了大多数 AI 产品最常见的形态。

但是，当我们开始设计 Agent，或者构建 AI 驱动的程序时，仅仅让大模型“像聊天一样回答”往往是不够的。

在真实的软件系统中，程序需要的是可以被稳定解析、可以被验证、可以被自动执行的结果。换句话说，我们不只是希望大模型“说得像那么回事”，而是希望它按照我们预先约定好的格式，产出结构化的数据。只有这样，AI 的输出才能更好地进入后续流程，被程序消费、判断和执行。

例如，一个普通的对话式 AI 可能会这样回答：

我认为用户想要查询天气，你可以调用天气接口获取北京今天的天气情况。

这对人来说很容易理解，但对程序来说却不够直接。程序还需要判断：用户的意图是什么？需要调用哪个工具？参数是什么？城市是哪里？时间范围是什么？

而如果我们要求大模型输出结构化结果，它可能会返回类似这样的 JSON：

{  "intent": "query_weather",  "tool": "weather_api",  "param": {      "city": "北京",      "date" : "today"   }}

这样的输出对于程序来说就清晰得多。程序可以直接读取 intent 判断用户意图，读取 tool 决定调用哪个工具，再从 parameters 中拿到具体参数，然后进入下一步执行流程。

这也是结构化输出在 AI 应用开发中非常重要的原因。

一 langchain中的结构化输出

如果使用的是 LangChain 的 Agent，则可以在创建 Agent 时通过 response_format 指定结构化输出格式. LangChain v1 文档说明，create_agent 可以自动处理结构化输出，模型生成的数据会被捕获、验证，并放到 Agent 状态中的 structured_response 字段里。

from pydantic import BaseModel, Fieldfrom langchain.agents import create_agentclass ContactInfo(BaseModel):    """联系人信息"""    name: str = Field(description="联系人姓名")    email: str = Field(description="联系人邮箱")    phone: str = Field(description="联系人电话")agent = create_agent(    model="openai:gpt-4o-mini",    tools=[],    response_format=ContactInfo)result = agent.invoke({    "messages": [        {            "role": "user",            "content": "张三的邮箱是 zhangsan@example.com，电话是 13800000000"        }    ]})print(result["structured_response"])

返回的结果如下:

ContactInfo(    name="张三",    email="zhangsan@example.com",    phone="13800000000")

对于 Agent 场景来说，这一点非常重要。因为 Agent 往往不是简单地回答用户，而是需要在多个步骤之间流转状态、调用工具、记录结果。结构化输出可以让 Agent 的最终结果更加可控，也更方便和业务系统集成。

二 langchain如何实现的结构化输出

LangChain 实现结构化输出，核心思路是：先定义一个 Schema，然后让模型按照这个 Schema 返回结果。

2.1 create_agent 归一化 response_format

if response_format is None:    initial_response_format = Noneelif isinstance(response_format, (ToolStrategy, ProviderStrategy)):    initial_response_format = response_formatelif isinstance(response_format, AutoStrategy):    initial_response_format = response_formatelse:    initial_response_format = AutoStrategy(schema=response_format)

可以看到这是langchain提供了两种结构化输出的策略,ToolStrategy 和 ProviderStrategy, 后面会讲到这两种策略的差异.

2.2 把 schema 包成 _SchemaSpec

langchina中支持下面的schema类型来约束结构化输出

Pydantic modeldataclassTypedDictJSON Schema dictUnion schema，主要用于 ToolStrategy

源码里的 _SchemaSpec 会把不同 schema 统一成一套描述,最后都会转变化统一的结构化描述. 源码中 _SchemaSpec 会根据 dict、Pydantic、dataclass、TypedDict 分别生成 JSON Schema；不支持的 schema 类型会抛 ValueError

if isinstance(schema, dict): schema_kind = "json_schema" json_schema = schemaelif issubclass(schema, BaseModel): schema_kind = "pydantic" json_schema = schema.model_json_schema()elif is_dataclass(schema): schema_kind = "dataclass" json_schema = TypeAdapter(schema).json_schema()elif is_typeddict(schema): schema_kind = "typeddict" json_schema = TypeAdapter(schema).json_schema()else: raise ValueError(...)

2.3 两种结构化输出策略ProviderStrategy 和 ToolStrategy

langchin提供了两种不同的结构化输出策略, 这两种的区别在于:

ProviderStrategy: 使用模型提供商原生 structured output 能力
ToolStrategy: 使用 tool calling 模拟 structured output

如果模型 provider 原生支持结构化输出，例如 OpenAI、Anthropic Claude、xAI Grok、Gemini，ProviderStrategy 通常更可靠；不支持时，LangChain 使用 tool calling 策略

同时langchain还提供了一个 AutoStrategy策略,它会根据大模型是否支持原始的结构化输出支持自动选择 ProviderStrategy 还是 ToolStrategy.

如果你直接传递的是schema

create_agent( model="openai:gpt-4o", response_format=ContactInfo,)

源码中显示会通过一个AutoStrategy进行包装AutoStrategy

AutoStrategy(schema=ContactInfo)

真正调用模型前，_get_bound_model(request) 里会判断：

ifisinstance(response_format, AutoStrategy): if _supports_provider_strategy(request.model, tools=request.tools): effective_response_format = ProviderStrategy(schema=response_format.schema) else: effective_response_format = ToolStrategy(schema=response_format.schema)

AutoStrategy  -> 如果模型支持原生 structured output       ProviderStrategy  -> 否则       ToolStrategy

2.4 ProviderStrategy 的实现----原生能力的依赖

ProviderStrategy主要是依赖模型的原生结构化输出能力,如openai在调用大模型的时候可以通过传递一个format的形式给远端模型.可以看到openai的调用方式.

所以ProviderStrategy只需要在调用的时候将schema绑定到openai的请求就行,langchain中具体的执行过程如下:

schema -> ProviderStrategyBinding -> model.bind_tools(..., structured output kwargs) -> provider 原生返回结构化结果 -> LangChain parse / validate -> state["structured_response"]

源码中 _get_bound_model 遇到 ProviderStrategy 时，会调用：

kwargs = effective_response_format.to_model_kwargs()return ( request.model.bind_tools( final_tools, strict=True, **kwargs, **request.model_settings, ), effective_response_format,)

也就是说，它并不是单独调用一个“structured output node”，而是在模型调用前把结构化输出参数绑定到 model 上。源码中这段逻辑在 _get_bound_model 里，ProviderStrategy 会转成 model kwargs 后传给 bind_tools。

模型返回 AIMessage 后，_handle_model_output 处理 ProviderStrategy：

ifisinstance(effective_response_format, ProviderStrategy): if not output.tool_calls: provider_strategy_binding = ProviderStrategyBinding.from_schema_spec( effective_response_format.schema_spec ) structured_response = provider_strategy_binding.parse(output) return { "messages": [output], "structured_response": structured_response, } return {"messages": [output]}

ProviderStrategy 期望 provider 直接在 AIMessage 里返回结构化数据；

LangChain 用 ProviderStrategyBinding.parse(output) 解析并验证；

成功后写入 structured_response。

源码里也明确：ProviderStrategy 分支会在没有 tool_calls 时 parse output；解析失败会包装成 StructuredOutputValidationError。

所以 ProviderStrategy 的流程是：

model.invoke(...) -> provider 原生 structured output -> AIMessage -> ProviderStrategyBinding.parse(output) -> Pydantic / dict -> structured_response

2.4 ToolStrategy 的实现----巧用tool args

ToolStrategy 是更有意思的一条路径。它的核心思想是：

把“结构化输出 schema”伪装成一个 tool。让模型通过 tool_call 提交结构化结果。LangChain 不真的执行这个 tool，而是拦截这个 tool_call，解析 args。

比如：

class ContactInfo(BaseModel): name: str email: str

LangChain 会构造一个“人工工具”：

Tool name: ContactInfoTool schema: name: string email: string

然后把它和普通工具一起绑定给模型。

源码里，在 create_agent 初始化阶段，如果发现 ToolStrategy，会提前创建 structured output tools：

structured_output_tools = {}if tool_strategy_for_setup: for response_schema in tool_strategy_for_setup.schema_specs: structured_tool_info = OutputToolBinding.from_schema_spec(response_schema) structured_output_tools[structured_tool_info.tool.name] = structured_tool_info

也就是说，每个 schema spec 会变成一个 OutputToolBinding，其中包含一个 artificial tool。源码中 AutoStrategy 会先转成 ToolStrategy 来提前 setup structured tools；这些工具不会放进默认工具列表，而是在调用模型时根据 effective response format 动态加入。

_get_bound_model(request) 里会构造最终 tool 列表：

final_tools = list(request.tools)if isinstance(effective_response_format, ToolStrategy): structured_tools = [info.tool for info in structured_output_tools.values()] final_tools.extend(structured_tools)

然后绑定模型：

request.model.bind_tools( final_tools, tool_choice=...)

所以 ToolStrategy 下，模型看到的 tools 包括：

用户传入的普通工具middleware 动态工具structured output artificial tools

但有个重要区别：

普通工具：

模型调用后，ToolNode 真正执行

structured output tool：

模型调用后，LangChain 拦截 tool_call，

parse args 成 structured_response，

不进入 ToolNode 执行真实工具

那最后返回的结果又是如何区分tool 和 structured output tool？

比如模型返回:

AIMessage( tool_calls=[   {   "name": "ContactInfo",   "args": {"name": "John", "email": "john@example.com"},   "id": "call_123"   } ])

_handle_model_output 里会筛选：

structured_tool_calls = [   tc for tc in output.tool_calls   if tc["name"] in structured_output_tools]

也就是说tool_call.name 在 structured_output_tools 里, 这就是一个结构化输出的tool call , 解析args就好了. 否则就调用工具.

好了,现在看看structured_output_tools 解析成功时候langchain中怎处理, 可不是简单的返回解析结构就好了.

源码里就是这个逻辑：找到对应 OutputToolBinding，用 parse(tool_call["args"]) 验证并转换，生成一个` ToolMessage`，并把结构化对象放进 structured_response。

tool_call = structured_tool_calls[0]structured_tool_binding = structured_output_tools[tool_call["name"]]structured_response = structured_tool_binding.parse(tool_call["args"])

return { "messages": [ output, ToolMessage( content=tool_message_content, tool_call_id=tool_call["id"], name=tool_call["name"], ), ], "structured_response": structured_response,}

为什么还要返回 ToolMessage？

因为从聊天协议看，模型发起了 tool call，就应该有一个对应的 tool result message。否则消息历史会不完整。

默认 tool message 内容类似：

Returning structured response: {...}

官方文档也说明，tool_message_content 可以自定义；不提供时默认显示 structured response 数据

如果模型结构化输出解析出错了,怎么处理?

如果模型错误地同时调用多个 structured output tools：

AIMessage(tool_calls=[    {"name": "ContactInfo", ...},      {"name": "EventDetails", ...},  ])

源码里会判断：

iflen(structured_tool_calls) > 1: exception = MultipleStructuredOutputsError(tool_names, output) should_retry, error_message = _handle_structured_output_error(...)

如果允许重试，则给每个错误 tool call 生成一个 ToolMessage：

ToolMessage(    content=error_message,     tool_call_id=tc["id"],   name=tc["name"],)

然后返回：

{"messages": [output, *tool_messages]}

注意这里没有 structured_response。图不会结束，会让模型看到错误消息后再重试。

官方文档也给了这个行为示例：模型错误返回多个结构化响应时，agent 会用 ToolMessage 提供错误反馈并提示模型重试。

2.5 ProviderStrategy vs ToolStrategy 对比

维度	ProviderStrategy	ToolStrategy
实现方式	用 provider 原生 structured output API	把 schema 伪装成 tool
模型输出	provider 返回结构化数据	模型返回 tool_call
LangChain 处理	`ProviderStrategyBinding.parse(output)`	`OutputToolBinding.parse(tool_call["args"])`
是否生成 ToolMessage	通常不需要	需要，补齐 tool_call 对应结果
错误恢复	provider/API 层 + LangChain parse	ToolMessage 反馈错误，让模型重试
适用模型	支持原生 structured output 的模型	支持 tool calling 的模型
返回位置	`state["structured_response"]`	`state["structured_response"]`

2.6 Model.with_structured_output() 的实现

如果不用 agent，而是直接：

structured_llm = model.with_structured_output(ContactInfo)structured_llm.invoke("...")

它走的是 ChatModel 层的结构化输出封装。官方文档说，with_structured_output() 会利用模型原生的结构化输出能力，例如工具/函数调用或 JSON mode；传入 schema 后返回一个 Runnable，输出不再是字符串或 message，而是对应 schema 的对象或 dict。

可以简单理解：

with_structured_output:   直接包装一个 ChatModel   输入 prompt   输出结构化对象create_agent(response_format=...):   在 agent model/tool 循环中内联结构化输出   输出放到 final state["structured_response"]

2.7 整体的源码级别的流程图

三总结

LangChain 的 Structured output 本质是：

把用户给的 Pydantic / dataclass / TypedDict / JSON Schema 统一转换成 schema spec；

然后根据模型能力选择两种策略：

ProviderStrategy：让 provider 原生强制模型按 schema 输出，LangChain 解析并验证 AIMessage。
ToolStrategy：把 schema 包装成人工 tool，让模型通过 tool_call 提交结构化参数，LangChain 拦截 tool_call，parse args，成功后写入 state["structured_response"]，失败则用 ToolMessage 反馈错误并让模型重试。