OpenClaw Agent
11c62dd383
- kb_workflow.py: 重构为编排者，使用sessions_spawn调用子技能 - skill_main.py: 新增技能入口，处理用户消息 - SKILL.md: 更新文档，明确架构和流程 - SkillBug.md: 新增问题记录文档 设计原则: - workflow不读文件内容，从文件名提取平台 - 使用sessions_spawn调用子agent - 子技能自己定义输入输出格式
134 lines
3.4 KiB
Python
134 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
KB Workflow Skill Main Entry
|
|
|
|
当用户发送包含触发词的消息时自动触发：
|
|
- "添加到知识库 + 链接"
|
|
- "保存到知识库 + 链接"
|
|
|
|
职责：
|
|
1. 接收用户消息
|
|
2. 识别触发词和链接
|
|
3. 调用 kb_workflow.process_url 处理
|
|
4. 汇报结果
|
|
"""
|
|
|
|
import re
|
|
import sys
|
|
import json
|
|
from pathlib import Path
|
|
|
|
# 触发词模式
|
|
TRIGGER_PATTERNS = [
|
|
r'添加到知识库\s+(https?://[^\s]+)',
|
|
r'保存到知识库\s+(https?://[^\s]+)',
|
|
]
|
|
|
|
|
|
def extract_link(message: str) -> str | None:
|
|
"""从消息中提取链接"""
|
|
for pattern in TRIGGER_PATTERNS:
|
|
match = re.search(pattern, message)
|
|
if match:
|
|
return match.group(1)
|
|
return None
|
|
|
|
|
|
def main(message: str) -> Dict[str, Any]:
|
|
"""
|
|
主入口函数 - 当 kb-workflow 作为子代理被调用时执行
|
|
"""
|
|
result = {
|
|
"success": False,
|
|
"url": None,
|
|
"workflow_result": None,
|
|
"error": None
|
|
}
|
|
|
|
try:
|
|
# 1. 提取链接
|
|
url = extract_link(message)
|
|
if not url:
|
|
result["error"] = "未找到链接"
|
|
return result
|
|
|
|
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
|
|
|
|
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:
|
|
message = sys.stdin.read()
|
|
|
|
if not message:
|
|
print("请提供消息内容")
|
|
sys.exit(1)
|
|
|
|
result = main(message)
|
|
print_result(result)
|
|
|
|
sys.exit(0 if result.get('success') else 1)
|