LLM Provider 多模型适配层
学习目标
理解 MetaGPT 的 LLM Provider 层如何通过注册表模式实现多模型适配，以及重试、成本和流式传输的实现细节。
项目实践
统一抽象：BaseLLM
所有 LLM 提供商继承 BaseLLM 抽象类：
class BaseLLM: def __init__(self, llm_config: LLMConfig): ... async def aask(self, msg, system_msgs=None, stream=True, temperature=None) -> str: ... async def _achat_completion(self, messages, timeout, temperature) -> ChatCompletion: ... def get_choice_text(self, rsp: ChatCompletion) -> str: ... def get_usage(self, messages, response_text) -> dict: ...公共能力：
aask()：统一的异步对话接口，支持流式和非流式_calc_usage()：计算 token 消耗和成本cost_manager：关联的成本管理器
注册表模式
llm_provider_registry.py 实现了 Provider 注册表：
# 装饰器：将 Provider 类绑定到多个 LLMTypedef register_provider(types: list[LLMType]): def decorator(cls): for t in types: LLM_API_REGISTRY[t] = cls return cls return decorator
# 工厂函数：根据配置创建实例def create_llm_instance(config: LLMConfig) -> BaseLLM: return LLM_API_REGISTRY[config.api_type](config)OpenAILLM：复用适配
OpenAILLM 通过注册表绑定到 10+ 个 LLMType：
@register_provider([ LLMType.OPENAI, LLMType.FIREWORKS, LLMType.OPEN_LLM, LLMType.MOONSHOT, LLMType.MISTRAL, LLMType.YI, LLMType.OPEN_ROUTER, LLMType.DEEPSEEK, LLMType.SILICONFLOW, LLMType.LLAMA_API,])class OpenAILLM(BaseLLM): def _init_client(self): self.aclient = AsyncOpenAI( api_key=self.config.api_key, base_url=self.config.base_url, )所有兼容 OpenAI API 的提供商复用同一个 OpenAILLM 类，只需在配置中指定不同的 base_url。
独立 Provider
需要自定义客户端的提供商独立实现：
| Provider | 用途 | SDK |
|---|---|---|
| AzureOpenAILLM | Azure OpenAI 服务 | openai.AsyncAzureOpenAI |
| AnthropicLLM | Claude 系列 | anthropic.AsyncAnthropic |
| GoogleGeminiLLM | Google Gemini | google.generativeai |
| OllamaLLM | 本地 Ollama | HTTP API |
| QianFanLLM | 百度千帆 | qianfan SDK |
| SparkLLM | 讯飞星火 | WebSocket |
| ZhiPuLLM | 智谱 AI | zhipuai SDK |
| BedrockLLM | AWS Bedrock | boto3 |
重试机制
OpenAILLM 使用 Tenacity 实现自动重试：
@retry( wait=wait_random_exponential(min=1, max=20), stop=stop_after_attempt(3), retry=retry_if_exception_type(APIConnectionError), after=after_log(logger, logger.level("WARNING").name), reraise=Exception,)async def _achat_completion_stream(self, messages, timeout, temperature) -> str: # 流式补全实现 ...- 指数退避：1-20 秒随机等待
- 最多 3 次重试
- 只重试
APIConnectionError（非业务错误）
成本追踪
每种模型类型关联对应的 CostManager：
def _select_costmanager(self, llm_config: LLMConfig) -> CostManager: if llm_config.api_type == LLMType.FIREWORKS: return FireworksCostManager() elif llm_config.api_type == LLMType.OPEN_LLM: return TokenCostManager() # 固定 token 计费 else: return self.cost_manager # 标准计费CostManager 内部维护：
class CostManager: total_cost: float = 0.0 max_budget: float = 100.0 # 按模型统计 token 消耗 def record_cost(self, model: str, prompt_tokens: int, completion_tokens: int): ...配置
~/.metagpt/config2.yaml：
llm: api_type: "openai" # 或 azure / ollama / groq 等 model: "gpt-4-turbo" # 或 gpt-3.5-turbo base_url: "https://api.openai.com/v1" api_key: "YOUR_API_KEY"代码中通过 Config.default() 加载：
class Config(BaseModel): llm: LLMConfig = Field(default_factory=LLMConfig)
@classmethod def default(cls) -> "Config": return cls.model_validate(read_yaml_file(CONFIG_ROOT / "config2.yaml"))问题与规避
max_tokens 参数名差异
- OpenAI 新模型使用
max_completion_tokens，旧模型和兼容 API 使用max_tokens get_max_completion_tokens()根据模型名判断使用哪个参数- 陷阱：自定义 OpenAI 兼容 API 可能不支持
max_completion_tokens
流式传输中断
- Tenacity 重试只覆盖
APIConnectionError，不覆盖流式传输中的中断 - 对策：
_achat_completion_stream()捕获asyncio.TimeoutError并返回已收到的内容
Rate Limit
- 不同提供商的 rate limit 策略不同
- 对策：指数退避重试（1-20s 随机）覆盖大多数 rate limit 场景
设计取舍
注册表 vs 工厂模式
- 注册表：
LLM_API_REGISTRY[api_type] = ProviderClass，查找 O(1) - 工厂：if-elif 链或 match-case，维护成本高
- MetaGPT 选择注册表，新提供商只需添加
@register_provider装饰器
OpenAI SDK 复用 vs 自定义客户端
- 兼容 OpenAI API 的提供商（Fireworks、Moonshot、Mistral 等）复用
OpenAILLM - 非兼容的提供商（Anthropic、Gemini、Spark 等）需要独立实现
- 好处：减少重复代码，新增兼容提供商只需配置变更
- 代价：OpenAILLM 内部需要处理不同提供商的边缘情况
参考来源
- 源码验证:
metagpt/provider/base_llm.py— BaseLLM 抽象类 - 源码验证:
metagpt/provider/openai_api.py:43— OpenAILLM 注册到 10+ 类型 - 源码验证:
metagpt/provider/llm_provider_registry.py— 注册表和工厂 - 源码验证:
metagpt/config2.py— Config 加载 - 源码验证:
metagpt/utils/cost_manager.py— CostManager 成本追踪