项目已在 Github 上开源,这里是一些基础的技术讲解。
在项目中,采用插件化架构进行系统构建,利用 Ncatbot 的插件模式,可以很好的将主程序与功能隔离开。
插件架构的优势在于模块边界清晰、依赖隔离度高、系统可拓展性强。标准的插件化极大程度上将代码之间解耦,无论是开发还是部署都是十分友好的。
项目系统目录结构如下:
chat.py
是系统的核心模块,主程序 main.py
负责接收、解析指令,并且路由消息到模型模块处理。在开发中这是十分友好的一种结构,使得专注于单个功能模块的开发与调试,极大降低了维护复杂度。
而配置项集中于 config.yml
,进一步提升了灵活性与环境适配能力。
采用临时 json 文件的方式,记录指令的调用与回复,并且传入 API 接口,可以使得大模型可以获得一定程度上的短期记忆功能,但一定程度上来说,这对系统并不友好,我认为更优解是采用数据库的形式,但使用数据库很大程度上增大了系统的复杂性,所以使用 json 文件是一个很好的代替选择。
main.py
是插件的主入口,通过 register_user_func
方法注册两个指令:/chat
和 /clear chat_history
,分别对应聊天功能和历史记录清除。
此外,主程序支持对图像消息的自动识别,提取其中的图像 URL 并传递给 chat_model_instance.recognize_image
方法,自动获取视觉描述。
获取到视觉描述后,再转到语言大模型来进行输出,其实这是一种很好的解决方法,在当前使用场景下,更多的是需要对图片进行识别后对内容进行分析处理,而不是处理图像本身,这可以很大程度上减少 API 调用,提高缓存命中率,减少 TOKEN 使用,从而降低 API 调用成本,并且可以使用云端大模型来进行识别后,再交给本地大模型回答,近一步压缩成本。
错误处理方面,try...except
包裹了整个聊天逻辑,避免了图像解码失败或 API 异常造成主流程崩溃,保持插件健壮性。
整合来看,main.py
是典型的“轻控制器”模式,仅协调各组件而不承担业务逻辑细节,使整个插件具备良好的工程可读性。
chat.py
是插件的核心逻辑。它负责处理模型调用、聊天历史记忆、图像识别等任务。为了兼容多种模型接口(如 OpenAI API 与 Ollama 本地服务),采用了统一封装接口的策略,使外部调用者无需关注模型细节,只需通过 useCloudModel()
或 useLocalModel()
两个方法即可完成对话。
值得注意的是,由于 OpenAI 接口调用和 Ollama 调用存在些许不同,并且某些模型的参数并不完全,所以相比起来,使用 OpenAI 接口,调用云端模型其实可以获得更好的体验,例如我们可以控制模型的 temperature 使它更具想象力亦或者是更注重实时,减少幻觉。
所有用户历史被存储在 cache/history.json
文件中,这是一种持久化保存的方案,并且可以具备一定程度的可追溯性,历史记录通过 _update_user_history
方法动态更新,控制在配置文件设定的最大轮数之内。这种方式防止了上下文过大带来的性能问题,同时确保模型能理解连续上下文,提升回答质量,使得即使对接接口也可以拥有近似记忆的能力。
类中还集成了 OpenAI 的图像识别模型,通过 _build_vision_messages
构建多模态消息结构。并且设计上我将图像处理、消息构造、异常处理、调用模型等函数分离,使得开发中可以更快定位问题所在,开源后也更利于其他开发者阅读。
云端模型的调用主要通过 openai
官方库封装,利用 chat.completions.create
方法完成上下文构建与回复生成。在每一次调用中都通过 _build_messages()
方法构造完整的对话上下文,加入系统提示词和使用 cache/history.json
保存的历史记录,实现多轮记忆式对话。
调用逻辑中封装了 temperature
参数,支持通过配置文件灵活控制模型输出的随机性。
当遇到 Bug report 时,统一使用 return 将报错反馈到用户面,这样可以减少大量的报错开发处理,并且将常见的由于配置失误导致的问题,更清晰的反馈出来,即统一模型+规则处理的两种方法来反馈运行时遇到的问题。
返回结果后,会将本轮问答同步到用户历史缓存中,并保存到本地文件中,确保下一轮能正常取回上下文。这样可以降低内存依赖、增强缓存命中率,同时为后续调试与行为复现提供依据。
本地模型的调用通过 ollama.chat()
完成,复用了 _build_messages()
的上下文构建逻辑,以确保调用逻辑与云端一致,保持接口一致性。
这种本地推理机制的优点非常明显:在无网络或私有部署环境下依旧能使用智能对话功能,极大增强了插件的部署灵活性和安全性。即便是在隐私敏感场景中,也可以做到本地化部署与运行。
设计上保持本地与云端调用接口一致(都封装为 use*Model()
),外部调用方无需判断模型来源,从而降低复杂度。除此之外,同样实现了历史记录更新、异常捕获机制,使本地模型具备与云端一致的功能完整度与稳定性。
图像识别功能是本插件的一大亮点。插件支持识别图像消息并通过 OpenAI 视觉模型进行处理。整个流程如下:
从图像消息中提取 URL;
image_url
和 text prompt
);这一机制有效解决了图文混合输入场景中的信息不对称问题,同时通过分级调节调用,可以仅利用云端高算力处理复杂问题,再交给本地处理简单化后的问题,极大程度上减少了 TOKEN 使用率。
异常处理方面,我们设计了两级降级策略:如主调用失败则尝试纯文本 fallback prompt;若仍失败,则提示用户检查 API 密钥或模型状态。这种容错设计使插件能在部分失败时依旧保持服务不中断。
聊天历史存储在 cache/history.json
文件中,按用户维度管理。该设计使系统可同时服务多用户,并为每个用户维护独立上下文。通过 _get_user_history
和 _update_user_history
方法,插件能自动在每轮对话中注入历史信息,实现类“记忆式”问答体验。
我们对历史记录长度做了窗口限制(默认10轮),以控制上下文规模并避免模型处理压力过大、过度消耗 TOKEN。缓存更新为同步写入的操作,确保在系统崩溃、断电等异常情况下不会造成信息丢失。
此外,还支持通过指令 /clear chat_history
主动清除用户历史,为隐私或重新对话提供了便利。这种机制让插件既具备持久性,又保留用户主动控制空间。
在调试处打断点、print 标志 这都是一个很好的测试习惯,我还从微信开发学到了一招 —— print("FUCK"),在长期运行时偶尔会出现崩溃的情况,这时候事实上可以在 log 中输出指定的字符,当查看日志定位问题的时候,可以直接搜索字符串来迅速定位,FUCK 无疑是一种有趣的方式。
ModuleChat/
├── main.py # 插件主入口,负责指令注册与调度逻辑
├── chat.py # 模型适配层,封装本地和云端模型的调用
├── config.yml # 配置文件,集中控制模型参数与启用选项
├── requirements.txt # 依赖库
└── cache/
└── history.json # 聊天历史记忆文件
if image_url and self.chat_model.get('enable_vision', True) and not self.chat_model.get('use_local_model'):
# 使用图像识别功能
image_description = await chat_model_instance.recognize_image(image_url)
user_input = f"用户发送了一张图片,图片描述是:{image_description}。用户说:{user_input}"
elif image_url and not self.chat_model.get('enable_vision', True):
# 图像识别功能未开启,但检测是否是本地模型
if self.chat_model.get('use_local_model'):
user_input = f"用户发送了一张图片,但用户使用的是本地模型,无法进行图像识别。用户说:{user_input}"
else:
user_input = f"用户发送了一张图片,但图像识别功能未开启。用户说:{user_input}"
async def useLocalModel(self, msg: BaseMessage, user_input: str):
"""使用本地模型处理消息"""
try:
# 构建消息列表,包含历史记录
messages = self._build_messages(user_input, msg.user_id if hasattr(msg, 'user_id') else None)
response: ChatResponse = chat(
model=self.config['model'],
messages=messages
)
reply = response.message.content.strip()
# 保存当前对话到历史记录
if hasattr(msg, 'user_id'):
self._update_user_history(msg.user_id, {"role": "user", "content": user_input})
self._update_user_history(msg.user_id, {"role": "assistant", "content": reply})
except Exception as e:
reply = f"请求出错了:{str(e)}"
return reply
def _build_messages(self, user_input: str, user_id: str = None):
"""构建消息列表"""
messages = []
# 添加系统提示词
system_prompt = self.config.get('system_prompt', "你是一名聊天陪伴机器人")
messages.append({"role": "system", "content": system_prompt})
if user_id:
history = self._get_user_history(user_id)
messages.extend(history)
# 添加当前用户输入
messages.append({"role": "user", "content": user_input})
return messages
if "401" in str(fallback_error) or "Unauthorized" in str(fallback_error):
raise Exception("模型API认证失败,请检查配置文件")
raise Exception(f"图像识别出错: {str(e)}, 备用方法也失败: {str(fallback_error)}")
for segment in msg.message:
if isinstance(segment, dict) and segment.get("type") == "image":
image_url = segment.get("data", {}).get("url")
break
response = requests.get(image_url)
response.raise_for_status()
return base64.b64encode(response.content).decode('utf-8')
# 获取并编码图片
image_data = self._encode_image_from_url(image_url)
# 构建消息
messages = self._build_vision_messages(image_data, prompt)
# 调用视觉模型
response = self.vision_client.chat.completions.create(
model=self.config.get('vision_model'),
messages=messages,
temperature=self.config.get('model_temperature', 0.6),
stream=False,
max_tokens=2048
)
async def clear_user_history(self, user_id: str):
"""清除指定用户的历史记录"""
user_id = str(user_id)
if user_id in self.history:
del self.history[user_id]
self._save_history()
reply = "已清空聊天记录"
else:
reply = "没有找到用户的聊天记录"
return reply
main.py
负责指令解耦与路由,chat.py
为核心逻辑模块,支持多模型调用(云端OpenAI与本地Ollama)和图像识别功能。聊天历史通过cache/history.json
实现多轮记忆,并支持主动清除。设计上注重接口一致性、异常处理和调试便利性。