refactor(kb-workflow): 改用Prompt-based模式

- SKILL.md: 重写为详细workflow指导,主代理直接读取执行
- kb_workflow.py: 简化为辅助工具函数库
- skill_main.py: 简化为触发检测入口
- SkillBug.md: 记录架构变更

主代理读取SKILL.md后,按照指导调用sessions_spawn,无需配置allowlist
This commit is contained in:
OpenClaw Agent
2026-02-10 20:46:06 +08:00
parent 679cca3b2c
commit e370b6b39f
4 changed files with 445 additions and 937 deletions
+33 -70
View File
@@ -3,21 +3,16 @@
"""
KB Workflow Skill Main Entry
当用户发送包含触发词的消息时自动触发:
- "添加到知识库 + 链接"
- "保存到知识库 + 链接"
当用户发送"添加到知识库 + 链接"时触发。
职责:
1. 接收用户消息
2. 识别触发词和链接
3. 调用 kb_workflow.process_url 处理
4. 汇报结果
此入口只负责:
1. 确认触发条件
2. 引导主代理读取 SKILL.md
3. 让主代理按照 SKILL.md 的指导执行完整 workflow
"""
import re
import sys
import json
from pathlib import Path
# 触发词模式
TRIGGER_PATTERNS = [
@@ -37,12 +32,15 @@ def extract_link(message: str) -> str | None:
def main(message: str) -> Dict[str, Any]:
"""
主入口函数 - 当 kb-workflow 作为子代理被调用时执行
主入口函数
此函数返回指导信息,让主代理知道如何执行 workflow。
实际的 workflow 逻辑在 SKILL.md 中定义。
"""
result = {
"success": False,
"url": None,
"workflow_result": None,
"instructions": None,
"error": None
}
@@ -55,69 +53,36 @@ def main(message: str) -> Dict[str, Any]:
result["url"] = url
# 2. 调用workflow处理
from kb_workflow import process_url
import os
base_path_str = os.environ.get('KB_BASE_PATH', '')
base_path = Path(base_path_str) if base_path_str else None
# 处理
workflow_result = process_url(url, base_path)
result["workflow_result"] = workflow_result
if workflow_result.get("success"):
result["success"] = True
# 2. 生成指导信息
result["success"] = True
result["instructions"] = {
"action": "execute_workflow",
"url": url,
"workflow_prompt": f"""请按照 SKILL.md 中的指导执行完整的知识库 workflow:
1. 查重 - 检查知识库是否已存在此内容
2. 调用 kb-link-parser 解析 URL
3. 根据平台调用对应元数据技能
4. 调用内容下载技能
5. 调用 OCR 处理(30分钟超时)
6. 调用深度分析
7. 归档到 archive/ 目录
8. Git commit
9. 汇报完成
当前处理的 URL: {url}
开始执行 workflow。"""
}
except Exception as e:
result["error"] = str(e)
import traceback
traceback.print_exc()
return result
def print_result(result: Dict[str, Any]):
"""打印结果"""
print("\n" + "="*50)
print("KB Workflow Result")
print("="*50)
print(f"\nURL: {result.get('url', 'N/A')}")
wr = result.get('workflow_result', {})
if wr:
print(f"\n处理状态: {'成功' if wr.get('success') else '失败'}")
if wr.get('stages'):
print(f"\n处理阶段:")
for stage in wr['stages']:
stage_name = stage.get('stage', 'unknown')
print(f" ✓ {stage_name}")
if wr.get('final_file'):
print(f"\n最终文件: {wr['final_file']}")
archive = wr.get('archive', {})
if archive:
print(f"\n归档:")
print(f" 归档路径: {archive.get('archive_path', 'N/A')}")
print(f" Git提交: {'是' if archive.get('git_commit') else '否'}")
push = archive.get('git_push', {})
if push:
print(f" Git推送: {'成功' if push.get('success') else '失败'}")
if wr.get('error'):
print(f"\n错误: {wr['error']}")
if result.get('error'):
print(f"\n错误: {result['error']}")
print()
if __name__ == "__main__":
# 从参数获取消息(作为子代理时)
# 从参数获取消息
if len(sys.argv) > 1:
message = sys.argv[1]
else:
@@ -128,6 +93,4 @@ if __name__ == "__main__":
sys.exit(1)
result = main(message)
print_result(result)
sys.exit(0 if result.get('success') else 1)
print(result)