LangChain结构化输出：Pydantic与JSON解析器实战

1. 为什么需要模型返回结构化数据？

在自然语言处理的实际应用中，我们经常会遇到这样的场景：模型输出的文本内容需要被后续程序解析和处理。比如你问模型"明天北京的天气如何？"，理想的回答可能是：

{ "city": "北京", "date": "2023-11-20", "weather": "晴", "temperature": { "high": 15, "low": 5 } }

而不是一段自由文本"明天北京天气晴朗，最高气温15度，最低气温5度"。后者虽然对人类友好，但程序需要额外编写复杂的解析逻辑才能提取关键信息。

1.1 结构化数据的优势

结构化数据相比自由文本有几个显著优势：

1. 程序可读性：JSON/YAML等格式可以直接被各种编程语言解析
2. 数据一致性：固定字段确保每次返回的数据结构相同
3. 接口标准化：便于不同系统间的数据交换
4. 类型安全：可以定义字段的数据类型，减少错误

1.2 LangChain中的结构化输出

LangChain提供了几种实现结构化输出的方式：

1. Pydantic输出解析器：利用Python的类型提示系统
2. JSON输出解析器：直接返回JSON格式
3. 自定义解析器：针对特定需求定制

提示：选择哪种方式取决于你的具体需求。Pydantic适合Python项目，JSON更通用，自定义解析器灵活性最高但开发成本也最高。

2. 使用Pydantic实现结构化输出

Pydantic是Python中用于数据验证和设置管理的库，LangChain内置了对Pydantic的支持。

2.1 定义输出模型

首先需要定义一个Pydantic模型来描述你期望的数据结构：

from pydantic import BaseModel, Field class WeatherInfo(BaseModel): city: str = Field(description="城市名称") date: str = Field(description="日期，格式为YYYY-MM-DD") weather: str = Field(description="天气状况") temperature: dict = Field(description="温度信息，包含最高和最低温度") class Config: schema_extra = { "example": { "city": "北京", "date": "2023-11-20", "weather": "晴", "temperature": {"high": 15, "low": 5} } }

2.2 创建解析器并绑定到链

from langchain.output_parsers import PydanticOutputParser from langchain.prompts import PromptTemplate from langchain.llms import OpenAI # 创建解析器 parser = PydanticOutputParser(pydantic_object=WeatherInfo) # 创建提示模板 prompt = PromptTemplate( template="回答用户问题。\n{format_instructions}\n问题：{query}\n", input_variables=["query"], partial_variables={"format_instructions": parser.get_format_instructions()} ) # 创建链 model = OpenAI(temperature=0) chain = prompt | model | parser # 执行查询 result = chain.invoke({"query": "明天北京的天气如何？"}) print(result)

2.3 关键参数解析

1. temperature=0：设置为0确保输出确定性高，适合结构化数据
2. get_format_instructions()：自动生成模型的结构描述
3. partial_variables：将格式指令作为固定部分加入提示

注意：Pydantic模型中的Field描述很重要，它们会被转换为模型的结构说明，影响LLM的输出。

3. JSON输出解析器实战

如果你不需要Pydantic的验证功能，或者需要与非Python系统交互，JSON格式是更好的选择。

3.1 基本使用

from langchain.output_parsers import StructuredOutputParser, ResponseSchema from langchain.prompts import ChatPromptTemplate # 定义响应模式 response_schemas = [ ResponseSchema(name="city", description="城市名称"), ResponseSchema(name="date", description="日期"), ResponseSchema(name="weather", description="天气状况"), ResponseSchema(name="temperature", description="温度范围") ] # 创建解析器 parser = StructuredOutputParser.from_response_schemas(response_schemas) # 创建提示 prompt = ChatPromptTemplate.from_template( """回答关于天气的问题，返回JSON格式。 {format_instructions} 问题：{query}""" ) # 创建链 chain = prompt | model | parser # 执行 result = chain.invoke({ "query": "明天上海的天气怎么样？", "format_instructions": parser.get_format_instructions() })

3.2 高级配置

# 自定义JSON格式 custom_parser = StructuredOutputParser.from_response_schemas( response_schemas, json_pattern='''{ "回答": { "城市": "...", "日期": "...", "天气": "...", "温度": {"最高": xx, "最低": xx} } }''' )

3.3 处理复杂结构

对于嵌套结构，可以这样定义：

response_schemas = [ ResponseSchema(name="city", description="城市名称"), ResponseSchema(name="date", description="日期"), ResponseSchema(name="weather", description="天气状况"), ResponseSchema( name="temperature", description="温度信息", type="object", schema={ "high": {"type": "number", "description": "最高温度"}, "low": {"type": "number", "description": "最低温度"} } ) ]

4. 常见问题与解决方案

4.1 模型不遵循格式

现象：返回自由文本而非指定格式

解决方案：

1. 加强提示词中的格式要求
2. 在示例中展示正确格式
3. 降低temperature值
4. 使用更强大的模型（如GPT-4）

改进后的提示模板：

prompt = PromptTemplate( template="""请严格按照指定格式回答问题。 格式要求： {format_instructions} 示例： 问题：明天北京的天气如何？ 回答： {{ "city": "北京", "date": "2023-11-20", "weather": "晴", "temperature": {{"high": 15, "low": 5}} }} 现在请回答： 问题：{query} 回答：""", input_variables=["query"], partial_variables={"format_instructions": parser.get_format_instructions()} )

4.2 字段缺失或错误

现象：缺少某些字段或字段值不符合预期

解决方案：

1. 在Pydantic模型中设置必填字段
2. 为字段添加更详细的描述
3. 使用try-catch处理解析错误

from pydantic import validator class WeatherInfo(BaseModel): city: str date: str weather: str temperature: dict @validator('date') def date_format(cls, v): if not re.match(r'\d{4}-\d{2}-\d{2}', v): raise ValueError("日期格式必须是YYYY-MM-DD") return v

4.3 处理多值返回

有时一个问题需要返回多个结构化结果：

class WeatherInfoList(BaseModel): items: List[WeatherInfo] # 提示词中明确说明 prompt = """返回以下城市未来三天的天气： {cities} 每个城市每天一个记录，共{count}条记录。 {format_instructions}"""

5. 性能优化技巧

5.1 批量处理

当需要处理多个相似查询时，可以使用批量模式：

queries = ["北京天气", "上海天气", "广州天气"] results = chain.batch([{"query": q} for q in queries])

5.2 缓存结果

对相同查询缓存结果：

from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())

5.3 流式处理

对于大结构数据，可以分段处理：

class ChunkedOutputParser(BaseOutputParser): def parse(self, text: str): # 实现分段解析逻辑 pass

6. 高级应用场景

6.1 动态结构

根据输入动态确定输出结构：

def dynamic_schema(query): if "天气" in query: return WeatherInfo elif "股票" in query: return StockInfo else: return BaseModel parser = PydanticOutputParser(pydantic_object=dynamic_schema(query))

6.2 多模型协作

让一个模型决定结构，另一个模型填充内容：

# 第一步：确定结构 schema_chain = prompt | model | SchemaParser() # 第二步：填充数据 data_chain = prompt | model | DataParser()

6.3 结合其他LangChain组件

from langchain.chains import LLMChain from langchain.agents import Tool weather_tool = Tool( name="Weather", func=lambda x: chain.run(query=x), description="获取天气信息，输入格式：'城市名 天气'" )

在实际项目中，我发现结构化输出特别适合以下场景：

1. 构建问答系统API
2. 数据提取和标准化
3. 多步骤工作流中的数据传递
4. 需要精确解析的自动化任务

一个实用的技巧是：先让模型用自由文本回答，再让另一个模型将其转换为结构化格式。这种方法结合了两种方式的优点，在复杂场景下效果更好。