1. 为什么要在 CrewAI 里自定义 LLM从 BaseLLM 说起CrewAI 是一个多智能体编排框架Agent 负责角色扮演Task 负责目标拆解Crew 负责把两者串成流水线。它默认通过 LiteLLM 去对接各家模型服务但 LiteLLM 的适配列表再长也覆盖不了所有内部网关、私有协议或带特殊鉴权头的模型服务。这时候BaseLLM抽象基类就是官方留出的扩展口子你继承它、实现call()就能把任意一个兼容 OpenAI Chat Completions 协议的服务接进 CrewAI 的编排流程。我这次要解决的具体问题是团队内部已经用 TaoToken 统一了模型 Key 通道所有模型调用都走同一个 Base URL 和同一把 Key模型 ID 按需切换。但 CrewAI 默认的 LiteLLM 路径要么要求你按它的命名规则传provider/model要么在鉴权头上做额外适配配置起来很别扭。与其在每个 Agent 上写一堆litellm_params不如直接写一个BaseLLM子类把 Base URL、Key、Model ID 三件套固定下来让 CrewAI 的每个 Agent 都复用这个自定义 LLM 实例。适合谁看已经跑通过 CrewAI 最小示例、想让智能体走自有模型通道的开发者或者你手上有一个兼容 OpenAI 协议的模型服务想接进 CrewAI 但不想改 LiteLLM 配置。前置知识只需要 Python 基础、requests库、以及能跑通pip install crewai的环境。下面从接口约定讲起再给可复制的子类骨架、环境变量配置、一次本地任务编排验证最后把常见报错逐个拆开。先明确BaseLLM的接口约定这是写子类的地基。构造函数必须调用super().__init__(model..., temperature...)否则父类内部的状态初始化不完整后续 CrewAI 读取self.model时会拿到 None。核心抽象方法是call()签名固定为call(self, messages, toolsNone, callbacksNone, available_functionsNone)返回值必须是字符串或可被 CrewAI 消费的对象。messages可能是字符串也可能是[{role: user, content: ...}]这种多轮消息列表你的实现要同时兼容两种形态。可选方法有三个supports_function_calling()决定 CrewAI 是否把 tools 传给你supports_stop_words()决定停止词由谁处理get_context_window_size()告诉框架上下文窗口大小默认 4096。这三个方法不实现也能跑但实现准确了能避免很多隐性 bug。2. TaoToken 前置准备Base URL、Key 与 Model ID 三件套在写子类之前先把模型服务侧的三个参数拿到手这是后面配置片段能直接复制的前提。TaoToken 的 API 入口是https://taotoken.net/api注意这个地址不带任何查询参数是纯粹的 API 根路径。你需要在这个根路径后面拼上/v1/chat/completions才是完整的对话补全端点也就是https://taotoken.net/api/v1/chat/completions。这一点很关键因为BaseLLM子类里的endpoint参数要填的是完整端点不是根路径。Key 的获取走控制台登录后在 API Keys 页面创建。创建时建议按用途命名比如crewai-agent方便后续在用量面板里区分是哪个项目在消耗。Key 只在创建时完整显示一次复制后立刻存进环境变量不要硬编码进代码。Model ID 则取决于你要调用的具体模型在模型列表或文档里能看到可用的模型标识比如常见的对话模型 ID。把这三个值记下来Base URL 根路径、Key、Model ID。环境变量我习惯这样组织写进.env文件用python-dotenv加载# .env TAOTOKEN_API_KEYsk-你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODEL_ID你的模型ID然后在代码里用os.getenv读取。这样做的好处是子类实例化时不用把 Key 写死在参数里换环境只改.env。如果你用 shell 直接导出也行export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODEL_ID你的模型ID这里有个容易踩的坑TAOTOKEN_BASE_URL存的是根路径https://taotoken.net/api而子类里的endpoint需要的是完整端点。我建议在子类构造函数里做拼接把根路径和/v1/chat/completions组合起来这样环境变量里只维护根路径端点路径由代码统一处理避免两处不一致。如果你更习惯直接存完整端点那环境变量名改成TAOTOKEN_ENDPOINT也行只要子类读取的键名对得上。另外提醒一点Key 属于敏感凭证.env文件要加进.gitignore不要提交到仓库。团队协作时每个人用自己的 Key或者用统一的测试 Key 但限制额度。这些准备工作做完下面就可以动笔写BaseLLM子类了。3. 可复制的 BaseLLM 子类骨架与配置片段这一节给完整的、能直接跑的代码。先看子类骨架我把它拆成构造函数、call()主逻辑、错误处理、可选方法四块每块都标了注释说明为什么这么写。# custom_llm.py import os import json import requests from typing import Any, Dict, List, Optional, Union from crewai import BaseLLM class TaoTokenLLM(BaseLLM): 把 TaoToken 统一 Key 通道接到 CrewAI 的 BaseLLM 子类。 def __init__( self, model: str, api_key: Optional[str] None, base_url: Optional[str] None, temperature: Optional[float] 0.7, timeout: int 60, ): # 必须调用父类构造函数传入 model 和 temperature super().__init__(modelmodel, temperaturetemperature) self.api_key api_key or os.getenv(TAOTOKEN_API_KEY) if not self.api_key: raise ValueError(缺少 API Key请设置 TAOTOKEN_API_KEY 环境变量) # 根路径拼接完整端点避免环境变量里维护两处 root base_url or os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) self.endpoint root.rstrip(/) /v1/chat/completions self.timeout timeout def call( self, messages: Union[str, List[Dict[str, str]]], tools: Optional[List[dict]] None, callbacks: Optional[List[Any]] None, available_functions: Optional[Dict[str, Any]] None, ) - Union[str, Any]: # 字符串统一转成消息列表 if isinstance(messages, str): messages [{role: user, content: messages}] payload: Dict[str, Any] { model: self.model, messages: messages, temperature: self.temperature, } # 只有声明支持函数调用时才把 tools 带上 if tools and self.supports_function_calling(): payload[tools] tools # 如果模型支持停止词把 CrewAI 注入的 stop 带上 if self.supports_stop_words() and getattr(self, stop, None): payload[stop] self.stop try: resp requests.post( self.endpoint, headers{ Authorization: fBearer {self.api_key}, Content-Type: application/json, }, jsonpayload, timeoutself.timeout, ) resp.raise_for_status() except requests.Timeout: raise TimeoutError(TaoToken 请求超时检查网络或调大 timeout) except requests.RequestException as e: raise RuntimeError(fTaoToken 请求失败: {e}) try: data resp.json() message data[choices][0][message] except (KeyError, IndexError, ValueError) as e: raise ValueError(f响应结构异常: {e}, 原始响应: {resp.text[:200]}) # 处理函数调用分支 if message.get(tool_calls) and available_functions: return self._handle_function_calls( message[tool_calls], messages, tools, available_functions ) content message.get(content) if content is None: raise ValueError(模型返回 content 为空) # 如果模型不支持停止词手动截断 if not self.supports_stop_words() and getattr(self, stop, None): for sw in self.stop: if sw in content: content content.split(sw)[0] break return content def _handle_function_calls(self, tool_calls, messages, tools, available_functions): for tc in tool_calls: fn_name tc[function][name] if fn_name not in available_functions: continue fn_args json.loads(tc[function][arguments]) fn_result available_functions[fn_name](**fn_args) messages.append({role: assistant, content: None, tool_calls: [tc]}) messages.append({ role: tool, tool_call_id: tc[id], name: fn_name, content: str(fn_result), }) return self.call(messages, tools, None, available_functions) def supports_function_calling(self) - bool: return True def supports_stop_words(self) - bool: return True def get_context_window_size(self) - int: return 8192这段代码里几个设计点值得说明。构造函数里api_key和base_url都允许从参数传也允许从环境变量兜底这样测试时可以直接传参生产时走环境变量。端点拼接用rstrip(/)处理根路径末尾可能带的斜杠避免出现//v1这种双斜杠。call()里先做消息格式归一化再按能力开关决定是否带 tools 和 stop最后统一错误处理。_handle_function_calls里把 assistant 的 tool_calls 消息和 tool 结果消息按顺序追加再递归调用call()这是 OpenAI 协议下函数调用的标准消息流。如果你用 CrewAI 的配置文件方式管理 Agent可以在agents.yaml里引用这个自定义 LLM。不过更直接的方式是在 Python 代码里实例化后传给 Agent。下面给一个settings风格的配置片段把三件套集中管理# config.py import os from dotenv import load_dotenv load_dotenv() TAOTOKEN_CONFIG { model: os.getenv(TAOTOKEN_MODEL_ID), api_key: os.getenv(TAOTOKEN_API_KEY), base_url: os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), temperature: 0.7, }然后在主流程里这样用from crewai import Agent, Task, Crew from custom_llm import TaoTokenLLM from config import TAOTOKEN_CONFIG llm TaoTokenLLM(**TAOTOKEN_CONFIG) agent Agent( role资料整理助手, goal把给定主题整理成结构化要点, backstory你擅长把零散信息归纳成清晰条目。, llmllm, verboseTrue, ) task Task( description整理 CrewAI 自定义 LLM 的三个关键接口方法, expected_output三条要点每条不超过 50 字, agentagent, ) crew Crew(agents[agent], tasks[task]) result crew.kickoff() print(result.raw)到这里配置片段就齐了环境变量三件套、子类骨架、实例化与 Agent 绑定。下一节做一次真实的本地任务编排验证确认自定义 LLM 在 CrewAI 流程里能正常返回。4. 验证请求一次本地任务编排的成功结果验证分两步走先单独测call()再跑完整 Crew。单独测的好处是能把问题定位在 LLM 层还是编排层。先写一个最小测试脚本# test_call.py from custom_llm import TaoTokenLLM from config import TAOTOKEN_CONFIG llm TaoTokenLLM(**TAOTOKEN_CONFIG) # 字符串输入 out1 llm.call(用一句话说明什么是多智能体编排) print(字符串输入 -, out1) # 多轮消息输入 msgs [ {role: system, content: 你是一个简洁的助手。}, {role: user, content: 列出 CrewAI 的三个核心概念。}, ] out2 llm.call(msgs) print(多轮消息 -, out2)跑python test_call.py如果配置正确你会看到两段模型返回的文本。字符串输入会被归一化成单条 user 消息多轮消息会原样传给服务端。这一步成功说明 Base URL、Key、Model ID 三件套和端点拼接都没问题。接着跑完整 Crew。用上一节的main.py执行python main.py。CrewAI 在verboseTrue下会打印 Agent 的思考过程和最终输出。成功时你会看到类似这样的结构Agent 先输出 Thought然后调用 LLM 得到结果最后result.raw打印出三条要点。我实测下来从 kickoff 到返回通常在几秒到十几秒取决于模型响应速度和任务复杂度。验证时重点看三个信号。第一call()返回的是非空字符串不是 None 也不是异常。第二Crew 的result.raw里包含任务要求的输出格式比如三条要点。第三如果开了verbose日志里能看到 Agent 确实调用了你的TaoTokenLLM实例而不是回退到默认 LiteLLM。如果这三点都满足说明自定义 LLM 已经稳定接入 CrewAI 编排流程。再补一个带工具调用的验证确认supports_function_calling分支正常。定义一个简单函数传给 Agent 的 toolsdef get_word_count(text: str) - int: return len(text) agent Agent( role计数助手, goal统计给定文本的字数, backstory你只做字数统计。, llmllm, tools[get_word_count], verboseTrue, )如果模型支持函数调用CrewAI 会把工具描述传给call()的tools参数你的子类带上tools发请求模型返回tool_calls_handle_function_calls执行本地函数并把结果回传最终得到基于真实计数的回答。这一步能跑通说明函数调用链路完整。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节把实际会撞上的报错逐个拆开每个都给现象、原因、修法。401 Unauthorized。现象是requests抛HTTPError: 401 Client Error。原因通常是 Key 没读到或格式不对。先确认TAOTOKEN_API_KEY环境变量在当前 shell 里可见echo $TAOTOKEN_API_KEY能打印出sk-开头的值。如果用了.env确认load_dotenv()在读取环境变量之前调用。另一个常见原因是 Key 前后带了空格或换行复制时容易带上用.strip()处理一下。修法在构造函数里self.api_key (api_key or os.getenv(TAOTOKEN_API_KEY, )).strip()。local proxy failed。现象是请求发不出去报连接错误或代理相关异常。这通常和本机网络环境有关比如系统级代理配置干扰了requests。修法是显式禁用代理在requests.post里加proxies{http: None, https: None}或者设置环境变量NO_PROXY。如果你在容器里跑检查容器网络是否能直连外网。这个报错和模型服务本身无关先把网络链路打通。reading choices 报错。现象是KeyError: choices或IndexError: list index out of range。原因是响应 JSON 结构和预期不符可能是服务端返回了错误对象而不是正常补全结果。修法是在解析前先打印resp.text看原始返回确认里面有没有choices字段。常见触发场景是 Model ID 填错服务端返回{error: {...}}。把TAOTOKEN_MODEL_ID换成文档里确认可用的模型标识即可。另外resp.raise_for_status()要在解析 JSON 之前调用这样 4xx/5xx 会先抛出来不会走到解析分支。OAuth 相关报错。如果你看到OAuth字样说明请求被路由到了需要 OAuth 鉴权的路径而不是 API Key 鉴权。检查endpoint是不是拼成了别的路径比如误把控制台地址当成了 API 地址。正确的端点是https://taotoken.net/api/v1/chat/completions鉴权头是Authorization: Bearer sk-xxx。确认base_url环境变量是https://taotoken.net/api没有多余路径段。构造函数报 missing required parameters。现象是TypeError: __init__() missing ...。原因是子类构造函数没调用super().__init__(model..., temperature...)或者调用时漏了参数。修法是确保第一行就调用父类构造函数把model和temperature传进去。父类内部依赖这两个值初始化状态漏传会在后续读取self.model时炸掉。函数调用不生效。现象是模型明明支持工具但 CrewAI 没触发工具执行。检查三点supports_function_calling()是否返回 Truecall()里是否在tools存在时把payload[tools]带上响应里message.get(tool_calls)是否被正确读取。如果模型返回的字段名不是tool_calls需要按实际协议适配。响应 content 为 None。现象是ValueError: 模型返回 content 为空。这通常发生在模型只返回了tool_calls而没有文本内容时。如果你的流程不需要工具把supports_function_calling()改成 False模型就不会走工具分支。如果需要工具确保_handle_function_calls递归调用后能拿到最终文本。把这些报错对照着排查基本能覆盖接入过程中的绝大多数问题。核心思路是先确认三件套配置对再确认端点拼接对最后确认响应解析对。6. 把自定义 LLM 用进日常编排下一步怎么走子类跑通之后日常使用就是把它当成一个普通 LLM 实例传给任意 Agent。多 Agent 协作时你可以让所有 Agent 共用一个TaoTokenLLM实例也可以按角色配不同模型 ID——比如研究型 Agent 用长上下文模型执行型 Agent 用响应快的模型。共用一个实例的好处是 Key 和端点只维护一份换模型只改TAOTOKEN_MODEL_ID。如果你要把这套配置沉淀成团队规范建议把custom_llm.py和config.py放进项目公共模块.env.example里列出三个变量名但不填值新成员复制成.env填自己的 Key 即可。CrewAI 的 Agent 定义里只引用llm变量不出现任何硬编码凭证。需要长期跑编码类或 Agent 类任务的话可以了解下 Coding Plan 这类按周期计费的方案适合高频调用场景。验证模型连通性时模型对话页面能快速确认某个 Model ID 是否可用省得在代码里反复试。接入文档里有完整的端点和参数说明遇到协议细节可以直接查。API Keys 页面负责创建和轮换 Key建议按项目分 Key方便用量归因。最后留一个实用技巧在call()里加一行请求耗时日志import time后记录start time.time()和elapsed time.time() - start打印出来。多 Agent 编排时你能一眼看出是哪个 Agent 的 LLM 调用拖慢了整体流程比在 Crew 层面猜要高效得多。这个日志在排查超时问题时特别有用配合timeout参数一起调能把稳定性问题定位到具体环节。 CrewAI智能体开发：自定义 LLM 实现——把 BaseLLM 子类接到 TaoToken 统一 Key 通道
1. 为什么要在 CrewAI 里自定义 LLM从 BaseLLM 说起CrewAI 是一个多智能体编排框架Agent 负责角色扮演Task 负责目标拆解Crew 负责把两者串成流水线。它默认通过 LiteLLM 去对接各家模型服务但 LiteLLM 的适配列表再长也覆盖不了所有内部网关、私有协议或带特殊鉴权头的模型服务。这时候BaseLLM抽象基类就是官方留出的扩展口子你继承它、实现call()就能把任意一个兼容 OpenAI Chat Completions 协议的服务接进 CrewAI 的编排流程。我这次要解决的具体问题是团队内部已经用 TaoToken 统一了模型 Key 通道所有模型调用都走同一个 Base URL 和同一把 Key模型 ID 按需切换。但 CrewAI 默认的 LiteLLM 路径要么要求你按它的命名规则传provider/model要么在鉴权头上做额外适配配置起来很别扭。与其在每个 Agent 上写一堆litellm_params不如直接写一个BaseLLM子类把 Base URL、Key、Model ID 三件套固定下来让 CrewAI 的每个 Agent 都复用这个自定义 LLM 实例。适合谁看已经跑通过 CrewAI 最小示例、想让智能体走自有模型通道的开发者或者你手上有一个兼容 OpenAI 协议的模型服务想接进 CrewAI 但不想改 LiteLLM 配置。前置知识只需要 Python 基础、requests库、以及能跑通pip install crewai的环境。下面从接口约定讲起再给可复制的子类骨架、环境变量配置、一次本地任务编排验证最后把常见报错逐个拆开。先明确BaseLLM的接口约定这是写子类的地基。构造函数必须调用super().__init__(model..., temperature...)否则父类内部的状态初始化不完整后续 CrewAI 读取self.model时会拿到 None。核心抽象方法是call()签名固定为call(self, messages, toolsNone, callbacksNone, available_functionsNone)返回值必须是字符串或可被 CrewAI 消费的对象。messages可能是字符串也可能是[{role: user, content: ...}]这种多轮消息列表你的实现要同时兼容两种形态。可选方法有三个supports_function_calling()决定 CrewAI 是否把 tools 传给你supports_stop_words()决定停止词由谁处理get_context_window_size()告诉框架上下文窗口大小默认 4096。这三个方法不实现也能跑但实现准确了能避免很多隐性 bug。2. TaoToken 前置准备Base URL、Key 与 Model ID 三件套在写子类之前先把模型服务侧的三个参数拿到手这是后面配置片段能直接复制的前提。TaoToken 的 API 入口是https://taotoken.net/api注意这个地址不带任何查询参数是纯粹的 API 根路径。你需要在这个根路径后面拼上/v1/chat/completions才是完整的对话补全端点也就是https://taotoken.net/api/v1/chat/completions。这一点很关键因为BaseLLM子类里的endpoint参数要填的是完整端点不是根路径。Key 的获取走控制台登录后在 API Keys 页面创建。创建时建议按用途命名比如crewai-agent方便后续在用量面板里区分是哪个项目在消耗。Key 只在创建时完整显示一次复制后立刻存进环境变量不要硬编码进代码。Model ID 则取决于你要调用的具体模型在模型列表或文档里能看到可用的模型标识比如常见的对话模型 ID。把这三个值记下来Base URL 根路径、Key、Model ID。环境变量我习惯这样组织写进.env文件用python-dotenv加载# .env TAOTOKEN_API_KEYsk-你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODEL_ID你的模型ID然后在代码里用os.getenv读取。这样做的好处是子类实例化时不用把 Key 写死在参数里换环境只改.env。如果你用 shell 直接导出也行export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODEL_ID你的模型ID这里有个容易踩的坑TAOTOKEN_BASE_URL存的是根路径https://taotoken.net/api而子类里的endpoint需要的是完整端点。我建议在子类构造函数里做拼接把根路径和/v1/chat/completions组合起来这样环境变量里只维护根路径端点路径由代码统一处理避免两处不一致。如果你更习惯直接存完整端点那环境变量名改成TAOTOKEN_ENDPOINT也行只要子类读取的键名对得上。另外提醒一点Key 属于敏感凭证.env文件要加进.gitignore不要提交到仓库。团队协作时每个人用自己的 Key或者用统一的测试 Key 但限制额度。这些准备工作做完下面就可以动笔写BaseLLM子类了。3. 可复制的 BaseLLM 子类骨架与配置片段这一节给完整的、能直接跑的代码。先看子类骨架我把它拆成构造函数、call()主逻辑、错误处理、可选方法四块每块都标了注释说明为什么这么写。# custom_llm.py import os import json import requests from typing import Any, Dict, List, Optional, Union from crewai import BaseLLM class TaoTokenLLM(BaseLLM): 把 TaoToken 统一 Key 通道接到 CrewAI 的 BaseLLM 子类。 def __init__( self, model: str, api_key: Optional[str] None, base_url: Optional[str] None, temperature: Optional[float] 0.7, timeout: int 60, ): # 必须调用父类构造函数传入 model 和 temperature super().__init__(modelmodel, temperaturetemperature) self.api_key api_key or os.getenv(TAOTOKEN_API_KEY) if not self.api_key: raise ValueError(缺少 API Key请设置 TAOTOKEN_API_KEY 环境变量) # 根路径拼接完整端点避免环境变量里维护两处 root base_url or os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) self.endpoint root.rstrip(/) /v1/chat/completions self.timeout timeout def call( self, messages: Union[str, List[Dict[str, str]]], tools: Optional[List[dict]] None, callbacks: Optional[List[Any]] None, available_functions: Optional[Dict[str, Any]] None, ) - Union[str, Any]: # 字符串统一转成消息列表 if isinstance(messages, str): messages [{role: user, content: messages}] payload: Dict[str, Any] { model: self.model, messages: messages, temperature: self.temperature, } # 只有声明支持函数调用时才把 tools 带上 if tools and self.supports_function_calling(): payload[tools] tools # 如果模型支持停止词把 CrewAI 注入的 stop 带上 if self.supports_stop_words() and getattr(self, stop, None): payload[stop] self.stop try: resp requests.post( self.endpoint, headers{ Authorization: fBearer {self.api_key}, Content-Type: application/json, }, jsonpayload, timeoutself.timeout, ) resp.raise_for_status() except requests.Timeout: raise TimeoutError(TaoToken 请求超时检查网络或调大 timeout) except requests.RequestException as e: raise RuntimeError(fTaoToken 请求失败: {e}) try: data resp.json() message data[choices][0][message] except (KeyError, IndexError, ValueError) as e: raise ValueError(f响应结构异常: {e}, 原始响应: {resp.text[:200]}) # 处理函数调用分支 if message.get(tool_calls) and available_functions: return self._handle_function_calls( message[tool_calls], messages, tools, available_functions ) content message.get(content) if content is None: raise ValueError(模型返回 content 为空) # 如果模型不支持停止词手动截断 if not self.supports_stop_words() and getattr(self, stop, None): for sw in self.stop: if sw in content: content content.split(sw)[0] break return content def _handle_function_calls(self, tool_calls, messages, tools, available_functions): for tc in tool_calls: fn_name tc[function][name] if fn_name not in available_functions: continue fn_args json.loads(tc[function][arguments]) fn_result available_functions[fn_name](**fn_args) messages.append({role: assistant, content: None, tool_calls: [tc]}) messages.append({ role: tool, tool_call_id: tc[id], name: fn_name, content: str(fn_result), }) return self.call(messages, tools, None, available_functions) def supports_function_calling(self) - bool: return True def supports_stop_words(self) - bool: return True def get_context_window_size(self) - int: return 8192这段代码里几个设计点值得说明。构造函数里api_key和base_url都允许从参数传也允许从环境变量兜底这样测试时可以直接传参生产时走环境变量。端点拼接用rstrip(/)处理根路径末尾可能带的斜杠避免出现//v1这种双斜杠。call()里先做消息格式归一化再按能力开关决定是否带 tools 和 stop最后统一错误处理。_handle_function_calls里把 assistant 的 tool_calls 消息和 tool 结果消息按顺序追加再递归调用call()这是 OpenAI 协议下函数调用的标准消息流。如果你用 CrewAI 的配置文件方式管理 Agent可以在agents.yaml里引用这个自定义 LLM。不过更直接的方式是在 Python 代码里实例化后传给 Agent。下面给一个settings风格的配置片段把三件套集中管理# config.py import os from dotenv import load_dotenv load_dotenv() TAOTOKEN_CONFIG { model: os.getenv(TAOTOKEN_MODEL_ID), api_key: os.getenv(TAOTOKEN_API_KEY), base_url: os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), temperature: 0.7, }然后在主流程里这样用from crewai import Agent, Task, Crew from custom_llm import TaoTokenLLM from config import TAOTOKEN_CONFIG llm TaoTokenLLM(**TAOTOKEN_CONFIG) agent Agent( role资料整理助手, goal把给定主题整理成结构化要点, backstory你擅长把零散信息归纳成清晰条目。, llmllm, verboseTrue, ) task Task( description整理 CrewAI 自定义 LLM 的三个关键接口方法, expected_output三条要点每条不超过 50 字, agentagent, ) crew Crew(agents[agent], tasks[task]) result crew.kickoff() print(result.raw)到这里配置片段就齐了环境变量三件套、子类骨架、实例化与 Agent 绑定。下一节做一次真实的本地任务编排验证确认自定义 LLM 在 CrewAI 流程里能正常返回。4. 验证请求一次本地任务编排的成功结果验证分两步走先单独测call()再跑完整 Crew。单独测的好处是能把问题定位在 LLM 层还是编排层。先写一个最小测试脚本# test_call.py from custom_llm import TaoTokenLLM from config import TAOTOKEN_CONFIG llm TaoTokenLLM(**TAOTOKEN_CONFIG) # 字符串输入 out1 llm.call(用一句话说明什么是多智能体编排) print(字符串输入 -, out1) # 多轮消息输入 msgs [ {role: system, content: 你是一个简洁的助手。}, {role: user, content: 列出 CrewAI 的三个核心概念。}, ] out2 llm.call(msgs) print(多轮消息 -, out2)跑python test_call.py如果配置正确你会看到两段模型返回的文本。字符串输入会被归一化成单条 user 消息多轮消息会原样传给服务端。这一步成功说明 Base URL、Key、Model ID 三件套和端点拼接都没问题。接着跑完整 Crew。用上一节的main.py执行python main.py。CrewAI 在verboseTrue下会打印 Agent 的思考过程和最终输出。成功时你会看到类似这样的结构Agent 先输出 Thought然后调用 LLM 得到结果最后result.raw打印出三条要点。我实测下来从 kickoff 到返回通常在几秒到十几秒取决于模型响应速度和任务复杂度。验证时重点看三个信号。第一call()返回的是非空字符串不是 None 也不是异常。第二Crew 的result.raw里包含任务要求的输出格式比如三条要点。第三如果开了verbose日志里能看到 Agent 确实调用了你的TaoTokenLLM实例而不是回退到默认 LiteLLM。如果这三点都满足说明自定义 LLM 已经稳定接入 CrewAI 编排流程。再补一个带工具调用的验证确认supports_function_calling分支正常。定义一个简单函数传给 Agent 的 toolsdef get_word_count(text: str) - int: return len(text) agent Agent( role计数助手, goal统计给定文本的字数, backstory你只做字数统计。, llmllm, tools[get_word_count], verboseTrue, )如果模型支持函数调用CrewAI 会把工具描述传给call()的tools参数你的子类带上tools发请求模型返回tool_calls_handle_function_calls执行本地函数并把结果回传最终得到基于真实计数的回答。这一步能跑通说明函数调用链路完整。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节把实际会撞上的报错逐个拆开每个都给现象、原因、修法。401 Unauthorized。现象是requests抛HTTPError: 401 Client Error。原因通常是 Key 没读到或格式不对。先确认TAOTOKEN_API_KEY环境变量在当前 shell 里可见echo $TAOTOKEN_API_KEY能打印出sk-开头的值。如果用了.env确认load_dotenv()在读取环境变量之前调用。另一个常见原因是 Key 前后带了空格或换行复制时容易带上用.strip()处理一下。修法在构造函数里self.api_key (api_key or os.getenv(TAOTOKEN_API_KEY, )).strip()。local proxy failed。现象是请求发不出去报连接错误或代理相关异常。这通常和本机网络环境有关比如系统级代理配置干扰了requests。修法是显式禁用代理在requests.post里加proxies{http: None, https: None}或者设置环境变量NO_PROXY。如果你在容器里跑检查容器网络是否能直连外网。这个报错和模型服务本身无关先把网络链路打通。reading choices 报错。现象是KeyError: choices或IndexError: list index out of range。原因是响应 JSON 结构和预期不符可能是服务端返回了错误对象而不是正常补全结果。修法是在解析前先打印resp.text看原始返回确认里面有没有choices字段。常见触发场景是 Model ID 填错服务端返回{error: {...}}。把TAOTOKEN_MODEL_ID换成文档里确认可用的模型标识即可。另外resp.raise_for_status()要在解析 JSON 之前调用这样 4xx/5xx 会先抛出来不会走到解析分支。OAuth 相关报错。如果你看到OAuth字样说明请求被路由到了需要 OAuth 鉴权的路径而不是 API Key 鉴权。检查endpoint是不是拼成了别的路径比如误把控制台地址当成了 API 地址。正确的端点是https://taotoken.net/api/v1/chat/completions鉴权头是Authorization: Bearer sk-xxx。确认base_url环境变量是https://taotoken.net/api没有多余路径段。构造函数报 missing required parameters。现象是TypeError: __init__() missing ...。原因是子类构造函数没调用super().__init__(model..., temperature...)或者调用时漏了参数。修法是确保第一行就调用父类构造函数把model和temperature传进去。父类内部依赖这两个值初始化状态漏传会在后续读取self.model时炸掉。函数调用不生效。现象是模型明明支持工具但 CrewAI 没触发工具执行。检查三点supports_function_calling()是否返回 Truecall()里是否在tools存在时把payload[tools]带上响应里message.get(tool_calls)是否被正确读取。如果模型返回的字段名不是tool_calls需要按实际协议适配。响应 content 为 None。现象是ValueError: 模型返回 content 为空。这通常发生在模型只返回了tool_calls而没有文本内容时。如果你的流程不需要工具把supports_function_calling()改成 False模型就不会走工具分支。如果需要工具确保_handle_function_calls递归调用后能拿到最终文本。把这些报错对照着排查基本能覆盖接入过程中的绝大多数问题。核心思路是先确认三件套配置对再确认端点拼接对最后确认响应解析对。6. 把自定义 LLM 用进日常编排下一步怎么走子类跑通之后日常使用就是把它当成一个普通 LLM 实例传给任意 Agent。多 Agent 协作时你可以让所有 Agent 共用一个TaoTokenLLM实例也可以按角色配不同模型 ID——比如研究型 Agent 用长上下文模型执行型 Agent 用响应快的模型。共用一个实例的好处是 Key 和端点只维护一份换模型只改TAOTOKEN_MODEL_ID。如果你要把这套配置沉淀成团队规范建议把custom_llm.py和config.py放进项目公共模块.env.example里列出三个变量名但不填值新成员复制成.env填自己的 Key 即可。CrewAI 的 Agent 定义里只引用llm变量不出现任何硬编码凭证。需要长期跑编码类或 Agent 类任务的话可以了解下 Coding Plan 这类按周期计费的方案适合高频调用场景。验证模型连通性时模型对话页面能快速确认某个 Model ID 是否可用省得在代码里反复试。接入文档里有完整的端点和参数说明遇到协议细节可以直接查。API Keys 页面负责创建和轮换 Key建议按项目分 Key方便用量归因。最后留一个实用技巧在call()里加一行请求耗时日志import time后记录start time.time()和elapsed time.time() - start打印出来。多 Agent 编排时你能一眼看出是哪个 Agent 的 LLM 调用拖慢了整体流程比在 Crew 层面猜要高效得多。这个日志在排查超时问题时特别有用配合timeout参数一起调能把稳定性问题定位到具体环节。