目录

第八章:核心能力详解 - Structured Output

在真实业务中,我们很多时候并不希望模型“自由发挥”写一段话,而是希望它返回一个可以直接被程序消费的结构。例如:

这时就需要结构化输出(Structured Output)。它的意义不在于“让 JSON 更漂亮”,而在于:让大模型输出进入可校验、可测试、可组合、可自动化处理的工程流程。

本章会系统讲解结构化输出的价值、Schema 设计、LangChain 中的 `response_format` 用法、校验与重试、联合类型、错误处理,以及多个可直接上手的代码案例。

8.1 为什么结构化输出如此关键

8.1.1 自然语言不利于程序消费

自然语言适合给人读,但对程序极不友好:

例如你要做“工单分流”,程序真正需要的不是一段解释,而是这些字段:

8.1.2 结构化输出让链路更可控

它带来几个直接收益:

8.2 在 LangChain 中定义 Schema

根据 LangChain 官方文档,`response_format` 支持多种 Schema 表达方式,最常见的是 Pydantic 模型和 TypedDict。对于教程和企业项目来说,Pydantic 最直观。

8.2.1 定义一个基础 Schema

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="是否需要人工介入")

8.2.2 字段描述不是装饰品

字段描述越清晰,模型越容易稳定生成。

例如:

描述中最好补充:

8.3 使用 `create_agent(..., response_format=...)`

LangChain 官方结构化输出文档给出的现代方法之一,是在 Agent 中直接指定 `response_format`。

8.3.1 最小示例

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` 实例,而不再是一段自然语言文本。

8.3.2 结构化响应的好处

一旦你获得稳定的结构化结果,就可以直接做:

8.4 ProviderStrategy 与 ToolStrategy

根据官方文档,LangChain 会根据模型能力选择合适的结构化输出策略:

对使用者来说,大多数时候可以直接写:

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[],
    response_format=TicketAnalysis
)

让 LangChain 自动决定底层策略。

8.5 联合类型与多种输出模式

有些场景不是只有一种结构化输出,而是根据输入不同返回不同类型。结构化输出文档中也支持 `Union` 这类模式。

8.5.1 示例:客服分流

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="请判断用户请求属于退款问题还是技术支持问题。"
)

这类设计很适合:

8.6 结构化输出案例一:会议纪要行动项提取

8.6.1 定义 Schema

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="后续行动项")

8.6.2 调用模型

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"])

8.7 结构化输出案例二:风险审核结果

有些场景需要的不仅是“分类”,而是可直接驱动业务流程的审核结果。

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="你是采购风险审核助手,请输出结构化审核结论。"
)

这类结构化输出非常适合接工作流:

8.8 校验、重试与错误处理

结构化输出不意味着永远零错误。即使启用了 Schema,仍可能出现:

8.8.1 使用 `ToolStrategy(..., handle_errors=...)`

官方文档中提到,可以通过 `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,请重新生成并仅返回合法结构化结果。"
    )
)

8.8.2 业务层二次校验

即便模型已经返回结构化对象,业务层仍建议做二次校验:

def validate_ticket(result: TicketAnalysis) -> None:
    if result.priority == "high" and len(result.summary) < 5:
        raise ValueError("高优先级工单摘要过短,可能不可靠")

这类业务校验适合补足“Schema 能校验格式,但无法校验业务合理性”的缺口。

8.9 结构化输出案例三:电商评论抽取

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"])

这种结果可以被后续系统直接用于:

8.10 结构化输出设计的实践建议

8.10.1 字段宁少勿乱

初期不要试图一次抽取 30 个字段。建议从真正驱动业务流程的 3~8 个字段开始。

8.10.2 尽量用枚举而不是自由文本

例如:

这样更便于后续流程判断。

8.10.3 将展示文案与业务字段拆开

例如风险审核结果可以拆成:

8.11 常见失败模式

8.12 本章小结

练习

1. 定义一个“会议纪要行动项”Schema,至少包含负责人、任务、截止时间、优先级 4 个字段。

2. 用 `create_agent(…, response_format=…)` 构建一个工单分流器,并输出结构化结果。

3. 设计一个联合类型 `Union` Schema,用于区分“退款请求”和“技术支持请求”。

4. 写一个二次校验函数,检查结构化输出的字段是否满足你的业务规则。

5. 将一个自然语言分析流程改造成结构化输出流程,并说明这样做对后续系统有什么帮助。

参考资源