一、概述

LangChain中的智能体（Agents）是具有自主决策能力的AI系统，能够根据用户输入、可用工具和环境反馈来决定执行哪些操作。一句话定义：Agent 是一个能自主思考、自主规划、自主调用工具、自主纠正错误的 AI 系统。

1.1 智能体核心定义

智能体 = 大模型 + 工具 + 记忆 + 决策能力。

• 突破大模型静态回答局限，能自主思考→选工具→执行→看结果→再思考，直到完成任务。
• 智能体不是“一次性回答”，它是一个会反复思考和行动的循环系统。
• LangChain 1.0 统一用 create_agent() 构建，替代旧版 create_react_agent 等，更简洁、生产就绪。

智能体与传统链式调用的区别：

• 普通链（Chain）：固定流程，输入→处理→输出
• 智能体（Agent）：动态流程，会思考、试错、循环、多步执行

1.2 智能体核心能力

1. 理解用户意图
2. 判断是否需要调用工具
3. 自主选择工具
4. 根据工具结果反思
5. 多轮循环直到完成任务
6. 记住上下文（记忆）
7. 可中断、可恢复、可持久化（LangGraph）

1.3 智能体核心范式：ReAct（Reason + Act）

智能体的 “思考 - 行动” 标准流程（create_agent 自动实现，无需手动写循环）：

1. Reason（推理）：LLM 分析用户问题，判断是否需要工具、用哪个工具、传什么参数。
2. Act（行动）：调用对应工具执行操作（如查天气、算数学、搜网页）。
3. Observe（观察）：接收工具返回结果，LLM 整合信息，判断是否继续或直接回答。
4. 循环直到：生成最终答案 / 达到最大迭代次数（默认 15 次）。

1.4 智能体核心组件

1. 模型（Model）：智能体的 “大脑”

   • 作用：负责推理、决策、生成回答，是智能体的核心
   • 支持：OpenAI、Anthropic、Google、DeepSeek 等主流 LLM，统一接口可无缝切换LangChain
   • 配置方式：
     • 静态（常用）：创建时固定模型（如 ChatOpenAI(model=“gpt-4o”)）
     • 动态：运行时根据状态选模型（成本优化 / 复杂路由）

2. 工具（Tools）：智能体的 “手脚”

   • 作用：让智能体与外部交互（联网、计算、查库、操作文件等）
   • 定义：用 @tool 装饰器快速封装函数，必须包含名称、描述、参数 schema（LLM 靠描述判断是否调用）
   • 内置工具：搜索（Tavily）、计算器、数据库、API 调用、文件读写等
   • 自定义工具：任何 Python 函数都能包装成工具

3. 系统提示词（System Prompt）：智能体的 “行为准则”

   • 作用：定义智能体角色、能力边界、输出格式、约束规则
   • 示例：“你是专业助手，仅用提供的工具回答，禁止编造信息，步骤清晰”
   • 优化：1.0 支持动态提示词（结合中间件实时修改）

4. 中间件（Middleware）：智能体的 “插件系统”（1.0 核心）

   • 作用：不修改核心逻辑，扩展 / 控制智能体行为（模型调用、工具执行、状态管理）
   • 常用内置中间件：
     • HumanInTheLoopMiddleware：敏感工具执行前需人工审批（如查隐私）
     • ModelCallLimitMiddleware：限制模型调用次数，防止无限循环
     • LoggingMiddleware：记录执行日志，调试 / 监控
     • 自定义中间件：实现个性化逻辑（如缓存、参数校验）

5. 状态（State）：智能体的 “短期记忆”

   • 作用：自动维护对话历史、工具调用记录、中间结果，支持多轮对话
   • 1.0 特性：基于 AgentState（TypedDict），可自定义扩展（如存储用户偏好）

二、LangChain 1.0 vs LangChain 0.x

2.1 关键区别

维度	LangChain 0.x	LangChain 1.0
创建方式	多入口（create_react_agent/create_json_agent）	统一 create_agent 一个函数搞定
架构	Chain 线性组合	Agent 循环 + 中间件（洋葱模型）
生产能力	弱，需手动实现	内置：状态管理、流式、重试、限流
扩展性	低，改源码	高，中间件无侵入扩展

2.2 0.x 时代的问题

在1.0之前，LangChain有一个严重问题：做Demo很快，上线就崩。典型痛点：

1. Chain太死板：业务逻辑稍微复杂一点，框架就套不住，最后不得不手写LLM调用。
2. 上下文溢出：对话一长，Token爆炸，系统直接挂掉。
3. 敏感信息泄露：用户的身份证、邮箱被直接发给了OpenAI。
4. 高危操作无审批：Agent一声不吭就把邮件发出去了、把数据库删了。
5. 换模型等于重写：OpenAI切到国产模型，Tool Calling格式不一样，适配代码重写一遍。

示例：

# LangChain 0.x - 参数爆炸，全靠手写
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=15,
    max_execution_time=300,
    handle_parsing_errors=True,
    return_intermediate_steps=True,
    trim_intermediate_steps=10,
    # 还有几十个参数...
)

2.3 1.0 的核心革新

LangChain团队分析了数百个生产级Agent，发现一个事实：绝大多数成功案例，底层都是同一个模式——ReAct（Reason + Act）。于是1.0做了一个极其大胆的决定：

1. 砍掉所有Chain，只保留ReAct Agent
2. 把Agent构建压缩成一个函数：create_agent
3. 把生产级能力做成“插件”：Middleware（中间件）

三、智能体的核心骨架（create_agent）

3.1 最小的智能体（3个参数就够了）

三个必填参数，缺一不可：

参数	作用	新手理解
model	谁在思考	智能体的大脑
tools	能干什么	智能体的手脚
system_prompt	遵守什么规则	智能体的岗位说明书

from langchain.agents import create_agent
from langchain.tools import tool

# 1. 定义一个工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}今天晴天，24℃"

# 2. 创建智能体——就这么简单！
agent = create_agent(
    model="openai:gpt-4o",           # ① 模型
    tools=[get_weather],            # ② 工具
    system_prompt="你是一个天气助手，回答简洁友好。"  # ③ 系统提示
)

# 3. 调用
result = agent.invoke({
    "messages": [{"role": "user", "content": "北京天气怎么样？"}]
})
print(result["messages"][-1]["content"])

3.2 底层原理：ReAct循环（3步走）

create_agent内部是一个无限循环，直到满足结束条件：

① 思考（Reason） → ② 行动（Act） → ③ 观察（Observe） → 回到①

用天气助手的例子走一遍：

1. 思考：用户问北京天气，我有get_weather工具，应该调用它
2. 行动：执行get_weather(city=“北京”)
3. 观察：得到返回“北京今天晴天，24℃”
4. 思考：信息已获取，直接回答用户 → 终止循环

四、生产级能力插件（Middleware）

4.1 中间件是什么？

技术定义：中间件是挂载在Agent执行流程上的钩子函数，可以在不修改核心循环的前提下，插入任何自定义逻辑。

类比：快递分拣线

• 智能体是快递员（负责跑腿）
• 中间件是分拣线上的检查站
  • PII检查站：把包裹上的手机号涂掉（脱敏）
  • 总结检查站：包裹太多时，写个摘要（长对话压缩）
  • 审批检查站：贵重包裹需要主管签字（人工审核）

4.2 中间件的5个挂载点

中间件可以在以下5个关键时刻介入：

用户输入 → [before_agent] → [before_model] → [wrap_model_call] → 模型调用 → 
[after_model] → [wrap_tool_call] → 工具执行 → [after_agent] → 输出

钩子	执行时机	典型用途
before_agent	智能体开始工作前	加载用户历史、权限校验
before_model	每次调用模型前	修剪太长对话、注入临时提示
wrap_model_call	包裹模型调用	动态切换模型/工具集
wrap_tool_call	包裹工具调用	工具权限控制、计费
after_model	每次模型返回后	输出格式校验、敏感词过滤
after_agent	智能体结束后	清理资源、保存对话

4.3 内置中间件（开箱即用）

LangChain 1.0内置了4个最常用的中间件，覆盖90%的生产需求。

4.3.1 PIIMiddleware——隐私保护

作用：在发送给模型之前，自动处理身份证、邮箱、手机号等敏感信息。

from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model="openai:gpt-4o",
    tools=[...],
    middleware=[
        # 邮箱脱敏：john@example.com → j***@e***.com
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        
        # 信用卡掩码：只显示后4位
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
        
        # 自定义API密钥：检测到直接阻断
        PIIMiddleware(
            "api_key",
            detector=r"sk-[a-zA-Z0-9]{32}",  # OpenAI密钥格式
            strategy="block"
        ),
    ]
)

为什么要在中间件做？不能靠模型自己识别吗？
答：永远不要把隐私安全交给模型。中间件是在发送请求之前就处理掉，模型根本看不到原始数据。

4.3.2 SummarizationMiddleware——对话总结

作用：对话历史太长时，自动生成摘要，避免Token溢出。

from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="openai:gpt-4o",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="openai:gpt-4o-mini",  # 用便宜的模型做总结
            max_tokens_before_summary=4000,  # 超过4000 Token触发
            messages_to_keep=20  # 保留最近20条不总结
        )
    ]
)

执行逻辑：

• 检查当前消息列表的Token总数
• 超过阈值 → 调用gpt-4o-mini生成摘要
• 将摘要插入对话，移除早期消息

4.3.3 HumanInTheLoopMiddleware——人工审批

作用：高危操作必须经过人类确认。

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
    model="claude-sonnet-4-5-20250929",
    tools=[send_email, delete_file],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": {  # 发送邮件前必须审批
                    "allowed_decisions": ["approve", "edit", "reject"]
                },
                "delete_file": {  # 删除文件直接禁止
                    "allowed_decisions": ["reject"]
                }
            }
        )
    ]
)

执行效果：当智能体试图调用send_email时，执行暂停，等待API返回审批结果。

4.3.4 ToolRetryMiddleware——工具重试

作用：工具调用失败时自动重试，支持指数退避。

from langchain.agents.middleware import ToolRetryMiddleware

agent = create_agent(
    model="openai:gpt-4o",
    tools=[database_tool],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,        # 最多重试3次
            backoff_factor=2.0,   # 延迟翻倍：1s → 2s → 4s
            initial_delay=1.0,    # 首次延迟1秒
            max_delay=60.0,       # 最大延迟60秒
            jitter=True           # 增加随机抖动，避免并发风暴
        )
    ]
)

4.4 自定义中间件

内置中间件覆盖通用场景，自定义中间件才是解决业务痛点的关键。

案例1：用户分级——给高手用重武器，给新手用安全模式

场景：根据用户技术水平，动态分配不同模型和工具集。

from dataclasses import dataclass
from langchain.agents.middleware import AgentMiddleware
from langchain_openai import ChatOpenAI

# 1. 定义上下文结构（存用户信息）
@dataclass
class Context:
    user_expertise: str = "beginner"  # 默认新手

# 2. 自定义中间件
class ExpertiseBasedToolMiddleware(AgentMiddleware):
    def wrap_model_call(self, request, handler):
        # 从运行时上下文读取用户等级
        user_level = request.runtime.context.user_expertise
        
        if user_level == "expert":
            # 专家用户：用最强模型 + 高级工具
            model = ChatOpenAI(model="gpt-5")
            tools = [advanced_search, data_analysis]
        else:
            # 普通用户：轻量模型 + 基础工具
            model = ChatOpenAI(model="gpt-5-nano")
            tools = [simple_search, basic_calculator]
        
        # 覆盖请求中的模型和工具
        return handler(request.override(model=model, tools=tools))

# 3. 使用
agent = create_agent(
    model="default",  # 会被中间件覆盖
    tools=[simple_search, advanced_search, basic_calculator, data_analysis],
    middleware=[ExpertiseBasedToolMiddleware()],
    context_schema=Context
)

# 调用时传入用户等级
result = agent.invoke(
    {"messages": [...]},
    context={"user_expertise": "expert"}  # 动态切换
)

这个模式解决了什么问题？

• 以前：写两个Agent，用if-else判断调用哪个
• 现在：一个Agent，运行时自动适配

案例2：质量门禁——自动校验输出质量

场景：研究助手必须输出至少3个核心观点、2个来源，否则让模型重试。

from langchain.agents.middleware import AgentMiddleware

class ValidationMiddleware(AgentMiddleware):
    def after_model(self, state, runtime):
        """模型返回后执行"""
        if "structured_response" in state:
            report = state["structured_response"]
            
            # 校验1：必须有3条以上的关键发现
            if len(report.key_findings) < 3:
                return {
                    "messages": [{
                        "role": "user",
                        "content": "请提供至少3个关键发现。"
                    }]
                }
            
            # 校验2：必须有2个以上的来源
            if len(report.sources) < 2:
                return {
                    "messages": [{
                        "role": "user", 
                        "content": "请引用至少2个权威来源。"
                    }]
                }
        
        return None  # 校验通过，继续执行

# 使用
agent = create_agent(
    model="gemini-2.0-flash-exp",
    tools=[...],
    middleware=[ValidationMiddleware()],
    response_format=ResearchReport
)

执行效果：如果模型偷懒只写了2个观点，中间件自动追加一条用户消息，让模型重试。用户无感知，但输出质量有保障。

4.5 中间件设计哲学总结

一句话记住中间件：Agent是固定的高速公路，中间件是沿途的服务区——想加什么功能，建个服务区就行。

维度	0.x 时代	1.0 中间件
上下文控制	散落在各个参数、回调函数	集中式、可组合的钩子
复用性	每个项目重写	内置+自定义，跨项目共享
测试难度	需要跑完整Agent	可单独测试中间件
学习曲线	参数爆炸，文档混乱	5个钩子，一通百通

五、系统提示（System Prompt）

系统提示（System Prompt）是开发者预先定义的、传递给大模型 / 智能体的 “规则说明书” —— 它不参与用户的对话交互，但会全程约束 AI 的行为、角色、能力边界、输出格式，是 LangChain 智能体（Agent）的 “行为准则”。

在 LangChain 中，系统提示区别于用户输入（HumanMessage）和 AI 回复（AIMessage），属于全局、前置、优先级最高的提示信息，直接决定智能体的核心表现。与普通提示词的区别：

类型	作用范围	优先级	核心目的	应用场景
系统提示（System Prompt）	全局 / 全程	最高	定义角色、规则、能力边界	智能体初始化时配置
用户提示（User Prompt）	单次交互	次高	传递用户具体需求	每次调用智能体时传入
工具提示（Tool Prompt）	工具调用场景	辅助	说明工具用法、参数	智能体绑定工具时自动生成

5.1 核心作用

1. 定义智能体的 “角色身份”

   • 让智能体明确自己的定位，避免回答偏离场景，比如：
     • “你是专业的电商客服智能体，仅处理订单查询、物流跟踪、退款申请相关问题”
     • “你是数据分析助手，仅使用提供的工具查询销售额数据，不回答无关问题”

2. 约束智能体的 “能力边界”

   • 避免智能体编造信息、越权回答，是生产级应用的核心风控手段，比如：
     • “仅使用提供的工具查询数据，不编造任何未验证的信息”
     • “拒绝回答与公司业务无关的问题，如时政、娱乐、敏感话题”

3. 规范智能体的 “输出格式”

   • 让智能体返回结构化、可直接解析的结果，避免手动处理文本，比如：
     • “回答必须以 JSON 格式返回，包含字段：result（结果）、source（数据来源）”
     • “分析报告需分 3 部分：数据摘要、核心结论、建议，每部分不超过 200 字”

4. 指导智能体的 “工具调用逻辑”

   • 这是 LangChain Agent 最核心的作用 —— 告诉智能体 “什么时候调用工具、怎么调用、调用失败怎么办”，比如：
     • “当用户询问销售额时，必须先调用 query_sales 工具，使用工具返回的结果回答”
     • “工具调用失败时，最多重试 2 次，重试失败后告知用户‘数据查询失败’”

5. 补充智能体的 “上下文信息”

   • 传递固定的背景信息，避免用户重复说明，比如：
     • “当前查询的是 2025 年 Q4 的销售数据，公司业务范围仅覆盖国内一线城市”
     • “用户 ID 为 1001，默认查询的是该用户名下的订单数据”

5.2 基础用法（单字符串）

最常用，直接传入简洁的字符串，框架会自动结合工具信息、记忆上下文生成完整提示词：

from langchain_core.agents import create_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# 1. 定义工具
@tool
def query_sales(quarter: str) -> str:
    """查询指定季度销售额，quarter格式为"2025年Q4"."""
    return f"{quarter}销售额：1280万元，同比增长15.3%"

# 2. 初始化模型
llm = ChatOpenAI(model="gpt-4o", api_key="你的API密钥")

# 3. 核心：定义系统提示（单字符串）
system_prompt = """
你是专业的销售数据分析智能体，严格遵守以下规则：
1. 角色：仅处理销售额查询、同比增长率分析相关问题；
2. 工具：必须调用 query_sales 工具获取数据，不编造信息；
3. 输出：回答简洁，仅返回数据+核心结论，不超过50字；
4. 边界：拒绝回答与销售额无关的问题。
"""

# 4. 创建Agent（直接传入system_prompt）
agent = create_agent(
    model=llm,
    tools=[query_sales],
    system_prompt=system_prompt  # LangChain 1.0 核心配置
)

# 5. 执行
executor = AgentExecutor(agent=agent, tools=[query_sales], verbose=True)
executor.invoke({"input": "2025年Q4销售额是多少？"})

5.3 进阶用法（动态模板）

结合变量动态生成系统提示，适配多场景复用，比如根据用户 ID 动态绑定上下文：

from langchain_core.prompts import SystemMessagePromptTemplate

# 1. 定义带变量的系统提示模板
template = """
你是销售数据分析智能体，规则如下：
1. 仅处理用户{user_id}名下的销售数据；
2. 默认查询季度为{default_quarter}；
3. 必须调用 query_sales 工具，输出格式为JSON。
"""

# 2. 渲染模板（动态传入变量）
system_prompt = SystemMessagePromptTemplate.from_template(template).format(
    user_id="1001",
    default_quarter="2025年Q4"
).content

# 3. 后续创建Agent的逻辑和基础用法一致
agent = create_agent(model=llm, tools=[query_sales], system_prompt=system_prompt)

5.4 结合工具信息（自动增强）

LangChain 1.0 会自动将工具的 name、description、parameters 整合到系统提示中，无需手动编写工具说明，比如：

• 你定义的工具：query_sales(quarter: str) -> str: “查询指定季度销售额”
• 框架自动补充到系统提示的内容：

你可以调用以下工具：
- 工具名：query_sales
- 描述：查询指定季度销售额
- 参数：quarter（字符串，格式为"2025年Q4"）

5.5 编写高质量系统提示的 6 条核心原则

1. 角色明确，不模糊
   • 错误：“你是一个助手”（太泛，智能体不知道该做什么）
   • 正确：“你是电商订单查询助手，仅处理订单状态、物流信息查询”
2. 规则可量化，不抽象
   • 错误：“回答要简洁”（无量化标准）
   • 正确：“回答不超过 50 字，仅包含核心数据，无多余解释”
3. 工具调用逻辑清晰
   • 错误：“可以调用工具查询数据”（未说明触发条件）
   • 正确：“当用户询问销售额时，必须调用 query_sales 工具，仅使用工具返回结果回答”
4. 边界清晰，拒绝兜底
   • 错误：“尽量回答用户的问题”（易编造信息）
   • 正确：“仅回答工具能查询到的信息，工具无结果时回复‘暂无相关数据’”
5. 输出格式标准化
   • 错误：“返回分析结果”（格式不固定）
   • 正确：“返回 JSON 格式，字段：{“quarter”: “季度”, “sales”: “销售额”, “growth”: “同比增长率”}”
6. 语言简洁，避免冗余
   • 错误：长篇大论讲背景、无关规则（模型易忽略核心）
   • 正确：只保留角色、规则、格式、工具逻辑 4 类核心信息

5.6 系统提示常见问题及解决方案

1. 问题 1：智能体不调用工具，直接编造答案
   • 原因：系统提示未明确 “必须调用工具”，或工具描述不清晰
   • 解决方案：

     system_prompt = """
     你必须调用 query_sales 工具获取销售额数据，禁止编造任何信息。
     步骤：
     1. 接收用户问题后，先调用工具；
     2. 仅使用工具返回的结果回答；
     3. 工具无结果时，回复“暂无该季度数据”。
     """
     

2. 问题 2：智能体输出格式不符合要求
   • 原因：系统提示未明确格式，或格式描述不具体
   • 解决方案：

     system_prompt = """
     回答必须严格遵循以下JSON格式，无任何额外文本：
     {
       "quarter": "2025年Q4",
       "sales": "1280万元",
       "growth": "15.3%"
     }
     """
     

3. 问题 3：智能体回答无关问题，越权响应
   • 原因：边界规则不清晰，未明确 “拒绝场景”
   • 解决方案：

     system_prompt = """
     仅回答销售额相关问题，拒绝回答以下类型问题：
     1. 时政、娱乐、敏感话题；
     2. 公司人事、财务（非销售）数据；
     3. 与2025年Q4无关的历史数据（除非用户明确指定）。
     拒绝时回复：“该问题超出我的处理范围，请咨询相关专业人员。”
     """
     

4. 问题 4：LangChain 1.0 系统提示不生效
   • 原因：
     • 系统提示传入位置错误（旧版本用 prompt，1.0 必须用 system_prompt）；
     • 系统提示过长，超出模型上下文窗口；
   • 解决方案：

     # 正确传入方式（1.0版本）
     agent = create_agent(
         model=llm,
         tools=[query_sales],
         system_prompt=short_system_prompt  # 精简提示词，控制长度
     )
     

六、结构化输出——让AI说“人话”变成说“数据”

结构化输出是指让 LangChain 调用的大模型 / 智能体，不再返回无规则的自然语言文本，而是返回固定格式、可直接解析的数据（如 JSON、字典、Pydantic 模型对象），是生产级 AI 应用的必备能力（避免手动解析文本的繁琐和错误）。

在 LangChain 中，结构化输出不是 “简单让模型输出 JSON 字符串”，而是通过框架层的约束 + 模型层的引导，实现：

• 输出格式 100% 符合预期（不会出现语法错误）；
• 输出内容可直接映射到代码中的数据结构（如类、字典）；
• 支持校验和纠错（格式错误时自动重试）。

6.1 为什么需要结构化输出？

简单来说：结构化输出是 AI 应用和业务系统对接的 “桥梁”。

场景	非结构化输出（自然语言）	结构化输出（JSON/Pydantic）
数据分析	“2025 年 Q4 销售额 1280 万，增长 15.3%”	{“quarter”:“2025Q4”,“sales”:1280,“growth”:15.3}
订单查询	“订单 12345 已发货，物流单号 7890”	{“order_id”:“12345”,“status”:“已发货”,“logistics_id”:“7890”}
代码调用	需要手动提取字段、处理类型转换，易出错	可直接作为参数传入函数，无需额外处理

6.2 结构化输出的实现方式

LangChain中实现结构化输出的各种方法，包括传统方式（OutputParser）以及LangChain 1.0的新方式（response_format），可能还有通过Tool calling的伪装等。可以按照以下几种实现方式分类：

1. 基于输出解析器（OutputParser）的传统方法：PydanticOutputParser, JsonOutputParser, StructuredOutputParser等。
2. 基于模型原生结构化输出（如OpenAI的JSON mode、函数调用）的方法：LangChain通过response_format参数或者绑定工具的方式。
3. LangChain 1.0中的新方式：create_agent中的response_format参数，以及底层如何适配不同模型。
4. 通过工具调用来伪装结构化输出（Tool Strategy）：对于不支持原生结构化的模型，将Pydantic模型包装成工具调用。

方式 1：输出解析器（OutputParser）——传统但灵活

这是 LangChain 0.x 时代就存在的方式，通过提示工程让模型输出特定格式的文本，然后用解析器将其转为结构化数据。

• 原理：提示模型输出特定格式，然后用解析器解析

  • 在提示词中加入格式说明（例如“请输出 JSON 格式”）。
  • 模型返回一段文本（如 JSON 字符串）。
  • 输出解析器解析这段文本，返回 Python 对象。

• 常用解析器：

  • PydanticOutputParser：基于 Pydantic 模型，最强大。
  • JsonOutputParser：直接解析 JSON，无需定义模型。
  • CommaSeparatedListOutputParser：解析逗号分隔的列表。
  • StructuredOutputParser：返回多个命名字段（值都是字符串）。
  • DatetimeOutputParser：解析日期时间。
  • EnumOutputParser：将输出映射到枚举值。

• 示例：PydanticOutputParser使用

  from langchain.output_parsers import PydanticOutputParser
  from langchain.prompts import PromptTemplate
  from langchain_openai import ChatOpenAI
  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 = PromptTemplate(
      template="从以下文本提取信息：\n{input}\n\n{format_instructions}",
      input_variables=["input"],
      partial_variables={"format_instructions": parser.get_format_instructions()}
  )
  
  # 4. 调用模型
  model = ChatOpenAI()
  chain = prompt | model | parser
  result = chain.invoke({"input": "张三今年25岁"})
  print(result)  # Person(name='张三', age=25)
  

• 优点：

  • 模型无关：任何文本生成模型都能用。
  • 高度可控：可以通过提示词精细控制格式。

• 缺点：

  • 依赖提示词质量：模型可能不遵守格式，导致解析失败。
  • 稳定性一般：复杂格式时模型容易出错。

方式 2：模型原生结构化输出（JSON mode / 函数调用）

现在的主流模型（OpenAI、Claude、Gemini 等）都提供了原生结构化输出能力，例如 OpenAI 的 JSON mode 和函数调用（Function Calling）。LangChain 1.0 对此做了深度集成，让开发者以统一的方式使用这些能力。

• 原理：利用模型API本身支持的结构化输出参数（如OpenAI的response_format、函数调用）

  • JSON mode：通过 API 参数强制模型输出合法 JSON。
  • 函数调用：让模型选择调用一个“函数”，并将函数参数作为结构化输出。

• LangChain中的实现：

  1. 使用 .with_structured_output()（LCEL 风格）

     • 这是最直接的 LCEL 方法，适用于任何支持结构化输出的模型。
     • 底层自动判断：若模型支持函数调用，就使用函数调用；若支持 JSON mode，就使用 JSON mode。

     from langchain_openai import ChatOpenAI
     from pydantic import BaseModel
     
     class Person(BaseModel):
         name: str
         age: int
     
     model = ChatOpenAI(model="gpt-4o")
     structured_model = model.with_structured_output(Person)
     
     result = structured_model.invoke("张三25岁")
     print(result)  # Person(name='张三', age=25)
     

  2. 在 create_agent 中使用 response_format 参数

     • 这是 LangChain 1.0 智能体中的标准用法。

     from langchain.agents import create_agent
     
     agent = create_agent(
         model="openai:gpt-4o",
         tools=[...],
         response_format=Person  # 直接传入 Pydantic 模型
     )
     result = agent.invoke({"messages": [{"role": "user", "content": "张三25岁"}]})
     person = result["structured_response"]  # 直接得到 Person 对象
     

  3. 通过绑定工具（bind_tools）让模型返回工具调用作为结构化数据

     from langchain.tools import tool
     from langchain_core.pydantic_v1 import BaseModel
     
     class Person(BaseModel):
         name: str
         age: int
     
     @tool(args_schema=Person)
     def dummy_tool(name: str, age: int):
         """这是一个占位工具，只用于获取参数"""
         pass
     
     model = ChatOpenAI().bind_tools([dummy_tool])
     response = model.invoke("张三25岁")
     # 解析 response.tool_calls[0]["args"] 得到结构化数据
     

• 优点：

  • 最稳定：模型 API 原生保证格式正确。
  • 无需提示工程：格式说明由框架自动处理。
  • 性能好：减少 token 浪费。

• 缺点：

  • 依赖模型能力：需要模型支持 JSON mode 或函数调用。
  • 可能有限制：某些模型对输出结构有大小限制。

方式 3：通过工具调用伪装（Tool Strategy）——兼容方案

当模型不支持原生结构化输出，但支持工具调用（例如大多数开源模型通过适配也支持）时，LangChain 1.0 采用了一种巧妙的方法：将 Pydantic 模型伪装成一个工具。

• 原理：对于不支持原生结构化的模型，将Pydantic模型包装成一个“工具”，让模型通过调用工具来返回结构化数据。模型会输出一个工具调用，我们解析工具调用的参数作为结构化输出。

  • 将目标数据结构定义为工具的 args_schema。
  • 在提示词中告诉模型“你需要调用工具才能完成任务”。
  • 模型会输出一个工具调用，其参数就是我们需要的数据。
  • 解析工具调用的参数，得到结构化对象。

• LangChain实现：在 LangChain 1.0 中，如果你在 create_agent 中设置了 response_format，而模型不支持原生结构化，框架会自动采用这个策略，开发者完全无感。

• 示例：手动绑定工具并解析

  from langchain.tools import tool
  from langchain_core.pydantic_v1 import BaseModel
  from langchain_openai import ChatOpenAI
  
  class Person(BaseModel):
      name: str
      age: int
  
  @tool(args_schema=Person)
  def return_person(name: str, age: int):
      """此工具用于返回人物信息"""
      return {"name": name, "age": age}
  
  model = ChatOpenAI(model="gpt-3.5-turbo").bind_tools([return_person])
  response = model.invoke("张三25岁")
  tool_call = response.tool_calls[0]
  args = tool_call["args"]
  person = Person(**args)  # 结构化输出
  print(person)
  

• 优点：

  • 兼容性好：适用于大量支持工具调用的模型。
  • 比传统解析器可靠：工具调用通常比自由文本格式稳定。

• 缺点：

  • 需要模型支持工具调用（大多数现代模型都支持）。
  • 可能会额外消耗一次工具调用额度（如果有计费）。

方式 4：自定义解析和验证（结合中间件）

在实际生产环境中，模型偶尔会输出格式错误的内容。这时我们可以通过输出修复解析器或中间件来增强鲁棒性。

• 原理：在中间件中对模型输出进行校验和修复，结合OutputFixingParser等，实现弹性结构化

  • 使用 OutputFixingParser 或 RetryWithErrorOutputParser 包裹基础解析器。
  • 当解析失败时，自动调用另一个 LLM 来修复或重试。

• 示例：使用OutputFixingParser包裹解析器

  from langchain.output_parsers import PydanticOutputParser, OutputFixingParser
  from langchain_openai import ChatOpenAI
  from pydantic import BaseModel
  
  class Person(BaseModel):
      name: str
      age: int
  
  base_parser = PydanticOutputParser(pydantic_object=Person)
  fixing_parser = OutputFixingParser.from_llm(
      parser=base_parser,
      llm=ChatOpenAI()
  )
  
  bad_json = '{"name": "张三", "age": 25'  # 缺少结束括号
  person = fixing_parser.parse(bad_json)   # 自动修复并解析
  print(person)
  

• 结合中间件（LangChain 1.0）
  可以在智能体的中间件中插入校验逻辑，当结构化输出不符合预期时，让模型重试。

  from langchain.agents.middleware import AgentMiddleware
  
  class ValidationMiddleware(AgentMiddleware):
      def after_model(self, state, runtime):
          if "structured_response" in state:
              data = state["structured_response"]
              if data.age < 0:
                  # 让模型重新生成
                  return {"messages": [{"role": "user", "content": "年龄不能为负，请修正"}]}
          return None
  

• 优点：

  • 极高的鲁棒性：能处理各种边缘情况。
  • 可自定义修复逻辑：不仅限于格式，还能校验业务规则。

• 缺点：

  • 增加额外 LLM 调用，可能提高成本和延迟。
  • 需要合理设计重试策略，避免死循环。

6.3 结构化输出的核心优化技巧

1. 模型选择：优先用强工具调用能力的模型
   • 推荐：GPT-4o、GPT-4 Turbo、通义千问 4、文心一言 4.0（结构化输出稳定性高）；
   • 不推荐：GPT-3.5 Turbo（低概率出现格式错误）；
   • 关键参数：temperature=0（关闭随机性，保证输出格式稳定）。
2. 格式说明：越具体越好
   • 错误：“输出 JSON 格式”（太泛，模型可能漏字段）；
   • 正确：使用 parser.get_format_instructions()（自动生成详细格式说明），包含字段名、类型、示例。
3. 错误处理：添加重试机制
   • 模型偶尔会输出格式错误，通过 LangChain 的 RetryOutputParser 实现自动重试：

     from langchain_core.output_parsers import RetryOutputParser
     from langchain_core.prompts import PromptTemplate
     
     # 包装解析器，添加重试逻辑
     retry_parser = RetryOutputParser.from_llm(
         parser=parser,
         llm=llm,
         prompt=prompt  # 重试时使用的提示词
     )
     
     # 调用（格式错误时自动重试）
     try:
         result = chain.invoke({"sales_info": sales_info})
     except Exception as e:
         result = retry_parser.parse_with_prompt(llm_response, prompt)
     

4. 字段约束：设置默认值和校验规则
   • 通过 Pydantic 模型添加字段校验，避免业务异常：

     from langchain_core.pydantic_v1 import BaseModel, Field, validator
     
     class SalesData(BaseModel):
         quarter: str = Field(description="季度")
         sales: float = Field(description="销售额", gt=0)  # 销售额必须大于0
         growth: float = Field(description="增长率", default=0.0)
     
         # 自定义校验规则
         @validator("quarter")
         def quarter_format(cls, v):
             if not v.startswith("202"):
                 raise ValueError("季度格式错误，必须以202开头（如2025Q4）")
             return v
     

5. response_format 本身是模型输出格式的参数，但在 LangChain 1.0 生产级落地中，需结合厂商适配（ProviderStrategy） 和工具调用（ToolStrategy） 形成完整策略体系：
   • ToolStrategy：使用人工工具调用生成结构化输出。这适用于任何支持工具调用的模型。

     from pydantic import BaseModel
     from langchain.agents import create_agent
     from langchain.agents.structured_output import ToolStrategy
     
     
     class ContactInfo(BaseModel):
         name: str
         email: str
         phone: str
     
     agent = create_agent(
         model="openai:gpt-4o-mini",
         tools=[search_tool],
         response_format=ToolStrategy(ContactInfo)
     )
     
     result = agent.invoke({
         "messages": [{"role": "user", "content": "从以下内容提取联系信息：John Doe, john@example.com, (555) 123-4567"}]
     })
     
     result["structured_response"]
     # ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')
     

   • ProviderStrategy：使用模型提供商的原生结构化输出生成。这更可靠，但仅适用于支持原生结构化输出的提供商（例如 OpenAI）：

     from langchain.agents.structured_output import ProviderStrategy
     
     agent = create_agent(
         model="openai:gpt-4o",
         response_format=ProviderStrategy(ContactInfo)
     )
     

6.4 结构化输出的核心优势

1. 与 LCEL 深度集成：支持 prompt | llm | parser 链式调用，代码简洁、可维护；
2. 强类型校验：通过 Pydantic 模型实现字段类型、范围、格式校验，避免业务错误；
3. 自动格式说明：parser.get_format_instructions() 自动生成详细格式要求，无需手动编写；
4. 与智能体无缝结合：1.0 版本的 create_agent() 可直接集成结构化输出规则，适配生产级场景；
5. 可扩展性强：支持自定义解析器，适配复杂业务格式（如 XML、CSV）。

七、标准内容块——跨模型兼容的终极方案

7.1 解决的问题

不同模型提供商的返回格式天差地别：

• OpenAI：response.choices[0].message.content
• Anthropic：response.content[0].text
• Gemini：response.candidates[0].content.parts[0].text

以前换个模型，解析代码全得重写。

7.2 1.0 的解决方案：content_blocks

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-sonnet-4-5-20250929")
response = model.invoke("What's the capital of France?")

# 统一的访问方式！
for block in response.content_blocks:
    if block["type"] == "reasoning":
        print(f"模型推理过程: {block['reasoning']}")
    elif block["type"] == "text":
        print(f"最终回答: {block['text']}")
    elif block["type"] == "tool_call":
        print(f"调用工具: {block['name']}({block['args']})")

优势：

• 一次编写，跑遍所有模型
• 支持推理轨迹、引用来源、内置工具等高级功能
• 完全类型提示，IDE自动补全

八、完整实战——把所有概念串起来

场景：
做一个真正的项目：自动写周报的智能体。

项目目标：
输入：“帮我写周报”，自动：

1. 从Jira获取本周完成的任务
2. 从Git统计代码提交
3. 生成结构化的周报Markdown
4. 敏感信息（任务ID）脱敏
5. 发送前请求人工审批

import os
from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.agents.middleware import (
    PIIMiddleware, 
    HumanInTheLoopMiddleware,
    AgentMiddleware
)
from langchain.tools import tool

# ---------- 1. 定义工具 ----------
@tool
def get_jira_tasks(assignee: str) -> str:
    """获取指定负责人在本周完成的任务"""
    # 真实场景会调Jira API，这里模拟数据
    return """
    PROJ-123: 修复登录页白屏问题
    PROJ-456: 优化数据库连接池配置
    PROJ-789: 更新用户文档
    """

@tool
def get_git_commits(author: str) -> str:
    """获取指定作者本周的代码提交"""
    return "合并PR #12, #15, #19；修复内存泄漏"

# ---------- 2. 定义输出结构 ----------
class WeeklyReport(BaseModel):
    summary: str = Field(description="本周工作整体概述")
    completed_tasks: list[str] = Field(description="完成任务列表")
    tech_details: str = Field(description="技术细节/代码提交")
    next_plan: str = Field(description="下周计划")

# ---------- 3. 自定义中间件：自动注入任务负责人 ----------
class InjectUserMiddleware(AgentMiddleware):
    def before_model(self, state, runtime):
        """模型调用前执行"""
        # 从上下文获取当前用户
        user = runtime.context.get("user", "zhangsan")
        # 将用户信息注入到消息中
        state["messages"].append({
            "role": "user",
            "content": f"我的Jira账号是{user}，Git账号也是{user}"
        })
        return None

# ---------- 4. 组装智能体 ----------
agent = create_agent(
    model="openai:gpt-4o",
    tools=[get_jira_tasks, get_git_commits],
    system_prompt="""你是一个周报助手。
    步骤1：调用get_jira_tasks获取任务
    步骤2：调用get_git_commits获取代码提交
    步骤3：按照规定的结构化格式生成周报
    """,
    middleware=[
        # ① 自动注入用户信息
        InjectUserMiddleware(),
        
        # ② 任务ID脱敏：PROJ-123 → PROJ-***
        PIIMiddleware(
            "jira_id",
            detector=r"PROJ-\d+",
            strategy="redact"
        ),
        
        # ③ 发送邮件前必须人工审批
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": {
                    "allowed_decisions": ["approve", "reject"]
                }
            }
        )
    ],
    response_format=WeeklyReport  # ④ 结构化输出
)

# ---------- 5. 执行 ----------
result = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "帮我写周报"}
        ]
    },
    context={"user": "wangwu"}  # 传入当前用户
)

# 6. 获取结构化数据
report = result["structured_response"]
print(f"## 周报摘要\n{report.summary}")
print("\n## 完成任务")
for task in report.completed_tasks:
    print(f"- {task}")