前言：
前面几章我们做的所有事情，本质上都是让大模型「说」：
写文案、答问题、做总结、抽信息，都是输出文字。
但真实业务里，光会说没用，你得会「做」：
查订单状态、查商品库存、算折扣价格、查天气、操作数据库、调用第三方接口…
这些事大模型本身做不了，它没有手，连不上你的业务系统。

怎么让大模型从「只会说」变成「会干活」？
答案就是 Tool 工具调用 + Agent 智能体。
这篇是 LangChain 系列第六篇，我带你从零搞懂 Agent 的核心原理，从定义第一个工具，到搭一个完整的电商客服 Agent，再到 MCP 协议、中间件、调试评估，全流程打通。
全是可直接落地的代码，跟着敲完，你就能让大模型真正帮你干活，而不是只会聊天。

一、先搞懂本质：Tool 和 Agent 到底是啥？

很多人学 Agent，上来就抄代码，结果抄完了也不知道为什么要这么写。
先把核心概念讲透，后面学起来就快了。

1. Tool：给大模型装「手」

大模型本身只会处理文字，它不能查数据库、不能调接口、不能算复杂的数。
Tool 是什么？
就是把外部能力封装成大模型可以调用的函数，相当于给大模型装了手。

普通函数 vs Tool 的区别：

类型 有什么 给谁用
普通 Python 函数 函数名、参数、函数体 程序员调用
LangChain Tool 函数名、参数、工具描述、函数体 大模型调用

多出来的「工具描述」是核心，大模型靠这个判断：

  • 这个工具是干嘛的？
  • 什么时候该用它？
  • 参数要传什么？

一句话记牢：
Tool 就是大模型能调用的外部能力，描述写得好不好，直接决定大模型会不会用、用得对不对。

2. Agent：给大模型装「大脑调度器」

有了 Tool，大模型就能用了吗？还不行。
谁来决定「什么时候调用哪个工具」？谁来处理工具返回的结果？谁来判断任务做完了没有？
这就是 Agent 干的事。

Agent 是什么？
具备自主思考、规划、调用工具、完成完整目标的人工智能程序。
和普通 Chat 的区别：

  • 普通 Chat：你问一句，它答一句，一步到位
  • Agent：你给一个目标，它自己拆解任务，自己调用工具，自己一步步完成，不用你指挥

Agent 的三大核心能力：

  1. 感知：获取外部信息（用户需求、工具返回结果、系统状态）
  2. 思考规划：拆解任务，判断下一步该做什么，要不要调用工具
  3. 行动执行：调用工具，拿到结果，继续思考，直到完成任务

3. Agent、Tool、Harness 的关系

LangChain 官方有个公式：

Agent = Model + Harness

很多人看不懂，我用大白话给你翻译一下：

角色 作用 类比
Model（大模型） 负责思考、决策、判断下一步做什么 谋士，只会动脑，出主意，但是手不能动，跑不出去
Tool（工具） 负责执行具体的操作 办事的方法、工具
Harness（调度框架） 负责整套流程调度：传话、调用工具、循环、把结果拿回来给模型 管家，跑腿的，负责把谋士的指令落地，循环往复
Agent 整个组合，能自主完成任务的智能体 整个团队，谋士+管家+工具

create_agent 就是 LangChain 给你提供的现成 Harness，你不用自己写循环调度的逻辑，直接用就行。

一句话总结：
大模型是脑子，Tool 是手，Harness 是连接脑子和手的神经系统，Agent 就是整个会思考会干活的人。

二、实战第一步：定义你的第一个 Tool

我们先从最简单的 Tool 开始，学会怎么定义工具，怎么写工具描述，怎么手动测试工具。

1. 用 @tool 装饰器定义工具

LangChain 里定义工具特别简单，用 @tool 装饰器就行：

from langchain_core.tools import tool

@tool
def get_order_status(order_id: str) -> str:
    """根据订单号查询订单状态，包括是否付款、是否发货、快递单号和签收状态。"""
    # 模拟数据库
    fake_orders = {
        "A1001": "已付款，等待发货",
        "A1002": "已发货，快递单号 SF123456",
        "A1003": "已签收",
    }
    return fake_orders.get(order_id, "没有查询到该订单")

就这么简单，一个普通函数加个 @tool 装饰器，就变成了大模型能调用的 Tool。

我们可以打印一下 Tool 的信息，看看它多了什么：

print("工具名称：", get_order_status.name)
print("工具描述：", get_order_status.description)
print("参数结构：", get_order_status.args)

你会看到，Tool 自动从函数名、docstring、类型注解里提取了名称、描述、参数结构，这些信息都会传给大模型，帮它判断怎么调用。

2. 手动调用 Tool，先测通再用

把工具交给 Agent 之前，一定要先手动调用测试，确保工具本身没问题：

result = get_order_status.invoke({
    "order_id": "A1002"
})
print(result)
# 输出：已发货，快递单号 SF123456

避坑提醒：
很多人定义完工具直接丢给 Agent，结果 Agent 调用报错，不知道是工具的问题还是 Agent 的问题。
先手动测通工具，再交给 Agent，出问题了好排查。

3. 工具描述怎么写？这是关键

工具描述写得好不好，直接决定大模型会不会用这个工具。
很多人写的描述跟没写一样：

# 反面教材：描述太模糊，大模型不知道什么时候用
@tool
def query(order_id: str) -> str:
    """查询信息。"""
    ...

好的工具描述应该包含这几点：

  1. 工具能做什么
  2. 什么时候应该用这个工具
  3. 每个参数是什么意思
  4. 返回什么结果
# 正面教材：描述清晰，大模型一看就知道什么时候用
@tool
def get_order_status(order_id: str) -> str:
    """
    根据订单号查询订单状态，包括付款状态、发货状态、快递单号和签收状态。
    当用户询问订单相关问题时使用此工具。
    参数 order_id 是订单编号，格式为字母A加4位数字，例如A1001。
    返回订单的详细状态信息。
    """
    ...

大佬经验：
工具名称用英文小写加下划线，比如 get_order_status、calculate_discount，兼容性好，大模型也更容易理解。
描述越具体，大模型用得越准，不要怕写得多。

4. 多参数 Tool + Pydantic 参数校验

业务工具通常有多个参数，而且需要校验参数格式，这时候用 Pydantic 定义参数结构最靠谱：

from pydantic import BaseModel, Field

class InventoryInput(BaseModel):
    product_id: str = Field(
        description="商品编号，格式为字母P加4位数字，例如P1001",
        pattern="^P[0-9]{4}$"  # 正则校验格式
    )
    warehouse: str = Field(
        description="仓库名称，可选值：上海仓、北京仓、广州仓"
    )

@tool(args_schema=InventoryInput)
def get_product_inventory(product_id: str, warehouse: str) -> str:
    """查询指定商品在指定仓库中的库存数量。"""
    fake_inventory = {
        ("P1001", "上海仓"): 35,
        ("P1001", "北京仓"): 12,
        ("P2001", "上海仓"): 0,
    }
    count = fake_inventory.get((product_id, warehouse))
    if count is None:
        return "没有查询到该商品的库存信息"
    return f"{product_id} 在 {warehouse} 当前库存为 {count} 件"

Pydantic 的好处：

  1. 每个字段都有清晰的描述，大模型更清楚传什么
  2. 可以加正则、范围等校验，参数不对直接报错，不会传到业务逻辑里
  3. 参数多的时候结构更清晰，好维护

划重点：
不要完全相信大模型一定会传正确的参数，工具内部一定要做参数校验。
尤其是生产环境，什么奇葩参数都可能出现。

三、实战第二步：搭你的第一个 Agent

有了 Tool，我们就可以创建 Agent 了，让大模型自己决定什么时候调用工具。

1. 最简单的单工具 Agent

我们用刚才的订单查询工具，搭一个最简单的 Agent：

from langchain.agents import create_agent
from utils.model_factory import get_deepSeek_model

# 初始化模型
model = get_deepSeek_model(temperature=0)

# 创建 Agent
agent = create_agent(
    model=model,
    tools=[get_order_status],
    system_prompt="你是一名电商客服助手。回答要礼貌、简洁，不要编造工具返回值中不存在的信息。"
)

# 调用 Agent
result = agent.invoke({
    "messages": [
        {
            "role": "user",
            "content": "帮我查一下订单 A1002 发货了吗？",
        }
    ]
})

# 打印最终回答
print("客服：", result["messages"][-1].content)

运行后你会看到，Agent 自动调用了订单查询工具，拿到结果后整理成自然语言回答用户。
就这么简单，几行代码，你就有了一个会查订单的智能客服。

2. 查看 Agent 的完整执行过程

很多人好奇 Agent 内部是怎么工作的，我们把所有消息都打印出来看看：

for message in result["messages"]:
    print(type(message).__name__)
    print(message.content)
    print("-" * 60)

你会看到完整的消息流转过程：

HumanMessage：用户问题
AIMessage：模型请求调用 get_order_status 工具，参数 order_id=A1002
ToolMessage：工具返回结果「已发货，快递单号 SF123456」
AIMessage：模型最终回答「订单 A1002 已发货，快递单号是 SF123456。」

这就是 Agent 的核心循环：

模型思考 → 调用工具 → 拿到结果 → 模型继续思考 → 生成最终答案

记住这个流程，以后 Agent 出问题了，你就打印消息列表，看看到底是哪一步错了：
是模型没调用工具？还是工具调用参数错了？还是工具返回结果错了？还是模型没看懂工具结果？
一目了然。

3. 多工具 Agent：自动选择工具

一个 Agent 可以有多个工具，它会根据问题自动选择用哪个：

@tool
def calculate_discount_price(original_price: float, discount_rate: float) -> str:
    """根据商品原价和折扣率计算折后价格。"""
    if original_price <= 0:
        return "原价必须大于 0"
    if discount_rate <= 0 or discount_rate > 1:
        return "折扣率必须在 0 到 1 之间"
    final_price = original_price * discount_rate
    return f"折后价格为 {final_price:.2f} 元"

# 创建多工具 Agent
agent = create_agent(
    model=model,
    tools=[get_order_status, get_product_inventory, calculate_discount_price],
    system_prompt="你是一个电商助手，可以查询订单状态、商品库存，以及计算折扣价格。"
)

我们测试几个不同的问题：

questions = [
    "订单 A1001 现在是什么状态？",
    "商品 P2001 还有库存吗？",
    "299 元的商品打八折后是多少钱？",
]

for question in questions:
    result = agent.invoke({
        "messages": [{"role": "user", "content": question}]
    })
    print(f"问题：{question}")
    print(f"回答：{result['messages'][-1].content}")
    print("~" * 40)

你会发现，Agent 会自动判断该用哪个工具，完全不用你写 if-else 判断。
这就是 Agent 比固定 Chain 灵活的地方：任务步骤不固定的时候，用 Agent 更合适。

四、企业级实战：电商客服 Agent

我们做一个完整的电商客服 Agent 案例，包含订单查询、库存查询、折扣计算、退款规则查询四个工具，模拟真实业务场景。

1. 项目结构

customer_service_agent/
├── tools.py          # 业务工具定义
├── customer_agent.py # Agent 封装
└── main.py           # 终端入口

2. 编写业务工具 tools.py

from pydantic import BaseModel, Field
from langchain.tools import tool

# 模拟数据库
ORDERS = {
    "A1001": {
        "status": "已付款，等待发货",
        "shipping_no": None,
        "product_id": "P1001",
    },
    "A1002": {
        "status": "已发货",
        "shipping_no": "SF123456",
        "product_id": "P2001",
    },
    "A1003": {
        "status": "已签收",
        "shipping_no": "YT998877",
        "product_id": "P3001",
    },
}

PRODUCTS = {
    "P1001": {"name": "无线静音鼠标", "stock": 35, "price": 129},
    "P2001": {"name": "蓝牙机械键盘", "stock": 0, "price": 299},
    "P3001": {"name": "Type-C 扩展坞", "stock": 8, "price": 199},
}

class DiscountInput(BaseModel):
    original_price: float = Field(description="商品原价，单位为元")
    discount_rate: float = Field(description="折扣率，例如 0.8 表示八折")

@tool
def get_order_status(order_id: str) -> str:
    """根据订单号查询订单状态、物流单号和商品编号。"""
    order = ORDERS.get(order_id)
    if not order:
        return "没有查询到该订单"
    shipping_no = order["shipping_no"] or "暂无快递单号"
    return (
        f"订单 {order_id} 状态：{order['status']}；"
        f"快递单号：{shipping_no}；"
        f"商品编号：{order['product_id']}"
    )

@tool
def get_product_inventory(product_id: str) -> str:
    """根据商品编号查询商品名称、库存数量和原价。"""
    product = PRODUCTS.get(product_id)
    if not product:
        return "没有查询到该商品"
    return (
        f"商品 {product_id}：{product['name']}；"
        f"库存：{product['stock']} 件；"
        f"原价：{product['price']} 元"
    )

@tool(args_schema=DiscountInput)
def calculate_discount_price(original_price: float, discount_rate: float) -> str:
    """根据商品原价和折扣率计算折后价格。"""
    if original_price <= 0:
        return "原价必须大于 0"
    if discount_rate <= 0 or discount_rate > 1:
        return "折扣率必须在 0 到 1 之间"
    final_price = original_price * discount_rate
    return f"折后价格为 {final_price:.2f} 元"

@tool
def get_refund_policy(order_status: str) -> str:
    """根据订单状态查询退款规则。order_status 可以是未发货、已发货、已签收。"""
    if "未发货" in order_status or "等待发货" in order_status:
        return "订单未发货时，用户可以直接申请退款。"
    if "已发货" in order_status:
        return "订单已发货时，需要等待商品送达后申请退货退款。"
    if "已签收" in order_status:
        return "订单签收后，如商品存在质量问题，可以在 7 天内申请售后。"
    return "没有匹配到明确的退款规则"

注意：这里用内存字典模拟数据库，真实项目里，工具内部可以查 MySQL、调 Redis、调 HTTP 接口、调用 RAG 检索，什么都可以。
Tool 本质就是 Python 函数，你能写的业务逻辑都能放进去。

3. 封装 Agent customer_agent.py

from langchain.agents import create_agent
from utils.model_factory import get_deepSeek_model
from tools import (
    get_order_status,
    get_product_inventory,
    calculate_discount_price,
    get_refund_policy,
)

SYSTEM_PROMPT = """
你是一名电商客服助手。
工作要求：
1. 根据用户问题选择合适工具。
2. 不要编造订单、库存、价格和退款规则。
3. 如果工具没有查到数据，要如实告诉用户。
4. 回答要礼貌、简洁、清楚。
5. 涉及订单状态、库存、价格时，优先调用工具确认。
"""

def create_customer_agent():
    return create_agent(
        model=get_deepSeek_model(temperature=0),
        tools=[
            get_order_status,
            get_product_inventory,
            calculate_discount_price,
            get_refund_policy,
        ],
        system_prompt=SYSTEM_PROMPT,
    )

4. 终端入口 main.py

from customer_agent import create_customer_agent

def main():
    agent = create_customer_agent()
    print("电商客服 Agent 已启动，输入 exit 退出。")
    while True:
        question = input("\n用户：").strip()
        if question.lower() == "exit":
            print("程序已退出。")
            break
        if not question:
            print("问题不能为空。")
            continue
        
        result = agent.invoke({
            "messages": [{"role": "user", "content": question}]
        })
        print(f"客服：{result['messages'][-1].content}")

if __name__ == "__main__":
    main()

5. 测试组合问题

我们测试一个需要调用多个工具的问题：

用户：订单 A1002 我还能退款吗？

你会发现 Agent 会自动分两步：

  1. 先调用 get_order_status 查订单状态，发现是已发货
  2. 再调用 get_refund_policy 查已发货的退款规则
  3. 最后整理成自然语言回答用户

完全不用你写步骤，它自己会规划，自己会调用工具，这就是 Agent 的威力。

五、进阶：MCP 协议，智能体世界的「USB 接口」

现在你定义的工具都是本地的，那如果我想用别人写的工具？或者用其他语言写的工具？或者远程服务上的工具？
总不能每个都自己重新写一遍吧？
这就是 MCP（模型上下文协议） 解决的问题。

1. MCP 是什么？

MCP 全称 Model Context Protocol，是 2024 年底 Anthropic 推出的协议，现在已经成了行业标准。
你可以把它理解成智能体世界的 USB 协议：

  • 以前每个工具都有自己的接口，用起来很麻烦
  • 现在只要符合 MCP 标准，不管是 Python 写的、Node 写的、本地的、远程的，都能直接插进来用
  • 不用关心底层怎么实现的，统一接口调用

MCP 是客户端-服务端架构：

  • MCP 服务端：定义并暴露工具，比如天气服务、数学计算服务、数据库服务
  • MCP 客户端：连接服务端，发现工具，把工具交给 Agent 调用

一句话总结：
MCP 让工具可以标准化复用，不用每个项目都重复造轮子。

2. 实战：搭一个 MCP 服务端

我们用 FastMCP 快速搭两个 MCP 服务端：一个数学工具（本地 stdio 传输），一个天气工具（HTTP 传输）。

数学工具 MCP 服务端 math_mcp_server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个数的和"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """计算两个数的乘积"""
    return a * b

if __name__ == "__main__":
    # stdio 传输：通过标准输入输出通信，适合本地工具
    mcp.run(transport="stdio")
天气工具 MCP 服务端 weather_mcp_server.py
from mcp.server.fastmcp import FastMCP
import httpx
from typing import Optional, Dict, Any

mcp = FastMCP("Weather")

# 免费天气 API，不用 Key
OPEN_METEO_WEATHER_URL = "https://api.open-meteo.com/v1/forecast"
OPEN_METEO_GEOCODE_URL = "https://geocoding-api.open-meteo.com/v1/search"

@mcp.tool()
def geocode_city(name: str, country: Optional[str] = None, language: str = "zh") -> Dict[str, Any]:
    """将城市名解析为经纬度"""
    params = {"name": name, "count": 1, "language": language, "format": "json"}
    if country:
        params["country"] = country
    with httpx.Client(timeout=10) as client:
        r = client.get(OPEN_METEO_GEOCODE_URL, params=params)
        r.raise_for_status()
        data = r.json()
    results = data.get("results") or []
    if not results:
        return {"error": f"未找到城市：{name}"}
    top = results[0]
    return {
        "name": top.get("name"),
        "lat": top.get("latitude"),
        "lon": top.get("longitude"),
        "country": top.get("country"),
    }

@mcp.tool()
def get_current_weather(lat: float, lon: float) -> Dict[str, Any]:
    """根据经纬度查询当前天气"""
    params = {"latitude": lat, "longitude": lon, "current_weather": True}
    with httpx.Client(timeout=10) as client:
        r = client.get(OPEN_METEO_WEATHER_URL, params=params)
        r.raise_for_status()
        payload = r.json()
    cw = payload.get("current_weather") or {}
    return {
        "latitude": lat,
        "longitude": lon,
        "temperature": cw.get("temperature"),
        "windspeed": cw.get("windspeed"),
        "weathercode": cw.get("weathercode"),
        "time": cw.get("time"),
    }

@mcp.tool()
def get_current_weather_by_city(name: str, country: Optional[str] = None, language: str = "zh") -> Dict[str, Any]:
    """根据城市名查询当前天气"""
    g = geocode_city(name=name, country=country, language=language)
    if "error" in g:
        return g
    w = get_current_weather(lat=g["lat"], lon=g["lon"])
    return {**g, **w}

if __name__ == "__main__":
    print("Weather MCP Server 启动中：http://localhost:8000/mcp")
    # HTTP 传输：通过网络提供服务，适合远程工具
    mcp.run(transport="streamable-http")

启动天气服务端：

python weather_mcp_server.py

3. 实战：MCP 客户端集成 Agent

现在我们用 MCP 客户端同时连接这两个服务端，把工具都注册到 Agent 里：

import asyncio
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from utils.model_factory import get_deepSeek_model

model = get_deepSeek_model()

# 同时连接多个 MCP 服务端
client = MultiServerMCPClient({
    "Math": {
        "transport": "stdio",
        "command": "python",
        "args": ["./math_mcp_server.py"],
    },
    "Weather": {
        "transport": "streamable_http",
        "url": "http://localhost:8000/mcp",
    },
})

# 从服务端获取所有工具
tools = asyncio.run(client.get_tools())
print("已加载工具：", [tool.name for tool in tools])

# 创建 Agent，直接用 MCP 工具
agent = create_agent(
    model=model,
    tools=tools,
    system_prompt="你是一个助理。涉及数学计算用 Math 工具，涉及天气用 Weather 工具。"
)

# 测试数学问题
result1 = asyncio.run(agent.ainvoke({
    "messages": [HumanMessage(content="请帮我计算 (3 + 5) × 12 的结果")]
}))
print("数学问题回答：", result1["messages"][-1].content)

# 测试天气问题
result2 = asyncio.run(agent.ainvoke({
    "messages": [HumanMessage(content="今天郑州天气怎么样？")]
}))
print("天气问题回答：", result2["messages"][-1].content)

运行后你会发现，Agent 可以直接调用两个不同 MCP 服务端的工具，完全不用关心工具是本地的还是远程的，是 stdio 还是 HTTP。
这就是 MCP 的魅力，标准化的工具接口，拿来就能用。

同步代码里怎么用异步的 MCP？
写个简单的同步封装就行：

def arun(coro):
    return asyncio.run(coro)

主逻辑还是同步写法，不影响使用。

六、生产进阶：中间件，给 Agent 加「安全护栏」

Agent 能干活了，但生产环境光会干活还不够，还要安全、稳定、可控。
LangChain 1.0 提供了中间件机制，可以在不修改核心业务逻辑的情况下，给 Agent 加各种能力：日志、重试、限流、人工审批、对话摘要等等。

1. 预置中间件：开箱即用的常用能力

LangChain 内置了很多常用的中间件，直接用就行：

中间件 作用 适用场景
Summarization 自动压缩对话历史，控制上下文长度 长对话、多轮对话，避免超 Token
Human-in-the-loop 高风险工具调用需要人工审批 写操作、支付、删除等危险操作
Tool retry 工具调用失败自动重试 网络不稳定、第三方接口偶尔失败
Model fallback 主模型挂了自动切备用模型 高可用场景，保证服务不中断
Model call limit 限制模型调用次数，防止成本失控 防止 Agent 死循环，控制成本
PII detection 自动检测并脱敏个人信息 合规场景，保护用户隐私

我们演示两个最常用的：

案例1：对话摘要中间件，自动压缩历史

对话多了容易超 Token，用 Summarization 中间件自动压缩早期对话：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

model = get_deepSeek_model()

summarization_middleware = SummarizationMiddleware(
    model=model,
    trigger=("tokens", 1000),  # 超过1000token触发摘要
    keep=("messages", 2),       # 保留最近2条原始消息
    summary_prompt="用20个字以内概括对话要点。",
)

agent = create_agent(
    model=model,
    middleware=[summarization_middleware],
    tools=[],
    system_prompt="回答要简洁，不超过30字。"
)

对话多了之后，早期的消息会被自动压缩成摘要，既保留了上下文，又不会超 Token。

案例2：人工审批中间件，高风险操作要人确认

涉及数据库写操作、删除操作、支付操作这种高风险的，不能让 Agent 直接执行，必须要人审批。
用 Human-in-the-loop 中间件：

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

@tool
def dangerous_write(sql: str) -> str:
    """对数据库执行写操作（插入/更新/删除）。"""
    return f"[模拟执行] {sql}"

# HITL 中间件：拦截 dangerous_write 工具，需要人工审批
hitl = HumanInTheLoopMiddleware(
    interrupt_on={
        "dangerous_write": {
            "allowed_decisions": ["approve", "reject", "edit"]
        }
    }
)

agent = create_agent(
    model=get_deepSeek_model(),
    middleware=[hitl],
    tools=[dangerous_write],
    system_prompt="凡是涉及数据库写操作，必须调用 dangerous_write 工具。",
    checkpointer=InMemorySaver(),  # HITL 必须有状态保存
)

# 调用配置，同一个 thread_id 可以恢复状态
CFG = {"configurable": {"thread_id": "hitl-demo"}}

# 用户提问，触发工具调用，会中断等待人工审批
result = agent.invoke(
    {"messages": [{"role": "user", "content": "SQL: DELETE FROM orders WHERE id=1;"}]},
    config=CFG
)

# 处理中断：人工输入决策
if result.get("__interrupt__"):
    decision = input("请输入决策 (approve/reject/edit)：").strip().lower()
    if decision == "approve":
        # 批准，继续执行
        result = agent.invoke(
            Command(resume={"decisions": [{"type": "approve"}]}),
            config=CFG
        )
    elif decision == "reject":
        # 拒绝，不执行工具
        result = agent.invoke(
            Command(resume={"decisions": [{"type": "reject", "override": {"content": "[操作已被人工拒绝]"}}]}),
            config=CFG
        )

print("最终回答：", result["messages"][-1].content)

划重点：
高风险操作一定要加人工审批，绝对不能让 Agent 直接执行。
不然哪天 Agent 抽风把你数据库删了，哭都来不及。

2. 自定义中间件：自己写钩子

内置中间件不够用？你可以自己写中间件，在 Agent 执行的各个节点插入逻辑。
中间件有三类钩子：

钩子类型 作用 常用场景
节点式钩子 在固定执行点触发，before/after 日志记录、输入校验、输出过滤
包裹式钩子 包裹整个调用过程，可以控制执行 重试、耗时统计、限流、缓存
便捷装饰器 常用场景的简化封装 动态修改 Prompt

举个最简单的例子，给模型调用加耗时统计和日志：

import time
from typing import Callable
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse

@wrap_model_call
def timing_and_log(request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]):
    start = time.time()
    print(f"[模型调用开始] 消息数：{len(request.messages)}")
    try:
        response = handler(request)
        cost = (time.time() - start) * 1000
        print(f"[模型调用结束] 耗时：{cost:.0f}ms")
        return response
    except Exception as e:
        print(f"[模型调用失败] 错误：{e}")
        raise

agent = create_agent(
    model=get_deepSeek_model(),
    middleware=[timing_and_log],
    tools=[],
    system_prompt="回答要简洁。"
)

就这么简单，几行代码，就给所有模型调用加上了日志和耗时统计，完全不用改业务逻辑。
你还可以加重试、加限流、加缓存，想加什么加什么。

七、调试与评估：Agent 出问题了怎么查？

Agent 比 RAG 更复杂，出问题的地方更多，所以调试和评估更重要。

1. 调试第一步：打印消息列表

Agent 出问题了，第一步永远是打印所有消息，看看到底哪一步错了：

for message in result["messages"]:
    print(type(message).__name__)
    print(message.content)
    if hasattr(message, "tool_calls") and message.tool_calls:
        print("工具调用：", message.tool_calls)
    print("-" * 60)

常见问题和排查方向：

问题现象 排查方向
Agent 不调用工具，直接回答 1. 工具描述不清楚 2. system prompt 没要求调用工具 3. 用户问题不需要工具
Agent 调用了错误的工具 1. 两个工具描述太像，区分度不够 2. 工具名称太模糊
工具调用参数错误 1. 参数描述不清楚 2. 没有加 Pydantic 校验 3. 模型理解错了参数含义
工具返回结果正确，但回答错了 1. 模型没看懂工具返回 2. system prompt 约束不够
Agent 死循环调用工具 1. 工具返回结果有问题 2. 模型理解错了 3. 加调用次数限制

2. 用 LangSmith 看完整 Trace

和 RAG 一样，Agent 也可以用 LangSmith 追踪完整调用链路：

  • 每一次模型调用
  • 每一次工具调用
  • 工具参数和返回结果
  • 每一步的耗时和 Token
  • 完整的消息流转过程

Agent 项目重点看：

  1. 工具调用是否正确：有没有调用该调用的工具
  2. 参数是否正确：传的参数对不对
  3. 工具返回结果是否正确：工具本身有没有问题
  4. 模型是否正确使用了工具结果：有没有理解返回的信息

大佬经验：
做 Agent 一定要开 LangSmith，不然出了问题你根本不知道中间发生了什么。
Agent 的执行过程是黑盒，LangSmith 就是给你开了个窗户，让你能看到里面的每一步。

3. Agent 效果评估

Agent 的评估比 RAG 复杂，不仅要看回答对不对，还要看工具调用对不对。
最简单的评估方法：测试集 + 关键词匹配 + 工具调用检查

test_cases = [
    {
        "question": "帮我查一下订单 A1002 发货了吗？",
        "expected_keyword": "SF123456",
        "expected_tool": "get_order_status",
    },
    {
        "question": "商品 P2001 还有库存吗？",
        "expected_keyword": "0",
        "expected_tool": "get_product_inventory",
    },
    {
        "question": "299 元打八折后是多少钱？",
        "expected_keyword": "239.20",
        "expected_tool": "calculate_discount_price",
    },
]

def extract_tool_names(messages):
    """从消息列表提取调用过的工具名"""
    tool_names = []
    for message in messages:
        tool_calls = getattr(message, "tool_calls", None)
        if not tool_calls:
            continue
        for tool_call in tool_calls:
            tool_names.append(tool_call["name"])
    return tool_names

def evaluate_agent(agent, test_cases):
    passed_count = 0
    for index, case in enumerate(test_cases, start=1):
        result = agent.invoke({
            "messages": [{"role": "user", "content": case["question"]}]
        })
        answer = result["messages"][-1].content
        tool_names = extract_tool_names(result["messages"])
        
        answer_passed = case["expected_keyword"] in answer
        tool_passed = case["expected_tool"] in tool_names
        passed = answer_passed and tool_passed
        
        if passed:
            passed_count += 1
        
        print("-" * 60)
        print(f"用例 {index}")
        print(f"问题：{case['question']}")
        print(f"答案：{answer}")
        print(f"调用工具：{tool_names}")
        print(f"预期工具：{case['expected_tool']}")
        print(f"是否通过：{passed}")
    
    print("\n===== Agent 评估结果 =====")
    print(f"通过数量：{passed_count}/{len(test_cases)}")
    print(f"通过率：{passed_count / len(test_cases):.2%}")

注意：
Agent 评估不能只看最终回答对不对，还要看工具调用对不对。
有时候回答对了，但工具没调用，说明模型是猜的，不是真的查了数据，这种也算失败。

八、Tool 设计原则与避坑指南

最后分享几个我踩过的坑，以及 Tool 设计的最佳实践，帮你少走弯路。

1. Tool 设计四大原则

原则一：工具职责要单一

推荐：

get_order_status
get_product_inventory
calculate_discount_price

不推荐：

handle_all_customer_questions

工具越小、越单一，模型越容易判断什么时候用，也越好测试、越好维护。
一个工具干一件事，永远没错。

原则二：返回结果要清楚

工具返回给模型看的内容要明确、结构化：

订单 A1002 状态：已发货；快递单号：SF123456；商品编号：P2001

不要返回含义不明的内容：

ok
1
true

模型看不懂，就会瞎编。

原则三：工具内部要做参数校验

不要完全相信模型会传正确的参数，各种奇葩参数都可能出现。
参数格式、范围、必填项，都要校验，错了就返回明确的错误信息，让模型自己修正。

原则四：高风险操作要加护栏

查询类工具风险低，随便用。
写入、删除、支付、发送消息这类高风险操作，一定要加：

  • 人工审批
  • 权限校验
  • 操作日志
  • 二次确认
    绝对不能让 Agent 直接执行。

2. Agent 常见问题排查

问题 解决方法
Agent 不调用工具 优化工具描述，system prompt 明确要求调用工具，给例子
Agent 调用错工具 优化工具名称和描述，增加区分度，减少工具数量
Agent 反复调用同一个工具 检查工具返回结果是否清楚，有没有解决问题，加调用次数限制
Agent 编造工具结果 加强 system prompt 约束，降低 temperature，工具返回结果要明确
Agent 理解错工具返回 优化工具返回格式，更结构化、更清楚，加说明

3. Agent 适合什么场景？

不是所有场景都适合用 Agent，不要为了用 Agent 而用 Agent。

✅ 适合用 Agent 的场景：

  • 任务步骤不固定，需要根据情况选择工具
  • 需要多步推理、多工具组合
  • 用户需求多样，没法用固定流程覆盖
  • 客服助手、数据分析助手、运维助手、办公助手

❌ 不适合用 Agent 的场景：

  • 流程非常固定，输入输出明确
  • 对稳定性要求极高，不能出错
  • 简单的单步任务
    比如「输入评论 → 情感分类 → 输出 JSON」这种任务，用普通 Chain 就行，没必要用 Agent，又慢又不稳定。

九、给新手的 5 条实战心法

1. 先把 Tool 写好，再谈 Agent

Tool 是 Agent 的基础，工具写得好不好，直接决定 Agent 好不好用。
花时间把工具描述写清楚，把参数校验做好，把返回结果写明白，比你折腾 Agent 本身有用得多。

2. 工具描述越具体越好

不要怕描述写得多，越具体模型越容易理解，用得越准。
把工具能做什么、什么时候用、参数是什么、返回什么，都写清楚。
你写得越模糊，模型越容易用错。

3. 调试从消息列表入手

Agent 出问题了，不要瞎猜，先把所有消息打印出来。
看看到底是没调用工具，还是调用错了，还是参数错了，还是返回结果错了，还是模型理解错了。
一步一步排查，比你瞎改 prompt 靠谱得多。

4. 不要上来就搞复杂 Agent

先从单工具 Agent 开始，跑通了再加第二个工具，再加第三个。
一步一步来，出了问题也好排查。
上来就搞十几个工具的复杂 Agent，出了问题你根本找不到原因。

5. 安全永远是第一位的

高风险操作一定要加人工审批，绝对不能让 Agent 直接执行。
生产环境一定要加调用次数限制，防止死循环烧钱。
用户输入一定要做校验，防止 Prompt 注入。
安全这根弦，永远不能松。

最后

到这里，Agent 的核心内容就全部讲完了：
从 Tool 定义到 Agent 创建，从 MCP 协议到中间件，从调试到评估，从 demo 到生产级应用。
现在你不仅能让大模型说，还能让它做，真正帮你干活。

整个 LangChain 核心系列到这里就差不多了，从基础的模型调用、Prompt、结构化输出，到 LCEL、文档处理、RAG，再到 Agent、工具调用、调试评估，完整的大模型应用开发链路你已经全部掌握了。希望以后再接再厉！

如果文章对你有帮助，欢迎点赞收藏，有问题评论区交流。

Logo
MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

cover

MateClaw 2.0 正式发布：从“一个能干活的人”到“一支能协作的队伍”

avatar MCP技术社区
cover

MCP更新： 2026-07-28：从 Agent 插件协议到企业 AI 工具总线

avatar MCP技术社区
cover

Nature正刊！| 基于可追溯推理的智能体系统重塑罕见病诊断

avatar MCP技术社区