输出解析与数据转换

为什么需要输出解析

问题：LLM 输出是不可控的文本

当你调用大语言模型时，模型的返回值是一个 AIMessage 对象，其中的 content 字段是一段纯文本字符串。这段文本可能包含多余的解释、格式不规范的标记、甚至一些「废话」。在简单对话场景中这没什么问题，但当你要构建一个自动化工作流时，就会遇到严重的问题：

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

response = llm.invoke("列出三种编程语言的名称")
print(response.content)
# 你可能期望得到：Python, Java, Go
# 但实际输出可能是：
# "当然！以下是三种常见的编程语言：\n\n1. Python\n2. Java\n3. Go\n\n这三种语言各有特色..."

可以看到，模型的输出中包含了大量不需要的内容。如果下游系统期望接收一个干净的列表，这段文本就无法直接使用。

核心矛盾

输出解析器（Output Parser） 就是解决这个矛盾的工具。它的工作是：

1. 指导模型按照指定格式输出（通过在提示词中注入格式说明）
2. 解析模型输出的文本，将其转换为 Python 数据结构（列表、字典、Pydantic 对象等）
3. 验证数据格式，确保解析结果符合预期

输出解析器的位置

在 LCEL 管道中，输出解析器通常位于链的末端：

# 输出解析器在 LCEL 链中的典型位置
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

chain = (
    ChatPromptTemplate.from_template("用一句话解释：{topic}")
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()  # <-- 输出解析器
)

result = chain.invoke({"topic": "机器学习"})
print(result)  # 直接得到字符串，不再是 AIMessage 对象

StrOutputParser

最简单的解析器

StrOutputParser 是 LangChain 中最基础、使用最频繁的输出解析器。它只做一件事：从 AIMessage 对象中提取 content 字段，返回一个纯字符串。

基本用法

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 创建组件
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("用50字解释：{concept}")

# 不使用 StrOutputParser
raw_response = (prompt | llm).invoke({"concept": "递归"})
print(type(raw_response))   # <class 'langchain_core.messages.ai.AIMessage'>
print(raw_response.content) # 需要手动访问 .content

# 使用 StrOutputParser
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"concept": "递归"})
print(type(result))  # <class 'str'>
print(result)        # 直接就是字符串内容

提示

几乎所有链的最后一步都会使用 StrOutputParser，因为大多数应用场景只需要文本内容，不需要 AIMessage 的元数据信息。

在流式输出中使用

StrOutputParser 与流式输出配合时特别有用，它能从流式分块中正确提取文本：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("写一首关于{topic}的四行诗")
chain = prompt | llm | StrOutputParser()

# 流式输出 -- StrOutputParser 确保每个 chunk 都是纯字符串
for chunk in chain.stream({"topic": "编程"}):
    print(chunk, end="", flush=True)
# 输出：
# 键盘如琴指尖舞，
# 代码似诗行行书。
# Bug 修尽天将明，
# 满屏绿意映朝阳。

注意

如果不使用 StrOutputParser，流式输出的每个 chunk 将是 AIMessage 对象而非字符串，需要手动调用 chunk.content 来获取文本内容。

CommaSeparatedListOutputParser

解析逗号分隔的列表

当你需要模型返回一个列表时，CommaSeparatedListOutputParser 可以让模型以逗号分隔的格式输出，并自动将其解析为 Python 列表。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import CommaSeparatedListOutputParser

# 创建解析器
parser = CommaSeparatedListOutputParser()

# 查看解析器自动生成的格式说明
print(parser.get_format_instructions())
# 输出：Your response should be a list of comma separated values, eg: `foo, bar, baz`
#       or `foo,bar,baz`. Do NOT include any other text.

实际使用示例

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import CommaSeparatedListOutputParser

# 1. 创建解析器
list_parser = CommaSeparatedListOutputParser()

# 2. 创建提示词模板，注入格式说明
prompt = ChatPromptTemplate.from_template(
    "列出5个适合初学者学习的{category}。\n"
    "{format_instructions}"
)

# 3. 将格式说明传入模板变量
chain = prompt | ChatOpenAI(model="gpt-4o-mini", temperature=0) | list_parser

result = chain.invoke({
    "category": "Python Web 框架",
    "format_instructions": list_parser.get_format_instructions(),
})

print(type(result))   # <class 'list'>
print(result)         # ['Django', 'Flask', 'FastAPI', 'Tornado', 'Sanic']

# 可以直接遍历使用
for i, framework in enumerate(result, 1):
    print(f"{i}. {framework}")

工作原理

[!warning] > CommaSeparatedListOutputParser 只适合简单的、元素不包含逗号的列表。如果列表元素本身包含逗号（比如「北京，中国」），就会导致解析错误。对于复杂场景，请使用 JsonOutputParser 或 PydanticOutputParser。

JsonOutputParser

解析 JSON 格式输出

JsonOutputParser 是最灵活、最常用的解析器之一。它可以让模型以 JSON 格式输出，并将结果自动解析为 Python 字典。你可以结合 Pydantic 模型来定义 JSON 的结构，解析器会自动生成格式说明注入到提示词中。

基本用法（不带 Schema）

如果不需要严格的 Schema 校验，可以直接使用 JsonOutputParser 不带 Pydantic 模型：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_template(
    "分析以下产品的用户评价，返回 JSON 格式，包含 sentiment（情感）、score（评分 0-1）"
    "和 keywords（关键词列表）。\n\n评价：{review}\n\n{format_instructions}"
)

parser = JsonOutputParser()

chain = prompt | llm | parser

result = chain.invoke({
    "review": "这款手机屏幕很清晰，电池续航也不错，但是拍照效果一般般",
    "format_instructions": parser.get_format_instructions(),
})

print(type(result))  # <class 'dict'>
print(result)
# {'sentiment': '中性', 'score': 0.6, 'keywords': ['屏幕清晰', '续航不错', '拍照一般']}

结合 Pydantic Schema（推荐）

通过 Pydantic 模型定义 JSON 结构，解析器会自动生成格式说明并注入提示词：

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

# 1. 定义输出结构的 Pydantic 模型
class BookAnalysis(BaseModel):
    """书籍分析结果"""
    title: str = Field(description="书名")
    author: str = Field(description="作者")
    genre: list[str] = Field(description="类型标签列表")
    rating: float = Field(description="评分，1-10之间")
    summary: str = Field(description="50字以内的内容摘要")
    suitable_for: list[str] = Field(description="适合的读者群体")

# 2. 创建解析器，传入 Pydantic 模型
parser = JsonOutputParser(pydantic_object=BookAnalysis)

# 3. 查看自动生成的格式说明
print(parser.get_format_instructions())
# 输出示例：
# Return a JSON object that conforms to the following schema:
# {"properties": {"title": {"description": "...", "title": "Title", "type": "string"}, ...}}

# 4. 构建链
prompt = ChatPromptTemplate.from_template(
    "分析以下书籍的信息：{book_info}\n\n{format_instructions}"
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | llm | parser

# 5. 调用
result = chain.invoke({
    "book_info": "《三体》是刘慈欣创作的科幻小说，讲述了地球文明与三体文明的碰撞",
    "format_instructions": parser.get_format_instructions(),
})

print(result)
# {
#     'title': '三体',
#     'author': '刘慈欣',
#     'genre': ['科幻', '硬科幻', '宇宙'],
#     'rating': 9.2,
#     'summary': '地球文明与三体文明跨越光年的碰撞与博弈',
#     'suitable_for': ['科幻爱好者', '理工科学生', '哲学思考者']
# }

print(type(result))  # <class 'dict'>  -- 注意：返回的是字典，不是 Pydantic 对象

[!tip] > JsonOutputParser 返回的是 Python 字典（dict），而不是 Pydantic 对象。如果你需要返回 Pydantic 模型实例以获得类型提示和校验能力，请使用 PydanticOutputParser。

复杂嵌套结构示例

JsonOutputParser 支持复杂的嵌套数据结构。以下是一个分析电商产品的完整示例：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import Optional

# 定义嵌套的 Pydantic 模型
class PriceInfo(BaseModel):
    """价格信息"""
    current: float = Field(description="当前价格")
    original: Optional[float] = Field(default=None, description="原价")
    currency: str = Field(default="CNY", description="货币单位")

class ReviewStats(BaseModel):
    """评价统计"""
    total_count: int = Field(description="总评价数")
    average_score: float = Field(description="平均评分，1-5分")
    positive_ratio: float = Field(description="好评率，0-1之间")

class FeatureHighlight(BaseModel):
    """产品亮点"""
    feature: str = Field(description="亮点名称")
    description: str = Field(description="亮点描述，20字以内")
    rating: int = Field(description="该维度的评分，1-5分")

class ProductAnalysis(BaseModel):
    """产品综合分析"""
    product_name: str = Field(description="产品名称")
    brand: str = Field(description="品牌")
    category: str = Field(description="产品类别")
    price: PriceInfo = Field(description="价格信息")
    review_stats: ReviewStats = Field(description="评价统计")
    highlights: list[FeatureHighlight] = Field(description="3-5个产品亮点")
    pros: list[str] = Field(description="优点列表，3-5条")
    cons: list[str] = Field(description="缺点列表，2-3条")
    recommendation: str = Field(description="推荐建议，50字以内")

# 创建解析器和链
parser = JsonOutputParser(pydantic_object=ProductAnalysis)

prompt = ChatPromptTemplate.from_template(
    "请根据以下产品描述，进行详细的产品分析：\n\n{product_desc}\n\n"
    "{format_instructions}"
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | llm | parser

result = chain.invoke({
    "product_desc": """
    MacBook Pro 14英寸，M3 Pro芯片，18GB内存，512GB SSD。
    售价12999元，原价14999元。累计评价50000+，平均4.8分，好评率97%。
    屏幕素质优秀，性能强劲，续航持久，但接口较少，价格偏高。
    """,
    "format_instructions": parser.get_format_instructions(),
})

# 结果是一个完整的嵌套字典
print(result["product_name"])           # MacBook Pro 14英寸
print(result["price"]["current"])       # 12999.0
print(result["review_stats"]["average_score"])  # 4.8
for highlight in result["highlights"]:
    print(f"  - {highlight['feature']}: {highlight['description']}")

PydanticOutputParser

严格类型校验的输出解析

PydanticOutputParser 与 JsonOutputParser 类似，但有一个关键区别：它返回的是 Pydantic 模型实例而非字典。这意味着你可以获得完整的类型提示、自动校验和 IDE 自动补全支持。

详细示例：简历信息提取

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from enum import Enum
from typing import Optional

# 定义枚举类型
class EducationLevel(str, Enum):
    HIGH_SCHOOL = "高中"
    BACHELOR = "本科"
    MASTER = "硕士"
    PHD = "博士"

# 定义嵌套模型
class WorkExperience(BaseModel):
    """工作经历"""
    company: str = Field(description="公司名称")
    position: str = Field(description="职位名称")
    duration: str = Field(description="在职时间，如 2020.06-2023.03")
    description: str = Field(description="工作职责描述，50字以内")

class EducationRecord(BaseModel):
    """教育经历"""
    school: str = Field(description="学校名称")
    major: str = Field(description="专业")
    level: EducationLevel = Field(description="学历层次")
    graduation_year: int = Field(description="毕业年份")

class ResumeInfo(BaseModel):
    """简历信息"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    email: Optional[str] = Field(default=None, description="邮箱地址")
    phone: Optional[str] = Field(default=None, description="手机号码")
    skills: list[str] = Field(description="技能列表")
    education: list[EducationRecord] = Field(description="教育经历")
    work_experience: list[WorkExperience] = Field(description="工作经历，最多5条")
    self_evaluation: str = Field(description="自我评价，100字以内")
    job_intention: Optional[str] = Field(default=None, description="求职意向")

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

# 查看格式说明（内容比 JsonOutputParser 更详细）
format_instructions = parser.get_format_instructions()

# 构建链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个简历信息提取助手。请从用户提供的简历文本中提取结构化信息。"),
    ("human", "请从以下简历中提取信息：\n\n{resume_text}\n\n{format_instructions}"),
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | llm | parser

# 调用
resume_text = """
张三，28岁，邮箱：zhangsan@email.com，手机：13800138000。

教育背景：
- 2015-2019，北京大学，计算机科学与技术，本科
- 2019-2022，清华大学，人工智能，硕士

工作经历：
- 2022.07-至今，字节跳动，高级算法工程师，负责推荐系统算法优化，提升点击率15%
- 2021.06-2022.06，腾讯实习，算法实习生，参与自然语言处理项目开发

技能：Python、PyTorch、TensorFlow、SQL、Linux、Docker
求职意向：AI算法工程师
自我评价：具有扎实的机器学习理论基础和丰富的工程实践经验，擅长将研究成果落地应用。
"""

result = chain.invoke({
    "resume_text": resume_text,
    "format_instructions": format_instructions,
})

# result 是 ResumeInfo 的 Pydantic 实例
print(type(result))              # <class 'ResumeInfo'>
print(result.name)               # 张三
print(result.age)                # 28
print(result.email)              # zhangsan@email.com
print(result.skills)             # ['Python', 'PyTorch', 'TensorFlow', ...]
print(result.education[0].school)  # 北京大学
print(result.education[0].level)   # EducationLevel.BACHELOR
print(result.work_experience[0].company)  # 字节跳动
print(result.work_experience[0].duration) # 2022.07-至今

format_instructions 的作用

PydanticOutputParser 的核心机制是通过 get_format_instructions() 方法生成一段格式说明文本，注入到提示词中。这段文本告诉模型应该以什么格式输出 JSON：

parser = PydanticOutputParser(pydantic_object=ResumeInfo)

instructions = parser.get_format_instructions()
print(instructions)
# 输出示例：
# The output should be formatted as a JSON instance that conforms to the JSON schema below.
#
# As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}
# the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema.
# The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.
#
# Here is the output schema:
# {"properties": {"name": {...}}, "required": [...], ...}

提示

务必在提示词模板中包含 {format_instructions} 占位符，否则解析器无法指导模型按正确格式输出，解析失败率会大幅上升。

解析错误的处理

当模型输出不符合 Schema 时，PydanticOutputParser 会抛出 OutputParserException：

from langchain_core.exceptions import OutputParserException

try:
    result = chain.invoke({"resume_text": resume_text, "format_instructions": format_instructions})
except OutputParserException as e:
    print(f"解析失败: {e}")
    # 常见原因：
    # 1. 模型输出的不是有效 JSON
    # 2. JSON 中缺少必填字段
    # 3. 字段类型不匹配（如期望 int 但得到 str）

Structured Output vs Parser

两种获取结构化输出的方式

LangChain 提供了两种主要方式来获取结构化输出，各有优劣：

对比表

维度	with_structured_output	OutputParser（Pydantic/Json）
工作原理	利用模型的 Function Calling / Tool Use 能力	通过提示词中的格式说明约束输出
可靠性	高，模型原生支持	中等，依赖模型遵循指令的能力
模型要求	模型必须支持工具调用	任何模型都能用
Token 消耗	较少，不需要格式说明	较多，格式说明会占用 Token
流式支持	部分支持	完全支持
返回类型	Pydantic 对象或 dict	Pydantic 对象或 dict
错误处理	模型层面校验	解析器层面校验
灵活性	较低，受限于模型能力	较高，可自定义解析逻辑
适用模型	GPT-4、Claude 等主流模型	所有模型，包括本地小模型

with_structured_output 示例

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

class MovieRecommendation(BaseModel):
    """电影推荐"""
    title: str = Field(description="电影名称")
    year: int = Field(description="上映年份")
    genre: list[str] = Field(description="类型标签")
    reason: str = Field(description="推荐理由，30字以内")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 使用 with_structured_output
structured_llm = llm.with_structured_output(MovieRecommendation)

result = structured_llm.invoke("推荐一部科幻电影")
print(type(result))     # <class 'MovieRecommendation'>
print(result.title)     # 星际穿越
print(result.year)      # 2014
print(result.genre)     # ['科幻', '冒险', '剧情']
print(result.reason)    # 诺兰执导的硬科幻巨作，探讨时间与爱的力量

OutputParser 示例（相同结果）

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

class MovieRecommendation(BaseModel):
    """电影推荐"""
    title: str = Field(description="电影名称")
    year: int = Field(description="上映年份")
    genre: list[str] = Field(description="类型标签")
    reason: str = Field(description="推荐理由，30字以内")

parser = PydanticOutputParser(pydantic_object=MovieRecommendation)

prompt = ChatPromptTemplate.from_template(
    "推荐一部{category}电影。\n\n{format_instructions}"
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | llm | parser

result = chain.invoke({
    "category": "科幻",
    "format_instructions": parser.get_format_instructions(),
})

print(type(result))     # <class 'MovieRecommendation'>
print(result.title)     # 星际穿越

如何选择

提示

实践建议：优先使用 with_structured_output，当模型不支持工具调用或需要更灵活的自定义解析逻辑时，再使用 OutputParser。

自定义输出解析器

什么时候需要自定义解析器

内置解析器覆盖了大多数场景，但有时你需要特殊的解析逻辑：

• 模型输出包含特殊标记，需要手动提取
• 需要自定义的数据清洗和转换逻辑
• 模型输出格式是非标准的，内置解析器无法处理

创建自定义解析器

继承 BaseOutputParser 并实现三个核心方法：

from langchain_core.exceptions import OutputParserException
from langchain_core.output_parsers import BaseOutputParser
from typing import List

class MarkdownCodeBlockParser(BaseOutputParser[List[str]]):
    """
    自定义解析器：从模型输出中提取所有 Markdown 代码块的内容。

    适用于让模型生成代码片段的场景，自动过滤掉非代码内容。

    示例输入：
        Here is a Python example:
        ```python
        print("hello")
        ```
        And a JavaScript one:
        ```javascript
        console.log("hello")
        ```

    示例输出：
        ['print("hello")', 'console.log("hello")']
    """

    def parse(self, text: str) -> List[str]:
        """解析模型输出，提取代码块内容"""
        import re

        # 匹配 Markdown 代码块：```lang\ncode\n```
        pattern = r"```(?:\w+)?\s*\n(.*?)```"
        matches = re.findall(pattern, text, re.DOTALL)

        if not matches:
            raise OutputParserException(
                f"未在输出中找到代码块。原始输出：\n{text[:200]}"
            )

        # 清理每段代码的首尾空白
        return [code.strip() for code in matches]

    @property
    def _type(self) -> str:
        """解析器类型标识"""
        return "markdown_code_block"

# 使用自定义解析器
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_template(
    "请分别用 Python 和 JavaScript 实现：{task}。每个实现都放在代码块中。"
)

chain = prompt | llm | MarkdownCodeBlockParser()

codes = chain.invoke({"task": "冒泡排序算法"})
print(f"提取到 {len(codes)} 段代码：")
for i, code in enumerate(codes, 1):
    print(f"\n--- 代码片段 {i} ---")
    print(code)

另一个示例：键值对解析器

from langchain_core.output_parsers import BaseOutputParser
from langchain_core.exceptions import OutputParserException
from typing import Dict

class KeyValueParser(BaseOutputParser[Dict[str, str]]):
    """
    自定义解析器：将模型的键值对输出解析为字典。

    期望模型输出格式：
        名称: Python
        类型: 编程语言
        创始人: Guido van Rossum
    """

    def parse(self, text: str) -> Dict[str, str]:
        result = {}
        for line in text.strip().split("\n"):
            line = line.strip()
            if not line:
                continue

            # 支持多种分隔符
            for sep in [":", "：", "-", "="]:
                if sep in line:
                    key, value = line.split(sep, 1)
                    result[key.strip()] = value.strip()
                    break

        if not result:
            raise OutputParserException(f"无法解析键值对。原始输出：\n{text[:200]}")

        return result

    @property
    def _type(self) -> str:
        return "key_value"

# 使用
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template(
    "请以键值对格式输出{subject}的基本信息，每行一个键值对，用冒号分隔。"
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | llm | KeyValueParser()

result = chain.invoke({"subject": "Python 编程语言"})
print(result)
# {'名称': 'Python', '类型': '编程语言', '创始人': 'Guido van Rossum',
#  '首次发布': '1991年', '最新版本': 'Python 3.12', '主要用途': 'Web开发、数据科学、AI'}

提示

自定义解析器实现 BaseOutputParser 后，会自动继承 invoke、stream、batch 等 Runnable 方法，可以直接在 LCEL 管道中使用。

输出修复与重试

模型输出不一定总是正确的

即使你在提示词中注入了详细的格式说明，模型有时仍然会输出不符合预期格式的文本。这可能是因为：

• 模型在 JSON 外面加了一层解释文字
• JSON 格式不完整（缺少闭合的括号）
• 字段值类型不正确（数字变成了字符串）

LangChain 提供了两种自动修复机制来应对这种情况。

OutputFixingParser

OutputFixingParser 在解析失败时，会将原始输出和错误信息发送给 LLM，让 LLM 修复格式问题：

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.output_parsers.fix import OutputFixingParser
from pydantic import BaseModel, Field

class PersonInfo(BaseModel):
    """个人信息"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    hobbies: list[str] = Field(description="爱好列表")

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

# 包装为自动修复解析器
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

fixing_parser = OutputFixingParser.from_llm(
    parser=base_parser,
    llm=llm,
)

# 模拟一个有问题的模型输出（JSON 外面有额外文字）
bad_output = """
当然，这是提取的信息：
```json
{
    "name": "李明",
    "age": "二十五岁",
    "hobbies": ["编程", "游泳"]
}```
希望这对你有帮助！
"""

# 基础解析器会失败
try:
    result = base_parser.parse(bad_output)
except Exception as e:
    print(f"基础解析器失败: {type(e).__name__}")

# 修复解析器会自动处理
result = fixing_parser.parse(bad_output)
print(result)
# name='李明' age=25 hobbies=['编程', '游泳']
# 注意：age 从 '二十五岁' 被自动修正为 25

RetryWithErrorOutputParser

RetryWithErrorOutputParser 比 OutputFixingParser 更强大。它不仅会将错误信息发给 LLM 修复，还会把原始的提示词一起发送，让 LLM 在理解完整上下文的情况下重新生成：

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

class RecipeInfo(BaseModel):
    """菜谱信息"""
    dish_name: str = Field(description="菜名")
    ingredients: list[str] = Field(description="食材列表")
    cook_time_minutes: int = Field(description="烹饪时间（分钟）")
    difficulty: str = Field(description="难度：简单/中等/困难")
    steps: list[str] = Field(description="烹饪步骤，3-5步")

base_parser = PydanticOutputParser(pydantic_object=RecipeInfo)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 创建重试解析器
retry_parser = RetryWithErrorOutputParser.from_llm(
    parser=base_parser,
    llm=llm,
)

prompt = ChatPromptTemplate.from_template(
    "请提供{dish}的做法。\n\n{format_instructions}"
)

# 构建链（使用重试解析器）
chain = prompt | llm | retry_parser

# 即使模型输出格式有问题，重试解析器也会自动修复
result = chain.invoke({
    "dish": "番茄炒蛋",
    "format_instructions": base_parser.get_format_instructions(),
})

print(result.dish_name)            # 番茄炒蛋
print(result.ingredients)          # ['番茄', '鸡蛋', '盐', '糖', '食用油']
print(result.cook_time_minutes)    # 10
print(result.difficulty)           # 简单
print(result.steps)                # ['番茄切块...', '鸡蛋打散...', ...]

两种修复解析器的对比

维度	OutputFixingParser	RetryWithErrorOutputParser
修复依据	原始输出 + 错误信息	原始输出 + 错误信息 + 原始提示词
修复能力	修复格式问题	修复格式问题 + 重新生成内容
Token 消耗	较少	较多（需要发送完整提示词）
额外 LLM 调用	1 次	1 次
适用场景	输出格式小错误	输出严重偏离预期

提示

对于大多数场景，OutputFixingParser 已经足够。只有当模型的输出严重偏离预期（比如完全忽略了格式要求）时，才需要使用 RetryWithErrorOutputParser。

数据转换工具

为什么需要数据转换

在实际应用中，链的各个组件之间往往需要对数据进行转换。比如：

• 从模型输出中提取特定字段，传给下一个组件
• 将一个输入同时传给多个处理分支，再合并结果
• 对中间结果进行格式转换、过滤、清洗

LangChain 提供了几个核心的 Runnable 工具来实现这些数据转换操作。

RunnableLambda

RunnableLambda 可以将任意 Python 函数包装为 Runnable 组件，从而在 LCEL 管道中使用。它是连接自定义逻辑与 LangChain 管道的桥梁。

from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 定义自定义转换函数
def extract_first_sentence(text: str) -> str:
    """提取第一句话"""
    sentences = text.replace("。", "。\n").replace("！", "！\n").replace("？", "？\n")
    first = sentences.split("\n")[0].strip()
    return first if first else text

def count_words(text: str) -> dict:
    """统计字数"""
    return {
        "text": text,
        "char_count": len(text),
        "word_count": len(text.split()),
    }

def format_output(data: dict) -> str:
    """格式化输出"""
    return f"文本：{data['text']}\n字数：{data['char_count']}\n词数：{data['word_count']}"

# 在 LCEL 管道中使用
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

chain = (
    ChatPromptTemplate.from_template("用一句话解释：{topic}")
    | llm
    | StrOutputParser()
    | RunnableLambda(extract_first_sentence)   # 自定义转换1
    | RunnableLambda(count_words)              # 自定义转换2
    | RunnableLambda(format_output)            # 自定义转换3
)

result = chain.invoke({"topic": "量子计算"})
print(result)
# 文本：量子计算是利用量子力学原理进行信息处理的计算方式
# 字数：24
# 词数：15

RunnableLambda 的便捷写法

你也可以直接使用 pipe 方法或 | 操作符，不需要显式创建 RunnableLambda：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 显式写法
chain1 = (
    ChatPromptTemplate.from_template("说一个{topic}的优点")
    | llm
    | StrOutputParser()
    | RunnableLambda(lambda x: x.upper())
)

# 简写：直接传入函数（LangChain 会自动包装为 RunnableLambda）
chain2 = (
    ChatPromptTemplate.from_template("说一个{topic}的优点")
    | llm
    | StrOutputParser()
    | (lambda x: x.upper())  # 自动包装
)

# 两种写法效果相同

注意

虽然可以传入 lambda 函数，但为了代码可读性和可调试性，建议对复杂的转换逻辑使用具名函数。

RunnableParallel

RunnableParallel 允许你将一个输入同时传给多个处理分支并行执行，然后将所有分支的结果合并为一个字典。它类似于 Promise.all 或 asyncio.gather 的概念。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 定义三个独立的处理分支
joke_chain = (
    ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")
    | llm
    | StrOutputParser()
)

fact_chain = (
    ChatPromptTemplate.from_template("说一个关于{topic}的有趣事实")
    | llm
    | StrOutputParser()
)

poem_chain = (
    ChatPromptTemplate.from_template("写一首关于{topic}的四行诗")
    | llm
    | StrOutputParser()
)

# 并行执行三个分支
parallel_chain = RunnableParallel(
    joke=joke_chain,
    fact=fact_chain,
    poem=poem_chain,
)

# 一次调用，三个分支同时执行
result = parallel_chain.invoke({"topic": "Python"})
print("=== 笑话 ===")
print(result["joke"])
print("\n=== 事实 ===")
print(result["fact"])
print("\n=== 诗歌 ===")
print(result["poem"])

RunnablePassthrough

RunnablePassthrough 用于在管道中直接传递输入数据，不做任何修改。通常与 RunnableParallel 配合使用，在并行分支中保留原始输入：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 场景：在生成内容的同时保留原始输入
chain = RunnableParallel(
    # 分支1：生成解释
    explanation=(
        ChatPromptTemplate.from_template("用一句话解释：{topic}")
        | llm
        | StrOutputParser()
    ),
    # 分支2：评估难度
    difficulty=(
        ChatPromptTemplate.from_template("评估以下主题的学习难度（简单/中等/困难）：{topic}")
        | llm
        | StrOutputParser()
    ),
    # 分支3：原样传递输入
    original_input=RunnablePassthrough(),
)

result = chain.invoke({"topic": "量子计算"})
print(result)
# {
#     'explanation': '量子计算是利用量子力学原理进行信息处理的技术。',
#     'difficulty': '困难',
#     'original_input': {'topic': '量子计算'}
# }

综合实战：多维度内容分析

以下是一个综合运用数据转换工具的实际案例，实现文章的多维度分析：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableLambda
from pydantic import BaseModel, Field

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 定义摘要分支
summary_chain = (
    ChatPromptTemplate.from_template(
        "请用100字以内总结以下文章的核心内容：\n\n{article}"
    )
    | llm
    | StrOutputParser()
)

# 定义关键词提取分支
class KeywordsResult(BaseModel):
    """关键词提取结果"""
    keywords: list[str] = Field(description="5-8个关键词")
    main_topic: str = Field(description="文章主题")

keyword_parser = JsonOutputParser(pydantic_object=KeywordsResult)

keywords_chain = (
    ChatPromptTemplate.from_template(
        "从以下文章中提取关键词和主题：\n\n{article}\n\n{format_instructions}"
    )
    | llm
    | keyword_parser
)

# 定义情感分析分支
class SentimentResult(BaseModel):
    """情感分析结果"""
    sentiment: str = Field(description="整体情感倾向：正面/负面/中性")
    confidence: float = Field(description="置信度 0-1")
    reason: str = Field(description="判断依据，30字以内")

sentiment_parser = JsonOutputParser(pydantic_object=SentimentResult)

sentiment_chain = (
    ChatPromptTemplate.from_template(
        "分析以下文章的情感倾向：\n\n{article}\n\n{format_instructions}"
    )
    | llm
    | sentiment_parser
)

# 合并结果并格式化
def format_analysis(data: dict) -> dict:
    """将多分支结果合并格式化"""
    return {
        "summary": data["summary"],
        "keywords": data["keywords"]["keywords"],
        "main_topic": data["keywords"]["main_topic"],
        "sentiment": data["sentiment"]["sentiment"],
        "sentiment_confidence": data["sentiment"]["confidence"],
        "sentiment_reason": data["sentiment"]["reason"],
        "original_article_length": len(data["original"]["article"]),
    }

# 组合完整管道
analysis_chain = (
    RunnableParallel(
        summary=summary_chain,
        keywords=keywords_chain,
        sentiment=sentiment_chain,
        original=RunnablePassthrough(),
    )
    | RunnableLambda(format_analysis)
)

# 执行分析
article = """
人工智能技术的快速发展正在深刻改变软件开发的方式。GitHub Copilot 等 AI 编程助手
已经帮助数百万开发者提高了编码效率。与此同时，AI 也在自动化测试、代码审查、Bug 修复
等方面展现出巨大潜力。不过，也有开发者担心 AI 会取代他们的工作。专家认为，AI 更可能
成为开发者的助手而非替代者，未来的软件开发将是人机协作的模式。
"""

result = analysis_chain.invoke({
    "article": article,
    "format_instructions": keyword_parser.get_format_instructions(),
})

print(f"摘要: {result['summary']}")
print(f"主题: {result['main_topic']}")
print(f"关键词: {', '.join(result['keywords'])}")
print(f"情感: {result['sentiment']}（置信度: {result['sentiment_confidence']}）")
print(f"判断依据: {result['sentiment_reason']}")
print(f"原文长度: {result['original_article_length']} 字")

本章小结

核心要点回顾

输出解析与数据转换是 LangChain 应用的关键环节。没有它，模型输出只是一段不可控的文本；有了它，模型输出就变成了结构化、可消费的 Python 数据。

关键知识点速查表

工具	用途	输入	输出	适用场景
StrOutputParser	提取纯文本	AIMessage	str	所有链的默认终端解析器
CommaSeparatedListOutputParser	解析逗号分隔列表	AIMessage	list[str]	简单列表提取
JsonOutputParser	解析 JSON	AIMessage	dict	需要结构化数据的场景
PydanticOutputParser	解析为 Pydantic 对象	AIMessage	BaseModel	需要类型校验的复杂结构
with_structured_output	模型层结构化输出	提示词	BaseModel / dict	模型支持工具调用时优先使用
OutputFixingParser	自动修复解析错误	解析失败的文本	修正后的对象	提高解析鲁棒性
RetryWithErrorOutputParser	重试并修复	解析失败的文本	重新生成的对象	输出严重偏离时使用
RunnableLambda	自定义数据转换	任意数据	任意数据	管道中插入自定义逻辑
RunnableParallel	并行处理	单一输入	dict（多分支结果）	多维度同时处理
RunnablePassthrough	原样传递	任意数据	原样输出	在并行分支中保留原始输入

最佳实践

1. 始终在链末端使用解析器：不要在下游代码中手动处理 AIMessage，用解析器在链内完成转换
2. 优先使用 with_structured_output：当模型支持时，这是最可靠的方式
3. 为 Pydantic 模型添加 description：每个字段的 description 会帮助模型理解应该填入什么内容
4. 使用修复解析器提高鲁棒性：在生产环境中，OutputFixingParser 可以有效减少解析失败
5. 利用 RunnableParallel 提高效率：多个独立的 LLM 调用应该并行执行，而非串行
6. 保持自定义转换函数的纯粹性：RunnableLambda 中的函数应该是无副作用的纯函数

学习路径