第 17 章：接入一个新模型

发表于 更新于 分类于 阅读次数：
实战开发：将 DeepSeek 和自定义模型接入 QwenPaw——OpenAI 兼容 Provider 的零代码配置方式，以及非兼容 API 需要自定义 Provider 类的完整实现步骤。
个人公众号
1
2
卷三：工坊实战
[15] 造新工具 -> [16] 造新技能 -> [17] 接入新模型 <- you are here

第 11 章我们理解了 Provider 的策略模式。现在动手——接入一个新的 LLM 提供商。分两种情况：OpenAI 兼容的（简单）和非兼容的（需要新类）。

目标

把 DeepSeek（OpenAI 兼容）和假设的 “XModel”（非兼容）接入 QwenPaw。

情况一：OpenAI 兼容——只需加配置

DeepSeek 的 API 兼容 OpenAI 格式。不需要写新类，只需在 provider_manager.py 中加一个实例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# provider_manager.py，约第 490 行

PROVIDER_DEEPSEEK = OpenAIProvider(
    id="deepseek",
    name="DeepSeek",
    base_url="https://api.deepseek.com/v1",
    api_key_prefix="sk-",
    models=[
        ModelInfo(id="deepseek-chat", name="DeepSeek-V3"),
        ModelInfo(id="deepseek-reasoner", name="DeepSeek-R1"),
    ],
    support_model_discovery=True,
    freeze_url=True,
)

然后在 _init_builtins() 中注册：

1
2
3
def _init_builtins(self):
    # ... 已有的注册 ...
    self._add_builtin(PROVIDER_DEEPSEEK)

完成。不需要：

  • 新建文件
  • 修改 _provider_from_data()
  • 改 Agent 代码

为什么？因为 _provider_from_data() 的默认分支就是 OpenAIProvider.model_validate(data)——所有未匹配的 Provider 都走 OpenAI 路径。

完整流程

1
2
3
4
5
6
1. 定义 PROVIDER_XXX 常量（OpenAIProvider 实例）
2. 在 _init_builtins() 中 _add_builtin()
3. 用户在 Web UI 填入 API Key
4. ProviderManager 保存到磁盘（api_key Fernet 加密）
5. Agent 通过 model_factory 找到 Provider
6. OpenAIChatModelCompat 处理实际 API 调用

情况二：非 OpenAI 兼容——需要新类

假设 “XModel” 使用完全不同的 API 格式。需要创建新的 Provider 类。

第一步：创建 Provider 类

在 src/qwenpaw/providers/ 下创建 xmodel_provider.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import logging
from typing import List, Tuple
from .provider import Provider, ModelInfo

logger = logging.getLogger(__name__)


class XModelProvider(Provider):
    """XModel uses its own SDK and API format."""

    async def check_connection(
        self, timeout: float = 5
    ) -> Tuple[bool, str]:
        try:
            import xmodel_sdk
            client = xmodel_sdk.Client(api_key=self.api_key)
            client.ping(timeout=timeout)
            return True, ""
        except Exception as e:
            return False, str(e)

    async def fetch_models(
        self, timeout: float = 5
    ) -> List[ModelInfo]:
        try:
            import xmodel_sdk
            client = xmodel_sdk.Client(api_key=self.api_key)
            models = client.list_models(timeout=timeout)
            return [ModelInfo(id=m.id, name=m.name) for m in models]
        except Exception as e:
            logger.error(f"Failed to fetch models: {e}")
            return []

    async def check_model_connection(
        self, model_id: str, timeout: float = 5
    ) -> Tuple[bool, str]:
        try:
            import xmodel_sdk
            client = xmodel_sdk.Client(api_key=self.api_key)
            client.chat(
                model=model_id,
                messages=[{"role": "user", "content": "hi"}],
            )
            return True, ""
        except Exception as e:
            return False, str(e)

    def get_chat_model_instance(self, model_id: str):
        # 需要 XModelChatModel 适配层
        return XModelChatModel(
            model_name=model_id,
            api_key=self.api_key,
            generate_kwargs=(
                self.get_effective_generate_kwargs(model_id)
            ),
        )

四个抽象方法各实现了什么：

  • check_connection：测试 API Key 是否有效
  • fetch_models：从 Provider 拉取可用模型列表
  • check_model_connection：测试特定模型是否能响应
  • get_chat_model_instance：返回 agentscope 的 ChatModel 实例（最难的部分）

第二步：注册到 ProviderManager

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# provider_manager.py
from .xmodel_provider import XModelProvider

PROVIDER_XMODEL = XModelProvider(
    id="xmodel",
    name="XModel",
    base_url="https://api.xmodel.ai",
    models=[
        ModelInfo(id="xmodel-pro", name="XModel Pro"),
    ],
)

# _init_builtins() 中添加：
self._add_builtin(PROVIDER_XMODEL)

# _provider_from_data() 中添加路由：
if data["id"] == "xmodel":
    return XModelProvider.model_validate(data)

get_chat_model_instance 是真正的难点

这个方法要返回一个 agentscope 的 ChatModelBase 子类。如果 XModel 的消息格式不是 OpenAI 兼容的，需要写适配器——把 agentscope 格式的消息转成 XModel 格式，再把响应转回来。

这就是为什么 OpenAI 兼容的 Provider 不需要新类——OpenAIChatModelCompat 已经做了消息格式转换。

现实中的例子

Provider实现方式原因
OpenAI独立类基础实现
Anthropic独立类不同消息格式
Gemini独立类Google SDK
Ollama继承 OpenAI兼容 /v1 端点
LMStudio继承 OpenAI几乎完全一样
DashScope/DeepSeek/…直接用 OpenAI 实例完全兼容

大部分新增 Provider 都走”OpenAI 兼容”路径——只需配置，不需要新类。

自检

  • 知道 OpenAI 兼容的 Provider 只需加一个 OpenAIProvider(...) 实例
  • 知道非兼容 Provider 需要实现四个抽象方法
  • 知道 get_chat_model_instance 是最难的部分
  • 知道 _provider_from_data() 负责从 JSON 反序列化为正确的类

下一章我们接入一个新的聊天频道——BaseChannel 的 6 个核心方法怎么实现？怎么把平台特有的消息格式翻译成 QwenPaw 的统一格式？