输出解析与结构化输出:OutputParser 三兄弟 + with_structured_output
链条 prompt | model 通了(D07),但模型吐的是 AIMessage,业务代码要的是字符串、dict、甚至强类型对象。今天补上最后一节车厢——libs/core/langchain_core/output_parsers/。弄清三件事:①Str/Json/Pydantic 三个解析器分别怎么"抠"出结果;②为什么"求模型输出 JSON"不靠谱、with_structured_output 怎么改用工具调用逼模型"填表";③解析失败怎么兜底(include_raw)。阶段2 到今天收官,prompt | model | parser 全链贯通。
with_structured_output 干脆换了思路——别让工匠自由发挥再验收,直接发一张结构化表单让他填(借用工具调用协议),从源头上杜绝"作文跑题"。痛点:模型只会写作文,业务要的是数据
{"name": ..., "age": ...},在提示词里千叮万嘱"只输出 JSON"。结果模型回:"好的!以下是您要的 JSON:```json {...} ```希望对您有帮助!"——json.loads 当场爆炸。就算这次抠出来了,下次它又漏个字段、写错个类型。把"自然语言 → 程序数据"这最后一公里踩实,是 LLM 应用工程化绕不开的坎。prompt | model | parser 接进 LCEL。它们做三件事:①转换(消息→目标类型)、②容错(从 markdown 围栏里抢救 JSON)、③反向指导(get_format_instructions() 生成"请按此格式输出"的说明,可以注入回提示词——质检员提前把验收标准贴在车间墙上)。| 解析器 | 位置 | 输入 → 输出 |
|---|---|---|
StrOutputParser | output_parsers/string.py:8 | AIMessage → str |
JsonOutputParser | output_parsers/json.py:31 | AIMessage → dict/list(免函数调用最可靠) |
PydanticOutputParser | output_parsers/pydantic.py:19 | AIMessage → Pydantic 对象(带校验) |
with_structured_output | language_models/chat_models.py:2357 | 不是解析器,是"模型包装":从源头逼填表 |
StrOutputParser:最简单的质检员
用得最多、也最简单的 StrOutputParser(libs/core/langchain_core/output_parsers/string.py:8):
# libs/core/langchain_core/output_parsers/string.py:8
class StrOutputParser(BaseTransformOutputParser[str]):
"""Extract text content from model outputs as a string.
Converts model outputs (such as `AIMessage` or `AIMessageChunk` objects)
into plain text strings. ...
Supports streaming, yielding text chunks as they're generated."""
@override
def parse(self, text: str) -> str: # string.py:61
"""Returns the input text with no changes."""
return text # ★没错,就一行
parse 只有一行?是的。真正的活(从 AIMessage 里取出 content 文本)在父类 BaseTransformOutputParser 里统一做了;到 parse 这一步已经是纯文本,原样返回即可。"撕包装"是所有解析器共享的逻辑,被抽到了基类。BaseTransformOutputParser名字里的 Transform 表示它支持流式:模型的 chunk 流过来,一块块转成 str 片段往下游吐——所以 chain.stream() 到终端逐字打印毫无压力(docstring 里就有 for chunk in parser.transform(stream) 的示例)。prompt | model | StrOutputParser() 之后,链的输出直接是 "你好!" 这样的字符串,再不用 .content 点来点去。90% 的纯文本场景就用它。JsonOutputParser:从作文里抠 JSON
JsonOutputParser(libs/core/langchain_core/output_parsers/json.py:31)的核心在 parse_result(json.py:61):
# libs/core/langchain_core/output_parsers/json.py:31
class JsonOutputParser(BaseCumulativeTransformOutputParser[Any]):
"""Parse the output of an LLM call to a JSON object.
Probably the most reliable output parser for getting structured data
that does *not* use function calling.
When used in streaming mode, it will yield partial JSON objects..."""
def parse_result(self, result, *, partial=False): # json.py:61
text = result[0].text
text = text.strip()
if partial: # 流式中途:解析失败不报错
try:
return parse_json_markdown(text) # ★从 markdown 围栏里抠 JSON
except JSONDecodeError:
return None # 没拼完整呢,先返回 None
try:
return parse_json_markdown(text)
except JSONDecodeError as e: # 最终结果:失败就诚实报错
msg = f"Invalid json output: {text}"
raise OutputParserException(msg, llm_output=text) from e
parse_json_markdown★容错主力:模型爱把 JSON 包在 ```json ... ``` 围栏里、前后再客套两句——这个函数负责从"作文"里定位并抠出 JSON 块再解析。这就是它 docstring 敢说"不用函数调用时最可靠"的底气。partial=True 分支流式时 JSON 只到了一半(如 {"name": "张),解析失败返回 None 而不是报错。配合基类 BaseCumulativeTransformOutputParser(累积式流解析),每来一块就重新解析一次"目前收到的全部文本",于是流式输出的是越来越完整的 dict:{} → {"name":"张三"} → {"name":"张三","age":18}。前端可以边收边渲染表单!OutputParserException带上原始输出 llm_output=text 一起抛——排查"模型到底吐了什么鬼"时救命。这个异常类型也是后面重试/兜底机制(with_fallbacks、OutputFixingParser)识别的信号。PydanticOutputParser:按图纸验收
dict 还不够——业务想要强类型对象。PydanticOutputParser(libs/core/langchain_core/output_parsers/pydantic.py:19)直接继承 JsonOutputParser,再加一道校验:
# libs/core/langchain_core/output_parsers/pydantic.py:19
class PydanticOutputParser(JsonOutputParser, Generic[TBaseModel]):
"""Parse an output using a Pydantic model."""
pydantic_object: Annotated[type[TBaseModel], SkipValidation()] # 你的"图纸"
def _parse_obj(self, obj: Any) -> TBaseModel: # pydantic.py:25
try:
if issubclass(self.pydantic_object, pydantic.BaseModel):
return self.pydantic_object.model_validate(obj) # ★Pydantic v2 校验
if issubclass(self.pydantic_object, pydantic.v1.BaseModel):
return self.pydantic_object.parse_obj(obj) # 兼容 v1
...
except (pydantic.ValidationError, pydantic.v1.ValidationError) as e:
raise self._parser_exception(e, obj) from e # 不合格 → 打回并说明原因
def parse_result(self, result, *, partial=False): # pydantic.py:57
try:
json_object = super().parse_result(result, partial=partial) # ① 先按 L03 抠出 JSON
return self._parse_obj(json_object) # ② 再按图纸验收
except OutputParserException:
if partial:
return None
raise
继承 JsonOutputParser两段式流水线:先复用 L03 的"抠 JSON",再叠加"按图纸验收"。校验是纯增量,抠取逻辑零重复。model_validate(obj)Pydantic 检查每个字段的存在性和类型(str 是 str、int 是 int),通过则返回真正的 Python 对象——IDE 能补全、mypy 能检查。_parser_exception(pydantic.py:39)校验失败时把"哪个模型、拿到什么 JSON、错在哪"拼成一条人话错误:Failed to parse Person from completion {...}. Got: 1 validation error...。with_structured_output 是"事前强制"路线:利用工具调用协议让 API 层面保证结构,代价是要求模型支持 bind_tools。能用协议就用协议,协议不可用才退回哀求。with_structured_output:不写作文,直接填表
压轴戏在 BaseChatModel.with_structured_output(libs/core/langchain_core/language_models/chat_models.py:2357)。跳过长长的 docstring,实现主体在 chat_models.py:2502-2537,惊人地短:
# libs/core/langchain_core/language_models/chat_models.py:2502
_ = kwargs.pop("method", None)
_ = kwargs.pop("strict", None)
if type(self).bind_tools is BaseChatModel.bind_tools:
msg = "with_structured_output is not implemented for this model."
raise NotImplementedError(msg) # 模型必须支持工具调用
llm = self.bind_tools( # ① ★把 schema 伪装成一个"工具"绑上
[schema],
tool_choice="any", # ② ★强制:必须调用工具(必须填表)
ls_structured_output_format={...},
)
if isinstance(schema, type) and is_basemodel_subclass(schema):
output_parser = PydanticToolsParser(tools=[schema], first_tool_only=True) # ③ Pydantic 图纸
else:
key_name = convert_to_openai_tool(schema)["function"]["name"]
output_parser = JsonOutputKeyToolsParser(key_name=key_name, first_tool_only=True)
...
return llm | output_parser # ④ ★返回的还是一条 LCEL 链!
bind_tools([schema])★神来之笔:你的 Pydantic 类被 convert_to_openai_tool 转成一份工具定义(函数名 = 类名,参数 = 字段),当作"工具"绑给模型。模型以为自己在调工具,实际是在照着你的字段清单填表。tool_choice="any"★关卡:告诉 API "这轮必须调用工具、不许自由发言"。于是返回的 AIMessage 里必有 tool_calls(D06 学的工单!),args 就是结构化数据——由 API 协议保证是合法 JSON。PydanticToolsParser从 tool_calls[0].args 里取出 dict、用你的 Pydantic 类校验实例化。first_tool_only=True:只取第一张工单。return llm | output_parser★整个"结构化输出"就是一条两节 LCEL 链:绑了工具的模型 | 工单解析器。没有任何魔法,全是 D03-D07 学过的积木。include_raw:解析失败怎么兜底
"十八")。默认直接抛异常——但有时你想"失败了也把原始回复给我,我自己处理"。这就是 include_raw=True 分支(libs/core/langchain_core/language_models/chat_models.py:2529):
# libs/core/langchain_core/language_models/chat_models.py:2529
if include_raw:
parser_assign = RunnablePassthrough.assign(
parsed=itemgetter("raw") | output_parser, parsing_error=lambda _: None
)
parser_none = RunnablePassthrough.assign(parsed=lambda _: None)
parser_with_fallback = parser_assign.with_fallbacks(
[parser_none], exception_key="parsing_error" # ★解析炸了 → 走备胎,异常装进字段
)
return RunnableMap(raw=llm) | parser_with_fallback
return llm | output_parser
RunnableMap(raw=llm)先把模型原始回复装进 {"raw": AIMessage}。parser_assign正常路径:往 dict 里追加 parsed=解析结果、parsing_error=None。with_fallbacks(...)★解析抛异常时自动切换到备胎 parser_none:parsed=None,异常对象放进 parsing_error 字段。于是无论成败,你都拿到 {"raw":…, "parsed":…, "parsing_error":…} 三件套,永不炸链。RunnablePassthrough.assign、with_fallbacks、RunnableMap 这些 LCEL 通用积木组合出来(D18 会专门拆这些积木)。框架自己也在用自己的积木盖楼,这是验证抽象好坏的金标准。串起来 + 今日小结
class AnswerWithJustification(BaseModel): answer: str; justification: strstructured_model = model.with_structured_output(AnswerWithJustification)structured_model.invoke("What weighs more a pound of bricks or a pound of feathers")→
AnswerWithJustification(answer='They weigh the same', justification='Both a pound of bricks and a pound of feathers weigh one pound...')而
include_raw=True 时返回三件套:{'raw': AIMessage(..., tool_calls=[{'name': 'AnswerWithJustification', 'args': {...}}]), 'parsed': AnswerWithJustification(...), 'parsing_error': None}——从 raw 里能清楚看到"填表"确实是靠 tool_calls 实现的。👶 小白:那我到底该用 PydanticOutputParser 还是 with_structured_output?
👨🏫 老师:先问一句——你的模型支持工具调用吗?支持(GPT/Claude/千问等主流模型都支持)就用 with_structured_output:API 层面保证 JSON 合法,成功率高得多,代码也短。不支持(一些本地小模型),才用 PydanticOutputParser:把 get_format_instructions() 注入提示词、事后抠取+校验,必要时再套重试。JsonOutputParser 的 docstring 说自己是"不用函数调用时最可靠的"——言下之意,能用函数调用就别用它硬抠。
🧠 今天你应该能回答
- StrOutputParser.parse 为什么只有一行?(取 content 的活在基类统一做了,string.py:61)
- JsonOutputParser 靠什么容错?(parse_json_markdown 从 markdown 围栏里抠 JSON,json.py:61)
- 流式时 partial=True 是什么行为?(解析半截 JSON 失败返回 None,逐步产出越来越完整的 dict)
- PydanticOutputParser 和 JsonOutputParser 什么关系?(继承:先抠 JSON 再按图纸校验,pydantic.py:19/57)
- with_structured_output 的三板斧?(schema 伪装成工具 bind_tools + tool_choice="any" 强制填表 + 工单解析器,chat_models.py:2512-2537)
- include_raw=True 返回什么?({'raw','parsed','parsing_error'} 三件套,解析失败不炸链,chat_models.py:2529)
✋ 10 分钟动手
cd /Users/bitmart/work/codes/github/AI_WORK/langchain/libs/core/langchain_core
# 1. 三个质检员
sed -n '8,35p' output_parsers/string.py # StrOutputParser
sed -n '61,92p' output_parsers/json.py # parse_result:抠 JSON + partial
sed -n '19,45p' output_parsers/pydantic.py # 按图纸验收
# 2. 结构化输出的三板斧
sed -n '2502,2538p' language_models/chat_models.py # bind_tools + tool_choice + parser
grep -n "def with_structured_output" language_models/chat_models.py
# 3. 兜底积木
grep -n "with_fallbacks\|RunnableMap(raw" language_models/chat_models.py | head
documents/ 的 Document(page_content + metadata)和 document_loaders/ 的 BaseLoader(load / lazy_load 为什么强调懒加载)。