实战：构建 AI Agent 工具链

Agent 定义 → Tool 注册 → 记忆管理 → LLM 调用循环 → 错误恢复全流程实战

项目概述

本文档将引导你从零构建一个能使用多种工具完成任务的 AI Agent。该 Agent 的核心能力包括：理解用户自然语言指令、自主选择并调用合适的工具、维护多轮对话上下文，以及在出现异常时进行错误恢复。

功能目标

构建一个能够通过自然语言与用户交互，并调用外部工具完成实际任务的智能 Agent。典型场景包括查询实时天气、执行数学计算、搜索网络信息等。Agent 需要能够自主判断何时调用工具、调用哪个工具，并将工具返回的结果整合为自然语言回复。

技术栈

LangChain：作为 Agent 编排框架，提供 Tool 注册、Agent 执行器、记忆管理等核心抽象
OpenAI：作为底层大语言模型驱动 Agent 的推理与决策
Tavily Search：作为网络搜索工具，为 Agent 提供实时信息获取能力

环境准备

首先安装所需的依赖包：

bash

pip install langchain langchain-openai langchain-community tavily-python

配置环境变量，用于访问 OpenAI 和 Tavily 的 API：

bash

export OPENAI_API_KEY="your-openai-api-key"
export TAVILY_API_KEY="your-tavily-api-key"

Step 1：Tool 注册

工具是 Agent 与外部世界交互的接口。LangChain 提供了多种方式来定义工具，最灵活的方式是使用 @tool 装饰器将任意 Python 函数注册为工具。

自定义工具示例

以下定义三个工具：获取天气、计算器、网页搜索。

python

from langchain_core.tools import tool
from langchain_community.tools.tavily_search import TavilySearchResults
from datetime import datetime

@tool
def get_weather(city: str) -> str:
    """获取指定城市的当前天气信息。"""
    # 实际项目中应调用天气 API
    return f"{city} 的天气：晴，温度 26°C，湿度 60%"

@tool
def calculator(expression: str) -> str:
    """执行数学计算，支持四则运算。"""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

@tool
def web_search(query: str) -> str:
    """搜索网络信息，返回最新的相关内容。"""
    search = TavilySearchResults(max_results=3)
    results = search.invoke(query)
    return "\n".join(
        f"[{r['title']}]({r['url']})：{r['content']}"
        for r in results
    )

关键要点：

函数的文档字符串（docstring）会被 LangChain 自动解析为工具描述，LLM 据此决定何时调用该工具
类型注解让 LangChain 能够自动生成工具的 JSON Schema 参数定义
@tool 装饰器将普通函数转换为 BaseTool 实例，可直接注册到 Agent

工具集合

将定义好的工具聚合为一个列表，供 Agent 初始化时使用：

python

tools = [get_weather, calculator, web_search]

Step 2：Agent 定义

Agent 是连接 LLM 和工具的核心组件。它负责接收用户输入，让 LLM 决定下一步动作（直接回答或调用工具），然后执行动作并反馈结果。

Agent 类型选择

LangChain 提供多种 Agent 类型，各有适用场景：

Agent 类型	特点	适用场景
OpenAI Tools Agent	原生支持 OpenAI 的 tool calling，性能最优	使用 gpt-3.5-turbo / gpt-4 等支持 tool calling 的模型
ReAct Agent	基于 ReAct（Reasoning + Acting）范式，兼容性广	需要与多种模型后端配合时
Structured Chat Agent	支持结构化多轮对话，记忆管理更灵活	需要复杂对话管理的场景

本文选用 OpenAI Tools Agent，它利用 OpenAI 原生 tool calling 能力，工具解析更准确，调用更稳定。

Agent 初始化配置

python

from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# 初始化 LLM
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    max_tokens=4096,
)

# 定义系统提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，可以使用工具帮助用户完成任务。"
     "请根据用户的问题选择合适的工具，并将工具返回的结果用自然语言回复给用户。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# 创建 Agent
agent = create_openai_tools_agent(
    llm=llm,
    tools=tools,
    prompt=prompt,
)

agent_scratchpad 是 LangChain 内部使用的占位符，用于存储 Agent 的中间推理步骤，帮助 LLM 跟踪当前执行状态。

Step 3：记忆管理

无记忆的 Agent 每次调用都是独立的，无法引用前文。通过 ConversationBufferMemory 可以让 Agent 记住对话历史。

python

from langchain.memory import ConversationBufferMemory
from langchain.agents import AgentExecutor

# 初始化对话记忆
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="output",
)

# 创建带记忆的 Agent 执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True,
    max_iterations=5,
    early_stopping_method="generate",
    handle_parsing_errors=True,
)

记忆管理的关键参数：

memory_key：必须与提示词模板中的占位符名称一致（本例中为 chat_history）
return_messages=True：以消息列表而非字符串形式存储历史，保留角色信息

Step 4：LLM 调用循环

AgentExecutor 封装了 Agent 的核心执行循环。理解其内部机制对调试和性能优化至关重要。

执行循环流程

接收输入：将用户消息与 chat_history、agent_scratchpad 组装为完整提示词
LLM 推理：将提示词发送给 LLM，获取响应
动作决策：
- 如果 LLM 返回的是最终答案，则结束循环并返回结果
- 如果 LLM 返回的是工具调用指令，则进入下一步
工具执行：解析工具名称和参数，调用对应的工具函数
结果反馈：将工具执行结果追加到 agent_scratchpad
循环迭代：回到步骤 2，继续推理直到得到最终答案或达到最大迭代次数

关键配置

python

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,              # 最大迭代次数，防止无限循环
    early_stopping_method="generate",  # 达到上限时让 LLM 生成一个合理回复
)

max_iterations：限制工具调用的最大轮次，防止 Agent 陷入死循环。对于简单任务设置为 3-5，复杂任务可适当调高
early_stopping_method="generate"：当达到最大迭代次数时，让 LLM 基于已有信息生成一个最终回复，比直接返回错误更友好

Step 5：错误恢复

生产环境中工具调用不可避免会出现异常。合理的错误恢复机制是 Agent 稳定运行的保障。

ToolException 处理

当工具执行失败时，LangChain 允许自定义错误处理逻辑：

python

from langchain_core.tools import ToolException

@tool
def safe_web_search(query: str) -> str:
    """带错误处理的网络搜索。"""
    try:
        search = TavilySearchResults(max_results=3)
        results = search.invoke(query)
        return "\n".join(
            f"[{r['title']}]({r['url']})：{r['content']}"
            for r in results
        )
    except Exception as e:
        raise ToolException(f"搜索服务暂时不可用：{str(e)}")

解析错误处理

当 LLM 返回的格式不符合预期时（例如 JSON 解析失败），handle_parsing_errors 配置可以让 Agent 自动重试：

python

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors=True,         # 自动处理解析错误
    max_iterations=5,
    early_stopping_method="generate",
)

handle_parsing_errors 支持三种配置方式：

True：使用默认错误提示让 LLM 重试
字符串：自定义错误消息模板
可调用对象：传入自定义的错误处理函数，实现更复杂的重试逻辑

重试机制示例

python

# 带重试的工具调用
import time
from functools import wraps

def retry_on_failure(max_retries=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        time.sleep(delay)
            raise ToolException(f"重试 {max_retries} 次后仍然失败：{last_exception}")
        return wrapper
    return decorator

@tool
@retry_on_failure(max_retries=2, delay=0.5)
def robust_web_search(query: str) -> str:
    """带重试机制的网络搜索。"""
    search = TavilySearchResults(max_results=3)
    results = search.invoke(query)
    return "\n".join(f"[{r['title']}]({r['url']})：{r['content']}" for r in results)

完整代码

以下整合以上所有步骤，形成一个完整的 Agent 脚本：

python

import os
from datetime import datetime
from functools import wraps
import time

from langchain_core.tools import tool, ToolException
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain.memory import ConversationBufferMemory
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.tools.tavily_search import TavilySearchResults

# ---------- Tool 注册 ----------

@tool
def get_weather(city: str) -> str:
    """获取指定城市的当前天气信息。"""
    return f"{city} 的天气：晴，温度 26°C，湿度 60%"

@tool
def calculator(expression: str) -> str:
    """执行数学计算，支持四则运算。"""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

def retry_on_failure(max_retries=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        time.sleep(delay)
            raise ToolException(
                f"重试 {max_retries} 次后仍然失败：{last_exception}"
            )
        return wrapper
    return decorator

@tool
@retry_on_failure(max_retries=2, delay=0.5)
def web_search(query: str) -> str:
    """搜索网络信息，返回最新的相关内容。"""
    search = TavilySearchResults(max_results=3)
    results = search.invoke(query)
    return "\n".join(
        f"[{r['title']}]({r['url']})：{r['content']}"
        for r in results
    )

tools = [get_weather, calculator, web_search]

# ---------- Agent 定义 ----------

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    max_tokens=4096,
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，可以使用工具帮助用户完成任务。"
     "请根据用户的问题选择合适的工具，并将工具返回的结果用自然语言回复给用户。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_openai_tools_agent(
    llm=llm,
    tools=tools,
    prompt=prompt,
)

# ---------- 记忆管理 ----------

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="output",
)

# ---------- LLM 调用循环 + 错误恢复 ----------

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True,
    max_iterations=5,
    early_stopping_method="generate",
    handle_parsing_errors=True,
)

# ---------- 启动交互 ----------

if __name__ == "__main__":
    print("AI Agent 已启动，输入 'exit' 退出")
    while True:
        user_input = input("\n用户：")
        if user_input.lower() in ("exit", "quit"):
            break
        response = agent_executor.invoke({"input": user_input})
        print(f"Agent：{response['output']}")

运行效果示例

用户：今天北京天气怎么样？
Agent：北京今天的天气：晴，温度 26°C，湿度 60%。

用户：帮我算一下 15 * 23 + 100 等于多少？
Agent：计算结果：15 * 23 + 100 = 445。

用户：搜索一下最近的 AI 新闻
Agent：以下是最近的 AI 相关新闻：
[新闻标题1](链接1)：内容摘要...
[新闻标题2](链接2)：内容摘要...

总结

本文从零构建了一个完整的 AI Agent 工具链，涵盖了工具注册、Agent 定义、记忆管理、执行循环和错误恢复五个核心环节。这套架构可以直接应用于实际项目，根据业务需求扩展更多自定义工具，或替换不同的 LLM 后端和记忆策略。

实战：构建 AI Agent 工具链 ​

项目概述 ​

功能目标 ​

技术栈 ​

环境准备 ​

Step 1：Tool 注册 ​

自定义工具示例 ​

工具集合 ​

Step 2：Agent 定义 ​

Agent 类型选择 ​

Agent 初始化配置 ​

Step 3：记忆管理 ​

Step 4：LLM 调用循环 ​

执行循环流程 ​

关键配置 ​

Step 5：错误恢复 ​

ToolException 处理 ​

解析错误处理 ​

重试机制示例 ​

完整代码 ​

运行效果示例 ​

总结 ​

实战：构建 AI Agent 工具链

项目概述

功能目标

技术栈

环境准备

Step 1：Tool 注册

自定义工具示例

工具集合

Step 2：Agent 定义

Agent 类型选择

Agent 初始化配置

Step 3：记忆管理

Step 4：LLM 调用循环

执行循环流程

关键配置

Step 5：错误恢复

ToolException 处理

解析错误处理

重试机制示例

完整代码

运行效果示例

总结