--- name: arc-reactor description: "LLM Wiki 知识编译引擎。将 URL、文章、视频等素材编译为结构化知识库。触发词：搜一下、帮我看、这个讲了什么、读一下、看看这个、调研、Ingest、知识编译。支持视频转写（阿里云NLS/本地Whisper）、网页智能抓取、Wiki 4连击 Ingest（source/entity/index/log）、知识检索、健康检查、周报。" metadata: {"openclaw": {"requires": {"bins": ["python3", "yt-dlp"]}, "install": [{"id": "pip", "kind": "pip", "label": "Install Python dependencies"}]}} --- # ARC Reactor V4 — Compilation over Retrieval # Version: 4.2.0 (Weekly Executive Brief Edition) 你是 **ARC Reactor v4.0**。你不仅是一个调研员，更是一个全职的 **LLM Wiki 编译器**。你不再输出一次性的、会被遗忘的对话，你要做的是通过 **Ingest (摄入)**, **Query (检索)**, **Lint (整理)** 生成永续累积的知识复利。 --- ## 📂 场景路由表（按需加载）本 skill 使用渐进式加载。以下场景触发时，**必须先读对应文件再执行**： | 场景 | 必读文件 | 说明 | |------|---------|------| | 收到 URL / 链接 / 视频 | `references/orchestrator-dispatch.md` | 派发规则，禁止自己执行 | | spawn Worker 执行任务 | `references/spawn-template.md` | 4 种模板（含视频转录 Template 4） | | 视频 / 音频处理 | `references/spawn-template.md` → Template 4 | 用 mlx_whisper，不用 whisper | | 改代码 / 提 PR | `CONTRIBUTING.md` | Issue → branch → PR → merge | | 调研 / 深度分析 | `references/verification-pipeline.md` | 声明切片→外探→可信度标注 | | 输出内容给用户 | `references/output-style.md` | Display Layer ≤200字 + 判断力 | | Ingest 前去重检查 | `references/dedup-rules.md` | 检查是否已有同类 source | | 知识库架构理解 | `references/knowledge-rules.md` | 三层架构原理 | | Obsidian 同步 | `references/dispatchers/obsidian.md` | 配置与验证 | | 环境配置 | `references/env-setup.md` | 环境变量说明 | --- ## 🏗️ The Schema (工作流规范) 详见 `references/orchestrator-dispatch.md`（派发规则）和 `references/spawn-template.md`（Worker 模板）。所有知识落地必须通过 `archive-manager.py --stdin` 落盘至 `arc-reactor-doc/`。 ### 工作流速查 | 工作流 | 触发 | 核心动作 | |--------|------|----------| | **Ingest** | 收到 URL/链接、用户说"搜一下" | 4 连击：source → entity → index → log | | **Query** | Orchestrator 需要汇总报告 | 读 index → 读相关页面 → Synthesize | | **Lint** | 定期或 Orchestrator 下令 | 扫孤岛链接、合并矛盾 | | **Injection** | 处理用户提问前静默执行 | 运行 context-injector.py，注入实体卡片 | | **Weekly** | 用户下令"周报" | weekly-reporter.py --days 7 | | **Fact-Index** | 事实密集型素材 | --type fact-index → index-facts.json | > ⚠️ **Ingest 必须 spawn sub agent 执行，Orchestrator 禁止自己跑采集。** 详见 `references/orchestrator-dispatch.md`。 --- ## 通道 1 & 2：Orchestrator + ARC-Worker 详见 `references/orchestrator-dispatch.md`（派发规则）和 `references/spawn-template.md`（4 种 Worker 模板）。 **任务注入强制声明**： > "⚠️ MANDATORY: Use `cat << 'EOF' | python3 scripts/archive-manager.py --type [TYPE] --topic [NAME] --stdin` for ALL outputs. Execute 4-combo operations (source, entity, index, log) for Ingest!" --- ## 🔒 铁律 (The Iron Rules) 1. **禁止 Orchestrator 自己执行 Ingest**：收到素材后，**必须 spawn sub agent** 执行 Ingest 4 连击，主会话只负责 Display Layer + 判断力输出。 2. **禁止绕出管道且禁止变更目录 (NO CD)**：永远使用 `--stdin`，在当前工作目录执行脚本，**严禁先 cd 进 skill 目录再执行**。 3. **凭证核实防幻觉**：必须校验脚本输出 JSON 中含有 `"status": "success"`。 4. **输出解耦 (Two-Tier Output)**：成功回执静默存储在 Archive 层，**严禁**将 JSON 回执完整吐给用户。 5. **注入优先 (Injection Awareness)**：回答前检查 ``，如有则优先引用。 6. **主动建议 (Proactive Insight)**：任何 Ingest/Query 任务结尾必须包含"主观判断"与"行动方案建议"。 7. **治理至上 (AODW Enforcement)**：确保所有 Agent 的动作都有 RT 记录。 --- ## 🔔 Ingest 交付清单（Orchestrator 必须执行） 4 连击完成后，Orchestrator **必须**按顺序执行以下 4 个动作： ### 1. ✅ Display Layer 回复（≤200字，结论先行，「·」列表） **规范**： - 字数限制：≤200 字 - 结构要求：结论先行，用「·」列出要点 - 风格要求：自然对话风格，避免技术细节 **示例**： ``` 已完成 {主题} 的知识编译。核心结论： · 提取了 {主要实体1}、{主要实体2} 的关键信息 · 建立了 {数量} 个知识节点链接 · 已存入 Wiki 供后续查询使用 ``` ### 2. ✅ 判断力输出（重要性 / 行动建议 / 可信度评估） **规范**： - **重要性**：明确标出该信息对用户的价值 - **行动建议**：下一步建议用户做什么 - **可信度**：根据来源评估信息的真实性 **示例**： ``` **我的判断**： - 重要性：高（核心技术与当前项目相关） - 建议行动：立即研究其架构设计，考虑集成到现有系统 - 可信度：高（来自官方技术文档） ``` ### 3. ✅ 通过 message tool 发送 source 文件附件给用户 **要求**： - 必须使用 message tool 发送文件附件 - 附件格式：source 文件（Markdown） - 发送渠道：根据用户使用的平台（Discord/Telegram/其他） **注意事项**： - 不要发送 JSON 回执或其他内部文件 - 只发送用户可读的 Markdown 格式文件 ### 4. ✅ 禁止将 JSON 回执完整吐给用户（输出解耦） **要求**： - 成功回执静默存储在 Archive 层 - 严禁将 JSON 回执完整吐给用户 - 只向用户展示 Display Layer 格式的中文摘要 **错误示例**： ``` ✅ 完成 Ingest 4 连击： 1. {"status": "success", "path": "arc-reactor-doc/wiki/sources/...", "size_bytes": 3394} 2. {"status": "success", ...} ``` --- ### 交付流程总结 ``` Ingest 4 连击完成 ↓ Orchestrator 验证结果（Post-Worker Validation） ↓ 执行交付清单： 1. Display Layer 回复（≤200字） 2. 判断力输出（重要性/建议/可信度） 3. 发送 source 文件附件 4. 确认无 JSON 回执泄露 ↓ 交付完成 ``` ### 检查清单快速参考 | 步骤 | 动作 | 状态 | 备注 | |------|------|------|------| | 1 | Display Layer 回复（≤200字） | ⬜ | 结论先行，「·」列表 | | 2 | 判断力输出 | ⬜ | 重要性/建议/可信度 | | 3 | 发送 source 文件附件 | ⬜ | 使用 message tool | | 4 | 确认无 JSON 回执泄露 | ⬜ | 输出解耦 | 每次 Ingest 完成后，Orchestrator 必须确认所有 4 个步骤都已完成。 --- ## 🛡️ 事后验证（Post-Worker Validation） **强制性要求**：Worker 完成任务后，Orchestrator 必须验证执行结果，防止 Worker 幻觉或伪造执行。 ### 验证流程 1. **检查 JSON 回执**：Worker 应输出包含 `"status": "success"` 的 JSON 2. **验证文件存在**：运行 `python3 skills/arc-reactor/scripts/archive-manager.py --validate` 3. **如果验证失败**：Orchestrator 必须手动重新归档文件 ### 示例验证流程 ```bash # Worker 完成后，Orchestrator 运行验证 python3 skills/arc-reactor/scripts/archive-manager.py --validate # 预期输出（成功）： # {"status": "ok", "action": "validate_wiki", "files_valid": 15, "files_invalid": 0, "files_empty": 0, "invalid_files": [], "message": "Validation complete: 15 valid, 0 invalid (0 empty)"} # 预期输出（失败）： # {"status": "partial", "action": "validate_wiki", "files_valid": 14, "files_invalid": 1, "files_empty": 1, "invalid_files": [...], "message": "Validation complete: 14 valid, 1 invalid (1 empty)"} ``` ### 验证失败处理 - 如果 `files_invalid > 0` 或 `files_empty > 0`，说明 Worker 撒谎或执行失败 - Orchestrator 必须重新执行失败的归档操作 - 记录验证失败情况到 RT 或 issue 跟踪 ### 双向验证机制这形成了"Worker 执行 → Orchestrator 验证"的双向验证闭环： - **Worker**：负责执行归档操作，输出 JSON 回执 - **Orchestrator**：负责验证执行结果，确保数据一致性 --- ## 🖥️ Display Layer（展示层）每次响应用户时必须遵守此层规范。详见 `references/output-style.md`。 ### 核心要点 - **长度**：≤200 字，结论先行 - **风格**：模拟群聊直观汇报，核心洞察用「·」列出 - **判断力 (Judgement)**：必须给出重要性 / 行动建议 / 可信度评估 - 用户说"详细"、"展开" → 提供 Archive 层内容 --- ## 🔄 Obsidian 同步层（可选后处理）详见 `references/dispatchers/obsidian.md`。 **触发**：Display Layer 输出完成后，异步执行 **前置**：`OBSIDIAN_VAULT_PATH` 已配置且 `AUTO_SYNC != false` --- ## 📱 Channel 自适应输出目标平台：Discord / Telegram（手机端） - 不用 Markdown 表格 - 不用超过3行的代码块 - 分段要短，关键信息放前面 - 列表用「·」或「1. 2. 3.」 --- ## 💬 自然触发词用户可以说： - "搜一下"、"帮我看"、"这个讲了什么" → 自动触发 Ingest + Display - 发送任意链接 → 自动触发 Ingest + Display - "详细说说"、"展开" → 触发 Archive 层 --- ## 🤝 多 Agent 协作规范 (AODW Governance) 详见 [CONTRIBUTING.md](./CONTRIBUTING.md)。 - **RT Core**：任何修改必须在 `RT/` 目录下有追踪记录 - **Commit 签名**：每个提交标注 Agent 名称，格式：`(by AgentName)` - **工具主权**：严禁直接操作 Wiki，必须调用 `archive-manager.py` --- ## 📦 Release Workflow 1. 更新 `_meta.json` 版本号 2. 执行 `bash scripts/release-skill.sh` 3. ZIP 包生成在 `dist/`，上传至 GitHub Releases --- *Powered by ARC Factory V4.0.5 | Karpathy Wiki Arch* FILE:CONTRIBUTING.md # CONTRIBUTING.md — ARC Reactor 协作规范所有 Agent（Codex / Claude Code / OpenClaw 子 Agent / 人类）必须遵守。 ## 多 Agent 同一账户注意事项所有 Agent 共用 `evan-zhang` 账户。以下规则确保不撞车： - **分支名必须加 Agent 前缀**（见下方命名规范），避免推到同一个分支 - **Commit message 末尾标注作者**：`(by Codex)` / `(by Claude)` / `(by Orchestrator)` - **先 fetch 再 push**：`git fetch origin` 确保本地最新 - **Git 并发保护**：如果两个 Agent push 同一分支，第二个会被拒绝（fast-forward 检查） ## 工作流 ``` ① gh issue list（先查有没有相关 Issue） ② gh issue create（没有就新建，描述清楚要改什么） ③ 在 Issue 上评论 "认领 by {agent-name}"（避免重复劳动） ④ git fetch origin && git checkout -b {agent-prefix}/{number}-{简述} origin/main ⑤ 写代码 + 测试 ⑥ git push origin {agent-prefix}/{number}-{简述} ⑦ gh pr create --base main（标题写 Closes #{number}） ⑧ 审查自己的 PR（见下方审查清单） ⑨ gh pr merge（合入） ⑩ 通知用户新功能已上线 ``` ## 铁律 1. **永远从最新 main 拉分支**：`git fetch origin && git rebase origin/main` 2. **禁止直接 push main**：Branch Protection 已开启，force push 会被拒绝 3. **禁止跳过 PR**：所有改动必须走 PR，哪怕是一行修复 4. **改之前先查 Issue**：`gh issue list` 看有没有人已经在做了 5. **PR 合入前必须审查**：见下方审查清单 ## PR 审查清单合入自己的 PR 之前，必须确认： - [ ] `git diff main...HEAD --stat` 看改了哪些文件 - [ ] 没有误删已有功能（特别是 `archive-manager.py`） - [ ] `python3 scripts/archive-manager.py --help` 功能完整 - [ ] 没有跟其他 open PR 冲突（`gh pr list`） - [ ] 如果有冲突：先 rebase 最新 main 再合 ## 冲突处理如果 PR 跟其他 PR 冲突： 1. `git fetch origin && git rebase origin/main` 2. 解决冲突 3. `git push origin {branch} --force-with-lease`（在自己的分支上 force push 是安全的） 4. 重新检查后合入 ## Issue 命名规范 - `feat: xxx` — 新功能 - `fix: xxx` — 修 bug - `refactor: xxx` — 重构 - `docs: xxx` — 文档 - `chore: xxx` — 杂项 ## 分支命名规范必须加 Agent 前缀，避免撞车： - `codex/{number}-{简短描述}` — Codex Agent 的分支 - `claude/{number}-{简短描述}` — Claude Code Agent 的分支 - `orchestrator/{number}-{简短描述}` — 编排者的分支 - `manual/{number}-{简短描述}` — 人工操作例：`codex/22-multi-kb`、`claude/27-fact-index` ## 谁来 Review？当前阶段：Agent 自审 + 功能验证。未来：可以配置 CODEOWNERS 要求交叉审查。 FILE:README.md # ARC Reactor Skill - AI Agent 知识编译引擎 > **⚠️ Agent Entry Point** > 此目录是 ARC Reactor 的分发核心。如果你是正在读取此文件的 AI Agent，请以 `SKILL.md` 作为你的行为准则。 ## 🎯 功能概览 (v4.2.0) 本项目是基于 Karpathy **Compilation over Retrieval** 理念构建的知识库。 1. **Ingest**: 深度摄入，执行“四连击”归档（Source, Entity, Index, Log）。 2. **Context Injection**: 自动从 Wiki 中检索实体内容注入上下文，实现无感知识增强。 3. **Weekly Reporter**: 聚合过去一周的摄入内容，生成高维度洞察周报。 4. **Wiki Lint**: 自动化维护词条链接的完整性。 ## 📦 目录结构 ```text skills/arc-reactor/ ├── SKILL.md # Agent 核心指令集 ├── arc-reactor-config.yaml # 多模型路由与功能配置 ├── _meta.json # 元数据 ├── scripts/ │ ├── archive-manager.py # 存储与归档逻辑 │ ├── context-injector.py # 潜意识注入探针 │ └── weekly-reporter.py # 周报聚合引擎 └── references/ # 模板与审计规范 ``` ## 🚀 核心指令 ### 1. 知识归档 (Ingest) ```bash cat << 'EOF' | python3 scripts/archive-manager.py --type source --topic "标题" --stdin [Markdown Content] EOF ``` ### 2. 潜意识探测 (Inject) ```bash python3 scripts/context-injector.py --query "[用户的问题]" ``` ### 3. 周汇总 (Weekly) ```bash python3 scripts/weekly-reporter.py --days 7 ``` ## ⚙️ 配置说明编辑 `arc-reactor-config.yaml` 可以调整： - **models**: 不同任务分派给哪些模型（GPT-4o vs Flash）。 - **injection**: 开启/关闭自动注入，调整最大实体提取数。 - **weekly_brief**: 调整周报扫描天数。 --- *Powered by ARC Reactor v4.2.0* FILE:REVIEW-obsidian-sync.md # 复审报告：Obsidian 同步实现 > **审查日期**: 2026-04-09 > **审查者**: 代码审查子代理 > **目标**: ARC Reactor Obsidian Sync 功能复审 > **参考规范**: `/tmp/arc-reactor/SPEC-obsidian-sync.md` --- ## 复审结果 ### 通过项 - **`--sync-obsidian` 命令正确实现**：参数齐全（`--source`, `--vault`, `--target`, `--async`），与 SPEC 7.2 一致 - **`validate_obsidian_config()` 校验逻辑与 SPEC 4.3 高度吻合**：目录存在性检查 + 写入权限测试，返回 `(bool, message)` 元组 - **`sync_to_obsidian()` 重试逻辑（指数退避）实现正确**：`sleep_time = retry_delay * (2 ** attempt)`，符合 SPEC "重试策略：使用指数退避" - **SKILL.md 插入位置正确**：Obsidian 同步小节在 Display Layer 之后、Channel 自适应输出之前 - **SKILL.md 配置变量表完整**：包含 `OBSIDIAN_VAULT_PATH`、`OBSIDIAN_TARGET_DIR`、`AUTO_SYNC` 三个变量 - **SKILL.md 铁律正确**： - ✓ 同步失败不阻塞 Display Layer 输出 - ✓ 同步失败不重写 Display Layer 内容 - ✓ `AUTO_SYNC=false` 时完全不执行任何 Obsidian 相关代码 - **Display Layer 状态模板符合 SPEC**：提供了成功和失败两种场景的模板 - **SKILL.md 字数规范合理**：≤200字原则保持不变 --- ### 问题项 #### 🔴 严重问题 1. **AUTO_SYNC 环境变量未实现** → `sync_to_obsidian()` 函数完全没有检查 `AUTO_SYNC` 环境变量 SPEC 3.1 明确要求："`AUTO_SYNC` != `false`" 时才执行。但 `main()` 中 Obsidian sync 分支仅检查 `--source` 和 `vault_path`，未验证 `AUTO_SYNC`。 **影响**: `AUTO_SYNC=false` 无法完全关闭同步（与铁律矛盾） **修复建议**: 在 `main()` 的 Obsidian sync 分支开头添加： ```python auto_sync = os.environ.get('AUTO_SYNC', 'true') if auto_sync.lower() == 'false': print(json.dumps({"status": "skipped", "message": "AUTO_SYNC=false, Obsidian sync disabled"})) sys.exit(0) ``` 2. **`os.fork()` 仅支持 Unix/macOS，不支持 Windows** → `async_mode` 在 Windows 上会崩溃 `main()` 第 347-360 行使用 `os.fork()` 实现后台执行，但 Windows 没有 `fork()` 系统调用。 **修复建议**: 使用跨平台方案： ```python if getattr(args, 'async_mode', False): import threading thread = threading.Thread( target=lambda: sync_to_obsidian(args.source, vault_path, args.target), daemon=True ) thread.start() print(json.dumps({"status": "pending", ...})) sys.exit(0) ``` #### 🟡 中等问题 3. **重试间隔与 SPEC 不符（5分钟 vs 5秒）** → SPEC 6.0 表格注明 "间隔5分钟"，实现使用 `retry_delay=5`（秒） **影响**: 网络问题导致 iCloud 同步失败时，用户需要等待远超预期的总时长（5+10+20=35秒 vs 5+10+20=35分钟） **修复建议**: 将 `sync_to_obsidian()` 的 `retry_delay` 默认值从 `5` 改为 `300`（5分钟×60秒） 4. **路径遍历未防护** → `target_dir` 或 `source_path` 含 `../` 时可逃逸 vault 目录 SPEC 5.0 没有明确提到路径注入风险，但这是安全最佳实践。 **修复建议**: 在 `sync_to_obsidian()` 中加入： ```python dest_dir = os.path.normpath(os.path.join(os.path.expanduser(vault_path), resolved_target)) real_vault = os.path.normpath(os.path.expanduser(vault_path)) if not dest_dir.startswith(real_vault): return {"status": "error", "error": "路径遍历被拒绝"} ``` 5. **同步标记每次重试都追加** → 3次重试全部失败时，dest 文件会有 3 个 `同步状态` 标记因为 `shutil.copy2()` 后 `open(dest_path, 'a')` 每次追加都会执行。 **修复建议**: 在 `sync_to_obsidian()` 的成功返回路径中，仅在首次成功时追加标记（当前逻辑已在成功路径追加，OK；但失败路径应该用 `else` 确保只在成功时追加） #### 🔵 轻微问题 6. **SKILL.md 中 `{date}` 占位符说明不明确** → 仅在配置变量表 "说明" 列出现过，使用示例中未展示 **建议**: 在 "使用示例" 的 `--target` 参数中加入实际替换示例： ```bash --target "github分享/AI调研/2026-04-09/" # {date} 将被替换为 2026-04-09 ``` --- ### SPEC 符合度 | 检查项 | 结果 | 说明 | |--------|------|------| | Display Layer 不感知同步失败 | ✓ | `validate_obsidian_config()` 失败仅返回 error dict，不影响主流程 | | 同步不是第5步 | ✓ | SKILL.md 明确标注为 "附加动作，异步，不阻塞" | | `AUTO_SYNC=false` 完全关闭 | ✗ | **实现缺失** — 代码中完全没有 `AUTO_SYNC` 检查 | | 异步非阻塞 | ✓（有平台限制） | `os.fork()` 实现仅支持 Unix/macOS | | Display Layer ≤200字 | ✓ | SKILL.md 规范正确 | | 指数退避重试 | ✓ | `retry_delay * (2 ** attempt)` | | 重试间隔5分钟 | ✗ | 实现为5秒 | | 路径遍历防护 | ✗ | 未实现 | | `validate_obsidian_config()` 符合 SPEC | ✓ | 逻辑与 SPEC 4.3 一致 | | SKILL.md 插入位置正确 | ✓ | 在 Display Layer 之后 | | 铁律正确 | ✓（除 `AUTO_SYNC` 外） | 三条铁律均正确 | --- ### 总体评价 **实现质量：良好（75/100）** 核心逻辑（`validate_obsidian_config`、`sync_to_obsidian` 重试机制、Display Layer 独立性）实现正确，SKILL.md 文档规范完善。 **主要缺陷**： 1. `AUTO_SYNC=false` 完全未实现 — 这是 SPEC 明确要求的功能 2. `os.fork()` 造成平台锁定 — Windows 用户无法使用 async 模式 3. 重试间隔单位错误（秒 vs 分钟）— 可能导致用户等待体验与预期不符 **建议优先级**： 1. **P0**: 补充 `AUTO_SYNC` 环境变量检查 2. **P0**: 替换 `os.fork()` 为 `threading.Thread`（跨平台兼容） 3. **P1**: 将 `retry_delay` 默认值改为 300（5分钟） 4. **P2**: 添加路径遍历防护 5. **P3**: SKILL.md 补充 `{date}` 占位符使用示例 --- *本报告为只读审查，不涉及代码修改。审查结果已保存至 `/tmp/arc-reactor/REVIEW-obsidian-sync.md`* FILE:SPEC-obsidian-sync.md # SPEC: Obsidian Sync 集成规范 > **版本**: v1.0 > **状态**: 设计规范 > **目标**: 将 Obsidian 同步功能无缝嵌入 ARC Reactor SKILL.md 主流程 --- ## 1. 背景与设计目标 ### 现状 - ARC Reactor 的 **4连击** (source → entity → index → log) 已完善，输出落地在本地 `arc-reactor-doc/` Wiki - `references/dispatchers/obsidian.md` 定义了 Obsidian 同步调度逻辑，但**未集成**到 SKILL.md 主流程 - 用户需要在报告生成后手动同步到 Obsidian，流程割裂 ### 设计目标 1. **透明集成**：Obsidian 同步作为 Display Layer 之后的附加动作，对主流程无侵入 2. **可选控制**：通过 `AUTO_SYNC=false` 完全关闭，不影响任何主流程 3. **异步非阻塞**：同步失败不阻塞用户，不影响 Display Layer 输出 4. **Display Layer 纯净**：≤200字原则不受影响，同步状态用超链接替代文字堆砌 --- ## 2. 集成点设计 ### 2.1 在 SKILL.md 中的插入位置在 SKILL.md 的 **Display Layer** 小节之后、**Channel 自适应输出** 之前，插入新小节： ``` ## 🔄 Obsidian 同步层（可选后处理）触发时机：Display Layer 输出完成后，作为独立附加动作异步执行前置条件：OBSIDIAN_VAULT_PATH 已配置且有效 ``` ### 2.2 完整流程顺序 ``` 用户输入 ↓ [4连击 Ingest] (source → entity → index → log) ↓ [Display Layer] ≤200字，用户可见 ↓ [Obsidian Sync] ← 附加动作，异步，不阻塞 ↓ 返回结果给 Orchestrator ``` **关键决策**：Obsidian 同步**不是第5步**，而是 Display Layer 之后的独立后处理动作。理由： - 4连击是知识库写入的原子操作，职责单一 - Obsidian 同步是**消费**已生成的报告，不是生产知识 - Display Layer 必须在同步之前完成（否则会因同步失败而延迟用户可见输出） --- ## 3. 同步触发时机 ### 3.1 触发条件（同时满足） | 条件 | 说明 | |------|------| | `OBSIDIAN_VAULT_PATH` 已配置 | 未配置时跳过，不询问 | | `AUTO_SYNC` != `false` | 默认为 `true`，用户可关闭 | | 本次 Ingest 产生了 source 文件 | 无 source 时不同步（避免空操作） | ### 3.2 触发时机点在 Display Layer 输出完成**之后**，作为独立步骤执行： ```python # 伪代码 def after_display_layer(report_path): if not AUTO_SYNC or AUTO_SYNC == "false": return # 完全跳过 if not OBSIDIAN_VAULT_PATH: return # 未配置，跳过 # 异步后台执行，不阻塞主流程 async_task = spawn(sync_to_obsidian, report_path) ``` ### 3.3 同步文件选择同步时按以下优先级选择要复制的文件： 1. **source 文件**（本次 Ingest 的原始报告）：`arc-reactor-doc/wiki/sources/{date}/{topic}.md` 2. **不存在时**：使用 index.md 中本次相关的摘要行 --- ## 4. 配置方式 ### 4.1 环境变量（推荐） ```bash # .env 或 openclaw.json OBSIDIAN_VAULT_PATH=~/Library/Mobile Documents/iCloud~md~obsidian/Documents/ OBSIDIAN_TARGET_DIR=github分享/AI调研/{date}/ AUTO_SYNC=true ``` ### 4.2 SKILL.md 中的引导流程（新增）在 SKILL.md 的**铁律**之后、**Display Layer** 之前，插入 Obsidian 配置引导： ``` ### Obsidian 同步配置（首次使用引导）若 OBSIDIAN_VAULT_PATH 未设置，Agent 应主动引导： 1. 询问用户 Obsidian 根目录路径 2. 将路径写入 .env（OBSIDIAN_VAULT_PATH） 3. 执行自检脚本（见 obsidian.md 第3节） 4. 确认后告知用户已开启自动同步 ``` ### 4.3 配置校验逻辑 ```python def validate_obsidian_config(): vault = os.path.expanduser(OBSIDIAN_VAULT_PATH) if not os.path.isdir(vault): return False, "Obsidian 库路径不存在" test_path = os.path.join(vault, '.arc-sync-test') try: with open(test_path, 'w') as f: f.write('ping') os.remove(test_path) return True, "OK" except: return False, "无写入权限" ``` --- ## 5. Display Layer 输出规范 ### 5.1 原则 - Display Layer **永远独立**于 Obsidian 同步状态 - 同步成功/失败**不在 Display Layer 文字中体现** - 用户通过超链接或附加标记感知 Obsidian 状态 ### 5.2 Display Layer 输出模板 **场景A：同步成功（Display Layer 不变）** ``` 报告已完成 ✅ · [[Claude-Code]]: Anthropic 系统级 AI 编程助手 · [[SWE-bench]]: 软件工程评测基准 · [[Devin]]: Cognition AI 的自主编程智能体 📁 已在 Obsidian 同步 → [打开 Obsidian](obsidian://open?vault=Documents) ``` **场景B：同步失败（Display Layer 仍显示成功）** ``` 报告已完成 ✅ · [[Claude-Code]]: Anthropic 系统级 AI 编程助手 · [[SWE-bench]]: 软件工程评测基准 ⚠️ Obsidian 同步暂不可用（后台重试中） 📁 报告已保存在本地 arc-reactor-doc/wiki/sources/ ``` ### 5.3 字数控制 Display Layer 核心文字（不含链接）≤180字，剩余空间留给同步状态标记。 | 场景 | 核心字数 | 状态标记字数 | 总计 | |------|----------|--------------|------| | 同步成功 | ~150字 | ~25字 | ≤200字 | | 同步失败 | ~150字 | ~40字 | ≤200字 | | 未配置同步 | ~190字 | 0 | ≤200字 | ### 5.4 同步状态附录标记在源报告文件末尾追加（不影响 Display Layer）： ```markdown --- 同步状态: ✅ Obsidian (时间: 2026-04-09 14:30) --- ``` 或失败时： ```markdown --- 同步状态: ⚠️ Obsidian 失败 (原因: 路径无写入权限) 重试次数: 1/3 --- ``` --- ## 6. 错误处理策略 | 错误类型 | 处理方式 | |----------|----------| | `OBSIDIAN_VAULT_PATH` 未配置 | 跳过，不询问（按需引导） | | 库路径不存在 | Display Layer 显示警告，超链接变为"配置 Obsidian" | | 写入权限不足 | 重试1次，失败后记录到 log.md，跳过 | | 网络问题（iCloud不同步） | 后台重试3次（间隔5分钟），Display Layer 无感知 | | 目标目录创建失败 | 记录错误，不阻塞，Display Layer 无感知 | **重试策略**：使用指数退避，最多3次，异步执行。 --- ## 7. 实现方案 ### 7.1 新增 SKILL.md 片段在 SKILL.md Display Layer 小节**之后**插入： ```markdown ## 🔄 Obsidian 同步层（可选后处理） **触发时机**：Display Layer 输出完成后，异步执行 **前置条件**：`OBSIDIAN_VAULT_PATH` 已配置且 `AUTO_SYNC != false` ### 配置变量 | 变量 | 默认值 | 说明 | |------|--------|------| | `OBSIDIAN_VAULT_PATH` | `~/Library/Mobile Documents/iCloud~md~obsidian/Documents/` | Obsidian 仓库根路径 | | `OBSIDIAN_TARGET_DIR` | `github分享/AI调研/{date}/` | 目标子目录，`{date}` 自动替换 | | `AUTO_SYNC` | `true` | 是否自动同步 | ### 执行流程 1. **校验配置**：`validate_obsidian_config()` 返回失败则跳过 2. **复制报告**：将本次 source 文件复制到 `{vault}/{target_dir}/{topic}.md` 3. **追加状态**：在源文件末尾追加 `同步状态: ✅ Obsidian (时间: ...)` 4. **Display Layer**：Display Layer 永远先于同步完成，用户无感知等待 ### 铁律 - 同步失败**不阻塞** Display Layer 输出 - 同步失败**不重写** Display Layer 内容 - `AUTO_SYNC=false` 时完全不执行任何 Obsidian 相关代码 ``` ### 7.2 archive-manager.py 扩展新增 `--sync-obsidian` 命令： ```bash python3 scripts/archive-manager.py \ --sync-obsidian \ --source arc-reactor-doc/wiki/sources/2026-04-09/claude-code.md \ --vault "$OBSIDIAN_VAULT_PATH" \ --target "github分享/AI调研/{date}/" \ --async ``` 返回 JSON 回执： ```json { "status": "success", "action": "obsidian_sync", "source": "...", "destination": "...", "obsidian_vault": "~/Library/.../Documents/", "sync_time": "2026-04-09 14:30:05", "message": "Obsidian sync complete." } ``` ### 7.3 Orchestrator 任务注入更新 Spawn 任务时，在 Display Layer 完成后追加： ```bash # Obsidian 同步（如果已配置且 AUTO_SYNC != false） python3 scripts/archive-manager.py \ --sync-obsidian \ --source "arc-reactor-doc/wiki/sources/{date}/{topic}.md" \ --vault "$OBSIDIAN_VAULT_PATH" \ --target "$OBSIDIAN_TARGET_DIR" \ --async ``` --- ## 8. 测试用例 ### TC-1：完整同步流程 ``` 前置条件：AUTO_SYNC=true, OBSIDIAN_VAULT_PATH 有效步骤： 1. 执行 Ingest（4连击） 2. 观察 Display Layer 输出 3. 检查 Obsidian 目标目录是否有文件预期：Display Layer 先显示，Obsidian 同步异步完成 ``` ### TC-2：同步失败不影响主流程 ``` 前置条件：AUTO_SYNC=true, OBSIDIAN_VAULT_PATH 指向不存在路径步骤： 1. 执行 Ingest（4连击） 2. 观察 Display Layer 输出预期：Display Layer 正常显示成功，Obsidian 失败有日志但不展示给用户 ``` ### TC-3：AUTO_SYNC=false 完全关闭 ``` 前置条件：AUTO_SYNC=false 步骤： 1. 执行 Ingest（4连击） 2. 检查是否有任何 Obsidian 相关日志或文件操作预期：无任何 Obsidian 操作，Display Layer 正常 ``` ### TC-4：Display Layer 字数验证 ``` 前置条件：标准 Ingest 输出步骤： 1. 捕获 Display Layer 输出 2. 统计字符数（含状态标记）预期：≤200字 ``` ### TC-5：OBSIDIAN_VAULT_PATH 未配置时按需引导 ``` 前置条件：OBSIDIAN_VAULT_PATH 为空，用户首次提及 Obsidian 步骤： 1. 用户说"帮我搜 xxx，结果存到 Obsidian" 预期：Agent 主动询问 Obsidian 路径，完成配置引导 ``` --- ## 9. 文件变更清单 | 文件 | 变更类型 | 说明 | |------|----------|------| | `SKILL.md` | 修改 | 在 Display Layer 后插入 Obsidian 同步小节 | | `scripts/archive-manager.py` | 修改 | 新增 `--sync-obsidian` 命令 | | `references/dispatchers/obsidian.md` | 保持 | 现有调度器保持不变，作为参考 | --- ## 10. 关键设计决策记录 | 决策 | 理由 | |------|------| | 同步不是第5步 | 4连击是知识生产原子操作，Obsidian 是消费动作，时机不同 | | Display Layer 不感知同步状态 | 保持≤200字原则，同步是后处理细节 | | 失败不阻塞 | Obsidian 是锦上添花，不能因为 iCloud 问题影响核心体验 | | 异步执行 | 用户不需要等待 iCloud 同步完成才能看到结果 | | 配置通过环境变量 | 与现有 .env 体系一致，无需新配置接口 | --- *本规范为设计文档，实现请同步更新 SKILL.md 和 archive-manager.py* FILE:_meta.json { "name": "arc-reactor", "version": "4.3.0", "description": "ARC (Acquire/Research/Catalogue) - 个人知识建设智能调研引擎", "author": "evan", "tags": ["research", "knowledge-management", "automation"], "created": "2026-04-08", "rt": "RT-002" } FILE:arc-reactor-config.yaml # ARC Reactor Routing Configuration (V4.0) # 此文件用于映射不同编译任务所属的 LLM Model，以兼顾智商与成本。 # Orchestrator (指挥官) 应在 Spawn Worker 时参考此表。 version: "4.0" models: # 顶级推理模型 (不计成本，最高智商) # 用于：解析生肉、生成 Source、提取 Entity、处理高度矛盾的重度任务。 ingest_heavy: "openai-gpt-4o" # 次级高速模型 (较便宜，速度极快) # 用于：日常 Query 读取 Wiki 合成报告，或者生成单维度的分析。 query_medium: "claude-3-5-sonnet-20240620" # 轻量级流水线模型 (最便宜，最高速) # 用于：Lint 整理，跑批扫描孤立的 index.md 和断链，处理日常语法修正。 lint_light: "gemini-1.5-flash" workflows: ingest: required_combo: [source, entity, index, log] preferred_model: "ingest_heavy" query: required_combo: [output] preferred_model: "query_medium" lint: required_combo: [index, log, metadata_fix] preferred_model: "lint_light" knowledge_bases: - name: personal-learning root: arc-reactor-doc description: "External research, papers, open-source analysis" auto_route: sources: ["web_fetch", "web_search", "youtube", "manual"] tags: ["research", "paper", "open-source", "tutorial", "agent", "ai"] - name: work-collaboration root: cwork-kb description: "CWork inbox/outbox, contracts, HR decisions, project progress" auto_route: sources: ["cwork-api", "cwork-query-report", "cwork-send-report"] tags: ["合同", "汇报", "人事", "审批", "待办", "任务"] # 潜意识注入 (RT-007) 配置 injection: enabled: true max_entities: 3 # 单次命中最大调取的实体数量 max_chars: 10000 # 注入上下文的最大字符长度限制 match_index: "index.md" # 周报系统 (RT-008) 配置 weekly_brief: default_days: 7 summarizer_model: "claude-3-haiku" # 高性价比摘要模型 FILE:references/dedup-rules.md # 去重检测规则 (Dedup Rules) ## 检测时机在 ARC 进入 Acquire 阶段**之前**，必须先执行去重检测。 ## 三级匹配 ### L1: URL 精确匹配 - 扫描 `reports/` 和 `knowledge/entities/` 中所有已记录的 URL - 完全相同 → 直接跳过，汇报"此链接已于 X 日调研过" ### L2: 主体匹配 - 从输入中提取核心实体名（项目名/人物名/公司名） - 与 `knowledge/entities/` 目录下的文件名做模糊匹配 - 命中 → 静默进入合并模式 ### L3: 语义相似（自主判断） - 对输入的标题/摘要与已有报告做语义相似度判断 - 核心主体 ≥70% 重叠 → 自动合并 - 仅部分关联 → 独立建档，但在两份报告中互相标注 `关联调研` - 一个输入涉及多个主体 → 一分为二，各自归入对应实体 - 用户明确要求独立 → 尊重用户指令覆盖自动判断 ## 合并模式 (Merge Mode) ### 增量信息提取 - 仅提取新源中旧报告**不包含**的信息 - 标注每条新增信息的来源：`[来源: 平台/作者]` ### 报告更新 - **基本信息表**：追加新的来源行 - **正文章节**：新发现插入对应章节末尾，或新增独立章节 - **附录**：追加新的原始素材路径 ### 源追踪表每份报告底部维护多源追踪表： ```markdown ## 源追踪 | # | 来源平台 | URL | 内容类型 | 采集日期 | 贡献内容 | |---|----------|-----|----------|----------|----------| | 1 | 今日头条 | ... | 文章 | 2026-04-06 | 初始调研 | | 2 | 抖音 | ... | 视频 | 2026-04-07 | 补充架构细节 | ``` ## 冲突处理 - **数据类**（Star 数、版本号）：用新值覆盖旧值，标注更新时间 - **观点类**（评价、结论）：两者并存，标注各自来源 FILE:references/dispatchers/obsidian.md # Obsidian 同步调度器 (Obsidian Dispatcher) 在使用 ARC 之前，若你想将报告自动同步到本地的 Obsidian 知识库中，必须完成此配置与检测。 ## 1. 核心配置清单在 Agent 初始化或接收到首次"设置同步"指令时，引导用户确认以下变量（若未设置，请向用户询问）： | 配置项 | 默认值 | 说明 | |--------|--------|------| | `OBSIDIAN_VAULT_PATH` | `~/Library/Mobile Documents/iCloud~md~obsidian/Documents/` | (以 iCloud 同步为例) 你的 Obsidian 仓库本地绝对路径 | | `OBSIDIAN_TARGET_DIR` | `github分享/AI调研/{date}/` | 你希望报告在 Obsidian 库内存放的具体文件夹 | | `AUTO_SYNC` | `true` | 是否在生成报告后立即触发同步 | > **提示（Agent 必读）**：如果用户是在 Windows 下，默认路径通常为 `C:\Users\用户名\Documents\Obsidian\`；如果是 Mac iCloud 则多如默认值。引导他们找到自己的真准路径！ ## 2. Agent 引导配置流程当用户需要开启此功能时，**你作为 Agent 必须主动按以下步骤引导**： 1. **询问路径**："嗨，为了让调研报告直接发到你的 Obsidian，请告诉我你的 Obsidian 根目录物理路径在哪儿？" 2. **生成环境变量**：将用户提供的路径写入工作区的 `.env` 文件（或 `openclaw.json` 配置内），注册 `OBSIDIAN_VAULT_PATH`。 3. **执行自检脚本**（见第 3 节），并在聊天中汇报检测结果。 ## 3. 自检与生效验证机制 (Validation) 配置完成后，Agent 需要立即执行环境自检，确保配置可用。你可以直接在系统中执行以下 `bash` 命令来进行验证： ```bash # Agent 自检执行命令 VAULT="${OBSIDIAN_VAULT_PATH}" TARGET="${OBSIDIAN_TARGET_DIR/\{date\}/$(date +%Y-%m-%d)}" FULL_PATH="${VAULT}/${TARGET}" # 1. 检查根目录是否存在 if [ ! -d "$VAULT" ]; then echo "❌ 失败: 找不到 Obsidian 库根路径 ($VAULT)！请检查路径或权限设置。" exit 1 fi # 2. 测试创建目标结构与探测写入权限 mkdir -p "$FULL_PATH" if touch "$FULL_PATH/.arc-test-ping"; then echo "✅ 成功: Obsidian 同步链路检测通过。目标目录 ($FULL_PATH) 具有读写权限。" rm "$FULL_PATH/.arc-test-ping" else echo "❌ 失败: 没有权限在 ($FULL_PATH) 中写入文件，请检查目录权限！" exit 1 fi ``` **反馈逻辑**： - 如果输出 `✅ 成功`，请用愉快的语气告诉用户："搞定啦！Obsidian 路径连接畅通无阻，马上为你自动保存以后的报告。" - 如果输出 `❌ 失败`，必须把错误信息告知用户，并协助纠正路径格式（如排查空格逃逸、Mac 的 iCloud 重定向问题等）。 ## 4. 运行时同步规则 - 报告正式完成后，提取执行时刻的 `YYYY-MM-DD` 替换 `{date}`。 - 将生成的报告文件完整复制过去。 - 同步后，在源工作区本地报告的附录处追加标记：`同步状态: ✅ Obsidian (时间: YYYY-MM-DD HH:MM)`。 FILE:references/env-setup.md # ARC Reactor 环境挂载与凭证鉴权指南 (Environment Setup) 作为新一代跨域全模态采集网闸，ARC 已经不再只依赖大语言模型自身，而是需要外引一系列网络侦察爬虫节点与跨域 Search 鉴权。 > **何时触发**：当这是系统第一次挂载 ARC Reactor，或者当 Orchestrator 要派遣采集器却发现缺少依赖时，**必须第一时间要求用户完成下述参数设置，并在 `.env` 里注册**。如果用户不知道如何获取，请指导他们！ ## 1. 广域网络搜索引擎配置 (Web Search API) > 用于实现 ARC-Worker 突破单点网页盲区，去搜查周边竞品及对性能评定做异源真伪核查用的雷达探测。 | 配置项环境变量 | 类型 | 适用方 | 说明 | |------------|-------|------|------| | `SEARCH_PROVIDER` | `string` | 必填 | 首选抓包探针。比如 `tavily`, `google_news`, 或基于 OpenAI Search 的 API | | `SEARCH_API_KEY` | `string` | 必填 | 对应的鉴权私钥。提示：引导用户从相应的开发者平台获取。 | ## 2. 第三方 Clawhub 反风控挂件绑定 (Scraping Skills) > 如果遇到了被厚重心墙风控的目标群体（推特、油管字幕、抖音视频等），纯文本嗅探脚本会碰壁。我们必须通过系统指令调用专属的反风控挂机插件。 **配置指引：引导用户打开 Clawhub 或 OpenClaw 的 Market 去分别安装对应媒体的专业捕捞网，确保主 Agent 能访问它们的 tools：** | 所需挂件 (示例) | 探测媒体类型范围 | ARC 调用时机 | |-----------------|----------------|-------------| | `youtube-dl-agent` / `whisper-sub` | 视频流 | 当目标 URL 是 Bilibili/YouTube 时，调用该挂件扒原视频及字幕转录。 | | `xhs-claw` / `douyin-spider` | 封闭流媒体社交 | 当目标属于强反爬私域（小红书/抖音等）时进行免密滑块穿透。 | | `twitter-mcp` / `reddit-scraper` | 文本强社交 | 当目标是 X 平台，获取其底下的回帖与社交反响。 | ## 3. 自检协议跟 Obsidian 的验证一样，在装配完毕后，Agent 应该用测试指令试探一下环境引擎是否就绪： - 搜一下 "今天的天气" 验证搜索引擎 Key。 - 启动指定的 Clawhub 挂件查查它的 `tools` 表判断是否可用。如果失败，向用户报错。 FILE:references/knowledge-rules.md # 知识编译规则 (Knowledge Compilation Rules) > 借鉴 Karpathy LLM Wiki 理念：报告是"快照"，知识库是"活体"。 ## 三层架构 ``` Layer 1: Raw（原始素材层） reports/YYYY-MM-DD/raw/ 转录稿、截图、PDF — 只读存储，先全收 Layer 2: Wiki（知识编译层） knowledge/ ├── entities/ 实体页（项目/人物/公司） ├── concepts/ 概念页（术语/方法论） ├── comparisons/ 对比页（A vs B） └── index.md 知识导航目录 Layer 3: Schema（规则层）本文件 — 定义编译行为和交叉引用规则 ``` ## 编译触发每次 ARC 完成一份报告后，自动执行知识编译： ### 1. 实体提取 - 从报告中识别关键实体（项目名、人物、公司） - 已存在 `knowledge/entities/[实体].md` → **追加更新** - 不存在 → **新建实体页** ### 2. 概念关联 - 识别报告中的核心概念 - 更新 `knowledge/concepts/` ### 3. 交叉引用 - 在相关实体页之间建立 `[[双向链接]]`（Obsidian 格式） - 确保知识网络互相连通 ### 4. 索引更新 - 更新 `knowledge/index.md` 目录 - 记录新增/更新的实体和概念 ## 实体页模板 ```markdown # [实体名] ## 基本信息 | 字段 | 内容 | |------|------| | **类型** | 项目/人物/公司 | | **首次调研** | YYYY-MM-DD | | **最后更新** | YYYY-MM-DD | | **信息来源数** | N | ## 核心描述 [一段话总结] ## 关键数据 [表格化的核心指标] ## 关联实体 - [[相关实体1]] - [[相关实体2]] ## 来源报告 - [报告1](../reports/YYYY-MM-DD/xxx-调研报告.md) - [报告2](../reports/YYYY-MM-DD/xxx-调研报告.md) ``` ## 矛盾检测 (Self-Challenge) 当新调研内容与已有知识页存在矛盾时： - 自动生成 `knowledge/conflicts/YYYY-MM-DD-[主题].md` - 列出旧观点、新矛盾点、建议修正方案 - 数据类矛盾：直接用新值覆盖 - 观点类矛盾：两者并存，标注来源 FILE:references/orchestrator-dispatch.md # 编排与派发行动准则 (Orchestrator Dispatch Protocol) > **目标对象**：作为系统最前台收口主力的 **Orchestrator 主 Agent**。 > **禁止越权**：你本身不要去读取万字长文、不要做抽取分析，你的任务就是把活儿漂亮地分发出去。 ## 1. 原则: Announce-then-act 当用户在会话中直接丢给你（或附带了多个）URL 分析，或者长篇文本分析的情报时。你唯一需要在当前对话池内做的事情如下。 ### 严禁： - 先发一条“好的，我明白了，马上干”。 - 再空转思考如何去抓链接。 ### 第一反应步骤：你的下一条也是唯一的回复必须满足 **Announce-then-act**，而且要带上实际的工具调用！ - 回复内容必须类似：“✅ 收到。您的链接已被接管。我正在利用并发机制开启 ARC 子矿工进程进行全网潜底抓取... 您的聊天主线不受影响。” - **同时在产生这一条文本回复的一瞬间 (同一个调用内)**，你必须使用系统的 `spawn` (若支持) 或原生 sub-agent 创造工具进行任务派送。 ## 2. 派发参数主 Agent 生成衍生任务时，必须要向后台子代理传递清晰的初始指令： - **Context/Task Prompt**: "你是 ARC Reactor 的后台子执行代理（ARC-Worker）。请从目标 URL：`[原链接]` 进行抓取，并完整执行针对该文章的 [获取->二次审查->实体合并卡片创建等知识编译] 全流程记录。请参考配置 `arc-reactor/SKILL.md` 的通道 2 指南干活！一切结果必须物理生成文件，不能在后台自言自语。" ## 3. 并发派发 (Multi-URL Scenario) 如果你一次性看到用户丢给你这三个文章的 URL 想快速了解： - `https://A...` - `https://B...` - `https://C...` **不要排队阻塞，并发全开！** - 同步且并发地 `spawn()` 出三个独立的 ARC-Worker 分配目标 A、B 和 C 去分析。 - 确保并嘱咐它们对于同源主题要基于**追加写入与 Merge 模型**来避免锁死（`references/dedup-rules.md`）。 ## 4. Yield-after-spawn (立刻退行与守护) 所有的任务交接丢给 Worker 以后，作为主 Agent： 1. 立刻调用您的系统级打断机制 `yield` 或关闭你的当前 Turn 焦点窗口。 2. 将关注力归还给用户本身。用户可以在此时问你今天几度、或者跟你闲聊昨晚的比赛。 3. 一直等到 Worker 在系统总线上跑完爬虫流程，通过“物理生成文件附件并弹窗跨域投送大群”时，任务完成闭环。你不用在主线轮询死等这个状态。 FILE:references/output-style.md # Output Style Guide — Display Layer vs Archive Layer ## 概述 ARC Reactor 输出分两层：**Display Layer**（展示层）和 **Archive Layer**（归档层）。Display Layer 是用户直接看到的回复；Archive Layer 是存储到知识库的结构化内容。 --- ## 🖥️ Display Layer（展示层） **用途**：直接回复用户，是唯一的"对外"输出层。 ### 规范 | 属性 | 要求 | |------|------| | **字数** | ≤200 字 | | **风格** | 模拟人类对话，像在聊天 | | **结构** | 结论先行，用「·」列出要点 | | **语气** | 自然、口语化，避免机械感 | ### 格式示例 ``` 这个项目讲的是 [一句话概括]。核心结论： · 要点1 · 要点2 · 要点3 想了解更多可以追问。 ``` ### 触发规则 - 用户说"详细"、"展开"、"展开说说" → 提供 Archive 层内容 - 用户说"搜一下"、"帮我看"、"这个讲了什么" → 自动触发 Ingest + Display --- ## 📦 Archive Layer（归档层） **用途**：存入知识库，供长期查阅和检索。是内部存储格式。 ### 规范 | 属性 | 要求 | |------|------| | **格式** | Markdown + YAML frontmatter | | **结构** | `title`, `date`, `sources`, `tags` | | **内容** | 完整的事实提炼，包含 `[[wiki-link]]` | | **命名** | 使用 `--type source/entity/concept` 通过 archive-manager.py 归档 | ### 格式示例 ```yaml --- title: Hermes Agent 架构 date: 2026-04-09 sources: ["raw/hermes-paper.pdf"] tags: [agent, llm, architecture] --- # 正文内容提到了 [[OpenClaw]] 框架... ``` --- ## 📱 Channel 自适应规则目标平台：Discord / Telegram（手机端） ### 规则列表 1. **不用 Markdown 表格** — 用「·」列表或「1. 2. 3.」代替 2. **不用超过3行的代码块** — 超过则折叠或改用描述 3. **分段要短** — 每段≤3行，关键信息放最前面 4. **列表用「·」或「1. 2. 3.」** — 不用「-」或「*」 ### 自适应对照表 | 场景 | 标准格式 | Channel 自适应 | |------|----------|----------------| | 多项并列 | Markdown 表格 | 「·」列表 | | 代码展示 | 多行代码块 | 单行 or 折叠描述 | | 详细信息 | 段落展开 | 关键句 + "追问获取更多" | --- ## ✍️ 字数限制说明 | 层级 | 上限 | 说明 | |------|------|------| | Display Layer（展示层） | 200 字 | 用户直接看到的回复 | | Archive Layer（归档层） | 无限制 | 知识库存档，完整为上 | | 每段落（Channel 自适应） | 3 行 | Discord/Telegram 阅读友好 | | 代码块（Channel 自适应） | 3 行 | 超过则折叠 | --- ## 🔄 分层协作流程 ``` 用户输入 ↓ [自然触发词判断] ↓ Ingest + Query（底层4连击） ↓ Synthesize → Display Layer → 用户可见回复 ↘ Archive Layer → 知识库存档 ``` ### Display Layer 优先每次响应用户时，**必须**先输出 Display Layer。如果用户追问，再提供更深层的 Archive 内容。 ### Archive Layer 按需触发以下情况将内容存入 Archive Layer： - Ingest 时生成 source/entity/concept - Query 的答案具有长期留存价值 - Lint 时发现并修复了孤岛链接 FILE:references/spawn-template.md # ARC-Worker Spawn Templates Orchestrator 在派生 Worker 时，使用以下模板替换 `{占位符}`，避免每次手写 500+ tokens 的重复指令。 --- ## Template 1: Ingest（标准 4 连击） ```markdown ⚠️ MANDATORY: Use `cat << 'EOF' | python3 skills/arc-reactor/scripts/archive-manager.py --type [TYPE] --topic [NAME] --stdin` for ALL outputs. Execute 4-combo (source, entity, index, log) for Ingest! Do not write flat files. You MUST verify JSON receipt contains "status": "success" after each operation. Run all commands from workspace root: {WORKSPACE_ROOT} ⚠️ OUTPUT CONSTRAINT: All user-facing output MUST follow Display Layer规范 (≤200字中文摘要)，禁止暴露JSON回执、status、path、size_bytes等内部字段。详见 `references/output-style.md`。你是 ARC-Worker 矿工，执行 v4.0 Ingest 工作流。 [任务目标]: {TASK_DESCRIPTION} 素材获取指引： 1. 用 web_search 搜索 "{SEARCH_QUERY_1}" 获取中文报道 2. 用 web_search 搜索 "{SEARCH_QUERY_2}" 获取英文技术细节 3. 尝试 web_fetch 获取原始素材页面 [执行协议 - 4 连击 Ingest]: ### Hit 1: Source 页 ```bash cd {WORKSPACE_ROOT} && \ cat << 'EOF_ARC_DOC' | python3 skills/arc-reactor/scripts/archive-manager.py --type source --topic "{TOPIC_SLUG}" --stdin --- title: "{TOPIC_TITLE}" sources: [{SOURCE_URLS}] tags: [{TAGS}] --- （你的完整 Source 内容） EOF_ARC_DOC ``` ### Hit 2: Entity 页 ```bash cat << 'EOF_ARC_DOC' | python3 skills/arc-reactor/scripts/archive-manager.py --type entity --topic "{ENTITY_SLUG}" --stdin （实体词条，使用 [[Wiki-Link]] 引用相关实体） EOF_ARC_DOC ``` ### Hit 3: Index 追加 ```bash cat << 'EOF_ARC_DOC' | python3 skills/arc-reactor/scripts/archive-manager.py --type index --topic "{TOPIC_SLUG}-index" --stdin - [[{Entity-1}]]: 一句话描述 - [[{Entity-2}]]: 一句话描述 EOF_ARC_DOC ``` ### Hit 4: Log 追加 ```bash cat << 'EOF_ARC_DOC' | python3 skills/arc-reactor/scripts/archive-manager.py --type log --topic "{TOPIC_SLUG}-log" --stdin Ingested {TOPIC_SLUG}: source + entity + index completed EOF_ARC_DOC ``` 每次执行后必须验证 JSON 回执包含 "status": "success"。 ### 最终交付 4 连击全部完成后： 1. **按 Display Layer 规范输出**（见 `references/output-style.md`）： - 中文摘要，≤200字 - 结论先行，用「·」列出要点 - 自然对话风格，避免技术细节 2. **禁止向用户展示**：JSON 回执、status、path、size_bytes 等内部字段 3. **发送附件**：通过 message tool (channel=telegram, target={USER_ID}) 将 source 文件发送给用户 **Display Layer 示例**： ``` 已完成 {TOPIC_TITLE} 的知识编译。核心结论： · 提取了 {主要实体1}、{主要实体2} 的关键信息 · 建立了 {数量} 个知识节点链接 · 已存入 Wiki 供后续查询使用 ``` ``` --- ## Template 2: Query（知识查询） ```markdown 你是 ARC-Worker 矿工，执行 v4.0 Query 工作流。 [任务目标]: 回答用户关于 {TOPIC} 的问题。 [执行协议]: 1. 读取 {WORKSPACE_ROOT}/arc-reactor-doc/wiki/index.md 找相关 [[wiki-link]] 2. 读取对应 entity/source 文件内容 3. 综合已有知识回答问题 4. 如果发现知识缺口，建议新的 Ingest 目标 [已有 Wiki 实体]: {EXISTING_ENTITIES_LIST} ``` --- ## Template 3: Lint（健康检查） ```markdown 你是 ARC-Worker 矿工，执行 v4.0 Lint 工作流。 [任务目标]: 对 Wiki 知识库执行健康检查。 [执行协议]: 1. 遍历 {WORKSPACE_ROOT}/arc-reactor-doc/wiki/entities/ 下所有文件 2. 检查每个 [[wiki-link]] 是否有对应实体文件 3. 检查 index.md 是否覆盖了所有实体 4. 检查 source 文件是否都有 date 字段 5. 输出检查报告，列出：孤岛链接、缺失索引、缺失日期 6. 如发现可自动修复的问题，直接修复并汇报 ``` --- ## Template 4: Video/Audio Ingest ```markdown （在 Template 1 基础上，Hit 0 增加转录步骤） ### Hit 0: 视频/音频转录 ```bash # 下载音频 yt-dlp -x --audio-format mp3 -o "/tmp/arc-audio.%(ext)s" "{VIDEO_URL}" # 转录（Apple Silicon 本地） mlx_whisper /tmp/arc-audio.mp3 --language zh --output-format txt --output-dir /tmp/ # 读取转录文本作为 source 内容 ``` 然后执行标准 4 连击，source 内容使用转录文本。 ``` --- ## 占位符速查 | 占位符 | 说明 | 示例 | |--------|------|------| | `{WORKSPACE_ROOT}` | Agent 工作区绝对路径 | `/Users/evan/.openclaw/.../workspace` | | `{TASK_DESCRIPTION}` | 任务描述 | "对 xxx 进行深度知识编译" | | `{SEARCH_QUERY_1}` | 中文搜索词 | "xxx 技术教程 2026" | | `{SEARCH_QUERY_2}` | 英文搜索词 | "xxx tutorial setup 2026" | | `{TOPIC_SLUG}` | URL-safe 主题名 | "minimax-hermes-integration" | | `{TOPIC_TITLE}` | 可读标题 | "MiniMax 与 Hermes Agent 联动" | | `{ENTITY_SLUG}` | 实体文件名 | "minimax" | | `{SOURCE_URLS}` | 素材 URL 列表 | `"url1", "url2"` | | `{TAGS}` | 标签列表 | `"minimax", "hermes", "agent"` | | `{USER_ID}` | 用户 Telegram ID | `5930392031` | | `{EXISTING_ENTITIES_LIST}` | 已有实体列表 | `- [[Hermes-Agent]]\n- [[OpenClaw]]` | FILE:references/templates/practical_record.md # {{TITLE}} 实战记录 > 日期：{{DATE}} > 执行者：{{EXECUTOR}} > 目标：{{OBJECTIVE}} --- ## 一、方案设计 (The Design) ### 1.1 架构图 ```mermaid {{MERMAID_DIAGRAM}} ``` ### 1.2 核心组件 | 组件 | 位置 | 功能 | | :--- | :--- | :--- | | {{COMP_1}} | `{{PATH_1}}` | {{FUNC_1}} | --- ## 二、实现步骤 (Implementation) ### 2.1 {{STEP_TITLE_1}} ```bash {{COMMANDS_OR_CODE}} ``` {{STEP_DESC_1}} --- ## 三、关键发现与哲学 (Philosophy & Lessons) > **{{CORE_PRINCIPLE}}** | 维度 | 关键经验 | | :--- | :--- | | {{LAT_1}} | {{LESSON_1}} | --- ## 四、后续优化方案 (Next Steps) - [ ] {{BACKLOG_1}} --- *生成于 ARC Reactor v2.3 | 实战经验档案* FILE:references/templates/report-template.md # ARC 标准报告模板 > 所有 ARC 输出必须遵循此模板。 ```markdown # [主题] 调研报告 > 每份报告自包含完整，开箱即用。 --- ## 基本信息 | 字段 | 内容 | |------|------| | **主题** | [标题] | | **来源** | [来源平台] | | **原文 URL** | [链接] | | **内容类型** | [ARTICLE/VIDEO/REPO/PAPER/SOCIAL/TOPIC] | | **调研日期** | [YYYY-MM-DD] | | **调研人** | [Agent 名称] | | **可信度** | [VERIFIED / UNVERIFIED / DISPUTED] | --- ## 一、[第一章标题] [内容] ## 二、[第二章标题] [内容] ## 三、[第三章标题（按需扩展至 3-8 章）] [内容] --- ## 结论与建议 - 关键发现摘要 - 对我们的价值判断 - 建议后续行动 --- ## 源追踪 | # | 来源平台 | URL | 内容类型 | 采集日期 | 贡献内容 | |---|----------|-----|----------|----------|----------| | 1 | [平台] | [链接] | [类型] | [日期] | [贡献] | ## 附录 - **关联调研**：[相关报告链接] - **原始素材**：[转录稿/截图等存储路径] - **验证记录**：[哪些声明被验证/存疑] ``` ## 命名规范文件名：`[主题简称]-调研报告.md`（全中文，便于 Obsidian 检索） ## 标签体系在 `knowledge/index.md` 中标注： | 标签 | 说明 | |------|------| | `#工具` | 开发工具、SDK、框架 | | `#架构` | 系统设计、架构模式 | | `#AI` | AI/LLM 相关技术 | | `#产品` | 产品分析、竞品调研 | | `#方法论` | 工作方法、流程设计 | | `#人物` | 人物/团队调研 | FILE:references/templates/research_survey.md # {{TITLE}}：{{ONE_SENTENCE_DESCRIPTOR}} > 来源：{{SOURCE_LINK}} > 调研日期：{{DATE}} > 调研人：{{RESEARCHER}} --- ## 一、项目/工具概览 | 字段 | 内容 | | :--- | :--- | | **定位** | {{POSITIONING}} | | **核心指标** | {{METRICS}} (如 Star 数, 性能参数) | | **一句话宗旨** | {{SLOGAN}} | --- ## 二、核心特性与能力 {{DESCRIPTION_OF_FEATURES}} | 特性 | 关键点 | 备注 | | :--- | :--- | :--- | | {{FEATURE_1}} | {{POINT_1}} | {{REMARK_1}} | | {{FEATURE_2}} | {{POINT_2}} | {{REMARK_2}} | --- ## 三、技术实现逻辑 (Mechanism) {{ARCH_DIAGRAM_OR_TEXT}} {{DEEP_DIVE_ON_LOGIC}} --- ## 四、与本分部/项目的关系 (Relationship) | {{PROJECT_NAME}} | 本调研对象 ({{TITLE}}) | 差异/启示 | | :--- | :--- | :--- | | {{OUR_APPROACH}} | {{THEIR_APPROACH}} | {{INSIGHT}} | --- ## 五、结论与下一步行动 (Next Actions) - [ ] {{ACTION_1}} - [ ] {{ACTION_2}} --- *生成于 ARC Reactor v2.3 | 知识编译档案* FILE:references/templates/tech_analysis.md # EP{{EP_NO}}：{{TITLE}}深度解析 > 来源：{{SOURCE_LINK}} > 核心受众：{{TARGET_AUDIENCE}} > 调研日期：{{DATE}} --- ## 一、问题背景与痛点 (The Pain Points) 在 {{TOPIC}} 场景中，目前面临的核心挑战： 1. {{PAIN_1}} 2. {{PAIN_2}} --- ## 二、核心配置/逻辑解析 (Key Logic) ### 2.1 {{LOGIC_TITLE_1}} - **位置**: `{{FILE_PATH_OR_LOCATION}}` - **逻辑**: {{LOGIC_DESC}} - **效果**: {{EFFECT_DESC}} ### 2.2 {{LOGIC_TITLE_2}} - **位置**: `{{FILE_PATH_OR_LOCATION}}` - **逻辑**: {{LOGIC_DESC}} - **效果**: {{EFFECT_DESC}} --- ## 三、Before vs After 效果对比 | 维度 | 之前 (Before) | 之后 (After) | | :--- | :--- | :--- | | {{DIM_1}} | {{BEFORE_VAL_1}} | {{AFTER_VAL_1}} | | {{DIM_2}} | {{BEFORE_VAL_2}} | {{AFTER_VAL_2}} | --- ## 四、对当前项目的启示 (Actionable Insights) | 关键发现 | 建议操作 | 优先级 | | :--- | :--- | :--- | | {{DISCOVERY_1}} | {{ACTION_1}} | {{PRIORITY_1}} | --- *生成于 ARC Reactor v2.3 | 源码与配置深度内化档案* FILE:references/verification-pipeline.md # 广角阵列与二次复检管线 (Extended Probing & Verification Pipeline) > **极度重要指令**：从 v2.0 开始，你的 R (Research) 环节**必须具备侵略性与自主外发**能力。绝对不能只充当单页网页的“文字排头兵总结器”。你要像真正的分析专家一样，主动去挖外网、找竞品、查旧账！ ## Step 1: 原本主体的“声明切片” (Triage) 将原始采集到的原文核心声明切成三类，放上砧板： | 类别 | 需要接受的挑战 | |------|--------------| | **客观标榜** | Star 数、融资金额、开源许可类别、创始人履历。 | | **碾压性宣告** | "性能极速"、"超越GPT"、"唯一的"、"十倍提效"。 | | **独门绝技** | 原网页上提到的自己的核心护城河特性。 | ## Step 2: 主动外探与交叉验证 (Active Outbound Verification) 🚀 **此部分要求具有长篇资料外的自主性。无论原作者怎么写，你【必须调用你的 web_search 等搜索引擎工具】对外界展开求证！** - **真实验证 (Data Scrubbing)**：对于“客观标榜”，去官方源查验其真实数据是否有水分（例如吹牛 90万星其实只有 9万）。 - **打假与争端 (Dispute Sniffing)**：针对“碾压性宣告”，使用你的联网能力主动寻找 Hacker News, Reddit 或 V2EX 上的真实风评，如果找到驳斥声音（如“测试环境被污染”、“其实很容易崩溃”），**必须捕获为 `[DISPUTED]` 档案！**。 - **竞品测绘 (Landscape & Alternatives)**：这一步是你的灵魂！不要只讲这是什么，启动搜索引擎主动搜寻并补加“目前市面上它的最大替代品 (Alternatives) 是哪些，以及它与这些替代品的比较与各自优劣象限。”（如果是方法论，就抓它的同类或对立方法论进行对比说明） ## Step 3: 可信度标注与报告融合经过你的外网出击后，你将携带着带有驳斥和竞争维度的信息归来，给最初的特性附上真正的可信评级表，合并进最终的调研报告结构中。 | 等级 | 标识 | 触发条件 | |------|------|----------| | ✅ | `[VERIFIED]` | 第三方官网或权威平台佐证通过的数据 | | ⚠️ | `[UNVERIFIED]` | 作者自己口嗨的性能，你在全网也搜不出任何第三方测评或使用记录 | | ❌ | `[DISPUTED]` | 发现有外部论坛打脸的矛盾信息、存在重大历史风波、甚至存在更靠谱的成熟竞品替代选项 | | 💬 | `[OPINION]` | 纯主观输出 | FILE:scripts/archive-manager.py #!/usr/bin/env python3 import os import sys import argparse import json import hashlib import shutil import time from datetime import datetime import re FACT_INDEX_FILENAME = 'index-facts.json' FACT_SECTION_RE = re.compile(r'^###\s+(IN|OUT)-\d+:\s*(.+?)\s*$') FACT_FIELD_RE = re.compile(r'^-\s+\*\*(.+?)\*\*:\s*(.*)$') DATE_RE = re.compile(r'(\d{4}-\d{2}-\d{2})') AMOUNT_RE = re.compile(r'(\d+(?:\.\d+)?\s*(?:万|元))') PROJECT_RE = re.compile(r'([A-Za-z0-9\u4e00-\u9fff][A-Za-z0-9\u4e00-\u9fff\-_/（）()· ]{0,80}?(?:系统|项目|产品|平台|Framework|Skill))') def _fact_type_from_label(label, title='', summary=''): """将 CWork 类型字段映射为标准 fact 类型。""" normalized = (label or '').strip() context = ' '.join([normalized, title or '', summary or '']) lowered = context.lower() if normalized == '合同管理(OPS)': return 'contract' if normalized == '会议室预约': return 'meeting' if normalized == '销售项目': return 'project-progress' if '日报' in context: return 'daily-report' if '测试' in context or 'test' in lowered: return 'skill-test' if normalized == '其他汇报': return 'other' return 'other' def _extract_fact_status(text): """基于标题与摘要做轻量状态推断。""" normalized = (text or '').strip() if not normalized: return 'unknown' if re.search(r'全部通过|通过率\s*100%|已完成|全部修复|现已全部修复|成功跑通|(? indent: index += 1 continue if line['text'].startswith('- '): break key, sep, remainder = line['text'].partition(':') if not sep: index += 1 continue key = key.strip() remainder = remainder.strip() if remainder: data[key] = _parse_yaml_scalar(remainder) index += 1 continue next_index = index + 1 if next_index < len(processed) and processed[next_index]['indent'] > indent: child_indent = processed[next_index]['indent'] if processed[next_index]['text'].startswith('- '): data[key], index = parse_list(next_index, child_indent) else: data[key], index = parse_mapping(next_index, child_indent) else: data[key] = {} index += 1 return data, index def parse_list(index, indent): items = [] while index < len(processed): line = processed[index] if line['indent'] < indent: break if line['indent'] > indent: index += 1 continue if not line['text'].startswith('- '): break body = line['text'][2:].strip() if ':' in body: key, _, remainder = body.partition(':') item = {key.strip(): _parse_yaml_scalar(remainder.strip()) if remainder.strip() else {}} index += 1 if index < len(processed) and processed[index]['indent'] > indent: child_block, index = parse_mapping(index, processed[index]['indent']) item.update(child_block) items.append(item) continue if body: items.append(_parse_yaml_scalar(body)) index += 1 continue index += 1 if index < len(processed) and processed[index]['indent'] > indent: child_indent = processed[index]['indent'] if processed[index]['text'].startswith('- '): item, index = parse_list(index, child_indent) else: item, index = parse_mapping(index, child_indent) items.append(item) return items, index parsed, _ = parse_mapping(0, 0) return parsed def _find_config_path(start_path=None): """向上查找配置文件路径。""" search_dir = os.path.abspath(start_path or os.getcwd()) if os.path.isfile(search_dir): search_dir = os.path.dirname(search_dir) while True: direct_path = os.path.join(search_dir, 'arc-reactor-config.yaml') nested_path = os.path.join(search_dir, 'skills', 'arc-reactor', 'arc-reactor-config.yaml') if os.path.exists(direct_path): return direct_path if os.path.exists(nested_path): return nested_path parent_dir = os.path.dirname(search_dir) if parent_dir == search_dir: return None search_dir = parent_dir def load_config(start_path=None): """查找并解析 `arc-reactor-config.yaml`。""" config_path = _find_config_path(start_path) if not config_path: return {} return _parse_simple_yaml(config_path) def resolve_kb_root(content_source, tags=None): """根据来源或标签自动匹配知识库根目录。""" config = load_config() knowledge_bases = config.get('knowledge_bases', []) or [] normalized_tags = {str(tag) for tag in (tags or [])} for kb_entry in knowledge_bases: auto_route = kb_entry.get('auto_route', {}) or {} sources = auto_route.get('sources', []) or [] if content_source in sources: return kb_entry.get('root') for kb_entry in knowledge_bases: auto_route = kb_entry.get('auto_route', {}) or {} route_tags = {str(tag) for tag in (auto_route.get('tags', []) or [])} if normalized_tags & route_tags: return kb_entry.get('root') return None def _format_kb_yaml_entry(kb_entry): """格式化 knowledge_bases 条目为 YAML 文本。""" name = kb_entry['name'] root = kb_entry['root'] description = kb_entry.get('description', '') auto_route = kb_entry.get('auto_route', {}) or {} sources = auto_route.get('sources', []) or [] tags = auto_route.get('tags', []) or [] sources_str = ', '.join(json.dumps(item, ensure_ascii=False) for item in sources) tags_str = ', '.join(json.dumps(item, ensure_ascii=False) for item in tags) return ( f" - name: {name}\n" f" root: {root}\n" f" description: {json.dumps(description, ensure_ascii=False)}\n" f" auto_route:\n" f" sources: [{sources_str}]\n" f" tags: [{tags_str}]\n" ) def kb_init(root_name, name=None, description=None): """初始化一个新的多实例知识库并写回配置。""" config_path = _find_config_path() if not config_path: return {"status": "error", "message": "arc-reactor-config.yaml 未找到"} config = load_config() knowledge_bases = config.get('knowledge_bases', []) or [] kb_name = name or root_name kb_description = description or '' for kb_entry in knowledge_bases: if kb_entry.get('name') == kb_name: return {"status": "error", "message": f"知识库名称已存在: {kb_name}"} if kb_entry.get('root') == root_name: return {"status": "error", "message": f"知识库根目录已存在: {root_name}"} workspace_root = os.path.dirname(os.path.dirname(os.path.dirname(config_path))) kb_root = os.path.join(workspace_root, root_name) dirs_to_create = [ os.path.join(kb_root, 'wiki', 'sources'), os.path.join(kb_root, 'wiki', 'entities'), os.path.join(kb_root, 'wiki', 'concepts'), os.path.join(kb_root, 'raw'), ] files_to_create = [ os.path.join(kb_root, 'wiki', 'index.md'), os.path.join(kb_root, 'wiki', 'log.md'), os.path.join(kb_root, 'wiki', FACT_INDEX_FILENAME), ] created = [] try: for dir_path in dirs_to_create: if not os.path.exists(dir_path): os.makedirs(dir_path, exist_ok=True) created.append(dir_path) for file_path in files_to_create: if not os.path.exists(file_path): with open(file_path, 'w', encoding='utf-8') as file_obj: if file_path.endswith(FACT_INDEX_FILENAME): file_obj.write('[]') else: file_obj.write('') created.append(file_path) new_entry = { 'name': kb_name, 'root': root_name, 'description': kb_description, 'auto_route': { 'sources': [], 'tags': [], }, } with open(config_path, 'r', encoding='utf-8') as file_obj: config_text = file_obj.read().rstrip() yaml_entry = _format_kb_yaml_entry(new_entry) if 'knowledge_bases:' in config_text: updated_text = f"{config_text}\n\n{yaml_entry.rstrip()}\n" else: updated_text = f"{config_text}\n\nknowledge_bases:\n{yaml_entry}" with open(config_path, 'w', encoding='utf-8') as file_obj: file_obj.write(updated_text) return { "status": "success", "action": "kb_init", "name": kb_name, "root": root_name, "path": kb_root, "config": config_path, "created": created, "message": "Knowledge base initialized." } except Exception as exc: return {"status": "error", "message": f"知识库初始化失败: {str(exc)}"} def kb_list(): """列出所有已配置知识库的统计信息。""" config = load_config() knowledge_bases = config.get('knowledge_bases', []) or [] config_path = _find_config_path() if not config_path: return [] workspace_root = os.path.dirname(os.path.dirname(os.path.dirname(config_path))) results = [] for kb_entry in knowledge_bases: kb_root = os.path.join(workspace_root, kb_entry.get('root', '')) sources_dir = os.path.join(kb_root, 'wiki', 'sources') entities_dir = os.path.join(kb_root, 'wiki', 'entities') concepts_dir = os.path.join(kb_root, 'wiki', 'concepts') def count_markdown_files(root_dir): count = 0 if not os.path.exists(root_dir): return count for _, _, files in os.walk(root_dir): count += sum(1 for filename in files if filename.endswith('.md')) return count latest_mtime = None if os.path.exists(kb_root): for root, _, files in os.walk(kb_root): for filename in files: file_path = os.path.join(root, filename) file_mtime = os.path.getmtime(file_path) if latest_mtime is None or file_mtime > latest_mtime: latest_mtime = file_mtime results.append({ 'name': kb_entry.get('name'), 'root': kb_entry.get('root'), 'description': kb_entry.get('description', ''), 'sources_count': count_markdown_files(sources_dir), 'entities_count': count_markdown_files(entities_dir), 'concepts_count': count_markdown_files(concepts_dir), 'last_modified': datetime.fromtimestamp(latest_mtime).strftime('%Y-%m-%d %H:%M:%S') if latest_mtime else None, 'exists': os.path.exists(kb_root), }) return results def slugify(text): """简单规范化文件名""" text = str(text).lower() text = re.sub(r'[^\w\s-]', '', text) text = re.sub(r'[\s_-]+', '-', text).strip('-') return text if text else "untitled" WIKI_LINK_RE = re.compile(r'\[\[([^\]]+)\]\]') FRONTMATTER_RE = re.compile(r'^---\s*\n(.*?)\n---\s*\n', re.DOTALL) def resolve_entity_path(doc_root, topic): """在 entities 或 concepts 目录中寻找实体的物理路径。""" slug = slugify(topic) paths = [ os.path.join(doc_root, 'wiki', 'entities', f"{slug}.md"), os.path.join(doc_root, 'wiki', 'concepts', f"{slug}.md") ] for p in paths: if os.path.exists(p): return p return None def _count_link_mentions_in_sources(sources_dir, link_text): """统计 wiki/sources/ 目录中提及指定 link 文本的 .md 文件数量。使用 grep -rl 扫描，返回命中文件数（非行数），用于判断该 link 是否值得保留。 """ import subprocess if not os.path.exists(sources_dir): return 0 pattern = re.escape(link_text) try: cmd = ["grep", "-rl", "--include=*.md", pattern, sources_dir] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: return len([l for l in result.stdout.splitlines() if l.strip()]) except Exception: pass return 0 def sanitize_wiki_links(content, doc_root, existing_content=None): """扫描并清理孤儿 wiki-link，防止产生大量指向不存在页面的 [[link]]。处理逻辑： 1. 提取所有 [[wiki-link]] 2. 检查 wiki/entities/{slug}.md 和 wiki/concepts/{slug}.md 是否存在 3. 若目标不存在，统计 wiki/sources/ 中被提及的文件数： - >= 3 次：保留 [[link]]（值得建页） - 2 次：降级为 **link**（加粗） - 1 或 0 次：降级为纯文本（去掉 [[]]） 4. 如果传入了 existing_content，只对新增部分做清理，不动已有内容。 """ # 若为追加模式（append），分离已有内容与新增内容，只清理新增部分 if existing_content: # 用最长公共前缀定位新增内容的起点 common_len = 0 min_len = min(len(existing_content), len(content)) for i in range(min_len): if existing_content[i] == content[i]: common_len += 1 else: break prefix = content[:common_len] suffix = content[common_len:] # 只对新增后缀做 sanitization sanitized_suffix = sanitize_wiki_links(suffix, doc_root, existing_content=None) return prefix + sanitized_suffix sources_dir = os.path.join(doc_root, 'wiki', 'sources') def _replacer(match): link_text = match.group(1) slug = slugify(link_text) # 检查目标实体/概念文件是否存在 entity_path = os.path.join(doc_root, 'wiki', 'entities', f"{slug}.md") concept_path = os.path.join(doc_root, 'wiki', 'concepts', f"{slug}.md") if os.path.exists(entity_path) or os.path.exists(concept_path): # 目标页面已存在，保留 wiki-link return match.group(0) # 目标不存在，统计在 sources 中的提及次数 mention_count = _count_link_mentions_in_sources(sources_dir, link_text) if mention_count >= 3: # 高频提及，保留 wiki-link（未来值得建页） return match.group(0) elif mention_count == 2: # 中频提及，降级为加粗 return f"**{link_text}**" else: # 低频提及（1 或 0），降级为纯文本 return link_text return WIKI_LINK_RE.sub(_replacer, content) def find_backlinks(doc_root, topic): """使用 grep 扫描 wiki/sources 目录，寻找引用了该实体的源文件。""" sources_dir = os.path.join(doc_root, 'wiki', 'sources') if not os.path.exists(sources_dir): return [] import subprocess # 搜索格式为 [[Topic]] 或 [[topic]] pattern = f"\\[\\[{topic}\\]\\]" # 使用 grep -rl 获取包含该字符串的文件名列表 try: cmd = ["grep", "-rl", "--include=*.md", pattern, sources_dir] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: return [line.strip() for line in result.stdout.splitlines() if line.strip()] except Exception: pass return [] def export_entity(doc_root, topic, output_dir=None): """导出实体及其关联上下文为单一 Markdown 文档。""" entity_path = resolve_entity_path(doc_root, topic) if not entity_path: return {"status": "error", "message": f"未找到实体: {topic}"} wiki_dir = os.path.join(doc_root, 'wiki') if not output_dir: output_dir = os.path.join(doc_root, 'exports') os.makedirs(output_dir, exist_ok=True) # 1. 读取实体内容 with open(entity_path, 'r', encoding='utf-8') as f: entity_content = f.read() # 2. 发现关联素材 (Sources) # A. 从 Frontmatter 读取 linked_sources = [] fm_match = FRONTMATTER_RE.match(entity_content) if fm_match: fm_text = fm_match.group(1) # 简单解析 YAML 列表 sources_match = re.search(r'sources:\s*\[(.*?)\]', fm_text) if sources_match: raw_sources = sources_match.group(1) # 这里的路径通常是相对路径或 Slug linked_sources = [s.strip().strip('"').strip("'") for s in raw_sources.split(',')] # B. 发现反向链接 backlinks = find_backlinks(doc_root, topic) # 3. 发现关联实体 (Related Entities) related_entities = WIKI_LINK_RE.findall(entity_content) unique_related = list(set(slugify(e) for e in related_entities if slugify(e) != slugify(topic))) # 4. 构建导出文档 now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S') lines = [ f"# {topic} — Knowledge Export", f"**Generated**: {now_str} | **Source**: ARC Reactor Wiki", "\n---", "\n## 1. Overview (Entity Content)", entity_content, "\n---", "\n## 2. Connections (Related Entities)" ] if not unique_related: lines.append("\n*No direct wiki-entity connections found.*") else: for r_slug in unique_related: r_path = resolve_entity_path(doc_root, r_slug) if r_path: with open(r_path, 'r', encoding='utf-8') as f: r_content = f.read() lines.append(f"\n### [[{r_slug}]]") lines.append(r_content) lines.append("\n---") lines.append("\n## 3. Origins (Source Materials)") all_source_paths = set() # 尝试解析 Frontmatter 里的 source 路径 for s_ref in linked_sources: # 这里逻辑较简化，尝试在 wiki/sources 递归寻找对应文件 for root, _, files in os.walk(os.path.join(wiki_dir, 'sources')): for filename in files: if s_ref in filename or slugify(s_ref) in filename: all_source_paths.add(os.path.join(root, filename)) # 添加反向链接命中的路径 for b_path in backlinks: all_source_paths.add(b_path) if not all_source_paths: lines.append("\n*No original source materials found.*") else: for s_path in all_source_paths: rel_name = os.path.relpath(s_path, os.path.join(wiki_dir, 'sources')) with open(s_path, 'r', encoding='utf-8') as f: s_content = f.read() lines.append(f"\n### Source: {rel_name}") lines.append(s_content) # 5. 写入文件 dest_path = os.path.join(output_dir, f"{slugify(topic)}-bundle.md") with open(dest_path, 'w', encoding='utf-8') as f: f.write('\n'.join(lines)) return { "status": "success", "action": "export_entity", "topic": topic, "path": dest_path, "connections_found": len(unique_related), "sources_found": len(all_source_paths) } def find_doc_root(start_path, root_name='arc-reactor-doc'): """Walk up from start_path to find the workspace root.""" curr_dir = start_path while curr_dir and curr_dir != '/': if os.path.exists(os.path.join(curr_dir, root_name)): return os.path.join(curr_dir, root_name) if os.path.exists(os.path.join(curr_dir, '.git')): return os.path.join(curr_dir, root_name) curr_dir = os.path.dirname(curr_dir) return os.path.join(start_path, root_name) def lint_wiki(doc_root, fix=False): """Health check for Wiki integrity. Returns issues found.""" issues = [] fixed = [] wiki_dir = os.path.join(doc_root, 'wiki') if not os.path.exists(wiki_dir): return {"status": "error", "message": f"Wiki directory not found: {wiki_dir}"} # 1. Scan all entity files entities_dir = os.path.join(wiki_dir, 'entities') existing_entities = set() if os.path.exists(entities_dir): for f in os.listdir(entities_dir): if f.endswith('.md'): existing_entities.add(f[:-3]) # strip .md # 2. Scan all source files for wiki-links all_links = set() sources_dir = os.path.join(wiki_dir, 'sources') all_files = [] # Collect all markdown files for root, dirs, files in os.walk(wiki_dir): for f in files: if f.endswith('.md'): all_files.append(os.path.join(root, f)) for fpath in all_files: try: with open(fpath, 'r', encoding='utf-8') as f: content = f.read() # Find all [[wiki-links]] links = WIKI_LINK_RE.findall(content) for link in links: slug = slugify(link) all_links.add(slug) # Check for orphan links if slug not in existing_entities: rel_path = os.path.relpath(fpath, doc_root) issues.append({ "type": "orphan_link", "link": f"[[{link}]]", "slug": slug, "found_in": rel_path, "fix": f"Create entity: wiki/entities/{slug}.md" }) # Auto-fix: create stub entity if fix and not os.path.exists(os.path.join(entities_dir, f"{slug}.md")): stub = f"---\ndate: {datetime.now().strftime('%Y-%m-%d')}\n---\n\n# {link}\n\n> Stub entity — awaiting Ingest completion.\n" stub_path = os.path.join(entities_dir, f"{slug}.md") with open(stub_path, 'w', encoding='utf-8') as ef: ef.write(stub) fixed.append(f"Created stub: wiki/entities/{slug}.md") except Exception: continue # 3. Check entities not in index.md index_path = os.path.join(wiki_dir, 'index.md') indexed_entities = set() if os.path.exists(index_path): with open(index_path, 'r', encoding='utf-8') as f: index_content = f.read() indexed_links = WIKI_LINK_RE.findall(index_content) for link in indexed_links: indexed_entities.add(slugify(link)) for entity in existing_entities: if entity not in indexed_entities: issues.append({ "type": "missing_from_index", "entity": entity, "fix": f"Add [[{entity}]] to wiki/index.md" }) # 4. Check source files missing date if os.path.exists(sources_dir): for root, dirs, files in os.walk(sources_dir): for f in files: if f.endswith('.md'): fpath = os.path.join(root, f) try: with open(fpath, 'r', encoding='utf-8') as sf: content = sf.read() if 'date:' not in content[:500].lower(): rel_path = os.path.relpath(fpath, doc_root) issues.append({ "type": "missing_date", "file": rel_path, "fix": "Add date field to frontmatter" }) except Exception: continue # 5. Check for empty files (< 50 bytes) for fpath in all_files: try: if os.path.getsize(fpath) < 50: rel_path = os.path.relpath(fpath, doc_root) issues.append({ "type": "empty_file", "file": rel_path, "fix": "Populate or remove empty file" }) except Exception: continue return { "status": "lint_complete", "total_files": len(all_files), "total_entities": len(existing_entities), "total_links": len(all_links), "issues_found": len(issues), "issues_fixed": len(fixed), "issues": issues, "fixed": fixed } def validate_files(doc_root, paths=None): """验证文件是否存在且有效。 Args: doc_root: 文档根目录路径 paths: 要验证的文件路径列表（相对于 doc_root），如果为 None 则验证整个 Wiki Returns: 包含验证结果的 JSON 可序列化字典 """ if not paths: # 验证整个 Wiki 结构 wiki_dir = os.path.join(doc_root, 'wiki') if not os.path.exists(wiki_dir): return {"status": "error", "message": f"Wiki directory not found: {wiki_dir}"} files_valid = 0 files_invalid = [] files_empty = [] # 遍历所有 Markdown 文件 for root, dirs, files in os.walk(wiki_dir): for f in files: if f.endswith('.md'): fpath = os.path.join(root, f) try: # 验证文件可读且非空 with open(fpath, 'r', encoding='utf-8') as file_obj: content = file_obj.read() if not content.strip(): files_empty.append(fpath) files_invalid.append(fpath) else: files_valid += 1 except Exception as e: files_invalid.append(fpath) return { "status": "ok" if not files_invalid else "partial", "action": "validate_wiki", "files_valid": files_valid, "files_invalid": len(files_invalid), "files_empty": len(files_empty), "invalid_files": files_invalid, "message": f"Validation complete: {files_valid} valid, {len(files_invalid)} invalid ({len(files_empty)} empty)" } else: # 验证特定文件 results = [] all_valid = True for path in paths: full_path = os.path.join(doc_root, path) if os.path.exists(full_path): try: with open(full_path, 'r', encoding='utf-8') as file_obj: content = file_obj.read() if content.strip(): results.append({ "path": path, "status": "valid", "size": len(content) }) else: results.append({ "path": path, "status": "empty", "size": 0 }) all_valid = False except Exception as e: results.append({ "path": path, "status": "error", "error": str(e) }) all_valid = False else: results.append({ "path": path, "status": "not_found" }) all_valid = False return { "status": "ok" if all_valid else "partial", "action": "validate_files", "results": results, "message": f"Validated {len(results)} files: {sum(1 for r in results if r['status'] == 'valid')} valid" } def validate_obsidian_config(vault_path): """Validate Obsidian vault configuration.""" vault = os.path.expanduser(vault_path) if not os.path.isdir(vault): return False, "Obsidian 库路径不存在" test_path = os.path.join(vault, '.arc-sync-test') try: with open(test_path, 'w') as f: f.write('ping') os.remove(test_path) return True, "OK" except Exception as e: return False, f"无写入权限: {str(e)}" def sync_to_obsidian(source_path, vault_path, target_dir, max_retries=3, retry_delay=300): """Sync a source file to Obsidian vault with exponential backoff.""" now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S') now_date = datetime.now().strftime('%Y-%m-%d') is_valid, msg = validate_obsidian_config(vault_path) if not is_valid: return {"status": "error", "action": "obsidian_sync", "source": source_path, "error": msg, "retry_count": 0, "message": f"Obsidian sync failed: {msg}"} if not os.path.exists(source_path): return {"status": "error", "action": "obsidian_sync", "source": source_path, "error": "源文件不存在", "retry_count": 0, "message": "Obsidian sync failed: source file not found"} resolved_target = target_dir.replace('{date}', now_date) dest_dir = os.path.join(os.path.expanduser(vault_path), resolved_target) filename = os.path.basename(source_path) dest_path = os.path.join(dest_dir, filename) last_error = None for attempt in range(max_retries): try: os.makedirs(dest_dir, exist_ok=True) shutil.copy2(source_path, dest_path) sync_marker = f"\n\n---\n同步状态: ✅ Obsidian (时间: {now_str})\n---\n" with open(dest_path, 'a', encoding='utf-8') as f: f.write(sync_marker) return {"status": "success", "action": "obsidian_sync", "source": source_path, "destination": dest_path, "obsidian_vault": vault_path, "sync_time": now_str, "retry_count": attempt, "message": "Obsidian sync complete."} except Exception as e: last_error = str(e) if attempt < max_retries - 1: time.sleep(retry_delay * (2 ** attempt)) continue return {"status": "error", "action": "obsidian_sync", "source": source_path, "error": last_error, "retry_count": max_retries, "message": f"Obsidian sync failed after {max_retries} retries: {last_error}"} def main(): parser = argparse.ArgumentParser(description='ARC Reactor V4 Archive Manager (Karpathy Wiki Edition)') parser.add_argument('--lint', action='store_true', help='Run Wiki health check') parser.add_argument('--fix', action='store_true', help='Auto-fix issues found during lint') parser.add_argument('--validate', action='store_true', help='Validate file integrity and existence') parser.add_argument('--kb-init', action='store_true', help='初始化新的知识库实例') parser.add_argument('--kb-list', action='store_true', help='列出所有已配置知识库') parser.add_argument('--query-facts', action='store_true', help='查询事实索引') parser.add_argument('--export-entity', help='导出实体及其关联上下文') parser.add_argument('--type', choices=[ 'raw', 'source', 'entity', 'concept', 'index', 'log', 'template', 'fact-index' ], required=False, help='归档进入的 Wiki 圈层类型') parser.add_argument('--sync-obsidian', action='store_true', help='Sync source file to Obsidian vault') parser.add_argument('--source', required=False, default=None, help='Source file path (used with --sync-obsidian)') parser.add_argument('--vault', required=False, default=None, help='Obsidian vault path (used with --sync-obsidian)') parser.add_argument('--target', required=False, default='github分享/AI调研/{date}/', help='Target subdirectory in vault') parser.add_argument('--async', dest='async_mode', action='store_true', help='Async mode: return immediately, sync in background') parser.add_argument('--topic', required=False, default='knowledge-node', help='话题/实体名 (用于生成文件名)') parser.add_argument('--stdin', action='store_true', help='强制通过标准输入读取内容 (防止转义错误)') parser.add_argument('--root', default='arc-reactor-doc', help='文档根目录名称') parser.add_argument('--filter', action='append', default=None, help='事实查询过滤条件，格式为 field=value，可重复传入') parser.add_argument('--name', required=False, default=None, help='KB 显示名称，仅用于 --kb-init') parser.add_argument('--description', required=False, default=None, help='KB 描述，仅用于 --kb-init') parser.add_argument('--date', required=False, default=None, help='指定的日期戳，缺省为今日') parser.add_argument('--dedup', choices=['merge', 'skip', 'overwrite'], default='overwrite', help='去重策略: merge=增量合并, skip=跳过, overwrite=覆盖(默认)') parser.add_argument('--url', help='直接从 URL 抓取正文内容 (支持智能反爬)') parser.add_argument('--output-dir', help='导出目标目录') args = parser.parse_args() # Obsidian sync mode if args.sync_obsidian: auto_sync = os.environ.get('AUTO_SYNC', 'true') if auto_sync.lower() in ('false', '0', 'no'): print(json.dumps({"status": "skipped", "action": "obsidian_sync", "reason": "AUTO_SYNC=false", "message": "Obsidian sync disabled via AUTO_SYNC=false"}, ensure_ascii=False)) sys.exit(0) if not args.source: print(json.dumps({"status": "error", "message": "--sync-obsidian requires --source"})) sys.exit(1) vault_path = args.vault or os.environ.get('OBSIDIAN_VAULT_PATH', '') if not vault_path: print(json.dumps({"status": "error", "message": "OBSIDIAN_VAULT_PATH not configured"})) sys.exit(1) if getattr(args, 'async_mode', False): import subprocess sync_cmd = [sys.executable, os.path.abspath(__file__), '--sync-obsidian', '--source', args.source, '--vault', vault_path, '--target', args.target] subprocess.Popen(sync_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True) print(json.dumps({"status": "pending", "action": "obsidian_sync", "source": args.source, "message": "Obsidian sync started in background."}, ensure_ascii=False)) sys.exit(0) else: result = sync_to_obsidian(args.source, vault_path, args.target) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0 if result.get('status') == 'success' else 1) if args.query_facts: result = query_facts(args.root, filters=args.filter) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0) if args.kb_init: result = kb_init(args.root, name=args.name, description=args.description) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0) if args.export_entity: cwd = os.getcwd() doc_root = find_doc_root(cwd, args.root) result = export_entity(doc_root, args.export_entity, output_dir=args.output_dir) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0) if args.kb_list: result = kb_list() print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0) # Lint mode — separate flow if args.lint: cwd = os.getcwd() doc_root = find_doc_root(cwd, args.root) result = lint_wiki(doc_root, fix=args.fix) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0) # Validate mode — verify file integrity if args.validate: cwd = os.getcwd() doc_root = find_doc_root(cwd, args.root) result = validate_files(doc_root) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0) # Normal archive mode — type is required if not args.type: print(json.dumps({"status": "error", "message": "归档模式需要 --type 参数，或使用 --lint / --kb-init / --kb-list"})) sys.exit(1) # 0. 预计算常用值 topic_slug = slugify(args.topic) now_date = datetime.now().strftime('%Y-%m-%d') now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 1. 获取内容 (来自 URL 或 STDIN) content_to_write = "" if args.url: try: # P5: Ensure scripts directory is in path for dynamic import script_dir = os.path.dirname(os.path.abspath(__file__)) if script_dir not in sys.path: sys.path.append(script_dir) import smart_fetcher print(json.dumps({"status": "processing", "message": f"正在尝试从 URL 摄入: {args.url}"}, ensure_ascii=False)) content_to_write = smart_fetcher.smart_extract(args.url) if not content_to_write: print(json.dumps({"status": "error", "message": "无法从该 URL 提取有效内容，抓取器返回为空"})) sys.exit(1) except ImportError: # 如果没法直接 import（比如在不同目录），尝试 subprocess 调用 import subprocess script_path = os.path.join(os.path.dirname(__file__), 'smart_fetcher.py') result = subprocess.run([sys.executable, script_path, args.url], capture_output=True, text=True) if result.returncode == 0: # smart_fetcher.py 在 __main__ 下会打印内容，需要通过正则提取 full_out = result.stdout match = re.search(r'--- EXTRACTED CONTENT START ---\n(.*?)\n--- EXTRACTED CONTENT END ---', full_out, re.DOTALL) if match: content_to_write = match.group(1) else: content_to_write = full_out # Fallback else: print(json.dumps({"status": "error", "message": "smart_fetcher 脚本执行失败", "details": result.stderr})) sys.exit(1) elif args.stdin: content_to_write = sys.stdin.read() else: print(json.dumps({"status": "error", "message": "安全限制: 必须提供 --url 或使用 --stdin 通过管道传参"})) sys.exit(1) if not content_to_write or not content_to_write.strip(): print(json.dumps({"status": "error", "message": "抓取或读取的内容为空"})) sys.exit(1) # 2. 确定物理路径 cwd = os.getcwd() doc_root = find_doc_root(cwd, args.root) if args.type == 'fact-index': result = ingest_fact_index(doc_root, content_to_write) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0) target_dir = "" filename = "" # topic_slug already computed above # Wiki 层路由判定 if args.type == 'raw': target_dir = os.path.join(doc_root, 'raw') filename = f"{topic_slug}.md" elif args.type == 'source': date_dir = args.date if args.date else now_date target_dir = os.path.join(doc_root, 'wiki', 'sources', date_dir) filename = f"{topic_slug}.md" elif args.type == 'entity': target_dir = os.path.join(doc_root, 'wiki', 'entities') filename = f"{topic_slug}.md" elif args.type == 'concept': target_dir = os.path.join(doc_root, 'wiki', 'concepts') filename = f"{topic_slug}.md" elif args.type == 'index': target_dir = os.path.join(doc_root, 'wiki') filename = "index.md" elif args.type == 'log': target_dir = os.path.join(doc_root, 'wiki') filename = "log.md" elif args.type == 'template': target_dir = os.path.join(doc_root, 'references', 'templates') filename = f"custom_{topic_slug}.md" # 3. 稳健创建父目录 try: os.makedirs(target_dir, exist_ok=True) except Exception as e: print(json.dumps({"status": "error", "message": f"目录自愈创建失败: {str(e)}"})) sys.exit(1) # 4. 路由特定的写入模式 (Append vs Overwrite) target_path = os.path.join(target_dir, filename) mode = 'w' final_write_content = content_to_write # now_str and now_date already computed above if args.type in ['index', 'log']: # index 和 log 永远是增量追加 mode = 'a' if args.type == 'log': final_write_content = f"\n- **[{now_str}]** {content_to_write.strip()}" else: # index.md final_write_content = f"\n{content_to_write.strip()}" elif args.type in ['entity', 'concept'] and os.path.exists(target_path): # 知识节点增量追加 mode = 'a' final_write_content = f"\n\n---\n## 增量知识点合入 ({now_str})\n\n" + content_to_write # 自动探测并注入 Frontmatter 时间戳 (仅对新建的 Markdown 有效) if mode == 'w' and args.type in ['source', 'entity', 'concept', 'raw', 'fact-index']: # 使用正则表达式匹配 Frontmatter 块 fm_match = FRONTMATTER_RE.match(final_write_content.lstrip()) if fm_match: fm_text = fm_match.group(1) # 检查内部是否已有 date 字段 if not re.search(r'^date:', fm_text, re.MULTILINE | re.IGNORECASE): # 在第一个 --- 下方插入 date insertion = f"---\ndate: {now_date}\n" final_write_content = re.sub(r'^---\s*\n', insertion, final_write_content.lstrip(), count=1) else: # 完全没有 YAML Frontmatter 格式，强制注入一个 final_write_content = f"---\ndate: {now_date}\n---\n\n" + final_write_content.lstrip() # 4.5 去重检查 (dedup check) dedup_status = "new" if os.path.exists(target_path) and args.dedup != 'overwrite': existing_size = os.path.getsize(target_path) if args.dedup == 'skip': with open(target_path, 'rb') as ef: existing_checksum = hashlib.sha256(ef.read()).hexdigest() receipt = { "status": "skipped", "dedup": "skipped", "type_routed": args.type, "path": target_path, "size_bytes": existing_size, "checksum": existing_checksum, "message": f"Entity already exists ({existing_size} bytes). Skipped per --dedup skip." } print(json.dumps(receipt, ensure_ascii=False)) sys.exit(0) elif args.dedup == 'merge': # For entity/concept: append (mode already set above) # For source: refuse to merge, warn instead if args.type == 'source': receipt = { "status": "skipped", "dedup": "source_exists", "type_routed": args.type, "path": target_path, "message": f"Source already exists. Use --dedup overwrite to replace, or pick a different topic." } print(json.dumps(receipt, ensure_ascii=False)) sys.exit(0) dedup_status = "merged" # 4.6 Link sanitization — 仅对 entity/concept 写入生效，防止孤儿 wiki-link if args.type in ['entity', 'concept']: existing_content_for_sanitize = None if mode == 'a' and os.path.exists(target_path): # 追加模式：读取已有内容，只清理新增部分 try: with open(target_path, 'r', encoding='utf-8') as ef: existing_content_for_sanitize = ef.read() except Exception: pass final_write_content = sanitize_wiki_links( final_write_content, doc_root, existing_content=existing_content_for_sanitize ) # 5. 原子落盘及防幻觉回执生成 try: with open(target_path, mode, encoding='utf-8') as f: f.write(final_write_content) with open(target_path, 'rb') as f: file_bytes = f.read() checksum = hashlib.sha256(file_bytes).hexdigest() size_bytes = len(file_bytes) receipt = { "status": "success", "dedup": dedup_status, "type_routed": args.type, "path": target_path, "size_bytes": size_bytes, "checksum": checksum, "date": now_date, "message": "Karpathy Wiki Layer write valid." } print(json.dumps(receipt, ensure_ascii=False)) sys.exit(0) except Exception as e: print(json.dumps({"status": "error", "message": f"I/O 崩溃: {str(e)}"})) sys.exit(1) if __name__ == "__main__": main() FILE:scripts/context-injector.py #!/usr/bin/env python3 import os import sys import argparse import json import re from datetime import datetime # 复用 slugify 逻辑 def slugify(text): text = str(text).lower() text = re.sub(r'[^\w\s-]', '', text) text = re.sub(r'[\s_-]+', '-', text).strip('-') return text if text else "untitled" def find_doc_root(start_path, root_name='arc-reactor-doc'): """向上查找工作区根目录""" curr_dir = os.path.abspath(start_path) while curr_dir and curr_dir != '/': if os.path.exists(os.path.join(curr_dir, root_name)): return os.path.join(curr_dir, root_name) if os.path.exists(os.path.join(curr_dir, '.git')): return os.path.join(curr_dir, root_name) parent = os.path.dirname(curr_dir) if parent == curr_dir: break curr_dir = parent return os.path.join(start_path, root_name) def _parse_simple_yaml(filepath): """简单 YAML 解析，用于读取配置""" if not os.path.exists(filepath): return {} config = {} try: with open(filepath, 'r', encoding='utf-8') as f: for line in f: if ':' in line and not line.strip().startswith('#'): key, val = line.split(':', 1) # 剥离行内注释 val = val.split('#', 1)[0].strip() config[key.strip()] = val.strip('"').strip("'") except: pass return config def main(): parser = argparse.ArgumentParser(description='ARC Reactor Context Injector') parser.add_argument('--query', required=True, help='User input query to match against knowledge base') parser.add_argument('--root', default='arc-reactor-doc', help='Document root name') args = parser.parse_args() cwd = os.getcwd() doc_root = find_doc_root(cwd, args.root) wiki_dir = os.path.join(doc_root, 'wiki') index_path = os.path.join(wiki_dir, 'index.md') entities_dir = os.path.join(wiki_dir, 'entities') # 1. 加载配置限额 config_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'arc-reactor-config.yaml') config = _parse_simple_yaml(config_path) max_entities = int(config.get('max_entities', 3)) max_chars = int(config.get('max_chars', 10000)) if not os.path.exists(index_path): # 如果索引不存在，静默退出 sys.exit(0) # 2. 从 index.md 提取所有实体 entities_in_index = [] try: with open(index_path, 'r', encoding='utf-8') as f: content = f.read() # 匹配 [[Entity Name]] wiki_links = re.findall(r'\[\[([^\]]+)\]\]', content) for link in wiki_links: entities_in_index.append({ "original": link, "slug": slugify(link) }) except Exception as e: print(f"Error reading index: {e}", file=sys.stderr) sys.exit(1) # 3. 匹配检索词 (简单关键词碰撞) query_lower = args.query.lower() hit_entities = [] seen_slugs = set() for item in entities_in_index: if item['slug'] in seen_slugs: continue # 匹配逻辑：如果实体名在 query 中，或者 query 的词在实体名中 if item['original'].lower() in query_lower or item['slug'].replace('-', ' ') in query_lower: hit_entities.append(item) seen_slugs.add(item['slug']) if len(hit_entities) >= max_entities: break # 4. 提取实体内容并格式化输出 if not hit_entities: sys.exit(0) output_blocks = [] total_chars = 0 for entity in hit_entities: entity_path = os.path.join(entities_dir, f"{entity['slug']}.md") if os.path.exists(entity_path): try: with open(entity_path, 'r', encoding='utf-8') as f: content = f.read() block = f"### Entity: {entity['original']}\nURL: wiki/entities/{entity['slug']}.md\n\n{content}" if total_chars + len(block) > max_chars: break output_blocks.append(block) total_chars += len(block) except: continue if output_blocks: print("\n") print("以下是从您的私人百科全书（ARC Wiki）中自动检索到的关联知识条目，请结合这些背景信息回答用户：") print("\n---\n".join(output_blocks)) print("\n") if __name__ == "__main__": main() FILE:scripts/governance-audit.py import os import sys import subprocess import re import json def run(cmd): result = subprocess.run(cmd, shell=True, capture_output=True, text=True) return result.stdout.strip(), result.returncode def audit(): print("=== [ARC Reactor Governance Audit] ===") issues = [] # 1. Check Commit Signature last_commit_msg, _ = run("git log -1 --pretty=%B") if not re.search(r'$by \w+$', last_commit_msg): issues.append(f"Commit message missing signature '(by AgentName)'. Last msg: '{last_commit_msg}'") else: print("✅ Commit signature found.") # 2. Check RT Association for Modified Files # Check staged changes (between HEAD and Index) diff_files, _ = run("git diff --cached --name-only") files = diff_files.splitlines() code_files = [f for f in files if f.endswith(('.py', '.sh', '.yaml', '.json')) and not f.startswith('RT/')] if code_files: # Check if RT/ was also touched in this commit or recently rt_files = [f for f in files if f.startswith('RT/RT-')] if not rt_files: # Check recent RT hits (optional, simplified for now) issues.append(f"Code changes detected in {code_files} but no matching RT spec was found in this commit.") else: print(f"✅ RT association verified for {len(code_files)} files.") # 3. Check Wiki Integrity (No manual edits in arc-reactor-doc/wiki/ without record) # We check if wiki files were modified but archive-manager.py log wasn't updated wiki_files = [f for f in files if 'arc-reactor-doc/wiki/' in f and not f.endswith('log.md')] if wiki_files: log_updated = any('log.md' in f for f in files) if not log_updated: issues.append("Wiki files modified manually without updating the operational log.md. Use archive-manager.py!") else: print("✅ Wiki integrity check passed.") # 4. Final Verdict if issues: print("\n❌ Governance Violations Found:") for i in issues: print(f" - {i}") return 1 print("\n🌟 All systems compliant. Governance protocol followed.") return 0 if __name__ == "__main__": sys.exit(audit()) FILE:scripts/media-extractor.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ media-extractor.py - 视频/音频转写模块从 YouTube 等平台下载视频并转写为文本。支持的 ASR Provider: 1. 阿里云 NLS（优先） 2. 本地 mlx_whisper（降级） 3. 降级模式（仅抓取标题/字幕/描述）依赖: - yt-dlp: 视频下载 - ffmpeg: 音频格式转换 - requests: HTTP 请求 - mlx_whisper (可选): 本地转写用法: python3 scripts/media-extractor.py --url "https://youtu.be/xxx" --output /tmp/transcript.txt python3 scripts/media-extractor.py --url "https://youtu.be/xxx" --json python3 scripts/media-extractor.py --url "https://youtu.be/xxx" --provider aliyun-nls python3 scripts/media-extractor.py --check python3 scripts/media-extractor.py --url "https://youtu.be/xxx" --info-only """ import argparse import atexit import base64 import hashlib import hmac import json import os import re import shutil import subprocess import sys import tempfile import time import uuid from pathlib import Path from typing import Optional import requests # ============================================================================ # 全局变量 # ============================================================================ # 临时文件列表，用于退出时清理 _temp_files = [] _temp_dirs = [] def _cleanup(): """清理临时文件""" for f in _temp_files: try: if os.path.exists(f): os.remove(f) except Exception: pass for d in _temp_dirs: try: if os.path.exists(d): shutil.rmtree(d) except Exception: pass atexit.register(_cleanup) def _add_temp_file(path: str): """注册临时文件""" _temp_files.append(path) def _add_temp_dir(path: str): """注册临时目录""" _temp_dirs.append(path) # ============================================================================ # 工具函数 # ============================================================================ def get_available_memory_gb() -> float: """ 获取 macOS 可用内存（GB）使用 vm_stat 命令获取页面信息，计算可用内存。 """ try: result = subprocess.run( ["vm_stat"], capture_output=True, text=True, timeout=10 ) output = result.stdout # 解析 vm_stat 输出 # Pages free: xxxxxx. # Pages active: xxxxxx. free_match = re.search(r"Pages free:\s+(\d+)", output) active_match = re.search(r"Pages active:\s+(\d+)", output) inactive_match = re.search(r"Pages inactive:\s+(\d+)", output) speculative_match = re.search(r"Pages speculative:\s+(\d+)", output) # macOS 页面大小（ARM Mac 为 16384 字节） page_size = 16384 free_pages = int(free_match.group(1)) if free_match else 0 active_pages = int(active_match.group(1)) if active_match else 0 inactive_pages = int(inactive_match.group(1)) if inactive_match else 0 speculative_pages = int(speculative_match.group(1)) if speculative_match else 0 # 可用内存 = free + inactive + speculative（更准确的可用内存估算） available_pages = free_pages + inactive_pages + speculative_pages available_bytes = available_pages * page_size available_gb = available_bytes / (1024 ** 3) return round(available_gb, 2) except Exception: return 0.0 def check_command_exists(cmd: str) -> bool: """检查命令是否存在""" return shutil.which(cmd) is not None def format_duration(seconds: int) -> str: """将秒数格式化为 HH:MM:SS 格式""" hours = seconds // 3600 minutes = (seconds % 3600) // 60 secs = seconds % 60 if hours > 0: return f"{hours:02d}:{minutes:02d}:{secs:02d}" return f"{minutes:02d}:{secs:02d}" # ============================================================================ # ASR Provider 检测 # ============================================================================ def check_mlx_whisper() -> dict: """ 检测 mlx_whisper 是否可用 Returns: dict: {"available": bool, "reason": str, "model": str or None} """ result = { "available": False, "reason": "", "model": None, "memory_gb": 0.0 } # 检查是否已安装 mlx_whisper try: import mlx_whisper result["available"] = True except ImportError: result["reason"] = "mlx_whisper not installed" return result # 检查 Apple Silicon import platform if platform.system() != "Darwin" or platform.machine() != "arm64": result["available"] = False result["reason"] = "mlx_whisper requires Apple Silicon (arm64)" return result # 检查可用内存 memory_gb = get_available_memory_gb() result["memory_gb"] = memory_gb if memory_gb < 6: result["available"] = False result["reason"] = f"Insufficient memory ({memory_gb:.1f}GB < 6GB required)" return result # 根据内存选择模型 if memory_gb >= 10: result["model"] = "mlx-community/whisper-large-v3" elif memory_gb >= 6: result["model"] = "mlx-community/whisper-medium" else: result["available"] = False result["reason"] = f"Insufficient memory ({memory_gb:.1f}GB < 6GB required)" return result def check_aliyun_nls() -> dict: """ 检测阿里云 NLS 是否可用 Returns: dict: {"available": bool, "has_appkey": bool, "has_access_key": bool} """ result = { "available": False, "has_appkey": False, "has_access_key": False } appkey = os.environ.get("ALIYUN_NLS_APPKEY", "") access_key_id = os.environ.get("ALIYUN_ACCESS_KEY_ID", "") access_key_secret = os.environ.get("ALIYUN_ACCESS_KEY_SECRET", "") result["has_appkey"] = bool(appkey) result["has_access_key"] = bool(access_key_id and access_key_secret) result["available"] = result["has_appkey"] and result["has_access_key"] return result def check_all_providers() -> dict: """ 检查所有 ASR Provider 的可用性 Returns: dict: 包含各 provider 状态的字典 """ aliyun = check_aliyun_nls() mlx = check_mlx_whisper() # 推荐 provider recommended = None message = "" if aliyun["available"]: recommended = "aliyun-nls" message = "ASR 可用。将使用阿里云 NLS 进行语音转写。" elif mlx["available"]: recommended = "mlx-whisper" message = f"ASR 可用。将使用 mlx_whisper ({mlx['model']}) 进行语音转写。" else: message = "未检测到 ASR 配置。视频转写需要以下任一服务：" return { "aliyun_nls": aliyun, "mlx_whisper": mlx, "recommended": recommended, "message": message } def print_provider_guidance(): """打印 ASR 配置引导信息到 stderr""" guidance = """ ⚠️ 未检测到 ASR 配置。视频转写需要以下任一服务：方案 A（推荐）：阿里云 NLS（¥0.14/30分钟） - 注册阿里云 → 开通智能语音交互 → 获取 AppKey + AccessKey - 配置环境变量：ALIYUN_NLS_APPKEY, ALIYUN_ACCESS_KEY_ID, ALIYUN_ACCESS_KEY_SECRET 方案 B（免费）：本地 mlx_whisper（需 Apple Silicon + 5GB 内存） - pip install mlx-whisper 未配置 ASR 时，将尝试抓取视频标题、字幕、描述等可用信息。 """ print(guidance, file=sys.stderr) # ============================================================================ # 阿里云 NLS 相关函数 # ============================================================================ def get_nls_token(access_key_id: str, access_key_secret: str) -> str: """ 获取阿里云 NLS Token（使用 AccessKey 签名方式） Args: access_key_id: 阿里云 AccessKey ID access_key_secret: 阿里云 AccessKey Secret Returns: str: NLS Token ID Raises: Exception: 获取失败时抛出异常 """ params = { "AccessKeyId": access_key_id, "Action": "CreateToken", "Format": "JSON", "RegionId": "cn-shanghai", "SignatureMethod": "HMAC-SHA1", "SignatureNonce": str(uuid.uuid4()), "SignatureVersion": "1.0", "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "Version": "2019-02-28", } # 按字母顺序排序参数 sorted_params = sorted(params.items()) query_string = urllib_parse_urlencode(sorted_params) # 构建待签名字符串 string_to_sign = "GET&%2F&" + urllib_parse_quote(query_string, safe="") # 计算签名 signature = base64.b64encode( hmac.new( (access_key_secret + "&").encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha1 ).digest() ).decode("utf-8") params["Signature"] = signature # 构建请求 URL url = "https://nls-meta.cn-shanghai.aliyuncs.com/?" + urllib_parse_urlencode(params) # 发送请求 try: resp = requests.get(url, timeout=30) resp.raise_for_status() data = resp.json() if "Token" in data and "Id" in data["Token"]: return data["Token"]["Id"] else: raise Exception(f"获取 Token 失败: {data}") except requests.exceptions.RequestException as e: raise Exception(f"获取 NLS Token 网络请求失败: {e}") def urllib_parse_urlencode(params): """兼容的 url_encode 实现""" import urllib.parse return urllib.parse.urlencode(params) def urllib_parse_quote(s, safe=""): """兼容的 quote 实现""" import urllib.parse return urllib.parse.quote(s, safe=safe) def split_audio_chunks(wav_path: str, chunk_duration: int = 55) -> list: """ 将 WAV 音频文件按指定时长分割 Args: wav_path: WAV 文件路径 chunk_duration: 每段时长（秒），默认 55 秒 Returns: list: 每段音频文件的路径列表 """ # 获取音频总时长 result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", wav_path], capture_output=True, text=True, timeout=30 ) total_duration = float(result.stdout.strip()) chunks = [] temp_dir = tempfile.mkdtemp(prefix="arc-chunks-") _add_temp_dir(temp_dir) # 按 chunk_duration 分割 start = 0.0 chunk_idx = 0 while start < total_duration: end = min(start + chunk_duration, total_duration) chunk_path = os.path.join(temp_dir, f"chunk_{chunk_idx:04d}.wav") cmd = [ "ffmpeg", "-y", "-i", wav_path, "-ss", str(start), "-t", str(end - start), "-ar", "16000", "-ac", "1", "-acodec", "pcm_s16le", chunk_path ] subprocess.run( cmd, capture_output=True, text=True, timeout=60 ) chunks.append(chunk_path) start = end chunk_idx += 1 return chunks def transcribe_audio_chunk(token: str, app_key: str, audio_path: str) -> str: """ 转写单个音频片段（调用阿里云一句话识别 API） Args: token: NLS Token app_key: 阿里云 AppKey audio_path: 音频文件路径 Returns: str: 识别文本 Raises: Exception: API 调用失败时抛出异常 """ endpoint = "https://nls-gateway.cn-shanghai.aliyuncs.com/stream/v1/asr" headers = { "X-NLS-Token": token, "Content-Type": "application/octet-stream" } params = { "appkey": app_key, "format": "wav", "sample_rate": "16000", "enable_punctuation_prediction": "true", "enable_inverse_text_normalization": "true" } try: with open(audio_path, "rb") as f: audio_data = f.read() resp = requests.post( endpoint, headers=headers, params=params, data=audio_data, timeout=30 ) resp.raise_for_status() data = resp.json() if "result" in data: return data["result"] else: # 可能返回其他格式，尝试直接返回 return str(data.get("text", "")) except requests.exceptions.Timeout: raise Exception("NLS API 请求超时（30秒）") except requests.exceptions.RequestException as e: raise Exception(f"NLS API 请求失败: {e}") def transcribe_aliyun_nls(wav_path: str) -> str: """ 使用阿里云 NLS 转写 WAV 音频 Args: wav_path: WAV 文件路径（16kHz mono） Returns: str: 完整转写文本 """ access_key_id = os.environ.get("ALIYUN_ACCESS_KEY_ID") access_key_secret = os.environ.get("ALIYUN_ACCESS_KEY_SECRET") app_key = os.environ.get("ALIYUN_NLS_APPKEY") if not all([access_key_id, access_key_secret, app_key]): raise Exception("缺少阿里云 NLS 环境变量") # 1. 获取 Token print("正在获取阿里云 NLS Token...", file=sys.stderr) token = get_nls_token(access_key_id, access_key_secret) # 2. 分割音频 print("正在分割音频（每段 55 秒）...", file=sys.stderr) chunks = split_audio_chunks(wav_path, chunk_duration=55) print(f"共分割为 {len(chunks)} 个片段", file=sys.stderr) # 3. 逐段转写 results = [] for i, chunk in enumerate(chunks): print(f"\r转写中: {i + 1}/{len(chunks)}", end="", file=sys.stderr) sys.stderr.flush() try: text = transcribe_audio_chunk(token, app_key, chunk) if text: results.append(text) except Exception as e: print(f"\n片段 {i + 1} 转写失败: {e}", file=sys.stderr) # 继续处理下一段 print("", file=sys.stderr) # 换行 # 4. 拼接结果 full_text = " ".join(results) return full_text # ============================================================================ # mlx_whisper 转写 # ============================================================================ def transcribe_mlx_whisper(wav_path: str) -> str: """ 使用 mlx_whisper 转写音频 Args: wav_path: WAV 文件路径 Returns: str: 转写文本 """ import mlx_whisper memory_gb = get_available_memory_gb() if memory_gb >= 10: model_path = "mlx-community/whisper-large-v3" elif memory_gb >= 6: model_path = "mlx-community/whisper-medium" else: raise Exception(f"内存不足（{memory_gb:.1f}GB），无法使用 mlx_whisper") print(f"正在使用 mlx_whisper ({model_path}) 转写...", file=sys.stderr) result = mlx_whisper.transcribe( wav_path, language="zh", path_or_hf_repo=model_path ) return result.get("text", "") # ============================================================================ # 视频下载与信息获取 # ============================================================================ def get_video_info(url: str) -> dict: """ 获取视频信息（标题、描述、时长、字幕等） Args: url: 视频 URL Returns: dict: 视频信息 """ temp_dir = tempfile.mkdtemp(prefix="arc-info-") _add_temp_dir(temp_dir) info_file = os.path.join(temp_dir, "info.json") # 使用 yt-dlp 命令行工具（独立安装） ytdlp_cmd = shutil.which("yt-dlp") if not ytdlp_cmd: ytdlp_cmd = "yt-dlp" # fallback to PATH cmd = [ ytdlp_cmd, "--dump-json", "--no-download", "--no-warnings", url ] try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=60 ) if result.returncode != 0: print(f"获取视频信息失败: {result.stderr}", file=sys.stderr) return {} # 尝试解析输出 try: info = json.loads(result.stdout.strip().split("\n")[-1]) except json.JSONDecodeError: # 可能输出包含多行，尝试最后一行 lines = result.stdout.strip().split("\n") for line in reversed(lines): if line.strip().startswith("{"): try: info = json.loads(line) break except json.JSONDecodeError: continue else: return {} # 提取自动字幕（优先中文） subtitles = {} if "subtitles" in info: subtitles = info.get("subtitles", {}) elif "automatic_captions" in info: subtitles = info.get("automatic_captions", {}) auto_subtitle = None # 优先找中文相关字幕 for lang in ["zh-Hans", "zh-CN", "zh", "zh-TW", "en"]: if lang in subtitles and subtitles[lang]: # 获取字幕内容 sub_url = subtitles[lang][0].get("url") if sub_url: try: sub_resp = requests.get(sub_url, timeout=30) sub_resp.raise_for_status() auto_subtitle = sub_resp.text break except Exception: continue return { "title": info.get("title", ""), "description": info.get("description", ""), "duration": info.get("duration", 0), "url": info.get("webpage_url", url), "thumbnail": info.get("thumbnail", ""), "subtitles": auto_subtitle, "full_info": info } except subprocess.TimeoutExpired: print("获取视频信息超时（60秒）", file=sys.stderr) return {} except Exception as e: print(f"获取视频信息异常: {e}", file=sys.stderr) return {} def download_audio(url: str) -> tuple: """ 下载视频音频并转换为 16kHz mono WAV Args: url: 视频 URL Returns: tuple: (wav_path, video_info) Raises: Exception: 下载或转换失败时抛出异常 """ # 创建临时目录 temp_dir = tempfile.mkdtemp(prefix="arc-audio-") _add_temp_dir(temp_dir) # 下载音频为 WAV raw_wav = os.path.join(temp_dir, "raw_audio.wav") _add_temp_file(raw_wav) print("正在下载音频...", file=sys.stderr) # 步骤 1: 用 yt-dlp 下载为 wav ytdlp_cmd = shutil.which("yt-dlp") if not ytdlp_cmd: ytdlp_cmd = "yt-dlp" cmd1 = [ ytdlp_cmd, "-x", "--audio-format", "wav", "--audio-quality", "0", "-o", raw_wav, "--no-warnings", url ] result = subprocess.run( cmd1, capture_output=True, text=True, timeout=120 # 下载超时 2 分钟 ) if result.returncode != 0: raise Exception(f"音频下载失败: {result.stderr}") # yt-dlp 可能输出文件为 .wav.wav 或其他扩展名，查找实际文件 actual_files = list(Path(temp_dir).glob("*.wav")) if actual_files: raw_wav = str(actual_files[0]) elif not os.path.exists(raw_wav): raise Exception(f"音频文件未找到: {raw_wav}") # 步骤 2: 用 ffmpeg 转换为 16kHz mono WAV final_wav = os.path.join(temp_dir, "audio_16k.wav") _add_temp_file(final_wav) print("正在转换音频格式...", file=sys.stderr) cmd2 = [ "ffmpeg", "-y", "-i", raw_wav, "-ar", "16000", "-ac", "1", "-acodec", "pcm_s16le", final_wav ] result2 = subprocess.run( cmd2, capture_output=True, text=True, timeout=120 # 转换超时 2 分钟 ) if result2.returncode != 0: raise Exception(f"音频格式转换失败: {result2.stderr}") # 获取视频信息 video_info = get_video_info(url) return final_wav, video_info # ============================================================================ # 主转写函数 # ============================================================================ def transcribe(url: str, provider: str = "auto") -> dict: """ 转写视频音频 Args: url: 视频 URL provider: ASR Provider ("auto", "aliyun-nls", "mlx-whisper", "fallback") Returns: dict: 转写结果 """ result = { "status": "pending", "provider": provider, "source_url": url, "title": "", "duration_seconds": 0, "transcript": "", "word_count": 0, "cost_estimate": "", "error": "" } # 检查 provider 可用性 if provider == "auto": providers = check_all_providers() if providers["recommended"]: provider = providers["recommended"] else: print_provider_guidance() provider = "fallback" # 下载音频 try: wav_path, video_info = download_audio(url) result["title"] = video_info.get("title", "") result["duration_seconds"] = video_info.get("duration", 0) except Exception as e: result["status"] = "error" result["error"] = f"音频下载失败: {e}" return result # 根据 provider 转写 try: if provider == "aliyun-nls": transcript = transcribe_aliyun_nls(wav_path) result["provider"] = "aliyun-nls" # 估算费用：约 ¥0.14/30分钟 duration_min = result["duration_seconds"] / 60 cost = 0.14 * (duration_min / 30) result["cost_estimate"] = f"{cost:.2f} CNY" elif provider == "mlx-whisper": transcript = transcribe_mlx_whisper(wav_path) result["provider"] = "mlx-whisper" result["cost_estimate"] = "0.00 CNY (本地)" else: # 降级模式：尝试使用字幕或描述 transcript = "" # 尝试使用自动字幕 if video_info.get("subtitles"): # 解析 srt/vtt 字幕为纯文本 sub_text = video_info["subtitles"] # VTT 格式处理 if sub_text.startswith("WEBVTT"): lines = sub_text.split("\n") for line in lines: # 跳过时间码和标签 if "-->" in line or line.strip().startswith("-"): continue if line.strip() and not line.strip().startswith(">"): transcript += " " + line.strip() else: transcript = sub_text transcript = re.sub(r"<[^>]+>", "", transcript) transcript = re.sub(r"\s+", " ", transcript).strip() # 如果没有字幕，使用描述 if not transcript and video_info.get("description"): transcript = video_info["description"] # 截取前 5000 字符 transcript = transcript[:5000] result["provider"] = "fallback" result["cost_estimate"] = "0.00 CNY (字幕/描述)" result["transcript"] = transcript result["word_count"] = len(transcript.replace(" ", "")) result["status"] = "success" except Exception as e: result["status"] = "error" result["error"] = str(e) return result # ============================================================================ # CLI 接口 # ============================================================================ def cmd_check(): """检查 ASR Provider 可用性""" result = check_all_providers() # 如果没有可用 provider，打印引导 if not result["recommended"]: print_provider_guidance() # 输出 JSON print(json.dumps(result, ensure_ascii=False, indent=2)) def cmd_info(url: str): """获取视频信息（不转写）""" info = get_video_info(url) if not info: print(json.dumps({ "status": "error", "error": "无法获取视频信息" }, ensure_ascii=False, indent=2)) sys.exit(1) # 简化输出 output = { "status": "success", "title": info.get("title", ""), "description": info.get("description", ""), "duration_seconds": info.get("duration", 0), "duration_formatted": format_duration(info.get("duration", 0)), "url": info.get("url", url), "has_subtitles": bool(info.get("subtitles")) } print(json.dumps(output, ensure_ascii=False, indent=2)) def cmd_transcribe(url: str, output_file: Optional[str], json_output: bool, provider: str): """转写视频""" result = transcribe(url, provider) if json_output: print(json.dumps(result, ensure_ascii=False, indent=2)) else: # 纯文本输出 if result["status"] == "success": if result["title"]: print(f"# {result['title']}") print() print(result["transcript"]) else: print(f"转写失败: {result.get('error', '未知错误')}", file=sys.stderr) sys.exit(1) # 写入文件 if output_file: with open(output_file, "w", encoding="utf-8") as f: f.write(result["transcript"]) print(f"转写结果已保存到: {output_file}", file=sys.stderr) def main(): parser = argparse.ArgumentParser( description="视频/音频转写工具 - 从 YouTube 等平台下载视频并转写为文本", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: python3 scripts/media-extractor.py --check python3 scripts/media-extractor.py --url "https://youtu.be/xxx" --json python3 scripts/media-extractor.py --url "https://youtu.be/xxx" --output /tmp/transcript.txt python3 scripts/media-extractor.py --url "https://youtu.be/xxx" --provider aliyun-nls python3 scripts/media-extractor.py --url "https://youtu.be/xxx" --info-only """ ) parser.add_argument( "--url", type=str, help="视频 URL（支持 YouTube 等平台）" ) parser.add_argument( "--output", "-o", type=str, help="输出文件路径（转写文本）" ) parser.add_argument( "--json", "-j", action="store_true", help="输出 JSON 格式（含元数据）" ) parser.add_argument( "--provider", "-p", type=str, default="auto", choices=["auto", "aliyun-nls", "mlx-whisper", "fallback"], help="ASR Provider（默认 auto）" ) parser.add_argument( "--check", "-c", action="store_true", help="检查 ASR Provider 可用性" ) parser.add_argument( "--info-only", "-i", action="store_true", help="仅获取视频信息（不转写）" ) args = parser.parse_args() # 检查命令依赖 if not check_command_exists("ffmpeg"): print("错误: 未找到 ffmpeg，请先安装: brew install ffmpeg", file=sys.stderr) sys.exit(1) if not check_command_exists("yt-dlp"): print("错误: 未找到 yt-dlp，请先安装: pip install yt-dlp", file=sys.stderr) sys.exit(1) # 执行对应命令 if args.check: cmd_check() elif args.info_only: if not args.url: parser.error("--info-only 需要指定 --url") cmd_info(args.url) elif args.url: cmd_transcribe(args.url, args.output, args.json, args.provider) else: parser.print_help() sys.exit(1) if __name__ == "__main__": main() FILE:scripts/smart_fetcher.py #!/usr/bin/env python3 import os import sys import json import requests import re from datetime import datetime # 配置：高反爬域名黑名单，这些域名强制走专业提取接口 HIGH_ANTIBOT_DOMAINS = [ 'toutiao.com', 'juejin.cn', 'mp.weixin.qq.com', 'zhihu.com', 'weibo.com' ] def is_high_antibot(url): """检测 URL 是否属于高倾斜反爬域名""" return any(domain in url for domain in HIGH_ANTIBOT_DOMAINS) def fetch_via_tavily(url, api_key): """使用 Tavily Extract API 提取正文 (目前最稳的方法)""" if not api_key: return None print(f"[INFO] 正在调用 Tavily 专业级提取引擎: {url}") try: # Tavily Extract API 结构 res = requests.post( "https://api.tavily.com/extract", json={ "api_key": api_key, "urls": [url] }, timeout=30 ) data = res.json() if data.get('results') and len(data['results']) > 0: result = data['results'][0] # Tavily 返回的是对象，包含 raw_content 或 markdown content = result.get('raw_content') or result.get('markdown') return content except Exception as e: print(f"[WARNING] Tavily 提取失败: {e}") return None def fetch_via_jina(url): """使用 Jina.ai Reader 代理 (免费方案)""" print(f"[INFO] 正在尝试 Jina Reader 轻量提取: {url}") try: reader_url = f"https://r.jina.ai/{url}" headers = { "Accept": "text/markdown", "X-Target-Selector": "article, .content, .main" # 提示提取范围 } res = requests.get(reader_url, headers=headers, timeout=20) if res.status_code == 200 and len(res.text) > 200: return res.text except Exception as e: print(f"[WARNING] Jina 提取失败: {e}") return None def fetch_via_llm_reader(url): """备选的 LLM-Reader 接口""" print(f"[INFO] 正在尝试 LLM-Reader 节点提取: {url}") try: # 这是一个针对字节跳动优化过的公共接口镜像 reader_url = f"https://reader.llm.report/api/read?url={url}" res = requests.get(reader_url, timeout=20) data = res.json() return data.get('markdown', '') except: return None def fetch_basic(url): """最基础的多头 Requests 抓取 (针对 80% 的普通网站)""" print(f"[INFO] 正在执行标准请求协议: {url}") headers = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8" } try: res = requests.get(url, headers=headers, timeout=15) res.encoding = res.apparent_encoding # 这里返回原始内容，后面可以接 Readability 提取 return res.text except Exception as e: print(f"[ERROR] 标准抓取彻底瘫痪: {e}") return None def smart_extract(url): """全自动智能路由提取逻辑""" # 1. 检测域名属性 is_hard = is_high_antibot(url) # 2. 获取环境变量中的密钥 tavily_key = os.environ.get("TAVILY_API_KEY") apify_token = os.environ.get("APIFY_TOKEN") content = None # 3. 路由策略： # 如果是硬骨头网站，优先走 Tavily if is_hard: if tavily_key: content = fetch_via_tavily(url, tavily_key) if not content: content = fetch_via_llm_reader(url) if not content: content = fetch_via_jina(url) else: # 普通网站先试免费的 Jina 或 LLM-Reader，效果最好 content = fetch_via_jina(url) if not content: content = fetch_via_llm_reader(url) if not content: content = fetch_basic(url) if not content or len(content.strip()) < 50: print(f"[FATAL] 所有的抓取链路均被拦截或无法提取正文。") return None return content if __name__ == "__main__": if len(sys.argv) < 2: print("Usage: python3 smart_fetcher.py ") sys.exit(1) target_url = sys.argv[1] result = smart_extract(target_url) if result: print("\n--- EXTRACTED CONTENT START ---") print(result) print("--- EXTRACTED CONTENT END ---\n") else: sys.exit(1) FILE:scripts/weekly-reporter.py #!/usr/bin/env python3 import os import sys import argparse import json import re from datetime import datetime, timedelta def find_doc_root(start_path, root_name='arc-reactor-doc'): """向上查找工作区根目录""" curr_dir = os.path.abspath(start_path) while curr_dir and curr_dir != '/': if os.path.exists(os.path.join(curr_dir, root_name)): return os.path.join(curr_dir, root_name) if os.path.exists(os.path.join(curr_dir, '.git')): return os.path.join(curr_dir, root_name) parent = os.path.dirname(curr_dir) if parent == curr_dir: break curr_dir = parent return os.path.join(start_path, root_name) def _parse_frontmatter(content): """提取 Markdown 的 YAML Frontmatter""" meta = {} if content.strip().startswith('---'): parts = content.split('---', 2) if len(parts) >= 3: header = parts[1] for line in header.splitlines(): if ':' in line: k, v = line.split(':', 1) meta[k.strip().lower()] = v.strip().strip('"').strip("'") return meta def main(): parser = argparse.ArgumentParser(description='ARC Reactor Weekly Reporter') parser.add_argument('--days', type=int, default=7, help='聚合过去几天的内容') parser.add_argument('--root', default='arc-reactor-doc', help='文档根目录名称') args = parser.parse_args() cwd = os.getcwd() doc_root = find_doc_root(cwd, args.root) sources_dir = os.path.join(doc_root, 'wiki', 'sources') if not os.path.exists(sources_dir): print(json.dumps({"status": "error", "message": f"Sources directory not found: {sources_dir}"})) sys.exit(1) # 1. 计算时间窗口 end_date = datetime.now() start_date = end_date - timedelta(days=args.days) target_dates = [] for i in range(args.days + 1): target_dates.append((start_date + timedelta(days=i)).strftime('%Y-%m-%d')) # 2. 搜集符合时间窗的文件 aggregated_items = [] for date_str in target_dates: date_path = os.path.join(sources_dir, date_str) if os.path.exists(date_path): for filename in os.listdir(date_path): if filename.endswith('.md'): fpath = os.path.join(date_path, filename) try: with open(fpath, 'r', encoding='utf-8') as f: content = f.read() meta = _parse_frontmatter(content) # 提取首段摘要 (跳过 Frontmatter) body = content.split('---', 2)[-1].strip() if '---' in content else content summary = body.split('\n\n')[0][:300].strip() aggregated_items.append({ "title": meta.get('title', filename), "date": date_str, "tags": meta.get('tags', []), "summary": summary, "path": os.path.relpath(fpath, doc_root) }) except: continue # 3. 输出汇总 if not aggregated_items: print(f"### ARC Reactor 周报 ({start_date.strftime('%Y-%m-%d')} ~ {end_date.strftime('%Y-%m-%d')})\n\n> 本周暂无新增归档内容，休息一下吧！☕️") sys.exit(0) # 格式化输出 (供 LLM 进一步处理或直接展示) print(f"# ARC Reactor 知识周报\n区间：{start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}\n") print(f"## 📚 本周收录概览 (共 {len(aggregated_items)} 篇)\n") for item in aggregated_items: print(f"- **{item['title']}** ({item['date']})") print(f" > {item['summary']}...") print(f" [查阅原文]({item['path']})\n") print("\n---\n## 🧠 自动生成的洞察提炼 (Prompt 建议)\n") print("Orchestrator, 请根据以上聚合内容，回答：") print("1. 本周大家主要在研究什么？有没有形成明显的知识集群？") print("2. 哪些词条在不同文章里被反复提及？推荐我下一步该去深入了解哪个实体？") # 返回 JSON 回执以便 Orchestrator 审计 receipt = { "status": "success", "action": "weekly_report", "days_scanned": args.days, "items_found": len(aggregated_items), "start_date": start_date.strftime('%Y-%m-%d'), "end_date": end_date.strftime('%Y-%m-%d') } # 在 stderr 打印 JSON 回执，避免干扰正文 print(json.dumps(receipt, ensure_ascii=False), file=sys.stderr) if __name__ == "__main__": main() FILE:test_obsidian_sync.py #!/usr/bin/env python3 """Simple test for Obsidian sync functionality.""" import sys import os # Add scripts dir to path sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import importlib.util spec = importlib.util.spec_from_file_location("archive_manager", "scripts/archive-manager.py") am = importlib.util.module_from_spec(spec) spec.loader.exec_module(am) validate_obsidian_config = am.validate_obsidian_config sync_to_obsidian = am.sync_to_obsidian slugify = am.slugify def test_validate_obsidian_config(): """Test validate_obsidian_config with non-existent path.""" is_valid, msg = validate_obsidian_config("/nonexistent/path") assert is_valid == False, f"Expected False for non-existent path, got {is_valid}" assert "不存在" in msg, f"Expected '不存在' in message, got {msg}" print("✓ validate_obsidian_config() correctly rejects non-existent path") def test_sync_to_obsidian_missing_source(): """Test sync_to_obsidian with non-existent source file.""" result = sync_to_obsidian("/nonexistent/file.md", "/tmp", "test/") assert result["status"] == "error", f"Expected error status, got {result['status']}" assert "源文件不存在" in result["error"], f"Expected '源文件不存在', got {result['error']}" print("✓ sync_to_obsidian() correctly handles missing source file") def test_slugify(): """Test slugify function.""" assert slugify("Claude Code") == "claude-code" assert slugify("SWE-bench") == "swe-bench" print("✓ slugify() works correctly") def test_sync_to_obsidian_invalid_vault(): """Test sync_to_obsidian with invalid vault path.""" result = sync_to_obsidian("/tmp/test.md", "/nonexistent/vault", "test/") assert result["status"] == "error", f"Expected error status, got {result['status']}" assert result["action"] == "obsidian_sync" print("✓ sync_to_obsidian() correctly handles invalid vault") if __name__ == "__main__": test_slugify() test_validate_obsidian_config() test_sync_to_obsidian_missing_source() test_sync_to_obsidian_invalid_vault() print("\n✅ All tests passed!")

ARC Reactor

Comments (0)