NcatBot 文档

外观

LLM_API 插件项目

约 974 字大约 3 分钟

2025-03-26

介绍

LLM_API 插件不直接提供大语言模型对话服务, 而是提供基于事件的对话接口和基本配置服务.

事件

通过 LLM_API.main 事件触发调用 LLM 的服务, Event 参数的 data 部分为 LLM 输入参数, 事件的处理结果为 LLM 的回复.

data 的构造如下:

{
    "history": [
        {
            "role": "system",
            "content": "系统提示内容"
        },
        {
            "role": "user",
            "content": "用户输入内容"
        },
    ], # 提示信息
    "max_tokens": 4096, # 最大长度
    "temperature": 0.7 # 温度,
}

result 的构造如下:

{
    "text": "回复内容",
    "status": "状态码" # 200 表示成功, 其他表示失败
    "error": "错误信息" # 
}

插件配置项

使用 NcatBot 的内置插件配置项功能, 三个核心配置项如下:

  • api: /cfg LLM_API.api <your api> api-key。
  • url: /cfg LLM_API.url <your url> 基准 url。
  • model: /cfg LLM_API.model <your model> 模型名。

例如 Kimi:

url: https://api.moonshot.cn/v1
model: moonshot-v1-8k
api: <KEY>

测试

拥有管理员权限可以发送 /tllma 触发大模型测试事件.

源代码

from ncatbot.plugin import BasePlugin, CompatibleEnrollment, Event
from ncatbot.core import GroupMessage, PrivateMessage
import asyncio
import httpx
import openai
from concurrent.futures import ThreadPoolExecutor

DEFAULT_URL = "url"
DEFAULT_API = "api"
DEFAULT_MODEL = "model"

class LLM_API(BasePlugin):
    name = "LLM_API" # 插件名称
    version = "0.0.1" # 插件版本

    async def on_load(self):
        print(f"{self.name} 插件已加载")
        print(f"插件版本: {self.version}")
        self.register_config("url", DEFAULT_URL)
        self.register_config("api", DEFAULT_API)
        self.register_config("model", DEFAULT_MODEL) # 注册三个配置项
        self.register_handler("LLM_API.main", self.main) # 注册事件(Event)处理器
        self.register_admin_func("test", self.test, prefix="/tllma", permission_raise=True) # 注册一个管理员功能, 需要提权以便在普通群聊中触发
    
    async def test(self, message: PrivateMessage):
        result = (await self.publish_async(Event("LLM_API.main", {
                "history": [
                {
                    "role": "system",
                    "content": "系统提示内容"
                },
                {
                    "role": "user",
                    "content": "用户输入内容"
                },
            ], # 提示信息
            "max_tokens": 4096, # 最大长度
            "temperature": 0.7 # 温度, 0-1, 越大越随机
        })))[0]
        await message.reply(text=result["text"] + result['error'])        
        
    async def main(self, event: Event):
        data = event.data
        url = self.data['config']["url"]
        api = self.data['config']["api"]
        model = self.data['config']["model"]
        
        if url == DEFAULT_URL or api == DEFAULT_API or model == DEFAULT_MODEL:
            event.add_result({
                "text": "",
                "status": 501,
                "error": "配置项错误"
            })
            return
        
        # 连接大预言模型的代码略去, 这里返回一个你好
        event.add_result({
            "text": "你好",
            "status": 200,
            "error": ""
        })
    
    async def on_unload(self):
        print(f"{self.name} 插件已卸载")

解析

on_load

async def on_load(self):
    print(f"{self.name} 插件已加载")
    print(f"插件版本: {self.version}")
    self.register_config("url", DEFAULT_URL)
    self.register_config("api", DEFAULT_API)
    self.register_config("model", DEFAULT_MODEL) # 注册三个配置项
    self.register_handler("LLM_API.main", self.main) # 注册事件(Event)处理器
    self.register_admin_func("test", self.test, prefix="/tllma", permission_raise=True) # 注册一个管理员功能, 需要提权以便在普通群聊中触发
  • 通过 register_config 注册所需要的配置项，配置项在正常退出时会保存，下次运行时自动加载。所有的配置项都需要在 on_load 中完成注册。
  • 通过 register_handler 注册事件处理器。事件总线在收到 LLM_API.main 事件时会调用 self.main 函数。
  • 通过 register_admin_func 注册一个管理员功能，功能的作用域只包括群聊和私聊，收到 /tllma 消息时，会进行鉴权，如果鉴权通过，则调用 self.test 函数。

self.main

async def main(self, event: Event):
    # 省略逻辑
    event.add_result({
        "text": "你好",
        "status": 200,
        "error": ""
    })
  • self.main 是插件服务的提供者，其它插件通过发布 LLM_API.main 事件来获得 self.main 提供的服务。
  • 通过 event.add_result 来添加事件处理结果，这个结果可以被事件发布者获取。

self.test

async def test(self, message: PrivateMessage):
    result = (await self.publish_async(Event("LLM_API.main", {
        # 此处省略大模型调用参数的构造
    })))[0]
    await message.reply(text=result["text"] + result['error'])
  • self.publish_async 是 BasePlugin 类提供的发布事件的方法。通过该方法向事件总线发布事件。
  • 事件总线收到 LLM_API.main 事件，调用 self.main 函数（所有订阅了该事件的处理器），self.main 添加事件处理结果。接着用 (await self.publish_async(...)) 获取所有的事件处理结果。
  • 只有一个处理器为该事件添加了结果，所以使用 (await self.publish_async(...))[0] 获取结果，再根据插件文档的说明正确解读。

版权归属：huan-yp

许可证：署名 4.0 国际 (CC-BY-4.0)

© 2025 NcatBot