大模型调度服务开发指南

📚 学习目标

1. 理解什么是大模型调度服务及其在系统中的作用
2. 掌握策略模式和工厂模式的实际应用
3. 学会如何设计统一的服务接口
4. 了解如何集成多个第三方大模型服务

---

第一部分：理解大模型调度服务

1.1 什么是大模型调度服务？

在智能问答系统中，有很多功能需要用到大模型，但是不同的大模型在不同的领域可能有不同的优势，比如：

• 智能问答：可能GPT回答的较好
• 图片理解：Qwen-vl的系可能效果较好
• 写代码：可能claude的效果较好

当然，市面上有很多大模型服务商：

• 阿里云 (通义千问)
• 百度 (文心一言)
• 火山（豆包）…

问题来了：

• 如果你的代码里到处都是 call_openai()、call_qwen_vl()，当你想让不同的场景用不同的模型（比如简单问答用便宜的模型，复杂推理用强大的模型），代码就会很乱，或者如果某个模型服务挂了，怎么快速切换到备用模型？

大模型调度服务就是为了解决这些问题！

它的核心思想是：

应用层 → 不用关心具体调用哪个模型
         ↓
    统一调度服务 → 负责选择和调用合适的模型
         ↓
  各种具体的模型实现 (OpenAI、Qwen...)

---

1.2 为什么需要统一的接口？

想象一下，如果没有统一接口：

# 不好的做法 ❌
if model_type == "openai":
    response = openai.ChatCompletion.create(
        model="gpt-5",
        messages=[{"role": "user", "content": prompt}]
    )
    result = response.choices[0].message.content
elif model_type == "doubao":
    response = anthropic.messages.create(
        model="claude-3",
        messages=[{"role": "user", "content": prompt}]
    )
    result = response.content[0].text
elif model_type == "qwen":
    # 又是完全不同的调用方式...
    ...

这样的代码每次添加新模型，都要修改这段代码，到处都是 if-else，难以维护；不同模型的调用方式差异大，容易出错

有了统一接口后：无论底层是什么模型，调用方式都可以是一样的！

model_service = ModelScheduler.get_model("gpt-4")
result = model_service.chat(messages=[
    {"role": "user", "content": prompt}
])

---

第二部分：设计模式讲解

构建大模型调度服务，使用什么设计模式比较好呢。

2.1 策略模式 (Strategy Pattern)

什么是策略模式？

策略模式就像你选择交通工具：

• 你要从家到学校（目标）
• 可以骑自行车、坐公交、打车（不同策略）
• 但不管用哪种方式，最终都是"到达学校"（统一接口）

用代码来理解：

# 步骤1: 定义统一的接口（所有策略都要遵守）
class TransportStrategy:
    def go_to_school(self):
        pass  # 这是个抽象方法，子类必须实现

# 步骤2: 实现不同的策略
class BikeStrategy(TransportStrategy):
    def go_to_school(self):
        return "骑自行车去学校，需要30分钟"

class BusStrategy(TransportStrategy):
    def go_to_school(self):
        return "坐公交去学校，需要20分钟"

class TaxiStrategy(TransportStrategy):
    def go_to_school(self):
        return "打车去学校，需要10分钟"

# 步骤3: 使用策略
def go_to_school(strategy: TransportStrategy):
    print(strategy.go_to_school())

# 你可以轻松切换策略
go_to_school(BikeStrategy())   # 今天骑车
go_to_school(TaxiStrategy())   # 明天打车

应用到大模型服务：

# 统一接口
class BaseLLMProvider:
    def chat(self, messages):
        pass  # 所有模型提供商都要实现这个方法

# 不同的模型实现
class OpenAIProvider(BaseLLMProvider):
    def chat(self, messages):
        # 调用 OpenAI API
        ...

class QwenProvider(BaseLLMProvider):
    def chat(self, messages):
        # 调用 Qwen API
        ...

策略模式的好处：

• 每个模型的实现是独立的，互不影响
• 添加新模型时，只需要新增一个类，不用改旧代码
• 可以在运行时动态切换模型

---

2.2 工厂模式 (Factory Pattern)

什么是工厂模式？

工厂模式就像一个餐厅厨房：

• 你只需要告诉服务员"我要一份宫保鸡丁"
• 你不需要知道厨师是谁、怎么做的
• 厨房（工厂）负责创建这道菜

用代码来理解：

# 不用工厂模式，你需要自己创建对象
if dish == "宫保鸡丁":
    chef = ChineseChef()
    dish_obj = KungPaoChicken(chef, ingredients, cook_time)
elif dish == "意大利面":
    chef = ItalianChef()
    dish_obj = Pasta(chef, ingredients, cook_time)
# 很复杂！

# 用工厂模式，一行搞定
dish_obj = DishFactory.create(dish_name="宫保鸡丁")

应用到大模型服务：

class ModelFactory:
    @staticmethod
    def create_model(model_name: str):
        if model_name == "gpt-4":
            return OpenAIProvider(model="gpt-4", api_key="...")
        elif model_name == "claude-3":
            return ClaudeProvider(model="claude-3", api_key="...")
        elif model_name == "glm-4":
            return GLMProvider(model="glm-4", api_key="...")
        else:
            raise ValueError(f"不支持的模型: {model_name}")

# 使用时非常简单
model = ModelFactory.create_model("gpt-4")
response = model.chat(messages)

工厂模式的好处：

• 建对象的逻辑集中管理
• 调用者不需要知道对象是怎么创建的

---

2.3 两种模式如何配合？

策略模式 → 定义每个模型的行为（怎么调用 API）
         ↓
工厂模式 → 负责创建对应的模型实例
         ↓
    调度器 → 统一对外接口，协调工厂和策略

完整的调用流程：

# 1. 应用层请求
scheduler = ModelScheduler()
response = scheduler.chat(
    model_name="gpt-4",
    messages=[{"role": "user", "content": "什么是大语言模型？"}]
)

# 2. 调度器内部流程
# scheduler.chat() 内部会：
#   a. 用工厂模式创建 OpenAIProvider 实例
#   b. 调用 provider.chat() (策略模式的统一接口)
#   c. 返回统一格式的结果

这样设计的优势：

• 应用层代码非常简洁
• 添加新模型只需要：① 写一个新的 Provider 类，② 在工厂里注册
• 可以轻松实现其他功能（如负载均衡、灾备、费用统计等）

---

第三部分：需求分析与架构设计

3.1 功能需求

大模型调度服务需要支持的功能：

1. 基础对话功能

   • 单轮对话
   • 多轮对话（带上下文）
   • 流式输出（逐字显示）

2. 模型管理

   • 支持多个模型提供商
   • 动态切换模型
   • 配置管理（API Key、模型参数等）

3. 可靠性保障

   • 错误处理和重试机制
   • 超时控制
   • 日志记录

4. 扩展性

   • 轻松添加新的模型提供商
   • 支持自定义模型参数

---

3.3 核心类设计

3.3.1 BaseLLMProvider（基础提供商类）

这是所有模型提供商的父类，定义了统一接口：

from abc import ABC, abstractmethod
from typing import List, Dict, Optional, Generator

class BaseLLMProvider(ABC):
    """
    大语言模型提供商的基类
    
    为什么要用抽象基类(ABC)？
    - 强制子类实现必要的方法
    - 提供类型检查，确保所有提供商都遵循同一规范
    """
    
    def __init__(self, model: str, api_key: str, **kwargs):
        """
        初始化提供商
        
        Args:
            model: 模型名称，如 "gpt-4", "qwen2.5-vl"
            api_key: API密钥
            **kwargs: 其他配置参数（如温度、最大token数等）
        """
        self.model = model
        self.api_key = api_key
        self.config = kwargs
    
    @abstractmethod
    def chat(self, messages: List[Dict[str, str]], **kwargs) -> str:
        """
        同步对话接口
        
        Args:
            messages: 消息列表，格式如 [{"role": "user", "content": "你好"}]
            **kwargs: 额外参数（如temperature、max_tokens等）
        
        Returns:
            模型的回复文本
        """
        pass
    
    @abstractmethod
    def stream_chat(self, messages: List[Dict[str, str]], **kwargs) -> Generator[str, None, None]:
        """
        流式对话接口（逐字输出）
        
        为什么需要流式输出？
        - 用户体验更好，不用等待完整回复
        - 对于长文本生成，可以边生成边展示
        
        Returns:
            生成器，逐步返回文本片段
        """
        pass
    
    def _handle_error(self, error: Exception) -> str:
        """
        统一的错误处理
        
        返回友好的错误信息，而不是直接抛出异常
        """
        return f"调用{self.model}时出错: {str(error)}"

设计要点讲解：

1. 为什么用 ABC（抽象基类）？

   • 如果有人写了一个新的 Provider 但忘记实现 chat() 方法，程序会立即报错
   • 这比运行时才发现问题要好得多

2. 为什么 chat() 和 stream_chat() 都是抽象方法？

   • 这两个是核心功能，每个提供商必须实现
   • 保证了所有模型都能提供同样的功能

3. **为什么用 kwargs？

   • 不同模型有不同的参数，用 **kwargs 可以灵活传递这些参数

---

3.3.2 ModelConfig（配置管理类）

from dataclasses import dataclass
from typing import Optional

@dataclass
class ModelConfig:
    """
    模型配置数据类
    
    为什么用 dataclass？
    - 自动生成 __init__, __repr__ 等方法
    - 代码更简洁，可读性更好
    - 支持类型提示
    """
    
    model_name: str                    # 模型名称
    provider: str                      # 提供商（openai, claude等）
    api_key: str                       # API密钥
    base_url: Optional[str] = None     # API基础URL（有些服务需要自定义）
    temperature: float = 0.7           # 温度参数（控制随机性）
    max_tokens: int = 2048             # 最大生成token数
    timeout: int = 60                  # 请求超时时间（秒）
    
    def __post_init__(self):
        """
        初始化后的验证
        
        在对象创建后自动调用，用于参数校验
        """
        if not self.api_key:
            raise ValueError("API密钥不能为空")
        if self.temperature < 0 or self.temperature > 2:
            raise ValueError("temperature必须在0-2之间")

为什么需要配置类？

• 把配置集中管理，而不是散落在代码各处
• 方便从配置文件（如JSON、YAML）加载
• 类型安全，IDE可以提供更好的代码提示

---

3.4 项目目录结构

在开始编码之前，我们先规划好目录结构：

Scholar-AI/
├── src/
│   └── llm_service/              # 大模型调度服务
│       ├── __init__.py
│       ├── base.py               # BaseLLMProvider 基类
│       ├── config.py             # 配置管理
│       ├── factory.py            # ModelFactory 工厂类
│       ├── scheduler.py          # ModelScheduler 调度器
│       ├── providers/            # 各个模型提供商实现
│       │   ├── __init__.py
│       │   ├── doubao_provider.py
│       │   ├── qwen_provider.py
│       │   └── deepseek_provider.py
│       └── utils/                # 工具函数
│           ├── __init__.py
│           ├── logger.py         # 日志工具
│           └── retry.py          # 重试机制
├── config/
│   └── models.yaml               # 模型配置文件
├── tests/                        # 测试文件
│   └── test_llm_service.py
└── requirements.txt              # 依赖管理

---

第一阶段总结

到此为止，我们目前学习了：

1. 概念理解：什么是大模型调度服务，为什么需要它
2. 设计模式：策略模式和工厂模式的原理与应用
3. 核心类设计：BaseLLMProvider 和 ModelConfig 的设计思路

---

如果想继续下一部分的学习，可以加入我们的群聊哦，国庆节后会有视频推出，企鹅群号：1062959816