智能体开发里有一个常见误区:以为把大模型 API 接进去,就算完成了一个智能体。实际上,大模型只负责“想”,真正让智能体“做”的,是工具。没有工具的 Agent 只是聊天机器人,有了自定义工具,它才能查天气、读文件、调接口、算数据、发通知。这次我们围绕“自定义工具实操”展开,从工具函数定义、注册、接入智能体,到封装成 API 服务,把这条链路完整走一遍。
这篇文章的内容对应厦门大学林子雨老师的“AI编程与智能体开发”课程中 8.8.3 自定义工具实操部分。我们不讲空概念,直接落到代码上。读完你能得到四样东西:自定义工具的标准写法、工具注册到智能体的完整流程、一个可以跑通的多工具 Agent 示例、一套批量调用工具服务的工程化思路。
本文适合谁?正在学智能体开发、想用 LangChain 或 Qwen Agent 做落地项目、需要给自己的 Agent 接私有数据或内部接口的开发者。如果你只是了解概念,也可以先看核心能力速览,再挑实操章节看。
1. 自定义工具是什么:智能体的“可执行能力”
智能体的工作流程可以简化成一条链:理解任务 -> 拆解步骤 -> 选择工具 -> 执行工具 -> 汇总结果。大模型负责理解和拆解,真正落地执行的那一步,靠的是工具。
所谓自定义工具,就是开发者自己写的函数,再按照框架要求的格式注册给模型。大模型在推理时看到用户问题,会判断“这个问题需要调用哪个工具”,然后生成一个结构化的调用请求。框架收到请求后,执行对应的 Python 函数,把结果返回给模型,模型再基于结果组织自然语言回复。
一次完整的工具调用,包含三个关键要素:
| 要素 | 作用 | 对应代码 |
|---|---|---|
| 函数本体 | 实际执行逻辑 | def get_weather(city): ... |
| 函数描述 | 告诉模型“这个工具是干什么的,什么时候用” | """查询指定城市的实时天气""" |
| 参数定义 | 说明工具需要什么参数,参数的格式和含义 | city: str以及参数注释 |
很多新手第一次写自定义工具失败,不是因为函数有 bug,而是因为描述写得太含糊。模型是“看描述选工具”的,描述不清楚,它就不会选。这点后面会专门展开。
2. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 适用方向 | AI编程、智能体开发、Function Calling 工具接入 |
| 核心功能 | 将任意 Python 函数封装为可供模型调用的工具 |
| 框架支持 | LangChain、Qwen Agent、OpenAI Function Calling 等 |
| 硬件需求 | 纯 API 调用模式无需 GPU;本地模型模式需要按模型要求配置 |
| 显存占用 | 取决于模型:使用云端 API 几乎不占显存;本地 7B 模型通常 6G 显存起 |
| 启动方式 | 脚本启动 / Jupyter 分步执行 / FastAPI 服务化 |
| 接口能力 | 可封装为 HTTP API,支持外部系统调用 |
| 批量任务 | 支持,可串行可并发,建议加队列和重试 |
| 适合场景 | 数据查询、办公自动化、内容生产、私有接口接入 |
从材料看,自定义工具这一节最关键的实践点是:把一个普通函数变成模型能主动调用的工具。做到这一步,后续的工具调用链、多智能体协作才有基础。
3. 适用场景与使用边界
自定义工具适合解决一类问题:用户用自然语言提出需求,程序需要到外部系统拿数据或执行操作。典型场景包括:
- 天气查询、新闻检索、商品比价。
- 读取本地文件、Excel 处理、PDF 解析。
- 调用企业内部 API,如订单查询、库存查询。
- 执行数学计算、代码运行、数据库查询。
- 发送邮件、推送消息、创建日程。
不适合的场景也要说清楚。如果任务不需要外部环境,纯靠模型内部知识就能回答,就不需要上工具;如果任务的失败成本很高,比如直接操作生产数据库、转账、删除文件,就不能让模型全自动调用,必须加人工确认环节。
使用边界第一条是授权边界。工具本质上是替模型“伸手”去访问外部资源,所以凡是涉及他人数据、版权内容、人脸信息、声音信息的操作,必须确认有合法授权。第二个边界是权限边界。给模型挂的工具,应该遵循最小权限原则。测试时给的 API Key 尽量只开只读权限,不要拿生产环境的管理员 Key 去调试。第三个边界是审计边界。每次工具调用的入参和返回值都要落日志,否则出了问题无法回溯。
4. 环境准备与前置条件
自定义工具本身不挑环境,Windows、macOS、Linux 都能跑。如果你用的是 Windows,建议直接装 Anaconda 或 Miniconda,创建一个独立环境,避免和现有 Python 环境冲突。
4.1 基础环境要求
建议先确认以下条件:
- Python 3.9 或更高版本。
- pip 能够正常安装依赖包。
- 有一个可用的模型 API Key,比如 OpenAI、通义千问、DeepSeek 等,用于跑模型调用部分。
- 不需要 GPU。使用云端 API 时,工具调用链路只消耗少量 CPU 内存。
4.2 创建虚拟环境
打开终端,执行以下命令:
# 创建独立环境,python 版本按本机实际可用的 3.9+ 选择 conda create -n agent-tool python=3.10 -y conda activate agent-tool4.3 安装依赖包
本实操示例主要依赖 LangChain 生态,安装命令如下:
pip install langchain langchain-openai python-dotenv如果使用 Qwen Agent,可以再装:
pip install qwen-agent具体安装哪个框架,看你自己在学哪一条技术路线。核心思想是一致的:写函数 -> 描述函数 -> 注册给模型。
4.4 配置 API Key
在项目目录下创建.env文件,把你的 Key 填进去:
OPENAI_API_KEY=sk-你的密钥 OPENAI_BASE_URL=https://api.openai.com/v1注意,.env文件不要提交到 Git 仓库,建议在.gitignore里加上它。
5. 第一个自定义工具:从普通函数到可调用工具
我们从一个最朴素的需求开始:让模型能够查询指定城市的天气。真实天气数据需要接第三方 API,这里先用一个模拟函数演示工具定义、注册和调用的完整机制。理解了机制,换成真实 API 只是替换函数体的问题。
5.1 先写普通函数
工具的本质是函数,所以第一步先写一个正常的 Python 函数:
def get_weather(city: str) -> str: """ 查询指定城市的实时天气。 参数: city: 城市名称,例如"厦门"、"上海"、"北京"。 返回: 该城市当前的天气描述和温度。 """ # 演示环境使用模拟数据,实际项目中替换为真实天气 API weather_data = { "厦门": "多云,26 摄氏度,东南风 3 级", "上海": "小雨,22 摄氏度,东风 2 级", "北京": "晴,18 摄氏度,北风 4 级", } return weather_data.get(city, f"暂未收录 {city} 的天气数据")注意这里已经出现了工具三要素中的两个:函数本体和文档字符串描述。文档字符串里的内容,模型会读到,所以你写清楚“参数是什么、返回什么、什么时候用”,比代码本身更重要。
5.2 用 LangChain 的 @tool 装饰器注册
LangChain 提供@tool装饰器,把一个普通函数变成工具对象:
from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市的实时天气。""" weather_data = { "厦门": "多云,26 摄氏度,东南风 3 级", "上海": "小雨,22 摄氏度,东风 2 级", "北京": "晴,18 摄氏度,北风 4 级", } return weather_data.get(city, f"暂未收录 {city} 的天气数据")定义变量时不要用get_weather(),括号会把函数直接执行掉。要传函数本身:
weather_tool = get_weather这句代码去看 LangChain 源码实现,@tool做的事情就是把函数、参数 schema、描述信息打包成一个BaseTool实例。框架会自动从类型注解city: str生成参数说明。
5.3 再写一个计算器工具
一个 Agent 通常需要多个工具。我们再注册一个可以安全的数学表达式计算器:
import ast import operator @tool def safe_calculator(expression: str) -> str: """ 计算数学表达式的值。 参数: expression: 一个数学表达式字符串,例如 "(1 + 2) * 3"。 返回: 计算结果。 """ # 用 ast 解析表达式,只允许基本四则运算,避免 eval 带来的注入风险 allowed_operators = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg, } def _eval(node): if isinstance(node, ast.Constant): return node.value if isinstance(node, ast.BinOp) and type(node.op) in allowed_operators: left = _eval(node.left) right = _eval(node.right) return allowed_operators[type(node.op)](left, right) if isinstance(node, ast.UnaryOp) and type(node.op) in allowed_operators: operand = _eval(node.operand) return allowed_operators[type(node.op)](operand) raise ValueError("不支持的表达式") try: tree = ast.parse(expression, mode="eval") result = _eval(tree.body) return str(result) except Exception as e: return f"计算失败: {e}"说明一下为什么使用ast而不是eval。直接eval("__import__('os').system('rm -rf /')")这类危险表达式在真实环境里是可能被模型生成的,虽然概率低,但不能赌。ast解析配合白名单操作符,只允许数学运算,从机制上阻断任意代码执行。这个思路适用于所有用户输入参与执行的工具。
5.4 验证工具是否能被框架识别
写一个独立脚本test_tools.py,把两个工具打印出来看看:
from tools import get_weather, safe_calculator print(get_weather.name) print(get_weather.description) print(get_weather.args_schema.schema()) print("---") print(safe_calculator.name) print(safe_calculator.description) print(safe_calculator.args_schema.schema())命令行运行:
python test_tools.py预期输出里能看到 LangChain 自动生成了工具的 JSON Schema,包含参数名、类型和是否必填。只要这一步没问题,工具定义就成功了,下面进入智能体接入环节。
6. 把工具接入智能体:让模型自动决定调用
工具定义好之后,需要被模型“看见”。LangChain 里,bind_tools或create_react_agent都可以把工具列表传给模型。下面用一个最小 Agent 来测试工具调用链路。
6.1 编写最小 Agent
创建agent_demo.py:
import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate from tools import get_weather, safe_calculator load_dotenv() model = ChatOpenAI( model="gpt-4o-mini", temperature=0, api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL"), ) tools = [get_weather, safe_calculator] prompt = PromptTemplate.from_template( """你是一个能调用工具的智能体。请根据用户的问题,选择并调用合适的工具。 工具列表: {tools} 工具名称格式: {tool_names} 思考过程: {agent_scratchpad} 用户问题: {input} """ ) agent = create_react_agent(llm=model, tools=tools, prompt=prompt) executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)然后写一个测试入口,连续问三个问题:
questions = [ "厦门今天天气怎么样?", "计算 (12 + 34) * 5 的结果", "你好,介绍一下你自己", ] for q in questions: print("=" * 40) print(f"用户提问:{q}") result = executor.invoke({"input": q}) print(f"最终回答:{result['output']}")运行:
python agent_demo.py如果你的模型和 Key 配置正确,观察点有两个:
- 前两个问题会触发“工具调用”,日志中会出现
Action: get_weather或Action: safe_calculator。 - 第三个问题没有任何工具可以调用,Agent 会直接回答,这说明模型具备“不调用工具”的判断能力。
6.2 多轮对话中的工具调用
上面的 ReAct Agent 每次invoke都是独立的一次调用,不会记住上轮内容。如果你需要多轮记忆,最简单的做法是把历史记录拼进 prompt:
from langchain_core.messages import HumanMessage, AIMessage history = [] def chat(message: str): history.append(HumanMessage(content=message)) response = executor.invoke({"input": message, "chat_history": history}) history.append(AIMessage(content=response["output"])) return response["output"] print(chat("厦门天气怎么样?")) print(chat("那上海呢?"))这里第二问如果模型足够聪明,可以理解“那上海呢”指的是“上海天气怎么样”,并复用get_weather工具。真实项目中,历史记忆可以直接交给 LangGraph 等框架来管理,比手动拼接更稳定。
7. 用 FastAPI 封装接口:工具服务的工程化落地
工具接入 Agent 并且能跑通之后,下一步就是服务化。把 Agent 包成一个 HTTP 接口,其他系统就可以通过curl或requests调用你的智能体能力。这也是把自定义工具接入业务系统的最常用方式。
7.1 创建 FastAPI 服务
安装依赖:
pip install fastapi uvicorn创建api_server.py:
import os from dotenv import load_dotenv from fastapi import FastAPI from pydantic import BaseModel from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate from tools import get_weather, safe_calculator load_dotenv() app = FastAPI(title="自定义工具 Agent 服务") model = ChatOpenAI( model="gpt-4o-mini", temperature=0, api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL"), ) tools = [get_weather, safe_calculator] prompt = PromptTemplate.from_template( """你是一个能调用工具的智能体。请根据用户的问题,选择并调用合适的工具。 工具列表: {tools} 工具名称格式: {tool_names} 思考过程: {agent_scratchpad} 用户问题: {input} """ ) agent = create_react_agent(llm=model, tools=tools, prompt=prompt) executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True) class ChatRequest(BaseModel): message: str class ChatResponse(BaseModel): reply: str @app.post("/chat", response_model=ChatResponse) async def chat(req: ChatRequest): result = executor.invoke({"input": req.message}) return ChatResponse(reply=result["output"]) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)7.2 启动与测试接口
启动服务:
python api_server.py另开一个终端,用 curl 验证:
curl -X POST http://127.0.0.1:8000/chat \ -H "Content-Type: application/json" \ -d '{"message": "厦门天气怎么样?"}'返回结果类似:
{ "reply": "厦门当前天气为多云,26 摄氏度,东南风 3 级。" }再用 Python requests 调用一次:
import requests resp = requests.post( "http://127.0.0.1:8000/chat", json={"message": "计算 (12 + 34) * 5 的结果"}, timeout=60, ) print(resp.json())接口能跑通,意味着自定义工具已经变成了一种可交付的服务能力。接下来可以接给前端页面、企业微信机器人、内部系统等。
7.3 批量任务调用
当你有大量问题要交给 Agent 处理时,不能每次同步等待,要用批量任务模式。最简单的实现是串行循环:
import time import requests questions = [ "北京天气怎么样?", "上海天气怎么样?", "计算 123 * 456 的结果", "厦门天气怎么样?", ] results = [] for q in questions: start = time.time() try: resp = requests.post( "http://127.0.0.1:8000/chat", json={"message": q}, timeout=60, ) resp.raise_for_status() results.append({"question": q, "answer": resp.json()["reply"], "status": "ok"}) except Exception as e: results.append({"question": q, "answer": str(e), "status": "failed"}) elapsed = time.time() - start print(f"问题:{q} | 耗时:{elapsed:.2f}s") # 将结果持久化保存 import json with open("batch_results.json", "w", encoding="utf-8") as f: json.dump(results, f, ensure_ascii=False, indent=2)批量任务切记两点:一是记录每个任务的耗时和状态,失败的要能定位;二是控制并发不要过大,API 有速率限制,盲目开线程很可能触发限流。
8. 资源占用与性能观察
很多读者关心自定义工具跑起来到底占多少资源。这里把情况说清楚。
如果你使用的是云端模型 API,本地进程只是做“用户问题转发 -> 模型返回 -> 执行工具函数 -> 结果返回模型”这几步,CPU 占用很低,内存一般不超过 300MB,不占用 GPU 显存。核心瓶颈在网络请求延迟和模型响应时间上。
如果你使用的是本地部署模型,资源占用主要由模型和推理框架决定。比如本地跑 7B 模型,显存占用通常在 6G 到 10G 之间;跑 14B 模型,建议 12G 以上。这部分占用跟自定义工具本身无关,模型加载进显存后,工具函数只是普通 Python 调用。
需要观察的资源指标有三个:
| 观察项 | 观察方式 | 关注点 |
|---|---|---|
| 单次请求耗时 | 在接口里记录时间戳 | 工具调用越多耗时越长,因为多了一到两轮模型往返 |
| 内存占用 | top或任务管理器 | Python 进程内存是否持续增长,异常增长要查是否有资源未释放 |
| API 调用次数 | 在 Agent 日志中统计 | 一次工具调用通常会产生多轮模型请求,直接决定费用 |
工具数量对性能的影响也要注意。每多一个工具,模型的 token 消耗就会增加,因为工具描述、参数 schema 都要作为上下文传给模型。生产环境里,工具数量控制在 10 个以内,描述尽量精简。
9. 常见问题与排查方法
自定义工具看起来简单,实际写起来容易踩坑。下面是按经验整理的高频问题。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 模型从不用某个工具 | 工具描述不清晰,模型不知道何时调用 | 查看工具 description 是否明确说明适用场景 | 重写描述,给出典型的调用示例 |
| 模型调用工具但参数传错 | 参数 schema 生成错误或用户输入缺少信息 | 检查 args_schema.schema() 的字段定义 | 给参数加 description,工具内部做默认值处理 |
| 工具函数报错导致整个对话失败 | Agent 没有做异常捕获,错误直接抛出 | 看控制台 Traceback 定位报错函数 | 在工具函数内部 try/except,返回友好错误信息 |
| 依赖安装失败 | Python 版本和包版本不匹配 | pip list查看当前版本的依赖 | 使用虚拟环境,按官方文档指定版本重装 |
| API Key 报 401 错误 | .env未加载或 Key 过期 | 在代码中打印 os.getenv 检查是否读到了 | 确认.env文件位置、格式和 Key 有效性 |
| FastAPI 无法启动,端口被占用 | 8000 端口被其他进程占用 | Windows: `netstat -ano | findstr 8000;Linux:lsof -i:8000` |
| 批量任务中途卡住 | 某个工具调用超时或网络阻塞 | 在批量脚本中加超时参数和日志 | 为每个请求设置 timeout,失败重试 2 次 |
| 模型陷入工具循环,反复调用同一工具 | 工具返回结果不满足模型预期 | 查看 Agent 日志中的多次 Action 序列 | 检查工具返回内容是否清晰,必要时设置最大迭代次数 |
| 中文输出乱码 | 控制台编码问题 | 检查终端编码格式 | Windows 下运行chcp 65001切换 UTF-8 |
最值得警惕的是“工具循环”。模型可能因为工具返回值不明确,连续调用同一个工具五六次,空转消耗 token。解决方法是:第一,工具返回信息要直接、结构化;第二,Agent 配置里设置max_iterations,比如 3 到 5 次后强制停止。
10. 最佳实践与使用建议
自定义工具做完能跑,只是第一步。要投入到真实项目,还需要注意下面这些点。
10.1 工具设计原则
一个工具只做一件事。不要写一个“万能函数”,然后靠参数分支判断走不同逻辑。模型是依据描述选工具的,工具越多、职责越单一,模型选择越准确。工具描述里要写出“什么时候用、什么时候不用”。比如:
@tool def get_weather(city: str) -> str: """仅在用户明确询问天气时使用。查询股票、新闻等无关信息时不要调用本工具。"""不要小看这句补充描述,它能把很多误调用直接挡掉。
10.2 安全边界设计
自定义工具是智能体最危险的组件,因为它能执行真实操作。建议按这个标准设计:
- 所有工具函数内部必须做入参校验,不能信任模型生成的内容。
- 涉及文件读写时,路径要限制在白名单目录内,防止路径穿越。
- 涉及网络请求时,目标 URL 要校验域名,防止 SSRF。
- 涉及删除、修改等危险操作时,接口层增加人工确认参数。
- 记录每次调用的入参、返回值和耗时,日志保留至少 30 天。
10.3 开发流程建议
推荐按下面的顺序开发,不要把工具和 Agent 一次写完再调:
- 先用纯 Python 测函数本身,确认输入输出正确。
- 再注册到框架,打印 schema,确认参数能被识别。
- 然后接一个最小 Agent,测单轮工具调用。
- 测多轮对话和工具组合。
- 最后才封装 API 和批量任务。
每一步都是上一层的“最小可验证单元”,这样出了问题能快速定位到函数层、描述层、还是编排层。
10.4 合规提醒
涉及人脸、声音、版权素材、个人数据的工具调用,必须确认授权链条完整。企业内部的工具服务,不建议直接暴露到公网,至少要加接口鉴权。批量爬取外部数据再通过工具提供给模型,需要评估目标网站的版权政策和 robots 协议,不要触碰法律风险边界。
11. 总结与下一步
自定义工具实操这条线,核心就三件事:写函数、写描述、注册给模型。函数是执行能力,描述是让模型理解能力,注册是把能力暴露给模型。三者缺一不可。建议你先拿天气查询这种带参数、带返回值的小函数练手,跑通后再逐步增加文件工具、接口工具和数据库工具。
最先要验证的,不是 Agent 能不能回答复杂问题,而是模型能不能在对话中正确选中工具、传对参数、拿到结果。这三点验证通过,整个智能体的工程基础就稳了。
最容易踩的坑有两个:第一是工具描述不清晰导致模型永远不调用;第二是直接eval用户输入导致安全风险。前者影响体验,后者影响安全,遇到要优先处理。
后续可以继续扩展的方向:把工具调用链抽成公共方法,统一处理日志、限流、重试;用 LangGraph 改造 Agent,支持多节点、多分支的复杂流程;再接一个多智能体协作层,让多个 Agent 各自持有不同的自定义工具,分工完成一个大任务。建议把这篇文章的示例代码保存下来,做成你自己的工具脚手架,后面每写一个 Agent 项目都可以直接复用。