Output Parser
🎯 学习目标
- 为链末端挂载 Output Parser，将文本转为 Python 对象
- 使用 PydanticOutputParser 配合 with_structured_output
- 设计解析失败时的重试与 fallback 策略
引言
Agent 与业务 API 需要可机器消费的输出。优先使用模型原生 JSON Schema；Parser 作为兼容层或后处理。
章节正文
第 1 步：常用 Parser 一览
| Parser | 输出 |
|---|---|
| StrOutputParser | 字符串 |
| JsonOutputParser | dict（要求模型输出合法 JSON） |
| PydanticOutputParser | Pydantic BaseModel |
| CommaSeparatedListOutputParser | list[str] |
python
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
parser = JsonOutputParser()
prompt = ChatPromptTemplate.from_messages([
("system", "输出 JSON，包含字段 answer, sources。\n{format_instructions}"),
("human", "{question}"),
]).partial(format_instructions=parser.get_format_instructions())
chain = prompt | llm | parser
chain.invoke({"question": "RAG 三步是什么？"})第 2 步：Pydantic 与 with_structured_output
python
from pydantic import BaseModel, Field
class Quote(BaseModel):
symbol: str = Field(description="股票代码")
price: float = Field(description="最新价格")
currency: str = "CNY"
structured_llm = llm.with_structured_output(Quote)
structured_llm.invoke("贵州茅台现在多少钱？") # → Quote 实例OpenAI、Anthropic 等支持 tool/schema 模式时，with_structured_output 比纯 Prompt 约束更可靠。
第 3 步：解析失败恢复
python
from langchain_core.runnables import RunnableLambda
def parse_with_retry(ai_message):
try:
return parser.parse(ai_message.content)
except Exception as e:
fix = llm.invoke(f"修正以下 JSON 错误 {e}：\n{ai_message.content}")
return parser.parse(fix.content)
chain = prompt | llm | RunnableLambda(parse_with_retry)生产环境应限制重试次数，并记录失败样本进入 Eval 集（7.1）。
动手练习
- 定义 Pydantic 模型 ContactInfo(name, email, phone)，用 structured output 抽取名片文本
- 故意让模型输出缺字段 JSON，实现一次重试修复
本节小结
Output Parser 是链的最后一环；优先 with_structured_output + Pydantic；JsonOutputParser 作 fallback；解析失败应重试上限 + 日志 + Eval。