147 lines
6.5 KiB
Python
147 lines
6.5 KiB
Python
"""管理员知识库教学命令:直接增删查 wtfllm 全局知识库。
|
|
|
|
用法(仅超级用户,群内 @机器人 或私聊):
|
|
教学 标题 | 内容
|
|
教学列表 [关键字]
|
|
修正 标题关键字 | 新内容
|
|
删除教学 标题关键字
|
|
"""
|
|
from nonebot import on_command, logger
|
|
from nonebot.adapters.onebot.v11 import Bot, MessageEvent
|
|
from nonebot.exception import IgnoredException
|
|
from nonebot.matcher import Matcher
|
|
from nonebot.message import run_preprocessor
|
|
from nonebot.permission import SUPERUSER
|
|
from nonebot.rule import to_me
|
|
|
|
from nonebot_plugin_wtfllm.memory.items.knowledge_base import KnowledgeEntry
|
|
from nonebot_plugin_wtfllm.utils import get_agent_id_from_bot
|
|
from nonebot_plugin_wtfllm.v_db.repositories import KnowledgeBaseRepository
|
|
|
|
WTFLLM_ENABLED_GROUPS = {"621016172"}
|
|
|
|
|
|
@run_preprocessor
|
|
async def gate_kb_teach_groups(matcher, event):
|
|
if matcher.module != "kb_teach":
|
|
return
|
|
if isinstance(event, MessageEvent) and getattr(event, "group_id", None):
|
|
if str(event.group_id) not in WTFLLM_ENABLED_GROUPS:
|
|
raise IgnoredException("kb_teach 仅在指定群启用")
|
|
|
|
|
|
def _split_args(text: str) -> tuple[str, str] | None:
|
|
if "|" not in text:
|
|
return None
|
|
left, _, right = text.partition("|")
|
|
left, right = left.strip(), right.strip()
|
|
return (left, right) if left and right else None
|
|
|
|
|
|
async def _repo(bot) -> tuple[KnowledgeBaseRepository, str]:
|
|
return KnowledgeBaseRepository(), get_agent_id_from_bot(bot)
|
|
|
|
|
|
teach_add = on_command("教学", rule=to_me(), permission=SUPERUSER, priority=2, block=True)
|
|
|
|
|
|
@teach_add.handle()
|
|
async def handle_add(matcher: Matcher, event: MessageEvent, bot: Bot):
|
|
args = _split_args(event.get_plaintext().replace("教学", "", 1))
|
|
if not args:
|
|
await matcher.finish("格式:教学 标题 | 内容")
|
|
title, content = args
|
|
repo, agent_id = await _repo(bot)
|
|
entry = KnowledgeEntry.create(
|
|
content=content,
|
|
title=title,
|
|
agent_id=agent_id,
|
|
category="规则",
|
|
tags=["manual"],
|
|
token_count=len(content),
|
|
)
|
|
await repo.save_knowledge(entry)
|
|
logger.info(f"[kb_teach] 新增知识: {title} ({agent_id})")
|
|
await matcher.finish(f"已记录知识:【{title}】")
|
|
|
|
|
|
teach_list = on_command("教学列表", rule=to_me(), permission=SUPERUSER, priority=2, block=True)
|
|
|
|
|
|
@teach_list.handle()
|
|
async def handle_list(matcher: Matcher, event: MessageEvent, bot: Bot):
|
|
keyword = event.get_plaintext().replace("教学列表", "", 1).strip()
|
|
repo, agent_id = await _repo(bot)
|
|
entries = await repo.search_by_category(agent_id=agent_id, category="规则", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="功能帮助", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="简单使用", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="脚本进阶", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="其它", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="general", limit=100)
|
|
if keyword:
|
|
entries = [e for e in entries if keyword in e.title or keyword in e.content]
|
|
if not entries:
|
|
await matcher.finish("没有匹配的知识条目")
|
|
lines = [f"共 {len(entries)} 条(显示前 20 条):"]
|
|
for e in entries[:20]:
|
|
lines.append(f"- 【{e.title}】{e.content[:60]}...")
|
|
await matcher.finish("\n".join(lines))
|
|
|
|
|
|
teach_update = on_command("修正", rule=to_me(), permission=SUPERUSER, priority=2, block=True)
|
|
|
|
|
|
@teach_update.handle()
|
|
async def handle_update(matcher: Matcher, event: MessageEvent, bot: Bot):
|
|
args = _split_args(event.get_plaintext().replace("修正", "", 1))
|
|
if not args:
|
|
await matcher.finish("格式:修正 标题关键字 | 新内容")
|
|
keyword, new_content = args
|
|
repo, agent_id = await _repo(bot)
|
|
entries = await repo.search_by_category(agent_id=agent_id, category="规则", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="功能帮助", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="简单使用", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="脚本进阶", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="其它", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="general", limit=100)
|
|
hits = [e for e in entries if keyword in e.title]
|
|
if not hits:
|
|
await matcher.finish(f"没有找到标题包含「{keyword}」的条目")
|
|
for e in hits:
|
|
await repo.delete_knowledge(e.storage_id)
|
|
new_entry = KnowledgeEntry.create(
|
|
content=new_content,
|
|
title=hits[0].title,
|
|
agent_id=agent_id,
|
|
category=hits[0].category,
|
|
tags=list(hits[0].tags) + ["manual"],
|
|
token_count=len(new_content),
|
|
)
|
|
await repo.save_knowledge(new_entry)
|
|
logger.info(f"[kb_teach] 修正 {len(hits)} 条知识: {keyword}")
|
|
await matcher.finish(f"已修正 {len(hits)} 条包含「{keyword}」的条目,新内容:【{new_entry.title}】")
|
|
|
|
|
|
teach_delete = on_command("删除教学", rule=to_me(), permission=SUPERUSER, priority=2, block=True)
|
|
|
|
|
|
@teach_delete.handle()
|
|
async def handle_delete(matcher: Matcher, event: MessageEvent, bot: Bot):
|
|
keyword = event.get_plaintext().replace("删除教学", "", 1).strip()
|
|
if not keyword:
|
|
await matcher.finish("格式:删除教学 标题关键字")
|
|
repo, agent_id = await _repo(bot)
|
|
entries = await repo.search_by_category(agent_id=agent_id, category="规则", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="功能帮助", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="简单使用", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="脚本进阶", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="其它", limit=100)
|
|
entries += await repo.search_by_category(agent_id=agent_id, category="general", limit=100)
|
|
hits = [e for e in entries if keyword in e.title]
|
|
if not hits:
|
|
await matcher.finish(f"没有找到标题包含「{keyword}」的条目")
|
|
for e in hits:
|
|
await repo.delete_knowledge(e.storage_id)
|
|
logger.info(f"[kb_teach] 删除 {len(hits)} 条知识: {keyword}")
|
|
await matcher.finish(f"已删除 {len(hits)} 条包含「{keyword}」的条目")
|