1. 项目缘起为什么我们需要一个“Provider抽象层”如果你正在尝试构建自己的AI Agent或者已经用LangChain、Dify这类框架做过一些实验那么“和LLM对话”这件事对你来说可能既熟悉又充满挫败感。熟悉的是无非就是调用一个API把用户的问题Prompt发过去然后等待模型返回结果。但挫败感往往接踵而至今天想试试Claude发现它的API参数命名和OpenAI完全不同明天公司要求接入一个本地部署的Qwen模型它的调用方式又是另一套后天你写的代码在某个云服务商那里因为网络问题超时了你想换个备用的模型提供商却发现整个Agent的核心逻辑和某个具体的SDK绑得死死的改起来牵一发而动全身。这就是我们今天要解决的问题。在“从零到一手撸Agent”系列的第二篇我们不急着去设计复杂的思维链或工具调用而是先扎扎实实地打好地基——构建一个LLM Provider抽象层。这个抽象层的核心目标是让你写的Agent核心业务逻辑完全不用关心背后具体是和GPT-4对话还是和Claude、Gemini抑或是本地跑的Qwen、ChatGLM对话。就像你家里的电器插头不管是插在国标的插座上还是通过转换器插在美标的插座上电器本身都能正常工作。最近在开发者社区里经常能看到类似这样的错误信息Claude显示your connection works, but the provider rejected a test request. often a model-access or quota issue.或者是qwen provider returned error: access to private networks is not allowed甚至是更通用的The model provider failed after retries。这些错误背后反映的正是与不同模型服务商Provider对接时的复杂性和差异性。一个健壮的Agent系统必须能优雅地处理这些差异而不是让业务代码被各种Provider特有的错误码和异常所淹没。因此这篇内容将带你从零开始设计并实现一个简洁、灵活、可扩展的Provider抽象层。我们会深入探讨其背后的设计模式核心就是策略模式并手把手实现一个具备重试、熔断、负载均衡等生产级特性的基础版本。当你完成这个抽象层后你的Agent就获得了第一项超能力模型无关性。这意味着你可以根据成本、性能、政策要求随时无缝切换底层的LLM而你的Agent大脑完全感知不到变化。2. 核心设计用策略模式解耦业务与具体实现在深入代码之前我们必须先想清楚设计。为什么是“策略模式”Strategy Pattern我们可以用一个更生活的例子来理解想象你是一个快递公司的调度中心Agent的核心逻辑。你需要把包裹用户请求送到客户手中。你有多种运输策略顺丰快递、中通快递、京东物流甚至自己公司的车队。调度中心不应该关心顺丰的快递员是骑电动车还是开货车也不应该关心中通的网点如何分拣。调度中心只定义标准接口“把这个包裹送到这个地址并返回签收结果”。至于具体是哪家物流公司、怎么送、用什么规则比如偏远地区加价那是具体“物流策略”LLM Provider要实现的。在我们的LLM对话场景中这个“标准接口”就是Provider抽象层。它定义了所有LLM提供商都必须实现的一组方法主要是generate生成文本和generate_stream流式生成。Agent的核心调度器只需要调用这个接口至于背后是调用OpenAI的/v1/chat/completions端点还是调用Anthropic的/v1/messages端点或是通过一个HTTP客户端调用本地模型的API调度器一概不知。2.1 定义抽象基类契约先行一切从定义一个“契约”开始。我们将创建一个抽象基类Abstract Base Class, ABC它规定了成为一个LLM Provider必须满足的最低要求。from abc import ABC, abstractmethod from typing import AsyncGenerator, Dict, Any, Optional, List from pydantic import BaseModel class LLMResponse(BaseModel): LLM响应的标准化数据结构 content: str # 模型返回的文本内容 model: str # 实际使用的模型名称 usage: Optional[Dict[str, int]] None # 令牌使用情况如 {prompt_tokens: 100, completion_tokens: 50} finish_reason: Optional[str] None # 停止原因如 stop, length raw_response: Optional[Dict[str, Any]] None # 原始响应用于调试 class Message(BaseModel): 对话消息的标准结构 role: str # system, user, assistant content: str class BaseLLMProvider(ABC): LLM提供商的抽象基类 def __init__(self, model_name: str, **kwargs): self.model_name model_name self.config kwargs # 保存各Provider特有的配置如api_key, base_url等 abstractmethod async def generate( self, messages: List[Message], **generation_params ) - LLMResponse: 同步生成文本的核心方法。 :param messages: 对话历史消息列表 :param generation_params: 生成参数如temperature, max_tokens等 :return: 标准化的LLMResponse对象 pass abstractmethod async def generate_stream( self, messages: List[Message], **generation_params ) - AsyncGenerator[str, None]: 流式生成文本的核心方法。 :param messages: 对话历史消息列表 :param generation_params: 生成参数 :return: 异步生成器逐个yield tokens pass abstractmethod async def health_check(self) - bool: 健康检查用于判断该Provider当前是否可用。 例如检查API密钥是否有效、网络是否可达、额度是否充足。 pass为什么这么设计使用Pydantic的BaseModelLLMResponse和Message使用Pydantic能自动进行类型验证和序列化。这比使用普通的字典更安全、更清晰。当你的Agent系统越来越复杂在不同模块间传递数据时强类型定义能避免很多低级错误。分离raw_response将原始响应保存在raw_response字段是一个非常重要的设计。在调试时尤其是处理那些令人头疼的provider rejected错误时你能直接看到API返回的原始错误信息而不是被我们抽象层过滤掉的关键细节。health_check方法这是实现智能路由和熔断的基础。一个Provider可能因为网络问题如cannot reach 127.0.0.1:15721、配额问题error code: 429或配置错误no inference provider configured而不可用。在将请求分发给它之前先做一次快速检查可以避免不必要的失败和等待。2.2 实现具体策略以OpenAI和Qwen为例定义了契约接下来就是让不同的“物流公司”来签约并实现它。我们以OpenAI官方格式和通义千问Qwen一种常见开源模型格式为例。OpenAI Provider 实现import aiohttp from typing import AsyncGenerator import logging logger logging.getLogger(__name__) class OpenAIProvider(BaseLLMProvider): OpenAI兼容接口的Provider实现 def __init__(self, model_name: str, api_key: str, base_url: str https://api.openai.com/v1): super().__init__(model_name, api_keyapi_key, base_urlbase_url) # 注意这里可以灵活支持OpenAI官方、Azure OpenAI或其他兼容此API的服务如某些本地部署模型 self.base_url base_url.rstrip(/) self.headers { Authorization: fBearer {api_key}, Content-Type: application/json } async def generate(self, messages: List[Message], **generation_params) - LLMResponse: url f{self.base_url}/chat/completions payload { model: self.model_name, messages: [msg.dict() for msg in messages], **generation_params # 将temperature, max_tokens等参数合并进来 } async with aiohttp.ClientSession() as session: try: async with session.post(url, jsonpayload, headersself.headers) as resp: resp.raise_for_status() data await resp.json() # 将OpenAI的响应格式转换为我们标准格式 choice data[choices][0] return LLMResponse( contentchoice[message][content], modeldata[model], usagedata.get(usage), finish_reasonchoice.get(finish_reason), raw_responsedata # 保存原始数据 ) except aiohttp.ClientError as e: logger.error(fOpenAI API请求失败: {e}) # 这里可以抛出自定义的业务异常便于上层统一处理 raise ProviderError(fOpenAI服务调用失败: {e}) from e async def generate_stream(self, messages: List[Message], **generation_params) - AsyncGenerator[str, None]: # 流式实现逻辑类似但需要处理Server-Sent Events (SSE) # 为简洁起见此处省略具体流式代码但原理是设置 streamTrue 并逐块解析返回的data pass async def health_check(self) - bool: 调用OpenAI的models端点验证API密钥和网络连通性 url f{self.base_url}/models async with aiohttp.ClientSession() as session: try: async with session.get(url, headersself.headers, timeout5) as resp: return resp.status 200 except (aiohttp.ClientError, asyncio.TimeoutError): return FalseQwen Provider 实现假设其API类似但略有不同class QwenProvider(BaseLLMProvider): 通义千问或类似风格API的Provider实现 def __init__(self, model_name: str, api_key: str, base_url: str): super().__init__(model_name, api_keyapi_key, base_urlbase_url) # Qwen的API可能要求不同的认证头例如 Authorization: Bearer {api_key} 或 X-API-Key: {api_key} self.headers { X-API-Key: api_key, # 注意这里和OpenAI不同 Content-Type: application/json } async def generate(self, messages: List[Message], **generation_params) - LLMResponse: url f{self.base_url}/generate # 注意端点路径可能不同不是 /chat/completions # Qwen的请求体格式也可能不同需要适配 payload { model: self.model_name, input: self._format_messages(messages), # 需要自定义一个消息格式化方法 parameters: generation_params # 生成参数可能放在嵌套的parameters字段里 } async with aiohttp.ClientSession() as session: try: async with session.post(url, jsonpayload, headersself.headers) as resp: resp.raise_for_status() data await resp.json() # 将Qwen的响应格式转换为我们标准格式 return LLMResponse( contentdata[output][text], # 字段名可能不同 modeldata.get(model, self.model_name), usagedata.get(usage), finish_reasondata.get(finish_reason), raw_responsedata ) except aiohttp.ClientError as e: logger.error(fQwen API请求失败: {e}) raise ProviderError(fQwen服务调用失败: {e}) from e def _format_messages(self, messages: List[Message]) - str: 将Message列表格式化为Qwen API所需的单一字符串Prompt # 这是一个简化的示例实际格式可能更复杂 formatted [] for msg in messages: formatted.append(f{msg.role}: {msg.content}) return \n.join(formatted) async def health_check(self) - bool: Qwen可能有一个特定的健康检查端点或者我们简单调用一次生成 # 方法1调用一个轻量级端点 # 方法2如果没有可以尝试一个极短的生成请求如max_tokens1来测试 # 这里采用方法2的简化版检查网络连通性 url self.base_url async with aiohttp.ClientSession() as session: try: async with session.get(url, timeout3) as resp: return resp.status 500 except (aiohttp.ClientError, asyncio.TimeoutError): return False关键点解析与避坑经验差异点封装注意两个实现类的__init__、请求payload构造、响应解析data[output][text] vs data[choices][0][message][content]以及health_check逻辑都不同。抽象层的价值就在于把这些差异全部封装在具体的Provider类内部。Agent业务代码看到的永远只是await provider.generate(messages, temperature0.7)。错误处理标准化我们定义了一个自定义的ProviderError异常。这样无论底层是OpenAI返回429还是Qwen返回网络错误上层业务逻辑都可以用try...except ProviderError来统一捕获和处理而不是写一堆if provider_type openai的条件判断。配置管理api_key,base_url等敏感或可变的配置通过__init__传入并保存在self.config或实例变量中。绝对不要硬编码在代码里。在实际项目中这些配置应该来自环境变量或配置中心。3. 进阶实现让抽象层具备生产级韧性一个只能简单调用的抽象层是远远不够的。在生产环境中我们需要它足够健壮能够应对网络抖动、服务限流、临时故障等常见问题。这就需要我们在抽象层之上再封装一层“智能管理器”。3.1 实现带重试和熔断的Provider包装器我们可以设计一个ResilientProvider类它包装一个具体的Provider实例并为其增加重试、熔断和回退策略。import asyncio import time from typing import Optional from circuitbreaker import circuit # 可以使用 circuitbreaker 库 class ResilientProvider: 为Provider增加弹性的包装器 def __init__(self, provider: BaseLLMProvider, max_retries: int 3): self.provider provider self.max_retries max_retries self.failure_count 0 self.last_failure_time: Optional[float] None self.circuit_open False self.circuit_open_until: Optional[float] None async def generate_with_retry(self, messages: List[Message], **kwargs) - LLMResponse: 带指数退避重试的生成方法 last_exception None for attempt in range(self.max_retries 1): # 尝试次数 重试次数 1 # 检查熔断器状态 if self.circuit_open: if self.circuit_open_until and time.time() self.circuit_open_until: raise ProviderError(f熔断器已开启跳过对 {self.provider.model_name} 的调用) else: # 熔断时间已过尝试半开状态 self.circuit_open False try: # 实际调用被包装的provider response await self.provider.generate(messages, **kwargs) # 调用成功重置失败计数 self._record_success() return response except (ProviderError, aiohttp.ClientError) as e: last_exception e self._record_failure() logger.warning(fProvider {self.provider.model_name} 第{attempt1}次调用失败: {e}) if attempt self.max_retries: # 计算指数退避的等待时间 wait_time (2 ** attempt) (random.random() * 0.1) # 加上一点随机性防止惊群 logger.info(f等待 {wait_time:.2f} 秒后重试...) await asyncio.sleep(wait_time) else: # 重试次数用尽判断是否触发熔断 if self.failure_count 5: # 连续失败5次 self._trip_circuit() raise ProviderError(f在{self.max_retries}次重试后仍失败: {last_exception}) from last_exception def _record_success(self): 记录成功重置失败状态 self.failure_count 0 self.last_failure_time None if self.circuit_open: logger.info(f熔断器对 {self.provider.model_name} 进入半开状态并测试成功关闭熔断) self.circuit_open False def _record_failure(self): 记录失败 self.failure_count 1 self.last_failure_time time.time() def _trip_circuit(self): 触发熔断 self.circuit_open True # 熔断30秒 self.circuit_open_until time.time() 30 logger.error(fProvider {self.provider.model_name} 因连续失败被熔断30秒内不再尝试)这个包装器解决了什么问题网络瞬断与抖动通过重试机制短暂的网络问题如cannot reach 127.0.0.1不会导致单次请求失败。服务限流429错误当遇到error code: 429 - {error: {message: the engine is currently overloaded}}这类错误时指数退避的重试策略能有效避免加重服务端压力并提高最终成功的概率。服务端故障如果某个Provider完全不可用如本地模型服务崩溃熔断机制能快速将其标记为“故障”避免后续请求继续发往该故障节点浪费资源和时间同时给服务端恢复的时间。30秒后进入“半开”状态尝试放行一个请求如果成功则关闭熔断。3.2 实现多Provider的负载均衡与路由当你的系统配置了多个同类型或不同类型的Provider时例如两个不同的OpenAI API密钥或者一个OpenAI加一个Claude一个更高级的抽象层可以实现简单的负载均衡或路由策略。class ProviderRouter: Provider路由器支持负载均衡和故障转移 def __init__(self, providers: List[BaseLLMProvider], strategy: str round_robin): :param providers: 可用的Provider列表 :param strategy: 路由策略可选 round_robin, random, fallback self.providers [ResilientProvider(p) for p in providers] # 每个Provider都包装上弹性层 self.strategy strategy self.current_index 0 # 用于轮询 async def generate(self, messages: List[Message], **kwargs) - LLMResponse: if self.strategy round_robin: return await self._round_robin_generate(messages, **kwargs) elif self.strategy fallback: return await self._fallback_generate(messages, **kwargs) elif self.strategy random: return await self._random_generate(messages, **kwargs) else: raise ValueError(f不支持的策略: {self.strategy}) async def _round_robin_generate(self, messages: List[Message], **kwargs) - LLMResponse: 轮询策略依次使用每个Provider for _ in range(len(self.providers)): provider self.providers[self.current_index] self.current_index (self.current_index 1) % len(self.providers) try: # 健康检查快速过滤 if not await provider.provider.health_check(): continue return await provider.generate_with_retry(messages, **kwargs) except ProviderError: continue # 当前Provider失败尝试下一个 raise ProviderError(所有Provider均不可用) async def _fallback_generate(self, messages: List[Message], **kwargs) - LLMResponse: 故障转移策略按列表顺序尝试直到成功 for provider in self.providers: try: if not await provider.provider.health_check(): continue return await provider.generate_with_retry(messages, **kwargs) except ProviderError: continue raise ProviderError(所有Provider均不可用)使用场景与心得成本优化你可以配置一个GPT-4 Provider和一个GPT-3.5-Turbo Provider。在ProviderRouter中使用fallback策略优先尝试GPT-3.5如果它的回答置信度不高这需要额外的逻辑判断再fallback到更强大但更贵的GPT-4。提升可用性接入多个同质化的Provider如两个不同账号的OpenAI使用round_robin策略既能分散请求负载也能在一个Provider出问题时自动切换到另一个。合规与数据主权某些业务可能要求特定类型的数据必须由部署在本地或特定区域的模型处理。你可以在路由逻辑中加入基于请求内容如语言、主题的判断将请求路由到对应的Provider。4. 集成与实战在Agent框架中使用抽象层现在我们已经有了一个功能完备的Provider抽象层。如何将它集成到你的Agent项目中呢关键在于你的Agent核心逻辑比如规划器、工具执行器、记忆模块不应该直接实例化某个具体的Provider而应该依赖于一个“工厂”或“配置”来获取Provider实例。4.1 使用工厂模式创建Providerclass LLMProviderFactory: Provider工厂根据配置创建对应的Provider实例 _providers: Dict[str, BaseLLMProvider] {} classmethod def register_provider(cls, name: str, provider_class): 注册Provider类可用于动态扩展 cls._providers[name] provider_class classmethod def create_provider(cls, provider_config: Dict) - BaseLLMProvider: 根据配置字典创建Provider。 配置示例 { type: openai, model_name: gpt-4, api_key: sk-..., base_url: https://api.openai.com/v1 } provider_type provider_config.pop(type) model_name provider_config.pop(model_name) if provider_type not in cls._providers: raise ValueError(f未注册的Provider类型: {provider_type}) provider_class cls._providers[provider_type] return provider_class(model_namemodel_name, **provider_config) # 在应用启动时注册已知的Provider LLMProviderFactory.register_provider(openai, OpenAIProvider) LLMProviderFactory.register_provider(qwen, QwenProvider) # 未来可以轻松扩展LLMProviderFactory.register_provider(claude, ClaudeProvider)4.2 在Agent核心中注入LLM能力假设我们有一个最简单的Agent类它的核心是chat方法。class SimpleAgent: 一个简单的对话Agent def __init__(self, llm_provider: BaseLLMProvider, system_prompt: str 你是一个有用的助手。): self.llm_provider llm_provider self.system_prompt system_prompt self.conversation_history: List[Message] [ Message(rolesystem, contentsystem_prompt) ] async def chat(self, user_input: str) - str: 处理用户输入返回Agent的回复 # 1. 将用户输入加入历史 self.conversation_history.append(Message(roleuser, contentuser_input)) # 2. 调用抽象层Agent核心不关心底层是哪个Provider try: response await self.llm_provider.generate( messagesself.conversation_history, temperature0.8, max_tokens500 ) except ProviderError as e: # 统一处理所有Provider错误 return f抱歉思考引擎暂时出了点问题{e} # 3. 将助手回复加入历史 assistant_reply response.content self.conversation_history.append(Message(roleassistant, contentassistant_reply)) # 4. 可选维护历史长度防止超出上下文窗口 if len(self.conversation_history) 20: # 保留系统提示和最近对话 self.conversation_history [self.conversation_history[0]] self.conversation_history[-19:] return assistant_reply启动你的Agentimport asyncio async def main(): # 从配置或环境变量读取 config { type: openai, # 想换Qwen只需改成 qwen model_name: gpt-3.5-turbo, api_key: os.getenv(OPENAI_API_KEY), base_url: https://api.openai.com/v1 } # 通过工厂创建Provider llm_provider LLMProviderFactory.create_provider(config) # 如果需要韧性可以包装一下 resilient_provider ResilientProvider(llm_provider, max_retries2) # 如果需要多Provider路由可以这样 # config2 {...} # 另一个Provider配置 # llm_provider2 LLMProviderFactory.create_provider(config2) # router ProviderRouter([llm_provider, llm_provider2], strategyround_robin) # 创建Agent注入Provider agent SimpleAgent(llm_providerresilient_provider) # 或 agent SimpleAgent(llm_providerrouter) # 开始对话 reply await agent.chat(你好介绍一下你自己。) print(fAgent: {reply}) if __name__ __main__: asyncio.run(main())4.3 处理那些“烦人”的错误让我们回到文章开头提到的那些错误看看有了抽象层之后如何更优雅地处理Claude显示your connection works, but the provider rejected a test request. often a model-access or quota issue.这通常意味着API密钥无效、没有权限或额度用完。在我们的架构里ClaudeProvider的health_check方法应该实现更精确的检查比如调用一个轻量的权限验证接口。如果检查失败ProviderRouter在_fallback_generate或_round_robin_generate中会跳过这个Provider尝试列表中的下一个。qwen provider returned error: access to private networks is not allowed这是一个网络策略错误。QwenProvider在请求时可能会抛出包含此错误信息的ProviderError。ResilientProvider的重试机制对于这种网络层面的错误通常是无效的重试会立刻失败。更好的做法是在QwenProvider的初始化或health_check中就尝试进行一次内网地址的连通性测试如果失败则直接标记该Provider为不健康避免业务请求打到它上面。The model provider failed after retries. I kept raw provider details...这正是我们设计LLMResponse.raw_response字段和自定义ProviderError的意义。当最终重试都失败后我们可以将最后一个异常抛出。上层业务如一个Web API接口可以捕获这个异常并将raw_response中的详细信息记录到日志或返回给管理员便于精准定位是哪个Provider、因为什么原因失败。(model provider error code: 1305, http status: 429)这是典型的速率限制错误。ResilientProvider的指数退避重试策略正是为应对此类错误而生。在收到429状态码时重试逻辑会等待一段时间再试这比立即重试或无限重试要友好和有效得多。5. 扩展与展望抽象层的更多可能性我们构建的这个抽象层已经是一个功能完整、可用于生产环境的核心组件了。但它的潜力远不止于此。基于这个设计你可以轻松地进行横向和纵向扩展横向扩展支持更多Provider实现一个ClaudeProvider适配Anthropic的API格式。实现一个OllamaProvider用于连接本地运行的Ollama服务。实现一个AzureOpenAIProvider虽然与OpenAI兼容但可能在认证API Key vs. Azure Active Directory和端点路径上略有不同。实现一个MockProvider用于单元测试返回预设的响应避免在测试中调用真实API。纵向扩展增强抽象层能力上下文窗口管理在BaseLLMProvider中增加一个max_context_tokens属性并在generate方法内部自动处理历史消息的截断或总结确保不会超出模型限制。Tokenizer集成集成各模型对应的tokenizer如tiktoken for OpenAItransformers for HuggingFace models在发送请求前精确计算token数量并给出提示或自动优化。请求与响应钩子Hooks在generate方法前后加入钩子机制方便实现日志记录、性能监控、敏感信息过滤、Prompt注入检测等功能。成本计算根据LLMResponse.usage和预设的单价如GPT-4每1000 tokens多少钱自动计算本次调用的成本并累计。与现有生态集成兼容LangChain你可以写一个LangChainCompatProvider它内部封装一个LangChain的LLMChain或ChatModel。这样你既可以利用LangChain丰富的周边生态工具、记忆、索引又能统一到你自己的抽象层管理之下。作为Dify或FastAPI项目的组件将你的ProviderRouter作为一个服务注入到Dify的Workflow中或者作为一个FastAPI的依赖项为不同的API路由提供不同策略的LLM调用能力。构建一个坚实的Provider抽象层看似是前期投入的“苦活”但它为你后续的Agent开发铺平了道路。当你想试验一个新模型时你不再需要重构业务代码只需要实现一个新的Provider类并注册到工厂中。当某个云服务出现故障时你的Agent可以依靠路由和熔断机制优雅地降级或切换保障服务的连续性。这份灵活性与鲁棒性正是专业AI应用与业余脚本之间的一道分水岭。