Python AI Agent 实战：使用 LangChain 构建具备工具调用能力的智能助手

一、AI Agent 的核心概念

AI Agent 是能够自主感知环境、做出决策并执行动作的软件实体。在 LLM 语境下，它将大模型作为“大脑”，通过提示词范式让模型产生“思考→行动→观察→再思考”的循环。与普通聊天机器人不同，Agent 可以：

• 调用工具：搜索引擎、计算器、API、数据库等。
• 多步规划：将复杂问题分解为多个子任务顺序执行。
• 记忆与反思：保持对话上下文，并根据行动结果调整后续决策。

目前主流的 Agent 实现范式是 ReAct（Reasoning + Acting），它交替进行推理和行动，直到得出最终答案。LangChain 提供了经过封装的 Agent 框架，让我们可以专注于工具的定义与组合。

二、环境准备与依赖安装

本教程使用 Python 3.10+ 和 LangChain 0.3 版本，并集成 OpenAI 的 gpt-3.5-turbo 模型。首先安装所需包：

pip install langchain langchain-openai python-dotenv

在项目根目录创建 .env 文件存储 API 密钥：

OPENAI_API_KEY=your_openai_api_key_here

加载环境变量并导入必要模块：

import os
from dotenv import load_dotenv

load_dotenv()

from langchain_openai import ChatOpenAI
from langchain.agents import load_tools, create_react_agent, AgentExecutor
from langchain.tools import tool
from langchain import hub

三、定义自定义工具：搜索、计算与数据库查询

工具是 Agent 的手和眼。我们将定义三个实用工具：网络搜索（模拟）、数学计算和 SQLite 数据库查询。每个工具使用 @tool 装饰器标注，并提供清晰的描述。

3.1 模拟搜索工具

@tool
def web_search(query: str) -> str:
    """当需要查找实时信息、百科知识或任何未知事实时使用此工具。输入一个搜索查询字符串，返回搜索结果摘要。"""
    # 实际项目中应调用 SerpAPI 或 Google Custom Search
    search_results = {
        "东京天气": "东京今天多云，气温 15°C 至 22°C。",
        "Python创始人": "Python 的创始人是 Guido van Rossum，于 1991 年首次发布。",
    }
    for key, value in search_results.items():
        if key in query:
            return value
    return f"未找到与 '{query}' 直接匹配的结果，请尝试更换关键词。"

3.2 数学计算工具

@tool
def calculator(expression: str) -> str:
    """执行数学表达式计算。输入一个有效的 Python 数学表达式，例如 '3 * 7 + 2'，返回计算结果。"""
    try:
        result = eval(expression, {"__builtins__": None}, {})
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算出错: {str(e)}"

3.3 数据库查询工具

该工具连接本地 SQLite 数据库，允许 Agent 查询订单信息。

import sqlite3

@tool
def query_database(query: str) -> str:
    """查询 SQLite 数据库中的订单表。输入 SQL SELECT 语句，例如 'SELECT * FROM orders WHERE id = 1'，返回查询结果。"""
    conn = sqlite3.connect("orders.db")
    cursor = conn.cursor()
    try:
        cursor.execute(query)
        rows = cursor.fetchall()
        if not rows:
            return "查询无结果。"
        # 简单格式化输出
        columns = [desc[0] for desc in cursor.description]
        result = str(columns) + "n"
        for row in rows:
            result += str(row) + "n"
        conn.close()
        return result
    except Exception as e:
        conn.close()
        return f"数据库查询失败: {str(e)}"

最后将所有工具放入列表，供 Agent 使用。

四、构建 ReAct Agent 并执行任务

LangChain 的 Hub 提供了大量预置的提示词模板。我们拉取经典的 ReAct 模板，创建 LLM 和 Agent。

# 创建 LLM 实例，降低温度以获得更确定的工具调用
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 定义工具列表
tools = [web_search, calculator, query_database]

# 从 LangChain Hub 加载 ReAct 提示词模板
prompt = hub.pull("hwchase17/react")

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

# 创建 Agent 执行器，verbose=True 会输出完整的思考过程
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=10,   # 最大推理步数
)

现在我们可以向 Agent 提问了。例如：“东京今天天气如何？将温度转换为华氏度，并查询订单ID为1的客户姓名。”

response = agent_executor.invoke({
    "input": "东京今天天气如何？将温度转换为华氏度（公式：F = C * 9/5 + 32），并查询订单ID为1的客户姓名。"
})
print(response["output"])

Agent 会经历以下步骤：

1. Thought：分析问题，决定先查天气。
2. Action：调用 web_search 工具，输入“东京天气”。
3. Observation：得到天气信息，包含摄氏度温度。
4. Thought：需要将摄氏度转为华氏度，使用 calculator。
5. Action：调用 calculator，输入“15 * 9/5 + 32”（以最低温为例）。
6. Observation：得到结果 59.0。
7. Thought：接下来查询订单数据库。
8. Action：调用 query_database，输入 SQL 语句。
9. Observation：返回订单信息。
10. Thought：我已获取所有信息，可以给出最终答案。
11. Final Answer：整合所有结果返回给用户。

五、自定义推理循环：深入理解 Agent 核心

如果不想依赖 LangChain 的高级封装，也可以手动实现 ReAct 循环，这有助于理解其本质。以下是一个简化的自定义 Agent：

import re

class SimpleReActAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.tool_descriptions = "n".join([f"- {t.name}: {t.description}" for t in tools])

    def run(self, user_input, max_steps=5):
        context = f"你有以下工具可用：n{self.tool_descriptions}nn用户问题：{user_input}n"
        for _ in range(max_steps):
            prompt = f"{context}n请思考下一步行动。输出格式：nThought: 你的思考nAction: 工具名称nAction Input: 工具输入n或者 Final Answer: 最终答案n"
            response = self.llm.invoke(prompt).content

            if "Final Answer:" in response:
                return response.split("Final Answer:")[-1].strip()

            action_match = re.search(r"Action: (.*)", response)
            input_match = re.search(r"Action Input: (.*)", response)
            if action_match and input_match:
                action_name = action_match.group(1).strip()
                action_input = input_match.group(1).strip()
                tool = self.tools.get(action_name)
                if tool:
                    observation = tool.invoke(action_input)
                    context += f"n观察结果：{observation}n"
                else:
                    context += f"n工具 '{action_name}' 不存在。n"
        return "无法在限定步骤内给出答案。"

这个精简版本展示了 Agent 的骨架：解析 LLM 输出、调用对应工具、将观察结果反馈给模型，直至得到最终答案。实际应用中 LangChain 的 AgentExecutor 还包含错误处理、输出解析器、记忆管理等，推荐直接使用。

六、添加记忆与多轮对话

为了让 Agent 记住之前的对话，可以引入 Memory 组件。在 LangChain 中，只需在 AgentExecutor 中传入 memory 即可：

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(memory_key="chat_history")
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

# 第一轮
agent_executor.invoke({"input": "我叫小明。"})
# 第二轮，Agent 能记住名字
agent_executor.invoke({"input": "我的名字是什么？"})

通过此方式，Agent 可以跨轮次维护上下文，实现更自然的多轮交互。

七、生产环境部署建议

• 工具安全：使用 eval 存在风险，计算器应使用 ast.literal_eval 或专门的数学库。数据库查询工具应限制权限，使用参数化查询防止注入。
• 错误重试：大模型的输出格式可能不符合预期，务必设置 handle_parsing_errors=True，并提供后备逻辑。
• 成本控制：限制最大迭代次数，避免 Agent 陷入死循环消耗大量 token。
• 异步调用：对于高并发场景，使用 AsyncAgentExecutor 提升吞吐量。

八、总结

本文从零构建了一个功能完整的 Python AI Agent，演示了如何定义工具、创建 ReAct Agent、手动实现推理循环以及添加记忆。通过这套模式，你可以轻松扩展自己的 Agent：添加邮件发送、日历操作、文件读写等工具，让大模型真正成为数字世界的行动者。

AI Agent 正处于爆发前夜，掌握其构建原理将让你在智能化浪潮中抢占先机。现在，就打开你的 Python 环境，开始构建属于你自己的智能助手吧。