第 28 期 | 复杂数据的提取 (Structured Outputs) — LangGraph 多智能体专家课：从底层逻辑到复杂工作流的重构

各位架构师，欢迎回到我们的《LangGraph 多智能体专家课》。我是你们的老朋友。

经过前面 27 期的浴血奋战，我们的「AI 万能内容创作机构 (AI Content Agency)」已经初具规模。Planner 运筹帷幄，Researcher 翻江倒海，Writer 奋笔疾书。看着终端里不断滚动的长篇大论，你是不是有一种“AI 已经统治世界”的错觉？

但作为一名有 10 年经验的开发者，我必须要给你泼一盆冷水：如果你的 AI 系统只能输出一坨纯文本，那么它在工程上几乎是不可用的。

想象一下，你的下游系统（比如公司的 CMS 内容管理系统、数据库、或者是前端的可视化大屏）需要的是什么？是精确的 Title，是结构化的 Outline，是数组格式的 Citations，是整型的 Word Count。

如果你现在还在用正则表达式（Regex）去解析大模型输出的字符串，或者在 Prompt 里苦苦哀求大模型：“请务必、一定、发誓只输出 JSON 格式，不要包含任何多余的废话”……那么这期课程，就是来拯救你的代码和发际线的。

今天，我们将为我们 Agency 的最后一道关卡——Editor (主编) Agent，注入“灵魂契约”：Structured Outputs（结构化输出）。我们要让图谱的最终产物，从不可控的自然语言，变成具备严格 JSON Schema 的数据结构。

🎯 本期学习目标

上完这节课，我要求你掌握以下几点，并且能立刻应用到生产环境中：

认知升级：彻底搞懂 Prompt Engineering 约束、JSON Mode 与 Tool Calling (结构化输出) 的底层逻辑差异。
核心 API 掌握：熟练使用 LangChain/LangGraph 中的 .with_structured_output() 方法。
架构重构：为 AI Content Agency 的 Editor Agent 引入 Pydantic 数据校验，输出包含标题、大纲、引文、字数等严格字段的最终 Payload。
状态图融合：将结构化数据无缝写入 LangGraph 的 State 中，完美对接传统软件的 API 接口。

📖 原理解析

在写代码之前，我们要先论一论“道”。为什么提取复杂数据这么难？

大语言模型（LLM）的本质是“概率预测机器”。它天生喜欢自由发挥。而我们传统的软件工程，讲究的是“确定性”（Determinism）。Structured Outputs 就是在概率和确定性之间，建立的一座桥梁。

业界为了让 LLM 输出结构化数据，经历了三个阶段：

阶段一：Prompt 约束（刀耕火种）。在提示词里写 Output as JSON only。结果 LLM 经常给你来一句：“好的，这是您的 JSON：...”，直接把 JSON 解析器干崩。
阶段二：JSON Mode（半自动步枪）。大厂 API 提供了 response_format={ "type": "json_object" }。这保证了输出一定是合法的 JSON，但它不保证 JSON 里面的字段对不对。你想要 title，它可能给你输出 heading。
阶段三：Function/Tool Calling 强制 Schema（现代战争）。我们将需要的 JSON 结构定义为一个“函数签名（Function Signature）”。LLM 为了调用这个函数，必须严格按照我们定义的参数类型（Pydantic Schema）来填充数据。这就是目前最稳定、最优雅的解法！

在我们 AI Content Agency 的业务流中，这个机制是如何运转的呢？请看下面的架构图：

graph TD
    subgraph LangGraph State Flow
        A[State: Writer 生成的初稿 Draft] --> B(Editor Agent 节点)
        R[State: Researcher 提供的参考链接 Links] --> B
    end

    subgraph Editor Agent 内部逻辑
        B -->|1. 组装 Prompt 与上下文| C{LLM with Structured Output}
        C -.->|2. 底层转换为 Tool Calling| D[OpenAI / Anthropic API]
        D -.->|3. 返回符合 Schema 的 JSON| E[Pydantic Validation]
    end

    E -->|校验失败 自动重试| C
    E -->|校验成功| F[State: Final Article Payload]
    
    F --> G[(下游系统: CMS / 数据库 / API)]

    classDef state fill:#e1f5fe,stroke:#01579b,stroke-width:2px;
    classDef agent fill:#fff3e0,stroke:#e65100,stroke-width:2px;
    classDef core fill:#e8f5e9,stroke:#1b5e20,stroke-width:2px;
    
    class A,R,F state;
    class B agent;
    class C,E core;

图解说明：

Editor Agent 接收到纯文本的初稿（Draft）和零散的参考资料。
我们不再直接调用 llm.invoke()，而是调用绑定了 Pydantic 模型的 llm.with_structured_output(ArticleSchema)。
底层大模型会被强制将提取出的内容填入 ArticleSchema 规定的坑位中。
Pydantic 会在本地进行强校验（类型对不对？必填项有没有？）。
最终输出一个完美的 Python 对象，存入 Graph State，直接供下游使用。

💻 实战代码演练

废话不多说，Show me the code。

我们将使用 Python、LangGraph 和 Pydantic 来重构 Editor 节点。请确保你已经安装了 langchain-openai, langgraph, pydantic。

第一步：定义严格的数据契约 (Pydantic Schema)

这是结构化输出的核心。你要把大模型当成一个“填表机器”，这张表设计得越严谨，大模型填得越好。

from pydantic import BaseModel, Field
from typing import List

# 定义我们希望 Editor 最终输出的结构
class ArticlePayload(BaseModel):
    """最终交付给 CMS 系统的文章数据结构"""
    
    title: str = Field(
        description="文章的最终标题，要求吸引人、符合 SEO 规范，不超过 20 个字"
    )
    outline: List[str] = Field(
        description="文章的大纲层级，提取出所有的 H2 和 H3 标题，组成数组"
    )
    citations: List[str] = Field(
        description="从原文和参考资料中提取的引用链接或文献出处，如果没有则为空数组"
    )
    word_count: int = Field(
        description="正文部分的大致字数统计（整数）"
    )
    final_content: str = Field(
        description="经过主编润色后的最终 Markdown 格式正文内容"
    )
    seo_keywords: List[str] = Field(
        description="提取 3-5 个核心 SEO 关键词"
    )

讲师点评： 注意这里的 description！在传统开发中，注释是给人看的；但在大模型开发中，Pydantic 的 description 是给 AI 看的 Prompt 的一部分！ 你在这里写得越清晰，AI 提取的数据就越准确。

第二步：定义 LangGraph 的 State

我们需要在全局状态中，为这个结构化对象留一个位置。

from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END

class AgencyState(TypedDict):
    writer_draft: str          # Writer 传过来的初稿
    research_links: List[str]  # Researcher 传过来的参考资料
    # 👇 这里是本期的核心，我们将结构化对象存入 State
    final_delivery: Optional[ArticlePayload]

第三步：编写 Editor 节点逻辑

这是见证奇迹的时刻。我们将使用 .with_structured_output() 方法。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

def editor_node(state: AgencyState):
    print("--- 👨‍⚖️ Editor Agent 开始工作：提取并重构复杂数据 ---")
    
    draft = state.get("writer_draft", "")
    links = state.get("research_links", [])
    
    # 1. 实例化 LLM (推荐使用支持良好 Tool Calling 的模型，如 GPT-4o)
    # 注意：temperature 设为 0，因为我们现在需要的是确定性的数据提取，而不是发散性的创作
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    
    # 2. 绑定结构化输出！！！(核心魔法)
    # 这行代码会在底层将 ArticlePayload 转换为 OpenAI 的 Function Calling Schema
    structured_llm = llm.with_structured_output(ArticlePayload)
    
    # 3. 编写 Editor 的 Prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一位严苛的 AI 顶级主编。
        你的任务是审阅 Writer 的初稿，并结合参考资料，输出最终的结构化文章数据。
        请严格按照要求提取标题、大纲、引文、字数、SEO关键词，并对正文进行最终润色。"""),
        ("user", "【初稿内容】:\n{draft}\n\n【参考资料】:\n{links}")
    ])
    
    # 4. 组装 Chain 并执行
    chain = prompt | structured_llm
    
    # 执行调用。注意：这里的 result 已经不再是字符串，而是一个原生的 ArticlePayload 对象！
    result: ArticlePayload = chain.invoke({
        "draft": draft,
        "links": "\n".join(links)
    })
    
    print(f"✅ 提取完成！标题: {result.title}, 字数: {result.word_count}")
    
    # 5. 更新 State
    return {"final_delivery": result}

第四步：组装图谱并模拟运行

让我们把图谱跑起来，看看效果。

# 构建图谱
workflow = StateGraph(AgencyState)
workflow.add_node("Editor", editor_node)

# 简单起见，我们直接将 Editor 作为入口和出口
workflow.set_entry_point("Editor")
workflow.add_edge("Editor", END)

app = workflow.compile()

# === 模拟运行 Demo ===
if __name__ == "__main__":
    # 模拟 Writer 和 Researcher 传递过来的上游数据
    mock_state = {
        "writer_draft": """
        # 为什么我们需要多智能体系统？
        在当今的 AI 领域，单体大模型已经遇到了瓶颈。多智能体系统（Multi-Agent Systems）通过分工协作，极大地提升了复杂任务的解决能力。
        ## 核心优势
        1. 分布式计算
        2. 角色专业化
        总之，多智能体是未来。
        """,
        "research_links": [
            "https://arxiv.org/abs/1234.5678",
            "https://github.com/langchain-ai/langgraph"
        ]
    }
    
    # 运行图谱
    final_state = app.invoke(mock_state)
    
    # 验证输出结果
    delivery = final_state["final_delivery"]
    
    print("\n" + "="*40)
    print("🚀 最终输出的 JSON 结构 (模拟发送给 CMS):")
    print("="*40)
    # 因为 delivery 是 Pydantic 对象，我们可以直接 model_dump_json()
    print(delivery.model_dump_json(indent=2))

预期的控制台输出：

{
  "title": "多智能体系统：突破单体大模型瓶颈的未来之路",
  "outline": [
    "为什么我们需要多智能体系统？",
    "核心优势"
  ],
  "citations": [
    "https://arxiv.org/abs/1234.5678",
    "https://github.com/langchain-ai/langgraph"
  ],
  "word_count": 85,
  "final_content": "# 为什么我们需要多智能体系统？\n\n在当今的 AI 领域，单体大语言模型（LLMs）在处理极度复杂的业务逻辑时逐渐显露出瓶颈。多智能体系统（Multi-Agent Systems）应运而生，它通过引入分工协作机制，极大地提升了系统解决复杂任务的上限。\n\n## 核心优势\n\n1. **分布式计算与推理**：将复杂问题拆解，由不同的 Agent 并行处理。\n2. **角色专业化**：类似人类团队，Planner、Researcher、Writer 各司其职，减少幻觉。\n\n综上所述，多智能体架构无疑是迈向 AGI 的重要基石。",
  "seo_keywords": [
    "多智能体系统",
    "Multi-Agent",
    "大模型瓶颈",
    "AI架构"
  ]
}

看！不再是脏乱差的纯文本，而是一个完美类型化、可直接入库、可直接被前端渲染的 JSON 数据！这才是工业级 AI 应用该有的样子。

坑与避坑指南 (高阶视角的排错经验)

作为你们的导师，我不能只教你们怎么写出漂亮的 Demo。在真实的生产环境中，结构化输出往往暗藏杀机。以下是我用血泪总结出来的避坑指南：

💣 坑一：大模型“自作聪明”导致 Schema 校验失败

有时候大模型会在 JSON 外面包裹 ```json ... ```，或者某些必填字段它觉得找不到，就直接不返回，导致 Pydantic 抛出 ValidationError。 🛡️ 避坑策略：

使用 OpenAI 的 Strict Mode：如果你使用的是最新的 OpenAI 模型，LangChain 底层已经支持了 OpenAI 的 Structured Outputs (strict=True)。这能在 API 层面保证 100% 符合 Schema。
容错与重试机制：在 LangGraph 中，你可以捕获 ValidationError，并将其作为一个边（Edge）路由回 Editor 节点，把错误信息作为 Prompt 喂给大模型：“你刚才生成的 JSON 报错了，错误原因是缺少了 title 字段，请修正后重新输出。”

💣 坑二：把太多的逻辑塞进一个大 Schema 里

我见过有学员定义了一个包含 50 多个字段、嵌套了 4 层的超大 Pydantic 模型，然后指望大模型一次性把几万字的文章完全解构成这个结构。结果不仅慢，而且幻觉极其严重。 🛡️ 避坑策略：分而治之 (Divide and Conquer)。不要让 Editor 一次性做完所有事。你可以拆分节点：

Extract_Meta_Node：只负责提取 SEO 关键词、字数。
Format_Content_Node：只负责排版 Markdown。保持 Schema 的扁平化，是大模型稳定输出的秘诀。

💣 坑三：并不是所有模型都支持 `.with_structured_output()`

如果你在本地部署了较弱的开源模型（比如 Llama-2-7B），调用这个方法可能会报错，或者输出完全不符合预期的垃圾数据。 🛡️ 避坑策略： 对于不支持原生 Tool Calling 的模型，LangChain 提供了 include_raw=True 或者使用 JsonOutputParser 的降级方案。但说句掏心窝子的话，在处理复杂结构化输出时，请务必使用 GPT-4o, Claude 3.5 Sonnet 等第一梯队的模型。 在数据契约这件事上，省 token 钱往往会带来巨大的工程维护成本。

📝 本期小结

今天，我们完成了 AI Content Agency 架构中最具工程价值的一次重构。

我们学习了：

为什么结构化输出是连接 AI 与传统软件的桥梁。
如何利用 Pydantic 定义严谨的数据契约 (ArticlePayload)。
如何通过 LangChain 的 .with_structured_output() 魔法，强制 LLM 输出结构化数据。
如何将这个过程融入到 LangGraph 的 State 中，形成闭环。

现在，我们的 Editor Agent 已经不再是一个只会“说废话”的聊天机器人，而是一个能产出标准数据接口的核心微服务。

下期预告： 数据结构虽然完美了，但如果 Editor 提取的标题老板不满意怎么办？AI 终究需要人类的监督。在第 29 期，我们将引入 LangGraph 最迷人的特性之一：Human-in-the-loop (人类在环)。我们将让图谱在 Editor 输出数据后暂停，等待主编（你）点击“Approve”或者修改后，再继续流转。

各位架构师，把今天的代码敲一遍，体会一下那种数据被精准拿捏的快感。我们下期见！

第 28 期 | 复杂数据的提取 (Structured Outputs)