2.4 记忆管理

2.4 结构化输出与输出解析 — LangChain框架精通 Agent 核心机制

本节导读：深入学习 LangChain Agent 的结构化输出和输出解析技术，学习如何使用 Pydantic 模型和 JSON Schema 来约束和解析 Agent 的输出，确保输出格式的一致性和可操作性。

学习目标

• 理解结构化输出的重要性和应用场景
• 掌握 Pydantic 模型和 JSON Schema 的使用方法
• 学习输出解析器的配置和自定义
• 实现复杂的结构化输出格式
• 掌握输出验证和错误处理机制
• 构建完整的结构化输出系统

核心概念

结构化输出的架构

结构化输出是确保 Agent 输出格式一致性和可操作性的关键技术。通过结构化输出，我们可以：

1. 输出格式约束：强制 LLM 输出特定格式的数据
2. 数据类型验证：确保输出数据的类型正确性
3. 数据完整性：验证输出数据的完整性和一致性
4. 业务逻辑集成：将结构化数据与业务逻辑无缝集成
5. 错误处理和恢复：处理输出格式异常和错误情况

输出解析器类型

环境准备 / 前置知识

基础依赖安装

# 结构化输出核心包
pip install langchain langchain-openai langchain-community
pip install pydantic jsonschema
​

核心导入模块

import os
import json
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field, validator
from enum import Enum

# 输出解析器
from langchain.output_parsers import (
    PydanticOutputParser,
    JsonOutputParser,
    OutputFixingParser,
    RetryOutputParser
)
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import BaseOutputParser

# Agent 相关
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage
​

分步实战

步骤 1：基础 Pydantic 模型输出

使用 Pydantic 模型来定义和解析结构化输出：

import os
from pydantic import BaseModel, Field, validator
from typing import List
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 设置 API 密钥
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 定义数据模型
class UserProfile(BaseModel):
    """用户资料模型"""
    name: str = Field(description="用户姓名")
    age: int = Field(description="用户年龄", ge=0, le=150)
    email: str = Field(description="用户邮箱")
    interests: List[str] = Field(description="用户兴趣爱好")
    
    @validator('email')
    def validate_email(cls, v):
        """验证邮箱格式"""
        if '@' not in v:
            raise ValueError('邮箱格式不正确')
        return v

class UserAnalysis(BaseModel):
    """用户分析模型"""
    user_profile: UserProfile
    behavior_score: float = Field(description="行为评分", ge=0, le=100)
    risk_level: str = Field(description="风险等级", regex="^(low|medium|high)$")
    recommendations: List[str] = Field(description="建议列表")

# 创建输出解析器
parser = PydanticOutputParser(pydantic_object=UserAnalysis)

# 创建提示词模板
prompt_template = PromptTemplate(
    template="""
你是一个用户分析专家，请根据以下用户信息进行分析：

用户信息：
{user_info}

{format_instructions}
""",
    input_variables=["user_info"],
    partial_variables={
        "format_instructions": parser.get_format_instructions()
    }
)

# 使用示例
user_info = """
姓名：张三
年龄：25
邮箱：zhangsan@example.com
兴趣爱好：编程, 读书, 运动
"""

# 生成提示词
prompt = prompt_template.format_prompt(user_info=user_info)

# 调用 LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.3)
result = llm.invoke([SystemMessage(content="你是一个用户分析专家"), HumanMessage(content=user_info)])

# 解析输出
try:
    analysis = parser.parse(result.content)
    print("分析结果：")
    print(f"用户姓名：{analysis.user_profile.name}")
    print(f"行为评分：{analysis.behavior_score}")
    print(f"风险等级：{analysis.risk_level}")
    print("建议：", "、".join(analysis.recommendations))
    
except Exception as e:
    print(f"解析失败：{e}")
​

步骤 2：JSON Schema 输出解析

使用 JSON Schema 来定义和解析结构化输出：

import os
from langchain.output_parsers import JsonOutputParser
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 定义 JSON Schema
analysis_schema = {
    "type": "object",
    "properties": {
        "user_profile": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "age": {"type": "integer", "minimum": 0, "maximum": 150},
                "email": {"type": "string", "format": "email"},
                "interests": {"type": "array", "items": {"type": "string"}}
            },
            "required": ["name", "age", "email", "interests"]
        },
        "behavior_score": {"type": "number", "minimum": 0, "maximum": 100},
        "risk_level": {"type": "string", "enum": ["low", "medium", "high"]},
        "recommendations": {"type": "array", "items": {"type": "string"}}
    },
    "required": ["user_profile", "behavior_score", "risk_level", "recommendations"]
}

# 创建 JSON 输出解析器
json_parser = JsonOutputParser()

# 创建提示词模板
json_prompt = PromptTemplate(
    template="""
你是一个用户分析专家，请根据以下用户信息进行分析：

用户信息：
{user_info}

请输出符合以下 JSON 格式的分析结果：
{schema}
""",
    input_variables=["user_info"],
    partial_variables={
        "schema": json.dumps(analysis_schema, ensure_ascii=False, indent=2)
    }
)

# 使用示例
user_info = """
姓名：李四
年龄：30
邮箱：lisi@example.com
兴趣爱好：音乐, 旅行, 摄影
"""

# 生成提示词
prompt = json_prompt.format_prompt(user_info=user_info)

# 调用 LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.3)
result = llm.invoke([SystemMessage(content="你是一个用户分析专家"), HumanMessage(content=user_info)])

# 解析输出
try:
    analysis = json_parser.parse(result.content)
    print("JSON 分析结果：")
    print(f"用户姓名：{analysis['user_profile']['name']}")
    print(f"行为评分：{analysis['behavior_score']}")
    print(f"风险等级：{analysis['risk_level']}")
    print("建议：", "、".join(analysis['recommendations']))
    
except Exception as e:
    print(f"JSON 解析失败：{e}")
​

步骤 3：自动修复输出解析器

实现能够自动修复错误的输出解析器：

import os
from langchain.output_parsers import OutputFixingParser, PydanticOutputParser
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 定义数据模型
class AnalysisResult(BaseModel):
    """分析结果模型"""
    sentiment: str = Field(description="情感倾向", regex="^(positive|negative|neutral)$")
    confidence: float = Field(description="置信度", ge=0, le=1)
    key_points: List[str] = Field(description="关键观点")
    summary: str = Field(description="总结")

# 创建标准解析器
base_parser = PydanticOutputParser(pydantic_object=AnalysisResult)

# 创建修复解析器
fixing_parser = OutputFixingParser(
    parser=base_parser,
    llm=ChatOpenAI(model="gpt-4-turbo", temperature=0.1)
)

# 创建提示词模板
fixing_prompt = PromptTemplate(
    template="""
你是一个文本分析专家，请分析以下文本并输出结构化结果：

文本内容：
{text}

{format_instructions}
""",
    input_variables=["text"],
    partial_variables={
        "format_instructions": base_parser.get_format_instructions()
    }
)

# 测试自动修复功能
test_text = """
今天天气很好，心情不错。这个产品功能很强大，用户体验也很好。
价格稍微有点高，但是物有所值。
推荐大家都来试试！
"""

# 生成提示词
prompt = fixing_prompt.format_prompt(text=test_text)

# 调用 LLM
result = llm.invoke([SystemMessage(content="你是一个文本分析专家"), HumanMessage(content=test_text)])

# 使用修复解析器
try:
    # 先尝试标准解析
    standard_result = base_parser.parse(result.content)
    print("标准解析成功：")
    print(f"情感倾向：{standard_result.sentiment}")
    print(f"置信度：{standard_result.confidence}")
    
except Exception as e:
    print(f"标准解析失败：{e}")
    
    # 尝试修复解析
    try:
        fixed_result = fixing_parser.parse(result.content)
        print("\n修复解析成功：")
        print(f"情感倾向：{fixed_result.sentiment}")
        print(f"置信度：{fixed_result.confidence}")
        print(f"关键观点：{fixed_result.key_points}")
        print(f"总结：{fixed_result.summary}")
        
    except Exception as e2:
        print(f"修复解析也失败：{e2}")
​

步骤 4：重试机制输出解析器

实现具有重试功能的输出解析器：

import os
from langchain.output_parsers import RetryOutputParser, PydanticOutputParser
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 定义数据模型
class ReviewAnalysis(BaseModel):
    """评论分析模型"""
    rating: int = Field(description="评分", ge=1, le=5)
    sentiment: str = Field(description="情感倾向", regex="^(positive|negative|neutral)$")
    keywords: List[str] = Field(description="关键词")
    confidence: float = Field(description="置信度", ge=0, le=1)

# 创建基础解析器
base_parser = PydanticOutputParser(pydantic_object=ReviewAnalysis)

# 创建重试解析器
retry_parser = RetryOutputParser(
    parser=base_parser,
    max_retries=3,
    delay=2
)

# 创建重试提示词
retry_prompt = PromptTemplate(
    template="""
你是一个评论分析专家，请分析以下用户评论：

评论内容：
{review}

{format_instructions}

如果无法按照要求格式输出，请说明原因并尝试重新格式化。
""",
    input_variables=["review"],
    partial_variables={
        "format_instructions": base_parser.get_format_instructions()
    }
)

# 测试重试功能
review_text = """
这款手机真的不错！屏幕很清晰，拍照效果也很好。电池续航能力很强，
一天重度使用都没有问题。唯一的缺点就是价格有点贵，但是总体来说还是很满意的。
强烈推荐购买！
"""

# 使用重试解析器
result = llm.invoke([SystemMessage(content="你是一个评论分析专家"), HumanMessage(content=review_text)])

try:
    analysis = retry_parser.parse(result.content)
    print("重试解析成功：")
    print(f"评分：{analysis.rating}")
    print(f"情感倾向：{analysis.sentiment}")
    print(f"关键词：{analysis.keywords}")
    print(f"置信度：{analysis.confidence}")
    
except Exception as e:
    print(f"重试解析失败：{e}")
​

完整示例

完整的结构化输出 Agent 系统

import os
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
from langchain.output_parsers import PydanticOutputParser, JsonOutputParser
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

class AnalysisRequest(BaseModel):
    """分析请求模型"""
    text: str = Field(description="待分析的文本")
    analysis_type: str = Field(description="分析类型", regex="^(sentiment|keyword|summary)$")
    requirements: Optional[List[str]] = Field(description="特殊要求")

class AnalysisResult(BaseModel):
    """分析结果模型"""
    analysis_type: str = Field(description="分析类型")
    confidence: float = Field(description="置信度", ge=0, le=1)
    key_findings: List[str] = Field(description="关键发现")
    summary: str = Field(description="总结")
    recommendations: List[str] = Field(description="建议")
    metadata: Dict[str, Any] = Field(description="元数据")

class StructuredOutputAgent:
    """结构化输出 Agent 系统"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.llm = ChatOpenAI(
            model="gpt-4-turbo",
            temperature=0.3,
            max_tokens=2000
        )
        self.setup_parsers()
        
    def setup_parsers(self):
        """设置解析器"""
        self.parsers = {
            'sentiment': PydanticOutputParser(pydantic_object=AnalysisResult),
            'keyword': JsonOutputParser(),
            'summary': PydanticOutputParser(pydantic_object=AnalysisResult)
        }
    
    def create_analysis_prompt(self, request: AnalysisRequest) -> PromptTemplate:
        """创建分析提示词"""
        
        prompts = {
            "sentiment": """
你是一个情感分析专家，请分析以下文本的情感倾向：

文本内容：{text}

要求：
1. 分析文本的整体情感倾向
2. 识别情感的关键触发点
3. 提供情感置信度
4. 给出基于情感的分析建议

{format_instructions}
""",
            "keyword": """
你是一个关键词提取专家，请从以下文本中提取关键词：

文本内容：{text}

要求：
1. 提取最重要的5-10个关键词
2. 为每个关键词分配权重
3. 分类关键词为不同类别

请输出JSON格式结果：
{format_instructions}
""",
            "summary": """
你是一个文本摘要专家，请为以下文本生成摘要：

文本内容：{text}

要求：
1. 保持原文的核心信息
2. 控制摘要长度在200-300字
3. 突出重要观点

{format_instructions}
"""
        }
        
        template = prompts.get(request.analysis_type, prompts["sentiment"])
        format_instructions = self.parsers[request.analysis_type].get_format_instructions()
        
        return PromptTemplate(
            template=template,
            input_variables=["text"],
            partial_variables={"format_instructions": format_instructions}
        )
    
    def analyze_text(self, request: AnalysisRequest) -> AnalysisResult:
        """分析文本"""
        
        prompt_template = self.create_analysis_prompt(request)
        prompt = prompt_template.format_prompt(text=request.text)
        
        result = self.llm.invoke([
            SystemMessage(content=f"你是一个{request.analysis_type}分析专家"),
            HumanMessage(content=request.text)
        ])
        
        try:
            parsed_result = self.parsers[request.analysis_type].parse(result.content)
            return parsed_result
        except Exception as e:
            return AnalysisResult(
                analysis_type=request.analysis_type,
                confidence=0.0,
                key_findings=[],
                summary="分析过程中出现错误",
                recommendations=["请检查输入文本或重试"],
                metadata={"error": str(e)}
            )

# 使用完整系统
print("=== 完整的结构化输出 Agent 系统 ===\")
agent = StructuredOutputAgent("your-openai-api-key")

# 创建分析请求
request = AnalysisRequest(
    text="今天天气很好，心情也很不错。这个产品功能很强大，用户体验也很好。",
    analysis_type="sentiment",
    requirements=["详细分析情感倾向", "提供改进建议"]
)

# 分析文本
result = agent.analyze_text(request)
print("分析结果：")
print(f"分析类型：{result.analysis_type}")
print(f"置信度：{result.confidence}")
print(f"关键发现：{result.key_findings}")
print(f"总结：{result.summary}")
print(f"建议：{result.recommendations}")
​

常见问题 FAQ

Q1：如何选择合适的输出解析器？

A：根据数据复杂度选择：简单数据用 JsonOutputParser，复杂数据结构用 PydanticOutputParser，需要容错时用 OutputFixingParser，需要可靠性时用 RetryOutputParser。

Q2：如何处理解析失败的情况？

A：首先检查 LLM 输出是否符合预期格式，然后使用修复解析器或重试解析器，最后可以实现降级策略，如提供默认值或错误信息。

Q3：如何优化结构化输出的性能？

A：合理设置 max_tokens 参数，使用温度参数控制输出确定性，提供清晰的格式说明，实现缓存机制，批量处理请求。

最佳实践与避坑

最佳实践

1. 模型设计：使用清晰的字段命名和类型注解
2. 格式说明：提供详细的格式说明和示例
3. 错误处理：实现完善的错误处理和恢复机制
4. 性能优化：合理设置参数，避免不必要的重试
5. 测试覆盖：为不同场景编写测试用例

常见陷阱

1. 格式不匹配：LLM 输出格式与预期不符
2. 数据类型错误：数据类型验证失败
3. 字段缺失：必需字段缺失
4. 嵌套结构复杂：复杂嵌套结构解析困难
5. 性能瓶颈：解析器性能问题

本节小结

本节详细介绍了 LangChain Agent 结构化输出和输出解析的核心概念和实现方法，包括：

• 结构化输出原理：理解结构化输出的重要性和基本原理
• Pydantic 模型应用：使用 Pydantic 模型定义复杂的数据结构
• JSON Schema 解析：使用 JSON Schema 灵活定义输出格式
• 自动修复机制：实现能够自动修复错误的解析器
• 重试机制：确保输出解析的可靠性
• 完整实践：从基础配置到多功能 Agent 系统的完整实现

掌握结构化输出是构建高质量 Agent 的关键技术。在下一节中，我们将继续学习 LangGraph 状态图基础，进一步提升 Agent 的编排和控制能力。

延伸阅读

• 官方文档：LangChain 输出解析器文档
• 相关章节：本教程 3.1 节 StateGraph 状态图基础
• 推荐资源：《Pydantic 数据模型设计》
• 开源项目：LangChain 结构化输出示例
• 技术博客：LLM 结构化输出最佳实践

关键词：结构化输出,LangChain,输出解析器,Pydantic,JSON Schema,Agent架构,数据模型,教程,实战,性能优化
难度：进阶
预计阅读：45分钟