feat(kb_teach): 新增管理员知识库教学命令(教学/教学列表/修正/删除教学,仅超级用户)

danding_bot/plugins/kb_teach.py

@@ -0,0 +1,146 @@
+"""管理员知识库教学命令:直接增删查 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}」的条目")