Day 50 / 共 60 天 · 阶段8 LLM 与集成

provider 适配:把统一接口翻译成各家方言

Day 49 我们知道 _get_native_provider 会返回 OpenAICompletion / AnthropicCompletion / GeminiCompletion... 这些"原生适配器"。今天进 llms/providers/ 目录,拆开其中最有代表性的 AnthropicCompletion:它同样继承 BaseLLM,同样实现 call,但内部要处理 Anthropic 特有的一切——系统消息要单拎出来、消息必须 user/assistant 交替、工具结果要塞进 user 消息、思考块要放最前。你会看到"统一接口 + 各自实现"这条设计线在真实差异面前是怎么落地的。

📍 你在 60 天里的位置(阶段8 LLM 与集成 · 共 6 天)
D49 LLM 抽象 D50 provider 适配 D51 函数调用/结构化 D52 上下文窗口 D53 token/成本 D54 遥测 阶段9 进阶
💡 先用一个类比兜住今天 provider 适配器就像翻译官。CrewAI 上层说的是一门"普通话"(统一的 messages 格式),但 OpenAI 听英语、Anthropic 听法语、Gemini 听德语。每个 provider 适配器就是一位专职翻译:进来普通话,它翻成对应语言递给那家 API;对方回话,它再翻回普通话交给上层。翻译官内部怎么处理语法差异是它的事,上层永远只说普通话。今天看 Anthropic 这位翻译官怎么干活。
L01

痛点:为什么每家 API 都要单独适配?

🤔 痛点都是"聊天补全",OpenAI 和 Anthropic 能差多少?差很多:OpenAI 把 system 消息和普通消息放同一个列表,Anthropic 要求 system 单独一个参数;OpenAI 消息可以连续两条 user,Anthropic 要求严格 user/assistant 交替;工具调用结果,OpenAI 用 role:"tool" 消息,Anthropic 要塞进 user 消息的 tool_result 内容块……如果 Day 49 那个统一 call 想适配所有家,代码里就得写满 if provider == "anthropic"。怎么优雅地隔离这些差异?
💡 一句话本质 每家 API 一个独立的适配器类,各自继承 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 里加一个分支。扩展点非常清晰。
L02

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 这种简写自动展开成配置对象。把宽容留给用户、把规范留给内部。
数据结构:provider 适配器的策略家族 BaseLLM.call() 契约 OpenAI Anthropic Gemini Bedrock Snowflake OpenAICompatible(继承 OpenAI,配置表驱动 7 家)
图注:都实现同一份 call 契约(虚线=is-a);OpenAICompatible 又继承 OpenAI,用配置表复用它的调用逻辑。
L03

客户端惰性构建: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 两套同步、异步客户端都建,分别服务 callacall
💡 设计取舍①:构建客户端,"急切" vs "惰性"怎么选? 纯急切(初始化就必须建成客户端):定义 LLM(model="anthropic/...") 的那一刻若环境变量还没注入,就直接崩——很多部署场景里模块 import 早于 env 注入,这会让人很痛苦。纯惰性(永远等到第一次调用才建):多一次判断、且延迟暴露配置错误。源码选了"能急切就急切,不行就静默降级到惰性":有 key 时 _init_clients 立刻建好(快);没 key 时吞掉异常,等 _get_sync_client 用时再补建(稳)。两头好处都要,用一个 try/except 把成本压到最低。注释里明明白白写了为什么这么做——这是高质量源码的标志。
L04

消息格式翻译: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 不行——这里要把它们合并/重排成严格交替。
📝 例子:同一段对话,两家格式差异 CrewAI 内部统一格式:[{system}, {user:问}, {assistant:调工具}, {tool:结果}, {user:继续}]
翻给 OpenAI:几乎原样,system 留在列表里,tool 消息用 role:"tool"
翻给 Anthropic:system 抽成独立参数 system="..."{tool:结果} 被塞进后面那条 user 消息的 content:[{type:"tool_result",...}] 块里;确保整体 user→assistant→user 严格交替。同一份对话,翻译官吐出两种截然不同的排布。
控制流:统一 messages → Anthropic 专属格式 统一 messages(含 system / tool 消息) super()._format_messages 通用格式化 ① system 抽成独立参数 ② tool 结果并入 user 块 ③ 强制 user/assistant 交替 返回 (formatted_messages, system_message)
图注:先做通用格式化,再套 Anthropic 三条重排规则,最后把 system 与对话分成两个返回值。
L05

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(让遥测/日志收到),再原样抛出——不吞错。
💡 共享行为放基类,差异化放子类三个 provider 的 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 这些"差异行为"由子类自己实现。这就是模板方法思想:骨架统一,具体步骤各自填。
L06

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 表驱动。加一家新服务只需往表里加一行。
💡 设计取舍②:为什么不给 deepseek/ollama 各写一个类? 朴素做法:deepseek、ollama、cerebras... 每家一个 XxxCompletion 类——但它们的调用逻辑 99% 相同(都是 OpenAI 协议),只有连接配置不同,分开写就是大量重复代码源码做法:一个 OpenAICompatibleCompletion 继承 OpenAI 的全部调用逻辑,差异只剩"配置表里的一行"(url/key/header)。用"配置驱动"替代"类爆炸"——加一家新服务的成本从"写一个类"降到"加一行配置"。当多个对象只有数据不同、行为相同时,就该用配置区分而非继承区分。
L07

能力方法的 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-20241022startswith("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 重写它,是为了给出贴合自家模型的准确值。这就是"覆盖"的意义:基类给个能跑的下限,子类给个更对的答案。

L08

边界 + 今日小结

⚠️ 边界:装了 crewai 不等于能用所有 provider 看 Anthropic 适配器文件顶部的 import(anthropic/completion.py:24):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  # 兼容层复用
明日预告 · Day 51:适配器的 call 里传了 toolsresponse_model——今天没细看它们怎么工作。明天专攻函数调用与结构化输出_handle_tool_call 怎么执行模型点名的函数、_validate_structured_output 怎么把模型吐的 JSON 校验成 Pydantic 对象、_prepare_completion_params 怎么把 response_format 塞进请求。
← Day 49 LLM 抽象 Day 51 · 函数调用与结构化输出 →