前言

在LangGraph构建的AI Agent生态中，Router（路由代理）作为逻辑分流的核心组件，其决策效率与可靠性高度依赖大模型输出的结构化数据。结构化输出通过明确的数据格式与字段定义，为Router提供了清晰的条件判断依据，实现了从“语义理解”到“逻辑控制”的专业分工。本文将深入探讨基于Pydantic、TypedDict和JSON Schema的结构化输出实践，及其与Router的深度整合方案。

一、Pydantic：类型安全的结构化输出方案

1.1 模型定义与数据验证

Pydantic通过定义继承自BaseModel的类，实现对输出数据结构、类型及约束的严格定义。其内置的验证系统可自动校验输入数据，确保字段符合预定义格式（如必填字段、数据类型、格式规则等），并在验证失败时抛出明确错误，显著提升数据处理的健壮性。

示例：用户信息提取模型

from typing import Optional
from pydantic import BaseModel, Field

class UserInfo(BaseModel):
    """Extracted user information, such as name, age, email, and phone number, if relevant."""
    name: str = Field(description="The name of the user")  # 必填字段，字符串类型
    age: Optional[int] = Field(description="The age of the user")  # 可选字段，整数类型
    email: str = Field(description="The email address of the user")  # 必填字段，字符串类型（通常需符合邮箱格式）
    phone: Optional[str] = Field(description="The phone number of the user")  # 可选字段，字符串类型

1.2 与LangGraph的集成

通过LangGraph的.with_structured_output()方法，可直接将Pydantic模型与大模型绑定，使大模型输出自动解析为指定的Pydantic对象，无需额外后处理逻辑。

代码实现：

import getpass
import os
from langchain_openai import ChatOpenAI
from langgraph.llms import OpenAI  # 假设LangGraph封装的OpenAI接口

# 配置OpenAI API密钥
if not os.environ.get("OPENAI_API_KEY"):
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

# 初始化大模型并绑定Pydantic输出
llm = OpenAI(model="gpt-4o-mini").with_structured_output(UserInfo)
result = llm.invoke({"input": "用户信息：张三，28岁，邮箱zhangsan@example.com，电话13812345678"})
# 输出：UserInfo(name="张三", age=28, email="zhangsan@example.com", phone="13812345678")

二、TypedDict：轻量级类型提示方案

若无需严格的数据验证，可使用Python的TypedDict定义结构化输出模式。结合Annotated语法，可添加字段描述与默认值，适合快速原型开发或对验证要求较低的场景。

2.1 模型定义与字段标注

from typing import Optional, Annotated
from typing_extensions import TypedDict  # 兼容Python 3.10+

class UserInfo(TypedDict):
    """Extracted user information from text"""
    name: Annotated[str, "The user's name"]  # 必填字段，描述为用户姓名
    age: Annotated[Optional[int], "The user's age"]  # 可选字段，描述为用户年龄
    email: Annotated[str, "The user's email address"]  # 必填字段，描述为邮箱地址
    phone: Annotated[Optional[str], "The user's phone number"]  # 可选字段，描述为电话号码

2.2 使用场景

适用于仅需类型提示而无需运行时验证的场景（如内部逻辑处理），或与Pydantic混合使用（如先通过TypedDict快速定义结构，再转换为Pydantic模型进行验证）。

三、JSON Schema：通用结构化描述语言

JSON Schema通过字典形式定义数据结构，无需引入额外类库，适合跨语言交互或需要可视化文档的场景。尽管代码较为冗长，但具备极高的通用性与可读性。

3.1 模型定义示例

user_info_schema = {
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "User Information",
    "type": "object",
    "properties": {
        "name": {
            "type": "string",
            "description": "The name of the user",
            "minLength": 1
        },
        "age": {
            "type": "integer",
            "description": "The age of the user",
            "minimum": 0,
            "maximum": 150
        },
        "email": {
            "type": "string",
            "description": "The email address of the user",
            "format": "email"
        },
        "phone": {
            "type": "string",
            "description": "The phone number of the user",
            "pattern": "^\\+?[1-9]\\d{1,14}$"  # 简单电话号码格式校验
        }
    },
    "required": ["name", "email"]  # 必填字段
}

3.2 与大模型的集成

通过LangGraph的提示工程或输出解析器，可引导大模型生成符合JSON Schema的输出，并通过工具库（如jsonschema）进行验证。

四、结构化输出驱动Router决策的实战路径

4.1 统一模型设计：基于Union类型的灵活路由

通过定义包含Union类型的父模型，可将多种业务逻辑的输出统一为单一结构，便于Router根据字段类型或值进行分流。

示例：多模态响应模型

from typing import Union
from pydantic import BaseModel, Field

# 数据库操作模型：提取用户信息用于存储
class UserInfo(BaseModel):
    name: str = Field(description="用户姓名")
    email: str = Field(description="用户邮箱")

# 对话响应模型：直接生成自然语言回答
class ConversationalResponse(BaseModel):
    response: str = Field(description="对话回复内容")

# 统一输出模型：支持两种类型的输出
class FinalResponse(BaseModel):
    final_output: Union[UserInfo, ConversationalResponse]

4.2 Router路由逻辑实现

结合上述统一模型，Router可通过解析final_output字段的类型或具体值，匹配对应的处理节点。

代码流程：

1. 大模型生成结构化输出：

   llm = OpenAI(model="gpt-4o-mini").with_structured_output(FinalResponse)
   result = llm.invoke({"input": "请记录用户信息：张三，邮箱zhangsan@example.com"})
   # 输出：FinalResponse(final_output=UserInfo(name="张三", email="zhangsan@example.com"))

2. Router根据类型分流：

   def routing_func(state):
      output_type = type(state["final_output"])
      if output_type == UserInfo:
          return "db_insert_node"  # 路由到数据库插入节点
      elif output_type == ConversationalResponse:
          return "response_node"  # 路由到回复节点

3. 构建状态图与条件边：

   builder = StateGraph(dict)
   builder.add_nodes_from([("db_insert_node", db_insert_func), ("response_node", response_func)])
   builder.add_conditional_edges(START, routing_func)  # 从起始节点触发路由

五、三种方案的对比与选型建议

六、总结：结构化输出的核心价值

结构化输出是连接大模型语义理解能力与Router逻辑控制能力的桥梁，其核心价值体现在：

1. 解耦语义与逻辑：大模型专注于内容理解与生成，Router专注于流程控制，提升系统可维护性。
2. 增强可控性：通过明确的数据格式约束，避免大模型直接参与流程决策的不可控性。
3. 标准化交互：为多代理协作、工具调用提供统一的数据接口，降低集成成本。

在LangGraph中，结合Pydantic的类型安全特性与Router的条件分流能力，可构建高可靠、易扩展的AI Agent系统。后续文章将进一步探讨结构化输出与工具代理的深度整合，以及如何通过Router实现复杂工具调用链路的动态调度。