内容纲要
一、结构化输出方式
内置工具方法
从结果可以明显看出，通过定制化的输出解析器得到的结果会更加的符合预期，而在 LangGraph 中更常用的，且效果更好的是，直接使用其内置的工具方法：.with_structured_output()。
这个方法通过接受一个定义了所需输出属性的名称、类型和描述的模式作为输入，进而生成一个类似模型的 Runnable 。不同于常规模型输出字符串或消息，这个 Runnable 输出一个与输入模式相匹配的对象。可以通过几种方式指定这种架构，包括 TypedDict 类、JSON Schema 或 Pydantic 类。如果采用 TypedDict 或 JSON Schema ，Runnable 将输出一个字典；若使用 Pydantic 类，则输出一个 Pydantic 对象。
先尝试和实践使用 Pydantic 类做格式化输出，应用的场景是：从文本中提取格式化的数据。
I 使用Pydantic做结构化输出（推荐）
使用 Pydantic 去限定输出格式，可以确保所有通过此模型处理的数据都会符合指定的结构和数据类型，从而减少数据处理中的错误并增加代码的健壮性。此外，Pydantic的验证系统还会自动确保所有字段都符合预定义的格式，如果输入数据不符合预期，则会抛出错误。比如下面是一个用 Pydantic 定义用户信息模型的示例：
from typing import Optional
from pydantic import BaseModel, Field
# 定义 Pydantic 模型
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")
'''
在这个 UserInfo 模型中：
name（必需）：存储用户的名字。
age（可选）：存储用户的年龄，这是一个可选字段。
email（必需）：存储用户的电子邮件地址。
phone（可选）：存储用户的电话号码，这也是一个可选字段。
对于 .with_structured_output() 方法，如果希望模型返回一个 Pydantic 对象，只需要传入所需的 Pydantic 类即可，即 UserInfo，代码如下所示：
'''
import getpass
import os
from dotenv import load_dotenv
# 加载 .env 文件中的环境变量
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key, base_url=base_url, temperature=0,)
structured_llm = llm.with_structured_output(UserInfo)
print(f"{"结构化大模型的输出结果："}\n{'*'*80}\n{structured_llm}\n{'*'*80}")
'''
打印结果：
first=_ChatModelBinding(bound=ChatOpenAI(metadata={'lc_versions': {'langchain-core': '1.4.7', 'langchain-openai': '1.3.2'}}, output_version=None, profile={'name': 'GPT-4o', 'release_date': '2024-05-13', 'last_updated': '2024-08-06', 'open_weights': False, 'max_input_tokens': 128000, 'max_output_tokens': 16384, 'text_inputs': True, 'image_inputs': True, 'audio_inputs': False, 'pdf_inputs': True, 'video_inputs': False, 'text_outputs': True, 'image_outputs': False, 'audio_outputs': False, 'video_outputs': False, 'reasoning_output': False, 'tool_calling': True, 'structured_output': True, 'attachment': True, 'temperature': True, 'image_url_inputs': True, 'pdf_tool_message': True, 'image_tool_message': True, 'tool_choice': True, 'tool_call_streaming': True}, client=<openai.resources.chat.completions.completions.Completions object at 0x000001828F547A40>, async_client=<openai.resources.chat.completions.completions.AsyncCompletions object at 0x000001828FEFBD10>, root_client=<openai.OpenAI object at 0x000001828F5450A0>, root_async_client=<openai.AsyncOpenAI object at 0x000001828F50BE30>, model_name='gpt-4o', temperature=0.0, model_kwargs={}, openai_api_key=SecretStr('**********'), openai_api_base='http://ai-network-qwen.top/minimax/v1', openai_proxy=None, stream_chunk_timeout=120.0), kwargs={'response_format': <class '__main__.UserInfo'>, 'ls_structured_output_format': {'kwargs': {'method': 'json_schema', 'strict': None}, 'schema': {'type': 'function', 'function': {'name': 'UserInfo', 'description': 'Extracted user information, such as name, age, email, and phone number, if relevant.', 'parameters': {'properties': {'name': {'description': 'The name of the user', 'type': 'string'}, 'age': {'anyOf': [{'type': 'integer'}, {'type': 'null'}], 'description': 'The age of the user'}, 'email': {'description': 'The email address of the user', 'type': 'string'}, 'phone': {'anyOf': [{'type': 'string'}, {'type': 'null'}], 'description': 'The phone number of the user'}}, 'required': ['name', 'age', 'email', 'phone'], 'type': 'object'}}}}}, config={}, config_factories=[]) middle=[] last=RunnableBinding(bound=RunnableLambda(...), kwargs={}, config={}, config_factories=[], custom_output_type=<class '__main__.UserInfo'>)
'''
# 从非结构化文本中提取用户信息
extracted_user_info = structured_llm.invoke("我叫奥特曼，今年38岁，邮箱地址是aoteman@qq.com，电话是1211111111")
# print(extracted_user_info)
print(f"{"从非结构化文本中提取用户信息："}\n{'*'*80}\n{extracted_user_info}\n{'*'*80}")
'''
打印结果：
name='奥特曼' age=38 email='aoteman@qq.com' phone='1211111111'
'''
# 换一条非结构化文本信息，再试一次，验证模型是否能够正确提取信息。
extracted_user_info = structured_llm.invoke("我叫三毛，今年38岁，邮箱地址是sanmao@qq.com，电话是1211111111")
print(f"{"换一条非结构化文本信息，再试一次，验证模型是否能够正确提取信息。："}\n{'*'*80}\n{extracted_user_info}\n{'*'*80}")
'''
打印结果：
name='三毛' age=38 email='sanmao@qq.com' phone='1211111111'
'''
# 它返回的是一个 UserInfo 的 Pydantic 对象，每个字段中则填充了在原始非结构化文本中提取出来的结构化信息。经过这样的格式化输出呢，对于 Router function 中，就可以通过类似这样的伪代码去继续路由分支的选择，比如：
# 用 isinstance 做条件判断，实现路由的分支
# isinstance 函数用于判断一个对象是否是一个已知的类型，或者是该类型的子类的实例
if isinstance(extracted_user_info, UserInfo):
print("如果extracted_user_info是 UserInfo的子类对象，则agent继续往下执行")
else:
print("如果不是，则agent放弃执行，报错，提示数据格式有问题")
'''
打印结果：
如果extracted_user_info是 UserInfo的子类对象，则agent继续往下执行
'''
extracted_user_info = "你好"
if isinstance(extracted_user_info, UserInfo):
print("如果extracted_user_info是 UserInfo的子类对象，则agent继续往下执行")
else:
print("如果不是，则agent放弃执行，报错，提示数据格式有问题")
'''
打印结果：
如果不是，则agent放弃执行，报错，提示数据格式有问题
'''
# 这就是结构化输出对于 LangGraph 中路由函数逻辑判断的意义所在。除此之外，还有 TypedDict 和 JSON Schema 能够做到相同的效果，下面依次对这两种模式进行尝试。II 使用TypedDict做结构化输出（不常用）
如果不想使用 Pydantic 去明确地验证输出参数，则可以使用 TypedDict 类定义结构化输出的模式。使用 Annotated 语法，添加对指定字段的默认值和描述。
from typing import Optional
from typing_extensions import Annotated, TypedDict
import getpass
import os
from dotenv import load_dotenv
# 加载 .env 文件中的环境变量
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key, base_url=base_url, temperature=0,)
# 定义 TypedDict 模型
class UserInfo(TypedDict):
"""Extracted user information from text"""
name: Annotated[str, ..., "The user's name"]
age: Annotated[Optional[int], None, "The user's age"]
email: Annotated[str, ..., "The user's email address"]
phone: Annotated[Optional[str], None, "The user's phone number"]
structured_llm = llm.with_structured_output(UserInfo)
# 从非结构化文本中提取用户信息
extracted_user_info = structured_llm.invoke("我叫奥特曼，今年38岁，邮箱地址是aoteman@qq.com，电话是1211111111")
print(extracted_user_info)
# 效果同pydantic一样
'''
打印结果：
{'name': '奥特曼', 'age': 38, 'email': 'aoteman@qq.com', 'phone': '1211111111'}
'''III Json Schema（繁琐-不推荐）
Json Schema 格式最为熟悉，不需要导入或类，可以直接通过字典的形式清楚地准确记录每个参数，但代价是代码会更加冗长。如下所示 user_info 的 JSON Schema，用于描述用户信息的结构，包括姓名、年龄、邮箱地址和电话号码。
# 定义 JSON Schema
import getpass
import os
from dotenv import load_dotenv
# 加载 .env 文件中的环境变量
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key, base_url=base_url, temperature=0,)
json_schema = {
"title": "user_info",
"description": "Extracted user information",
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "The user's name",
},
"age": {
"type": "integer",
"description": "The user's age",
"default": None,
},
"email": {
"type": "string",
"description": "The user's email address",
},
"phone": {
"type": "string",
"description": "The user's phone number",
"default": None,
},
},
"required": ["name", "email"],
}
structured_llm = llm.with_structured_output(json_schema)
# 从非结构化文本中提取用户信息
extracted_user_info = structured_llm.invoke("我叫奥特曼，今年38岁，邮箱地址是aoteman@qq.com，电话是1211111111")
print(extracted_user_info)
# 效果同pydantic和typeDict效果一样
'''
打印效果：
{'name': '奥特曼', 'age': 38, 'email': 'aoteman@qq.com', 'phone': '1211111111'}
'''二、 结合结构化输出构建路由图
I 概念
三种不同的结构化输出方法，更常使用的是用 Pydantic 来处理路由决策。在这种策略下，可以通过定义一个包含 Union 类型属性的父模型来灵活地从多种模式中选择适当的路由分支。例如，如果我们想根据输出决定是查询数据库还是直接回答问题，可以创建一个统一的模型来封装可能的输出类型。代码如下所示：
from typing import Union, Optional
from pydantic import BaseModel, Field
# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
model_name = os.environ["OPENAI_MODEL_NAME"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=20)
# 定义数据库插入的用户信息模型
# class UserInfo(BaseModel):
# # """Extracted user information, such as name, age, email, and phone number, if relevant."""
# """当用户消息中包含可提取的个人信息字段（姓名、年龄、邮箱、电话号码中的任意一项或多项）时，必须使用本工具。
# 只要出现'我叫/我是/姓名/今年X岁/邮箱/电话/手机'等信号词，就选本工具，不要选对话工具。"""
# 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")
# 给字段加examples示例，让模型"认得"要提取什么。例子比描述更能锚定模型对字段的理解，尤其对中文输入。
class UserInfo(BaseModel):
"""当用户消息中包含可提取的个人信息字段（姓名、年龄、邮箱、电话号码中的任意一项或多项）时，必须使用本工具。
只要出现'我叫/我是/姓名/今年X岁/邮箱/电话/手机'等信号词，就选本工具，不要选对话工具。"""
name: str = Field(description="用户姓名", examples=["奥特曼", "张三"])
age: Optional[int] = Field(description="用户年龄", examples=[38])
email: str = Field(description="用户邮箱地址", examples=["aoteman#qq.com"])
phone: Optional[str] = Field(description="用户电话号码", examples=["12111111111"])
# 定义正常生成模型回复的模型
class ConversationalResponse(BaseModel):
"""仅当用户消息是纯寒暄、提问、闲聊，且不包含任何可提取的个人信息字段时，才使用本工具。
例如：你好、谢谢、帮我解释一下XX、今天天气怎么样。"""
response: str = Field(description="A conversational response to the user's query")
# 定义最终响应模型，可以是用户信息或一般响应
class FinalResponse(BaseModel):
final_output: Union[UserInfo, ConversationalResponse]
"""
这个扩展后的代码将用于提取和存储用户的基本信息，包括姓名、年龄、电子邮件地址和电话号码的 UserInfo 模型与
用于生成面向用户的交流响应的 ConversationalResponse 模型统一的放在了 FinalResponse 模型中，使用 Union 类型来支持灵活的输出选项。
final_output 属性可以是 UserInfo 类型，也可以是 ConversationalResponse 类型，这使得系统可以根据不同的业务逻辑和用户输入决定输出的具体形式。
例如，在用户请求个人数据时可以返回 UserInfo，而在普通查询时则提供 ConversationalResponse。
"""
structured_llm = llm.with_structured_output(FinalResponse)
from langchain_core.messages import SystemMessage
system = SystemMessage(content=(
"从用户消息中提取个人信息字段（姓名、年龄、邮箱、电话/手机号）。\n"
"只要消息中包含上述任意字段，调用 UserInfo 工具填入对应值，缺失字段留空。\n"
"仅当消息是纯寒暄/提问、不含任何个人信息字段时，才调用 ConversationalResponse 工具回复。"
))
# 从非结构化文本中提取用户信息或进行一般对话响应
extracted_user_info = structured_llm.invoke([system,"你好"])
print(f"{"进行一般对话响应："}\n{'*'*80}\n{extracted_user_info}\n{'*'*80}")
# 输出：FinalResponse(final_output=ConversationalResponse(response='你好！请问有什么我可以帮助你的吗？'))
print(f"{"进行一般对话响应，不带结构信息："}\n{'*'*80}\n{extracted_user_info.final_output.response}\n{'*'*80}")
# 输出：'你好！请问有什么我可以帮助你的吗？'
"""
暴露出的问题：模型调用会造成无限制卡住
class FinalResponse(BaseModel):
final_output: Union[UserInfo, ConversationalResponse]
LangChain 对 Union 类型默认走 tool-calling 方式：它把 UserInfo 和 ConversationalResponse
各注册成一个 tool，让模型自己决定调用哪一个。这比普通结构化输出多了一层"路由决策"，具体慢在哪：
1、多一次决策开销：模型要先判断输入属于哪一类，再生成对应 schema，而不是直接填一个固定 schema。
2、重试/纠错循环：模型如果没按 tool-call 格式返回，SDK 会再发一轮 prompt 让它纠正，每轮都是一次完整的网络往返。
3、prompt 变长：两个 schema 的 JSON 描述都塞进 system prompt，token 多了，推理也慢。
解决办法：给 ChatOpenAI 加超时，如果失败就尽快失败。并且防止无限次内部的重试和纠错循环
llm = ChatOpenAI(model=model_name, api_key=key, base_url=base_url, temperature=0, request_timeout=5)
一点提醒：
request_timeout 控制的是单次 HTTP 请求。如果 Union schema 触发了 SDK 内部的重试/纠错循环，
每轮重试自己各有 30 秒，多轮累加起来仍可能等上一两分钟才最终抛错——它只是保证不会"无限卡死"，
不会让成功返回变快。要真正变快，还是得去掉 Union
"""
# 从非结构化文本中提取用户信息
# extracted_user_info = structured_llm.invoke("我叫奥特曼，今年38岁，邮箱地址是aoteman#qq.com，电话是12111111111")
extracted_user_info = structured_llm.invoke([system, "我叫奥特曼，今年38岁，邮箱地址是aoteman#qq.com，电话是12111111111"])
print(f"{"从非结构化文本中提取用户信息："}\n{'*'*80}\n{extracted_user_info}\n{'*'*80}")
# 输出：FinalResponse(final_output=UserInfo(name='奥特曼', age=38, email='aoteman#qq.com', phone='12111111111'))
print(f"{"从非结构化文本中提取用户信息，不带结构信息："}\n{'*'*80}\n{extracted_user_info.final_output}\n{'*'*80}")
# 输出：UserInfo(name='奥特曼', age=38, email='aoteman#qq.com', phone='12111111111')
'''
路由错误怎么解决？
1、改写两个 schema 的 description，让边界清晰、有优先级（最有效）
class UserInfo(BaseModel):
"""当用户消息中包含可提取的个人信息字段（姓名、年龄、邮箱、电话号码中的任意一项或多项）时，必须使用本工具。
只要出现'我叫/我是/姓名/今年X岁/邮箱/电话/手机'等信号词，就选本工具，不要选对话工具。
"""
class ConversationalResponse(BaseModel):
"""仅当用户消息是纯寒暄、提问、闲聊，且不包含任何可提取的个人信息字段时，才使用本工具。
例如：你好、谢谢、帮我解释一下XX、今天天气怎么样。"""
2、在 system prompt 里再强调一次路由规则
with_structured_output 之外，给 LLM 加 system message，双保险：
from langchain_core.messages import SystemMessage
system = SystemMessage(content=(
"从用户消息中提取个人信息字段（姓名、年龄、邮箱、电话/手机号）。\n"
"只要消息中包含上述任意字段，调用 UserInfo 工具填入对应值，缺失字段留空。\n"
"仅当消息是纯寒暄/提问、不含任何个人信息字段时，才调用 ConversationalResponse 工具回复。"
))
extracted_user_info = structured_llm.invoke([system, "我叫奥特曼，今年38岁，邮箱地址是aoteman#qq.com，电话是12111111111"])
3、给字段加 examples，让模型"认得"要提取什么
class UserInfo(BaseModel):
name: str = Field(description="用户姓名", examples=["奥特曼", "张三"])
age: Optional[int] = Field(description="用户年龄", examples=[38])
email: str = Field(description="用户邮箱地址", examples=["aoteman#qq.com"])
phone: Optional[str] = Field(description="用户电话号码", examples=["12111111111"])
'''