萌新教你对接大模型?
背景
最近在倒腾大模型,各家模型都有优缺点,比如有的人格模拟体验更好,有的理解推理更好;
~~众所周不知,鱼和熊掌我全都要~~,但在项目中调用代码遍布各处着实有点难看,所以想着说说对大模型调用进行统一封装,切换不同厂家模型时可以调用统一的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功底有些薄,可能还会很多漏洞和错误,欢迎大佬多多指点小子