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

LLM 抽象:一个类装下所有大模型

前 48 天我们一直在说"问一次 LLM"(get_llm_responseself.llm.call(...)),但从没打开过 self.llm 到底是什么。今天进入阶段8,第一站就是 crewai/llm.py——CrewAI 里所有大模型调用的总闸门。你会看到:一个 BaseLLM 抽象基类定义"所有 LLM 长什么样",一个 LLM 类用 __new__ 工厂在"原生 SDK / LiteLLM 兜底"之间路由,还有 call / supports_function_calling / get_context_window_size 这些每个执行器都会用到的方法。

📍 你在 60 天里的位置(阶段8 LLM 与集成 · 共 6 天)
阶段7 Flow D41-48 D49 LLM 抽象 D50 provider 适配 D51 函数调用/结构化 D52 上下文窗口 D53 token/成本 D54 遥测 阶段9 进阶
💡 先用一个类比兜住今天 LLM 类就像一个万能遥控器。你不用关心家里电视是索尼还是小米(OpenAI 还是 Claude),拿起这个遥控器按"开机/换台"(call),它内部会自动识别你面对的是哪台机器,然后发对应品牌的红外码。你按下按钮那一刻,遥控器先做的事是"我要不要用这台电视自带的原厂遥控(原生 SDK),还是用一个学习型万能码库(LiteLLM)"。今天就是拆开这个遥控器的外壳。
L01

痛点:为什么不直接调 OpenAI 就好?

🤔 痛点如果 CrewAI 只支持 OpenAI,代码里到处写 openai.chat.completions.create(...) 不就完了?可现实是:用户可能用 Claude、Gemini、Bedrock、本地 Ollama,甚至自己公司搭的兼容网关。每家 SDK 的参数名、消息格式、返回结构、工具调用协议都不一样。如果 Agent 执行器里直接写死某一家,换模型就得改一堆代码。怎么让"上层执行器只管说一句 llm.call(messages),底下自动适配几十种模型"?
💡 一句话本质 CrewAI 用一个抽象基类 BaseLLM 定义统一接口(所有 LLM 都得有 call / supports_function_calling / get_context_window_size),再用一个 LLM 工厂类在实例化那一刻决定"走哪条实现路":模型是主流大厂的 → 用 CrewAI 自己写的原生 provider(更快更准);是冷门的 → 交给 LiteLLM 这个"万能翻译层"兜底。上层永远只依赖 BaseLLM 这个接口,从不关心底下是谁。这就是面向接口编程 + 工厂模式的经典组合。

先看 llm.py 的整体骨架(llm.py,2714 行的核心文件):

# llm.py:368
class LLM(BaseLLM):                              # 继承统一接口
    llm_type: Literal["litellm"] = "litellm"
    def __new__(cls, model, is_litellm=False, **kwargs): ...   # :393 工厂路由
    def _get_native_provider(cls, provider): ...              # :665 provider 映射表
    def call(self, messages, tools=None, ...): ...            # :1820 高层调用入口
    def supports_function_calling(self): ...                  # :2402 会不会原生工具调用
    def get_context_window_size(self): ...                    # :2443 上下文窗口多大
    def _prepare_completion_params(self, messages, tools): ...# :748 组装请求参数
大白话看方法名就懂这一层干嘛的:__new__ 是"出厂时决定用哪套零件",call 是"对外的按钮",supports_* 是"自我介绍我会哪些本事",get_context_window_size 是"我脑容量多大"。上层执行器(Day 08 那个 while 循环)拿到的就是这么一个对象,只管按按钮。
L02

BaseLLM:所有 LLM 必须长这样

抽象基类定义了"一个 LLM 至少要有哪些字段和方法"(llms/base_llm.py:150):

# llms/base_llm.py:150
class BaseLLM(BaseModel, ABC):               # 既是 Pydantic 模型,又是抽象基类
    model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)
    llm_type: str = "base"
    model: str                               # ★唯一必填:模型名
    temperature: float | None = None
    max_tokens: int | float | None = None
    stream: bool | None = None
    api_key: str | None = None
    base_url: str | None = None
    provider: str = Field(default="openai")
    stop: list[str] = Field(default_factory=list,
        validation_alias=AliasChoices("stop", "stop_sequences"))   # 两种名字都认
    additional_params: dict[str, Any] = Field(default_factory=dict)

以及那个所有子类都必须实现的抽象方法 callllms/base_llm.py:312):

# llms/base_llm.py:312
def call(
    self,
    messages: str | list[LLMMessage],
    tools: list[dict[str, BaseTool]] | None = None,
    callbacks: list[Any] | None = None,
    available_functions: dict[str, Any] | None = None,
    from_task: Task | None = None,
    from_agent: BaseAgent | None = None,
    response_model: type[BaseModel] | None = None,
) -> str | Any:
    """Call the LLM with the given messages. ...
    Returns: Either a text response (str) or the result of a tool call (Any)."""
BaseModel, ABC 双继承既享受 Pydantic 的字段校验/序列化,又用 ABC 强制"抽象方法必须被子类实现"。一个类扮两个角色。
model: str(无默认)唯一必填字段——没有模型名就没法用。其余全有默认值,可选。
AliasChoices("stop","stop_sequences")贴心兼容:用户写 stop=stop_sequences= 都能被同一个字段接住,减少踩坑。
call 的签名★这是全框架"问一次 LLM"的统一契约:进 messages+tools,出"文本或工具结果"。所有 provider 都得按这个签名实现。
-> str | Any返回类型故意宽松:普通对话返回 str;如果这次调用触发了工具,返回工具执行结果(任意类型)。
💡 为什么要单独抽一个 BaseLLM,而不是让大家都继承 LLM?因为 BaseLLM 是"接口契约",LLM 是"其中一种实现(litellm 版)"。用户想接一个 CrewAI 没内置的模型(比如公司自研),只要继承 BaseLLM 实现 call 就能插进整个框架,完全不必碰 LiteLLM。抽象和实现分离,扩展点就清清楚楚。
数据结构:LLM 类的继承体系 BaseLLM (ABC 接口) call() · supports_function_calling() · get_context_window_size() LLM (litellm 版) 原生 provider 你自定义的 LLM 虚线 = "is-a":三者都实现同一份 call 契约 上层执行器只 import BaseLLM 类型 → 换实现零成本
图注:BaseLLM 是接口,其余都是它的实现。上层只认接口,所以三种实现可以自由替换。
L03

LLM 类:litellm 版实现的全部字段

LLMBaseLLM 基础上又加了一大堆"LiteLLM 支持的高级参数"(llm.py:368):

# llm.py:368
class LLM(BaseLLM):
    llm_type: Literal["litellm"] = "litellm"
    completion_cost: float | None = None       # 本次调用花了多少钱
    timeout: float | int | None = None
    top_p: float | None = None
    max_completion_tokens: int | None = None
    presence_penalty: float | None = None
    frequency_penalty: float | None = None
    response_format: JsonResponseFormat | type[BaseModel] | None = None  # 结构化输出(D51)
    seed: int | None = None
    reasoning_effort: Literal["none","low","medium","high"] | None = None # o 系列推理力度
    stream: bool = False
    thinking: Any = None                        # 思考/reasoning 配置
    context_window_size: int = 0                # 上下文窗口(D52 惰性算)
    is_anthropic: bool = False
llm_type = "litellm"标记"我是 litellm 实现"。原生 provider 会写别的值,方便框架区分。
completion_cost成本字段——LiteLLM 能估算每次调用的美元花费,D53 会讲怎么用。
response_format结构化输出的钥匙:传一个 Pydantic 类进来,就能让模型直接吐 JSON(D51 主题)。
reasoning_effort / thinking为 o1/o3、Claude thinking 这类"会先想再答"的模型准备的旋钮。
context_window_size = 0默认 0 = "还没算过"。首次调用 get_context_window_size() 时才惰性计算并缓存(D52 讲)。
📝 例子:一行创建就带出这些字段 llm = LLM(model="gpt-4o", temperature=0.7, response_format=MyModel, stream=True)
这行代码创建的 llm 对象里,model="gpt-4o"temperature=0.7response_format=MyModelstream=True,其余字段都是默认值。context_window_size 此刻还是 0——要等第一次问它"你脑容量多大"才会算出 128000 × 0.85 = 108800
L04

__new__ 工厂:出厂时决定走哪条路

最精妙的地方——LLM(...) 这行代码执行时,先进的不是 __init__ 而是 __new__llm.py:393)。它是个"路由器",注释里写死了 4 条优先级:

# llm.py:393
def __new__(cls, model: str, is_litellm: bool = False, **kwargs: Any) -> LLM:
    """Routing priority:
        1. custom_openai=True → 强制原生 OpenAI(需自定义 endpoint)
        2. provider 显式给了 → 用那个 provider
        3. 模型名带 "/" → 拆前缀,是原生大厂且模型在常量表里 → 原生;否则 LiteLLM
        4. 都不满足 → 从模型名推断 provider"""
    if not model or not isinstance(model, str):
        raise ValueError("Model must be a non-empty string")   # ★边界:空模型名直接拦
    custom_openai = bool(kwargs.pop("custom_openai", False))
    explicit_provider = kwargs.get("provider")
    if custom_openai:
        provider = "openai"; use_native = True                 # 路①
    elif explicit_provider:
        provider = explicit_provider; use_native = True         # 路②
    elif "/" in model:                                          # 路③
        prefix, _, model_part = model.partition("/")
        provider_mapping = {"openai":"openai","anthropic":"anthropic","claude":"anthropic",
            "azure":"azure","google":"gemini","gemini":"gemini","bedrock":"bedrock",
            "aws":"bedrock","ollama":"ollama", ...}
        canonical_provider = provider_mapping.get(prefix.lower())
        valid_native_model = bool(canonical_provider
            and cls._validate_model_in_constants(model_part, canonical_provider))
        if canonical_provider and valid_native_model:
            provider = canonical_provider; use_native = True
        # 否则落到 LiteLLM
__new__ 而非 __init__★关键:__new__ 在对象还没创建时就介入,可以决定"到底 new 出哪个类的实例"。这正是工厂模式在 Python 里的标准落点。
路① custom_openai公司自建的 OpenAI 兼容网关:强制走原生 OpenAI 客户端,但要求你给了 base_url/endpoint。
路② explicit_provider你直接写 provider="anthropic",那就别猜了,听你的。
路③ 模型名带 "/""anthropic/claude-...":拆出前缀 anthropic,查映射表,再核对模型名在不在常量表里——都对才走原生。
_validate_model_in_constants防手滑:前缀对但模型名拼错(不在已知列表)→ 不敢当原生,退回 LiteLLM 兜底。
控制流:__new__ 的四条路由 LLM(model=...) 调用 ① custom_openai=True? ② 显式 provider? ③ 带"/"且模型在常量表? ④ 从模型名推断 provider 命中→原生 provider OpenAI/Anthropic/... 都不满足→LiteLLM 兜底 否则依次下探
图注:四条路依次判断,任一命中就定 provider 走原生;全不命中则退回 LiteLLM 兜底。
💡 设计取舍①:为什么用 __new__ 而不是简单地写个 create_llm() 函数? 朴素做法:写个工厂函数 make_llm(model) 返回不同对象——但这样用户就得记住"要用 make_llm() 不能直接 LLM()",且历史代码里满是 LLM(...) 没法平滑迁移。源码做法:把路由藏进 __new__,用户依旧写 LLM(model=...),但拿到的可能是 OpenAICompletion、AnthropicCompletion 或 LLM 自己。对用户是零感知的升级——老代码不改一个字就自动享受原生 provider 提速。代价是 __new__ 逻辑复杂、可读性下降,属于"把复杂度关进一个盒子换取调用方的简单"。
L05

_get_native_provider:provider 名 → 实现类

路由算出 provider 字符串后,靠这张"查表"拿到真正的实现类(llm.py:665):

# llm.py:665
def _get_native_provider(cls, provider: str) -> type | None:
    if provider == "openai":
        from crewai.llms.providers.openai.completion import OpenAICompletion
        return OpenAICompletion
    if provider == "anthropic" or provider == "claude":
        from crewai.llms.providers.anthropic.completion import AnthropicCompletion
        return AnthropicCompletion
    if provider == "azure" or provider == "azure_openai":
        from crewai.llms.providers.azure.completion import AzureCompletion
        return AzureCompletion
    if provider == "google" or provider == "gemini":
        from crewai.llms.providers.gemini.completion import GeminiCompletion
        return GeminiCompletion
    if provider == "bedrock":
        from crewai.llms.providers.bedrock.completion import BedrockCompletion
        return BedrockCompletion
    if provider == "snowflake":
        from crewai.llms.providers.snowflake.completion import SnowflakeCompletion
        return SnowflakeCompletion
    openai_compatible_providers = {"openrouter","deepseek","ollama","ollama_chat",
        "hosted_vllm","cerebras","dashscope"}
    if provider in openai_compatible_providers:
        from crewai.llms.providers.openai_compatible.completion import OpenAICompatibleCompletion
        return OpenAICompatibleCompletion
    return None                              # 没有原生实现 → 上层退回 LiteLLM
函数内 import★每个分支都在函数内部 import,不在文件顶部。这是"惰性导入":只有真用到 anthropic 时才 import anthropic 包,避免没装的包在启动时就报错。
"anthropic" or "claude"两个别名指向同一个类——用户写哪个都行。
openai_compatible_providers 集合一批"长得像 OpenAI"的服务(openrouter/deepseek/ollama...)共用一个 OpenAICompatibleCompletion,不用每家单写。
return None★兜底信号:没有对应原生实现,上层就知道"这个得交给 LiteLLM"。
这里为什么不用一个大 dict 映射而用一串 if?因为每个分支要触发不同的 import——dict 的值若写成类引用,就得在文件顶部全 import,惰性加载的好处就没了。用 if + 函数内 import 是"惰性 + 可读"的折中。

👶 小白:那 LiteLLM 到底是啥?为什么当兜底?

👨‍🏫 老师:LiteLLM 是个第三方库,它把100+ 家模型 API都翻译成 OpenAI 的调用格式。好处是覆盖广、你几乎啥模型都能用;代价是多一层封装、不一定能用上某家最新的原生特性。所以 CrewAI 的策略是:主流大厂我自己写原生 provider(快、能用最新特性),长尾的交给 LiteLLM(广、省心)。两全其美。

L06

call:对外那颗按钮按下去发生什么

无论走哪条实现,执行器调的都是 call。看 LiteLLM 版的 callllm.py:1820):

# llm.py:1820
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():                         # 生成本次调用的唯一 call_id
        self._emit_call_started_event(...)           # ① 广播"LLM 调用开始"事件(D26 事件系统)
        self._validate_call_params()                 # ② 校验参数合法
        if isinstance(messages, str):
            messages = [{"role": "user", "content": messages}]   # ③ 字符串→标准消息列表
        if "o1" in self.model.lower():
            for message in messages:
                if message.get("role") == "system":
                    message["role"] = "assistant"    # ④ o1 不支持 system,降级为 assistant
        if not self._invoke_before_llm_call_hooks(messages, from_agent):
            raise ValueError("LLM call blocked by before_llm_call hook")  # ⑤ 钩子可拦截
        with suppress_warnings():
            params = self._prepare_completion_params(messages, tools)     # ⑥ 组装请求参数
            if self._effective_stream():
                result = self._handle_streaming_response(params=params, ...)  # 流式
            else:
                result = self._handle_non_streaming_response(params=params, ...)  # 一次性
            if isinstance(result, str):
                result = self._invoke_after_llm_call_hooks(messages, result, ...) # 后置钩子
llm_call_context()上下文管理器:给这一次调用分配唯一 call_id,让开始/结束/失败事件能对上号。
_emit_call_started_event调用开始就往事件总线广播——监听器(遥测、日志、UI)都能收到。这是 D54 遥测的挂载点。
字符串 → [{"role":"user"...}]贴心:你传一个纯字符串,它自动包成标准消息格式。
o1 的 system → assistant★边界:o1 系列模型不接受 system 消息,框架自动把 system 降级成 assistant,免得报错。
before/after 钩子前置钩子返回 False 就直接拦下(安全审查场景);后置钩子能改写结果。这是 hooks 系统(阶段9)的入口。
_prepare_completion_params把 self 上那一堆字段(temperature、max_tokens...)打包成给底层 API 的 dict——见 D51。
大白话call 就是"按下按钮"到"拿到回答"之间的全部流水线:广播开始 → 校验 → 规整消息 → 特殊模型打补丁 → 钩子放行 → 组装参数 → 真调 API(流式或一次性)→ 钩子加工结果。执行器只看到"进 messages、出结果",中间这些它一概不管。
L07

能力探测:LLM 的"自我介绍"方法族

还记得 Day 08 那句 self.llm.supports_function_calling() 吗?它就在这(llm.py:2402):

# llm.py:2402
def supports_function_calling(self) -> bool:
    """Note: 只用于 litellm 兜底路径。原生 provider 会各自覆盖此方法。"""
    if not _ensure_litellm():
        return True                          # litellm 没装 → 假设现代模型都支持
    try:
        provider = self._get_custom_llm_provider()
        return litellm.utils.supports_function_calling(self.model, custom_llm_provider=provider)
    except Exception as e:
        logging.error(f"Failed to check function calling support: {e!s}")
        return True                          # 查不到 → 默认 True

def supports_stop_words(self) -> bool:       # :2422
    model_lower = self.model.lower() if self.model else ""
    if "gpt-5" in model_lower:
        return False                         # ★硬编码:gpt-5 不支持 stop words
    ...
只用于 litellm 兜底★注释点破:原生 provider(D50)会各自覆盖这些方法,用更准的答案。这里是 litellm 版的默认实现。
litellm 没装 → return True降级策略:查不了就乐观假设"现代模型都支持",不因为探测失败就一刀切禁掉功能。
gpt-5 硬编码 return False特例硬写:某些新模型不支持 stop words,来不及等 litellm 更新,直接在代码里打补丁。
except → return True探测出错也默认 True——"宁可试了失败,不要因探测挂了就少一个功能"。
💡 设计取舍②:能力探测失败时,默认"支持"还是"不支持"? 源码一律选默认支持(return True)。为什么?因为如果错判成"不支持",就会白白关掉一个本可用的功能(比如明明能用原生工具调用,却退回慢又易错的文本 ReAct);而错判成"支持",最坏情况是真调用时报错,那时有 D08 讲的降级机制兜着(原生报错→切文本)。两种错都有代价,但"乐观 + 有兜底"比"悲观 + 少功能"整体体验更好。这是"失败安全的方向选择"——朝着损失更小的那边倒。
L08

边界 + 今日小结

⚠️ 边界:__new__ 返回的类型不再是 LLM,isinstance 还成立吗? 当你写 LLM(model="anthropic/claude-...")__new__ 可能返回一个 AnthropicCompletion 实例。这时 isinstance(obj, LLM) 会是 False(AnthropicCompletion 不继承 LLM),但 isinstance(obj, BaseLLM)True。所以框架内部判断"这是不是一个 LLM"时,一律用 isinstance(x, BaseLLM) 而非 isinstance(x, LLM)(你在 Day 07 的 crew.py:calculate_usage_metrics 里已经见过 if isinstance(agent.llm, BaseLLM))。教训:面向工厂返回多态对象时,类型判断要盯着"接口基类",不能盯着"某个具体实现类"。

👶 小白:我只写 LLM(model="gpt-4o")(没带斜杠、没写 provider),走哪条路?

👨‍🏫 老师:走路④——_infer_provider_from_modelllm.py:634)从模型名 gpt-4o 推断出 provider 是 openai,然后查 _get_native_provider 拿到 OpenAICompletion。所以你其实用上了原生 OpenAI 实现,而不是 LiteLLM。想强制用 LiteLLM?传 is_litellm=True

🧠 今天你应该能回答

  • 为什么要 BaseLLM(接口)和 LLM(实现)两层?
  • __new__ 工厂的 4 条路由优先级分别是什么?
  • "原生 provider"和"LiteLLM 兜底"各自的取舍是什么?
  • call 从按下到出结果经过哪些环节?o1 的 system 消息怎么处理?
  • 能力探测失败为什么默认 True?
  • 为什么类型判断要用 isinstance(x, BaseLLM)

✋ 10 分钟动手

P=lib/crewai/src/crewai
sed -n '150,215p'  $P/llms/base_llm.py    # BaseLLM 接口
sed -n '393,465p'  $P/llm.py              # __new__ 工厂路由
sed -n '665,715p'  $P/llm.py              # _get_native_provider 映射表
# 亲眼看 LLM() 返回的到底是哪个类
python -c "
from crewai import LLM
for m in ['gpt-4o','anthropic/claude-sonnet-4-5','ollama/llama3','foo/bar-999']:
    print(m, '->', type(LLM(model=m)).__name__)
"
明日预告 · Day 50:今天 _get_native_provider 返回的那些 OpenAICompletion / AnthropicCompletion 到底长啥样?明天进 llms/providers/ 目录,逐个拆原生适配器——看它们怎么各自实现 call、怎么把 CrewAI 的统一消息格式翻译成各家 SDK 的专属格式。
← Day 48 Flow vs Crew Day 50 · provider 适配 →