AI Agent入门：ReAct框架与工具调用

大模型会"说"但不会"做"——它能生成文本但无法查数据库、调 API、执行代码。AI Agent 就是给大模型加上手脚：让它能推理（Reasoning）、行动（Acting）、观察结果（Observation），形成闭环。本文从 ReAct 框架讲起，用 LangChain 实现一个能调用工具的 Agent。

Agent 的核心循环

一个 Agent 的运行逻辑可以抽象为：

循环:
  1. 思考 (Thought): 分析当前状态，决定下一步
  2. 行动 (Action): 调用一个工具
  3. 观察 (Observation): 获取工具返回的结果
  4. 判断: 是否已经得到最终答案？
     - 是 → 输出结果，结束
     - 否 → 回到步骤 1

这就是 ReAct（Reasoning + Acting）框架的核心。它由 Yao et al. 在 2022 年提出，思路很朴素但效果显著。

ReAct 的 Prompt 结构

ReAct 通过精心设计的 Prompt 让 LLM 按照 Thought → Action → Observation 的格式输出：

Answer the following questions as best you can. You have access to the following tools:

search: Search the web for information
calculator: Do math calculations

Use the following format:

Question: the input question
Thought: reason about what to do
Action: the tool name
Action Input: the input to the tool
Observation: the result of the tool
... (repeat Thought/Action/Observation as needed)
Thought: I now know the final answer
Final Answer: the answer

Begin!

Question: 2024年美国GDP是多少万亿美元？换算成人民币是多少？
Thought: 我需要先搜索2024年美国GDP数据，然后用计算器做汇率换算。
Action: search
Action Input: 2024年美国GDP总量
Observation: 2024年美国GDP约为28.78万亿美元。
Thought: 得到了GDP数据，现在需要乘以汇率。假设汇率约7.2。
Action: calculator
Action Input: 28.78 * 7.2
Observation: 207.216
Thought: I now know the final answer
Final Answer: 2024年美国GDP约为28.78万亿美元，按7.2汇率换算约为207.2万亿人民币。

LLM 每次只生成到 Observation 之前停止，由框架执行工具并填入 Observation，再把完整文本送回 LLM 继续生成。

工具定义

工具（Tool）是 Agent 的手脚。定义一个工具需要三要素：

1. 名称：让 LLM 知道有哪些工具可用
2. 描述：让 LLM 理解什么时候该用这个工具
3. 输入格式：让 LLM 知道怎么构造输入

from langchain.tools import Tool, tool
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field


# 方式1：用 @tool 装饰器（最简单）
@tool
def search_web(query: str) -> str:
    '''搜索互联网获取实时信息。当需要查询最新数据、新闻、事实时使用。'''
    # 实际实现：调用搜索 API
    import requests
    resp = requests.get(f"https://api.search.example/search?q={query}")
    return resp.json()["answer"]


# 方式2：结构化工具（复杂输入）
class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")
    date: str = Field(description="日期，格式 YYYY-MM-DD", default="today")

@tool(args_schema=WeatherInput)
def get_weather(city: str, date: str = "today") -> str:
    '''查询指定城市的天气预报。'''
    return f"{city}在{date}的天气：晴，25°C"


# 方式3：Tool 对象（最灵活）
def run_python_code(code: str) -> str:
    '''执行 Python 代码并返回结果'''
    try:
        result = eval(code)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

python_tool = Tool(
    name="python_executor",
    func=run_python_code,
    description="执行 Python 表达式。用于数学计算、数据处理等。输入应为合法的 Python 表达式。",
)

工具描述（description）的质量直接影响 Agent 的表现。描述要说清楚：

• 这个工具做什么
• 什么时候该用它
• 输入是什么格式

LangChain Agent 实现

用 LangChain 实现一个完整的 ReAct Agent：

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

os.environ["OPENAI_API_KEY"] = "your-key"
os.environ["OPENAI_BASE_URL"] = "https://api.your-provider.com/v1"

# === 定义工具 ===

@tool
def search_knowledge_base(query: str) -> str:
    '''从内部知识库搜索信息。适用于查询公司产品、政策、技术文档等内部信息。'''
    # 模拟知识库搜索
    kb = {
        "退货政策": "自签收之日起7天内可无理由退货，商品需保持原包装完好。",
        "VIP折扣": "VIP用户享受全场9折优惠，部分商品可叠加满减活动。",
        "配送时间": "同城次日达，跨省2-5个工作日。偏远地区可能延迟。",
    }
    for key, value in kb.items():
        if key in query or any(c in query for c in key):
            return value
    return "未找到相关信息"


@tool
def query_order(order_id: str) -> str:
    '''查询订单状态。输入订单编号，返回订单详情。'''
    # 模拟订单查询
    orders = {
        "ORD20240601": "已发货，预计6月3日送达，快递单号SF1234567890",
        "ORD20240530": "已签收，6月1日14:32签收",
    }
    return orders.get(order_id, f"订单 {order_id} 不存在")


@tool
def calculator(expression: str) -> str:
    '''计算数学表达式。输入一个合法的数学表达式字符串。'''
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误: {e}"


# === 构建 Agent ===

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

tools = [search_knowledge_base, query_order, calculator]

# 获取 ReAct prompt 模板
prompt = hub.pull("hwchase17/neact")

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

# AgentExecutor 负责执行循环
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,           # 打印推理过程
    max_iterations=5,       # 最大循环次数，防止无限循环
    handle_parsing_errors=True,  # 自动处理输出格式错误
)

# === 运行 ===

result = agent_executor.invoke({
    "input": "我的订单ORD20240601到哪了？另外VIP用户有什么优惠？"
})

print(result["output"])

运行输出（verbose=True 时）：

> Entering new AgentExecutor chain...
Thought: 用户问了两个问题：订单状态和VIP优惠。我先查订单。
Action: query_order
Action Input: ORD20240601
Observation: 已发货，预计6月3日送达，快递单号SF1234567890

Thought: 订单信息拿到了，现在查VIP优惠。
Action: search_knowledge_base
Action Input: VIP折扣优惠
Observation: VIP用户享受全场9折优惠，部分商品可叠加满减活动。

Thought: 两个问题都有答案了。
Final Answer: 您的订单 ORD20240601 已发货，预计6月3日送达，
快递单号 SF1234567890。关于VIP优惠，VIP用户享受全场9折，
部分商品还可以叠加满减活动。

Function Calling

OpenAI 的 Function Calling 是比 ReAct Prompt 更原生的方案。它不是通过 Prompt 让 LLM 输出特定格式的文本，而是在 API 层面支持工具调用：

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 使用 Function Calling 的 Agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的客服助手。尽可能帮助用户解决问题。"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# create_tool_calling_agent 使用 Function Calling 而非 ReAct Prompt
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({"input": "帮我查一下 ORD20240601"})

Function Calling 的优势：

• 更可靠：不依赖 LLM 精确输出特定文本格式
• 并行调用：支持一次返回多个工具调用
• 结构化输入：参数以 JSON 传递，不是纯文本解析

ReAct Prompt 的优势：

• 模型无关：任何 LLM 都能用（不需要模型支持 Function Calling）
• 推理过程可见：Thought 部分暴露了 LLM 的推理链
• 灵活：可以自定义 Prompt 格式

自己实现一个最小 Agent

不依赖 LangChain，用纯 OpenAI API 实现一个最小的 ReAct Agent：

import openai
import json

client = openai.OpenAI()

# 工具定义
tools_schema = [
    {
        "type": "function",
        "function": {
            "name": "search",
            "description": "搜索信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "数学计算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "数学表达式"}
                },
                "required": ["expression"]
            }
        }
    },
]

# 工具实现
def execute_tool(name: str, args: dict) -> str:
    if name == "search":
        return f"搜索结果：关于「{args['query']}」的信息..."
    elif name == "calculate":
        return str(eval(args["expression"]))
    return "未知工具"


def run_agent(user_input: str, max_steps: int = 5) -> str:
    messages = [
        {"role": "system", "content": "你是一个有用的助手。"},
        {"role": "user", "content": user_input},
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools_schema,
            tool_choice="auto",
        )

        msg = response.choices[0].message

        # 如果没有工具调用，说明得到了最终答案
        if not msg.tool_calls:
            return msg.content

        # 执行工具调用
        messages.append(msg)
        for tool_call in msg.tool_calls:
            fn_name = tool_call.function.name
            fn_args = json.loads(tool_call.function.arguments)

            print(f"  [Tool] {fn_name}({fn_args})")
            result = execute_tool(fn_name, fn_args)
            print(f"  [Result] {result}")

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })

    return "达到最大步数，未能得出结论。"


# 测试
answer = run_agent("今年全球智能手机出货量是多少？换算成每秒出货多少台？")
print(f"\n最终答案: {answer}")

Agent 的局限

Agent 很强大但不是银弹：

1. 延迟高：每次循环都要调一次 LLM，多步推理可能要 10-30 秒
2. 成本高：每次工具调用都消耗 token，复杂任务可能需要很多轮
3. 不稳定：LLM 可能输出格式错误、选错工具、陷入死循环
4. 安全风险：如果工具能执行代码或操作数据库，需要严格的权限控制

生产环境中的 Agent 需要额外考虑：

• 设置 max_iterations 防止无限循环
• 工具执行结果做截断，避免超长 context
• 对工具输入做校验和沙箱隔离
• 记录完整的推理链用于审计
• 失败时的降级策略（fallback 到人工）

Agent 是 LLM 应用从"生成文本"走向"执行任务"的关键一步。理解 ReAct 框架和工具调用的原理，是构建更复杂 AI 应用的基础。