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, tools=None, callbacks=None, available_functions=None)，返回值必须是字符串或可被 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_KEY=sk-你的Key TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL_ID=你的模型ID然后在代码里用os.getenv读取。这样做的好处是子类实例化时不用把 Key 写死在参数里，换环境只改.env。如果你用 shell 直接导出也行：
export TAOTOKEN_API_KEY="sk-你的Key" export TAOTOKEN_BASE_URL="https://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__(model=model, temperature=temperature) 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": f"Bearer {self.api_key}", "Content-Type": "application/json", }, json=payload, timeout=self.timeout, ) resp.raise_for_status() except requests.Timeout: raise TimeoutError("TaoToken 请求超时，检查网络或调大 timeout") except requests.RequestException as e: raise RuntimeError(f"TaoToken 请求失败: {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="你擅长把零散信息归纳成清晰条目。", llm=llm, verbose=True, ) task = Task( description="整理 CrewAI 自定义 LLM 的三个关键接口方法", expected_output="三条要点，每条不超过 50 字", agent=agent, ) 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 在verbose=True下会打印 Agent 的思考过程和最终输出。成功时你会看到类似这样的结构：Agent 先输出 Thought，然后调用 LLM 得到结果，最后result.raw打印出三条要点。我实测下来，从 kickoff 到返回通常在几秒到十几秒，取决于模型响应速度和任务复杂度。
验证时重点看三个信号。第一，call()返回的是非空字符串，不是 None 也不是异常。第二，Crew 的result.raw里包含任务要求的输出格式，比如三条要点。第三，如果开了verbose，日志里能看到 Agent 确实调用了你的TaoTokenLLM实例，而不是回退到默认 LiteLLM。如果这三点都满足，说明自定义 LLM 已经稳定接入 CrewAI 编排流程。
再补一个带工具调用的验证，确认supports_function_calling分支正常。定义一个简单函数，传给 Agent 的 tools：
def get_word_count(text: str) -> int: return len(text) agent = Agent( role="计数助手", goal="统计给定文本的字数", backstory="你只做字数统计。", llm=llm, tools=[get_word_count], verbose=True, )如果模型支持函数调用，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()是否返回 True；call()里是否在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参数一起调，能把稳定性问题定位到具体环节。
