4.2 工具调用进阶与中间件

4.2 工具调用进阶与中间件 — LangChain框架精通 Agent高级能力

本节导读：学完本节，你将掌握 LangChain Agent 工具调用的进阶技巧——包括运行时上下文访问、复杂工具 Schema 定义、中间件机制、以及动态工具选择，让你的 Agent 从"能用"升级到"好用"。

学习目标

• 理解 ToolRuntime 机制，让工具访问会话状态、用户上下文和长期记忆
• 掌握 Pydantic 模型和 JSON Schema 定义复杂工具输入
• 理解中间件的工作原理，能用中间件实现日志记录、输入过滤和行为控制
• 掌握动态工具选择模式，让 Agent 根据场景自动获取不同的工具集

核心概念

上节我们学会了用 create_agent 创建基础 Agent，工具都是"无状态"的——它们只接收输入参数、返回结果，对 Agent 的运行环境一无所知。但在真实业务中，工具往往需要知道"这是哪个用户的请求""之前对话了什么""用户的历史偏好是什么"。

LangChain 通过 ToolRuntime 解决这个问题。ToolRuntime 是工具函数的一个特殊注入参数，它让工具能够访问 Agent 的运行时上下文，包括会话状态（State）、调用配置（Context）、长期存储（Store）等。

关键设计：runtime 参数对模型不可见。模型只会看到你定义的业务参数（如 query、user_id），而 runtime 是 Harness 自动注入的。这意味着你不需要在工具描述中解释运行时概念，模型也不会被无关信息干扰。

环境准备 / 前置知识

# 确保已安装 LangChain 核心和至少一个模型集成
pip install -qU langchain "langchain[openai]"
export OPENAI_API_KEY="your-api-key-here"
​

前置要求：

• 已掌握本教程 4.1 节的 create_agent 基础用法
• 理解 Python dataclass 和 Pydantic BaseModel
• 了解基本的中间件（Middleware）概念

分步实战

步骤 1：工具访问会话状态（State）

State 是 Agent 的短期记忆——当前会话中的所有消息和自定义字段。下面的例子展示了一个"智能计数器"工具，它能感知 Agent 已经调用了多少次工具：

from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain.messages import HumanMessage

@tool
def get_conversation_summary(runtime: ToolRuntime) -> str:
    """获取当前对话的摘要信息，包括消息轮数和最近的话题。"""
    messages = runtime.state["messages"]

    # 统计对话轮数
    user_messages = [m for m in messages if isinstance(m, HumanMessage)]
    turn_count = len(user_messages)

    # 获取最近 3 条消息的摘要
    recent = []
    for msg in messages[-3:]:
        role = "用户" if isinstance(msg, HumanMessage) else "助手"
        content = msg.content[:50] + "..." if len(msg.content) > 50 else msg.content
        recent.append(f"[{role}] {content}")

    return f"对话轮数：{turn_count}\n最近消息：\n" + "\n".join(recent)


@tool
def count_tool_calls(runtime: ToolRuntime) -> str:
    """统计本次会话中已经调用了多少次工具。"""
    messages = runtime.state["messages"]
    from langchain.messages import ToolMessage
    tool_count = len([m for m in messages if isinstance(m, ToolMessage)])
    return f"本次会话已执行 {tool_count} 次工具调用"


agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[get_conversation_summary, count_tool_calls],
    system_prompt="你是一个对话分析助手，可以查看对话统计信息。",
)

# 模拟多轮对话
result = agent.invoke({
    "messages": [
        {"role": "user", "content": "我们聊聊天气"},
        {"role": "assistant", "content": "好的，你想了解哪里的天气？"},
        {"role": "user", "content": "帮我看看当前对话的情况"},
    ]
})
print(result["messages"][-1].content)
​

注意 runtime: ToolRuntime 这个参数——它不需要你在调用时传入，Harness 会自动注入。模型看到的工具 schema 中也不会出现 runtime 参数。

步骤 2：工具访问用户上下文（Context）

Context 是在 invoke 时传入的不可变配置，典型用途是携带用户身份信息：

from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: str
    tier: str  # 用户等级

@tool
def get_user_recommendations(runtime: ToolRuntime) -> str:
    """根据用户的等级获取个性化推荐内容。"""
    ctx = runtime.context  # 获取传入的上下文
    tier = ctx.tier

    recommendations = {
        "free": "推荐免费课程和基础功能介绍",
        "pro": "推荐进阶教程和 API 高级用法",
        "enterprise": "推荐企业级部署方案和定制化服务",
    }
    return recommendations.get(tier, "未知用户等级")


# 创建 Agent 并传入用户上下文
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[get_user_recommendations],
    system_prompt="你是一个推荐助手，根据用户等级提供个性化推荐。",
)

# 调用时传入 context
result = agent.invoke(
    {"messages": [{"role": "user", "content": "有什么适合我的内容？"}]},
    config={
        "configurable": {"thread_id": "demo-thread"},
        "context": UserContext(user_id="user-123", tier="pro"),
    },
)
print(result["messages"][-1].content)
​

Context 和 thread_id 的区别很重要：thread_id 决定"这是哪段对话"（会话作用域），context 决定"这次调用附带了什么配置"（请求作用域）。在多轮对话中，thread_id 不变但每次 invoke 可以传不同的 context。

步骤 3：用 Pydantic 定义复杂工具 Schema

当工具的输入参数比较复杂时（有嵌套结构、枚举值、必选/可选字段混合），用 Pydantic 模型来定义 Schema 是最清晰的方式：

from pydantic import BaseModel, Field
from typing import Literal, Optional
from langchain.tools import tool

class DatabaseQueryInput(BaseModel):
    """数据库查询工具的输入参数定义。"""
    table: str = Field(description="要查询的表名，如 users / orders / products")
    conditions: dict = Field(
        default_factory=dict,
        description="查询条件，例如 {'status': 'active', 'age': {'gt': 18}}"
    )
    fields: list[str] = Field(
        default=["*"],
        description="要返回的字段列表，空列表或 ['*'] 表示返回所有字段"
    )
    sort_by: Optional[str] = Field(
        default=None,
        description="排序字段，加 '-' 前缀表示降序，如 '-created_at'"
    )
    limit: int = Field(default=10, ge=1, le=100, description="返回结果数量上限")

@tool(args_schema=DatabaseQueryInput)
def query_database(
    table: str,
    conditions: dict = None,
    fields: list[str] = None,
    sort_by: str = None,
    limit: int = 10
) -> str:
    """查询业务数据库，支持条件筛选、字段选择和排序。"""
    # 生产环境中这里会连接真实数据库
    conditions = conditions or {}
    fields = fields or ["*"]

    result = {
        "table": table,
        "conditions": conditions,
        "fields": fields,
        "sort_by": sort_by,
        "limit": limit,
        "rows": [
            {"id": 1, "name": "示例数据", "status": "active"},
            {"id": 2, "name": "更多数据", "status": "inactive"},
        ][:limit],
        "total": 2,
    }
    import json
    return json.dumps(result, ensure_ascii=False, indent=2)

# 枚举类型参数的示例
class FileType(str, Literal["pdf", "csv", "json", "excel"]):
    pass

class ExportInput(BaseModel):
    """数据导出工具的输入参数。"""
    data_source: str = Field(description="数据源标识，如 'sales_report_2026'")
    file_type: FileType = Field(description="导出文件格式")
    date_range: tuple[str, str] = Field(
        description="日期范围，格式为 (start_date, end_date)，如 ('2026-01-01', '2026-06-30')"
    )
    include_headers: bool = Field(default=True, description="是否包含表头")

@tool(args_schema=ExportInput)
def export_data(
    data_source: str,
    file_type: str,
    date_range: tuple,
    include_headers: bool = True
) -> str:
    """将指定数据源的数据导出为文件。"""
    return f"已将 {data_source} 的数据导出为 {file_type} 格式，日期范围 {date_range[0]} 至 {date_range[1]}，包含表头：{include_headers}"
​

Pydantic Schema 的优势：

• 类型安全：Field 的 ge/le 约束自动转为 JSON Schema 的 minimum/maximum
• 自描述：每个字段的 description 会成为模型理解参数含义的依据
• 枚举支持：Literal 类型会生成 enum 约束，模型只能选择预定义的值
• 可选字段：Optional + default 让模型知道哪些参数可以省略

步骤 4：中间件——拦截和修改 Agent 行为

中间件（Middleware）是 Harness 中最强大的扩展机制。它可以在模型调用前后插入自定义逻辑，类似于 Web 框架中的中间件。

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

@tool
def search(query: str) -> str:
    """搜索信息。"""
    return f"关于'{query}'的搜索结果：..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式。"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

# 中间件函数示例 1：日志记录
def log_middleware(state, config, next_):
    """记录每次模型调用的输入和输出。"""
    print(f"[LOG] 模型收到 {len(state['messages'])} 条消息")
    result = next_(state, config)
    print(f"[LOG] 模型返回，新增 {len(result['messages']) - len(state['messages'])} 条消息")
    return result

# 中间件函数示例 2：输入过滤
def content_filter_middleware(state, config, next_):
    """过滤用户输入中的敏感词。"""
    messages = state["messages"]
    filtered = []
    for msg in messages:
        if hasattr(msg, 'content') and isinstance(msg.content, str):
            # 简单的敏感词过滤演示
            filtered_content = msg.content.replace("密码", "***")
            if filtered_content != msg.content:
                msg = msg.model_copy(update={"content": filtered_content})
        filtered.append(msg)
    
    filtered_state = {**state, "messages": filtered}
    return next_(filtered_state, config)

# 中间件函数示例 3：动态注入系统提示
def inject_context_middleware(state, config, next_):
    """根据当前对话状态动态修改系统提示词。"""
    message_count = len(state["messages"])
    
    # 如果对话轮数较多，追加一条"简洁回复"的提醒
    extra_prompt = ""
    if message_count > 10:
        extra_prompt = "\n\n注意：对话已经比较长了，请尽量简洁回复。"
    
    if extra_prompt:
        # 在第一条系统消息后追加提醒
        messages = list(state["messages"])
        for i, msg in enumerate(messages):
            if isinstance(msg, SystemMessage):
                messages[i] = SystemMessage(content=msg.content + extra_prompt)
                break
        state = {**state, "messages": messages}
    
    return next_(state, config)

# 创建带中间件的 Agent
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[search, calculate],
    system_prompt="你是一个智能助手。",
    middleware=[
        content_filter_middleware,    # 先过滤输入
        inject_context_middleware,    # 再注入上下文
        log_middleware,               # 最后记录日志
    ],
)
​

中间件的执行顺序就是列表中的声明顺序——第一个中间件最先处理请求、最后处理响应。这类似于洋葱模型。

中间件的核心用途：

步骤 5：动态工具选择

有些场景下，Agent 不应该一开始就拥有所有工具。例如：一个客服 Agent 在用户未验证身份前不应该看到"查看订单"工具。LangChain 支持通过中间件实现动态工具选择：

def dynamic_tools_middleware(state, config, next_):
    """根据对话上下文动态调整可用工具集。"""
    messages = state["messages"]
    
    # 检查用户是否已验证身份（简单判断：是否提到了"验证码"相关内容）
    authenticated = any(
        "验证码" in (getattr(m, 'content', '') or '')
        for m in messages
    )
    
    # 根据认证状态决定工具集
    if authenticated:
        # 已认证：提供完整工具集
        state["tools"] = [search, calculate, check_order, create_order]
    else:
        # 未认证：只提供基础工具
        state["tools"] = [search, calculate]
    
    return next_(state, config)
​

这种模式在实际业务中非常实用。我的建议是：如果工具集的切换逻辑简单（基于用户等级、是否认证等），用中间件实现；如果工具集的切换逻辑复杂（需要查询数据库、调用外部服务），考虑在调用 Agent 前在业务层做好工具过滤，再把过滤后的工具列表传给 create_agent。

完整示例

下面是一个集成了 ToolRuntime、复杂 Schema 和中间件的完整客服 Agent：

from dataclasses import dataclass
from pydantic import BaseModel, Field
from typing import Optional
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain.messages import SystemMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.utils.uuid import uuid7
import json

# ========== 上下文定义 ==========
@dataclass
class CustomerContext:
    customer_id: str
    name: str
    tier: str

# ========== 工具定义 ==========

class RefundInput(BaseModel):
    """退款申请参数。"""
    order_id: str = Field(description="要退款的订单编号")
    reason: str = Field(description="退款原因")
    amount: Optional[float] = Field(default=None, description="部分退款金额，不填则全额退款")

@tool(args_schema=RefundInput)
def process_refund(order_id: str, reason: str, amount: float = None, runtime: ToolRuntime) -> str:
    """处理退款申请。需要管理员权限才能执行。"""
    ctx = runtime.context
    if ctx.tier != "admin":
        return f"权限不足：{ctx.name}（{ctx.tier}）无权执行退款操作，请联系管理员。"
    
    refund_amount = amount or "全额"
    return json.dumps({
        "status": "退款已提交",
        "order_id": order_id,
        "reason": reason,
        "refund_amount": refund_amount,
        "processing_time": "1-3 个工作日",
    }, ensure_ascii=False, indent=2)

@tool
def get_customer_info(runtime: ToolRuntime) -> str:
    """获取当前客户的账户信息和权益。"""
    ctx = runtime.context
    return json.dumps({
        "customer_id": ctx.customer_id,
        "name": ctx.name,
        "tier": ctx.tier,
        "benefits": {
            "free": "基础客服支持",
            "vip": "优先客服 + 9.5 折优惠",
            "admin": "全部权限 + 管理后台",
        }.get(ctx.tier, "无特殊权益"),
    }, ensure_ascii=False, indent=2)

# ========== 中间件 ==========

def audit_log_middleware(state, config, next_):
    """审计日志中间件：记录所有工具调用。"""
    result = next_(state, config)
    new_messages = result["messages"][len(state["messages"]):]
    for msg in new_messages:
        if hasattr(msg, 'tool_calls') and msg.tool_calls:
            for tc in msg.tool_calls:
                print(f"[AUDIT] 工具调用：{tc['name']}({json.dumps(tc['args'], ensure_ascii=False)})")
    return result

def tier_based_prompt_middleware(state, config, next_):
    """根据用户等级动态调整系统提示词。"""
    ctx = config.get("context")
    if not ctx:
        return next_(state, config)
    
    tier_suffix = {
        "free": "\n对于免费用户，优先推荐自助服务选项。",
        "vip": "\n对于 VIP 用户，提供更详细和个性化的服务。",
        "admin": "\n你拥有管理员权限，可以执行退款等敏感操作。",
    }.get(ctx.tier, "")
    
    messages = list(state["messages"])
    for i, msg in enumerate(messages):
        if isinstance(msg, SystemMessage) and tier_suffix:
            messages[i] = SystemMessage(content=msg.content + tier_suffix)
            break
    
    return next_({**state, "messages": messages}, config)

# ========== Agent 创建和运行 ==========

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[process_refund, get_customer_info],
    system_prompt="你是一个电商客服 Agent，帮助客户查询信息和处理售后。",
    checkpointer=InMemorySaver(),
    middleware=[tier_based_prompt_middleware, audit_log_middleware],
)

# 运行
thread_id = str(uuid7())
config = {
    "configurable": {"thread_id": thread_id},
    "context": CustomerContext(customer_id="C001", name="张三", tier="vip"),
}

result = agent.invoke(
    {"messages": [{"role": "user", "content": "帮我看看我的账户信息和权益"}]},
    config=config,
)
print(result["messages"][-1].content)
​

常见问题 FAQ

Q1：工具函数中用了 runtime 参数，但直接调用函数测试会报错怎么办？

A：这是正常的。runtime 参数由 Harness 在 Agent 运行时自动注入，你不能在 Agent 外部直接调用带 runtime 的工具函数。测试时可以构造一个 ToolRuntime 对象传入，或者写单元测试时不测带 runtime 的函数，而是测试业务逻辑。

Q2：中间件里能修改工具的返回结果吗？

A：可以。中间件的 next_ 函数返回的 result 包含完整的 state，你可以在返回前修改其中的消息内容。但要注意：如果你修改了 ToolMessage 的内容，模型看到的就是你修改后的内容，可能影响后续推理。

Q3：Pydantic 的 args_schema 和函数参数类型标注会冲突吗？

A：不会。当你指定了 args_schema，LangChain 会优先使用 Pydantic 模型来生成工具的 JSON Schema，函数的参数类型标注会被忽略。但函数签名中的参数名和类型标注应该与 Pydantic 模型保持一致，否则运行时可能出错。

Q4：Store（长期记忆）怎么用？和 State 有什么区别？

A：State 是会话级的，随 checkpointer 持久化，生命周期和 thread_id 绑定；Store 是跨会话的持久化存储，通过 runtime.store 访问。典型用法是：State 存"当前对话中收集到的用户偏好"，Store 存"用户历史偏好数据"。Store 需要配合 BaseStore 实现类（如 InMemoryStore）使用。

最佳实践与避坑

实践 1：中间件顺序很重要

请求处理是按照列表顺序的，但响应处理是反过来的。把日志中间件放在最后，这样它的请求日志在所有处理之前、响应日志在所有处理之后，能看到完整的输入输出。

实践 2：工具描述中写清"何时用"而非仅仅"做什么"

"搜索数据库"是做什么；"当用户询问订单、账户、交易记录时使用此工具"是何时用。后者对模型的决策帮助大得多。

实践 3：复杂 Schema 用 Pydantic，简单 Schema 用类型标注

不要为了炫技给只有两个字符串参数的工具也写一个 Pydantic 模型。简单的类型标注 + 好的 docstring 就够了。Pydantic 的价值在于有嵌套结构、枚举约束、字段校验的复杂场景。

坑点 1：config 和 runtime 是保留参数名

不要在你的工具函数中使用 config 或 runtime 作为业务参数名。这两个名称被 LangChain 保留用于内部机制，用作业务参数会导致运行时错误。如果确实需要传递配置信息，用 context 通过 ToolRuntime 注入。

坑点 2：中间件中修改 state 要小心不可变对象

LangGraph 的 state 通常是不可变的（基于 Pydantic 或 TypedDict）。不要直接修改 state["messages"] 列表，而是创建新列表再构造新的 state 字典传给 next_。

坑点 3：Context 对象必须在每次 invoke 时传入

Context 不是持久化的，每次调用 agent.invoke 时需要重新传入。如果你忘了传 context，runtime.context 会是 None，工具中访问 context 属性会报错。建议在工具中加一个 None 检查作为防御性编程。

本节小结

本节我们从 ToolRuntime 出发，让工具从"无状态函数"进化为"能感知上下文的能力单元"。通过 State 访问会话历史、Context 获取用户身份、Pydantic 定义复杂输入 Schema、中间件拦截和修改行为，你的 Agent 已经具备了应对真实业务场景的能力。

下一节我们将探讨多 Agent 协作——当单个 Agent 的能力不够时，如何让多个 Agent 分工合作完成复杂任务。

延伸阅读

• 官方文档：LangChain Tools 页面（docs.langchain.com/oss/python/langchain/tools）
• 官方文档：LangChain Middleware 页面
• 相关章节：本教程 4.1 节介绍了 create_agent 的基础用法，本节在其基础上展开了工具和中间件的进阶内容
• 相关章节：本教程 2.5 节介绍了工具的基本概念

关键词：LangChain框架精通, ToolRuntime, 中间件, Middleware, 工具调用, Pydantic, Agent, 教程, 实战, 最佳实践
难度：进阶
预计阅读：18 分钟