LangChain 设计与实现

第7章 输出解析与结构化输出

作者 杨艺韬 · 9,199 字 · 发布于 · 更新于

语言模型的原始输出是自由格式的文本流。对于构建可靠的 AI 应用而言，仅仅拿到一段文字是远远不够的 -- 我们需要将模型的非结构化响应转换为程序可以直接操作的结构化数据。这一需求催生了 LangChain 的输出解析系统。

从最简单的字符串透传，到 JSON/Pydantic 对象的精确验证，再到 XML 流式增量解析，LangChain 构建了一套层次分明、功能递进的输出解析器体系。更重要的是，它巧妙地将解析器统一到 Runnable 协议之下，使得解析操作能够天然地融入 LCEL 链式调用。本章将深入解析这套体系的设计思路与实现细节。

本章要点

  • 理解 BaseLLMOutputParser -> BaseOutputParser -> BaseTransformOutputParser -> BaseCumulativeTransformOutputParser 的四层继承体系
  • 掌握 StrOutputParser、JsonOutputParser、PydanticOutputParser、XMLOutputParser 四大核心解析器的实现原理
  • 理解流式 Transform 解析的累积与增量差异机制
  • 了解 ListOutputParser 族的流式分块解析策略
  • 理解 with_structured_output 与传统输出解析器的关系

7.1 从 Generation 到结构化数据：解析器的使命

语言模型的每一次调用，无论是补全模型（LLM）还是聊天模型（ChatModel），最终都会产出 Generation 对象。Generation 封装了模型输出的原始文本（text 属性）及其元数据。而输出解析器的核心使命，就是将这些 Generation 对象转换为应用程序所需的结构化类型。

这里有一个关键的设计考量：解析器需要同时支持两种输入 -- 原始字符串和 BaseMessage 对象。因为在 LCEL 链中，解析器既可能直接接收模型的文本输出，也可能接收 AIMessage 等消息对象。LangChain 通过在 invoke 方法中进行类型分发来解决这一问题：

# libs/core/langchain_core/output_parsers/base.py:136
class BaseOutputParser(BaseLLMOutputParser, RunnableSerializable[LanguageModelOutput, T]):
    def invoke(
        self,
        input: str | BaseMessage,
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> T:
        if isinstance(input, BaseMessage):
            return self._call_with_config(
                lambda inner_input: self.parse_result(
                    [ChatGeneration(message=inner_input)]
                ),
                input, config, run_type="parser",
            )
        return self._call_with_config(
            lambda inner_input: self.parse_result([Generation(text=inner_input)]),
            input, config, run_type="parser",
        )

这段代码揭示了一个精巧的设计：无论输入是字符串还是消息，都会被包装成 Generation 或 ChatGeneration，然后统一交给 parse_result 处理。_call_with_config 确保了回调系统的正确触发，run_type="parser" 则让 LangSmith 等可观测工具能够正确识别这是一次解析操作。

7.2 解析器类层级结构

LangChain 的输出解析器采用了经典的模板方法模式，构建了四层继承体系。每一层都引入了特定的能力维度：

classDiagram
    class BaseLLMOutputParser~T~ {
        <<abstract>>
        +parse_result(result: list~Generation~, partial: bool) T
        +aparse_result(result: list~Generation~, partial: bool) T
    }
    class BaseGenerationOutputParser~T~ {
        +invoke(input, config) T
        +ainvoke(input, config) T
    }
    class BaseOutputParser~T~ {
        <<abstract>>
        +parse(text: str) T*
        +parse_result(result, partial) T
        +parse_with_prompt(completion, prompt) Any
        +get_format_instructions() str
        +invoke(input, config) T
    }
    class BaseTransformOutputParser~T~ {
        +_transform(input: Iterator) Iterator~T~
        +_atransform(input: AsyncIterator) AsyncIterator~T~
        +transform(input, config) Iterator~T~
        +atransform(input, config) AsyncIterator~T~
    }
    class BaseCumulativeTransformOutputParser~T~ {
        +diff: bool
        +_diff(prev, next) T
        +_transform(input) Iterator~Any~
        +_atransform(input) AsyncIterator~T~
    }
    BaseLLMOutputParser <|-- BaseGenerationOutputParser
    BaseLLMOutputParser <|-- BaseOutputParser
    BaseOutputParser <|-- BaseTransformOutputParser
    BaseTransformOutputParser <|-- BaseCumulativeTransformOutputParser

7.2.1 BaseLLMOutputParser -- 最底层的协议

BaseLLMOutputParser（langchain-core-1.2.26/libs/core/langchain_core/output_parsers/base.py:30）是整个体系的根基，它定义了一个极简接口：唯一的抽象方法 parse_result（同文件 :34）接收 list[Generation]，返回类型 T。它是泛型的，声明为 class BaseLLMOutputParser(ABC, Generic[T])。注意它并不继承 Runnable，因此不能直接用于 LCEL 链——上图里的 BaseGenerationOutputParser（langchain-core-1.2.26/libs/core/langchain_core/output_parsers/base.py:70）之所以有 invoke，是因为它同时混入了 RunnableSerializable（为保持图形简洁，这条继承边没画出来）。

7.2.2 BaseOutputParser -- 核心抽象层

BaseOutputParser（langchain-core-1.2.26/libs/core/langchain_core/output_parsers/base.py:136）同时继承了 BaseLLMOutputParser 和 RunnableSerializable，这是一个关键的设计决策。通过混入 RunnableSerializable，解析器获得了完整的 Runnable 协议支持：invoke/ainvoke/batch/stream 等方法，使其能够作为 LCEL 链中的一个环节。

它引入了三个核心方法（只有 parse 是 @abstractmethod）：

方法 职责 备注
parse(text) 将字符串解析为目标类型 @abstractmethod，子类必须实现（langchain-core-1.2.26/libs/core/langchain_core/output_parsers/base.py:267）
get_format_instructions() 返回格式指令字符串 基类直接 raise NotImplementedError（langchain-core-1.2.26/libs/core/langchain_core/output_parsers/base.py:330），需要格式指令的子类才覆盖
parse_result(result, partial) 从 Generation 列表提取并解析 默认取第一个 Generation 的 text（langchain-core-1.2.26/libs/core/langchain_core/output_parsers/base.py:246）

parse_result 的默认实现非常简洁：return self.parse(result[0].text)。它总是取候选列表中的第一个结果（通常是最高置信度的输出），然后委托给 parse 方法。

7.2.3 BaseTransformOutputParser -- 流式能力

BaseTransformOutputParser（langchain-core-1.2.26/libs/core/langchain_core/output_parsers/transform.py:28）在 BaseOutputParser 的基础上增加了 _transform（同文件 :31）和 _atransform 方法，使解析器能够处理流式输入。其默认实现是"逐块独立解析"策略：

# libs/core/langchain_core/output_parsers/transform.py:28
class BaseTransformOutputParser(BaseOutputParser[T]):
    def _transform(self, input: Iterator[str | BaseMessage]) -> Iterator[T]:
        for chunk in input:
            if isinstance(chunk, BaseMessage):
                yield self.parse_result([ChatGeneration(message=chunk)])
            else:
                yield self.parse_result([Generation(text=chunk)])

这种策略适用于那些每个 chunk 都是完整可解析单元的场景。对于需要累积 chunk 才能解析的场景，就需要下一层的 BaseCumulativeTransformOutputParser。

7.2.4 BaseCumulativeTransformOutputParser -- 累积解析

这是解析器体系中最复杂的一层（langchain-core-1.2.26/libs/core/langchain_core/output_parsers/transform.py:99，diff 字段在 :102、_diff 在 :107、_transform 在 :126）。它的核心思想是：不断累积输入 chunk，每次累积后尝试解析，只在解析结果发生变化时才 yield 输出。

# libs/core/langchain_core/output_parsers/transform.py:99（简化）
class BaseCumulativeTransformOutputParser(BaseTransformOutputParser[T]):
    diff: bool = False

    def _transform(self, input: Iterator[str | BaseMessage]) -> Iterator[Any]:
        prev_parsed = None
        acc_gen = None
        for chunk in input:
            # 将 chunk 包装为 GenerationChunk 并累积
            if isinstance(chunk, BaseMessageChunk):
                chunk_gen = ChatGenerationChunk(message=chunk)
            else:
                chunk_gen = GenerationChunk(text=chunk)

            acc_gen = chunk_gen if acc_gen is None else acc_gen + chunk_gen

            # 尝试解析累积结果
            parsed = self.parse_result([acc_gen], partial=True)
            if parsed is not None and parsed != prev_parsed:
                if self.diff:
                    yield self._diff(prev_parsed, parsed)
                else:
                    yield parsed
                prev_parsed = parsed

这段代码有几个精妙之处：

  1. 累积机制：利用 GenerationChunk 的 __add__ 运算符进行文本拼接，实现增量累积
  2. 变化检测：通过 parsed != prev_parsed 过滤掉重复的中间结果，避免不必要的输出
  3. 差异模式：当 diff=True 时，不输出完整的解析结果，而是输出前后两次解析结果的差异
flowchart TD
    A[输入 chunk 流] --> B{chunk 类型?}
    B -->|BaseMessageChunk| C[包装为 ChatGenerationChunk]
    B -->|str| D[包装为 GenerationChunk]
    C --> E[累积: acc_gen = acc_gen + chunk_gen]
    D --> E
    E --> F[parse_result with partial=True]
    F --> G{parsed 有变化?}
    G -->|否| A
    G -->|是| H{diff 模式?}
    H -->|否| I[yield parsed]
    H -->|是| J[yield _diff prev, parsed]
    I --> K[更新 prev_parsed]
    J --> K
    K --> A

7.3 StrOutputParser：最简单却最常用的解析器

StrOutputParser 是整个体系中最简单的实现 -- 它的 parse 方法只有一行代码：return text。但这并不意味着它不重要。恰恰相反，它是 LCEL 链中使用频率最高的解析器。

# langchain_core/output_parsers/string.py
class StrOutputParser(BaseTransformOutputParser[str]):
    @override
    def parse(self, text: str) -> str:
        """Returns the input text with no changes."""
        return text

它继承自 BaseTransformOutputParser[str]，因此天然支持流式操作。当你在 LCEL 链中写 chain = prompt | llm | StrOutputParser() 时，StrOutputParser 的作用是将模型返回的 AIMessage 对象"剥壳"，提取其中的文本内容。

因为 BaseTransformOutputParser._transform 的默认实现是逐块解析，而 StrOutputParser.parse 只是透传文本，所以在流式场景下，每个 token chunk 都会被原样 yield 出来。这使得 StrOutputParser 在流式场景下几乎是零开销的。

# 典型用法：流式提取文本
from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
# model 为任一聊天模型实例（如 ChatOpenAI）
stream = model.stream("Tell me a story")
for chunk in parser.transform(stream):
    print(chunk, end="", flush=True)  # 逐 token 输出

7.4 JsonOutputParser：结构化 JSON 输出

JsonOutputParser 是实现结构化输出最直接的方式之一。它继承自 BaseCumulativeTransformOutputParser，支持流式增量解析 JSON。

7.4.1 核心解析逻辑

# libs/core/langchain_core/output_parsers/json.py:31（docstring 省略；
# pydantic_object 的注解原为 Annotated[..., SkipValidation()]，此处简化）
class JsonOutputParser(BaseCumulativeTransformOutputParser[Any]):
    pydantic_object: type[TBaseModel] | None = None

    def parse_result(self, result: list[Generation], *, partial: bool = False) -> Any:
        text = result[0].text.strip()
        if partial:
            try:
                return parse_json_markdown(text)
            except JSONDecodeError:
                return None
        else:
            try:
                return parse_json_markdown(text)
            except JSONDecodeError as e:
                msg = f"Invalid json output: {text}"
                raise OutputParserException(msg, llm_output=text) from e

partial 参数的引入是流式解析的关键。在流式模式下（partial=True），JSON 解析失败不会抛出异常，而是返回 None。这允许解析器在 JSON 字符串尚未完整时优雅地等待更多数据。parse_json_markdown 函数能够处理被 Markdown 代码块包裹的 JSON，这在实际应用中非常常见。

容易被忽略的是：parse_json_markdown 的默认 parser 就是 parse_partial_json（libs/core/langchain_core/utils/json.py:58、:142），非流式的 partial=False 分支用的也是它。它的容错规则如下：

  1. 先原样 json.loads，成功就直接返回；
  2. 否则逐字符扫描，记录是否在字符串内、以及未闭合的 {/[ 栈；字符串内未转义的换行替换成 \n；遇到与栈顶不匹配的 }/] 直接返回 None；
  3. 扫描结束仍在字符串内，就去掉末尾悬空的反斜杠、补上 "；
  4. 按栈逆序补齐右括号后尝试解析，失败就从末尾删一个字符再试，直到成功；删光仍不行，才对原串 json.loads 抛出 JSONDecodeError。

parse_json_markdown 外层还有一步：整段（去掉首尾空白与反引号）解析失败时，用正则 ```(json)?(.*) 取第一个代码围栏之后的全部内容再解析一次。在 pin 环境实跑 JsonOutputParser().parse(...)：'{"a": 1, "b": "hel' 得到 {'a': 1, 'b': 'hel'}，'[1, 2, ' 得到 [1, 2]，'{"a": tru' 得到 {}，'{"a": 1} trailing' 得到 {'a': 1}，都不抛异常；括号错配的 '{"a": [1}' 则静默返回 None；而 'Here is: {"a": 1}' 这种开头不是 JSON、又没有代码围栏的文本才会抛 OutputParserException。所以非流式模式下拿到截断的 JSON 不会报错，下游若要求字段齐全，应改用 PydanticOutputParser 或自行校验。

7.4.2 差异模式与 JSONPatch

当 diff=True 时，JsonOutputParser 利用 jsonpatch 库生成 JSONPatch 操作序列：

def _diff(self, prev: Any | None, next: Any) -> Any:
    return jsonpatch.make_patch(prev, next).patch

在 pin 环境用 GenericFakeChatModel（按空白切块）流式输出 {"title": "Inception", "rating": 9.5, "tags": ["sci-fi", "heist"]}，接 JsonOutputParser() 与 JsonOutputParser(diff=True) 的逐块输出分别是：

# JsonOutputParser()
{}
{'title': 'Inception'}
{'title': 'Inception', 'rating': 9.5}
{'title': 'Inception', 'rating': 9.5, 'tags': ['sci-fi']}
{'title': 'Inception', 'rating': 9.5, 'tags': ['sci-fi', 'heist']}

# JsonOutputParser(diff=True)
[{'op': 'replace', 'path': '', 'value': {}}]
[{'op': 'add', 'path': '/title', 'value': 'Inception'}]
[{'op': 'add', 'path': '/rating', 'value': 9.5}]
[{'op': 'add', 'path': '/tags', 'value': ['sci-fi']}]
[{'op': 'add', 'path': '/tags/1', 'value': 'heist'}]

第一块 {"title": 被删到只剩 { 再补齐，所以先吐出 {}。如果切块更细，半截的字符串值和数字也会被吐出：输入 '{"ti'、'tle": "Inc'、'ep'、'tion", "r'、'ating": 9'、'.5}' 时，输出依次是 {}、{'title': 'Inc'}、{'title': 'Incep'}、{'title': 'Inception'}、{..., 'rating': 9}、{..., 'rating': 9.5}。

这对于前端实时渲染部分 JSON 结果特别有用 -- 前端可以直接应用 JSONPatch 操作来增量更新 UI，而不需要每次都处理完整的 JSON 对象。

7.4.3 格式指令生成

get_format_instructions 方法根据是否提供了 pydantic_object 来生成不同的指令：

# libs/core/langchain_core/output_parsers/json.py:104（JsonOutputParser；
# 同名方法在 base.py:330 / pydantic.py:93 / xml.py:202 / list.py:157 各有一份，
# 这里是 JSON 这一份）—— docstring 与行内注释已省略
def get_format_instructions(self) -> str:
    if self.pydantic_object is None:
        return "Return a JSON object."
    schema = dict(self._get_schema(self.pydantic_object).items())
    reduced_schema = schema
    if "title" in reduced_schema:
        del reduced_schema["title"]
    if "type" in reduced_schema:
        del reduced_schema["type"]
    schema_str = json.dumps(reduced_schema, ensure_ascii=False)
    return JSON_FORMAT_INSTRUCTIONS.format(schema=schema_str)

这个方法会将 Pydantic 模型的 JSON Schema 嵌入到格式指令中，告诉模型应当输出什么样的 JSON 结构。指令文本中特别强调不要使用 Markdown 代码块包裹，确保输出可以被直接解析。

flowchart LR
    subgraph 非流式
        A1[模型输出文本] --> B1[parse_json_markdown]
        B1 --> C1{解析成功?}
        C1 -->|是| D1[返回 dict/list]
        C1 -->|否| E1[抛出 OutputParserException]
    end
    subgraph 流式累积
        A2[chunk 流] --> B2[累积文本]
        B2 --> C2[parse_json_markdown]
        C2 --> D2{解析成功?}
        D2 -->|是| E2{结果有变化?}
        D2 -->|否| F2[返回 None, 继续累积]
        E2 -->|是| G2{diff 模式?}
        E2 -->|否| F2
        G2 -->|否| H2[yield 完整 JSON]
        G2 -->|是| I2[yield JSONPatch]
    end

7.5 PydanticOutputParser：类型安全的解析

PydanticOutputParser 继承自 JsonOutputParser，在 JSON 解析的基础上增加了 Pydantic 模型验证。它是类型安全的结构化输出解析器的代表。

# langchain_core/output_parsers/pydantic.py（简化：省略两种版本都不匹配时抛 OutputParserException 的分支）
class PydanticOutputParser(JsonOutputParser, Generic[TBaseModel]):
    pydantic_object: type[TBaseModel]

    def _parse_obj(self, obj: dict) -> TBaseModel:
        try:
            if issubclass(self.pydantic_object, pydantic.BaseModel):
                return self.pydantic_object.model_validate(obj)
            if issubclass(self.pydantic_object, pydantic.v1.BaseModel):
                return self.pydantic_object.parse_obj(obj)
        except (pydantic.ValidationError, pydantic.v1.ValidationError) as e:
            raise self._parser_exception(e, obj) from e

    def parse_result(self, result: list[Generation], *, partial: bool = False):
        try:
            json_object = super().parse_result(result)
            return self._parse_obj(json_object)
        except OutputParserException:
            if partial:
                return None
            raise

设计亮点在于：

  1. 双版本兼容：同时支持 Pydantic v1（parse_obj）和 v2（model_validate），这是 LangChain 在 Pydantic 版本迁移期间的务实选择
  2. 优雅的流式降级：在 partial=True 模式下，验证失败返回 None 而非抛出异常；反过来，只要累积出的部分 JSON 恰好能通过校验就会产出对象。实跑中按 '..."r'、'ating": 9'、'.5}' 切块时，先产出 rating=9.0，下一块再产出 rating=9.5，所以流式时拿到的第一个对象未必是最终值
  3. 继承复用：通过继承 JsonOutputParser，自动获得了 JSON 解析和流式累积的全部能力

典型用法：

from pydantic import BaseModel
from langchain_core.output_parsers import PydanticOutputParser

class MovieReview(BaseModel):
    title: str
    rating: float
    summary: str

parser = PydanticOutputParser(pydantic_object=MovieReview)

# 获取格式指令，注入到 Prompt 中
instructions = parser.get_format_instructions()
# 解析模型输出
result: MovieReview = parser.parse('{"title": "Inception", "rating": 9.5, "summary": "A mind-bending thriller"}')

7.6 XMLOutputParser：XML 格式的流式解析

XMLOutputParser 是一个独具特色的解析器。与 JSON 不同，XML 的流式解析可以利用 SAX/Pull 解析器的事件驱动特性，在每个叶子标签闭合时就产出结果，而不需要等待整个文档完成。

7.6.1 流式解析架构

XMLOutputParser 使用了一个内部的 _StreamingParser 类来封装流式 XML 解析的状态：

# langchain_core/output_parsers/xml.py（省略 parser_ 的构造：defusedxml 时为
# XMLParser(target=TreeBuilder())，否则为 None）
class _StreamingParser:
    def __init__(self, parser: Literal["defusedxml", "xml"]) -> None:
        self.pull_parser = ET.XMLPullParser(["start", "end"], _parser=parser_)
        self.xml_start_re = re.compile(r"<[a-zA-Z:_]")
        self.current_path: list[str] = []
        self.current_path_has_children = False
        self.buffer = ""
        self.xml_started = False

关键状态包括：

  • pull_parser：Python 标准库的 XMLPullParser，支持增量喂入数据
  • current_path：维护当前 XML 路径（如 ["root", "item"]），用于构建嵌套结构
  • buffer：未处理的文本缓冲区
  • xml_started：标记是否已经遇到了第一个 XML 标签
stateDiagram-v2
    [*] --> WaitingForXml: 初始化
    WaitingForXml --> WaitingForXml: 非 XML 文本
    WaitingForXml --> Parsing: 检测到 XML 起始标签
    Parsing --> Parsing: start 事件 - 压入 path
    Parsing --> YieldElement: end 事件 - 叶子节点
    YieldElement --> Parsing: 继续解析
    Parsing --> WaitingForXml: 根标签闭合
    WaitingForXml --> [*]: 流结束

7.6.2 安全考量

XML 解析器默认使用 defusedxml 库，以防御 XML 炸弹（Billion Laughs Attack）等安全威胁：

class XMLOutputParser(BaseTransformOutputParser):
    parser: Literal["defusedxml", "xml"] = "defusedxml"

这是一个值得注意的安全设计决策。在处理不可信的 LLM 输出时，标准库的 XML 解析器可能受到某些 XML 漏洞的攻击，而 defusedxml 默认启用了所有安全防护。

但 defusedxml 并不在 langchain-core 的依赖列表里（libs/core/pyproject.toml 的 dependencies 只有 langsmith、tenacity、jsonpatch、PyYAML 等 8 项）。没装它时，默认构造的 XMLOutputParser() 一调用 parse 或流式 _transform 就抛 ImportError: defusedxml is not installed...（libs/core/langchain_core/output_parsers/xml.py:62-69、:223-231，pin 环境实跑确认）。要么 pip install defusedxml，要么在确认运行环境安全时显式传 parser="xml"。

7.6.3 嵌套结果构建

流式解析产出的结果使用 AddableDict 类型，支持通过 + 运算符合并：

def nested_element(path: list[str], elem: ET.Element) -> Any:
    if len(path) == 0:
        return AddableDict({elem.tag: elem.text})
    return AddableDict({path[0]: [nested_element(path[1:], elem)]})

这个递归函数将 XML 路径和元素转换为嵌套字典。AddableDict 的可加性使得多次 yield 的结果可以被上层（如 LCEL 的 stream 聚合器）正确地合并在一起。以 parser="xml" 流式解析 <movie> <title>Inception</title> <cast> <actor>Leo</actor> <actor>Tom</actor> </cast> </movie>，实跑逐块输出为：

{'movie': [{'title': 'Inception'}]}
{'movie': [{'cast': [{'actor': 'Leo'}]}]}
{'movie': [{'cast': [{'actor': 'Tom'}]}]}

只有叶子元素闭合时产出，</cast>、</movie> 这类父元素闭合不单独产出；整条链 invoke 时这三块相加，结果与一次性 parse 相同：{'movie': [{'title': 'Inception'}, {'cast': [{'actor': 'Leo'}, {'actor': 'Tom'}]}]}。

7.7 ListOutputParser 族：列表解析

LangChain 提供了三种列表解析器，均继承自 ListOutputParser：

解析器 格式 分隔模式
CommaSeparatedListOutputParser foo, bar, baz CSV 解析
NumberedListOutputParser 1. foo\n2. bar 正则 \d+\.\s([^\n]+)
MarkdownListOutputParser - foo\n- bar 正则 ^\s*[-*]\s([^\n]+)$

7.7.1 流式列表解析的 droplastn 策略

列表解析器的流式实现面临一个独特挑战：在文本持续到达的过程中，最后一个元素可能是不完整的。解决方案是一个巧妙的 droplastn 函数：

# langchain_core/output_parsers/list.py
def droplastn(iter: Iterator[T], n: int) -> Iterator[T]:
    """Drop the last n elements of an iterator."""
    buffer: deque[T] = deque()
    for item in iter:
        buffer.append(item)
        if len(buffer) > n:
            yield buffer.popleft()

这个函数使用一个长度为 n 的 deque 缓冲区，总是延迟 n 个元素输出。在列表解析的流式场景中，n=1 意味着最后一个匹配项被暂时保留（因为它可能是不完整的），只有当下一个匹配项到达后，前一个才会被确认输出。

# libs/core/langchain_core/output_parsers/list.py:73
# ListOutputParser._transform 中的核心逻辑（简化：省略 chunk 为 BaseMessage 时
# 取 content、非字符串 content 直接跳过的分支）
def _transform(self, input: Iterator[str | BaseMessage]) -> Iterator[list[str]]:
    buffer = ""
    for chunk in input:
        buffer += chunk
        try:
            done_idx = 0
            for m in droplastn(self.parse_iter(buffer), 1):
                done_idx = m.end()
                yield [m.group(1)]
            buffer = buffer[done_idx:]
        except NotImplementedError:
            # 回退到非流式解析
            parts = self.parse(buffer)
            if len(parts) > 1:
                for part in parts[:-1]:
                    yield [part]
                buffer = parts[-1]
    # 流结束，输出最后的缓冲内容
    for part in self.parse(buffer):
        yield [part]

这段代码的结构非常精巧：优先尝试基于正则迭代器的流式解析（parse_iter），如果子类未实现，则回退到基于完整解析的分块策略。两种路径都保持了"最后一个元素延迟输出"的不变量。

flowchart TD
    A[输入 chunk] --> B[追加到 buffer]
    B --> C{parse_iter 实现?}
    C -->|是| D[正则匹配迭代]
    C -->|否| E[parse 完整解析]
    D --> F[droplastn 延迟 1 个]
    F --> G[yield 已确认的匹配项]
    G --> H[截断 buffer 到最后匹配位置]
    E --> I{结果 > 1 项?}
    I -->|是| J[yield 除最后一项外的所有项]
    J --> K[buffer = 最后一项]
    I -->|否| L[保持 buffer 不变]
    H --> M{还有更多 chunk?}
    K --> M
    L --> M
    M -->|是| A
    M -->|否| N[yield buffer 中剩余内容]

7.8 OpenAI Tools/Functions 输出解析器

除了基于文本格式的解析器，LangChain 还提供了专门解析 OpenAI tool_calls 结构的解析器。这些解析器位于 libs/core/langchain_core/output_parsers/openai_tools.py 和 libs/core/langchain_core/output_parsers/openai_functions.py 模块中。

7.8.1 JsonOutputToolsParser

JsonOutputToolsParser 继承自 BaseCumulativeTransformOutputParser，它的输入不是纯文本，而是 AIMessage 中的 tool_calls 字段。其 parse_result 方法直接从 ChatGeneration 的消息对象中提取结构化数据：

# 简化的核心逻辑
def parse_result(self, result, *, partial=False):
    generation = result[0]
    if not isinstance(generation, ChatGeneration):
        raise OutputParserException(
            "This output parser can only be used with a chat generation.")
    message = generation.message
    if isinstance(message, AIMessage) and message.tool_calls:
        tool_calls = message.tool_calls
        # 从 tool_calls 中提取参数（return_id=False 时去掉 id）
        return [{"type": tc["name"], "args": tc["args"]} for tc in tool_calls]
    # 省略：没有 tool_calls 时回退解析 additional_kwargs["tool_calls"]；first_tool_only 处理

这种解析方式比文本解析更加可靠：tool_calls 的参数已在消息层解析成 dict，省去了从文本里抠 JSON 的步骤。但它并非没有失败的可能——回退到 additional_kwargs["tool_calls"] 时，参数不是合法 JSON 仍会抛 OutputParserException（libs/core/langchain_core/output_parsers/openai_tools.py:62-70）。流式时它同样靠累积：实跑中三个 tool_call_chunks（参数分别为 '{"ti'、'tle": "Incep'、'tion", "rating": 9.5}'）依次产出 args 为 {}、{'title': 'Incep'}、{'title': 'Inception', 'rating': 9.5} 的结果。

7.8.2 PydanticToolsParser

PydanticToolsParser 在 JsonOutputToolsParser 的基础上增加了 Pydantic 模型验证，将原始字典转换为类型安全的 Pydantic 实例。它接受一个 tools 参数，包含一个或多个 Pydantic 模型类，并根据 tool_call 的 name 字段选择对应的模型进行验证。

7.8.3 两种解析路径的对比

flowchart LR
    subgraph 文本解析路径
        A1[模型输出文本] --> B1[JsonOutputParser]
        B1 --> C1["parse_json_markdown(text)"]
        C1 --> D1[dict]
    end
    subgraph Tool Call 解析路径
        A2[模型输出 tool_calls] --> B2[JsonOutputToolsParser]
        B2 --> C2["message.tool_calls"]
        C2 --> D2["list of dict"]
    end

文本解析路径需要处理 Markdown 代码块、不完整 JSON 等各种边界情况；Tool Call 解析路径则直接从 API 结构中提取数据，省去了文本解析的复杂性。在使用支持 function calling 的模型时，Tool Call 路径是更推荐的选择。

7.9 with_structured_output：模型层面的结构化输出

LangChain 在模型层面提供了 with_structured_output 方法，这是一种更现代、更推荐的结构化输出方式：

# libs/core/langchain_core/language_models/chat_models.py:1604（方法本体；
# BaseChatModel 类声明在同文件 :250，下面只是把方法放回类里便于阅读）
class BaseChatModel:
    def with_structured_output(
        self,
        schema: builtins.dict[str, Any] | type,
        *,
        include_raw: bool = False,
        **kwargs: Any,
    ) -> Runnable[LanguageModelInput, builtins.dict[str, Any] | BaseModel]:
        ...

with_structured_output 返回一个新的 Runnable，它将模型的结构化输出能力（如 function calling、tool use、JSON mode）与输出解析组合在一起。与传统的 OutputParser 相比，它有几个显著优势：

flowchart TD
    subgraph 传统方式
        A1[Prompt + 格式指令] --> B1[模型生成文本]
        B1 --> C1[OutputParser 解析文本]
        C1 --> D1{解析成功?}
        D1 -->|否| E1[OutputParserException]
        D1 -->|是| F1[结构化结果]
    end
    subgraph with_structured_output
        A2[Prompt] --> B2[模型直接生成结构化数据]
        B2 --> C2[API 返回 tool_calls]
        C2 --> D2[Pydantic 验证]
        D2 --> F2[结构化结果]
    end

关键区别：

维度 OutputParser 方式 with_structured_output 方式
结构约束 通过 Prompt 指令引导 通过 API 参数强制
可靠性 依赖模型遵循指令 API 强制调用工具，参数仍须校验
流式支持 需要累积解析 同样靠累积解析（tool call 参数的部分 JSON）
适用范围 所有模型 支持 tool calling 的模型
错误处理 解析异常 验证异常

include_raw=True 选项可以同时返回原始消息和解析结果，便于调试和错误处理：

from langchain_openai import ChatOpenAI  # MovieReview 见 §7.5

model = ChatOpenAI(model="gpt-4o")
structured_model = model.with_structured_output(MovieReview, include_raw=True)
result = structured_model.invoke("Review the movie Inception")
# result = {
#     "raw": AIMessage(...),
#     "parsed": MovieReview(title="Inception", ...),
#     "parsing_error": None
# }

7.9.1 基类默认实现的真实组装：tool_choice + Parser + Fallback

前面表格把 OutputParser 和 with_structured_output 并排比，给人感觉这是两套完全不同的机制——其实打开 libs/core/langchain_core/language_models/chat_models.py:1755 的基类默认实现会发现：with_structured_output 内部本质上是 bind_tools + OutputParser 的组合，只是选了一个比裸文本更可靠的解析起点。贴出主体（省略了开头 5 行：静默丢弃 method、strict 两个关键字参数，其余关键字参数一律 ValueError）：

# chat_models.py:1755 （基类默认实现）
if type(self).bind_tools is BaseChatModel.bind_tools:
    raise NotImplementedError(
        "with_structured_output is not implemented for this model.")

llm = self.bind_tools(
    [schema],
    tool_choice="any",
    ls_structured_output_format={
        "kwargs": {"method": "function_calling"},
        "schema": schema,
    },
)

if isinstance(schema, type) and is_basemodel_subclass(schema):
    output_parser = PydanticToolsParser(
        tools=[cast("TypeBaseModel", schema)], first_tool_only=True
    )
else:
    key_name = convert_to_openai_tool(schema)["function"]["name"]
    output_parser = JsonOutputKeyToolsParser(
        key_name=key_name, first_tool_only=True
    )

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

四层结构拆开看：

1. bind_tools([schema], tool_choice="any") 是强制。tool_choice="any" 告诉模型"下次回复必须调用 tools 列表里的某个工具"——而列表里只有一个工具（就是用户传入的 schema）、所以模型别无选择只能按这个 schema 输出。这就是原章节"通过 API 参数强制"那句抽象描述的具体机制：不是 API 层有特殊 flag、而是利用 tool_calling 协议的"必选其一"语义化成"必选这一个"。

2. 解析器按 schema 类型分流：

  • Pydantic 类 → PydanticToolsParser(first_tool_only=True)：直接拿第一个 tool_call 的 arguments、用 Pydantic 校验实例化；验证失败时原样抛出 Pydantic 的 ValidationError（libs/core/langchain_core/output_parsers/openai_tools.py:368-381，实跑确认），只有工具名对不上时才抛 OutputParserException。
  • 其他（dict / TypedDict） → JsonOutputKeyToolsParser(key_name=..., first_tool_only=True)：只返回原始 JSON dict、不做类型实例化、key_name 从 convert_to_openai_tool(schema)["function"]["name"] 取出以便从响应里正确路由。

"with_structured_output 用的是 OutputParser"——这句话本身没问题，只不过用的是作用在 tool_calls 字段而非文本的特殊解析器。可靠性提升不来自"API 有什么神奇 flag"，而来自输入形态更规整：tool_calls 已经是结构化字段、解析器不需要做"从 markdown 里抓 JSON 块"这种脆弱的文本处理。

3. include_raw=True 分支用 RunnablePassthrough.assign + with_fallbacks(exception_key="parsing_error")——和第 3 章讲的 RunnableWithFallbacks 的 exception_key 机制是同一个东西。这里的手法：

  • 主路径 parser_assign：执行 output_parser(raw) 成功就写 parsed、parsing_error=None
  • fallback 路径 parser_none：parsed=None，不做解析
  • exception_key="parsing_error" 让解析异常被捕获并挂到 fallback 的输入上

结果是解析阶段的异常不会向外抛（模型调用 raw=llm 在 fallback 之外，它出错照样抛）：成功时返回 {raw, parsed, parsing_error: None}、失败时返回 {raw, parsed: None, parsing_error: <exception>}。这正是 include_raw=True 文档承诺的"出错也不丢信息"——实现由 LCEL 原语组合出来、不需要新语法。

4. bind_tools 未覆盖时的 NotImplementedError（line 1755）——基类的 with_structured_output 依赖 bind_tools 已被具体模型类（ChatOpenAI、ChatAnthropic 等）重写。如果 type(self).bind_tools is BaseChatModel.bind_tools——子类没覆盖——直接报错。type(self).bind_tools 而非 self.bind_tools 是精确到类方法解析的检查，不被子类实例级的临时 monkey-patch 骗过。

这也说明了一件重要事情：具体模型（ChatOpenAI / ChatAnthropic / ChatOllama 等）可以重写 with_structured_output 走更高效的专属路径——比如 OpenAI 可以用 JSON Mode 或 json_schema、Anthropic 可以用 json_schema（见第 5 章 §5.8）、某些 provider 甚至有 schema-enforced constrained decoding。基类只提供一个建立在 bind_tools + Parser 通用组合上的兜底实现——所有支持 tool calling 的模型开箱即用、特定优化留给具体模型的作者自己发挥。这是 LangChain "通用兜底 + 专属超车" 的典型扩展点设计。

7.10 OutputParserException：统一的错误处理

当解析失败时，所有解析器都通过 OutputParserException 报告错误。这个异常类携带了关键的调试信息：

# langchain_core/exceptions.py（签名去掉了类型注解，docstring 省略）
class OutputParserException(ValueError, LangChainException):
    def __init__(self, error, observation=None, llm_output=None, send_to_llm=False):
        if isinstance(error, str):
            error = create_message(
                message=error, error_code=ErrorCode.OUTPUT_PARSING_FAILURE
            )
        super().__init__(error)
        if send_to_llm and (observation is None or llm_output is None):
            msg = (
                "Arguments 'observation' & 'llm_output'"
                " are required if 'send_to_llm' is True"
            )
            raise ValueError(msg)
        self.observation = observation
        self.llm_output = llm_output
        self.send_to_llm = send_to_llm

它同时继承 ValueError 和 LangChainException，所以既能被普通的 except ValueError 捕获，也能与框架内其他异常一起按 LangChainException 归类。构造时如果 error 是字符串，会先套一层带 ErrorCode.OUTPUT_PARSING_FAILURE 的 create_message——错误信息里因此会附上排障文档链接。另外 send_to_llm=True 时 observation 与 llm_output 缺一不可，否则构造本身就抛 ValueError。

属性 含义
observation 可选的人类可读错误描述
llm_output 导致解析失败的原始 LLM 输出
send_to_llm 是否应将错误信息返回给 LLM 进行修正

send_to_llm 标志特别有趣 -- 当设为 True 时，Agent 框架可以将解析错误作为反馈发送给模型，让模型重新生成符合格式要求的输出。在 pin 树里读这个标志的只有 langchain-classic 的 AgentExecutor（libs/langchain/langchain_classic/agents/agent.py:1337）：handle_parsing_errors=True 时，它把异常的 observation 与 llm_output 作为一步 _Exception 动作回喂给模型。另一类"自我修正"是 OutputFixingParser / RetryOutputParser（同样不在 core，而在 langchain-classic 的 langchain_classic/output_parsers/fix.py、retry.py）：它们包装一个解析器，捕获到任何 OutputParserException 就把出错的输出（RetryOutputParser 还带上原始 prompt）交给另一条 LLM 链改写后重解析，默认 max_retries=1，并不看 send_to_llm。注意 RetryOutputParser.parse 直接抛 NotImplementedError，只能通过 parse_with_prompt 调用。

在实际开发中，一个常见的模式是将格式指令和错误反馈组合使用：

from langchain_core.exceptions import OutputParserException
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel

class Answer(BaseModel):
    result: str
    confidence: float

parser = PydanticOutputParser(pydantic_object=Answer)

# 将格式指令注入 Prompt
prompt = f"""Answer the question.
{parser.get_format_instructions()}
Question: What is 2+2?"""

# 如果解析失败，捕获异常并获取原始输出（model_output 为模型返回的文本）
try:
    answer = parser.parse(model_output)
except OutputParserException as e:
    print(f"Failed to parse: {e.llm_output}")

7.11 解析器在 LCEL 链中的位置

理解解析器在典型 LCEL 链中的位置和数据流向，有助于在实践中正确使用它们：

flowchart LR
    A[PromptTemplate] -->|PromptValue| B[ChatModel]
    B -->|AIMessage| C[OutputParser]
    C -->|T| D[下游处理]
    style A fill:#e1f5fe
    style B fill:#fff3e0
    style C fill:#e8f5e9
    style D fill:#f3e5f5

在流式场景下，数据流变为：

PromptTemplate -> ChatModel.stream() -> AIMessageChunk流 -> OutputParser.transform() -> T类型chunk流

每种解析器在流式链中的行为不同：

解析器 流式行为 输出粒度
StrOutputParser 每个 chunk 直接输出 token 级别
JsonOutputParser 累积后输出部分 JSON 每当累积结果变化时（含半截字符串值）
PydanticOutputParser 累积结果能通过校验时输出 完整对象，可能多次（字段值仍在增长）
XMLOutputParser 每个叶子标签闭合时输出 叶子元素级别
ListOutputParser 确认一项完成时输出 列表项级别

这种差异化的流式行为使得开发者可以根据 UI 需求选择合适的解析器。例如，实时显示聊天回复使用 StrOutputParser；渐进式填充表单使用 JsonOutputParser；流式解析结构化配置使用 XMLOutputParser。

7.12 设计决策分析

为什么解析器是 Runnable？

将解析器统一为 Runnable 是 LangChain 设计中最重要的决策之一。这意味着解析器可以通过 | 运算符与其他组件自由组合：

chain = prompt | model | JsonOutputParser()

这不仅带来了语法上的简洁，更重要的是它使得回调、追踪、重试等横切关注点可以统一处理。每次解析操作都会触发 on_chain_start/on_chain_end 回调，LangSmith 可以记录每一次解析的输入输出。

为什么同时支持 parse 和 parse_result？

parse(text: str) 是面向开发者的简洁接口，而 parse_result(result: list[Generation]) 是面向系统内部的通用接口。大多数子类只需要实现 parse，而 parse_result 提供了一个扩展点 -- 某些解析器可能需要从 Generation 的其他字段（如 generation_info）中提取信息。

流式解析的两种范式

LangChain 提供了两种流式解析范式：

  1. 逐块独立解析（BaseTransformOutputParser）：适用于每个 chunk 都是完整可解析单元的场景，如 StrOutputParser
  2. 累积后解析（BaseCumulativeTransformOutputParser）：适用于需要累积完整文本才能解析的场景，如 JsonOutputParser

这种分层设计让子类可以根据自己的语义选择合适的基类，而不需要关心流式调度的细节。

格式指令的局限性

get_format_instructions() 本质上是一种"协作式"约束 -- 它通过自然语言告诉模型应该输出什么格式。这种方式的问题在于模型不一定会遵循指令。LangChain 的解决方案是双管齐下：一方面通过格式指令尽量引导模型，另一方面通过严格的解析逻辑和 OutputParserException 来检测并报告格式错误。with_structured_output 则通过 API 层面的约束（强制工具调用）大幅缓解了这一问题，但工具参数仍要经过校验，校验失败同样会报错（§7.9.1）。

7.12.1 实测：langchain_core/output_parsers/ 11 文件 2227 行

把 output_parsers/ 目录按文件实测——

文件 行 角色
openai_tools.py 384 本目录最大——JsonOutputToolsParser + PydanticToolsParser（§7.8 主角）
base.py 348 BaseLLMOutputParser + BaseOutputParser + BaseGenerationOutputParser 三层抽象（§7.2.1-7.2.2 全部基类）
openai_functions.py 313 OpenAI Function Calling 旧版兼容（openai_tools.py 之前的方案，本章未提）
xml.py 300 XMLOutputParser（§7.6）
list.py 249 CommaSeparatedListOutputParser / NumberedListOutputParser / MarkdownListOutputParser（§7.7）
transform.py 175 BaseTransformOutputParser + BaseCumulativeTransformOutputParser（§7.2.3-7.2.4 流式抽象）
json.py 139 JsonOutputParser（§7.4）
pydantic.py 139 PydanticOutputParser（§7.5）
__init__.py 101 公共 export
string.py 63 StrOutputParser（§7.3）——本目录最薄、印证 §7.3 标题"最简单却最常用"
format_instructions.py 16 通用格式指令模板字符串

两条值得记住的物理事实——

  1. openai_tools.py 384 行 + openai_functions.py 313 行 = 697 行解析工具调用 / 函数调用——是整个目录 31%。文件名沿用 OpenAI 的叫法，但 JsonOutputToolsParser 读的是标准化的 message.tool_calls，任何支持 tool calling 的模型都适用（§7.9.1 的基类默认实现就对所有模型用它）；这些类也同样继承 BaseCumulativeTransformOutputParser，与文本解析器同属一个类体系。openai_functions.py 解析的是 additional_kwargs["function_call"]，即 OpenAI 早期 function calling 协议，本章未展开但仍在 codebase
  2. StrOutputParser 仅 63 行 vs XMLOutputParser 300 行，约 4.8 倍差距——真实工程量因输入复杂度（plain string vs XML 流式 + 嵌套）差别很大——这种"抽象等价、实现差距大"是 §7.2 类层级的具体体现：每种解析器在统一 Runnable 协议下做完全不同复杂度的事

7.12.2 流式解析的源码边界：逐块解析 vs 累积解析

输出解析器之所以值得单独成章，是因为它们不只是 parse(text)。libs/core/langchain_core/output_parsers/base.py:30-67 的最底层接口是 BaseLLMOutputParser.parse_result，输入是 list[Generation]，而不是裸字符串；base.py:136-170 的 BaseOutputParser 又把解析器接入 RunnableSerializable，所以解析器才能放进 LCEL 链里。

流式边界在 transform.py 更清楚。BaseTransformOutputParser 位于 libs/core/langchain_core/output_parsers/transform.py:28-96，它的 _transform 对每个字符串 chunk 或 BaseMessage chunk 立即构造 Generation / ChatGeneration，然后调用 parse_result。这适合 StrOutputParser 这样的解析器：每个 chunk 都可以独立变成输出。

BaseCumulativeTransformOutputParser 位于 libs/core/langchain_core/output_parsers/transform.py:99-175，它维护 acc_gen，把 chunk 累加成 GenerationChunk 或 ChatGenerationChunk，再用 parse_result(..., partial=True) 尝试解析。解析结果和上一次不同才产出；如果 diff=True，还可以输出差量。这就是 JsonOutputParser 能流式吐出部分 JSON 的基础。

基类 状态 适合解析器 风险
BaseTransformOutputParser 默认 _transform 不保留历史 chunk；子类可覆写并自带缓冲 字符串；列表、XML 解析器覆写了 _transform 默认实现不能处理跨 chunk 的结构
BaseCumulativeTransformOutputParser 累积历史 chunk JSON、Pydantic、tool call 解析器 需要处理半截结构和重复输出

JsonOutputParser 的源码验证了这点。libs/core/langchain_core/output_parsers/json.py:31-42 明确说流式模式会 yield 已返回的 partial JSON；json.py:61-91 在 partial=True 时捕获 JSONDecodeError 并返回 None，完整模式下才抛 OutputParserException。PydanticOutputParser 在 pydantic.py:55-80 复用 JSON 解析，再做模型校验；partial 模式下如果解析失败也返回 None。这说明"流式结构化输出"不是模型一次性给出合法 JSON，而是解析器持续尝试，把尚未闭合的结构当作暂时不可用。

工程建议很直接：如果 UI 只需要展示文本，选逐块解析；如果 UI 要渐进填充结构化表单，选累积解析；如果下游必须拿到完整业务对象，不要在 partial 阶段触发不可逆副作用。解析器接入 Runnable 以后，错误和回调会沿 LCEL 传播，这很好，但也意味着一个解析器的 partial 策略会影响整条链的用户体验。

7.13 小结

LangChain 的输出解析系统展现了精心设计的层次结构：从 BaseLLMOutputParser 的最小接口，到 BaseOutputParser 的 Runnable 集成，再到 BaseTransformOutputParser 和 BaseCumulativeTransformOutputParser 的流式支持，每一层都在前一层的基础上增添恰到好处的能力。

核心解析器家族 -- StrOutputParser、JsonOutputParser、PydanticOutputParser、XMLOutputParser、以及 ListOutputParser 族 -- 覆盖了绝大多数结构化输出需求。它们通过统一的 Runnable 协议无缝融入 LCEL 链，通过分层的流式解析策略（逐块 vs 累积）适应不同的解析语义。

with_structured_output 代表的是从"解析模型的文本输出"转向"让模型直接生成结构化数据"的思路；可以预期，随着模型能力的演进，这会成为结构化输出的主流做法（这是预测，不是现状）。但传统的输出解析器在兼容性、灵活性和对非 tool-calling 模型的支持方面，仍然有着不可替代的价值。