智能体工具调用实战:从零到一完整教程

2026-06-20阅读 0热度 0
智能体

前几篇文章搭建了一个完整的Agent基础框架(第01篇),加装了“操作系统”——重试机制、超时控制、步数限制(第02篇),又帮它省下80%的Token消耗(第03篇)。

智能体开发实战04|工具调用从零到一

但你有没有发现一个致命缺口——

这个Agent直到现在还是个“光说不练”的空壳:能陪你聊天,却一件事都执行不了。

  • 查实时天气?“抱歉,我没有联网能力。”
  • 写个文件?“建议您手动操作。”
  • 查询数据库?“我没有数据库访问权限。”

这就是缺失工具调用的Agent——只有大脑,没有四肢。

今天,我们来给它装上双手。

一、工具调用是什么

工具调用(Tool Calling),在OpenAI生态中叫Function Calling,在Anthropic体系里叫Tool Use。叫法不同,本质一致:让大语言模型能够激活外部函数和API。

没有工具调用的模型:

用户:北京今天的天气怎么样?
模型:抱歉,我的训练数据截止到2025年,
      无法获取实时天气信息。建议您打开天气App查询。

有工具调用的模型:

用户:北京今天的天气怎么样?
模型:[调用函数 get_weather(city="北京") → 返回 "晴 22-28°C"]
模型:北京今天晴,气温22到28度,
      适合外出,紫外线中等,建议防晒。

区别不在模型“知道”多少——而在于它能否“行动”。工具调用赋予模型一个关键能力:在生成回复之前,它可以选择调用某个函数,拿到实际结果后再组织语言。

1.1 工作流程

完整的工具调用流程分为三个步骤:

  1. 注册函数:明确定义函数名称、参数和用途,告知模型“你有这些工具可用”
  2. 模型决策:模型解析用户意图,判断“是否需要调用工具、调用哪个、传什么参数”
  3. 执行-返回:你的代码真正执行该函数,将结果回填到上下文,模型据此生成最终回复

注意:模型不负责执行函数。它只负责“决定是否调用”和“给出参数建议”。真正执行的是你的代码——这正是Agent架构的核心:模型是决策者,代码是执行者。

二、从零实现┃Function Calling 实战

我们用OpenAI兼容的API演示。目前DeepSeek、Qwen、Kimi均已支持Function Calling,接口完全兼容。

2.1 第一步:定义你的第一个工具

工具定义是一个JSON Schema,向模型描述“这个函数的名称、作用、所需参数”:

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的实时天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称,如北京、上海、深圳"
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位",
                        "default": "celsius"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

关键字段说明:

  • name:函数标识符,模型用此名称来“调用”该函数
  • description:简短描述,告诉模型何时使用这个函数。极其重要——模型完全依赖这段描述做决策
  • parameters:参数定义(JSON Schema格式),模型按此结构生成参数
  • required:必填参数列表,未列出的均为可选

2.2 第二步:实际的函数实现

# 核心执行函数:获取天气
import requestsdef get_weather(city: str, units: str = "celsius") -> str:
    """调用天气API获取实时数据"""
    api_key = "your_api_key"
    url = f"https://api.openweathermap.org/data/2.5/weather"
    params = {
        "q": city,
        "appid": api_key,
        "units": "metric" if units == "celsius" else "imperial"
    }
    resp = requests.get(url, params=params)
    data = resp.json()
    
    temp = data["main"]["temp"]
    desc = data["weather"][0]["description"]
    humidity = data["main"]["humidity"]
    
    return f"{city}天气:{desc},温度{temp}°{'C' if units == 'celsius' else 'F'},湿度{humidity}%"

2.3 第三步:完整的调用循环

这是最核心的环节——将模型决策与函数执行串联起来:

from openai import OpenAIclient = OpenAI(
    api_key="your-api-key",
    base_url="https://api.deepseek.com/v1"
)# 工具注册表——名称到函数的映射
function_registry = {
    "get_weather": get_weather
}messages = [
    {"role": "system", "content": "你是一个智能助手,可以通过工具获取实时信息。"},
    {"role": "user", "content": "北京今天天气怎么样?适合去公园吗?"}
]# 第一轮调用
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    tools=tools,  # 传入工具定义
    tool_choice="auto"  # 让模型自动决定是否调用
)# 检查模型是否想要调用工具
tool_calls = response.choices[0].message.tool_calls
if tool_calls:
    # 模型决定调用了!
    for tool_call in tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)
        
        # 从注册表找到函数并执行
        func = function_registry[func_name]
        result = func(**func_args)
        
        # 将结果追加到消息列表
        messages.append(response.choices[0].message)
        messages.append({
            "tool_call_id": tool_call.id,
            "role": "tool",
            "name": func_name,
            "content": str(result)
        })
    
    # 带着工具结果再调一次模型,生成最终回复
    final_response = client.chat.completions.create(
        model="deepseek-chat",
        messages=messages
    )
    print(final_response.choices[0].message.content)
else:
    # 模型未调用工具,直接输出
    print(response.choices[0].message.content)

这个循环模式(模型的回复 → 检查工具调用 → 执行 → 返回结果 → 再次调用模型)是所有Agent框架的底层逻辑——OpenAI SDK、LangChain、AutoGen、Claude Tool Use均沿用这一范式。

三、高级模式┃多工具协作

一个工具不够用?那就给它一组工具。多工具协作才是Agent真正的能力所在。

3.1 多工具注册

tools = [    {        "type": "function",        "function": {            "name": "web_search",            "description": "搜索互联网获取最新信息",            "parameters": {                "type": "object",                "properties": {                    "query": {"type": "string", "description": "搜索关键词"}                },                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_webpage",
            "description": "读取指定URL的页面内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "网页URL"}
                },
                "required": ["url"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "sa ve_to_file",
            "description": "将内容保存到本地文件",
            "parameters": {
                "type": "object",
                "properties": {
                    "filename": {"type": "string", "description": "文件名"},
                    "content": {"type": "string", "description": "文件内容"}
                },
                "required": ["filename", "content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "run_code",
            "description": "执行Python代码并返回结果",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {"type": "string", "description": "Python代码"}
                },
                "required": ["code"]
            }
        }
    }
]# 注册表也相应扩展
function_registry = {
    "web_search": web_search_func,
    "read_webpage": read_webpage_func,
    "sa ve_to_file": sa ve_to_file_func,
    "run_code": run_code_func
}

有了这组工具,你的Agent就能完成这类任务:

场景示例:一个科研助手

用户:帮我查一下最新的Agent框架对比,整理一下保存到文件。第1步:模型调用 web_search(query="2025年 AI Agent框架对比")
第2步:查看搜索结果,发现几个关键文章
第3步:调用 read_webpage(url=...) 逐篇阅读
第4步:调用 sa ve_to_file(filename="agent_frameworks_对比.md", content=...)最终回复:已为您整理好5个主流Agent框架的对比,
          已保存到 agent_frameworks_对比.md

注意:模型并非一次性调用所有工具,而是一步一步推演——每步调用一个工具,观察结果后再决定下一步。这就是Agent的“思维链”。

3.2 并行工具调用

很多场景下,多个工具调用之间没有依赖关系,可以同时执行。DeepSeek和OpenAI都支持并行工具调用:

# 模型可能在同一回复中请求多个工具调用
# 例如:对比三家公司的财报
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    tools=tools
)tool_calls = response.choices[0].message.tool_calls
# tool_calls 可能是数组,每个独立import asyncioasync def execute_tool_calls(tool_calls, registry):
    """并行执行无依赖的工具调用"""
    async def execute_one(tc):
        func = registry[tc.function.name]
        args = json.loads(tc.function.arguments)
        result = await func(**args)
        return {
            "tool_call_id": tc.id,
            "role": "tool",
            "name": tc.function.name,
            "content": str(result)
        }
    
    # 全部并行执行
    results = await asyncio.gather(*[
        execute_one(tc) for tc in tool_calls
    ])
    return results

并行调用能大幅缩短多步骤任务的执行时间。例如查询三个城市的天气,串行需6秒,并行仅需2秒。

四、实战┃在 Agent 框架中集成工具系统

前几篇我们一直在搭建Agent框架,现在是时候把工具系统集成进去了。先回顾一下已有的Agent结构:

class Agent:
    def __init__(self, model="deepseek-chat"):
        self.model = model
        self.tools = []       # 工具定义列表
        self.registry = {}    # 函数注册表
        self.history = []     # 对话历史
        self.max_steps = 10   # 最大执行步数
    
    def register_tool(self, name: str, fn, schema: dict):
        """注册一个工具"""
        schema["name"] = name
        self.tools.append({
            "type": "function",
            "function": schema
        })
        self.registry[name] = fn
    
    async def run(self, user_input: str):
        messages = [
            {"role": "system", "content": self.system_prompt},
            *self.history[-10:],  # 仅保留最近10轮
            {"role": "user", "content": user_input}
        ]
        
        for step in range(self.max_steps):
            response = await self._call_llm(messages)
            msg = response.choices[0].message
            
            if not msg.tool_calls:
                # 模型没调用工具,直接输出
                self.history.append({"role": "user", "content": user_input})
                self.history.append({"role": "assistant", "content": msg.content})
                return msg.content
            
            # 模型调用了工具
            messages.append(msg)
            for tc in msg.tool_calls:
                fn = self.registry.get(tc.function.name)
                if not fn:
                    result = f"错误:未注册的函数 {tc.function.name}"
                else:
                    try:
                        args = json.loads(tc.function.arguments)
                        result = await fn(**args)
                    except Exception as e:
                        result = f"执行错误:{str(e)}"
                
                messages.append({
                    "tool_call_id": tc.id,
                    "role": "tool",
                    "name": tc.function.name,
                    "content": str(result)[:5000]  # 限制结果长度
                })
        
        return "已达到最大执行步数,任务未完成。"

关键设计要点:

  • 步数限制:防止工具调用死循环(例如搜索→看到结果→再搜索→无尽循环)
  • 结果长度限制:工具返回内容可能过大,在超出上下文窗口前果断截断
  • 异常处理:工具可能抛出异常,不能让整个Agent崩溃
  • 历史管理:每轮对话后保留历史,但只保留最近N轮(参照第03篇的上下文压缩)

4.1 实用工具库

下面这几个工具是Agent的“标配”,几乎每个Agent都用得上:

tools_library = {
    # 1. 网络搜索
    "web_search": {
        "fn": lambda q: requests.get(
            f"https://api.duckduckgo.com/?q={q}&format=json"
        ).json(),
        "schema": {
            "description": "在互联网上搜索信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"]
            }
        }
    },
    
    # 2. 网页读取
    "read_url": {
        "fn": lambda url: requests.get(url).text[:10000],
        "schema": {
            "description": "读取指定URL的文本内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {"type": "string"}
                },
                "required": ["url"]
            }
        }
    },
    
    # 3. 文件操作
    "read_file": {
        "fn": lambda path: open(path, "r").read(),
        "schema": {
            "description": "读取本地文件内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"}
                },
                "required": ["path"]
            }
        }
    },
    "write_file": {
        "fn": lambda path, content: (
            open(path, "w").write(content), f"已保存到 {path}"
        )[1],
        "schema": {
            "description": "将内容写入本地文件",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "content": {"type": "string"}
                },
                "required": ["path", "content"]
            }
        }
    },
    
    # 4. 代码执行
    "python_exec": {
        "fn": exec_sandboxed_python,  # 安全沙箱
        "schema": {
            "description": "在安全的沙箱中执行Python代码",
            "parameters": {
                "type": "object",
                "properties": {
                    "code": {"type": "string"}
                },
                "required": ["code"]
            }
        }
    }
}

安全提醒:代码执行工具(python_exec)务必跑在沙箱内,切勿在宿主环境直接exec模型生成的代码。推荐使用Docker容器、pyodide沙箱或子进程加权限限制。

五、踩坑指南┃工具调用的 10 个坑

实战中,工具调用的陷阱远比你想象的多。以下是我亲身踩过的坑:

5.1 一个经典的死循环处理

# 检测重复调用——防止死循环
class ToolCallTracker:
    def __init__(self, max_repeats=3):
        self.call_history = []
        self.max_repeats = max_repeats
    
    def track(self, name: str, args: dict) -> bool:
        """跟踪工具调用,检测是否陷入死循环"""
        self.call_history.append((name, str(args)))
        
        # 检查最近N次是否都是同一个工具调用
        recent = self.call_history[-self.max_repeats:]
        if len(recent) >= self.max_repeats:
            # 检查是否全是同一个调用
            first = recent[0]
            if all(c == first for c in recent):
                return False  # 死循环,需要干预
        return True
    
    def force_stop(self, messages, step_info):
        """强制打断并注入上下文"""
        messages.append({
            "role": "system",
            "content": f"你已连续重复调用工具 {self.call_history[-1][0]} {self.max_repeats} 次。"
                        f"请分析已获得的信息,直接回答用户问题,不要继续调用工具。"
        })
        return messages

六、对比┃各家工具调用的差异

不同厂商的工具调用接口存在细微差别,但核心模式一致。以下总结关键差异:

建议做法:统一使用OpenAI兼容接口格式,这样切换模型时只需修改base_url和api_key,无需改动工具定义。

七、最佳实践总结

  1. 工具描述要精确:“搜索互联网获取最新信息”远优于“帮助用户搜索信息”
  2. 参数尽量少:最小化原则,能不传的参数就不定义
  3. 设置步数限制:始终添加max_steps,Agent不是永动机
  4. 避免返回过大的结果:工具返回数据要精炼,而非全量dump
  5. 加检测机制:监控循环调用、幻觉函数名、参数解析失败
  6. 用标准接口:OpenAI兼容格式,便于模型切换
  7. 函数体要健壮:try/except兜底,不让Agent因工具异常而崩溃
  8. 安全第一:代码执行强制沙箱化,文件操作限定路径范围

八、下期预告

  • 第05篇: Agent 记忆系统——从“转身就忘”到“过目不忘”
  • 第06篇: Multi-Agent 模式——把 Agent 变成团队
  • 第07篇: Skills 工程——如何给 Agent 批量安装技能

没有工具的Agent,就像一个超级聪明却被捆住手脚的人——他能理解你的每个字,知道该怎么做,但就是做不出来。而一旦装上工具,你的Agent才能真正“动手干活”。

很多人在搭建Agent时,把大量精力花在写Prompt上,却忽视了工具定义的质量。实际上,工具定义的精度直接决定Agent的能力上限。工具是Agent的接口——接口质量 = 能力上限。

最后提醒一句:工具不是越多越好。5个精心设计的工具,远强过50个随意堆砌的。

下一篇,我们来攻克Agent的“失忆症”——记忆系统。


(本文内容仅保留正文,已按规范清理无关信息)

免责声明

本网站新闻资讯均来自公开渠道,力求准确但不保证绝对无误,内容观点仅代表作者本人,与本站无关。若涉及侵权,请联系我们处理。本站保留对声明的修改权,最终解释权归本站所有。

相关阅读

更多
欢迎回来 登录或注册后,可保存提示词和历史记录
登录后可同步收藏、历史记录和常用模板
注册即表示同意服务条款与隐私政策