从模型返回结构化数据指南
1. 前提概念
开始之前，请确保你熟悉以下概念：
- 聊天模型 (Chat Models)函数/工具调用 (Function/Tool Calling)
获取符合特定模式的模型输出通常很有用，例如从文本中提取数据以插入数据库或用于下游系统。本指南介绍了从模型获取结构化输出的几种策略。
2. 使用 .with_structured_output() 方法
这是获取结构化输出最简单、最可靠的方法。它适用于原生支持结构化输出（如工具/函数调用或 JSON 模式）的模型。
- 输入： 此方法接受一个模式 (Schema) 作为输入，该模式指定所需输出属性的名称、类型和描述。模式可以是：
TypedDict 类JSON Schema 字典Pydantic 类- 如果使用
TypedDict 或 JSON Schema，输出为字典。如果使用 Pydantic 类，输出为 Pydantic 对象。支持此方法的模型：
你可以在这里找到支持此方法的模型列表（请查找具有 "Structured Output" 徽章的模型）。
示例：生成结构化笑话
我们将让模型生成一个笑话，并将“铺垫” (setup) 与“笑点” (punchline) 分开。
环境设置 (以 TogetherAI + Mixtral 为例):
import getpassimport osfrom langchain_openai import ChatOpenAIif "TOGETHER_API_KEY" not in os.environ: os.environ["TOGETHER_API_KEY"] = getpass.getpass("请输入 Together AI API 密钥：")llm = ChatOpenAI( base_url="https://api.together.xyz/v1", api_key=os.environ["TOGETHER_API_KEY"], model="mistralai/Mixtral-8x7B-Instruct-v0.1", )2.1 使用 Pydantic 类
传入所需的 Pydantic 类。Pydantic 的主要优点是会对模型生成的输出进行验证。如果缺少必需字段或字段类型错误，Pydantic 会引发错误。
from typing import Optionalfrom pydantic import BaseModel, Fieldclass Joke(BaseModel): """讲给用户的笑话。""" setup: str = Field(description="笑话的铺垫") punchline: str = Field(description="笑话的笑点") rating: Optional[int] = Field( default=None, description="笑话的有趣程度，从 1 到 10" )structured_llm = llm.with_structured_output(Joke)result = structured_llm.invoke("给我讲个关于猫的笑话")print(result)提示: Pydantic 类的名称、文档字符串 (docstring)、参数名称和描述都很重要。它们通常会被添加到模型提示中，尤其是在使用函数/工具调用 API 时。
2.2 使用 TypedDict 或 JSON Schema
如果你不想使用 Pydantic，或者希望能够流式处理输出，可以使用 TypedDict。
要求:
- 核心库:
langchain-core>=0.2.26类型扩展: 强烈建议从 typing_extensions 导入 Annotated 和 TypedDict。from typing import Optional from typing_extensions import Annotated, TypedDictclass JokeTypedDict(TypedDict): """讲给用户的笑话。""" setup: Annotated[str, ..., "笑话的铺垫"] punchline: Annotated[str, ..., "笑话的笑点"] rating: Annotated[Optional[int], None, "笑话的有趣程度，从 1 到 10"] structured_llm_typeddict = llm.with_structured_output(JokeTypedDict)result_typeddict = structured_llm_typeddict.invoke("给我讲个关于猫的笑话")print(result_typeddict)或者，你可以传入一个 JSON Schema 字典。这不需要导入或类定义，但会稍微冗长一些。
json_schema = { "title": "joke", "description": "讲给用户的笑话。", "type": "object", "properties": { "setup": { "type": "string", "description": "笑话的铺垫", }, "punchline": { "type": "string", "description": "笑话的笑点", }, "rating": { "type": "integer", "description": "笑话的有趣程度，从 1 到 10", "default": None, }, }, "required": ["setup", "punchline"], }structured_llm_json = llm.with_structured_output(json_schema)result_json = structured_llm_json.invoke("给我讲个关于猫的笑话")print(result_json)2.3 在多个模式之间选择
创建一个包含联合类型 (Union) 属性的父模式，让模型从中选择。
from typing import Unionfrom pydantic import BaseModel, Field class Joke(BaseModel): """讲给用户的笑话。""" setup: str = Field(description="笑话的铺垫") punchline: str = Field(description="笑话的笑点") rating: Optional[int] = Field( default=None, description="笑话的有趣程度，从 1 到 10" )class ConversationalResponse(BaseModel): """以对话方式回应。要友好且乐于助人。""" response: str = Field(description="对用户查询的对话式回应")class FinalResponse(BaseModel): final_output: Union[Joke, ConversationalResponse]structured_llm_union = llm.with_structured_output(FinalResponse)result_joke = structured_llm_union.invoke("给我讲个关于猫的笑话")print(result_joke)result_convo = structured_llm_union.invoke("你今天怎么样？")print(result_convo)或者，如果模型支持，可以直接使用工具调用让模型在选项间选择（设置更复杂，但可能性能更好）。
2.4 流式处理
当输出类型为字典时（即模式为 TypedDict 或 JSON Schema），可以流式传输输出。
注意： 当前流式输出产生的是聚合块，而非逐字增量。
from typing import Optional from typing_extensions import Annotated, TypedDictstructured_llm_typeddict = llm.with_structured_output(JokeTypedDict)print("流式输出：")for chunk in structured_llm_typeddict.stream("给我讲个关于猫的笑话"): print(chunk)2.5 使用少量示例 (Few-Shot Prompting)
对于复杂模式，在提示中添加少量示例很有帮助。
方法一：添加到系统消息
from langchain_core.prompts import ChatPromptTemplatesystem = """你是一个搞笑的喜剧演员。你的专长是敲门笑话 (knock-knock jokes)。返回一个包含铺垫（对“谁在那儿？”的回答）和最终笑点（对“<铺垫> 谁？”的回答）的笑话。以下是一些笑话示例：示例用户: 给我讲个关于飞机的笑话示例助手: {{"setup": "为什么飞机从不累？", "punchline": "因为它们有休息的翅膀！", "rating": 2}}示例用户: 再给我讲个关于飞机的笑话示例助手: {{"setup": "货物 (Cargo)", "punchline": "汽车 (Car) go 'vroom vroom'，但飞机 (plane) go 'zoom zoom'！", "rating": 10}}示例用户: 现在讲个关于毛毛虫的示例助手: {{"setup": "毛毛虫 (Caterpillar)", "punchline": "毛毛虫爬得很慢，但看我变成蝴蝶然后抢尽风头！", "rating": 5}}"""prompt = ChatPromptTemplate.from_messages([ ("system", system), ("human", "{input}")])structured_llm_typeddict = llm.with_structured_output(JokeTypedDict) few_shot_structured_llm = prompt | structured_llm_typeddictresult_few_shot = few_shot_structured_llm.invoke({"input": "关于啄木鸟有什么好笑的？"})print(result_few_shot)方法二：使用显式工具调用 (如果模型支持)
检查模型的 API 参考是否使用工具调用。
from langchain_core.messages import AIMessage, HumanMessage, ToolMessagefrom langchain_core.prompts import ChatPromptTemplate structured_llm_pydantic = llm.with_structured_output(Joke) examples = [ HumanMessage("给我讲个关于飞机的笑话", name="示例用户"), AIMessage( content="", name="示例助手", tool_calls=[ { "name": "Joke", "args": { "setup": "为什么飞机从不累？", "punchline": "因为它们有休息的翅膀！", "rating": 2, }, "id": "tool_call_1", } ], ), ToolMessage("工具调用成功", tool_call_id="tool_call_1"), HumanMessage("再给我讲个关于飞机的笑话", name="示例用户"), AIMessage( content="", name="示例助手", tool_calls=[ { "name": "Joke", "args": { "setup": "货物 (Cargo)", "punchline": "汽车 (Car) go 'vroom vroom'，但飞机 (plane) go 'zoom zoom'！", "rating": 10, }, "id": "tool_call_2", } ], ), ToolMessage("工具调用成功", tool_call_id="tool_call_2"), HumanMessage("现在讲个关于毛毛虫的", name="示例用户"), AIMessage( content="", name="示例助手", tool_calls=[ { "name": "Joke", "args": { "setup": "毛毛虫 (Caterpillar)", "punchline": "毛毛虫爬得很慢，但看我变成蝴蝶然后抢尽风头！", "rating": 5, }, "id": "tool_call_3", } ], ), ToolMessage("工具调用成功", tool_call_id="tool_call_3"),]system = """你是一个搞笑的喜剧演员。你的专长是敲门笑话。返回一个包含铺垫（对“谁在那儿？”的回答）和最终笑点（对“<铺垫> 谁？”的回答）的笑话。"""prompt_tool_call = ChatPromptTemplate.from_messages( [ ("system", system), ("placeholder", "{examples}"), ("human", "{input}") ])few_shot_structured_llm_tool = prompt_tool_call | structured_llm_pydanticresult_tool_few_shot = few_shot_structured_llm_tool.invoke({"input": "鳄鱼", "examples": examples})print(result_tool_few_shot)2.6 (高级) 指定结构化输出的方法
对于支持多种方式（如工具调用和 JSON 模式）的模型，可以使用 method= 参数指定方法。
JSON 模式 (JSON mode):
如果使用 method="json_mode"，你仍需在模型提示中告知模型期望的 JSON 格式。传递给 with_structured_output 的模式仅用于解析输出，不会像工具调用那样传递给模型。检查模型的 API 参考以确认是否支持 JSON 模式。
structured_llm_json_mode = llm.with_structured_output(None, method="json_mode")result_json_mode = structured_llm_json_mode.invoke( "给我讲个关于猫的笑话，用包含 `setup` 和 `punchline` 键的 JSON 格式回应")print(result_json_mode)2.7 (高级) 获取原始输出
LLM 在生成复杂结构化输出时可能不完美。传递 include_raw=True 可以避免解析错误时引发异常，并允许你自行处理原始输出。输出格式将变为包含原始消息、解析后的值（如果成功）和任何解析错误的字典。
structured_llm_raw = llm.with_structured_output(Joke, include_raw=True)raw_output_result = structured_llm_raw.invoke("给我讲个关于猫的笑话")print(raw_output_result)3. 直接提示和解析模型输出
当模型不支持 .with_structured_output()（即没有工具调用或 JSON 模式支持）时，需要：
- 直接提示模型使用特定格式。使用输出解析器 (Output Parser) 从原始模型输出中提取结构化响应。
3.1 使用 PydanticOutputParser
此解析器用于解析被提示以匹配给定 Pydantic 模式的聊天模型输出。将 format_instructions 添加到提示中。
from typing import Listfrom langchain_core.output_parsers import PydanticOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom pydantic import BaseModel, Fieldclass Person(BaseModel): """关于一个人的信息。""" name: str = Field(..., description="这个人的名字") height_in_meters: float = Field( ..., description="这个人的身高，以米表示。" )class People(BaseModel): """文本中所有人的身份信息。""" people: List[Person]parser = PydanticOutputParser(pydantic_object=People)prompt_parser = ChatPromptTemplate.from_messages( [ ( "system", "回答用户查询。将输出包裹在 `json` 标签中。\n{format_instructions}", ), ("human", "{query}"), ]).partial(format_instructions=parser.get_format_instructions()) query = "安娜 23 岁，她身高 6 英尺"chain_parser = prompt_parser | llm | parserresult_parser = chain_parser.invoke({"query": query})print(result_parser)3.2 自定义解析
使用 LangChain 表达式语言 (LCEL) 创建自定义提示和解析器函数。
import jsonimport refrom typing import List, Dict, Any from langchain_core.messages import AIMessagefrom langchain_core.prompts import ChatPromptTemplatefrom pydantic import BaseModel, Field prompt_custom = ChatPromptTemplate.from_messages( [ ( "system", "回答用户查询。将你的答案输出为符合给定 Schema 的 JSON：\n" "```json\n{schema}\n```\n" "确保将答案包裹在 ```json 和 ``` 标签中。", ), ("human", "{query}"), ]).partial(schema=People.schema_json(indent=2)) def extract_json(message: AIMessage) -> List[Dict[str, Any]]: """从包含 ```json ... ``` 标签的文本中提取 JSON 内容。""" text = message.content pattern = r"```json(.*?)```" matches = re.findall(pattern, text, re.DOTALL) parsed_json = [] errors = [] for match in matches: try: parsed = json.loads(match.strip()) parsed_json.append(parsed) except json.JSONDecodeError as e: errors.append(f"解析失败的块: {match.strip()}, 错误: {e}") if not parsed_json and errors: raise ValueError(f"无法从模型输出中解析 JSON。原始输出: {text}\n错误: {errors}") elif not parsed_json: raise ValueError(f"在模型输出中未找到 ```json ... ``` 块。原始输出: {text}") return parsed_jsonquery_custom = "安娜 23 岁，她身高 6 英尺"chain_custom = prompt_custom | llm | extract_jsonresult_custom = chain_custom.invoke({"query": query_custom})print(result_custom)4. 参考文献
LangChain 官方教程：www.langchain.com.cn/
