什么是输出解析器

graph LR
    A[输入/指令] --> B[大模型处理]
    B --> C[文本]
    B --> D[json]
    B --> E[xml]
    B --> F[...]

大模型本质上是在用概率处理自然语言序列，一般而言输出为文本
然而在现实中我们主要使用机器友好的序列语言来构建应用
让大模型说机器懂的话，而不是人懂的话，就是outputParser的主要任务

LangChain输出解析器（Output Parsers）是将大语言模型（LLM）的原始文本响应转换为结构化、可操作数据的关键组件。输出解析器通过提供格式化指令并解析模型输出，实现文本到结构化数据的高效转换。

常见的输出解析器一览

LangChain内置多种解析器，如JSON、XML、CSV等结构化数据格式，部分支持流式输出。使用时需注意模型是否支持特定功能，并确保提示词模板包含格式要求，以避免输出错误。不同模型特性差异较大，需测试验证其兼容性与输出质量。

名称	是否支持流	是否有格式要求	输入	输出
str	✅	❌	str｜message	string
json	✅	✅	str｜message	json object
xml	✅	✅	str｜message	dict
csv	✅	✅	str｜message	List[str]
pydantic	❌	✅	str｜message	Pydantic:Basemodel
yaml	❌	✅	str｜message	Pydantic:Basemodel

不同的输出器在输出上都会不同
注意有一些输出器是不支持流式输出的
当解析器有格式要求的时候，需要在提示词模版格式化的时候，将格式要求实例化进去
下游承接输出时，要注意数据格式兼容问题

主流模型的支持情况

Langchain官方对不同模型输出结构化情况说明：https://python.langchain.com/docs/integrations/chat/

以Deepseek举例：https://python.langchain.com/docs/integrations/chat/deepseek/

注意并不是所有的模型都支持结构化的输出
不支持结构化输出的模型只能输出字符串
在使用某个模型前最好先对支持特定进行测试

LangChain支持的主要输出格式(解析器)

在使用langchain框架时，不推荐直接用原生方式获取输出文本，因为会导致数据格式不统一、后续处理复杂。推荐使用框架自带的解析器（如Stream Output Parser、Pydantic Output Parser等），它们能自动提取并标准化输出结构，方便与Agent或Chain等组件协作。特别是Pydantic解析器，它是一个Python数据验证工具，能确保数据符合预期格式、自动转换类型、出错立即报错，常用于生成结构化数据，提升代码清晰度和安全性。同时介绍了JSON和chamr2等常见解析器及其使用场景，并强调使用框架内置解析器可避免重复造轮子、减少错误。

基础解析器

基础解析器处理最简单的数据格式转换：

StrOutputParser：直接提取模型返回的原始文本，不做任何结构化处理。
CommaSeparatedListOutputParser：将逗号分隔的文本转换为Python列表。例如，将”apples, bananas, oranges”解析为[‘apples’, ‘bananas’, ‘oranges’]。
BooleanOutputParser：解析文本为布尔值（True/False）。模型输出必须是”yes”或”no”（不区分大小写），解析器会统一转为大写后判断。
SimpleJsonOutputParser：将文本简单处理后转换为JSON格式，通常用于模型已经正确输出JSON的情况。

这些基础解析器适合处理简单数据结构，但缺乏验证和容错机制，适用于对格式要求不严格的场景。

# 纯文本
str_parser = StrOutputParser()

# 分号
comma_parser = CommaSeparatedListOutputParser()
comma_parser_instructions = comma_parser.get_format_instructions()

# 布尔值
boolean_parser = BooleanOutputParser()

# 布尔值校验
chain = llm | boolean_parser
resp = chain.invoke("""
要求：仅返回 YES 或者 NO
--
问题：“你好”这句话是否包含敏感信息？
""")

print(resp)

示例代码

import os
from langchain_openai import ChatOpenAI

api_base = os.getenv("OPENAI_API_BASE")
api_key = os.getenv("OPENAI_API_KEY")

llm = ChatOpenAI(
    temperature=0.0,
    base_url=api_base,
    api_key=api_key
)

# 定义一个假的事实天气预报工具，用来传入地名返回实时天气
from langchain_core.tools import tool

@tool
def get_weather(location: str) -> str:
    """根据location地名返回当地实时天气"""
    return "天气小雨18度"
  
# 绑定工具
llm_with_tools = llm.bind_tools([get_weather])
# 调用
response = llm_with_tools.invoke("北京市当前的天气如何？")
response.content

# 引入StrOutputParser
from langchain_core.output_parsers import StrOutputParser

# 临时构造一个链
chain = llm_with_tools | StrOutputParser()
# 调用
response = chain.invoke("北京市当前的天气如何？")
print(response)

import os
from langchain_openai import ChatOpenAI

api_base = os.getenv("OPENAI_API_BASE")
api_key = os.getenv("OPENAI_API_KEY")

llm = ChatOpenAI(
    temperature=0.0,
    base_url=api_base,
    api_key=api_key
)

# 执行链
response = llm.invoke('列出3种适合春季穿搭的颜色，用逗号分隔。')
print("原始输出：", response.content)
# 初始化输出解析器
parser = CommaSeparatedListOutputParser()
# 解析为结构化列表
structured = parser.parse(response.content)
print("结构化列表：", structured)

Pydantic解析器

Pydantic解析器通过Pydantic模型定义复杂结构，提供更严格的验证和数据校验：

PydanticOutputParser：将模型输出解析为Pydantic定义的模型对象。需要预先定义Pydantic模型，支持嵌套数据结构和类型验证。
PydanticToolsParser：专门处理OpenAI工具调用中的Pydantic对象，将工具调用参数解析为Pydantic模型。

Pydantic解析器适合需要严格验证的数据结构，如用户信息、订单详情等，确保输出的完整性和正确性。

示例代码

import os
from langchain_openai import ChatOpenAI

api_base = os.getenv("OPENAI_API_BASE")
api_key = os.getenv("OPENAI_API_KEY")

llm = ChatOpenAI(
    temperature=0.0,
    base_url=api_base,
    api_key=api_key
)

from pydantic import BaseModel, Field, model_validator


# 定义一个名为Joke的数据模型
# 必须要包含的数据字段：铺垫(setup)、抖包袱(punchline)
class Joke(BaseModel):
    setup: str = Field(description="笑话中的铺垫问题，必须以?结尾")
    punchline: str = Field(description="笑话中回答铺垫问题的部分，通常是一种抖包袱方式回答铺垫问题，例如谐音、会错意等")

    # 验证器，你可以根据自己的数据情况进行自定义
    # 注意mode=before意思是数据被转成pydantic模型的字段之前，对原始数据进行验证
    @model_validator(mode="before")
    @classmethod
    def question_ends_with_question_mark(cls, values: dict) -> dict:
        setup = values.get("setup")
        # 如果铺垫问题没有以问号结尾则抛出错误
        if setup and setup[-1] != "?":
            raise ValueError("Badly formed question!")
        return values

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate


# 实例化解析器、提示词模板
parser = PydanticOutputParser(pydantic_object=Joke)
# 注意，提示词模板中需要部分格式化解析器的格式要求format_instructions
prompt = PromptTemplate(
    template="回答用户的查询.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

print("pydanticoutparser的格式指令要求为：")
print(parser.get_format_instructions())

print("最终输出结果：")
# 使用LCEL语法组合一个简单的链
prompt_and_model = prompt | llm
output = prompt_and_model.invoke({"query": "给我讲一个笑话"})
parser.invoke(output)

扩展：什么是Pydantic

定义：Pydantic是一个Python的数据验证工具，它就像一个“数据把关员”，帮你确保数据符合你预期的格式和规则
作用：
- 确保所有信息都是按规定填写
- 自动转换合适数据类型
- 如果有问题就立即报告
好处：
- 省心：不用写很多if-else来检查数据
- 安全：减少因数据格式错误导致的程序崩溃
- 清晰：代码更容易理解，因为数据结构一目了然

Pydantic In LangChain

V0.1: Pydantic 1
V0.2: Pydantic 1 & Pydantic 2
V0.3: Pydantic 2（Rust重构，语法变化）

JSON解析器

JsonOutputParser

示例代码

import os
from langchain_openai import ChatOpenAI

api_base = os.getenv("OPENAI_API_BASE")
api_key = os.getenv("OPENAI_API_KEY")

llm = ChatOpenAI(
    temperature=0.0,
    base_url=api_base,
    api_key=api_key
)

# 使用Pydantic快速生成json
from pydantic import BaseModel, Field


# 定义一个名为Joke的数据模型
# 必须要包含的数据字段：铺垫(setup)、抖包袱(punchline)
class Joke(BaseModel):
    setup: str = Field(description="笑话中的铺垫问题，必须以?结尾")
    punchline: str = Field(description="笑话中回答铺垫问题的部分，通常是一种抖包袱方式回答铺垫问题，例如谐音、会错意等")

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate


# 实例化解析器、提示词模板
parser = JsonOutputParser(pydantic_object=Joke)
# 注意，提示词模板中需要部分格式化解析器的格式要求format_instructions
prompt = PromptTemplate(
    template="回答用户的查询.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

# 使用LCEL语法组合一个简单的链
chain = prompt | llm | parser
output = chain.invoke({"query": "给我讲一个笑话"})
print(output)
# 流式输出
parser.get_format_instructions()
for s in chain.stream({"query":"给我讲一个关于程序员编程的笑话"}):
   print(s)

不使用Pydantic生成JSON，缺点是无法控制精准字段

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate


joke_query = "Tell me a joke."

parser = JsonOutputParser()

prompt = PromptTemplate(
    template="Answer the user query.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | llm | parser

chain.invoke({"query": joke_query})

XML解析器

XMLOutputParser

示例代码

import os
from langchain_openai import ChatOpenAI

api_base = os.getenv("OPENAI_API_BASE")
api_key = os.getenv("OPENAI_API_KEY")

llm = ChatOpenAI(
    temperature=0.0,
    base_url=api_base,
    api_key=api_key
)

# 引入依赖包
from langchain_core.output_parsers import XMLOutputParser
from langchain_core.prompts import PromptTemplate

actor_query = "Generate the shortened filmography for Tom Hanks."

# parser = XMLOutputParser()

# We will add these instructions to the prompt below
parser = XMLOutputParser(tags=["movies", "actor", "film", "name", "genre"])

# We will add these instructions to the prompt below
# parser.get_format_instructions()
prompt = PromptTemplate(
    template="""{query}\n{format_instructions}""",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | llm | parser

output = chain.invoke({"query": actor_query})
print(output)

函数调用解析器

函数调用解析器处理支持OpenAI函数调用的模型（如GPT-4）的响应：

OutputFunctionsParser：解析模型返回的函数调用参数，通常返回JSON格式的函数参数。
JsonOutputFunctionsParser：提取函数调用中的JSON参数。

这些解析器专为需要调用特定函数或工具的场景设计，常见于Agent系统或复杂任务处理。

时间/枚举解析器

处理特定类型的数据：

DatetimeOutputParser：将文本解析为标准日期时间格式（”%Y-%m-%dT%H:%M:%S.%fZ”）。

from langchain.output_parsers import DatetimeOutputParser
from langchain.prompts import ChatPromptTemplate

from app.common import llm

# 创建DatetimeOutputParser
output_parser = DatetimeOutputParser()

# 获取格式指令
format_instructions = output_parser.get_format_instructions()

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", f"你必须按照以下格式返回日期时间：{format_instructions}"),
    ("human", "请将以下自然语言转换为标准日期时间格式：{text}")
])


chain = prompt | llm | output_parser

resp = chain.invoke({"text": "二零二五年五月十三日"})

print(resp)
print(type(resp))

EnumOutputParser：将文本解析为预定义的枚举值，适用于需要严格限制输出选项的场景。

from enum import Enum
from langchain.output_parsers import EnumOutputParser
from langchain_core.prompts import ChatPromptTemplate

from app.common import llm


# 定义枚举类
class Color(Enum):
    RED = "red"
    GREEN = "green"
    BLUE = "blue"

# 创建解析器
output_parser = EnumOutputParser(enum=Color)

# 提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "请选择一种颜色："),
    ("human", "选项：{options}，直接返回颜色文本")
])

# 构建链
chain = prompt | llm | output_parser

# 测试输入
result = chain.invoke({"options": "red, green, blue"})
print("解析后的枚举值：", result)
print(type(result))

这些解析器针对特定数据类型进行了优化，提供更精准的解析结果。

正则解析器

通过正则表达式提取特定模式的数据：

RegexParser：单字段正则匹配。
RegexDictParser：多字段正则提取为字典。

正则解析器适合需要灵活匹配文本模式的场景，如提取电话号码、邮箱地址等。

复合解析器

处理复杂或多部分的输出：

CombiningOutputParser：组合多个解析器，将文本通过分隔符拆分后分别解析。
RetryWithErrorOutputParser：在解析失败时重试并尝试修复输出。

复合解析器提供了更强大的处理能力，可以应对复杂或多任务的输出场景。

强校验解析器

提供更严格的输出验证：

GuardrailsOutputParser：基于Guardrails库，对输出进行强校验，支持自定义规则（如过滤不当内容、校验数据格式）。

Guardrails解析器特别适合安全敏感场景，确保输出内容符合预设规则和标准。

输出阶段的容错处理

是在使用大模型时，由于其输出存在不确定性，可能会导致错误，为此Langchain提供了两种容错机制：一种是“重试”，即当解析器输出错误时，重新调用模型进行修正；另一种是“修复”，通过特定方法直接修复错误输出，确保最终结果符合预期格式和要求。

错误重试

使用RetryOutputParser进行自动重试，示例代码

import os
from langchain_openai import ChatOpenAI

api_base = os.getenv("OPENAI_API_BASE")
api_key = os.getenv("OPENAI_API_KEY")

llm = ChatOpenAI(
    temperature=0.0,
    base_url=api_base,
    api_key=api_key
)

from pydantic import BaseModel, Field


class Action(BaseModel):
    action: str = Field(description="action to take")
    action_input: str = Field(description="input to the action")
    
# 使用RetryOutputParser进行自动重试
from langchain.output_parsers import RetryOutputParser
from langchain_core.exceptions import OutputParserException
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate

template = """Based on the user question, provide an Action and Action Input for what step should be taken.
{format_instructions}
Question: {query}
Response:"""


parser = PydanticOutputParser(pydantic_object=Action)

prompt = PromptTemplate(
    template="Answer the user query.\n{format_instructions}\n{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)
# 假设的用户输入合成提示值
prompt_value = prompt.format_prompt(query="北京今天天气如何?")

# 假设得到的一个错误回答，不符合pydantic的字段要求
bad_response = '{"action": "search"}'
# 运行抛出错误
try:
   parser.parse(bad_response)
except OutputParserException as e:
   print(e)

# 使用RetryOutputParser实现错误重试
# 定义使用哪个模型进行重试
retry_parser = RetryOutputParser.from_llm(parser=parser, llm=llm)
# 传入错误信息以及原始的提示值
retry_parser.parse_with_prompt(bad_response, prompt_value)

错误修复

使用OutputFixingParser进行错误修复，示例代码

import os
from langchain_openai import ChatOpenAI

api_base = os.getenv("OPENAI_API_BASE")
api_key = os.getenv("OPENAI_API_KEY")

llm = ChatOpenAI(
    temperature=0.0,
    base_url=api_base,
    api_key=api_key
)

from typing import List
from pydantic import BaseModel, Field


# 定义一个演员模型，有两个字段
class Actor(BaseModel):
    name: str = Field(description="name of an actor")
    film_names: List[str] = Field(description="list of names of films they starred in")
    

actor_query = "Generate the filmography for a random actor."
parser = PydanticOutputParser(pydantic_object=Actor)

# 假设生成的错误值
misformatted = "{'name': 'Tom Hanks', 'film_names': ['Forrest Gump']}"
# 运行的时候抛出错误
try:
   parser.parse(misformatted)
except OutputParserException as e:
   print(e)
    

# 使用OutputFixingParser进行错误修复
from langchain.output_parsers import OutputFixingParser

# 使用OutputFixingParser可以修复错误
# 定义修复所依赖的LLM
new_parser = OutputFixingParser.from_llm(parser=parser, llm=llm)
# 传入报错信息
new_parser.parse(misformatted)

自定义输出解析器

纳美星输出器

纳美星语网址 https://learnnavi.org/

示例代码

import os
from langchain_openai import ChatOpenAI

api_base = os.getenv("OPENAI_API_BASE")
api_key = os.getenv("OPENAI_API_KEY")

llm = ChatOpenAI(
    temperature=0.0,
    base_url=api_base,
    api_key=api_key
)

import re
from langchain_core.messages import AIMessage

def parse(ai_message: AIMessage) -> str:
    """Convert message to Na'vi style language."""
    content = ai_message.content
    
    # 纳美语特征转换规则
    navi_rules = {
        # 常见问候语转换
        r'\b[Hh]ello\b': 'Kaltxì',
        r'\b[Hh]i\b': 'Kaltxì',
        r'\b[Tt]hank you\b': 'Irayo',
        r'\b[Gg]oodbye\b': 'Eywa ngahu',
        
        # 添加纳美语特征的后缀
        r'\b(friend|friends)\b': 'eylan',
        r'\b(brother|brothers)\b': 'tsmukan',
        r'\b(sister|sisters)\b': 'tsmuke',
        
        # 添加纳美语语气词
        r'[!.]\s*$': ' kxe!',
        r'\?\s*$': ' srak?',
        
        # 在某些词前添加特征前缀
        r'\b(beautiful|pretty)\b': 'na\'vi',
        r'\b(sacred|holy)\b': 'txe\'lan',
    }
    
    # 应用转换规则
    result = content
    for pattern, replacement in navi_rules.items():
        result = re.sub(pattern, replacement, result)
    
    # 添加纳美语特征词缀
    result = f"Oel ngati kameie... {result}"
    
    return result

chain = llm | parse
chain.invoke("hello")

布尔值解析器

通过继承BaseOutputParser的方式来扩展自定义输出器

示例代码

import os
from langchain_openai import ChatOpenAI

api_base = os.getenv("OPENAI_API_BASE")
api_key = os.getenv("OPENAI_API_KEY")

llm = ChatOpenAI(
    temperature=0.0,
    base_url=api_base,
    api_key=api_key
)

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


# [bool] 描述了一个泛型的参数化。
# 它基本上说明了解析的返回类型是什么
# 在这种情况下，返回类型为 True 或 False
# 自定义解析器将YES、NO字符串解析为布尔值
class BooleanOutputParser(BaseOutputParser[bool]):
    """Custom boolean parser."""

    true_val: str = "YES"
    false_val: str = "NO"

    def parse(self, text: str) -> bool:
        cleaned_text = text.strip().upper()
        if cleaned_text not in (self.true_val.upper(), self.false_val.upper()):
            raise OutputParserException(
                f"BooleanOutputParser expected output value to either be "
                f"{self.true_val} or {self.false_val} (case-insensitive). "
                f"Received {cleaned_text}."
            )
        return cleaned_text == self.true_val.upper()
    #设置统一接口名
    @property
    def _type(self) -> str:
        return "boolean_output_parser"

# 如果传入值不是YES NO会报错
try:
   parser = BooleanOutputParser()
   parser.invoke("MEOW")
except Exception as e:
   print(f"Triggered an exception of type: {type(e)}")

# 我们也可以更换参数
parser = BooleanOutputParser(true_val="OKAY")
parser.invoke("OKAY")

# 简单调用
parser = BooleanOutputParser()
parser.invoke("YES")

Jean's Blog

LangChain组件OutputParsers解析器

什么是输出解析器

常见的输出解析器一览

主流模型的支持情况

LangChain支持的主要输出格式(解析器)

基础解析器

Pydantic解析器

扩展：什么是Pydantic

JSON解析器

XML解析器

函数调用解析器

时间/枚举解析器

正则解析器

复合解析器

强校验解析器

输出阶段的容错处理

错误重试

错误修复

自定义输出解析器

纳美星输出器

布尔值解析器