Python Tool Calling Agent

手写一个带工具调用的 Agent，理解 tool calling 的完整流程。

#type / howto #status / growing #tech / lang / python #resource / python #tech / ai #resource / agent

[!info] related notes

前置笔记: Python 调用 AI 接口

所属 MOC: 学习 AI Agent MOC

相关概念: Function Calling, Agent 中的 Tool Use

相关资源: OpenRouter

Python Tool Calling Agent

目标

手写一个带工具调用的 Agent，理解 Agent = LLM + tools + loop 的完整流程。

和 Day 1 的区别

	Day 1（调通 API）	Day 2（Tool Agent）
流程	用户 → 模型 → 回答	用户 → 模型 → 可能调用工具 → 基于结果回答
模型能力	只用自己知道的知识	能使用外部能力（查天气、算数学、搜笔记…）
响应方式	流式 SSE	非流式（tool_calls 是结构化数据，流式处理更复杂）

前置条件

复用 Day 1 的 .env（OPENROUTER_API_KEY）
uv add requests python-dotenv
项目结构：

project/
├── .env
├── main.py      # Agent 主循环
└── tools.py     # 工具函数 + 工具描述

核心理解

Tool Calling 的工作原理

用户: "东京天气怎么样？"
        ↓
   调用 API（附带 tools 描述）
        ↓
   模型返回: tool_calls = [{name: "get_weather", arguments: {city: "东京"}}]
        ↓
   代码执行: get_weather("东京") → "25°C，晴天"
        ↓
   把结果作为 tool message 发回模型
        ↓
   模型基于结果生成最终回答: "东京现在 25°C，晴天..."

关键点：模型不执行工具，只决定”要不要调用、调哪个、传什么参数”。执行是代码的事。

Tool 描述的结构

每个工具有两部分：

函数本身（Python 代码，真正执行）
JSON Schema 描述（告诉模型这个工具能做什么、需要什么参数）

{
  "type": "function",
  "function": {
    "name": "get_weather",
    "description": "获取指定城市的当前天气信息",
    "parameters": {
      "type": "object",
      "properties": {
        "city": { "type": "string", "description": "城市名称" }
      },
      "required": ["city"]
    }
  }
}

描述写得好不好，直接影响模型判断”什么时候该调用”。

完整代码

tools.py — 工具定义

[!note]- 展开查看 tools.py

"""
工具函数定义。

每个工具有两个部分：
1. 函数本身（Python 代码，真正执行逻辑）
2. 工具描述（JSON schema，告诉模型这个工具能做什么、需要什么参数）

模型靠"工具描述"来决定什么时候调用哪个工具。
所以描述写得好不好，直接影响模型的判断。
"""


# ============================================================
# 工具函数（真正执行的代码）
# ============================================================


def get_weather(city: str) -> str:
    """获取天气（模拟数据，实际项目会调用天气 API）"""
    mock_data = {
        "东京": "25°C，晴天，湿度 60%",
        "北京": "18°C，多云，空气质量 良",
        "上海": "22°C，小雨，记得带伞",
        "new york": "15°C，cloudy, wind 12mph",
        "london": "12°C，rainy, bring an umbrella",
    }
    city_lower = city.lower().strip()
    if city_lower in mock_data:
        return f"{city}的天气：{mock_data[city_lower]}"
    return f"{city}的天气数据暂不可用（这是模拟数据，只支持几个城市）"


def calculate(expression: str) -> str:
    """计算数学表达式"""
    allowed = set("0123456789+-*/.() ")
    if not all(c in allowed for c in expression):
        return f"错误：表达式 '{expression}' 包含不允许的字符"
    try:
        result = eval(expression)
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误：{e}"


def search_notes(keyword: str) -> str:
    """搜索本地笔记（模拟数据）"""
    notes = [
        {"title": "Python 装饰器", "content": "装饰器是一个接收函数并返回函数的函数..."},
        {"title": "Git 常用命令", "content": "git add . && git commit -m 'msg' && git push"},
        {"title": "Docker 入门", "content": "docker build -t myapp . && docker run -p 8080:80 myapp"},
        {"title": "FastAPI 教程", "content": "用 @app.get('/path') 定义路由..."},
        {"title": "Python 虚拟环境", "content": "uv init 创建项目，uv add 添加依赖..."},
    ]
    results = []
    for note in notes:
        if keyword.lower() in note["title"].lower() or keyword.lower() in note["content"].lower():
            results.append(f"  [{note['title']}] {note['content']}")
    if results:
        return f"找到 {len(results)} 条相关笔记：\n" + "\n".join(results)
    return f"没有找到包含 '{keyword}' 的笔记"


def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    now = datetime.now()
    return f"当前时间是 {now.strftime('%Y-%m-%d %H:%M:%S')}"


# ============================================================
# 工具描述（JSON schema，告诉模型有哪些工具可用）
# ============================================================

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前天气信息。当用户询问天气时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，例如 '东京'、'北京'、'New York'",
                    }
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "计算数学表达式。支持加减乘除和括号。当用户需要计算时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，例如 '2 + 3 * 4' 或 '(10 - 2) / 4'",
                    }
                },
                "required": ["expression"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_notes",
            "description": "搜索本地笔记库。当用户想找笔记、回忆某个知识点时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "keyword": {
                        "type": "string",
                        "description": "搜索关键词，例如 '装饰器'、'docker'",
                    }
                },
                "required": ["keyword"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取当前时间。当用户询问当前时间时使用。",
            "parameters": {
                "type": "object",
                "properties": {},
            },
        },
    },
]

# 工具名 → 函数的映射
TOOL_MAP = {
    "get_weather": get_weather,
    "calculate": calculate,
    "search_notes": search_notes,
    "get_current_time": get_current_time,
}

main.py — Agent 主循环

[!note]- 展开查看 main.py

"""
Day 2: Tool Agent — 手写工具调用

核心思想：
  Agent = LLM + tools + loop
  模型决定"要不要调用工具"，代码负责"执行工具"，然后把结果还给模型。
"""

import os
import json
import requests
from dotenv import load_dotenv
from tools import TOOLS, TOOL_MAP

load_dotenv()

api_key = os.environ.get("OPENROUTER_API_KEY")
if not api_key:
    print("Error: 在 .env 文件中设置 OPENROUTER_API_KEY")
    exit(1)

API_URL = "https://openrouter.ai/api/v1/chat/completions"
MODEL = "openai/gpt-oss-120b:free"

SYSTEM_INSTRUCTION = {
    "role": "system",
    "content": (
        "You are a helpful assistant with access to tools. "
        "Use tools when the user's question needs real-time data or calculations. "
        "Answer in the same language the user uses. "
        "Keep answers concise."
    ),
}

history = [SYSTEM_INSTRUCTION]


def call_api(messages: list[dict]) -> dict:
    """
    调用 OpenRouter API（非流式）。

    和 Day 1 不同，这次不用流式输出。
    因为工具调用时，API 返回的不是文字，而是结构化的 tool_calls 数据。
    流式处理 tool_calls 会更复杂，Day 2 先用非流式理解原理。
    """
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": MODEL,
        "messages": messages,
        "tools": TOOLS,  # 关键：告诉模型有哪些工具可用
        "stream": False,
    }
    try:
        r = requests.post(API_URL, headers=headers, json=payload, timeout=(10, 60))
        if r.status_code != 200:
            print(f"\nAPI 错误 (HTTP {r.status_code}): {r.text[:200]}")
            return {}
        return r.json()
    except requests.exceptions.RequestException as e:
        print(f"\n网络请求失败：{e}")
        return {}


def execute_tool(tool_name: str, arguments: dict) -> str:
    """
    执行一个工具函数。

    参数:
        tool_name: 工具名（如 "get_weather"）
        arguments: 参数字典（如 {"city": "东京"}）
    """
    func = TOOL_MAP.get(tool_name)
    if not func:
        return f"错误：未知工具 '{tool_name}'"
    try:
        result = func(**arguments)  # **arguments 把字典展开为关键字参数
        return result
    except Exception as e:
        return f"工具执行错误：{e}"


def agent_loop(user_input: str) -> str:
    """
    Agent 的核心循环。

    流程：
    1. 把用户消息加入历史
    2. 调用 API
    3. 检查返回：
       - 有 content → 直接返回（不需要工具）
       - 有 tool_calls → 执行工具 → 把结果加入历史 → 回到第 2
    4. 重复直到模型给出最终文字回答
    """
    history.append({"role": "user", "content": user_input})

    while True:
        response = call_api(history)
        if not response or "choices" not in response:
            return "API 返回异常，请重试。"

        message = response["choices"][0]["message"]

        # 情况 1：模型直接回答（没有调用工具）
        if message.get("content") and not message.get("tool_calls"):
            reply = message["content"]
            history.append({"role": "assistant", "content": reply})
            return reply

        # 情况 2：模型想调用工具
        if message.get("tool_calls"):
            history.append(message)  # 保存模型的 tool_calls 请求

            for tool_call in message["tool_calls"]:
                func_name = tool_call["function"]["name"]
                func_args = json.loads(tool_call["function"]["arguments"])
                tool_call_id = tool_call["id"]

                print(f"  🔧 调用工具: {func_name}({func_args})")
                result = execute_tool(func_name, func_args)
                print(f"  📋 工具结果: {result}")

                # 把工具结果加入历史
                history.append({
                    "role": "tool",
                    "tool_call_id": tool_call_id,
                    "content": result,
                })

            continue  # 工具结果已加入历史，继续循环让模型生成回答

        # 情况 3：异常
        return "模型返回了未知格式的响应。"


# --- 主循环 ---
print(f"Tool Agent (model: {MODEL})")
print("我有 3 个工具：查天气、算数学、搜笔记")
print("输入 'quit' 退出 | 'history' 查看历史 | 'clear' 清空")
print("-" * 60)

while True:
    user_input = input("\nYou: ").strip()

    if user_input.lower() in ("quit", "exit", "q"):
        print("Goodbye!")
        break
    if user_input.lower() == "history":
        for msg in history:
            role = msg.get("role", "?")
            if role == "system":
                continue
            if role == "tool":
                print(f"  [tool] {msg['content'][:80]}")
            elif msg.get("tool_calls"):
                for tc in msg["tool_calls"]:
                    print(f"  [call] {tc['function']['name']}({tc['function']['arguments']})")
            elif msg.get("content"):
                label = "You" if role == "user" else "Agent"
                print(f"  [{label}] {msg['content'][:80]}")
        continue
    if user_input.lower() == "clear":
        history = [SYSTEM_INSTRUCTION]
        print("历史已清空。")
        continue
    if not user_input:
        continue

    reply = agent_loop(user_input)
    print(f"\nAgent: {reply}")

关键流程解析

1. API 请求多了 `tools` 参数

payload = {
    "model": MODEL,
    "messages": messages,
    "tools": TOOLS,      # ← Day 1 没有这行
    "stream": False,     # ← Day 1 用的 True
}

tools 把工具的 JSON Schema 传给模型，模型据此决定是否调用。

2. 模型返回 `tool_calls` 而非文字

# 模型返回的消息可能长这样：
{
    "role": "assistant",
    "content": None,
    "tool_calls": [
        {
            "id": "call_abc123",
            "type": "function",
            "function": {
                "name": "get_weather",
                "arguments": "{\"city\": \"东京\"}"
            }
        }
    ]
}

注意 arguments 是 JSON 字符串，需要 json.loads() 解析。

3. 工具结果用 `role: "tool"` 回传

{
    "role": "tool",
    "tool_call_id": "call_abc123",   # 对应上面的 id
    "content": "东京的天气：25°C，晴天，湿度 60%"
}

tool_call_id 告诉模型”这是哪个工具调用的结果”。

4. Agent Loop

用户输入 → API → 有 tool_calls? → 执行工具 → 结果加入历史 → 再调 API
                                    ↓ 没有
                                 返回文字回答

这就是 Agent 的 “loop” — 不是调一次 API 就结束，而是可能来回多轮。

运行效果

You: 东京的天气怎么样
  🔧 调用工具: get_weather({'city': '东京'})
  📋 工具结果: 东京的天气：25°C，晴天，湿度 60%

Agent: 东京目前气温约 25 °C，天气晴朗，湿度约 60%。

You: 现在几点了
  🔧 调用工具: get_current_time({'': {}})       ← ⚠️ 第一次：传了奇怪的空参数
  📋 工具结果: 工具执行错误                        ← 失败了
  🔧 调用工具: get_current_time({})              ← 第二次：传了空字典
  📋 工具结果: 当前时间是 2026-06-06 20:01:32    ← 成功了

Agent: 当前时间是 2026 年 6 月 6 日 20:01。

踩坑记录：无参数工具的 `parameters` 描述

上面 get_current_time 被调用了两次才成功。原因是工具描述写得不够好。

错误写法

{
    "type": "function",
    "function": {
        "name": "get_current_time",
        "description": "获取当前时间"
        // ← 没有 parameters 字段
    }
}

模型看到没有 parameters，不确定该怎么传参，就自己瞎猜了一个 {'': {}}。

正确写法

{
    "type": "function",
    "function": {
        "name": "get_current_time",
        "description": "获取当前日期和时间。当用户问现在几点、今天几号时使用。",
        "parameters": {
            "type": "object",
            "properties": {}    // 空的，但必须有
        }
    }
}

教训

parameters 字段不能省。 即使工具不需要参数，也要写完整的 schema：

"parameters": {
    "type": "object",
    "properties": {},
},

模型靠这个 schema 理解”怎么调用你的工具”。描述不完整 → 模型猜测 → 调用失败 → 浪费一轮 API 请求。

常见问题

模型不调用工具，直接编造答案：检查工具描述是否清晰，description 要明确说”当用户…时使用”
tool_call_id 不匹配：回传工具结果时，tool_call_id 必须和模型返回的一致
arguments 解析失败：模型有时返回不合法的 JSON，需要 try/except 兜底
无限循环：如果模型反复调用工具不给最终回答，需要加最大轮次限制
无参数工具传了奇怪参数：parameters 字段必须写，即使没有参数也要写 "properties": {}，否则模型会瞎猜

Python Tool Calling Agent

目标

和 Day 1 的区别

前置条件

核心理解

Tool Calling 的工作原理

Tool 描述的结构

完整代码

tools.py — 工具定义

main.py — Agent 主循环

关键流程解析

1. API 请求多了 tools 参数

2. 模型返回 tool_calls 而非文字

3. 工具结果用 role: "tool" 回传

4. Agent Loop

运行效果

踩坑记录：无参数工具的 parameters 描述

错误写法

正确写法

教训

常见问题

1. API 请求多了 `tools` 参数

2. 模型返回 `tool_calls` 而非文字

3. 工具结果用 `role: "tool"` 回传

踩坑记录：无参数工具的 `parameters` 描述