萌新的大模型适配器？

背景

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

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

创建一个大模型基类

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, Callable

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

    @abstractmethod
    async def async_chat(
        self,
        messages: list,
        prompt_word: str = '',
        model:str = '',
        **kwargs
    ) -> Dict[str, Any]:
        """ 统一聊天接口 """
        pass
  

    @abstractmethod
    async def async_stream_chat(
        self,
        callback: Callable[[str], None],
        prompt_word: str = '',
        model:str = '',
        **kwargs
    ):
        """ 统一流式聊天接口 """
        pass
  

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

实现具体大模型适配器

例如DeepSeek的API调用进行适配

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

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

import os
import openai
import httpx
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, Callable
from nonebot import get_driver

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

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

        messages.insert(0, {"role": "system", "content": prompt_word})
        params = {
            "model": self.config.get("model", model),
            "messages": messages,
            **self._standardize_params(**kwargs)
        }

        try:
            response = await client.chat.completions.create(**params)
            return {
                "content": response.choices[0].message.content,
                "usage": dict(response.usage)
            }
        except Exception as e:
            return {"error": str(e)}


    async def async_stream_chat(
        self,
        messages: list,
        callback: Callable[[str], None],
        prompt_word: str = '',
        model:str = 'deepseek-chat',
        **kwargs
    ):
        client = openai.OpenAI(base_url= self.base_url, api_key=self.config["api_key"])
        if not prompt_word:
            prompt_word = get_prompt_word('xxx')
        messages.insert(0, {"role": "system", "content": prompt_word})
        params = {
            "model": self.config.get("model", model),
            "messages": messages,
            **self._standardize_params(**kwargs),
            "stream": True
        }
        try:
            accumulated_text = ''
            response = client.chat.completions.create(**params)
            for chunk in response:
                # 提取流式内容
                content = chunk.choices[0].delta.content or ""
                accumulated_text += content
                # 执行回调处理
                if len(accumulated_text) > split_len and callback is not None:
                    callback(accumulated_text)
                    accumulated_text = accumulated_text[split_len:]
            if len(accumulated_text) > 0  and callback is not None:
                callback(accumulated_text)
                accumulated_text = ''
        except Exception as e:
            return {"error": str(e)}

做一些初始化工作

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

import os
import openai
import httpx
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, Callable
from nonebot import get_driver

class ModelFactory:
    """ 维护各厂商的大模型适配器 """
    MODEL_MAP = {
        "deepseek": DeepSeekModel,
    }

    @classmethod
    def create_model(
        cls,
        model_type: str,
        config: Dict[str, Any]
    ) -> Optional[ModelAdapter]:
        """ 创建模型实例 """
        model_class = cls.MODEL_MAP.get(model_type.lower())
        return model_class(config) if model_class else None

modelFactory:ModelFactory = ModelFactory()
deepSeek_model: ModelAdapter | None

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


def init_deep_seek():
    """ 初始化deepseek适配器"""
    global deepSeek_model
    conf = get_driver().config
    # 初始化deepSeek配置
    deepseek_api_key = os.environ.get("deepseek_api_key") or conf.deepseek_api_key
    config = {
	 "api_key": deepseek_api_key,
	}
    deepSeek_model = modelFactory.create_model("deepseek", config)

实测调用

modelFactory:ModelFactory = ModelFactory()
deepSeek_model: ModelUnifiedAdapter | None


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


async def init_chatGpt():
    global modelFactory
    global deepSeek_model
    """初始化"""
    model_config = {
        "api_key": "你猜",
    }
    deepSeek_model = modelFactory.create_model("deepseek", model_config)


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

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


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


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

在NoneBot中怎么用？

初始化

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

nonebot.init(session_expire_timeout=120)

driver = nonebot.get_driver()
# 注册适配器​
driver.register_adapter(Adapter)
# 初始化大模型适配器
model_init()
# ...
# 加载功能插件
nonebot.load_from_toml("pyproject.toml")

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

具体实践

比如说在群聊中进行搭话

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

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

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

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

@chat_msg.handle()
async def _(bot: Bot, event: GroupMessageEvent):
    text = event.get_plaintext()
    try:
        if ":reply" in event.raw_message:
            return
        if event.is_tome() and text:
            msg = await chat_msg(int(bot.self_id), event, text)
        if msg:
            await chat_msg.finish(msg)
    except Exception as e:
         pass

后话

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