从零手撸一个会调工具的 Agent 执行循环（ReAct：思考到工具到观察）

我们只需要一个支持 tools 参数的 SDK。这里用官方的 openai Python 包，它同样能连 DeepSeek。

pip install openai

# 在终端导出密钥（不要写进代码里）
export OPENAI_API_KEY="sk-你的密钥"
export OPENAI_BASE_URL="https://api.deepseek.com/v1"

Agent 的工具就是一个函数。我们先用普通 Python 写两个工具，再把它们登记进一张「工具表」，交给模型。

import math

def calc(expression: str) -> str:
    # 教学用：只接受四则运算，避免 eval 执行任意代码
    allowed = set("0123456789.+-*/() ")
    if not set(expression) <= allowed:
        return "错误：表达式含非法字符"
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"错误：{e}"

def fake_db(query: str) -> str:
    # 模拟一个本地知识库
    data = {
        "北京": "北京到上海高铁约 4.5 小时，二等座 553 元",
        "上海": "上海今日多云，气温 22 到 28 度",
    }
    return data.get(query, "未收录该城市")

TOOLS = {
    "calc": calc,
    "fake_db": fake_db,
}

模型看不懂 Python 函数，它只看 JSON 描述的工具清单。我们按 OpenAI tools 格式写一份。

tool_schemas = [
    {
        "type": "function",
        "function": {
            "name": "calc",
            "description": "计算一个四则运算表达式，例如 553 * 2",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "四则运算表达式"}
                },
                "required": ["expression"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "fake_db",
            "description": "查询本地城市知识库",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "城市名"}
                },
                "required": ["query"],
            },
        },
    },
]

messages 里维护完整上下文：系统提示、用户问题、以及后续每一步的工具调用与结果。

from openai import OpenAI
client = OpenAI(base_url=OPENAI_BASE_URL, api_key=OPENAI_API_KEY)

messages = [
    {"role": "system", "content": "你是会调用工具的助手，优先用工具回答，后用一句话给结论。"},
    {"role": "user", "content": "北京到上海高铁二等座 553 元，买 2 张加 1 张儿童半价票一共多少钱？"},
]

resp = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    tools=tool_schemas,
    tool_choice="auto",
)
msg = resp.choices[0].message
print(msg.content)            # 可能是 None，因为模型决定先调工具
print(msg.tool_calls)         # 模型想调的工具与参数

拿到 tool_calls 后，我们本地执行对应函数，把结果作为 tool 角色消息追加进上下文，再请求一次。

import json

# 1) 把模型的回复（含 tool_calls）原样放回上下文
messages.append(msg)

# 2) 逐个执行模型点名的工具
for call in msg.tool_calls:
    name = call.function.name
    args = json.loads(call.function.arguments)
    result = TOOLS[name](**args)          # 本地执行
    messages.append({
        "role": "tool",
        "tool_call_id": call.id,
        "content": str(result),
    })

# 3) 带着工具结果再问一次模型
resp2 = client.chat.completions.create(
    model="deepseek-chat", messages=messages, tools=tool_schemas)
print(resp2.choices[0].message.content)

只要模型还在要工具，就继续循环；一旦不再返回 tool_calls，说明它已经能给最终答案。

def run_agent(user_input, max_steps=8):
    messages = [
        {"role": "system", "content": "你是会调用工具的助手，优先用工具，后给一句话结论。"},
        {"role": "user", "content": user_input},
    ]
    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model="deepseek-chat", messages=messages, tools=tool_schemas)
        msg = resp.choices[0].message
        if not msg.tool_calls:
            return msg.content          # 没有工具调用 = 结束
        messages.append(msg)
        for call in msg.tool_calls:
            args = json.loads(call.function.arguments)
            result = TOOLS[call.function.name](**args)
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": str(result),
            })
    return "步骤超限，未能在限定轮次内完成"

把前面的工具表、schema、run_agent 拼起来，运行下面这句，模型应当先查 fake_db 拿到「553 元」，再调 calc 算出总价。

answer = run_agent("北京到上海高铁二等座 553 元，买 2 张成人票加 1 张儿童半价票，一共多少钱？")
print(answer)
# 预期：先 fake_db 得到单价，再 calc 计算 553*2 + 553*0.5 = 1382.5