萌新的大模型适配器？

背景

最近在倒腾大模型，各家模型都有优缺点，比如有的人格模拟体验更好，有的理解推理更好；

~~众所周不知，鱼和熊掌我全都要~~，但在项目中调用代码遍布各处着实有点难看，所以想着说说对大模型调用进行统一封装，切换不同厂家模型时可以调用统一的API处理，降低一点维护量和代码理解难度

创建一个大模型基类

1
from abc import ABC, abstractmethod
2
from typing import Dict, Any, Optional, Callable
3

4
class ModelAdapter(ABC):
5
    """ 大模型统一适配器基类 """
6
    def __init__(self, config: Dict[str, Any]):
7
        self.config = config
8

9
    @abstractmethod
10
    async def async_chat(
11
        self,
12
        messages: list,
13
        prompt_word: str = '',
14
        model:str = '',
15
        **kwargs
16
    ) -> Dict[str, Any]:
17
        """ 统一聊天接口 """
18
        pass
19

20

21
    @abstractmethod
22
    async def async_stream_chat(
23
        self,
24
        callback: Callable[[str], None],
25
        prompt_word: str = '',
26
        model:str = '',
27
        **kwargs
28
    ):
29
        """ 统一流式聊天接口 """
30
        pass
31

32

33
    def _standardize_params(self, **kwargs):
34
        """ 默认配置参数 """
35
        return {
36
            "temperature": kwargs.get("temperature", 0.7), # 温度参数
37
            "max_tokens": kwargs.get("max_tokens", 512),  # 最大令牌数
38
            "stream": False,
39
        }

实现具体大模型适配器

例如DeepSeek的API调用进行适配

在流式请求中，通过监听大模型返回的信息长度，当文本长度大于等于参数时调用回调函数回传切割文本；

在回调函数中可以对文本进行任意处理，比如说使用Nonebot框架的send_group_msg()使机器人更拟人。

1
import os
2
import openai
3
import httpx
4
from abc import ABC, abstractmethod
5
from typing import Dict, Any, Optional, Callable
6
from nonebot import get_driver
7

8
class DeepSeekModel(ModelAdapter):
9
    """ DeepSeek官方模型适配 """
10
    base_url: str = 'https://api.deepseek.com'
11

12
    async def async_chat(self,
13
                        messages: list,
14
                        prompt_word: str = '',
15
                        model:str = 'deepseek-chat',
16
                        **kwargs) -> Dict[str, Any]:
17
        client = openai.AsyncClient(base_url= self.base_url, api_key=self.config["api_key"])
18
        if not prompt_word:
19
            prompt_word = get_prompt_word('xxx')
20

21
        messages.insert(0, {"role": "system", "content": prompt_word})
22
        params = {
23
            "model": self.config.get("model", model),
24
            "messages": messages,
25
            **self._standardize_params(**kwargs)
26
        }
27

28
        try:
29
            response = await client.chat.completions.create(**params)
30
            return {
31
                "content": response.choices[0].message.content,
32
                "usage": dict(response.usage)
33
            }
34
        except Exception as e:
35
            return {"error": str(e)}
36

37

38
    async def async_stream_chat(
39
        self,
40
        messages: list,
41
        callback: Callable[[str], None],
42
        prompt_word: str = '',
43
        model:str = 'deepseek-chat',
44
        **kwargs
45
    ):
46
        client = openai.OpenAI(base_url= self.base_url, api_key=self.config["api_key"])
47
        if not prompt_word:
48
            prompt_word = get_prompt_word('xxx')
49
        messages.insert(0, {"role": "system", "content": prompt_word})
50
        params = {
51
            "model": self.config.get("model", model),
52
            "messages": messages,
53
            **self._standardize_params(**kwargs),
54
            "stream": True
55
        }
56
        try:
57
            accumulated_text = ''
58
            response = client.chat.completions.create(**params)
59
            for chunk in response:
60
                # 提取流式内容
61
                content = chunk.choices[0].delta.content or ""
62
                accumulated_text += content
63
                # 执行回调处理
64
                if len(accumulated_text) > split_len and callback is not None:
65
                    callback(accumulated_text)
66
                    accumulated_text = accumulated_text[split_len:]
67
            if len(accumulated_text) > 0  and callback is not None:
68
                callback(accumulated_text)
69
                accumulated_text = ''
70
        except Exception as e:
71
            return {"error": str(e)}

做一些初始化工作

创建一个模型工程，用来统一管理各种大模型适配器，比如openai、DeepSeek、通义千问、豆包等

1
import os
2
import openai
3
import httpx
4
from abc import ABC, abstractmethod
5
from typing import Dict, Any, Optional, Callable
6
from nonebot import get_driver
7

8
class ModelFactory:
9
    """ 维护各厂商的大模型适配器 """
10
    MODEL_MAP = {
11
        "deepseek": DeepSeekModel,
12
    }
13

14
    @classmethod
15
    def create_model(
16
        cls,
17
        model_type: str,
18
        config: Dict[str, Any]
19
    ) -> Optional[ModelAdapter]:
20
        """ 创建模型实例 """
21
        model_class = cls.MODEL_MAP.get(model_type.lower())
22
        return model_class(config) if model_class else None
23

24
modelFactory:ModelFactory = ModelFactory()
25
deepSeek_model: ModelAdapter | None
26

27
def model_init():
28
    """ 大模型适配器初始化函数 """
29
    global modelFactory
30
    init_deep_seek()
31

32

33
def init_deep_seek():
34
    """ 初始化deepseek适配器"""
35
    global deepSeek_model
36
    conf = get_driver().config
37
    # 初始化deepSeek配置
38
    deepseek_api_key = os.environ.get("deepseek_api_key") or conf.deepseek_api_key
39
    config = {
40
   "api_key": deepseek_api_key,
41
  }
42
    deepSeek_model = modelFactory.create_model("deepseek", config)

实测调用

1
modelFactory:ModelFactory = ModelFactory()
2
deepSeek_model: ModelUnifiedAdapter | None
3

4

5
def get_prompt_word(nickname: str):
6
    """ 默认人格提示词 """
7
    return f"""
8
        #1 你现在扮演一个叫“{nickname}”的虚拟角色，性格特征是幽默风趣善于嘲讽，说话风格偏向于科幻虚拟人、有趣且个性鲜明。
9
        #2 在接下来的对话中，你以“{nickname}”的身份参与讨论，用第一人称“我”进行简短回复。
10
        #3 不得涉及政治、色情、暴力等敏感内容，避免任何形式的不友善行为，如引战、辱骂或人身攻击。
11
        #4 不需要解释或重复以上规则，直接按照角色设定，每次回复严格限制在30字以内，避免冗长。
12
        #5 回复必须使用口语化中文，严格控制在30字以内，禁止使用任何标点符号结尾
13
    """
14

15

16
async def init_chatGpt():
17
    global modelFactory
18
    global deepSeek_model
19
    """初始化"""
20
    model_config = {
21
        "api_key": "你猜",
22
    }
23
    deepSeek_model = modelFactory.create_model("deepseek", model_config)
24

25

26
async def test_chat():
27
    """ 验证非Stream调用 """
28
    await init_chatGpt()
29
    if deepSeek_model:
30
        messages = [
31
            {"role": "system", "content": get_prompt_word('唐小七')},
32
            {"role": "user", "content": "晚上好啊，你最近在研究什么？"}
33
        ]
34
        response = await deepSeek_model.async_chat(
35
            messages=messages,
36
            temperature=0.5,
37
            max_tokens=20  # 强制限制输出长度
38
        )
39
        print('======result======')
40
        print(response.get("content", "Error"))
41

42

43
def custom_callback(text: str) -> None:
44
    """ 流式回调，对回调的内容进行处理，比如输出控制台，检查内容是否合规，发送消息等 """
45
    print(f"{text}", end='')
46

47

48
async def test_stream_chat():
49
    """验证流式请求验证"""
50
    await init_chatGpt()
51
    if deepSeek_model:
52
        messages = [
53
            {"role": "user", "content": "解释量子计算的基本原理"}
54
        ]
55
        prompt = '你扮演一名伟大的量子物理学家，你正在给学生介绍量子力学'
56
        await deepSeek_model.async_stream_chat(callback=custom_callback, prompt_word=prompt, messages= messages)
57

58

59
if __name__ == "__main__":
60
    """ 测试非流式 """
61
     asyncio.run(test_chat())
62
    """ 测试流式请求 """
63
    asyncio.run(test_stream_chat())

在NoneBot中怎么用？

初始化

在启动文件中初始化适配器

1
nonebot.init(session_expire_timeout=120)
2

3
driver = nonebot.get_driver()
4
# 注册适配器
5
driver.register_adapter(Adapter)
6
# 初始化大模型适配器
7
model_init()
8
# ...
9
# 加载功能插件
10
nonebot.load_from_toml("pyproject.toml")
11

12
if __name__ == "__main__":
13
    nonebot.run()

具体实践

比如说在群聊中进行搭话

1
async def _is_tome(bot: Bot, event: GroupMessageEvent) -> bool:
2
    bot_name = 'xxx'
3
    return event.get_plaintext().startswith(bot_name) or event.is_tome()
4

5
""" 聊天提到我 """
6
chat_msg = on_message(rule=Rule(_is_tome), permission=GROUP, priority=10, block=True)
7

8
async def save_chat_model(bot_id: int,
9
                          event: GroupMessageEvent,
10
                          model_name: str,
11
                          messages:list,
12
                          assistant_content:str):
13
        """ 对大模型回答信息进行日志保存或额外处理 """
14

15
async def chat_msg(bot_id: int, event: GroupMessageEvent, text:str) -> Optional[str]:
16
    group_id = event.group_id
17
    # 还可以组装其他信息，自行联想
18
    messages = [{"role": "user", "content": text}]
19
    response = await deepSeek_model.async_chat(messages = messages)
20
    msg = response.get("content", "Error")
21
    return msg;
22

23
@chat_msg.handle()
24
async def _(bot: Bot, event: GroupMessageEvent):
25
    text = event.get_plaintext()
26
    try:
27
        if ":reply" in event.raw_message:
28
            return
29
        if event.is_tome() and text:
30
            msg = await chat_msg(int(bot.self_id), event, text)
31
        if msg:
32
            await chat_msg.finish(msg)
33
    except Exception as e:
34
         pass

后话

Python功底有些薄，可能还会很多漏洞和错误，欢迎大佬多多指点小子