Day 08 / 共 20 天 · 阶段2 模型·消息·提示·输出

输出解析与结构化输出: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 全链贯通。

📍 你在 20 天里的位置(阶段2:模型·消息·提示·输出 · D05-08)
D05 ChatModel D06 消息体系 D07 Prompt 模板 D08 输出解析 S3 数据与RAG S4 工具/Agent S5 进阶收官
💡 先用两个类比兜住今天 类比一:OutputParser 是流水线末端的质检员——模型是个才华横溢但不拘小节的工匠,交付的活儿(AIMessage)得经质检员整理成合格出厂品:StrOutputParser 只撕掉包装(取纯文本);JsonOutputParser 从一堆刨花里抠出零件(从 markdown 里抽 JSON);PydanticOutputParser 还要按图纸逐项验收(schema 校验),不合格就打回。类比二:with_structured_output 干脆换了思路——别让工匠自由发挥再验收,直接发一张结构化表单让他填(借用工具调用协议),从源头上杜绝"作文跑题"。
L01

痛点:模型只会写作文,业务要的是数据

🤔 痛点你想让模型抽取 {"name": ..., "age": ...},在提示词里千叮万嘱"只输出 JSON"。结果模型回:"好的!以下是您要的 JSON:```json {...} ```希望对您有帮助!"——json.loads 当场爆炸。就算这次抠出来了,下次它又漏个字段、写错个类型。把"自然语言 → 程序数据"这最后一公里踩实,是 LLM 应用工程化绕不开的坎。
💡 本质:解析器也是 Runnable,是链条的第三节车厢所有 OutputParser 都继承自 Runnable(基类在 output_parsers/base.py),所以能直接 prompt | model | parser 接进 LCEL。它们做三件事:①转换(消息→目标类型)、②容错(从 markdown 围栏里抢救 JSON)、③反向指导get_format_instructions() 生成"请按此格式输出"的说明,可以注入回提示词——质检员提前把验收标准贴在车间墙上)。
解析器位置输入 → 输出
StrOutputParseroutput_parsers/string.py:8AIMessage → str
JsonOutputParseroutput_parsers/json.py:31AIMessage → dict/list(免函数调用最可靠)
PydanticOutputParseroutput_parsers/pydantic.py:19AIMessage → Pydantic 对象(带校验)
with_structured_outputlanguage_models/chat_models.py:2357不是解析器,是"模型包装":从源头逼填表
L02

StrOutputParser:最简单的质检员

用得最多、也最简单的 StrOutputParserlibs/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% 的纯文本场景就用它。
L03

JsonOutputParser:从作文里抠 JSON

JsonOutputParserlibs/core/langchain_core/output_parsers/json.py:31)的核心在 parse_resultjson.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)识别的信号。
L04

PydanticOutputParser:按图纸验收

dict 还不够——业务想要强类型对象PydanticOutputParserlibs/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_exceptionpydantic.py:39)校验失败时把"哪个模型、拿到什么 JSON、错在哪"拼成一条人话错误:Failed to parse Person from completion {...}. Got: 1 validation error...
💡 取舍:提示词哀求 vs 协议强制Str/Json/Pydantic 三兄弟都属于"事后质检"路线:靠提示词请求格式 + 事后解析校验,优点是任何模型都能用(哪怕不支持工具调用);缺点是模型不听话就只能报错重试。下一节的 with_structured_output 是"事前强制"路线:利用工具调用协议让 API 层面保证结构,代价是要求模型支持 bind_tools能用协议就用协议,协议不可用才退回哀求。
L05

with_structured_output:不写作文,直接填表

压轴戏在 BaseChatModel.with_structured_outputlibs/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
PydanticToolsParsertool_calls[0].args 里取出 dict、用你的 Pydantic 类校验实例化。first_tool_only=True:只取第一张工单。
return llm | output_parser★整个"结构化输出"就是一条两节 LCEL 链:绑了工具的模型 | 工单解析器。没有任何魔法,全是 D03-D07 学过的积木。
两条路线:事后质检 vs 事前强制(with_structured_output) 路线A 事后质检 提示词哀求出JSON 模型自由作文 "好的!```json…```" parse_json_markdown 抠 + Pydantic 验收,可能打回 路线B 事前强制 schema 伪装成工具 bind_tools([schema]) tool_choice="any" 必须填表,不许作文 tool_calls[0].args API 协议保证合法 JSON Pydantic 对象 路线B = chat_models.py:2512 bind_tools + :2537 return llm | output_parser —— 就是一条 LCEL 链
图注:路线 A 靠质检员事后抢救;路线 B 从源头发表单。生产环境优先 B,模型不支持工具调用时才退回 A。
L06

include_raw:解析失败怎么兜底

🤔 痛点就算走路线 B,Pydantic 校验也可能失败(比如模型把年龄填成 "十八")。默认直接抛异常——但有时你想"失败了也把原始回复给我,我自己处理"。

这就是 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_noneparsed=None,异常对象放进 parsing_error 字段。于是无论成败,你都拿到 {"raw":…, "parsed":…, "parsing_error":…} 三件套,永不炸链。
💡 本质:兜底也是用积木搭的注意这段实现没写一行 try/except——全靠 RunnablePassthrough.assignwith_fallbacksRunnableMap 这些 LCEL 通用积木组合出来(D18 会专门拆这些积木)。框架自己也在用自己的积木盖楼,这是验证抽象好坏的金标准。
L07

串起来 + 今日小结

📝 真实值:源码 docstring 里的完整例子 这是 chat_models.py:2416 docstring 原例(略去 import):
class AnswerWithJustification(BaseModel): answer: str; justification: str
structured_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
明日预告 · Day 09:阶段2 完结,进入阶段3"数据与 RAG"。模型要回答"你公司文档里的问题",得先把文档搬进来——明天看 documents/Document(page_content + metadata)和 document_loaders/BaseLoader(load / lazy_load 为什么强调懒加载)。
← Day 07 Prompt 模板 Day 09 · Document 与加载 →