provider 适配:把统一接口翻译成各家方言
Day 49 我们知道 _get_native_provider 会返回 OpenAICompletion / AnthropicCompletion / GeminiCompletion... 这些"原生适配器"。今天进 llms/providers/ 目录,拆开其中最有代表性的 AnthropicCompletion:它同样继承 BaseLLM,同样实现 call,但内部要处理 Anthropic 特有的一切——系统消息要单拎出来、消息必须 user/assistant 交替、工具结果要塞进 user 消息、思考块要放最前。你会看到"统一接口 + 各自实现"这条设计线在真实差异面前是怎么落地的。
痛点:为什么每家 API 都要单独适配?
role:"tool" 消息,Anthropic 要塞进 user 消息的 tool_result 内容块……如果 Day 49 那个统一 call 想适配所有家,代码里就得写满 if provider == "anthropic"。怎么优雅地隔离这些差异?BaseLLM 并覆盖 call 及一堆格式化方法。差异被封装在各自的类里——AnthropicCompletion 里全是 Anthropic 的怪癖,OpenAICompletion 里全是 OpenAI 的规矩,彼此互不干扰。上层 call(messages) 时多态分发到对应适配器,"翻译"这件脏活就地消化。这是"策略模式":同一个接口,多种可互换的具体策略。目录布局一眼看清有哪些原生 provider(llms/providers/):
llms/providers/
├── openai/completion.py # OpenAICompletion
├── anthropic/completion.py # AnthropicCompletion(今天主角)
├── azure/completion.py # AzureCompletion
├── gemini/completion.py # GeminiCompletion
├── bedrock/completion.py # BedrockCompletion
├── snowflake/completion.py # SnowflakeCompletion
├── openai_compatible/completion.py # 一批"类 OpenAI"服务共用
└── utils/common.py # 共享工具(safe_tool_conversion 等)
completion.py,里面一个 XxxCompletion 类。想加新模型?照葫芦画瓢新建一个文件夹、写一个类、在 Day 49 的 _get_native_provider 里加一个分支。扩展点非常清晰。AnthropicCompletion:也是一个 BaseLLM
它同样继承 BaseLLM,但字段是 Anthropic 专属的(anthropic/completion.py:148):
# anthropic/completion.py:148
class AnthropicCompletion(BaseLLM):
"""Anthropic native completion implementation. ...native tool use,
streaming support, and proper message formatting."""
llm_type: Literal["anthropic"] = "anthropic"
model: str = "claude-3-5-sonnet-20241022"
max_retries: int = 2
max_tokens: int = 4096 # ★Anthropic 要求 max_tokens 必填
stream: bool = False
client_params: dict[str, Any] | None = None
interceptor: BaseInterceptor[...] | None = None
thinking: AnthropicThinkingConfig | None = None # 思考模式配置
response_format: JsonResponseFormat | type[BaseModel] | None = None
tool_search: AnthropicToolSearchConfig | None = None
is_claude_3: bool = False
supports_tools: bool = True
_client: Any = PrivateAttr(default=None) # SDK 同步客户端(私有)
_async_client: Any = PrivateAttr(default=None) # SDK 异步客户端
llm_type = "anthropic"身份标记,和 Day 49 的 "litellm" 区分开。max_tokens = 4096(有默认值)★关键差异:Anthropic 的 API 强制要 max_tokens。OpenAI 可以不给,Anthropic 不给会报错,所以这里给了默认值兜底。thinking / tool_searchAnthropic 独有特性——扩展思考、内置工具搜索。原生适配器才能用上这些,LiteLLM 兜底路一般吃不到。_client / _async_client (PrivateAttr)真正的 Anthropic SDK 客户端对象,用 PrivateAttr 存——不参与 Pydantic 序列化,是纯运行时状态。初始化前还有一步字段归一化(anthropic/completion.py:176):
# anthropic/completion.py:176
@model_validator(mode="before")
def _normalize_anthropic_fields(cls, data):
popped = data.pop("stop_sequences", None)
seqs = popped if popped is not None else (data.get("stop") or [])
if isinstance(seqs, str): seqs = [seqs]
data["stop"] = seqs # 统一成 list
data["is_claude_3"] = "claude-3" in data.get("model", "").lower() # 顺手判代际
ts = data.get("tool_search")
if ts is True: data["tool_search"] = AnthropicToolSearchConfig() # True→默认配置对象
return data
@model_validator(mode="before") 在 Pydantic 真正校验字段之前运行,专门用来"把用户传的各种花样归一化成规范形状"。比如 stop 既接受字符串又接受列表,这里统一成列表;tool_search=True 这种简写自动展开成配置对象。把宽容留给用户、把规范留给内部。客户端惰性构建:import 时还没 key 也不崩
构建 SDK 客户端这一步做得很谨慎(anthropic/completion.py:195):
# anthropic/completion.py:195
@model_validator(mode="after")
def _init_clients(self) -> AnthropicCompletion:
"""Eagerly build clients when the API key is available, otherwise
defer so LLM(model="anthropic/...") can be constructed at module
import time even before deployment env vars are set."""
try:
self._client = self._build_sync_client()
self._async_client = self._build_async_client()
except ValueError:
pass # ★没 key → 静默跳过,不报错
return self
def _get_sync_client(self) -> Any: # :222
if self._client is None: # 用的时候若还没建,补建
self._client = self._build_sync_client()
return self._client
mode="after"字段都校验完了才跑——此时能安全读 self.api_key 等。try...except ValueError: pass★核心边界:构建客户端需要 API key,可有人会在没设 key的时候就 import 定义了 LLM 的模块。这里吞掉 ValueError,让"定义"和"真正调用"解耦。_get_sync_client 补建真调用时若客户端还是 None(因为当时没 key),此刻再建一次。这时 key 应该已就位。sync + async 两套同步、异步客户端都建,分别服务 call 和 acall。LLM(model="anthropic/...") 的那一刻若环境变量还没注入,就直接崩——很多部署场景里模块 import 早于 env 注入,这会让人很痛苦。纯惰性(永远等到第一次调用才建):多一次判断、且延迟暴露配置错误。源码选了"能急切就急切,不行就静默降级到惰性":有 key 时 _init_clients 立刻建好(快);没 key 时吞掉异常,等 _get_sync_client 用时再补建(稳)。两头好处都要,用一个 try/except 把成本压到最低。注释里明明白白写了为什么这么做——这是高质量源码的标志。消息格式翻译:Anthropic 的四条铁律
最能体现"翻译官"价值的方法(anthropic/completion.py:653),docstring 列出了 Anthropic 的硬性要求:
# anthropic/completion.py:653
def _format_messages_for_anthropic(self, messages):
"""Anthropic has specific requirements:
- System messages are separate from conversation messages # ① system 单拎出来
- Messages must alternate between user and assistant # ② 必须交替
- First message must be from user # ③ 首条必是 user
- Tool results must be in user messages with tool_result blocks # ④ 工具结果进 user
- When thinking is enabled, assistant messages must start with thinking blocks
Returns: Tuple of (formatted_messages, system_message)."""
base_formatted = super()._format_messages(messages) # 先做通用格式化
formatted_messages: list[LLMMessage] = []
system_message: str | None = None
pending_tool_results: list[dict[str, Any]] = []
for message in base_formatted:
... # 逐条按四条铁律重排
返回 (messages, system)★关键:把 system 消息从列表里剥出来单独返回,因为 Anthropic 的 API 把 system 当独立参数,不能混在对话里。super()._format_messages先复用父类的通用格式化(字符串转列表等),再在结果上做 Anthropic 专属重排。共性交给父类、个性自己处理。pending_tool_results暂存工具结果,因为它们要合并进下一条 user 消息的 tool_result 块,而不是自成一条。user/assistant 交替OpenAI 允许连续多条同角色消息,Anthropic 不行——这里要把它们合并/重排成严格交替。[{system}, {user:问}, {assistant:调工具}, {tool:结果}, {user:继续}]。翻给 OpenAI:几乎原样,system 留在列表里,tool 消息用
role:"tool"。翻给 Anthropic:system 抽成独立参数
system="...";{tool:结果} 被塞进后面那条 user 消息的 content:[{type:"tool_result",...}] 块里;确保整体 user→assistant→user 严格交替。同一份对话,翻译官吐出两种截然不同的排布。provider 版 call:同一个契约,各自的流水线
对比 Day 49 的 LiteLLM 版 call,Anthropic 版走自己的流程(anthropic/completion.py:277):
# anthropic/completion.py:277
def call(self, messages, tools=None, callbacks=None, available_functions=None,
from_task=None, from_agent=None, response_model=None) -> str | Any:
with llm_call_context():
try:
self._emit_call_started_event(...) # ① 同样广播开始事件
formatted_messages, system_message = \
self._format_messages_for_anthropic(messages) # ② 翻译成 Anthropic 格式
if not self._invoke_before_llm_call_hooks(formatted_messages, from_agent):
raise ValueError("LLM call blocked by before_llm_call hook") # ③ 同样过钩子
completion_params = self._prepare_completion_params(
formatted_messages, system_message, tools, available_functions) # ④ 组参数
effective_response_model = response_model or self.response_format
if self._effective_stream():
return self._handle_streaming_completion(completion_params, ...) # 流式
return self._handle_completion(completion_params, ...) # 一次性
except Exception as e:
self._emit_call_failed_event(error=f"Anthropic API call failed: {e!s}", ...)
raise # ⑤ 失败也广播事件再抛
签名一模一样★和 Day 49 BaseLLM.call 的契约完全一致——这才能让上层多态调用。差异全在方法体内部。_emit_call_started_event开始/失败事件的广播每个 provider 都做——因为它们都继承了 BaseLLM 里的这些 _emit_* 方法。事件语义统一。_prepare_completion_params(4 个参数)注意它比 Day 49 的版本多了 system_message 参数——因为 Anthropic 的 system 是独立参数,必须单独传进去。_handle_completion / streaming真正调 Anthropic SDK 的地方,返回后还要把 Anthropic 的响应结构翻回统一格式。except → emit_failed → raise失败先广播 LLMCallFailedEvent(让遥测/日志收到),再原样抛出——不吞错。call 都调 _emit_call_started_event / _invoke_before_llm_call_hooks / _emit_call_failed_event——这些"共性行为"实现在 BaseLLM(Day 53/54 会看到它们的定义),子类直接继承。而 _format_messages_for_anthropic / _prepare_completion_params 这些"差异行为"由子类自己实现。这就是模板方法思想:骨架统一,具体步骤各自填。OpenAI 兼容层:一个类服务七家服务
最省心的复用——一批"长得像 OpenAI"的服务共用一个类(openai_compatible/completion.py:113):
# openai_compatible/completion.py:113
class OpenAICompatibleCompletion(OpenAICompletion): # ★直接继承 OpenAICompletion
"""Supported providers:
- openrouter / deepseek / ollama / ollama_chat
- hosted_vllm / cerebras / dashscope
Example:
llm = LLM(model="deepseek/deepseek-chat")
llm = LLM(model="llama3", provider="ollama")"""
@model_validator(mode="before")
def _resolve_provider_config(cls, data):
provider = data.get("provider", "")
config = OPENAI_COMPATIBLE_PROVIDERS.get(provider)
if config is None:
supported = ", ".join(sorted(OPENAI_COMPATIBLE_PROVIDERS.keys()))
raise ValueError(f"Unknown OpenAI-compatible provider: {provider}. "
f"Supported providers: {supported}") # ★未知 provider 明确报错
data["api_key"] = cls._resolve_api_key(data.get("api_key"), config, provider)
data["base_url"] = cls._resolve_base_url(data.get("base_url"), config, provider)
data["default_headers"] = cls._resolve_headers(data.get("default_headers"), config)
return data
继承 OpenAICompletion★不是继承 BaseLLM,而是继承 OpenAI 适配器——因为这些服务的 API 就是 OpenAI 协议,调用逻辑完全复用,只需改"连到哪、用什么 key、带什么头"。_resolve_provider_config按 provider 名查一张配置表,自动填好 base_url(deepseek 连 deepseek.com、ollama 连本地)、api_key(读对应环境变量)、请求头。未知 provider → raise★边界:写了个不认识的 provider,立刻报错并列出所有支持的名字——比默默连错地方强太多。_resolve_api_key/base_url/headers三个专职方法,把"每家的连接细节"抽出来,config 表驱动。加一家新服务只需往表里加一行。XxxCompletion 类——但它们的调用逻辑 99% 相同(都是 OpenAI 协议),只有连接配置不同,分开写就是大量重复代码。源码做法:一个 OpenAICompatibleCompletion 继承 OpenAI 的全部调用逻辑,差异只剩"配置表里的一行"(url/key/header)。用"配置驱动"替代"类爆炸"——加一家新服务的成本从"写一个类"降到"加一行配置"。当多个对象只有数据不同、行为相同时,就该用配置区分而非继承区分。能力方法的 provider 覆盖:更准的答案
Day 49 说"原生 provider 会覆盖能力探测方法",这就是例子(anthropic/completion.py:1846):
# anthropic/completion.py:1846
def get_context_window_size(self) -> int:
from crewai.llm import CONTEXT_WINDOW_USAGE_RATIO
context_windows = {
"claude-3-5-sonnet": 200000, "claude-3-5-haiku": 200000,
"claude-3-opus": 200000, "claude-3-7-sonnet": 200000,
"claude-2.1": 200000, "claude-2": 100000,
"claude-instant": 100000,
}
for model_prefix, size in context_windows.items():
if self.model.startswith(model_prefix):
return int(size * CONTEXT_WINDOW_USAGE_RATIO) # 命中→按前缀返回
return int(200000 * CONTEXT_WINDOW_USAGE_RATIO) # 默认 200k
def supports_function_calling(self) -> bool: # :1838
return True # Anthropic 都支持,直接 True
Anthropic 专属的窗口表不查 Day 49 的通用表,而是用 Anthropic 自己更准的一份——claude 系列基本都是 200k。startswith 前缀匹配模型名 claude-3-5-sonnet-20241022 用 startswith("claude-3-5-sonnet") 命中,不用列全每个日期版本。× CONTEXT_WINDOW_USAGE_RATIO★复用 Day 49 的 0.85 系数——留 15% 余量。这个常量跨 provider 共享,D52 详解。supports_function_calling → True比 Day 49 那个"问 litellm"的版本干脆——Anthropic 我自己家的能力我最清楚,直接给答案。👶 小白:既然都继承 BaseLLM,为啥 Anthropic 要重写 get_context_window_size?基类不是有默认实现吗?
👨🏫 老师:基类的默认实现(base_llm.py:494)只会返回一个保守的 DEFAULT_CONTEXT_WINDOW_SIZE=8192——对 claude 的 200k 窗口来说太小了,会导致明明能塞得下的对话被误判"超长"而提前压缩。原生 provider 重写它,是为了给出贴合自家模型的准确值。这就是"覆盖"的意义:基类给个能跑的下限,子类给个更对的答案。
边界 + 今日小结
try: from anthropic import Anthropic ... except ImportError: raise ImportError('Anthropic native provider not available, to install: uv add "crewai[anthropic]"')。这意味着:核心 crewai 包并不强制安装每家的 SDK——你想用 Anthropic 原生 provider,得额外装 crewai[anthropic]。如果没装就写 LLM(model="anthropic/..."),会在 _get_native_provider 触发 import 时抛出这条带安装命令的清晰错误。这是"可选依赖"的标准做法:核心轻量,按需装重。踩坑点在于——报错信息其实已经告诉你怎么修,别忽略它。🧠 今天你应该能回答
- 为什么每家 API 要单独一个适配器类?封装了什么差异?
- Anthropic 消息格式的四条铁律是什么?system 为何单拎出来?
- 客户端为什么"能急切就急切、不行就惰性"?解决什么部署痛点?
- provider 版
call和 Day 49 的call有哪些共性、哪些差异? - OpenAI 兼容层为什么用"一个类 + 配置表"而非"类爆炸"?
- 原生 provider 为什么要覆盖
get_context_window_size?
✋ 10 分钟动手
P=lib/crewai/src/crewai
ls $P/llms/providers/ # 有哪些原生 provider
sed -n '148,215p' $P/llms/providers/anthropic/completion.py # 字段 + 归一化
sed -n '195,232p' $P/llms/providers/anthropic/completion.py # 客户端惰性构建
sed -n '277,345p' $P/llms/providers/anthropic/completion.py # provider 版 call
sed -n '113,166p' $P/llms/providers/openai_compatible/completion.py # 兼容层复用
call 里传了 tools 和 response_model——今天没细看它们怎么工作。明天专攻函数调用与结构化输出:_handle_tool_call 怎么执行模型点名的函数、_validate_structured_output 怎么把模型吐的 JSON 校验成 Pydantic 对象、_prepare_completion_params 怎么把 response_format 塞进请求。