在真实业务中,我们很多时候并不希望模型“自由发挥”写一段话,而是希望它返回一个可以直接被程序消费的结构。例如:
这时就需要结构化输出(Structured Output)。它的意义不在于“让 JSON 更漂亮”,而在于:让大模型输出进入可校验、可测试、可组合、可自动化处理的工程流程。
本章会系统讲解结构化输出的价值、Schema 设计、LangChain 中的 `response_format` 用法、校验与重试、联合类型、错误处理,以及多个可直接上手的代码案例。
自然语言适合给人读,但对程序极不友好:
例如你要做“工单分流”,程序真正需要的不是一段解释,而是这些字段:
它带来几个直接收益:
根据 LangChain 官方文档,`response_format` 支持多种 Schema 表达方式,最常见的是 Pydantic 模型和 TypedDict。对于教程和企业项目来说,Pydantic 最直观。
from pydantic import BaseModel, Field from typing import Literal class TicketAnalysis(BaseModel): category: str = Field(description="工单类别,例如报销、账号、采购、IT") priority: Literal["low", "medium", "high"] = Field(description="优先级") summary: str = Field(description="不超过 80 字的工单摘要") need_human: bool = Field(description="是否需要人工介入")
字段描述越清晰,模型越容易稳定生成。
例如:
描述中最好补充:
LangChain 官方结构化输出文档给出的现代方法之一,是在 Agent 中直接指定 `response_format`。
from langchain.agents import create_agent agent = create_agent( model="openai:gpt-4o-mini", tools=[], response_format=TicketAnalysis, system_prompt="你是工单分诊助手,请把用户问题归类为结构化结果。" ) result = agent.invoke({ "messages": [{ "role": "user", "content": "VPN 无法连接,导致我今天在家无法访问内网。" }] }) print(result["structured_response"])
此时 `structured_response` 通常就是一个 `TicketAnalysis` 实例,而不再是一段自然语言文本。
一旦你获得稳定的结构化结果,就可以直接做:
根据官方文档,LangChain 会根据模型能力选择合适的结构化输出策略:
对使用者来说,大多数时候可以直接写:
agent = create_agent( model="openai:gpt-4o-mini", tools=[], response_format=TicketAnalysis )
让 LangChain 自动决定底层策略。
有些场景不是只有一种结构化输出,而是根据输入不同返回不同类型。结构化输出文档中也支持 `Union` 这类模式。
from typing import Union from pydantic import BaseModel, Field class RefundRequest(BaseModel): type: str = Field(default="refund") order_id: str reason: str class TechSupportRequest(BaseModel): type: str = Field(default="tech_support") issue: str severity: str RoutingSchema = Union[RefundRequest, TechSupportRequest]
然后将其作为 `response_format`:
router_agent = create_agent( model="openai:gpt-4o-mini", tools=[], response_format=RoutingSchema, system_prompt="请判断用户请求属于退款问题还是技术支持问题。" )
这类设计很适合:
from pydantic import BaseModel, Field from typing import List, Literal class ActionItem(BaseModel): owner: str = Field(description="负责人姓名") task: str = Field(description="任务内容") deadline: str = Field(description="截止日期,如未知则写待确认") priority: Literal["low", "medium", "high"] class MeetingSummary(BaseModel): topic: str = Field(description="会议主题") decisions: List[str] = Field(description="会议达成的关键结论") action_items: List[ActionItem] = Field(description="后续行动项")
meeting_agent = create_agent( model="openai:gpt-4o-mini", tools=[], response_format=MeetingSummary, system_prompt="你是会议纪要分析助手,请从纪要中提取结论和行动项。" ) result = meeting_agent.invoke({ "messages": [{ "role": "user", "content": """ 会议主题:四月产品发布准备 决定:本周完成官网上线文案终稿;演示视频由市场部统一剪辑。 待办: - 张三在 4 月 8 日前完成官网首页文案定稿; - 李四本周内确认演示视频脚本; - 王五跟进展会物料制作,优先级高。 """ }] }) print(result["structured_response"])
有些场景需要的不仅是“分类”,而是可直接驱动业务流程的审核结果。
from typing import Literal class RiskReview(BaseModel): risk_level: Literal["low", "medium", "high"] need_manager_approval: bool blocked: bool reasons: list[str] recommendation: str
对应调用:
risk_agent = create_agent( model="openai:gpt-4o-mini", tools=[], response_format=RiskReview, system_prompt="你是采购风险审核助手,请输出结构化审核结论。" )
这类结构化输出非常适合接工作流:
结构化输出不意味着永远零错误。即使启用了 Schema,仍可能出现:
官方文档中提到,可以通过 `ToolStrategy` 设置错误处理行为,例如定制校验失败后如何反馈给模型。
from langchain.agents.structured_output import ToolStrategy agent = create_agent( model="openai:gpt-4o-mini", tools=[], response_format=ToolStrategy( schema=TicketAnalysis, handle_errors="输出不符合 schema,请重新生成并仅返回合法结构化结果。" ) )
即便模型已经返回结构化对象,业务层仍建议做二次校验:
def validate_ticket(result: TicketAnalysis) -> None: if result.priority == "high" and len(result.summary) < 5: raise ValueError("高优先级工单摘要过短,可能不可靠")
这类业务校验适合补足“Schema 能校验格式,但无法校验业务合理性”的缺口。
class ReviewExtraction(BaseModel): sentiment: Literal["positive", "neutral", "negative"] pros: list[str] cons: list[str] need_followup: bool followup_reason: str review_agent = create_agent( model="openai:gpt-4o-mini", tools=[], response_format=ReviewExtraction, system_prompt="你是电商评论分析助手,请结构化提取情感和售后风险。" ) result = review_agent.invoke({ "messages": [{ "role": "user", "content": "外观不错,物流也很快,但用了三天就自动关机两次,客服暂时还没回复。" }] }) print(result["structured_response"])
这种结果可以被后续系统直接用于:
初期不要试图一次抽取 30 个字段。建议从真正驱动业务流程的 3~8 个字段开始。
例如:
这样更便于后续流程判断。
例如风险审核结果可以拆成:
1. 定义一个“会议纪要行动项”Schema,至少包含负责人、任务、截止时间、优先级 4 个字段。
2. 用 `create_agent(…, response_format=…)` 构建一个工单分流器,并输出结构化结果。
3. 设计一个联合类型 `Union` Schema,用于区分“退款请求”和“技术支持请求”。
4. 写一个二次校验函数,检查结构化输出的字段是否满足你的业务规则。
5. 将一个自然语言分析流程改造成结构化输出流程,并说明这样做对后续系统有什么帮助。