跳转至

LangChain Output Parser 深度解析:从结构化提取到自定义容错

在现代大模型(LLM)应用开发中,获取结构化、可预测的输出是连接自然语言与程序逻辑的关键。LangChain 的 Output Parser(输出解析器)框架正是为了解决这一痛点而生。它不仅封装了格式指令的生成,还提供了自动解析、验证、甚至自我修复的完整机制。下面将逐一拆解这些核心问题,深入每一个技术细节。

为什么要使用 Output Parser?它解决了直接从 LLM 获取结构化输出的哪些问题?

语言模型本质上是一个文本生成器,它接收一段提示(Prompt),然后基于概率预测下一个 Token,最终生成一段自由文本。然而,现代应用程序通常需要处理结构化数据(如 JSON、CSV、特定格式的指令)。直接从 LLM 获取结构化输出面临三个核心痛点,而 Output Parser 正是为根除它们而设计的。

1.1 痛点一:格式的极端不确定性

即使我们在提示中明确要求“请用 JSON 格式输出”,模型也可能:

  • 在 JSON 前后添加“客套话”:例如,输出变成 “好的,根据您的要求,这是结果:\n{\"name\": \"张三\"}”。程序无法直接解析这个字符串。

  • 遗漏关键符号:如缺少闭合的引号、花括号或逗号,导致 JSON.parse 直接抛出异常。

  • 多语言符号混淆:中英文标点混用,如使用了中文分号或中文引号“”,这些在标准 JSON 中是非法字符。

1.2 痛点二:字段缺失或类型错误

模型可能会“擅自发挥”,导致数据不完整或不准确:

  • 遗漏必填字段:提示要求输出姓名和年龄,但模型可能只返回了姓名。

  • 数据类型不匹配:要求年龄是数字,模型却输出了字符串 “二十五” 或者 “25岁”

  • 结构嵌套错误:对于复杂的嵌套对象或列表,模型可能弄错层级关系。

1.3 痛点三:解析逻辑的脆弱性与不可维护性

在没有专用解析器的情况下,开发者通常会写大量正则表达式或字符串分割逻辑来提取数据。这种方式极不可靠:

  • 对 Prompt 修改极度敏感:一旦微调了提示词,模型输出的文本结构稍有变化,原来的正则可能完全失效。

  • 难以处理复杂结构:用正则解析多层嵌套的 JSON 几乎是不可能完成的任务。

  • 缺乏错误反馈机制:解析失败时,通常只能抛出异常,无法自动修复或重试。

1.4 Output Parser 的解决之道

LangChain 的 Output Parser 提供了一套声明式、可复用的解决方案:

  • 自动生成格式指令:通过 get_format_instructions() 方法,自动生成清晰、结构化的格式说明,并注入 Prompt。这相当于给模型提供了一份“模板”,告诉它必须严格按照何种结构输出。

  • 自动解析与验证:它不仅能将文本转为 Python 对象(字典、列表、Pydantic 模型),还能进行类型校验(如使用 Pydantic)。

  • 自我修复与重试机制:当解析失败时,有专门的 OutputFixingParserRetryWithErrorOutputParser 可以自动调用 LLM 修复错误,或者将错误信息反馈给模型重试,形成闭环。

  • 无缝集成 LCEL:通过 | 管道操作符,可以像搭积木一样将解析器嵌入到链中,数据流畅流转。

写一个 PydanticOutputParser 的例子,定义 Pydantic 类,并将其用于提取 JSON 结果。

PydanticOutputParser 是 LangChain 中最强大、最常用的解析器之一。它利用 Python 的 Pydantic 库来定义数据模型,自动生成 JSON Schema,并在解析时进行严格的类型检查和数据验证。

下面是一个详细的实战例子,演示如何从一段学术文本中提取论文信息。

# 导入必要的模块
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List

# 第一步:定义我们期望的数据结构
# 使用 Pydantic 的 BaseModel,可以定义字段、类型和描述
class Paper(BaseModel):
    # Field 中的 description 会帮助 LLM 理解该字段的含义
    title: str = Field(description="论文标题")
    authors: List[str] = Field(description="作者列表,即使只有一个作者也要用列表格式")
    year: int = Field(description="发表年份,必须是整数")
    keywords: List[str] = Field(description="关键词列表")

# 第二步:创建解析器实例
parser = PydanticOutputParser(pydantic_object=Paper)

# 第三步:查看并理解解析器自动生成的格式指令
# 这是 Output Parser 的精髓之一,它告诉 LLM 该怎么输出
format_instructions = parser.get_format_instructions()
print("生成的格式指令如下:\n", format_instructions)
# 输出示例:
# The output should be a JSON object with the following keys:
# - "title": (论文标题)
# - "authors": (作者列表)
# - "year": (发表年份)
# ...

# 第四步:构建 Prompt 模板
# 将格式指令嵌入到系统提示中,让 LLM 严格遵守
template = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的学术信息提取助手。请严格按照以下JSON格式输出提取结果,不要输出任何其他文字:\n{format_instructions}"),
    ("human", "请从以下文本中提取论文信息:\n{text}")
])

# 第五步:初始化模型并组装链
model = ChatOpenAI(model="gpt-4o", temperature=0)  # temperature=0 能让输出更稳定
chain = template | model | parser  # 使用 LCEL 管道符串联

# 第六步:准备数据并调用
text = "《Attention Is All You Need》由Ashish Vaswani等人于2017年发表,提出了Transformer架构,关键词包括注意力机制、机器翻译。"
result = chain.invoke({
    "text": text,
    "format_instructions": format_instructions  # 注入格式指令
})

# 第七步:查看和使用结果
print("解析后的对象类型:", type(result))  # <class '__main__.Paper'>
print("论文标题:", result.title)
print("发表年份:", result.year)
print("作者数量:", len(result.authors))
# 输出:
# 解析后的对象类型: <class '__main__.Paper'>
# 论文标题: Attention Is All You Need
# 发表年份: 2017
# 作者数量: 1

背后机制深挖:

  1. get_format_instructions() 方法内部会根据 Pydantic 模型生成一个详细的 JSON Schema 描述,包括每个字段的名称、类型和 description 内容。这个描述清晰到足以让 LLM 理解并遵守。

  2. 当链被调用时,template 首先将变量替换,生成一个包含具体格式指令的完整提示。

  3. model 接收提示,生成符合格式的 JSON 字符串。

  4. parser 接收模型输出,底层调用 json.loads 将字符串解析为 Python 字典,然后使用 Paper(**dict) 实例化 Pydantic 对象。Pydantic 会自动进行类型转换(例如将 "2017" 转为 2017)和验证(例如检查 year 是否为整数)。

当 LLM 返回的 JSON 不合法时,LangChain 的 OutputFixingParser 是如何修正的?

尽管我们提供了严格的格式指令,LLM 还是可能偶尔“犯错”,比如在 JSON 前后多说了几句话,或者遗漏了一个引号。这时,OutputFixingParser 就登场了。它实际上是一个装饰器(Wrapper),包装了另一个基础解析器(如 PydanticOutputParser),提供了兜底修复能力。

3.1 工作机制

OutputFixingParser 的工作流程如下:

  1. 尝试解析:它首先将 LLM 的原始输出传递给被包装的基础解析器(base_parser)进行解析。

  2. 捕获异常:如果基础解析器解析失败,抛出了异常(例如 JSONDecodeError, ValidationError),OutputFixingParser 会捕获这个异常,而不是让程序崩溃。

  3. 构建修复提示:它动态创建一个新的 Prompt,这个 Prompt 包含三部分关键信息:

  4. 原始格式指令:提醒 LLM 应该遵循的正确格式。
  5. LLM 的原始输出:也就是那个不合法的文本。
  6. 详细的错误信息:例如 JSONDecodeError 中的具体位置和原因(Expecting property name enclosed in double quotes)。

  7. 调用修复 LLM:将这个修复提示发送给另一个专门的 LLM(通常就是同一个模型实例,但可以配置),要求它根据错误信息修正原始输出。

  8. 再次解析:OutputFixingParser 拿到修复 LLM 返回的文本后,再次调用基础解析器。如果成功,则返回最终结果;如果仍然失败,则会抛出异常。

3.2 代码示例

from langchain_core.output_parsers import PydanticOutputParser, OutputFixingParser
from langchain_openai import ChatOpenAI

# 1. 定义基础解析器
base_parser = PydanticOutputParser(pydantic_object=Paper)

# 2. 创建 OutputFixingParser
# 它需要两个参数:被包装的解析器,和用于修复的 LLM
fixing_parser = OutputFixingParser.from_llm(
    parser=base_parser,
    llm=ChatOpenAI(model="gpt-4o", temperature=0)
)

# 3. 将修复解析器集成到链中
chain = template | model | fixing_parser

# 当 model 输出 `"好的,这是提取的论文信息:\n{\"title\": ...}"` 时,
# base_parser 解析失败,OutputFixingParser 会启动修复流程

3.3 成本与最佳实践

  • 额外成本:修复流程会额外消耗一次 LLM 调用,这意味着更高的延迟和费用。

  • 最佳实践:不要将它作为常态依赖。首先应该通过优化 Prompt(如使用更清晰的指令、few-shot 示例)来提高 LLM 的首次输出成功率。OutputFixingParser 是最后一道“安全网”。

RetryWithErrorOutputParser 的重试机制是怎样的?它如何将上次的错误信息反馈给 LLM?

RetryWithErrorOutputParser 的核心理念与 OutputFixingParser 不同。它不是调用一个“修复专家”,而是让原模型进行“自我反省”并重试。

4.1 重试机制详解

  1. 首次尝试与失败:与之前一样,它先用基础解析器解析 LLM 的原始输出。一旦失败,捕获异常。

  2. 构建重试提示:它巧妙地构造一个新的 Prompt,这个 Prompt 不仅包含原始的用户输入,还包括:

  3. LLM 上一次生成的(错误)输出。
  4. 解析失败的具体错误信息(例如 JSONDecodeError 的 traceback)。 这种设计的意图是让模型“看到”自己犯的错误以及程序给出的反馈,就像老师批改作业一样。

  5. 发送重试请求:将这个包含错误信息的 Prompt 再次发送给同一个 LLM。模型结合错误反馈和原始任务,通常会生成一个修正后的、更符合要求的输出。

  6. 再次解析:用基础解析器解析重试后的输出。如果成功,返回;如果失败,可以配置再次重试,直到达到最大重试次数。

4.2 代码示例与对比

from langchain_core.output_parsers import RetryWithErrorOutputParser

# 创建 RetryWithErrorOutputParser
retry_parser = RetryWithErrorOutputParser.from_llm(
    parser=base_parser,
    llm=ChatOpenAI(model="gpt-4o", temperature=0)
)
# 链的其他部分与 OutputFixingParser 类似

两种容错机制对比:

查看内嵌表格

4.3 组合使用策略

在实际工程中,可以将两者结合起来,实现更强大的容错链:

  1. 首先使用 RetryWithErrorOutputParser 让模型自省重试 1-2 次。

  2. 如果仍然失败,再使用 OutputFixingParser 作为最后的兜底修复。 这种分层容错的设计,能在保证成功率的同时,最大限度地利用模型的自我修正能力。

比较 Output Parser 和 OpenAI 原生 Function Calling,各自的优劣和适用场景。

这是 LLM 应用架构中的一个核心选型问题。两者都能实现结构化输出,但原理和侧重点完全不同。

查看内嵌表格

选型建议:

  • 优先使用 Function Calling:如果你的应用只使用 OpenAI 或 Claude 等支持该功能的模型,并且需要极高的格式可靠性(如金融、医疗数据提取),这是首选。

  • 使用 Output Parser:当你需要兼容多种模型(特别是本地部署的开源模型),或者你的输出结构非常特异,或者你正在快速原型设计阶段,需要频繁调整输出结构。

两者并非完全互斥。一个混合策略是:使用 Function Calling 来定义顶层的工具调用,而在工具内部的详细参数提取中,使用 Output Parser 来处理更复杂的自由文本。

如果你需要提取一段文本中的多个实体(如人名、地名、时间),应该用哪种 Output Parser?

对于命名实体识别(NER) 式的多实体提取任务,我们需要的是一个能够输出实体列表的解析器,每个实体最好还带有类别、原文等属性。显然,只能输出简单字符串列表的 CommaSeparatedListOutputParser 不够用。PydanticOutputParser 是处理这类复杂嵌套结构的最佳选择。

6.1 定义模型

首先,我们使用 Pydantic 来定义一个既包含列表、又包含嵌套对象的复杂数据模型。

from typing import List, Literal
from pydantic import BaseModel, Field

# 定义单个实体的结构
class Entity(BaseModel):
    name: str = Field(description="实体名称")
    type: Literal["PERSON", "LOC", "ORG", "TIME", "EVENT"] = Field(description="实体类型")
    mention: str = Field(description="实体在文本中的原始表述")

# 定义输出结构:一个包含实体的列表
class EntityExtraction(BaseModel):
    entities: List[Entity] = Field(description="从文本中提取出的所有实体列表")

# 创建解析器
parser = PydanticOutputParser(pydantic_object=EntityExtraction)

6.2 关键设计点

  • 使用 Literal 约束类型:type 字段使用了 Literal["PERSON", "LOC", ...],这会在生成的 JSON Schema 中限制类型必须是这几个枚举值之一。这能有效降低模型的“幻觉”,避免它创造出奇怪的实体类型。

  • 保留 mention 字段:mention 记录了实体在原文中的原始表述,这对于回溯和校验非常重要,因为模型可能会规范化实体名称(如将“小张”提取为“张伟”)。

  • 包装在列表对象中:直接要求模型输出一个 List[Entity] 有时不太稳定。像上面那样用一个 EntityExtraction 对象包裹 entities: List[Entity] 字段,结构更清晰,模型也更容易理解,符合 JSON 最佳实践。

通过这种方式,我们就能可靠地将一段自由文本转化为一个包含丰富信息的实体对象列表,供下游应用使用。

如何自定义一个 Output Parser?需要继承哪个基类,实现哪些方法?

当你的 LLM 输出格式非常特殊,既不是 JSON,也不是简单列表时,自定义 Output Parser 就派上了用场。例如,你要求模型输出一种自定义的标记语言。LangChain 让这个过程变得非常简单。

7.1 继承基类与实现方法

你需要做的是:

  1. 继承 BaseOutputParser 基类。

  2. 实现三个核心方法:parseget_format_instructions_type

下面我们实现一个解析“姓名:xxx,年龄:yyy”这种特定格式的解析器。

from langchain_core.output_parsers import BaseOutputParser
from typing import Dict, Any

class NameAgeParser(BaseOutputParser[Dict[str, Any]]):
    """一个用于解析 '姓名:张三,年龄:25' 这种自定义格式的解析器"""

    # 核心解析逻辑
    def parse(self, text: str) -> Dict[str, Any]:
        # 1. 去除首尾空白
        text = text.strip()
        result = {}

        # 2. 按逗号分割字符串
        parts = text.split(",")
        for part in parts:
            # 3. 解析每个 "键:值" 对
            if ":" in part:
                key, value = part.split(":", 1)
                key = key.strip()
                value = value.strip()
                # 4. 尝试类型转换,增强健壮性
                if key == "年龄":
                    try:
                        value = int(value)
                    except ValueError:
                        raise ValueError(f"年龄必须是数字,但得到: {value}")
                result[key] = value

        # 5. 验证必填字段
        if "姓名" not in result or "年龄" not in result:
            raise ValueError(f"解析失败:缺少必填字段 '姓名' 或 '年龄'。解析的文本为: {text}")

        return result

    # 返回格式说明,会被注入到 Prompt 中引导模型
    def get_format_instructions(self) -> str:
        return '请严格使用以下格式回答:\n姓名:<名字>,年龄:<数字>'

    # 返回一个唯一标识,用于日志和序列化
    @property
    def _type(self) -> str:
        return "name_age_parser"

7.2 集成到 LCEL 链中

自定义解析器就像任何内置解析器一样,可以直接通过 | 操作符集成到你的链中。

# 初始化自定义解析器
my_parser = NameAgeParser()

# 使用 LCEL 组装链
chain = prompt | model | my_parser

# 调用链
result = chain.invoke({"text": "请提取信息。"})
print(result)  # 输出: {'姓名': '张三', '年龄': 25}

自定义 Output Parser 的设计哲学在于,它将非标准化的输出处理逻辑封装在一个独立的、可复用的组件中,保持了代码的整洁和可维护性,同时完全保留了 LangChain 的流式、异步等高级功能。

在 LCEL 中,如何将 Output Parser 集成到链里?写出链的组装代码。

LCEL(LangChain Expression Language)通过 | 管道操作符提供了极为优雅的链式组装方式。Output Parser 作为一个 Runnable,可以无缝串联在模型之后,自动接收模型输出并解析。

8.1 基础集成

最基本的集成模式是:prompt | model | parser。数据流如下:

  1. prompt 接收用户输入,生成 ChatPromptValue 或字符串。

  2. model 接收 Prompt 的输出,生成 AIMessage 或字符串。

  3. parser 接收模型的输出,调用其 parse 方法,将文本转换为目标对象(如 Pydantic 实例、字典、列表等)。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

# 1. 定义数据结构
class Person(BaseModel):
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")

# 2. 创建解析器
parser = PydanticOutputParser(pydantic_object=Person)

# 3. 构建 Prompt,注入格式指令
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个信息提取助手。{format_instructions}"),
    ("human", "从以下文本提取信息:{text}")
])

# 4. 模型
model = ChatOpenAI(model="gpt-4o", temperature=0)

# 5. LCEL 链:使用管道符串联
chain = prompt | model | parser

# 6. 调用
result = chain.invoke({
    "text": "我叫张三,今年28岁。",
    "format_instructions": parser.get_format_instructions()
})
print(result)  # Person(name='张三', age=28)

8.2 带预处理的链

如果模型输出可能夹杂额外文本,可以在 modelparser 之间插入一个预处理步骤。RunnableLambda 可以将任意 Python 函数包装为 Runnable

from langchain_core.runnables import RunnableLambda
import re

def extract_json(text: str) -> str:
    # 提取第一个完整 JSON 对象
    match = re.search(r'\{.*\}', text, re.DOTALL)
    if match:
        return match.group(0)
    raise ValueError("未找到有效JSON")

# 链:prompt -> model -> 提取JSON -> parser
chain = prompt | model | RunnableLambda(extract_json) | parser

8.3 带容错的链

结合 OutputFixingParserRetryWithErrorOutputParser,可以进一步提升链的鲁棒性。

from langchain_core.output_parsers import OutputFixingParser, RetryWithErrorOutputParser

# 基础解析器
base_parser = PydanticOutputParser(pydantic_object=Person)

# 修复解析器
fixing_parser = OutputFixingParser.from_llm(
    parser=base_parser,
    llm=ChatOpenAI(model="gpt-4o", temperature=0)
)

# 链:prompt -> model -> 修复解析器
chain = prompt | model | fixing_parser

或者使用重试解析器:

retry_parser = RetryWithErrorOutputParser.from_llm(
    parser=base_parser,
    llm=ChatOpenAI(model="gpt-4o", temperature=0)
)
chain = prompt | model | retry_parser

8.4 多分支链

在更复杂的场景中,可能需要根据条件选择不同的解析器。RunnableBranch 可以实现这一点。

from langchain_core.runnables import RunnableBranch

# 假设有两种可能的输出格式
parser_type1 = PydanticOutputParser(pydantic_object=Person)
parser_type2 = PydanticOutputParser(pydantic_object=Paper)  # 另一个 Pydantic 类

# 根据输入中的 "type" 字段进行路由
branch = RunnableBranch(
    (lambda x: x["type"] == "person", parser_type1),
    (lambda x: x["type"] == "paper", parser_type2),
)

chain = prompt | model | branch

通过 LCEL,Output Parser 的集成变得非常直观,同时保留了链的可组合性、可测试性和流式支持。


当 LLM 输出不是纯 JSON,而是“解释文本 + JSON”时,你如何预处理再解析?

这是一个非常常见的问题:模型可能在返回的 JSON 前后附加一些自然语言解释,例如 “好的,这是您要的数据:\n{\"name\": \"张三\"}”。直接将其传递给 JSON 解析器会引发异常。解决方法是在解析前对输出进行预处理。

9.1 使用 RunnableLambda 进行预处理

最灵活的方法是利用 RunnableLambda 将自定义的清理函数嵌入到链中。清理函数的核心任务是提取出纯粹的 JSON 字符串。

from langchain_core.runnables import RunnableLambda
import re
import json

def extract_and_clean_json(text: str) -> str:
    """
    从包含解释文字的文本中提取并修复 JSON 字符串。
    """
    # 1. 尝试直接解析,如果成功则返回原始文本(可能已经是纯 JSON)
    try:
        json.loads(text)
        return text
    except json.JSONDecodeError:
        pass

    # 2. 使用正则查找第一个 '{' 和最后一个 '}' 之间的内容
    start = text.find('{')
    end = text.rfind('}')
    if start != -1 and end != -1 and end > start:
        json_candidate = text[start:end+1]
        # 3. 可选:尝试修复常见的 JSON 错误
        # 例如替换中文引号等
        json_candidate = json_candidate.replace('“', '"').replace('”', '"')
        return json_candidate

    # 4. 如果仍然无法提取,抛出异常
    raise ValueError(f"无法从文本中提取JSON: {text}")

# 将清理函数包装为 Runnable
cleanup_runnable = RunnableLambda(extract_and_clean_json)

# 集成到链中:prompt -> model -> cleanup -> parser
chain = prompt | model | cleanup_runnable | parser

9.2 使用 LangChain 内置的 parse_partial_json

LangChain 提供了一些内置工具来处理不完整或混杂的 JSON。langchain_core.utils.json 中的 parse_partial_json 可以解析不完整的 JSON(适合流式场景),但它无法处理夹杂的非 JSON 文本。

对于“文本+JSON”的情况,更好的方式是使用 langchain_core.output_parsers.json 中的 parse_json_markdown,它能从 Markdown 代码块中提取 JSON(例如 \``json ... ````)。

from langchain_core.output_parsers.json import parse_json_markdown

def extract_json(text: str) -> str:
    try:
        return parse_json_markdown(text)
    except Exception:
        # 回退到自定义正则提取
        return extract_and_clean_json(text)

.3 使用 OutputFixingParser 作为兜底

即使经过了预处理,JSON 可能仍然不合法。此时,OutputFixingParser 可以作为最后的保险。它会在内部调用修复 LLM,根据错误信息重新生成合法的 JSON。

from langchain_core.output_parsers import OutputFixingParser

# 基础解析器
base_parser = PydanticOutputParser(pydantic_object=Person)

# 修复解析器
fixing_parser = OutputFixingParser.from_llm(
    parser=base_parser,
    llm=ChatOpenAI(model="gpt-4o", temperature=0)
)

# 链:prompt -> model -> 预处理 -> 修复解析器
chain = prompt | model | RunnableLambda(extract_and_clean_json) | fixing_parser

这样,我们就构建了一个三层防护:Prompt 指令 → 预处理提取 → 修复解析,确保即使模型输出混杂了额外文本,也能得到正确的结构化数据。


你如何测试 Output Parser 的健壮性?如果 LLM 反复输出错误格式,怎么办?

在生产环境中,Output Parser 的健壮性直接决定了应用的可靠性。我们需要构建一套完善的测试体系,并制定应对反复出错的策略。

10.1 测试 Output Parser 的健壮性

测试可以分为单元测试和集成测试两个层面。

单元测试

单元测试专注于解析器本身的逻辑,无需实际调用 LLM。我们可以手动构造各种合法和不合法的输入,验证解析器的行为是否符合预期。

import pytest
from your_parser_module import NameAgeParser

def test_valid_input():
    parser = NameAgeParser()
    result = parser.parse("姓名:张三,年龄:25")
    assert result == {"姓名": "张三", "年龄": 25}

def test_missing_field():
    parser = NameAgeParser()
    with pytest.raises(ValueError, match="缺少必填字段"):
        parser.parse("姓名:张三")

def test_invalid_age():
    parser = NameAgeParser()
    with pytest.raises(ValueError, match="年龄必须是数字"):
        parser.parse("姓名:张三,年龄:二十五")

def test_extra_spaces():
    parser = NameAgeParser()
    result = parser.parse("  姓名:  李四  ,年龄: 30  ")
    assert result == {"姓名": "李四", "年龄": 30}

def test_partial_json():
    # 测试预处理函数
    from your_utils import extract_and_clean_json
    text = "好的,结果如下:\n{\"name\": \"张三\"}"
    cleaned = extract_and_clean_json(text)
    assert cleaned == '{"name": "张三"}'

集成测试

集成测试调用真实的 LLM 接口,评估整个链在真实场景下的表现。可以准备一批测试样本,运行链并统计解析成功率。

import pytest
from your_chain import build_chain

@pytest.mark.integration
def test_chain_on_dataset():
    chain = build_chain()
    test_data = [
        {"text": "我叫张三,今年28岁。", "expected_name": "张三"},
        {"text": "李四,34岁,工程师", "expected_name": "李四"},
        # ...更多测试用例
    ]
    failures = []
    for item in test_data:
        try:
            result = chain.invoke({"text": item["text"]})
            if result.name != item["expected_name"]:
                failures.append(f"预期{item['expected_name']},实际{result.name}")
        except Exception as e:
            failures.append(f"文本'{item['text']}'解析失败: {e}")

    success_rate = 1 - len(failures) / len(test_data)
    assert success_rate >= 0.95, f"解析成功率 {success_rate:.2f} 低于 95%"

10.2 如果 LLM 反复输出错误格式,怎么办?

当 LLM 频繁产生不合法的输出时,可以采取以下递进式策略:

  1. 优化 Prompt(最根本):
  2. 使用更清晰、更具体的指令,明确要求“只输出 JSON,不要任何解释”。
  3. 提供 Few-Shot 示例,展示正确的输出格式。
  4. 将格式指令放在 Prompt 的末尾,因为模型对靠近生成位置的指令更敏感。
  5. 使用 ChatPromptTemplateSystemMessage 强调格式要求。

  6. 增强预处理:

  7. 使用前面提到的 extract_and_clean_json 函数,增强正则表达式的鲁棒性,覆盖更多边界情况。
  8. 结合 json.loads 的异常信息,尝试自动修复简单的错误(如缺少引号、多余逗号等)。

  9. 引入修复/重试机制:

  10. 使用 RetryWithErrorOutputParser 让模型自我纠正。
  11. 使用 OutputFixingParser 调用修复 LLM 进行二次修正。

  12. 降级与兜底策略:

  13. 如果所有解析都失败,应用层应设计降级逻辑。例如,返回一个默认的错误响应,或将原始文本记录到日志中供人工审核,避免系统崩溃。
  14. 可以尝试使用更简单的解析方法(如正则)作为最后的回退手段。

  15. 监控与告警:

  16. 使用 LangSmith 或自定义回调追踪解析失败率。当失败率超过设定阈值时,自动触发告警,提示开发者检查模型版本、Prompt 有效性或 API 服务状态。

总之,健壮性的提升是一个从 “Prompt 工程 -> 解析层防护 -> 应用层容错” 的纵深防御过程。


什么是 StructuredOutputParser?它和 PydanticOutputParser 的区别在哪里?

StructuredOutputParser 是 LangChain 中用于提取结构化数据的基础解析器,它允许你通过预定义的 Response Schema(响应模式)来告诉 LLM 你想要哪些字段以及字段的类型。而 PydanticOutputParser 是它的高级封装,利用 Pydantic 模型来定义 Schema。

11.1 StructuredOutputParser

StructuredOutputParser 使用一个简单的 JSON Schema 字典来定义输出结构。例如:

from langchain_core.output_parsers import StructuredOutputParser, ResponseSchema

# 定义字段
response_schemas = [
    ResponseSchema(name="answer", description="用户问题的回答"),
    ResponseSchema(name="source", description="回答所依据的参考来源"),
]

# 创建解析器
parser = StructuredOutputParser.from_response_schemas(response_schemas)

# 自动生成格式指令
format_instructions = parser.get_format_instructions()
# 输出类似:Your response should be a JSON object with the following keys:
# - "answer": 用户问题的回答
# - "source": 回答所依据的参考来源

它的 parse 方法会将模型输出解析为一个 Python 字典,键就是 ResponseSchema 中定义的 name

11.2 PydanticOutputParser

PydanticOutputParser 使用 Pydantic 的 BaseModel 来定义数据结构,它提供了更强大的类型校验、默认值、复杂嵌套和自定义验证器等高级功能。

from pydantic import BaseModel, Field

class AnswerWithSource(BaseModel):
    answer: str = Field(description="用户问题的回答")
    source: str = Field(description="回答所依据的参考来源")

parser = PydanticOutputParser(pydantic_object=AnswerWithSource)

11.3 核心区别

查看内嵌表格

总结:PydanticOutputParserStructuredOutputParser 的升级版和替代品。在几乎所有新项目中,如果你需要提取多个字段,都推荐直接使用 PydanticOutputParser,以获得更严格的保障和更清晰的代码。


12. 在 Agent 中,Output Parser 是如何解析 LLM 的“Action”和“Action Input”的?举例说明 ReAct 解析器。

Agent 的核心是让 LLM 进行推理(Reasoning)和行动(Acting)。LangChain 中最经典的 Agent 范式之一是 ReAct(Reasoning + Acting)。LLM 被要求按照特定的格式输出,例如:

Thought: 我需要查询天气。
Action: weather_api
Action Input: 北京

为了让程序能理解这个输出,就需要一个专门的 Output Parser 来提取 ActionAction Input

12.1 ReAct 解析器的工作原理

ReActSingleInputOutputParser 是 LangChain 中用于解析单轮 Agent 输出的解析器。它负责从 LLM 的文本输出中识别出 ActionAction Input 部分。

其核心逻辑是:

  1. 在 Prompt 中,我们已经要求 LLM 以 Thought/Action/Action Input/Observation 的格式输出。

  2. ReActSingleInputOutputParser 内部使用正则表达式来匹配这些模式。

  3. 如果匹配到 Action:Action Input:,则返回一个 AgentAction 对象,包含工具名和输入。

  4. 如果匹配到 Final Answer:,则返回一个 AgentFinish 对象,包含最终答案。

  5. 如果两者都没匹配到,或者格式不完整,它会抛出 OutputParserException

12.2 代码示例:集成 ReAct 解析器

from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain.agents.react.output_parser import ReActSingleInputOutputParser

# 1. 定义工具
def get_weather(city: str) -> str:
    return f"{city}的天气是晴天,25摄氏度。"

tools = [Tool(name="weather_api", func=get_weather, description="查询指定城市的天气")]

# 2. 定义 Prompt 模板(简化版)
template = """你是一个助手。你可以使用以下工具:

{tools}

使用以下格式回答:

Question: 用户的问题
Thought: 你应该思考要做什么
Action: 要采取的行动,应该是 [{tool_names}] 之一
Action Input: 行动的输入
Observation: 行动的结果
... (这个 Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我现在知道最终答案了
Final Answer: 原始问题的最终答案

开始!

Question: {input}
Thought:"""

prompt = PromptTemplate.from_template(template)

# 3. 创建模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 4. 创建 Agent,它会自动使用内置的 ReAct 解析器
agent = create_react_agent(llm=llm, tools=tools, prompt=prompt)

# 5. 创建 Agent 执行器
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 6. 运行
result = agent_executor.invoke({"input": "北京今天天气怎么样?"})
print(result)

create_react_agent 内部,它已经将 ReActSingleInputOutputParser 集成进去了。当 LLM 返回文本后,解析器会提取 ActionAction InputAgentExecutor 据此调用工具,再将工具的返回结果(Observation)追加到对话中,继续调用 LLM,直到得到 Final Answer

12.3 自定义解析器的扩展

如果你需要解析非标准的 Agent 格式,可以继承 BaseOutputParser 并实现类似的逻辑,然后将其传递给 Agent 构造函数。这种灵活性使得 LangChain 的 Agent 框架可以适应各种提示词策略。


谈谈 LangChain 中 Output Parser 和 Prompt Template 的协作机制:格式说明是如何自动注入的?

Output Parser 和 Prompt Template 的协作是 LangChain 中自动化程度最高的部分之一,它解决了“如何让模型知道并遵循输出格式”这一关键问题。其核心机制是:Output Parser 通过 get_format_instructions() 方法生成格式说明字符串,Prompt Template 通过变量占位符将其注入到提示词中。

13.1 协作流程

  1. 定义 Output Parser:你创建一个解析器实例,例如 PydanticOutputParser。这个解析器内部封装了目标数据结构的完整知识。

  2. 生成格式指令:当你调用 parser.get_format_instructions() 时,解析器会根据其类型(JSON、Pydantic等)动态生成一段人类可读、模型可理解的格式指令。例如,PydanticOutputParser 会生成包含 JSON Schema 的详细说明。

  3. 设计 Prompt Template:在你的 ChatPromptTemplatePromptTemplate 中,预留一个占位符(如 {format_instructions})。

  4. 注入格式指令:在调用链时,你显式地将 parser.get_format_instructions() 的返回值作为 format_instructions 变量的值传入。

  5. 模型遵循指令:当 LLM 收到这个包含格式指令的完整 Prompt 后,它会根据指令的要求,尝试生成符合特定格式的文本。

13.2 代码演示

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

# 1. 定义数据模型和解析器
class BookInfo(BaseModel):
    title: str = Field(description="书名")
    author: str = Field(description="作者")

parser = PydanticOutputParser(pydantic_object=BookInfo)

# 2. 获取格式指令
format_instructions = parser.get_format_instructions()
print("生成的格式指令:\n", format_instructions)
# 输出类似:
# The output should be a JSON object with the following keys:
# - "title": 书名
# - "author": 作者

# 3. 设计 Prompt,使用 {format_instructions} 占位符
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个图书信息提取助手。请严格按照以下JSON格式输出提取结果:\n{format_instructions}"),
    ("human", "请从以下文本中提取图书信息:\n{text}")
])

# 4. 组装链并调用
chain = prompt | model | parser
result = chain.invoke({
    "text": "《三体》是刘慈欣创作的科幻小说。",
    "format_instructions": format_instructions  # 注入格式指令
})
print(result)  # BookInfo(title='三体', author='刘慈欣')

13.3 这种设计模式的优点

  • 解耦:Prompt 模板不需要知道输出格式的细节,它只负责留出位置。解析器不需要知道 Prompt 长什么样,只负责提供指令和解析。两者可以独立开发和修改。

  • 自动化:开发者无需手动为每个任务编写冗长的格式说明,解析器会自动完成,减少了错误和不一致。

  • 动态性:你可以根据不同的解析器在运行时动态生成不同的格式指令,同一个 Prompt 模板可以适配多种输出格式。

  • 可扩展:自定义的解析器只需实现 get_format_instructions 方法,就可以无缝地融入这个协作体系。

通过这种精巧的协作,LangChain 将“引导模型输出”这一复杂过程标准化和自动化了,让开发者可以专注于业务逻辑,而非 Prompt 字符串的拼接。

介绍一种你在项目中使用过的复杂输出解析方式,比如嵌套 JSON 或列表。

在一个智能合同审查项目中,我们需要从一份长达数十页的租赁合同中提取结构化的关键条款信息。目标输出结构包括合同基本信息、租赁双方详情、以及一个包含多个条款的列表,每个条款下面又有子字段(如条款内容、风险等级、相关法律依据等)。这是一个典型的“列表嵌套对象”结构,使用 Pydantic 配合 PydanticOutputParser 可以精准定义并提取。

14.1 定义多层嵌套的 Pydantic 模型

我们首先用 Pydantic 定义一个可以递归嵌套的模型,清晰表达业务逻辑。

from pydantic import BaseModel, Field
from typing import List, Optional, Literal

class LegalReference(BaseModel):
    """法律依据"""
    law: str = Field(description="法律名称,如'合同法'")
    article: str = Field(description="条款编号,如'第13条'")
    description: str = Field(description="相关法律内容摘要")

class ContractClause(BaseModel):
    """单个合同条款"""
    clause_id: str = Field(description="条款编号,如'3.2'")
    title: str = Field(description="条款标题")
    content: str = Field(description="条款原文摘要")
    risk_level: Literal["高", "中", "低"] = Field(description="风险等级")
    suggestion: Optional[str] = Field(description="修改建议")
    legal_refs: Optional[List[LegalReference]] = Field(description="涉及的法律依据列表")

class ContractInfo(BaseModel):
    """合同主体信息"""
    contract_name: str = Field(description="合同名称")
    party_a: str = Field(description="甲方全称")
    party_b: str = Field(description="乙方全称")
    start_date: str = Field(description="合同生效日期")
    end_date: str = Field(description="合同终止日期")

class ContractExtraction(BaseModel):
    """最终提取结果"""
    basic_info: ContractInfo
    clauses: List[ContractClause] = Field(description="提取的所有条款")
    overall_risk: Literal["高", "中", "低"] = Field(description="整体风险评级")
    summary: str = Field(description="合同摘要")

14.2 应对超长文本的挑战

合同文本很长,直接输入会超出模型的上下文窗口。我们采用 RunnableLambda 对文本进行分段处理,但解析器本身并不需要修改。关键技巧是在 Prompt 中利用 format_instructions 自动注入完整的 JSON Schema,让模型明白期望的复杂结构。

from langchain_core.runnables import RunnableLambda

parser = PydanticOutputParser(pydantic_object=ContractExtraction)

# 针对长文本,可以先用 LangChain 的 TextSplitter 分块,再逐块提取,最后合并
def extract_from_long_text(text: str, chain) -> dict:
    # 这里是分块逻辑的示意
    chunks = split_text(text)
    results = []
    for chunk in chunks:
        res = chain.invoke({"text": chunk, "format_instructions": parser.get_format_instructions()})
        results.append(res)
    return merge_results(results)

# 链定义
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是法律合同审查专家。请提取合同信息。{format_instructions}"),
    ("human", "{text}")
])
chain = prompt | model | parser

14.3 解析器的容错与后处理

对于嵌套列表,模型偶尔会遗漏某个字段。我们在自定义的解析逻辑中增加了一个后处理步骤,对于 Optional 字段自动填充默认值,避免下游因 None 值崩溃。这种“Pydantic + 自定义后处理”的组合拳在实际项目中非常有效。


在流式输出时,Output Parser 还能正常工作吗?会遇到什么问题?

流式输出(Streaming)是提升用户体验的关键,但对 Output Parser 提出了挑战。

15.1 工作原理

当链以 stream 方式调用时,LangChain 会检查每个组件的流式能力。大多数 BaseOutputParser 本身不支持“逐 Token”解析,因为它们需要完整的输出字符串才能进行 JSON 解析或正则匹配。因此,默认行为是等待模型流式输出完成,拼接成完整字符串后,再一次性调用 parser.parse()

15.2 常见问题

  1. 输出被截断:如果流式传输中断或模型生成了不完整的 JSON(例如最后一个 } 缺失),解析器会失败。

  2. 延迟反馈:用户在前端看到文字逐个打出,但最终解析却要等到最后,可能会感觉到“卡顿”。

  3. 无法处理部分 JSON:标准 JSON 解析器不能处理不完整的 JSON 片段。例如 {"name": "张三", "age": 3 还未闭合,json.loads 会报错。

15.3 解决方案

  • 使用 parse_partial_json:LangChain 提供了 langchain_core.utils.json.parse_partial_json,可以解析不完整的 JSON 字符串,返回当前已成功解析的部分。你可以在流式回调中逐步更新解析结果。

  • 自定逐步解析器:继承 BaseOutputParser,重写 stream 方法,在内部维护一个缓冲区,每当收到新的 Token,尝试用 parse_partial_json 提取当前结果,并通过回调通知上层。

  • 在 Prompt 中强调格式完整性:虽然不能完全避免,但可以要求模型“输出紧凑的 JSON 且不要添加额外内容”,减少流式异常。

from langchain_core.utils.json import parse_partial_json

class StreamingJsonParser(BaseOutputParser):
    def parse(self, text: str) -> Any:
        return json.loads(text)

    def stream(self, input: AsyncIterator[str]) -> AsyncIterator[Any]:
        buffer = ""
        async for chunk in input:
            buffer += chunk
            try:
                # 尝试部分解析
                partial = parse_partial_json(buffer)
                yield partial
            except Exception:
                pass  # 解析失败,等待更多数据

如何为 Output Parser 编写单元测试?你会 mock LLM 的输出吗?

单元测试是保证解析器健壮性的基石。我们遵循“不依赖外部服务”的原则,一定会 mock LLM 的输出。

16.1 测试策略

  • 单元测试:针对 parse 方法,提供各种合法的、非法的、边界值的输入字符串,断言输出或异常。

  • 集成测试:使用 mock 的 LLM 响应(FakeChatModelunittest.mock.patch)测试整个链的串联行为。

16.2 代码示例

import pytest
from unittest.mock import patch, MagicMock
from langchain_core.messages import AIMessage
from your_parser_module import NameAgeParser

# 单元测试:直接测试解析逻辑
def test_parse_valid():
    parser = NameAgeParser()
    result = parser.parse("姓名:张三,年龄:25")
    assert result == {"姓名": "张三", "年龄": 25}

def test_parse_missing_field_raises():
    parser = NameAgeParser()
    with pytest.raises(ValueError, match="缺少必填字段"):
        parser.parse("姓名:张三")

# 集成测试:mock 模型输出
@patch('your_module.ChatOpenAI.invoke')
def test_chain_with_mock(mock_invoke):
    # 准备模拟的 AI 消息
    mock_response = AIMessage(content="姓名:李四,年龄:30")
    mock_invoke.return_value = mock_response

    chain = build_chain()  # 返回 prompt | model | parser
    result = chain.invoke({"text": "无关输入"})

    assert result == {"姓名": "李四", "年龄": 30}
    mock_invoke.assert_called_once()

16.3 使用 FakeChatModel

LangChain 提供了 FakeChatModel,可以预置一系列响应,非常适合测试。

from langchain_core.language_models.fake import FakeChatModel

fake_model = FakeChatModel(responses=[
    AIMessage(content="姓名:李四,年龄:30"),
    AIMessage(content="Invalid JSON"),
])

# 测试成功和失败路径

你认为 Output Parser 的局限性在哪?未来是否会被原生 JSON mode 完全取代?

17.1 局限性

  • 依赖 Prompt 质量:解析的成功率高度依赖提示词的撰写技巧,模型可能“不听话”。

  • 不确定性:即使使用了 PydanticOutputParser,模型仍可能输出结构错误的数据,需要多层容错。

  • 性能开销:格式指令会消耗 Token,解析过程需要 CPU 时间,流式兼容性差。

  • 无法应对动态 Schema:如果输出结构需要根据输入动态变化,静态的 Parser 难以处理。

17.2 与原生 JSON Mode 的关系

OpenAI、Claude 等推出的原生 JSON Mode 通过在 API 层面约束输出,显著降低了格式错误率。但我认为 Output Parser 不会被完全取代,理由如下:

  • 多模型兼容性:开源模型、本地模型可能不支持 JSON Mode,而 Output Parser 是通用的。

  • 复杂逻辑处理:Parser 可以包含自定义的清洗、验证、转换逻辑,而 JSON Mode 只能限制输出格式,无法进行业务规则校验。

  • 向后兼容:大量现有应用基于 Prompt 工程,迁移成本高。

  • 混合模式:未来趋势可能是“原生 JSON Mode + Output Parser”结合,前者保证格式,后者进行语义级验证和转换。


总结一下:设计一个健壮的 LLM 输出管道,Prompt + Parser 应遵循哪些最佳实践?

结合上述所有经验,一个生产级的输出管道应遵循以下实践:

  1. 明确的角色与格式隔离:使用 SystemMessage 强调格式要求,HumanMessage 承载业务数据。格式指令放在 System 末尾或 Human 开头,确保模型“看到”。

  2. 用 Pydantic 定义严格的 Schema:利用 BaseModelField(description=...)LiteralOptional 等特性,让模型理解字段的语义和约束。

  3. Few-Shot 示例必不可少:在 Prompt 中提供 1-3 个格式完全正确的示例,尤其对于复杂嵌套结构,能极大幅度提升准确性。

  4. 预处理与后处理双保险:

  5. 预处理:用 RunnableLambda 清洗 Markdown 标记、提取 JSON 块。
  6. 后处理:在 parse 方法中增加默认值填充、类型强制转换等容错逻辑。

  7. 分层容错机制:先尝试 RetryWithErrorOutputParser 让模型自省,失败再用 OutputFixingParser 修复,最后应用层兜底(如返回默认错误响应)。

  8. 编写完善的单元测试:Mock LLM 输出,覆盖正常、异常、边界场景,确保解析器自身的逻辑无懈可击。

  9. 流式场景特殊处理:使用 parse_partial_json 或自定义流式解析器,避免因等待完整输出而丢失流式优势。

  10. 持续监控与迭代:通过 LangSmith 等工具追踪解析失败率,当失败率异常时及时优化 Prompt 或升级模型。

遵循这些实践,你将能构建一个既灵活又坚如磐石的 LLM 输出管道,让大模型真正成为可靠的“结构化数据生成器”。