# 1858 LLM Interface E8e67950 > LLM 接口层 - Skill: `tools-only/1858-llm-interface-e8e67950` (Agent Skill, multi-file: 3 files) - Install (CLI): `npx skillmds@latest add tools-only/1858-llm-interface-e8e67950` - Raw SKILL.md: https://api.skillmd.com/api/skills/tools-only/1858-llm-interface-e8e67950/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - Author: tools-only (https://skillmd.com/u/tools-only) - Updated: 2026-09-22 - Page: https://skillmd.com/skills/tools-only/1858-llm-interface-e8e67950 --- # LLM 接口层 本文档介绍 SimpleLLMFunc 的 LLM 接口层设计，包括接口抽象、密钥管理和流量控制等核心组件。 ## 概述 SimpleLLMFunc 提供了完整的 LLM 接口框架，旨在统一不同 LLM 服务提供商的调用方式。该框架包含三个核心组件： 1. **LLM_Interface** - 抽象基类，定义标准接口规范 2. **APIKeyPool** - API 密钥池，实现负载均衡 3. **TokenBucket** - 令牌桶算法，提供流量控制 ## 快速开始 ### 最简单的使用方式 ```python from SimpleLLMFunc import OpenAICompatible # 从配置文件加载 LLM 接口 llm = OpenAICompatible.load_from_json_file("provider.json")["openai"]["gpt-3.5-turbo"] # 现在可以在装饰器中使用 from SimpleLLMFunc import llm_function @llm_function(llm_interface=llm) async def my_task(text: str) -> str: """处理文本任务""" pass ``` ### 配置文件格式 创建 `provider.json` 文件： ```json { "openai": { "gpt-3.5-turbo": { "api_keys": ["sk-key1", "sk-key2"], "base_url": "https://api.openai.com/v1", "model": "gpt-3.5-turbo", "max_retries": 5, "retry_delay": 1.0, "rate_limit_capacity": 20, "rate_limit_refill_rate": 3.0 }, "gpt-4": { "api_keys": ["sk-key3"], "base_url": "https://api.openai.com/v1", "model": "gpt-4" } }, "deepseek": { "v3-turbo": { "api_keys": ["your-deepseek-key"], "base_url": "https://api.deepseek.com/v1", "model": "deepseek-chat" } } } ``` ## LLM_Interface 基类 ### 设计理念 `LLM_Interface` 是所有 LLM 实现的抽象基类，定义了统一的接口规范。所有具体实现（如 `OpenAICompatible`）都继承自这个类。 ### 核心特性 - **标准化接口**: 定义了 `chat()` 和 `chat_stream()` 两个核心方法 - **类型安全**: 使用 Python 类型注解确保类型正确性 - **异步原生**: 完全基于异步编程，支持高并发 - **可扩展性**: 通过继承轻松添加新的 LLM 服务支持 ### 接口定义 ```python from abc import ABC, abstractmethod from typing import Dict, Any, AsyncGenerator, List class LLM_Interface(ABC): """LLM 接口抽象基类""" @abstractmethod async def chat( self, messages: List[Dict[str, str]], stream: bool = False, timeout: int | None = None, **kwargs ) -> Dict[str, Any]: """ 执行非流式对话请求 Args: messages: 消息列表，每条消息为 {"role": "user"|"assistant"|"system", "content": "..."} stream: 流式标志（False 表示非流式） timeout: 请求超时时间（秒） **kwargs: 其他 LLM 参数（temperature、top_p 等） Returns: LLM 响应数据 """ pass @abstractmethod async def chat_stream( self, messages: List[Dict[str, str]], stream: bool = True, timeout: int | None = None, **kwargs ) -> AsyncGenerator[Dict[str, Any], None]: """ 执行流式对话请求 Args: messages: 消息列表 stream: 流式标志（True 表示流式） timeout: 请求超时时间（秒） **kwargs: 其他 LLM 参数 Yields: LLM 响应数据块 """ if False: yield {} ``` ## OpenAICompatible 实现 ### 什么是 OpenAICompatible `OpenAICompatible` 是 `LLM_Interface` 的具体实现，支持任何兼容 OpenAI API 格式的服务，包括： - OpenAI (GPT-4, GPT-3.5) - Deepseek - Anthropic Claude - 火山引擎 Ark - 百度千帆 - 本地 LLM (Ollama, vLLM) - 任何兼容 OpenAI API 的服务 ### 创建接口 #### 方式 1: 从配置文件加载（推荐） ```python from SimpleLLMFunc import OpenAICompatible # 从 JSON 配置文件加载 all_models = OpenAICompatible.load_from_json_file("provider.json") # 访问具体的模型 gpt35 = all_models["openai"]["gpt-3.5-turbo"] gpt4 = all_models["openai"]["gpt-4"] deepseek = all_models["deepseek"]["v3-turbo"] ``` #### 方式 2: 直接创建 ```python from SimpleLLMFunc import OpenAICompatible, APIKeyPool # 创建 API 密钥池 key_pool = APIKeyPool( api_keys=["sk-key1", "sk-key2", "sk-key3"], provider_id="openai-gpt35" ) # 创建 LLM 接口 llm = OpenAICompatible( api_key_pool=key_pool, base_url="https://api.openai.com/v1", model="gpt-3.5-turbo", max_retries=5, retry_delay=1.0, rate_limit_capacity=20, rate_limit_refill_rate=3.0 ) ``` ### 核心特性 - **自动重试**: 请求失败时自动重试（可配置次数） - **令牌统计**: 自动统计 input 和 output token 数量 - **流量控制**: 使用令牌桶算法防止速率限制 - **密钥轮换**: 自动在多个密钥间负载均衡 ## APIKeyPool - 密钥管理 ### 设计理念 `APIKeyPool` 使用**小根堆**数据结构实现 API 密钥的负载均衡，确保请求均匀分布到不同的密钥上。 ### 核心特性 - **小根堆轮换**: 自动选择负载最低的密钥 - **负载均衡**: 实时跟踪每个密钥的任务数量 - **线程安全**: 使用锁保护并发访问 - **单例模式**: 相同 provider_id 的密钥池共享状态 ### 工作原理 ``` 堆结构：[(任务数, API密钥), ...] 堆顶永远是任务数最少的密钥 示例： heap = [(0, "key1"), (1, "key2"), (2, "key3")] 获取最低负载密钥 → "key1" key1 任务完成后重新整理堆... ``` ### 使用示例 ```python from SimpleLLMFunc.interface import APIKeyPool # 创建密钥池 key_pool = APIKeyPool( api_keys=["sk-key1", "sk-key2", "sk-key3"], provider_id="my-provider" ) # 获取最低负载的密钥 key = key_pool.get_least_loaded_key() # 任务开始时增加计数 key_pool.increment_task_count(key) # 任务完成后减少计数 key_pool.decrement_task_count(key) ``` ## TokenBucket - 流量控制 ### 设计理念 `TokenBucket` 实现经典的**令牌桶算法**，用于 API 请求的流量控制和速率限制。 ### 算法原理 令牌桶工作原理： 1. **令牌生成**: 以固定速率向桶中添加令牌 2. **令牌消费**: 每次请求消费一个或多个令牌 3. **容量限制**: 桶有最大容量，多余令牌会被丢弃 4. **突发支持**: 当桶中有足够令牌时，允许突发请求 示例： ``` 容量=5，补充速率=2 tokens/秒 时间: 0s 1s 2s 3s 桶: [●●●●●] [●●●●●] [●●●●●] [●●●] 请求: 0个 0个 2个 2个 ``` ### 核心特性 - **平滑流量**: 避免突发请求冲击后端 API - **可配置参数**: 支持自定义容量和补充速率 - **异步支持**: 非阻塞的令牌获取，支持超时 - **线程安全**: 统一的锁保护所有操作 ### 配置参数 | 参数 | 类型 | 说明 | 推荐值 | |------|------|------|--------| | `capacity` | int | 令牌桶容量（最大令牌数） | 10-50 | | `refill_rate` | float | 令牌补充速率（tokens/秒） | 0.5-5.0 | ### 使用场景 ```python # 高频 API（如 OpenAI） capacity=20, refill_rate=3.0 # 标准 API capacity=10, refill_rate=1.0 # 受限 API（如百度千帆） capacity=5, refill_rate=0.5 ``` ## 完整的生产级示例 ### 1. 多模型配置 ```python from SimpleLLMFunc import OpenAICompatible, llm_function # 加载多个模型 models = OpenAICompatible.load_from_json_file("provider.json") # 定义不同的任务用不同的模型 fast_llm = models["openai"]["gpt-3.5-turbo"] # 快速、便宜 powerful_llm = models["openai"]["gpt-4"] # 强大、昂贵 deepseek_llm = models["deepseek"]["v3-turbo"] # 国内服务 @llm_function(llm_interface=fast_llm) async def simple_task(text: str) -> str: """简单任务使用快速模型""" pass @llm_function(llm_interface=powerful_llm) async def complex_task(text: str) -> str: """复杂任务使用强大模型""" pass ``` ### 2. 错误处理和重试 ```python import asyncio from SimpleLLMFunc import llm_function @llm_function(llm_interface=llm) async def robust_call(text: str) -> str: """可靠的 LLM 调用""" pass async def call_with_fallback(): """带备选方案的调用""" try: result = await robust_call("test") return result except Exception as e: print(f"主模型失败: {e}") # 使用备选模型 return await backup_call("test") ``` ### 3. 监控和调试 ```python # 检查 token 使用量 print(f"输入 tokens: {llm.input_token_count}") print(f"输出 tokens: {llm.output_token_count}") # 检查密钥池状态 least_loaded = llm.key_pool.get_least_loaded_key() print(f"最低负载密钥: {least_loaded}") # 检查令牌桶状态 print(f"可用令牌: {llm.token_bucket.tokens}") ``` ## 最佳实践 ### 1. 密钥管理 ```python # ✅ 推荐：为每个项目/环境配置独立的密钥 { "openai": { "prod": { "api_keys": ["prod-key1", "prod-key2"], "max_retries": 5 }, "dev": { "api_keys": ["dev-key1"], "max_retries": 2 } } } # ✅ 推荐：为不同模型配置不同的速率限制 { "openai": { "gpt-4": { "rate_limit_capacity": 10, # 小容量，保守 "rate_limit_refill_rate": 1.0 }, "gpt-3.5-turbo": { "rate_limit_capacity": 50, # 大容量，激进 "rate_limit_refill_rate": 5.0 } } } ``` ### 2. 流量控制 ```python # ✅ 推荐：根据 API 限制调整参数 # OpenAI：rate_limit_capacity=10-20, refill_rate=1.0-3.0 # Deepseek：rate_limit_capacity=10-15, refill_rate=2.0-3.0 # 本地 LLM：capacity=100, refill_rate=10.0（几乎无限） ``` ### 3. 错误处理 ```python import asyncio from typing import Optional async def call_with_exponential_backoff( llm_call, max_retries: int = 3, base_delay: float = 1.0 ) -> Optional[str]: """带指数退避的重试""" for attempt in range(max_retries): try: return await llm_call() except Exception as e: if attempt == max_retries - 1: raise delay = base_delay * (2 ** attempt) print(f"第 {attempt + 1} 次失败，等待 {delay} 秒后重试...") await asyncio.sleep(delay) ``` ## 故障排除 ### 常见问题 1. **"Rate limit exceeded" 错误** - 增加 `rate_limit_capacity` 或 `rate_limit_refill_rate` - 检查配置中的速率限制是否与 API 限制匹配 2. **密钥持续失败** - 检查 API 密钥是否有效且有足够配额 - 验证 `base_url` 是否正确 3. **Token 统计不准确** - 某些 API 可能不返回 token 统计信息 - 框架会尽力估算，但可能不完全准确 ### 调试技巧 ```python import logging # 启用详细日志 logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger("SimpleLLMFunc") logger.setLevel(logging.DEBUG) # 查看请求/响应详情 ``` ## 总结 SimpleLLMFunc 的接口层通过以下设计提供强大的 LLM 服务： 1. **LLM_Interface**: 统一的抽象基类 2. **OpenAICompatible**: 开箱即用的 OpenAI 兼容实现 3. **APIKeyPool**: 智能的密钥负载均衡 4. **TokenBucket**: 可靠的流量控制 这种设计既保证了易用性，又提供了企业级的功能和可靠性。