最近在项目里折腾CrewAI多智能体开发,最让我上头的不是Agent怎么编排,而是“自定义工具”这块。团队的需求很直白:让AI自动查库存、核订单、跟进物流状态。听起来简单,可CrewAI自带的那几个工具根本碰不到企业内部接口,最后还是得老老实实写自己的工具。这篇就把我在CrewAI里从零创建自定义工具的设计思路、代码实现、踩坑记录一起整理出来。适合刚把Agent跑通、却发现内置工具不够用的同学,也适合准备在业务场景里扩展智能体能力的开发者。不需要懂框架源码,只要会点Python,跟着走一遍就能自己写出第一支工具。
1. 为什么要在CrewAI里写自定义工具
1.1 智能体与工具的“手脚关系”
CrewAI的核心模型很简单:Agent是大脑,负责理解任务、拆解计划、判断下一步做什么;但大脑不会真的去调外部系统,真正动手的是Tool。模型本身不具备“查询数据库”“调用订单接口”“读本地文件”这些能力,它能做的只是“决定调用哪个工具、传什么参数、怎么解读返回结果”。
如果没有自定义工具,Agent的能力边界就非常有限。它只能靠训练时学到的知识和内置工具提供的实时信息来回答问题,一旦遇到私有系统、内部API、特定业务规则,就会开始编答案。所以自定义工具实际上是在给智能体“长手脚”,每加一个工具,就相当于给Agent增加一种可以信赖的实操能力。
在CrewAI里,一个工具本质上是一个可以被模型调用的函数包装,包含名称、描述、参数定义和执行逻辑。模型会从工具描述里判断“这个工具是干什么的”“什么情况下使用它”。所以工具写得好不好,直接决定智能体能不能正确完成任务。
1.2 内置工具解决不了什么问题
CrewAI插装包提供了一些常用工具,比如网页搜索、文件读取、网站内容抓取、RAG检索等。它们胜在通用,开箱即用,但问题也很明显:它们只面向“公开、通用、无业务规则”的场景。
拿我手里的供应链项目来说,我需要查询内部订单系统的订单状态,这个接口有内网访问限制,需要带token认证,返回的是我们自定义的JSON结构。内置的网页抓取工具根本不认识这个接口,也没法处理认证逻辑。更重要的是,很多业务操作不只是“读”,还包括“写”——比如审批、提交工单、标记异常。内置工具不会也不敢封装这些有业务逻辑和权限控制的操作。
所以自定义工具的核心价值在于:
- 封装内部系统的访问逻辑,把认证、请求、解析细节收进函数里;
- 把领域规则和校验逻辑放到可执行代码中,模型不需要自己“推理”这些规则;
- 控制返回给模型的内容格式,避免无关信息挤占上下文;
- 对写操作做权限校验和审计,让智能体的行为可控。
1.3 自定义工具应覆盖的现实场景
从实际项目看,最值得自定义成工具的场景通常有几类。
第一类是内部数据查询。比如查库存、查订单、查客户信息、查工单进度。这类接口一般都在内网,而且数据结构是公司内部定义的,模型没法凭空猜到,只能通过工具去拿。
第二类是业务计算和规则判断。比如计算运费、判断是否满足发货条件、校验订单地址格式。这些规则用代码写清楚,比让模型“看着办”靠谱得多。
第三类是写操作。比如创建工单、提交审批、发送消息。这类操作必须控制在工具层,不能允许模型即兴发挥,否则容易产生不可控的副作用。
第四类是外部系统的集成。比如调用天气接口、查询物流轨迹、获取汇率等,只要是有固定API的服务,都可以包成工具。
一句话:凡是模型不能凭常识完成的、需要实时数据或业务口径支撑的动作,都应该考虑做成自定义工具。
2. 创建前的设计:决定工具好用不好用
2.1 工具本质是给模型看的“API文档”
很多人第一次写自定义工具时,注意力全放在“功能怎么实现”上,结果功能写对了,模型就是不会调用。问题往往出在描述上。
一个工具对模型来说就是一份“API文档”:工具名叫什么、它是干什么的、参数是什么含义、返回值长什么样。模型通过这份文档来决定是否调用。如果文档写得含糊,模型要么不敢用,要么乱用。
我习惯把工具描述当成“给一个认真但不太了解业务的新人写的操作说明”。要告诉他:什么时候该用这个工具,什么时候不该用;参数应该填什么格式;返回结果里哪些信息是有用的。比如“order_status_query”的描述,我通常会写成:当用户询问订单状态、物流节点、签收情况时使用。参数order_id是订单号,格式如SO-2025-0001。工具会返回订单当前状态和物流节点,如果订单不存在返回NOT_FOUND。
这样模型一看就知道,用户问“我的单到哪了”时,应该拿order_id调用这个工具。
2.2 粒度怎么控制
自定义工具最怕两个极端:一是功能太粗,一个工具里又查库存又改价格又发消息,模型用起来完全失控;二是功能太细,查一个订单要分“查基本信息”“查物流信息”“查商品明细”三个工具,模型容易选错,任务流程也变得冗长。
我的经验是:按“业务动作的最小完整单元”来切分。也就是说,一个工具应该完整回答一类问题,而不是做一些零碎的操作。例如“查询订单详情”是一个完整动作,它应该返回模型回答“订单现在什么状态、预计什么时候送达”所需的核心信息。“更新订单地址”是另一个完整动作,它负责校验新地址、调用更新接口、返回更新结果。
粒度控制也不需要一开始就追求完美。我一般是先根据真实业务问题列一个工具清单,然后拿几个典型问题走一遍流程,发现模型频繁组合调用多个工具,再考虑是不是要合并;发现某个工具容易被误用,再考虑是不是要拆分。
2.3 描述与参数Schema的拿捏
工具设计里最容易翻车的两个点,一是description写得不够“触发”,二是args_schema定义得不够清楚。
description的写法有个小技巧:把触发条件明确写出来。不要只写“查库存”,要写“当用户询问某个SKU在当前仓库是否有货、可用库存数量是多少时,使用本工具”。触发条件越具体,模型调用准确率越高。可以在描述里加场景示例,比如“例如用户问‘SKU-10086还有多少货’,就适合调用本工具”。
参数Schema要尽量用Field把每个字段的含义讲清楚,必要时给示例值。比如:
from pydantic import BaseModel, Field from typing import Type class StockInput(BaseModel): sku_id: str = Field(..., description="商品SKU编码,例如SKU-10086") warehouse: str = Field("default", description="仓库编码,缺省为default仓")这样模型在生成参数时,可以根据描述填出正确的sku_id,而不是随便传个“苹果手机”之类的模糊值。如果字段是必填的,用...表示;如果不是必填的,给出默认值。类型也要卡紧,别用object或dict,否则模型不知道该传什么结构。
2.4 错误处理与返回值设计
工具返回值会被拼到模型上下文里。模型会基于这段内容组织回答。所以返回值设计有一个核心原则:返回“模型可以直接引用”的结论,而不是返回一团原始数据。
比如查询订单接口返回了一大段JSON,里面有创建时间、修改时间、内部备注、嵌套的商品列表、物流轨迹数组。如果直接把这段JSON扔给模型,模型也能解析,但会浪费大量token,而且容易被无关字段干扰。更聪明的做法是在工具内部提取关键信息,整理成“订单SO-2025-0001当前状态为已发货,物流公司顺丰,当前节点为运输中,预计明天18点前送达”这样的文本。
错误处理同样重要。工具执行时如果抛异常,轻则让本次调用失败,重则让整个Crew任务中断。我通常会在工具内部捕获所有异常,把它转换成人类可读的错误消息。模型看到“订单接口请求超时,请稍后重试”后,会自然地转述给用户,而不是输出一堆堆栈信息。
3. 实操:用BaseTool从零写一个自定义工具
3.1 环境准备与项目目录
先准备好环境,建议用独立的虚拟目录。
mkdir crew-tools-demo cd crew-tools-demo python -m venv .venv source .venv/bin/activate pip install crewai crewai-tools安装完成后,创建一个简单的项目结构:
crew-tools-demo/ ├── main.py ├── .env └── tools/ ├── __init__.py ├── holiday_tool.py └── order_tool.py把工具放在独立文件夹里,主要是为了复用和测试。一个工具文件只负责一个领域,主流程文件只负责Agent和Crew的编排,这样后面维护起来非常清爽。工具内部需要调用外部API时,把地址和密钥放在.env里,用环境变量读取,不要硬编码在代码中。
3.2 第一支工具:节假日计算器
先写一个简单的工具,用来熟悉BaseTool的基本结构。
# tools/holiday_tool.py from datetime import datetime from pydantic import BaseModel, Field from crewai.tools import BaseTool from typing import Type class HolidayInput(BaseModel): year: int = Field(..., description="年份,例如2025") country: str = Field("CN", description="国家代码,CN代表中国,US代表美国") class HolidayTool(BaseTool): name: str = "holiday_calculator" description: str = ( "当用户询问某个年份、某个国家的法定节假日数量," "或最近的一个节假日日期时,使用本工具。" ) args_schema: Type[BaseModel] = HolidayInput def _run(self, year: int, country: str = "CN") -> str: # 这里只是演示数据,生产环境请替换为真实节假日API holiday_map = { "CN": ["2025-01-01", "2025-01-28", "2025-04-04"], "US": ["2025-01-01", "2025-01-20"], } holidays = holiday_map.get(country, []) if not holidays: return f"没有找到{country}的节假日数据,请确认国家代码。" return ( f"{year}年{country}共返回{len(holidays)}个节假日," f"最近的一个是{holidays[0]}。" )这段代码的核心结构是:定义输入参数模型HolidayInput;继承BaseTool;设置name和description;在_run方法里实现业务逻辑。注意_run方法的参数名和类型,必须和args_schema里的字段对应。模型会按照schema生成参数,然后框架把这些参数传给_run。
3.3 第二支工具:查询内部订单系统
节假日工具只是热身,真正派得上用场的是连接内部系统的工具。这里用requests调一个订单API,并做好错误处理。
# tools/order_tool.py import os import requests from pydantic import BaseModel, Field from crewai.tools import BaseTool from typing import Type class OrderQueryInput(BaseModel): order_id: str = Field(..., description="订单号,例如SO-2025-0001") include_items: bool = Field(False, description="是否返回商品明细数量") class OrderQueryTool(BaseTool): name: str = "order_status_query" description: str = ( "当用户询问订单状态、物流节点、签收情况时,使用本工具。" "参数order_id为订单号,格式如SO-2025-0001。" "如果订单不存在,返回NOT_FOUND。" ) args_schema: Type[BaseModel] = OrderQueryInput def _run(self, order_id: str, include_items: bool = False) -> str: api_base = os.getenv("ORDER_API_BASE", "http://localhost:8000") try: resp = requests.get( f"{api_base}/api/orders/{order_id}", params={"include_items": include_items}, timeout=5, ) resp.raise_for_status() data = resp.json() except requests.exceptions.Timeout: return "订单接口请求超时,请稍后重试。" except requests.exceptions.HTTPError as e: return f"订单接口返回错误:{e}。" except Exception as e: return f"订单查询失败:{e}。" if resp.status_code == 404: return "NOT_FOUND" items_text = "" if include_items and data.get("items"): items_text = f",商品明细共{len(data['items'])}件" return ( f"订单{order_id}状态为{data.get('status')}," f"物流公司{data.get('logistics')}," f"当前节点{data.get('node')}{items_text}。" )这个工具做了几件重要的事:设置超时时间,避免接口卡死;捕获异常并返回可读信息;整理返回结果,只保留模型回答问题所需的信息。如果include_items为True,也只返回商品数量,不会把明细节全部塞进上下文。这样模型获得的是干净、可直接引用的答案。
3.4 注册到Agent并跑通Crew
工具写好后,在main.py里把它挂到Agent上。
# main.py from crewai import Agent, Task, Crew, Process from tools.order_tool import OrderQueryTool from tools.holiday_tool import HolidayTool order_tool = OrderQueryTool() holiday_tool = HolidayTool() support_agent = Agent( role="订单客服专员", goal="准确回答用户关于订单和假期的询问", backstory="你是一名细心的客服,只使用工具提供的事实回答,不编造信息。", tools=[order_tool, holiday_tool], verbose=True, ) query_task = Task( description="用户刚刚问:SO-2025-0001这个订单什么时候能送到?请先查订单状态再回答。", expected_output="给出订单当前所处节点与预计送达时间", agent=support_agent, ) crew = Crew( agents=[support_agent], tasks=[query_task], process=Process.sequential, verbose=True, ) result = crew.kickoff() print(result)执行时,Crew会先把任务交给Agent处理。模型看到任务描述里的“订单状态”,再看到可用工具里有名称和描述匹配的order_status_query,就会自动生成调用参数并执行工具。工具返回结果被放回上下文,模型再组织成最终答复。
如果你用的是其他模型服务,只需要提前配置好对应的API Key和模型名,CrewAI本身并不绑定某个厂商。这里不展开具体配置,按你平时用CrewAI的方式设置即可。
3.5 使用工具时常见配置细节
新手第一次接自定义工具,最容易在导入路径和类定义上卡住。
先说导入路径。不同版本CrewAI的BaseTool位置不完全一样,有的从crewai.tools导入,有的从crewai_tools导入。我实验过几个版本,建议直接查看你安装版本的官方文档,或者使用pip show crewai确认版本。代码层面只要导入路径统一,一般不会有大问题。
再说工具实例。一个工具类可以实例化多次,比如订单工具可以根据环境不同创建测试实例和生产实例。实例传给Agent时,要放在tools列表里。有些版本还支持在Task级别临时传工具,但我更推荐统一放在Agent上,这样Agent相关的所有任务都能复用,不会出现某个Task忘了挂工具、模型瞎编的情况。
还有一点:BaseTool类本身是Pydantic模型,所以类属性里的name和description要定义为类字段并给出值。如果名字取得太随意,比如“tool1”,模型很难理解它的用途。工具命名建议用小写字母和下划线,比如order_status_query,和Python函数命名规范保持一致。
4. 进阶:让工具更稳、更快、更省token
4.1 状态管理与线程安全
多数自定义工具是无状态的:输入参数进来,调用外部接口,返回结果。这种设计最安全,因为CrewAI可能并行执行多个任务,多个Agent也可能共享同一个工具实例。如果你在工具内部用self.xxx保存可变状态,就可能出现竞态条件。
如果确实需要统计调用次数、维护临时缓存,建议用锁来保护共享状态。比如给订单工具加一个调用计数器:
import threading class OrderQueryTool(BaseTool): def __init__(self, **kwargs): super().__init__(**kwargs) self._count = 0 self._lock = threading.Lock() def _run(self, order_id: str, include_items: bool = False) -> str: with self._lock: self._count += 1 current = self._count return "第{current}次调用,..." # 实际内容省略这段代码只是示意,实际项目中这种计数器多用于监控和限流。核心思路是:任何需要修改实例变量的地方都要考虑线程安全。能不用可变状态就不用,能用局部变量就用局部变量。
4.2 缓存与幂等设计
有些API查询逻辑比较重,同一个订单号短期内可能被模型反复查询。如果能做一层缓存,可以显著减少外部接口压力。查询类工具的缓存很好加,用内存里的字典或者Redis都可以。
from functools import lru_cache @lru_cache(maxsize=128) def _fetch_order_api(order_id: str, include_items: bool) -> dict: # 实际请求逻辑 ...但要注意,缓存会带来数据陈旧的问题。订单状态是会变化的,如果你把“运输中”的状态缓存了30秒,模型可能给用户一个已经“已签收”的旧答案。所以我通常只对“短时间内不会变化”的数据做缓存,或者给缓存设置很短的过期时间。写入类操作则要额外注意幂等性,同一个操作不能被重复提交,工具内部要做防重校验。
4.3 外部接口调用的超时与重试
自定义工具一旦接通外部API,稳定性就成了最大的问题。外部接口可能慢、可能超时、可能返回5xx错误。requests库的timeout参数一定要设,否则一个接口卡住,整个Crew任务都可能被拖死。
我一般的做法是:先设一个较短的连接超时(比如3秒),再设一个稍长的读取超时(比如5秒)。失败后可以重试,但别无限重试,通常两到三次就够了。重试之间加一点退避时间,避免把下游接口打爆。
import time for attempt in range(3): try: resp = requests.get(url, timeout=(3.05, 5)) resp.raise_for_status() break except requests.exceptions.Timeout: if attempt == 2: return "订单接口超时,请稍后重试。" time.sleep(0.5 * (attempt + 1))这种重试逻辑写起来不难,但能给整个智能体系统省下很多“看起来像傻了”的故障。模型面对超时错误时,有时候会反复调用同一个工具,试图“碰运气”,加上了重试之后,至少外部接口层面已经尽量可靠了。
4.4 输出精简与上下文控制
一个很多人会忽略的问题是:工具返回值会被拼到模型的上下文中,如果返回内容太长,会带来两个问题:一是token消耗剧增,成本变高;二是上下文窗口被无关信息塞满,模型的注意力会被稀释,反而更容易答错。
所以工具返回一定不能“有言必录”。我见过有人把整个数据库表结构返回给模型,结果模型分不清哪些字段是给用户看的,哪些是内部状态。正确的做法是:只返回“回答用户问题所需的最小信息集”。如果某个信息用户不关心,就不要返回。
如果结果是列表,比如查到了50条待处理工单,不要全部输出。可以返回“共50条,前5条为xxxx”,同时提供另一个分页查询工具,让模型在用户要求更多的时候再调下一步。这种设计既控制了上下文,又保留了扩展空间。
5. 实际运行中的问题与排查技巧
5.1 模型不会调用工具:先改描述
最常见的现象是:Agent跑完了,但完全是靠模型“脑补”回答,根本没有调用你的工具。打开verbose日志,如果看不到Tool调用记录,基本可以确定是描述没有触发模型。
先检查description是不是写得“太文绉绉”。模型不是靠语义联想来猜工具的,它是根据任务文本和工具描述的相关性来判断的。如果你在描述里只写“查询订单状态”,可能不够;应该写成“当用户询问订单状态、物流节点、何时送达、签收情况时,必须使用本工具查询,不要自行猜测”。把触发词写得越具体,模型越容易调用。我还习惯在描述里补一句“如果订单不存在,请不要编造,直接返回NOT_FOUND给用户”。
5.2 参数传错或类型不符
另一个高频问题:模型倒是调用工具了,但参数传得离谱。比如把订单号传成“那笔订单”,或者把year传成“今年”。这通常是Schema描述不够清楚导致的。
解决办法有三个层面:一是给Field加更详细的描述,注明格式和示例;二是给参数做兜底处理,在_run里做类型转换或默认值填充;三是工具内部对非法参数返回明确错误,让模型有机会重试。比如:
if not order_id.startswith("SO-"): return "订单号格式不正确,应以SO-开头,请确认后重试。"这样即模型传错了,也能得到一个可理解的反馈,而不是直接抛异常。
5.3 工具抛异常导致对话中断
工具代码里如果存在未捕获的异常,整个Crew任务经常会中断,而且日志里全是堆栈。用户体验极差。正确的做法是把异常拦截在工具内部。
前面订单工具已经演示了try-except的写法。需要注意的一点是:返回错误信息时,不要返回一堆技术细节,比如“KeyError: 'status'”。应该转译成“订单数据缺少状态字段,暂时无法获取完整信息”。模型看到这样的内容,至少能组织出一句“系统暂时查询不到该订单的完整状态”给用户。
5.4 智能体陷入循环或长时间不返回
运行过程中可能遇到Agent反复调用同一工具,比如因为工具返回了一个错误,模型不死心,又用同样的参数调了一次,形成死循环。CrewAI里可以给Agent设置max_iter,限制最大迭代次数。如果超过次数还没完成,任务会以失败或部分结果结束,总比无限循环好。
另外,工具本身的耗时也要设上限。如前所述,requests必须设timeout,重试要设次数。如果一个工具的平均耗时就超过30秒,那整个Crew的交互体验会非常差。遇到这种情况,要考虑异步处理或把长任务拆出去,而不是让Agent一直等着同一个同步接口。
5.5 调试CrewAI应用的轻量方法
调试自定义工具,我会分三步走。
第一步,脱离框架单独测工具。直接写一个脚本,实例化工具类,调用_run方法,确认返回值符合预期。这一步能过滤掉80%的逻辑问题。
第二步,用一个极小Crew做集成测试。只放一个Agent、一个Task、一个工具,任务描述是固定的真实业务问题。打开verbose=True,观察模型是否调用工具、调用参数是什么、返回结果如何被使用。
第三步,逐步增加复杂度。先把一个工具跑顺,再加第二个工具;先跑单Agent,再加多Agent协作。每次只变更一个变量,出了问题就能立刻锁定原因。
我还习惯在工具的关键位置加print或log。CrewAI的verbose输出会显示一部分日志,但工具内部的print内容更直接。上生产前再把这些调试输出删掉或改成logging级别。
最后说一点个人体会
在多个项目里改过自定义工具之后,我发现最深的坑往往不是代码,而是工具描述。刚开始我会把description写得很“像人话”,什么“获取订单运输轨迹信息”,结果模型就是不爱用。后来改成“当用户问我的订单到哪了、快递到哪了、什么时候能送到时,必须使用本工具”,调用准确率立刻上来了。另一个很深的体会是:工具返回一定要精简,别一股脑把原始数据丢给模型,模型不差信息,差的是结构清晰、可直接引用的答案。
如果手头正在搭CrewAI智能体,我建议从一个小工具跑起。先挑一个你每天都要重复查的内部接口,做成工具,挂到一个最简单Agent上跑通。跑通之后再加第二个工具、再加第二个Agent。这套节奏看着慢,但每一步都能积累可复现的配置和排查经验,后面多智能体协作起来会稳得多。