From e370b6b39f9a0bc6006608b7e71af5809de99fe9 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Tue, 10 Feb 2026 20:46:06 +0800 Subject: [PATCH] =?UTF-8?q?refactor(kb-workflow):=20=E6=94=B9=E7=94=A8Prom?= =?UTF-8?q?pt-based=E6=A8=A1=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SKILL.md: 重写为详细workflow指导，主代理直接读取执行 - kb_workflow.py: 简化为辅助工具函数库 - skill_main.py: 简化为触发检测入口 - SkillBug.md: 记录架构变更 主代理读取SKILL.md后，按照指导调用sessions_spawn，无需配置allowlist --- skills/kb-workflow/SKILL.md | 430 +++++++++--------- skills/kb-workflow/SkillBug.md | 125 +++--- skills/kb-workflow/kb_workflow.py | 724 ++++++------------------------ skills/kb-workflow/skill_main.py | 103 ++--- 4 files changed, 445 insertions(+), 937 deletions(-) diff --git a/skills/kb-workflow/SKILL.md b/skills/kb-workflow/SKILL.md index 962805d..50df0a5 100644 --- a/skills/kb-workflow/SKILL.md +++ b/skills/kb-workflow/SKILL.md @@ -1,244 +1,258 @@ --- name: kb-workflow -description: 知识库工作流编排技能，提供收件箱扫描、内容处理、归档、订阅检测、Git版本管理和自动清理功能。 +description: 知识库工作流编排技能。接收"添加到知识库 + 链接"消息，编排各子代理完成完整处理流程。 metadata: short-description: 知识库工作流编排 --- # KB Workflow -## Description +## 触发条件 -KB Workflow 是一个完整的知识库管理系统，包含： - -**核心功能**： -- 📥 **收件箱扫描** - 扫描四个子目录 (pending, metadata, content, analyzed) -- 🔄 **内容处理** - 调用子agent处理各阶段 -- 📦 **Git版本管理** - 归档后自动提交并推送 -- 🔔 **订阅管理** - 跟踪订阅源更新 -- 🧹 **自动清理** - 定期清理临时文件 -- 📊 **汇报生成** - 生成处理统计 - -## 触发方式 - -### 自然语言触发 - -当用户发送包含以下模式的消息时，技能会自动触发： +当用户发送包含以下模式的消息时，此技能自动触发： | 触发词 | 行为 | 示例 | |--------|------|------| -| `添加到知识库` + 链接 | 添加到收件箱并处理 | "添加到知识库 https://www.bilibili.com/video/BVxxx" | -| `保存到知识库` + 链接 | 添加到收件箱并处理 | "保存到知识库 https://www.bilibili.com/video/BVxxx" | +| `添加到知识库` + 链接 | 添加到知识库并处理 | "添加到知识库 https://www.bilibili.com/video/BVxxx" | +| `保存到知识库` + 链接 | 同上 | "保存到知识库 https://www.bilibili.com/video/BVxxx" | -### 支持的平台 +## Workflow 步骤 -| 平台 | 子技能 | -|------|--------| -| Bilibili | kb-bilibili-metadata, kb-bilibili-content | -| 知乎 | kb-zhihu-metadata | +当触发后，按照以下步骤执行： -## 架构设计 +### 步骤 1: 查重 -``` -用户: "添加到知识库 https://www.bilibili.com/video/BV1ci4y1V7tb" - │ - ▼ -┌─────────────────────────────────────────────────────────┐ -│ kb-workflow (主代理) │ -│ │ -│ 职责: │ -│ 1. 接收用户消息，识别触发词和链接 │ -│ 2. 查重（根据URL/文件名） │ -│ 3. 调用子agent处理各阶段 │ -│ 4. 文件流转（mv） │ -│ 5. Git版本管理 │ -│ 6. 归档和清理 │ -└─────────────────────────────────────────────────────────┘ - │ │ │ │ - ▼ ▼ ▼ ▼ -┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ -│ kb-link-parser│ │kb-bilibili- │ │kb-bilibili- │ │ kb-deep- │ -│ (子agent) │ │ metadata │ │ content │ │ analyzer │ -│ │ │ (子agent) │ │ (子agent) │ │ (子agent) │ -└──────────────┘ └──────────────┘ └──────────────┘ └──────────────┘ +**操作：** 检查知识库是否已存在此内容。 + +1. 根据 URL 中的 ID（如 `BVxxx`）扫描 `knowledge-base/metadata/` 和 `knowledge-base/archive/` 目录 +2. 如果存在： + - 汇报用户内容已存在及其状态 + - 询问用户是要更新还是跳过 + - 等待用户指示 +3. 如果不存在：继续步骤 2 + +### 步骤 2: 调用 kb-link-parser + +**操作：** 解析 URL，获取平台和内容 ID。 + +```python +# 调用方式 +result = sessions_spawn( + agentId="kb-link-parser", + task="""解析以下URL，将结果保存到指定目录： + +URL: [用户的URL] + +输出目录: ~/work/knowledge-base/pending +输出文件名格式: {platform}_{id}.link.json + +请解析URL中的平台和内容ID，将解析结果保存为JSON文件。 +返回结果时告诉我输出文件的完整路径。""", + timeoutSeconds=60 +) ``` -## 处理流程 +**预期输出：** `pending/{platform}_{id}.link.json` -### 完整流程（6个阶段） +### 步骤 3: 提取平台并调用对应元数据技能 +**操作：** 从文件名提取平台（如 `bilibili`），调用对应平台的元数据技能。 + +```python +# 从返回结果中提取 output_file +output_file = result.get("output_file") +platform = output_file.split("/")[-1].split("_")[0] # 从文件名提取平台 +content_id = output_file.split("/")[-1].split("_")[1].split(".")[0] ``` -用户发送"添加到知识库 + 链接" - │ - ▼ -kb-workflow: - 1. 查重（根据URL/文件名） - 2. 确认目录存在 - 3. 调用 kb-link-parser (sessions_spawn) - │ - ▼ -kb-link-parser (子agent): - 解析URL - 输出: pending/bilibili_BV1ci4y1V7tb.link.json - 返回: {"success": true/false, "output_file": "pending/..."} - │ - ▼ -kb-workflow: - 从文件名提取平台 "bilibili" → 调用 kb-bilibili-metadata - │ - ▼ -kb-bilibili-metadata (子agent): - 获取元数据 - 输出: metadata/bilibili_BV1ci4y1V7tb.meta.json - 返回: {"success": true/false, "output_file": "metadata/..."} - │ - ▼ -kb-workflow: - 调用 kb-bilibili-content - │ - ▼ -kb-bilibili-content (子agent): - 下载视频+字幕 - 输出: content/downloads/*.mp4, *.srt - 返回: {"success": true/false, "output_file": "...", "subtitles": "..."} - │ - ▼ -kb-workflow: - 拿到字幕文件 → 调用 kb-ocr-processor - │ - ▼ -kb-ocr-processor (子agent): - OCR处理 - 输出: content/bilibili_BV1ci4y1V7tb.txt - 返回: {"success": true/false, "output_file": "content/..."} - │ - ▼ -kb-workflow: - 拿到txt文件 → 调用 kb-deep-analyzer - │ - ▼ -kb-deep-analyzer (子agent): - 深度分析 - 输出: analyzed/bilibili_BV1ci4y1V7tb.md - 返回: {"success": true/false, "output_file": "analyzed/..."} - │ - ▼ -kb-workflow: - mv → archive/YYYY-MM-DD/ - git add → commit → push - 汇报完成 + +**调用元数据技能：** + +```python +if platform == "bilibili": + agent = "kb-bilibili-metadata" +elif platform == "zhihu": + agent = "kb-zhihu-metadata" + +result = sessions_spawn( + agentId=agent, + task="""获取链接的元数据： + +链接文件: [上一步的output_file] +输出目录: ~/work/knowledge-base/metadata + +请读取链接文件，提取URL，然后获取该链接的元数据（标题、作者、播放量等）。 +将元数据保存为JSON文件，文件名格式: {platform}_{id}.meta.json + +返回结果时告诉我输出文件的完整路径。""", + timeoutSeconds=120 +) +``` + +**预期输出：** `metadata/{platform}_{id}.meta.json` + +### 步骤 4: 调用内容下载技能 + +**操作：** 根据平台调用内容下载技能。 + +```python +if platform == "bilibili": + agent = "kb-bilibili-content" + +result = sessions_spawn( + agentId=agent, + task="""下载视频内容： + +元数据文件: [上一步的output_file] +下载目录: ~/work/knowledge-base/content + +请读取元数据文件，获取视频信息，然后下载视频和字幕。 +视频保存到: content/downloads/ +字幕保存到: content/subtitles/ + +文件名格式: {platform}_{id}.mp4 和 {platform}_{id}.srt + +返回结果时告诉我下载的视频文件和字幕文件的完整路径。""", + timeoutSeconds=600 +) +``` + +**预期输出：** +- `content/downloads/{platform}_{id}.mp4` +- `content/subtitles/{platform}_{id}.srt`（可能没有） + +### 步骤 5: OCR 处理 + +**操作：** 对视频或字幕进行 OCR 提取文字。 + +```python +subtitle_file = result.get("subtitle") or result.get("subtitles") +video_file = result.get("output_file") + +# 如果有字幕用字幕，没有就用视频 +ocr_source = subtitle_file or video_file + +result = sessions_spawn( + agentId="kb-ocr-processor", + task="""处理文件，提取文字内容： + +源文件: [ocr_source] +输出目录: ~/work/knowledge-base/content + +请读取文件，提取文字内容。 +输出文件名格式: {platform}_{id}.txt + +返回结果时告诉我输出文件的完整路径。""", + timeoutSeconds=1800 # 30分钟 +) +``` + +**预期输出：** `content/{platform}_{id}.txt` + +### 步骤 6: 深度分析 + +**操作：** 调用深度分析生成最终报告。 + +```python +result = sessions_spawn( + agentId="kb-deep-analyzer", + task="""深度分析内容： + +元数据文件: [步骤3的output_file] +文字内容文件: [步骤5的output_file]（如果存在，否则用元数据文件） +输出目录: ~/work/knowledge-base/analyzed + +请读取元数据和文字内容，进行深度分析。 +生成分析报告，保存为Markdown文件。 +文件名格式: {platform}_{id}.md + +分析报告应包含： +- 一句话总结 +- 核心要点 +- 时间轴（如适用） +- 推荐标签 + +返回结果时告诉我输出文件的完整路径。""", + timeoutSeconds=180 +) +``` + +**预期输出：** `analyzed/{platform}_{id}.md` + +### 步骤 7: 归档和 Git + +**操作：** 移动文件到归档目录，提交 Git。 + +```python +# 1. 创建归档目录 +date_dir = Path.home() / "work/knowledge-base/archive" / datetime.now().strftime('%Y-%m-%d') +date_dir.mkdir(parents=True, exist_ok=True) + +# 2. 移动分析报告 +analyzed_file = Path(result.get("output_file")) +archive_file = date_dir / analyzed_file.name +shutil.copy(analyzed_file, archive_file) + +# 3. 复制元数据 +meta_file = Path.home() / "work/knowledge-base/metadata" / f"{analyzed_file.stem}.meta.json" +if meta_file.exists(): + shutil.copy(meta_file, date_dir / meta_file.name) + +# 4. Git commit（如果已初始化） +git_root = Path.home() / "work/knowledge-base/archive" +subprocess.run(['git', 'add', '.'], cwd=str(git_root)) +subprocess.run(['git', 'commit', '-m', f"Archive: {analyzed_file.name}"], cwd=str(git_root)) +``` + +### 步骤 8: 汇报完成 + +**操作：** 向用户汇报处理结果。 + +```python +汇报内容： +- 视频标题 +- 处理阶段完成情况 +- 最终文件路径 +- Git commit 信息 ``` ## 目录结构 ``` -知识库根目录 KB_BASE_PATH/ -├── archive/ # Git归档目录 -│ ├── YYYY-MM-DD/ # 按日期归档 -│ └── subscriptions/ # 订阅数据 -├── pending/ # 链接解析结果 (*.link.json) -├── metadata/ # 元数据 (*.meta.json) +~/work/knowledge-base/ +├── archive/ +│ └── YYYY-MM-DD/ +│ └── {platform}_{id}.md # 分析报告 +│ └── {platform}_{id}.meta.json # 元数据 +├── pending/ +│ └── {platform}_{id}.link.json # 链接解析结果 +├── metadata/ +│ └── {platform}_{id}.meta.json # 元数据 ├── content/ -│ ├── downloads/ # 视频文件 (*.mp4) - 7天后清理 -│ └── subtitles/ # 字幕文件 (*.srt) - 7天后清理 -├── frames/ # 视频截帧 (*.jpg) - 7天后清理 -└── analyzed/ # 分析报告 (*.md) +│ ├── downloads/ +│ │ └── {platform}_{id}.mp4 # 视频文件 +│ ├── subtitles/ +│ │ └── {platform}_{id}.srt # 字幕文件 +│ └── {platform}_{id}.txt # OCR结果 +└── analyzed/ + └── {platform}_{id}.md # 分析报告 ``` -## 设计原则 - -### workflow 职责 - -| 操作 | 说明 | -|------|------| -| 查重 | 根据URL/文件名查重 | -| 创建目录 | 确保 base_path 下目录存在 | -| 调用子agent | 使用 sessions_spawn 调用子技能 | -| 提取文件名 | 从子agent返回的output_file提取 | -| 提取平台 | 从文件名第一个下划线之前提取（不读内容） | -| 传递信息 | 给子agent提示词 + 需要的路径 | -| 文件流转 | 最后归档 | -| Git | commit + push | - -### workflow 不管 - -- 不读文件内容 -- 不关心子技能内部格式 -- 子技能自己定义输入输出规范 - -### 子agent 调用方式 - -```python -# kb-workflow 调用子agent -result = sessions_spawn( - agentId="子技能名称", - task="提示词 + 需要的路径信息", - timeoutSeconds=120 -) -``` - -### 子技能返回格式 - -所有子技能应返回JSON： - -```json -{ - "success": true, - "output_file": "完整文件路径", - "subtitles": "字幕文件路径（可选）" -} -``` - -## 环境变量 +## 环境变量（可选） ```bash -export KB_BASE_PATH="/path/to/knowledge-base" # 知识库根目录 -export KB_GIT_ROOT="/path/to/archive" # Git仓库目录 -export KB_GIT_REMOTE="git@github.com:user/repo.git" # 远程仓库 -export KB_GIT_USER="Username" # 用户名 -export KB_GIT_EMAIL="user@example.com" # 邮箱 +export KB_BASE_PATH="~/work/knowledge-base" # 知识库根目录 ``` -## Usage +## 支持的平台 -### 作为 OpenClaw 技能 +| 平台 | 元数据技能 | 内容技能 | +|------|-----------|---------| +| Bilibili | kb-bilibili-metadata | kb-bilibili-content | +| 知乎 | kb-zhihu-metadata | - | -当用户发送消息时自动触发： -``` -用户: "添加到知识库 https://www.bilibili.com/video/BVxxx" -``` +## 注意事项 -### 作为子代理 - -```python -# 从其他技能调用 -result = sessions_spawn( - agentId="kb-workflow", - task="添加到知识库 https://www.bilibili.com/video/BVxxx" -) -``` - -### 命令行 - -```bash -# 处理单个URL -python kb_workflow.py --url "https://www.bilibili.com/video/BVxxx" -``` - -## 子技能集成 - -| 阶段 | 子技能 | 输出文件 | -|------|--------|----------| -| 链接解析 | kb-link-parser | pending/{platform}_{id}.link.json | -| 元数据 | kb-bilibili-metadata | metadata/{platform}_{id}.meta.json | -| 内容 | kb-bilibili-content | content/downloads/{platform}_{id}.mp4, content/subtitles/{platform}_{id}.srt | -| OCR | kb-ocr-processor | content/{platform}_{id}.txt | -| 分析 | kb-deep-analyzer | analyzed/{platform}_{id}.md | - -## Related Skills - -- **kb-link-parser** - 链接解析 -- **kb-bilibili-metadata** - B站元数据 -- **kb-bilibili-content** - B站内容获取 -- **kb-deep-analyzer** - 深度分析 -- **kb-ocr-processor** - OCR处理 +1. **文件名提取平台：** 从文件名第一个下划线之前提取（如 `bilibili_BVxxx.link.json` → `bilibili`） +2. **不读文件内容：** workflow 只根据文件名判断，不读取文件内部 +3. **超时设置：** OCR 步骤较长，建议 30 分钟超时 +4. **子代理权限：** 确保子代理已在 allowlist 中配置 diff --git a/skills/kb-workflow/SkillBug.md b/skills/kb-workflow/SkillBug.md index 7d38f50..05b02c0 100644 --- a/skills/kb-workflow/SkillBug.md +++ b/skills/kb-workflow/SkillBug.md @@ -2,98 +2,75 @@ ## 问题记录 -### 2026-02-10: 缺少自动写入pending目录的功能 +### 2026-02-10: 子代理权限问题 -**状态：** ✅ 已修复 +**问题：** `sessions_spawn` 调用子代理时返回 `forbidden` 错误。 -**解决方案：** -1. 创建 `skill_main.py` 作为技能入口 -2. 创建 `kb_workflow.py` 实现完整流程 -3. 接收用户消息后自动处理 +**原因：** OpenClaw 配置中 allowlist 为空，Python 脚本没有权限调用子代理。 + +**解决方案：** 改用 Prompt-based 模式（方案B）。 --- -### 2026-02-10: kb-workflow 自己实现链接解析而非调用子技能 +## 架构重构 -**状态：** ✅ 已修复 +### 之前（Script-based） -**解决方案：** -重构 `kb_workflow.py`： -1. 使用 `sessions_spawn` 调用子agent -2. 不自己实现链接解析，调用 `kb-link-parser` -3. 从文件名提取平台（不读内容） - -```python -# 调用子agent -result = sessions_spawn( - agentId="kb-link-parser", - task="解析URL，输出到pending目录" -) +``` +kb_workflow.py (Python脚本) + ↓ +sessions_spawn() # ❌ 失败（脚本没有权限） ``` ---- +### 之后（Prompt-based） -### 2026-02-10: _extract_metadata 是空壳 - -**状态：** ✅ 已修复 - -**解决方案：** -实现真正的子agent调用： -```python -def call_metadata_agent(platform, link_file, metadata_dir): - agent_id = METADATA_AGENTS.get(platform) # bilibili → kb-bilibili-metadata - return call_subagent(agent_id, task, timeout=120) ``` - ---- - -## 重构总结 - -### 架构变化 - -**之前（错误）：** +主代理读取 SKILL.md + ↓ +按照指导调用 sessions_spawn # ✅ 成功（主代理有权限） ``` -kb_workflow.py (自己造轮子) -├── _extract_link() - 自己解析URL -├── _extract_metadata() - 空壳 -``` - -**之后（正确）：** -``` -kb-workflow (主代理) -├── sessions_spawn() 调用子agent -│ ├── kb-link-parser → pending/*.link.json -│ ├── kb-bilibili-metadata → metadata/*.meta.json -│ ├── kb-bilibili-content → content/downloads/*.mp4 -│ ├── kb-ocr-processor → content/*.txt -│ └── kb-deep-analyzer → analyzed/*.md -└── 文件流转 + Git归档 -``` - -### 设计原则 - -| workflow 管 | workflow 不管 | -|-------------|---------------| -| 查重 | 读文件内容 | -| 调用子agent | 子技能内部格式 | -| 提取文件名（从返回结果） | 子技能逻辑 | -| 提取平台（从文件名） | | -| 传递路径信息 | | -| Git归档 | | ### 文件变更 -| 文件 | 操作 | 说明 | +| 文件 | 变更 | 说明 | |------|------|------| -| `kb_workflow.py` | 重构 | 使用 sessions_spawn 调用子agent | -| `skill_main.py` | 新增 | 技能入口 | -| `SKILL.md` | 更新 | 明确架构和流程 | +| `SKILL.md` | 重写 | 详细定义 workflow 步骤，主代理直接读取执行 | +| `kb_workflow.py` | 简化 | 保留为辅助工具函数库（查重、归档、Git） | +| `skill_main.py` | 简化 | 只做触发检测，返回指导信息 | +| `SkillBug.md` | 更新 | 记录架构变更 | + +### SKILL.md 结构 + +``` +1. 触发条件 +2. 步骤 1: 查重 +3. 步骤 2: kb-link-parser +4. 步骤 3: 元数据技能 +5. 步骤 4: 内容下载 +6. 步骤 5: OCR（30分钟超时） +7. 步骤 6: 深度分析 +8. 步骤 7: 归档和 Git +9. 步骤 8: 汇报完成 +10. 目录结构 +11. 支持平台 +12. 注意事项 +``` + +### kb_workflow.py 保留的函数 + +```python +extract_platform(filename) # 从文件名提取平台 +extract_base_name(filename) # 从文件名提取基础名称 +get_knowledge_base_path() # 获取知识库根目录 +ensure_directories(base_path) # 确保目录存在 +check_duplicate(url) # 查重 +archive_and_git(analyzed_file) # 归档和 Git +generate_summary(...) # 生成汇报 +``` --- ## 待办 -- [ ] 测试完整的处理流程 -- [ ] 验证 sessions_spawn 调用是否正常 -- [ ] 完善错误处理 -- [ ] 添加单元测试 +- [ ] 测试 Prompt-based 模式是否正常工作 +- [ ] 验证主代理能否正确执行 SKILL.md 中的步骤 diff --git a/skills/kb-workflow/kb_workflow.py b/skills/kb-workflow/kb_workflow.py index 2ae3b94..9cac8e2 100644 --- a/skills/kb-workflow/kb_workflow.py +++ b/skills/kb-workflow/kb_workflow.py @@ -3,21 +3,10 @@ """ KB Workflow - Knowledge Base Workflow Orchestrator -知识库工作流编排技能。 +知识库工作流辅助工具函数。 -核心职责： -1. 接收用户消息，识别触发词和链接 -2. 查重（根据URL/文件名） -3. 调用子agent处理各阶段 -4. 文件流转（mv） -5. Git版本管理 -6. 归档和清理 - -设计原则： -- workflow不读文件内容 -- 从文件名提取平台（第一个下划线之前） -- 子技能自己定义输入输出格式 -- workflow只传递提示词 + 需要的路径信息 +此模块提供查重、文件操作、Git 管理等工具函数， +供主代理在执行 SKILL.md 中定义的 workflow 时调用。 """ import os @@ -31,20 +20,21 @@ from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Any + # ==================== 子技能映射 ==================== -LINK_PARSER_AGENT = "kb-link-parser" METADATA_AGENTS = { "bilibili": "kb-bilibili-metadata", "zhihu": "kb-zhihu-metadata", } + CONTENT_AGENTS = { "bilibili": "kb-bilibili-content", } -OCR_AGENT = "kb-ocr-processor" -ANALYZER_AGENT = "kb-deep-analyzer" +# ==================== 工具函数 ==================== + def extract_platform(filename: str) -> str: """从文件名提取平台（第一个下划线之前）""" match = re.match(r'^([a-zA-Z0-9]+)_(.+)$', filename) @@ -58,9 +48,21 @@ def extract_base_name(filename: str) -> str: return Path(filename).stem -def ensure_directories(base_path: Path) -> Dict[str, Path]: +def get_knowledge_base_path() -> Path: + """获取知识库根目录""" + base_path_str = os.environ.get('KB_BASE_PATH', '') + if base_path_str: + return Path(base_path_str).expanduser() + return Path.home() / "work" / "knowledge-base" + + +def ensure_directories(base_path: Path = None) -> Dict[str, Path]: """确保目录存在""" + if base_path is None: + base_path = get_knowledge_base_path() + dirs = { + "base": base_path, "pending": base_path / "pending", "metadata": base_path / "metadata", "content": base_path / "content", @@ -72,19 +74,24 @@ def ensure_directories(base_path: Path) -> Dict[str, Path]: } for name, dir_path in dirs.items(): - dir_path.mkdir(parents=True, exist_ok=True) + if isinstance(dir_path, Path): + dir_path.mkdir(parents=True, exist_ok=True) return dirs -def check_duplicate(base_path: Path, url: str) -> Dict[str, Any]: +def check_duplicate(base_path: Path = None, url: str = None, platform: str = None, content_id: str = None) -> Dict[str, Any]: """查重检查""" + if base_path is None: + base_path = get_knowledge_base_path() + metadata_dir = base_path / "metadata" archive_dir = base_path / "archive" # 从URL提取ID - bv_match = re.search(r'BV[a-zA-Z0-9]+', url) - content_id = bv_match.group() if bv_match else None + if url and not content_id: + bv_match = re.search(r'BV[a-zA-Z0-9]+', url) + content_id = bv_match.group() if bv_match else None result = {"duplicate": False, "status": None, "files": []} @@ -107,364 +114,26 @@ def check_duplicate(base_path: Path, url: str) -> Dict[str, Any]: return result -def call_subagent(agent_id: str, task: str, timeout: int = 120) -> Dict[str, Any]: - """调用子agent（通过 OpenClaw sessions_spawn 或命令行）""" - print(f"\n{'='*50}") - print(f"调用子技能: {agent_id}") - print(f"{'='*50}") - - # 尝试使用 sessions_spawn（OpenClaw 环境） - try: - from openclaw_tool import sessions_spawn - result = sessions_spawn( - agentId=agent_id, - task=task, - timeoutSeconds=timeout - ) - print(f"\n子技能返回结果:") - print(json.dumps(result, ensure_ascii=False, indent=2)) - return result - except ImportError: - # 回退到命令行调用 - pass - - # 命令行回退：调用子技能的脚本 - print(f"使用命令行方式调用: {agent_id}") - - # 技能目录路径 - skills_path = Path(__file__).resolve().parent.parent - - # 从任务中提取关键信息 - import re - - # 提取URL - url_match = re.search(r'URL: (\S+)', task) - url = url_match.group(1) if url_match else "" - - # 提取输出目录 - output_match = re.search(r'输出目录: (\S+)', task) - output_dir = output_match.group(1) if output_match else "" - - # 提取输出文件名格式 - filename_match = re.search(r'输出文件名格式: (\S+)', task) - filename_format = filename_match.group(1) if filename_match else "{platform}_{id}.json" - - # 提取链接文件路径 - link_file_match = re.search(r'链接文件: (\S+)', task) - link_file = link_file_match.group(1) if link_file_match else "" - - # 提取元数据文件路径 - meta_file_match = re.search(r'元数据文件: (\S+)', task) - meta_file = meta_file_match.group(1) if meta_file_match else "" - - # kb-link-parser - if agent_id == "kb-link-parser": - cmd = ['python3', str(skills_path / "kb-link-parser" / "link-parser.py"), '--url', url] - try: - proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) - if proc.returncode == 0: - try: - data = json.loads(proc.stdout) - if data.get('success'): - platform = data.get('platform', 'unknown') - content_id = data.get('id', 'unknown') - ext = '.link.json' - filename = f"{platform}_{content_id}{ext}" - output_file = Path(output_dir) / filename - output_file.parent.mkdir(parents=True, exist_ok=True) - with open(output_file, 'w', encoding='utf-8') as f: - json.dump(data, f, ensure_ascii=False, indent=2) - print(f"保存文件: {output_file}") - return {"success": True, "output_file": str(output_file)} - else: - return {"success": False, "error": data.get('error', 'unknown')} - except json.JSONDecodeError: - return {"success": False, "error": f"解析失败: {proc.stdout[:200]}"} - else: - return {"success": False, "error": proc.stderr} - except Exception as e: - return {"success": False, "error": str(e)} - - # kb-bilibili-metadata - elif agent_id == "kb-bilibili-metadata": - # 从链接文件或URL中提取BV号 - bv = None - if link_file and Path(link_file).exists(): - with open(link_file, 'r', encoding='utf-8') as f: - link_data = json.load(f) - bv = link_data.get('id') - if not bv: - bv_match = re.search(r'(BV[a-zA-Z0-9]+)', url or link_file) - bv = bv_match.group(1) if bv_match else None - - if not bv: - return {"success": False, "error": "无法提取BV号"} - - # 生成输出文件名 - filename = f"bilibili_{bv}.meta.json" - output_file = Path(output_dir) / filename - output_file.parent.mkdir(parents=True, exist_ok=True) - - cmd = [ - 'python3', str(skills_path / "kb-bilibili-metadata" / "bilibili_metadata.py"), - '--id', bv, - '--type', 'video', - '--output', str(output_file) - ] - - try: - proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) - if proc.returncode == 0: - print(f"保存文件: {output_file}") - return {"success": True, "output_file": str(output_file)} - else: - return {"success": False, "error": proc.stderr} - except Exception as e: - return {"success": False, "error": str(e)} - - # kb-bilibili-content - elif agent_id == "kb-bilibili-content": - bv_match = re.search(r'(BV[a-zA-Z0-9]+)', url or meta_file) - bv = bv_match.group(1) if bv_match else None - - if not bv: - return {"success": False, "error": "无法提取BV号"} - - downloads_dir = Path(output_dir) / "downloads" - subtitles_dir = Path(output_dir) / "subtitles" - downloads_dir.mkdir(parents=True, exist_ok=True) - subtitles_dir.mkdir(parents=True, exist_ok=True) - - video_file = None - subtitle_file = None - - # 下载视频 - cmd_video = [ - 'python3', str(skills_path / "kb-bilibili-content" / "bilibili_content.py"), - '--id', bv, - '--type', 'video', - '--output', output_dir - ] - - try: - proc = subprocess.run(cmd_video, capture_output=True, text=True, timeout=timeout) - if proc.returncode == 0: - # 查找生成的视频文件（可能包含BV号） - for f in Path(output_dir).iterdir(): - if f.is_file() and bv in f.name and f.suffix in ['.flv', '.mp4']: - # 重命名为标准格式 - new_name = f"bilibili_{bv}{f.suffix}" - new_path = downloads_dir / new_name - f.rename(new_path) - video_file = str(new_path) - break - except Exception as e: - print(f"视频下载警告: {e}") - - # 下载字幕 - cmd_subtitle = [ - 'python3', str(skills_path / "kb-bilibili-content" / "bilibili_content.py"), - '--id', bv, - '--type', 'subtitle', - '--output', output_dir - ] - - try: - proc = subprocess.run(cmd_subtitle, capture_output=True, text=True, timeout=timeout) - if proc.returncode == 0: - # 查找生成的字幕文件 - for f in Path(output_dir).iterdir(): - if f.is_file() and bv in f.name and f.suffix in ['.srt', '.lrc', '.ass']: - # 重命名为标准格式 - new_name = f"bilibili_{bv}{f.suffix}" - new_path = subtitles_dir / new_name - f.rename(new_path) - subtitle_file = str(new_path) - break - except Exception as e: - print(f"字幕下载警告: {e}") - - if video_file or subtitle_file: - return {"success": True, "output_file": video_file, "subtitles": subtitle_file} - else: - return {"success": False, "error": "下载未生成任何文件"} - - # kb-ocr-processor - elif agent_id == "kb-ocr-processor": - # 查找字幕文件 - subtitles_dir = Path(output_dir) / "subtitles" - subtitle_file = None - if subtitles_dir.exists(): - for f in subtitles_dir.iterdir(): - if f.suffix in ['.srt', '.lrc']: - subtitle_file = str(f) - break - - if not subtitle_file: - return {"success": False, "error": "未找到字幕文件"} - - # 生成输出文件名 - text_file = Path(output_dir) / f"{Path(subtitle_file).stem}.txt" - - cmd = [ - 'python3', str(skills_path / "kb-ocr-processor" / "ocr_processor.py"), - '--input', subtitle_file, - '--output', str(text_file) - ] - - try: - proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) - if proc.returncode == 0: - print(f"保存文件: {text_file}") - return {"success": True, "output_file": str(text_file)} - else: - return {"success": False, "error": proc.stderr} - except Exception as e: - return {"success": False, "error": str(e)} - - # kb-deep-analyzer - elif agent_id == "kb-deep-analyzer": - # 生成输出文件名 - meta_name = Path(meta_file).stem if meta_file else "unknown" - filename = f"{meta_name}.md" - output_file = Path(output_dir) / filename - output_file.parent.mkdir(parents=True, exist_ok=True) - - # 从任务中提取文字内容文件路径 - text_file_match = re.search(r'文字内容文件: (\S+)', task) - text_content_file = text_file_match.group(1) if text_file_match else "" - - # 如果没有文字内容文件，使用 metadata 作为 content - content_file = meta_file - if text_content_file and Path(text_content_file).exists(): - # 从文字文件创建 content JSON - content_json = { - "type": "transcript", - "source": text_content_file, - "text": Path(text_content_file).read_text(encoding='utf-8')[:10000] if Path(text_content_file).exists() else "" - } - content_file = Path(output_dir) / f"{meta_name}.content.json" - with open(content_file, 'w', encoding='utf-8') as f: - json.dump(content_json, f, ensure_ascii=False, indent=2) - - cmd = [ - 'python3', str(skills_path / "kb-deep-analyzer" / "deep_analyzer.py"), - '--metadata', meta_file or output_dir, - '--content', str(content_file), - '--output', str(output_file) - ] - - try: - proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) - if proc.returncode == 0: - print(f"保存文件: {output_file}") - return {"success": True, "output_file": str(output_file)} - else: - return {"success": False, "error": proc.stderr} - except Exception as e: - return {"success": False, "error": str(e)} - - return {"success": False, "error": f"暂未实现命令行回退: {agent_id}"} - - -def call_link_parser(url: str, pending_dir: Path) -> Dict[str, Any]: - """调用 kb-link-parser""" - task = f"""解析以下URL，将结果保存到指定目录： - -URL: {url} - -输出目录: {pending_dir} -输出文件名格式: {{platform}}_{{id}}.link.json - -请解析URL中的平台和内容ID，将解析结果保存为JSON文件。 -返回结果时告诉我输出文件的完整路径。 - -只需要解析URL并保存结果，不需要下载内容。""" - - return call_subagent(LINK_PARSER_AGENT, task, timeout=60) - - -def call_metadata_agent(platform: str, link_file: Path, metadata_dir: Path) -> Dict[str, Any]: - """调用对应的元数据获取技能""" - agent_id = METADATA_AGENTS.get(platform) - if not agent_id: - return {"success": False, "error": f"不支持的平台: {platform}"} - - task = f"""获取链接的元数据： - -链接文件: {link_file} -输出目录: {metadata_dir} - -请读取链接文件，提取URL，然后获取该链接的元数据（标题、作者、播放量等）。 -将元数据保存为JSON文件，文件名格式: {{platform}}_{{id}}.meta.json - -返回结果时告诉我输出文件的完整路径。""" - - return call_subagent(agent_id, task, timeout=120) - - -def call_content_agent(platform: str, meta_file: Path, content_dir: Path) -> Dict[str, Any]: - """调用对应的内容获取技能""" - agent_id = CONTENT_AGENTS.get(platform) - if not agent_id: - return {"success": False, "error": f"不支持的平台: {platform}"} - - task = f"""下载视频内容： - -元数据文件: {meta_file} -下载目录: {content_dir} - -请读取元数据文件，获取视频信息，然后下载视频和字幕。 -视频保存到: content/downloads/ -字幕保存到: content/subtitles/ - -文件名格式: {{platform}}_{{id}}.mp4 和 {{platform}}_{{id}}.srt - -返回结果时告诉我下载的视频文件和字幕文件的完整路径。""" - - return call_subagent(agent_id, task, timeout=600) - - -def call_ocr_processor(source_file: Path, content_dir: Path, timeout: int = 1800) -> Dict[str, Any]: - """调用 OCR 处理""" - task = f"""处理文件，提取文字内容： - -源文件: {source_file} -输出目录: {content_dir} - -请读取文件，提取文字内容。 -输出文件名格式: {{platform}}_{{id}}.txt - -返回结果时告诉我输出文件的完整路径。""" - - return call_subagent(OCR_AGENT, task, timeout=timeout) - - -def call_deep_analyzer(meta_file: Path, text_file: Path, analyzed_dir: Path) -> Dict[str, Any]: - """调用深度分析""" - task = f"""深度分析内容： - -元数据文件: {meta_file} -文字内容文件: {text_file} -输出目录: {analyzed_dir} - -请读取元数据和文字内容，进行深度分析。 -生成分析报告，保存为Markdown文件。 -文件名格式: {{platform}}_{{id}}.md - -分析报告应包含： -- 一句话总结 -- 核心要点 -- 时间轴（如适用） -- 推荐标签 - -返回结果时告诉我输出文件的完整路径。""" - - return call_subagent(ANALYZER_AGENT, task, timeout=180) - - -def archive_and_git(base_path: Path, analyzed_file: Path, git_root: Path, git_user: str, git_email: str, git_remote: str) -> Dict[str, Any]: +def archive_and_git( + analyzed_file: str, + base_path: Path = None, + git_root: Path = None, + git_user: str = None, + git_email: str = None, + git_remote: str = None +) -> Dict[str, Any]: """归档并提交Git""" + if base_path is None: + base_path = get_knowledge_base_path() + if git_root is None: + git_root = base_path / "archive" + if git_user is None: + git_user = os.environ.get('KB_GIT_USER', '') + if git_email is None: + git_email = os.environ.get('KB_GIT_EMAIL', '') + if git_remote is None: + git_remote = os.environ.get('KB_GIT_REMOTE', '') + result = { "success": False, "archive_path": "", @@ -473,239 +142,124 @@ def archive_and_git(base_path: Path, analyzed_file: Path, git_root: Path, git_us "error": None } - # 1. 创建归档目录 - date_str = datetime.now().strftime('%Y-%m-%d') - date_dir = base_path / "archive" / date_str - date_dir.mkdir(parents=True, exist_ok=True) - - # 2. 复制文件到归档目录 - archive_file = date_dir / analyzed_file.name - shutil.copy(analyzed_file, archive_file) - result["archive_path"] = str(archive_file) - - # 3. 复制元数据 - meta_file = base_path / "metadata" / f"{extract_base_name(analyzed_file.name)}.meta.json" - if meta_file.exists(): - shutil.copy(meta_file, date_dir / meta_file.name) - - # 4. Git操作 - git_dir = git_root / ".git" - if not git_dir.exists(): - subprocess.run(['git', 'init'], cwd=str(git_root), capture_output=True) - if git_user: - subprocess.run(['git', 'config', 'user.name', git_user], cwd=str(git_root)) - if git_email: - subprocess.run(['git', 'config', 'user.email', git_email], cwd=str(git_root)) - - if (git_root / ".git").exists(): - subprocess.run(['git', 'add', '.'], cwd=str(git_root), capture_output=True) - commit_msg = f"Archive: {analyzed_file.name} - {datetime.now().isoformat()}" - commit_result = subprocess.run( - ['git', 'commit', '-m', commit_msg], - cwd=str(git_root), capture_output=True, text=True - ) - result["git_commit"] = commit_result.returncode == 0 - - if git_remote and commit_result.returncode == 0: - push_result = subprocess.run( - ['git', 'push', 'origin', 'master'], - cwd=str(git_root), capture_output=True, text=True, timeout=60 - ) - result["git_push"] = {"success": push_result.returncode == 0} - - result["success"] = True - return result - - -def process_url(url: str, base_path: Path = None) -> Dict[str, Any]: - """处理单个URL的完整流程""" - if not base_path: - base_path = Path.home() / "work" / "knowledge-base" - - git_root = os.environ.get('KB_GIT_ROOT', str(base_path / 'archive')) - git_user = os.environ.get('KB_GIT_USER', '') - git_email = os.environ.get('KB_GIT_EMAIL', '') - git_remote = os.environ.get('KB_GIT_REMOTE', '') - - dirs = ensure_directories(base_path) - - result = { - "url": url, - "success": False, - "stages": [], - "final_file": None, - "archive": None, - "error": None - } - try: - # 查重 - print(f"\n{'='*50}") - print(f"步骤1: 查重") - print(f"{'='*50}") + # 1. 创建归档目录 + date_str = datetime.now().strftime('%Y-%m-%d') + date_dir = base_path / "archive" / date_str + date_dir.mkdir(parents=True, exist_ok=True) - dup_result = check_duplicate(base_path, url) - if dup_result["duplicate"]: - result["error"] = f"已存在: {dup_result['status']}" - result["duplicate"] = True - result["existing_files"] = dup_result["files"] - return result + # 2. 复制分析报告 + src_file = Path(analyzed_file) + archive_file = date_dir / src_file.name + shutil.copy(src_file, archive_file) + result["archive_path"] = str(archive_file) - # kb-link-parser - print(f"\n{'='*50}") - print(f"步骤2: 链接解析 (kb-link-parser)") - print(f"{'='*50}") + # 3. 复制元数据 + meta_file = base_path / "metadata" / f"{src_file.stem}.meta.json" + if meta_file.exists(): + shutil.copy(meta_file, date_dir / meta_file.name) - link_result = call_link_parser(url, dirs["pending"]) - if not link_result.get("success"): - result["error"] = f"链接解析失败: {link_result.get('error')}" - return result + # 4. Git 操作 + git_dir = git_root / ".git" + if not git_dir.exists(): + subprocess.run(['git', 'init'], cwd=str(git_root), capture_output=True) + if git_user: + subprocess.run(['git', 'config', 'user.name', git_user], cwd=str(git_root)) + if git_email: + subprocess.run(['git', 'config', 'user.email', git_email], cwd=str(git_root)) - link_file = link_result.get("output_file") - if not link_file: - result["error"] = "链接解析未返回输出文件" - return result - - link_file = Path(link_file) - result["stages"].append({"stage": "link_parsing", "file": str(link_file)}) - - # 提取平台 - filename = link_file.name - platform = extract_platform(filename) - content_id = extract_base_name(filename) - - print(f"\n平台: {platform}") - print(f"内容ID: {content_id}") - - # 元数据 - print(f"\n{'='*50}") - print(f"步骤3: 元数据提取 ({METADATA_AGENTS.get(platform, 'unknown')})") - print(f"{'='*50}") - - meta_result = call_metadata_agent(platform, link_file, dirs["metadata"]) - if not meta_result.get("success"): - result["error"] = f"元数据获取失败: {meta_result.get('error')}" - return result - - meta_file = meta_result.get("output_file") - if not meta_file: - result["error"] = "元数据获取未返回输出文件" - return result - - meta_file = Path(meta_file) - result["stages"].append({"stage": "metadata", "file": str(meta_file)}) - - # 内容下载 - print(f"\n{'='*50}") - print(f"步骤4: 内容下载 ({CONTENT_AGENTS.get(platform, 'unknown')})") - print(f"{'='*50}") - - content_result = call_content_agent(platform, meta_file, dirs["content"]) - if not content_result.get("success"): - result["error"] = f"内容下载失败: {content_result.get('error')}" - return result - - video_file = content_result.get("output_file") - subtitle_file = content_result.get("subtitle") or content_result.get("subtitles") - - result["stages"].append({"stage": "content", "video": video_file, "subtitles": subtitle_file}) - - # OCR - text_file = None - print(f"\n{'='*50}") - print(f"步骤5: OCR处理 (kb-ocr-processor)") - print(f"{'='*50}") - - # 无论有没有字幕，都尝试OCR处理 - ocr_source = subtitle_file - if not ocr_source and video_file: - ocr_source = video_file - - if ocr_source: - # OCR超时设置为30分钟 - ocr_result = call_ocr_processor(Path(ocr_source), dirs["content"], timeout=1800) - text_file = ocr_result.get("output_file") if ocr_result.get("success") else None + if (git_root / ".git").exists(): + subprocess.run(['git', 'add', '.'], cwd=str(git_root), capture_output=True) + commit_msg = f"Archive: {src_file.name} - {datetime.now().isoformat()}" + commit_result = subprocess.run( + ['git', 'commit', '-m', commit_msg], + cwd=str(git_root), capture_output=True, text=True + ) + result["git_commit"] = commit_result.returncode == 0 - if text_file: - result["stages"].append({"stage": "ocr", "file": text_file}) - else: - print(f"OCR处理未生成文件") - else: - print(f"无字幕文件，跳过OCR处理") + if git_remote and commit_result.returncode == 0: + push_result = subprocess.run( + ['git', 'push', 'origin', 'master'], + cwd=str(git_root), capture_output=True, text=True, timeout=60 + ) + result["git_push"] = {"success": push_result.returncode == 0} - # 深度分析 - print(f"\n{'='*50}") - print(f"步骤6: 深度分析 (kb-deep-analyzer)") - print(f"{'='*50}") - - analyze_result = call_deep_analyzer( - meta_file, - Path(text_file) if text_file else meta_file, - dirs["analyzed"] - ) - if not analyze_result.get("success"): - result["error"] = f"深度分析失败: {analyze_result.get('error')}" - return result - - analyzed_file = analyze_result.get("output_file") - if not analyzed_file: - result["error"] = "深度分析未返回输出文件" - return result - - analyzed_file = Path(analyzed_file) - result["stages"].append({"stage": "analyzer", "file": str(analyzed_file)}) - result["final_file"] = str(analyzed_file) - - # 归档和Git - print(f"\n{'='*50}") - print(f"步骤7: 归档和Git") - print(f"{'='*50}") - - archive_result = archive_and_git( - base_path, analyzed_file, - Path(git_root), git_user, git_email, git_remote - ) - - result["archive"] = archive_result result["success"] = True - - print(f"\n{'='*50}") - print(f"处理完成！") - print(f"{'='*50}") except Exception as e: result["error"] = str(e) - import traceback - traceback.print_exc() return result +def generate_summary(url: str, stages: List[Dict], final_file: str, archive_result: Dict = None) -> str: + """生成处理完成汇报""" + # 提取视频标题（如果有） + title_match = re.search(r'com/video/([^?]+)', url) + title = title_match.group(1) if title_match else "未知" + + summary = f""" +✅ **处理完成！** + +**视频:** {title} + +**处理阶段:** +""" + for stage in stages: + stage_name = stage.get('stage', 'unknown') + emoji = "✓" if stage.get('success') else "○" + summary += f"{emoji} {stage_name}\n" + + if final_file: + summary += f"\n**最终文件:** `{final_file}`" + + if archive_result: + if archive_result.get('git_commit'): + summary += f"\n**Git:** 已提交" + if archive_result.get('git_push', {}).get('success'): + summary += " ✓" + elif archive_result.get('git_push'): + summary += " ✗" + + return summary + + +# ==================== CLI ==================== + def main(): - """CLI入口""" + """CLI入口 - 用于命令行测试""" parser = argparse.ArgumentParser( description='KB Workflow - Knowledge Base Workflow Orchestrator', - formatter_class=argparse.RawDescriptionHelpFormatter + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +使用示例: + # 查重 + python kb_workflow.py --check --url "https://www.bilibili.com/video/BVxxx" + + # 归档 + python kb_workflow.py --archive "/path/to/analyzed.md" + """ ) - parser.add_argument('--url', type=str, help='要处理的URL') + parser.add_argument('--check', action='store_true', help='查重') + parser.add_argument('--url', type=str, help='要检查的URL') + parser.add_argument('--archive', type=str, help='要归档的文件路径') parser.add_argument('--base-path', type=str, help='知识库根目录') args = parser.parse_args() - if not args.url: - print("请提供URL: --url https://...") - sys.exit(1) - base_path = Path(args.base_path) if args.base_path else None - result = process_url(args.url, base_path) - print(f"\n{'='*50}") - print("KB Workflow Result") - print(f"{'='*50}") - print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) + if args.check and args.url: + result = check_duplicate(base_path, args.url) + print(json.dumps(result, ensure_ascii=False, indent=2)) - sys.exit(0 if result.get('success') else 1) + elif args.archive: + result = archive_and_git(args.archive, base_path) + print(json.dumps(result, ensure_ascii=False, indent=2)) + + else: + parser.print_help() if __name__ == '__main__': diff --git a/skills/kb-workflow/skill_main.py b/skills/kb-workflow/skill_main.py index 80c3ba5..478b998 100644 --- a/skills/kb-workflow/skill_main.py +++ b/skills/kb-workflow/skill_main.py @@ -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)