输出解析器 API

BaseOutputParser 与结构化输出

概述

输出解析器将 LLM 的原始输出转换为结构化数据(如 JSON、对象、列表等)。

graph TD
                    A[BaseOutputParser] --> B[StrOutputParser]
                    A --> C[PydanticOutputParser]
                    A --> D[JsonOutputParser]
                    A --> E[CommaSeparatedListOutputParser]
    A --> F[StructuredOutputParser]
    A --> G[XMLOutputParser]

    style A fill:#e1f5fe
    style C fill:#c8e6c9

基类

BaseOutputParser

输出解析器抽象基类。

from langchain_core.output_parsers import BaseOutputParser

class BaseOutputParser(Generic[T], ABC):
    """输出解析器基类

    Type Parameters:
        T: 输出类型
    """

    @property
    @abstractmethod
    def _type(self) -> str:
        """返回解析器类型标识"""

    @abstractmethod
    def parse(self, text: str) -> T:
        """
        解析文本

        Args:
            text: LLM 输出文本

        Returns:
            解析后的数据

        Raises:
            OutputParserException: 解析失败
        """

    def parse_result(
        self,
        result: Generation,
        *,
        partial: bool = False,
    ) -> T:
        """
        解析 Generation 对象

        Args:
            result: Generation 对象
            partial: 是否允许部分解析

        Returns:
            解析后的数据
        """

    def get_format_instructions(self) -> str:
        """
        获取格式说明

        Returns:
            格式说明字符串(用于 Prompt)
        """

    @property
    def OutputType(self) -> Type[T]:
        """返回输出类型"""
        return self.__orig_bases__[0].__args__[0]

    def _fix_json(self, text: str) -> str:
        """
        修复 JSON 格式

        Args:
            text: 可能有问题的 JSON 字符串

        Returns:
            修复后的 JSON 字符串
        """

解析器类型

StrOutputParser

字符串输出解析器(最常用)。

from langchain_core.output_parsers import StrOutputParser

class StrOutputParser(BaseOutputParser[str]):
    """字符串输出解析器"""

    def parse(self, text: str) -> str:
        """
        解析为字符串

        Args:
            text: LLM 输出(可能是 AIMessage 或字符串)

        Returns:
            字符串内容
        """

使用示例

python
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# 在链中使用
chain = llm | StrOutputParser()

result = chain.invoke("讲一个笑话")
print(result)  # 纯字符串

PydanticOutputParser

Pydantic 模型输出解析器。

from langchain_core.output_parsers import PydanticOutputParser

class PydanticOutputParser(BaseOutputParser[BaseModel]):
    """Pydantic 输出解析器"""

    pydantic_object: Type[BaseModel]
    """目标 Pydantic 模型"""

    def __init__(
        self,
        pydantic_object: Type[BaseModel],
    ):
        """
        初始化

        Args:
            pydantic_object: 目标 Pydantic 模型类
        """

    def get_format_instructions(self) -> str:
        """
        获取格式说明

        Returns:
            JSON Schema 格式的说明
        """

    def parse(self, text: str) -> BaseModel:
        """
        解析为 Pydantic 对象

        Args:
            text: LLM 输出(应该是 JSON)

        Returns:
            Pydantic 模型实例
        """

使用示例

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

# 定义输出模型
class Answer(BaseModel):
    summary: str = Field(description="答案摘要")
    confidence: float = Field(description="置信度 (0-1)")
    sources: list[str] = Field(description="信息来源")

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

# 获取格式说明
format_instructions = parser.get_format_instructions()

# 在提示中使用
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手"),
    ("user", "{format_instructions}\n\n问题: {query}")
])

chain = prompt | ChatOpenAI(model="gpt-4o") | parser

result = chain.invoke({
    "query": "什么是 Python?",
    "format_instructions": format_instructions
})

print(result.summary)     # "Python 是一种编程语言"
print(result.confidence)  # 0.95
print(result.sources)     # ["官方文档", "维基百科"]

JsonOutputParser

JSON 输出解析器。

from langchain_core.output_parsers import JsonOutputParser

class JsonOutputParser(BaseOutputParser[Union[Dict, List]]):
    """JSON 输出解析器"""

    def __init__(
        self,
        pydantic_object: Optional[Type[BaseModel]] = None,
    ):
        """
        初始化

        Args:
            pydantic_object: 可选的 Pydantic 模型(用于验证)
        """

使用示例

python
from langchain_core.output_parsers import JsonOutputParser

# 基础使用
parser = JsonOutputParser()

# 解析 JSON 字符串
result = parser.parse('{"name": "Alice", "age": 30}')
# {"name": "Alice", "age": 30}

# 解析 JSON 数组
result = parser.parse('["apple", "banana", "orange"]')
# ["apple", "banana", "orange"]

# 在链中使用
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

prompt = PromptTemplate.from_template(
    "返回 {topic} 的 3 个特点,用 JSON 数组格式: {format_instructions}"
)

chain = prompt | ChatOpenAI(model="gpt-4o") | parser

result = chain.invoke({
    "topic": "Python",
    "format_instructions": parser.get_format_instructions()
})
# ["简洁易学", "功能强大", "应用广泛"]

CommaSeparatedListOutputParser

逗号分隔列表解析器。

from langchain_core.output_parsers import CommaSeparatedListOutputParser

class CommaSeparatedListOutputParser(BaseOutputParser[List[str]]):
    """逗号分隔列表解析器"""

    def parse(self, text: str) -> List[str]:
        """
        解析为列表

        Args:
            text: 逗号分隔的字符串

        Returns:
            字符串列表
        """

使用示例

python
from langchain_core.output_parsers import CommaSeparatedListOutputParser

parser = CommaSeparatedListOutputParser()

result = parser.parse("apple, banana, orange")
# ["apple", "banana", "orange"]

# 在链中使用
chain = llm | parser
result = chain.invoke("列出3种水果,用逗号分隔")

StructuredOutputParser

结构化输出解析器(旧版)。

from langchain_core.output_parsers import StructuredOutputParser

class StructuredOutputParser(BaseOutputParser[Dict[str, Any]]):
    """结构化输出解析器(旧版,推荐使用 PydanticOutputParser)"""

    response_schema: Dict[str, Any]
    """输出模式(字段名到描述的映射)"""

XMLOutputParser

XML 输出解析器。

from langchain_core.output_parsers import XMLOutputParser

class XMLOutputParser(BaseOutputParser[Dict[str, Any]]):
    """XML 输出解析器"""

    tags: List[str]
    """要提取的标签列表"""

    def __init__(
        self,
        tags: List[str],
        *,
        strip_tags: bool = False,
    ):
        """
        初始化

        Args:
            tags: 要提取的 XML 标签
            strip_tags: 是否移除标签本身
        """

使用示例

python
from langchain_core.output_parsers import XMLOutputParser

parser = XMLOutputParser(tags=["name", "age", "city"])

xml_text = """

    Alice
    30
    NYC

"""

result = parser.parse(xml_text)
# {"name": "Alice", "age": "30", "city": "NYC"}

RetryOutputParser

带重试的输出解析器。

from langchain_core.output_parsers import OutputFixingParser

class OutputFixingParser(BaseOutputParser[T]):
    """输出修复解析器"""

    parser: BaseOutputParser[T]
    """底层解析器"""

    retry_llm: BaseLanguageModel
    """用于修复的 LLM"""

    def __init__(
        self,
        parser: BaseOutputParser[T],
        retry_llm: BaseLanguageModel,
    ):
        """
        初始化

        Args:
            parser: 可能失败的解析器
            retry_llm: 用于修复输出的 LLM
        """

使用示例

python
from langchain_core.output_parsers import OutputFixingParser
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

class Output(BaseModel):
    answer: str

base_parser = PydanticOutputParser(pydantic_object=Output)
fixing_parser = OutputFixingParser.from_llm(
    parser=base_parser,
    llm=ChatOpenAI(model="gpt-4o")
)

# 即使第一次输出格式错误,也会自动修复
chain = ChatOpenAI(model="gpt-4o") | fixing_parser

Custom 装饰器

创建自定义解析器的装饰器。

from langchain_core.output_parsers import @output_parser

@output_parser
def parse_uppercase(text: str) -> str:
    """将输出转为大写"""
    return text.upper()

# 使用
chain = llm | parse_uppercase

使用示例

python
# ========== 示例1: 基础字符串解析 ==========
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
chain = llm | StrOutputParser()

result = chain.invoke("讲一个笑话")

# ========== 示例2: Pydantic 解析 ==========
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser

class MovieReview(BaseModel):
    title: str = Field(description="电影标题")
    rating: float = Field(description="评分 (0-10)")
    summary: str = Field(description="剧情简介")

parser = PydanticOutputParser(pydantic_object=MovieReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个影评人。{format_instructions}"),
    ("user", "评价电影: {movie_title}")
])

chain = prompt | ChatOpenAI(model="gpt-4o") | parser

review = chain.invoke({
    "movie_title": "盗梦空间",
    "format_instructions": parser.get_format_instructions()
})

print(review.title)    # "盗梦空间"
print(review.rating)   # 9.5
print(review.summary)  # "一部关于梦境的科幻惊悚片..."

# ========== 示例3: JSON 解析 ==========
from langchain_core.output_parsers import JsonOutputParser

parser = JsonOutputParser()

prompt = PromptTemplate.from_template(
    "返回 {query} 的 3 个相关信息,用 JSON 格式: {format_instructions}"
)

chain = prompt | ChatOpenAI(model="gpt-4o") | parser

result = chain.invoke({
    "query": "Python 编程",
    "format_instructions": parser.get_format_instructions()
})
# ["Python 是一种高级编程语言", "由 Guido van Rossum 创建", "广泛应用于数据科学"]

# ========== 示例4: 自定义解析器 ==========
from langchain_core.output_parsers import BaseOutputParser

class DateOutputParser(BaseOutputParser[str]):
    """日期解析器"""

    def parse(self, text: str) -> str:
        # 提取日期
        import re
        date_pattern = r'\d{4}-\d{2}-\d{2}'
        match = re.search(date_pattern, text)
        if match:
            return match.group(0)
        raise ValueError(f"未找到日期: {text}")

    def get_format_instructions(self) -> str:
        return "请使用 YYYY-MM-DD 格式返回日期"

chain = llm | DateOutputParser()
date = chain.invoke("今天是几号?")

# ========== 示例5: 错误处理 ==========
from langchain_core.exceptions import OutputParserException

try:
    result = parser.parse("invalid json")
except OutputParserException as e:
    print(f"解析失败: {e}")

# 使用带重试的解析器
from langchain_core.output_parsers import OutputFixingParser

robust_parser = OutputFixingParser.from_llm(
    parser=base_parser,
    llm=ChatOpenAI(model="gpt-4o")
)

相关 API