LLM 抽象:一个类装下所有大模型
前 48 天我们一直在说"问一次 LLM"(get_llm_response、self.llm.call(...)),但从没打开过 self.llm 到底是什么。今天进入阶段8,第一站就是 crewai/llm.py——CrewAI 里所有大模型调用的总闸门。你会看到:一个 BaseLLM 抽象基类定义"所有 LLM 长什么样",一个 LLM 类用 __new__ 工厂在"原生 SDK / LiteLLM 兜底"之间路由,还有 call / supports_function_calling / get_context_window_size 这些每个执行器都会用到的方法。
LLM 类就像一个万能遥控器。你不用关心家里电视是索尼还是小米(OpenAI 还是 Claude),拿起这个遥控器按"开机/换台"(call),它内部会自动识别你面对的是哪台机器,然后发对应品牌的红外码。你按下按钮那一刻,遥控器先做的事是"我要不要用这台电视自带的原厂遥控(原生 SDK),还是用一个学习型万能码库(LiteLLM)"。今天就是拆开这个遥控器的外壳。痛点:为什么不直接调 OpenAI 就好?
openai.chat.completions.create(...) 不就完了?可现实是:用户可能用 Claude、Gemini、Bedrock、本地 Ollama,甚至自己公司搭的兼容网关。每家 SDK 的参数名、消息格式、返回结构、工具调用协议都不一样。如果 Agent 执行器里直接写死某一家,换模型就得改一堆代码。怎么让"上层执行器只管说一句 llm.call(messages),底下自动适配几十种模型"?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 循环)拿到的就是这么一个对象,只管按按钮。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)
以及那个所有子类都必须实现的抽象方法 call(llms/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 是"其中一种实现(litellm 版)"。用户想接一个 CrewAI 没内置的模型(比如公司自研),只要继承 BaseLLM 实现 call 就能插进整个框架,完全不必碰 LiteLLM。抽象和实现分离,扩展点就清清楚楚。LLM 类:litellm 版实现的全部字段
LLM 在 BaseLLM 基础上又加了一大堆"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.7、response_format=MyModel、stream=True,其余字段都是默认值。context_window_size 此刻还是 0——要等第一次问它"你脑容量多大"才会算出 128000 × 0.85 = 108800。__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 兜底。make_llm(model) 返回不同对象——但这样用户就得记住"要用 make_llm() 不能直接 LLM()",且历史代码里满是 LLM(...) 没法平滑迁移。源码做法:把路由藏进 __new__,用户依旧写 LLM(model=...),但拿到的可能是 OpenAICompletion、AnthropicCompletion 或 LLM 自己。对用户是零感知的升级——老代码不改一个字就自动享受原生 provider 提速。代价是 __new__ 逻辑复杂、可读性下降,属于"把复杂度关进一个盒子换取调用方的简单"。_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(广、省心)。两全其美。
call:对外那颗按钮按下去发生什么
无论走哪条实现,执行器调的都是 call。看 LiteLLM 版的 call(llm.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、出结果",中间这些它一概不管。能力探测: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——"宁可试了失败,不要因探测挂了就少一个功能"。边界 + 今日小结
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_model(llm.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__)
"
_get_native_provider 返回的那些 OpenAICompletion / AnthropicCompletion 到底长啥样?明天进 llms/providers/ 目录,逐个拆原生适配器——看它们怎么各自实现 call、怎么把 CrewAI 的统一消息格式翻译成各家 SDK 的专属格式。