LangChain Agent 完整实战：从工具定义到部署

前几章分别讲了 Tools、Memory、Retriever、Callback 等组件，本章把它们组合起来，构建一个完整的、生产可用的 Agent。目标是搭建一个能查天气、搜索新闻、执行数学计算的助手，并覆盖工具定义、错误处理、调试和部署的全流程。

1. LangChain Agent 的完整构建流程

LangChain Agent完整架构图
LangChain Agent 完整架构 — 从工具定义到 API 服务部署

2. create_tool_calling_agent vs create_react_agent

LangChain 提供两种主流 Agent 创建方式：

方式	适用模型	工具调用格式	推荐程度
`create_react_agent`	所有模型（包括不支持 function calling 的）	文本格式（Action/Observation）	兼容性好，但容易解析出错
`create_tool_calling_agent`	支持 function calling（函数调用：LLM 原生支持调用外部工具的能力，模型会输出结构化 JSON 指定要调用哪个函数和参数）的模型（GPT-4、Claude、Gemini、DeepSeek）	结构化 JSON（原生 API 支持）	推荐，稳定可靠

优先用 create_tool_calling_agent。它依赖模型原生的 function calling 能力，工具调用通过结构化 JSON 传递，不依赖模型能否正确格式化文本，解析错误率极低。

3. 工具定义

3.1. 使用 @tool 装饰器

最简单的工具定义方式，函数的 docstring 就是工具的描述（模型靠这个理解何时调用这个工具）：

python

import os
import json
import math
import requests
from typing import Optional
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """
    查询指定城市的当前天气信息。
    当用户询问某个城市的天气、温度、是否下雨时使用此工具。

    Args:
        city: 城市名称，支持中文（如"北京"）或英文（如"Beijing"）
    """
    # 这里使用 OpenWeatherMap 免费 API，需要注册获取 API_KEY
    api_key = os.getenv("OPENWEATHER_API_KEY")
    if not api_key:
        # 开发测试时的模拟数据，避免依赖真实 API
        return json.dumps({
            "city": city,
            "temperature": 22,
            "condition": "晴天",
            "humidity": 65,
            "wind_speed": "3m/s",
        }, ensure_ascii=False)

    url = f"http://api.openweathermap.org/data/2.5/weather"
    params = {"q": city, "appid": api_key, "units": "metric", "lang": "zh_cn"}

    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        return json.dumps({
            "city": city,
            "temperature": round(data["main"]["temp"]),
            "condition": data["weather"][0]["description"],
            "humidity": data["main"]["humidity"],
            "wind_speed": f"{data['wind']['speed']}m/s",
        }, ensure_ascii=False)
    except requests.RequestException as e:
        # 工具函数返回错误描述，而不是抛出异常
        # 让 Agent 知道工具失败了，可以选择重试或给用户解释
        return f"天气查询失败：{str(e)}。请检查城市名称是否正确。"


@tool
def search_news(query: str, max_results: int = 3) -> str:
    """
    搜索最新新闻。当用户询问最近发生的事件、新闻或实时信息时使用此工具。
    注意：此工具返回的是近期新闻，不适合查询历史事实。

    Args:
        query: 搜索关键词，尽量用中文关键词
        max_results: 返回的最大新闻条数，默认3条
    """
    # 实际项目中接入 Serper API、Bing News API 等
    # 这里使用模拟数据演示结构
    mock_news = [
        {"title": f"关于'{query}'的最新进展", "source": "科技日报", "summary": f"{query}领域近期有重要突破..."},
        {"title": f"{query}行业分析报告", "source": "财经网", "summary": f"专家分析{query}未来发展趋势..."},
        {"title": f"深度解读：{query}", "source": "36kr", "summary": f"本文将深入探讨{query}的技术细节..."},
    ]

    results = mock_news[:max_results]
    return json.dumps(results, ensure_ascii=False, indent=2)


@tool
def calculate(expression: str) -> str:
    """
    执行数学计算。支持加减乘除、幂运算、三角函数、对数等。
    当用户需要计算数字结果时使用此工具。

    支持的运算示例：
    - 基本运算："2 + 3 * 4"
    - 幂运算："2 ** 10"
    - 数学函数："math.sqrt(144)"、"math.log(100, 10)"

    Args:
        expression: 数学表达式字符串
    """
    # 限制可用的命名空间，防止代码注入
    # 只允许 math 模块和基本数字类型，禁止 __import__、open 等危险操作
    safe_globals = {
        "__builtins__": {},  # 禁用所有内置函数
        "math": math,
        "abs": abs,
        "round": round,
        "int": int,
        "float": float,
    }

    try:
        result = eval(expression, safe_globals)
        return f"计算结果：{expression} = {result}"
    except ZeroDivisionError:
        return "计算失败：除数不能为零"
    except (SyntaxError, NameError) as e:
        return f"计算失败：表达式格式错误 - {str(e)}"
    except Exception as e:
        return f"计算失败：{str(e)}"

3.2. 使用 StructuredTool 定义复杂工具

当工具参数较多、需要参数验证时，用 Pydantic 模型定义参数结构更合适：

python

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class TranslateInput(BaseModel):
    text: str = Field(description="需要翻译的原文")
    target_language: str = Field(
        description="目标语言，使用语言代码：zh（中文）、en（英文）、ja（日文）、ko（韩文）",
        default="zh"
    )
    formal: bool = Field(description="是否使用正式语气", default=False)

def translate_text(text: str, target_language: str = "zh", formal: bool = False) -> str:
    """调用翻译服务（实际项目中接入 DeepL 或 Google Translate API）"""
    # 模拟翻译
    style = "正式" if formal else "口语"
    return f"[模拟翻译 -> {target_language} ({style}风格)] {text}"

translate_tool = StructuredTool.from_function(
    func=translate_text,
    name="translate",
    description="翻译文本到指定语言。当用户需要翻译文字时使用此工具。",
    args_schema=TranslateInput,
)

4. 构建完整 Agent

python

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.messages import HumanMessage, AIMessage

# 初始化 LLM
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0,  # Agent 场景用 temperature=0，推理更稳定
)

# 工具列表
tools = [get_weather, search_news, calculate]

# Agent Prompt：必须包含 agent_scratchpad 占位符，Agent 用它存放工具调用的中间结果
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个功能全面的智能助手，可以查询天气、搜索新闻和进行数学计算。

工作原则：
1. 需要实时信息（天气、新闻）时，必须调用相应工具，不要凭记忆回答
2. 数学计算统一使用计算器工具，避免心算出错
3. 工具调用失败时，向用户说明情况并提供替代帮助
4. 回答要简洁，重点突出"""),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),  # 必须包含，存放 ReAct 中间步骤
])

# 创建 Agent（注意：create_tool_calling_agent 返回的是 Runnable，不是 AgentExecutor）
agent = create_tool_calling_agent(llm, tools, prompt)

# AgentExecutor 是 Agent 的运行容器，负责循环执行工具调用直到得出最终答案
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,          # 最多调用工具 5 次，防止无限循环
    handle_parsing_errors=True, # 工具调用解析失败时自动恢复，而不是抛异常
    verbose=True,               # 开发时开启，查看推理过程
    return_intermediate_steps=False,  # 生产环境不需要返回中间步骤
)

5. 工具错误处理：Agent 如何恢复

工具失败是常见情况，handle_parsing_errors=True 只处理 Agent 输出格式解析错误，实际工具执行失败需要在工具函数内部处理。

更健壮的做法是给工具加上重试机制和降级逻辑：

python

from functools import wraps
import time
from typing import Callable

def with_retry(max_retries: int = 2, delay_seconds: float = 1.0):
    """工具重试装饰器：网络类工具失败时自动重试"""
    def decorator(func: Callable):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries:
                        # 最后一次失败，返回错误描述（不抛出异常）
                        return f"工具执行失败（已重试 {max_retries} 次）：{str(e)}"
                    time.sleep(delay_seconds * (attempt + 1))  # 指数退避
        return wrapper
    return decorator

# 对网络工具加重试
@tool
@with_retry(max_retries=2, delay_seconds=1.0)
def get_weather_with_retry(city: str) -> str:
    """查询天气（带重试机制）"""
    # 函数体同前面的 get_weather
    ...

6. AgentExecutor 关键配置

python

# 完整配置示例
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,

    # 防止工具调用死循环（重要！）
    max_iterations=10,

    # 按执行时间限制（秒），适合有 SLA（Service Level Agreement，服务级别协议：约定系统响应时间上限）要求的生产环境
    max_execution_time=30.0,

    # 工具调用解析失败时的行为
    # True：把错误信息传给模型让它重试
    # 字符串：直接返回该字符串作为最终答案
    handle_parsing_errors="我遇到了一些问题，请重新描述您的需求。",

    # 开发调试用，生产环境设为 False
    verbose=False,

    # 是否在结果中包含中间步骤（工具调用记录）
    # 设为 True 方便调试，生产环境通常 False
    return_intermediate_steps=True,
)

7. 完整对话实战

python

def run_agent_conversation():
    """演示多轮对话，Agent 保持历史上下文"""
    chat_history = []

    test_queries = [
        "北京今天天气怎么样？",
        "如果我要从北京飞上海，飞行距离大约1200公里，假设飞机时速900km/h，飞行时间是多少分钟？",
        "最近有什么关于人工智能的新闻？",
        "综合前面查到的信息，给我一个去上海出行的建议",  # 需要记忆前面的对话
    ]

    for query in test_queries:
        print(f"\n{'='*50}")
        print(f"用户：{query}")

        result = agent_executor.invoke({
            "input": query,
            "chat_history": chat_history,
        })

        response = result["output"]
        print(f"Agent：{response}")

        # 更新历史，保持多轮对话上下文
        chat_history.append(HumanMessage(content=query))
        chat_history.append(AIMessage(content=response))

        # 只保留最近 10 轮历史，避免 context 过长
        if len(chat_history) > 20:
            chat_history = chat_history[-20:]


if __name__ == "__main__":
    run_agent_conversation()

8. 调试技巧：verbose=True 查看推理过程

开启 verbose=True 后，AgentExecutor 会打印完整的推理过程：

code

> Entering new AgentExecutor chain...

调用工具：get_weather
参数：{"city": "北京"}
返回：{"city": "北京", "temperature": 22, "condition": "晴天", "humidity": 65}

调用工具：calculate
参数：{"expression": "1200 / 900 * 60"}
返回：计算结果：1200 / 900 * 60 = 80.0

最终答案：北京今天晴天，气温22℃，湿度65%，适合出行。
北京到上海飞行时间约80分钟。

> Finished chain.

除了 verbose，还可以用 Callback 获取更结构化的调试信息：

python

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict, List, Union

class AgentDebugCallback(BaseCallbackHandler):
    """记录 Agent 的每一步操作，方便调试和监控"""

    def on_tool_start(self, serialized: Dict, input_str: str, **kwargs):
        tool_name = serialized.get("name", "unknown")
        print(f"[DEBUG] 调用工具：{tool_name}，参数：{input_str[:100]}")

    def on_tool_end(self, output: str, **kwargs):
        print(f"[DEBUG] 工具返回：{output[:100]}")

    def on_tool_error(self, error: Exception, **kwargs):
        print(f"[ERROR] 工具出错：{error}")

    def on_agent_action(self, action, **kwargs):
        print(f"[DEBUG] Agent 决策：调用 {action.tool}")

    def on_agent_finish(self, finish, **kwargs):
        print(f"[DEBUG] Agent 完成，最终答案长度：{len(finish.return_values.get('output', ''))}")


# 注册 Callback
debug_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,
    handle_parsing_errors=True,
    callbacks=[AgentDebugCallback()],
)

9. 封装为 API 服务

实际部署时，Agent 通常以 HTTP API 的形式对外提供服务。以下示例使用 FastAPI（Python 的高性能 Web 框架，用于快速构建 HTTP 接口）和 uvicorn（异步 Web 服务器，负责运行 FastAPI 应用）：

python

# app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn

app = FastAPI(title="AI Agent API", version="1.0.0")

class ChatRequest(BaseModel):
    message: str
    session_id: str = "default"
    history: Optional[List[dict]] = []

class ChatResponse(BaseModel):
    answer: str
    session_id: str

# 会话历史存储（生产环境用 Redis 存储）
session_store: dict = {}

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        # 从存储恢复会话历史
        history = session_store.get(request.session_id, [])

        # 执行 Agent
        result = agent_executor.invoke({
            "input": request.message,
            "chat_history": history,
        })

        answer = result["output"]

        # 更新会话历史
        history.append(HumanMessage(content=request.message))
        history.append(AIMessage(content=answer))
        # 保留最近 20 条消息，控制内存使用
        session_store[request.session_id] = history[-20:]

        return ChatResponse(answer=answer, session_id=request.session_id)

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent 执行失败：{str(e)}")

@app.delete("/session/{session_id}")
async def clear_session(session_id: str):
    """清除会话历史"""
    session_store.pop(session_id, None)
    return {"message": f"会话 {session_id} 已清除"}

@app.get("/health")
async def health_check():
    return {"status": "healthy"}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

启动服务：

bash

pip install fastapi uvicorn

# 开发模式，支持热重载
uvicorn app:app --reload --port 8000

# 测试
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "北京今天天气怎么样？", "session_id": "user_001"}'

10. Agent 执行完整流程

11. 小结

一个生产可用的 LangChain Agent 需要关注的要点：

优先使用 create_tool_calling_agent，稳定性远高于 ReAct 文本格式
工具函数返回错误描述而不是抛异常，让 Agent 能够感知失败并恢复
设置 max_iterations 和 max_execution_time，防止工具循环调用导致的超时和费用失控
开发时开 verbose=True，生产时关掉，改用结构化 Callback 记录日志
会话历史限制长度，避免 context 越来越长导致性能下降和费用增加

LangChain 的核心组件——LCEL、Memory、Tools、OutputParser、Retriever、Callback、DocumentLoader、TextSplitter、Agent——到这里都讲完了。接下来可以找一个真实项目来用，理解会比单独读文档快得多。