From 3e3c6a24d291b9f47b97d41e5e11cbbc70d658da Mon Sep 17 00:00:00 2001 From: root Date: Fri, 3 Jul 2026 08:04:39 +0800 Subject: [PATCH] v4 architecture: pipeline stages, SDK, event bus, LLM-driven memory consolidation - SDK PluginAPI (internal/plugin/sdk/): RegisterTool/RegisterStage/Subscribe/Publish - EventBus (internal/events/): system-level pub/sub with wildcard support - StageHost (internal/agent/core/stages.go): 7-stage message pipeline - Agent core: on_input/pre_action/post_action/before_toolcall/after_toolcall/before_output/after_output - Plugin Registry: SDK plugin registration and tool routing - GraphDB.MergeEntities: entity consolidation with relation redirection - memory_merge tool: allows LLM to merge similar entities - Consolidation task: heartbeat detects conflicts, enqueues via IO for LLM decision - _consolidation_ internal channel for system-level memory maintenance - Comprehensive documentation: ARCHITECTURE.md, PLAN.md, DESIGN.md, README.md - 54 tests across all packages, all passing --- DESIGN.md | 247 +------ PLAN.md | 132 ++++ README.md | 120 ++-- cmd/homed/main.go | 44 +- docs/ARCHITECTURE.md | 726 +++++++++++++------- internal/agent/core/agent.go | 472 +++++++++++-- internal/agent/core/agent_functions_test.go | 18 +- internal/agent/core/context.go | 82 ++- internal/agent/core/context_test.go | 12 +- internal/agent/core/stages.go | 74 ++ internal/agent/core/stages_test.go | 183 +++++ internal/api/plugin.go | 68 ++ internal/events/bus.go | 69 ++ internal/events/bus_test.go | 102 +++ internal/memory/graph.go | 92 +++ internal/memory/graph_test.go | 89 +++ internal/plugin/plugin.go | 76 ++ internal/plugin/sdk/api.go | 169 +++++ internal/plugin/sdk/bus.go | 42 ++ internal/plugin/sdk/bus_test.go | 133 ++++ 20 files changed, 2318 insertions(+), 632 deletions(-) create mode 100644 PLAN.md create mode 100644 internal/agent/core/stages.go create mode 100644 internal/agent/core/stages_test.go create mode 100644 internal/api/plugin.go create mode 100644 internal/events/bus.go create mode 100644 internal/events/bus_test.go create mode 100644 internal/plugin/sdk/api.go create mode 100644 internal/plugin/sdk/bus.go create mode 100644 internal/plugin/sdk/bus_test.go diff --git a/DESIGN.md b/DESIGN.md index ed4ee25..c177033 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -1,240 +1,19 @@ -# HomeAgent 架构设计 v3 +# HomeAgent 架构设计 v4 -## 一、核心理念 +完整架构文档参见 [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md)。 -24 小时陪伴用户、随时待命的智能管家。**单会话·单 Agent**，身份不漂移。 +## 核心原则 -### 设计原则 -- **所有输入走 IO 抽象层（中断模式）**，不直调 agent 方法 -- **所有 LLM 调用走 Provider 接口**，不直连 API -- **DeepSeek v4 flash** 为默认 LLM，thinking 模式关闭 -- **人格固定**（personal.md），记忆分层管理防止性格突变 -- **知识独立于记忆**，agent 主动学习 -- **插件 = 容器**，IO 通道是插件的内嵌组件 +- **核心零 IO** — 无任何硬编码 IO 能力，所有 IO 来自插件 +- **输出是工具调用** — Agent 必须显式 `output_send` 才能通信 +- **三通道插件** — 工具 (RegisterTool)、阶段 (RegisterStage)、事件 (Subscribe/Publish) +- **阶段管道** — 7 个 hook 点让插件干预消息处理流：`on_input` → `pre_action` → `post_action` ↔ `before_toolcall`/`after_toolcall` → `before_output` → `after_output` +- **三层记忆** — Context (内存) → Document (JSON+向量) → Graph (SQLite) +- **知识独立** — 独立 TF-IDF 向量索引，不与记忆耦合 ---- - -## 二、核心抽象 - -### IO 抽象层（唯一输入路径） +## 快速启动 +```bash +make build # 编译 +make run # 编译并启动（数据 /tmp/homeagent） ``` -外部设备 (Mic/Camera/OneBot/GPIO/HTTP) - │ - ▼ -IOManager - ├─ Device 接口（每个设备实现） - │ ├─ Name() / Type() / Description() - │ ├─ Tools() → 给 LLM 的 Function Calling 工具 - │ ├─ Execute() → 工具调用分发 - │ ├─ Start() / Stop() - │ └─ OutputCapabilities() → 输出能力声明 - ├─ InputEvent{Source, Type, Payload} → inputCh - ├─ OutputEvent{Target, Type, Payload} → outputCh - └─ 路由表: 输入源 → 默认输出通道 -``` - -**输出通道能力声明**（`OutputCapability` 位掩码）： -| 能力 | 说明 | -|------|------| -| CapText | 文本输出 | -| CapFile | 文件传输 | -| CapImage | 图片输出 | -| CapAudio | 音频输出 | -| CapStructured | 结构化数据（JSON/卡片） | - -**Agent 可调用通道工具**： -- `output_list_channels` — 查看所有可用通道及其能力 -- `output_set_channel` — 切换当前回复的输出通道 -- `output_send` — 通过指定通道异步发送消息（校验能力） - -### API 抽象层（唯一输出路径） - -``` -Agent Core → Provider.Chat() → LLM API - │ - ┌───────┴───────┐ - ▼ ▼ - OpenAIProvider LuaAdapter - (DeepSeek API) (格式转换) -``` - -### 插件系统（OpenClaw 兼容） - -``` -Plugin（容器） - ├─ 元数据: name, version, author, description - ├─ IOConfig（可选）: 声明 IO 端口 - ├─ Device（可选）: 原生 Go IO 设备实现 - └─ Tools: 给 LLM 的工具定义 -``` - -插件来源： -- `plugins/` 目录热加载，agent 通过 `plgreload` 工具显式控制重载 -- 有原生工厂注册的（如 QQ/OneBot）→ 构造原生设备 -- 纯 SKILL.md → 用 PluginDevice 包装为 IO 设备 -- 原子化替换：新设备 Start → 原子换路由 → 旧设备 Stop - ---- - -## 三、记忆体系（三层） - -``` -用户输入 → Context（内存, 30条, TF-IDF排序） - │ 每次响应后裁剪最不相关的 - ▼ - Document（JSON + TF-IDF向量, 冷72h→图） - │ 心跳蒸馏（30min） - ▼ - Graph（SQLite三元组, 定期重整+同义合并） -``` - -| 层 | 存储 | 容量 | 裁剪策略 | -|---|---|---|---| -| Layer 1: 上下文 | `RelevanceContext`（内存环形缓冲） | 30 条 | TF-IDF 余弦相似度排序，低分→文档 | -| Layer 2: 文档 | `document.Store`（JSON 文件 + 向量索引） | 无上限 | 72h 未访问 + ≤2 次命中→图 | -| Layer 3: 图 | `memory.GraphDB`（SQLite 三元组） | 无上限 | 定期重整 + bigram Jaccard 同义合并 | - -**注入策略**： -- 只注入**图索引**（实体名+类型+提及数）到 prompt，不注入全文 -- agent 通过 `memory_recall` 主动查询详情 -- 文档摘要按相关性注入前 3 条 - ---- - -## 四、知识体系 - -独立于记忆，agent 主动学习。 - -``` -knowledge/ - smart_home/content.md - cooking/content.md - │ - ▼ - TF-IDF 向量索引（字符 bigram） - │ - ▼ - knowledge_search / knowledge_create / knowledge_list -``` - ---- - -## 五、人格内核 - -```personal.md``` → 加载一次 → 固定在 system prompt 最前 → 永不漂移。 - ---- - -## 六、OneBot QQ 通道 - -``` -go-cqhttp / Lagrange（OneBot 前端） - │ Reverse WebSocket - ▼ -OneBot Client（internal/onebot/） - ├─ 事件循环 → Event → IO InputEvent - ├─ Action 调用 → send_private_msg / send_group_msg / ... - └─ 自动重连 + 心跳检测 - │ - ▼ -Device 注册 → IOManager → Agent -``` - ---- - -## 七、数据流 - -``` -用户消息 → IO InputEvent → eventLoop - │ - ├─ 1. 追加到 RelevanceContext - ├─ 2. 构建 prompt: - │ personal.md + 图索引 + 文档摘要 + 上下文 + 工具 - ├─ 3. 工具循环（最多 10 轮） - │ LLM → tool_calls → Execute → 结果 → LLM → ... - ├─ 4. 追加响应到上下文 - ├─ 5. 相关性裁剪（Prune → 不相关的归档到文档） - ├─ 6. OutputEvent → 输出通道 - └─ 7. 记忆候选 → TextMemory(JSONL) + Distiller → GraphDB - -心跳（30min）: - ├─ Indexer.Sync() — 图→向量 - ├─ DocStore.Reindex() — 文档向量重建 - ├─ 冷文档→图归化 - └─ 图同义合并 - -插件重载（plgreload）: - ├─ 扫描 plugins/ 目录 - ├─ 加载新插件，Start 新设备 - ├─ 原子替换 IOManager 设备表 + 路由表 - └─ Stop 旧设备 -``` - ---- - -## 八、配置 - -```yaml -daemon: - listen_addr: ":8080" - data_dir: "/var/lib/homeagent" - heartbeat_interval: 15s -llm: - model: "deepseek-v4-flash" - base_url: "https://api.deepseek.com/v1" - api_key: "${DEEPSEEK_API_KEY}" - temperature: 0.7 - max_tokens: 4096 -``` - ---- - -## 九、关键文件 - -``` -cmd/homed/main.go — 入口：组装所有子系统 -internal/agent/core/agent.go — Agent 核心：事件循环、工具循环、心跳 -internal/agent/core/context.go — RelevanceContext：TF-IDF 上下文管理 -internal/agent/personal.go — 人格加载 -internal/agent/api/provider.go — Provider 接口 + DeepSeek 实现 -internal/agent/io/channel.go — IO 抽象层：Device/InputEvent/OutputEvent/路由 -internal/api/handler.go — HTTP API 端点 -internal/knowledge/knowledge.go — 知识系统 -internal/memory/graph.go — SQLite 图数据库 -internal/memory/indexer.go — 图索引器 -internal/memory/vector/store.go — TF-IDF 向量存储 -internal/memory/document/doc.go — 文档记忆 -internal/memory/text/text.go — 文本记忆（JSONL） -internal/memory/pipeline/ — 蒸馏器 -internal/onebot/ — OneBot V11 协议实现 -internal/plugin/plugin.go — 插件系统 -internal/tracker/tracker.go — 变更追踪 -internal/supervisor/daemon.go — 守护进程 -internal/lua/vm.go — Lua 适配器 VM -plugins/ — 插件目录 - qq/SKILL.md — QQ 插件 SKILL.md - qq/skill.json — QQ 插件 JSON 元数据 -config/config.go — 配置加载 -pkg/types/ — 类型定义 -DESIGN.md — 本架构文档 -``` - ---- - -## 十、与旧设计的核心区别 - -| 维度 | 旧设计 (v1) | 当前设计 (v3) | -|------|-------------|---------------| -| 上下文裁剪 | 固定 FIFO 20 条 | TF-IDF 相关性排序 top-K | -| 图记忆注入 | 关键词 LIKE 查询 | 向量搜索实体名，只注索引 | -| 文档→图 | 无 | 72h 冷文档自动归化 | -| 图重整 | 无 | 向量同步 + bigram Jaccard 同义合并 | -| 知识系统 | 无 | `knowledge/` 目录 + TF-IDF 向量 | -| 人格 | 无 | `personal.md` 固定注入 | -| 向量引擎 | 无 | 自研 TF-IDF + 倒排索引（字符 bigram） | -| 部署 | Docker 容器 + 快照 | 单二进制 + overlayfs 追踪 | -| Agent 模型 | 多 Agent 编排 | 单 Agent + 工具循环 | -| 输出通道 | 无 | 能力声明 + 路由 + 校验 | -| 插件系统 | 无 | OpenClaw SKILL.md + 原生工厂 | -| QQ 通道 | 无 | OneBot V11 Reverse WS | diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 0000000..7ea3eec --- /dev/null +++ b/PLAN.md @@ -0,0 +1,132 @@ +# HomeAgent 实施计划 + +## 已完成 + +### Phase 0 — 核心基础设施 ✅ + +| 任务 | 文件 | 状态 | +|------|------|------| +| SDK 接口定义 | `internal/plugin/sdk/api.go` | ✅ | +| PluginAPI（RegisterTool/RegisterStage/Subscribe/Publish） | `internal/plugin/sdk/api.go` | ✅ | +| 插件内部 EventBus | `internal/plugin/sdk/bus.go` | ✅ | +| 系统 EventBus | `internal/events/bus.go` | ✅ | +| StageHost 编排器 | `internal/agent/core/stages.go` | ✅ | +| Agent 阶段注入（7 个 hook 点） | `internal/agent/core/agent.go` | ✅ | +| 插件注册表 SDK 支持 | `internal/plugin/plugin.go` | ✅ | +| main.go 接入 EventBus + StageHost | `cmd/homed/main.go` | ✅ | +| 架构文档 v4 | `docs/ARCHITECTURE.md` | ✅ | + +--- + +## 待实施 + +### Phase 1 — 插件 SDK 迁移（当前） + +| # | 任务 | 说明 | 优先级 | +|---|------|------|--------| +| 1.1 | SDK 添加 `ToolDef` 参数描述支持 | `RegisterTool` 接受 `ToolDef` 结构体（含 parameters）而非纯 handler | high | +| 1.2 | StageHost 收集完整 ToolDef | 目前只传 name，需传完整 description + parameters 给 LLM | high | +| 1.3 | Registry.AddPluginAPI 自动构建 StageHost | 替代手动 `syncFromRegistry` | high | +| 1.4 | 添加 `before_toolcall` deny 机制的测试 | 确保 `StageContext.Response` 在工具级别生效 | medium | +| 1.5 | 添加 `on_input` 改写消息的测试 | `stageCtx.RawMessage` 在阶段后被正确使用 | medium | + +### Phase 2 — 迁移 WebUI 到 SDK 模式 + +| # | 任务 | 说明 | 优先级 | +|---|------|------|--------| +| 2.1 | WebUI 改为通过 `PluginAPI` 注册 | 不再依赖 `Device` 接口 | high | +| 2.2 | WebUI 通过 `Subscribe(EventAll)` 获取所有事件 | 取代 OutputChan 监听 | high | +| 2.3 | WebUI 注册 `output_send` 工具 | 通过 `RegisterTool` 暴露给 LLM | high | +| 2.4 | 删除 `internal/api/plugin.go` 的 Device 包装 | 不再需要 `Device` 适配器 | medium | +| 2.5 | Handler 改为通过 EventBus 获取 IOManager 引用 | 减少直接依赖 | low | + +### Phase 3 — 迁移 QQ/OneBot 到 SDK 模式 + +| # | 任务 | 说明 | 优先级 | +|---|------|------|--------| +| 3.1 | OneBot 插件改为 `PluginAPI.RegisterTool` | 注册 `qq_send_private_msg` 等工具 | high | +| 3.2 | OneBot 接管后通过 `Publish(raw_input)` 发布事件 | 取代 IOManager.InjectInput | high | +| 3.3 | OneBot 注册阶段钩子 | 可接入群聊特定的 `pre_action` 逻辑 | medium | +| 3.4 | 删除 `internal/onebot/device.go` 的 Device 包装 | SDK 模式原生支持 | medium | + +### Phase 4 — 迁移 OutputBus 到 SDK + +| # | 任务 | 说明 | 优先级 | +|---|------|------|--------| +| 4.1 | 创建 `internal/outputbus/` 插件 | 管理 `output_send`/`output_list_channels` | high | +| 4.2 | 通过 `RegisterTool` 注册输出工具 | LLM 可直接调用 | high | +| 4.3 | 通过 `RegisterStage(before_output)` 拦截最终文本 | 渠道适配 | medium | +| 4.4 | Agent 内置的 output_* 工具改为委托给 outputbus | 解耦核心 | medium | + +### Phase 5 — 清理旧组件 + +| # | 任务 | 说明 | 优先级 | +|---|------|------|--------| +| 5.1 | 删除 `Device` 接口定义 | 全部迁移后移除 | high | +| 5.2 | 删除 `IOManager.ExecuteTool` | 工具路由走 StageHost | high | +| 5.3 | 删除 `IOManager.EmitOutput`/`EmitOutputTo` | 走 EventBus | medium | +| 5.4 | 删除 `IOManager.AtomicSwapDevices` | 不再需要设备热替换 | medium | +| 5.5 | 删除 `PluginDevice` 包装器 | SDK 模式替代 | medium | +| 5.6 | 删除 `internal/onebot/device.go` | 已迁移到 SDK | high | +| 5.7 | 删除 `internal/api/plugin.go` | 已迁移到 SDK | medium | +| 5.8 | 精简 `cmd/homed/main.go` | 移除设备相关初始化 | medium | + +### Phase 6 — 进程隔离 + +| # | 任务 | 说明 | 优先级 | +|---|------|------|--------| +| 6.1 | 实现 Unix Socket JSON-RPC 传输层 | 进程隔离模式 | low | +| 6.2 | `sdk.Run()` 自动检测 in-process/external | 开发 vs 生产 | low | +| 6.3 | 插件进程管理（启动/停止/健康检查） | Supervisor 扩展 | low | + +### Phase 7 — 增强功能 + +| # | 任务 | 说明 | 优先级 | +|---|------|------|--------| +| 7.1 | WebUI D3.js 力导向图记忆星图 | 已有 API `GET /api/v1/memory/star` | low | +| 7.2 | Model Context Protocol (MCP) 支持 | 标准工具协议 | low | +| 7.3 | 多 Agent 支持 | 每个 Agent 独立上下文 | low | +| 7.4 | Python 插件 SDK | 扩展生态 | low | + +--- + +## 文件最终结构（Phase 5 完成后） + +``` +HomeAgent/ +├── cmd/homed/main.go — 入口 +├── internal/ +│ ├── agent/ +│ │ ├── core/ +│ │ │ ├── agent.go — Agent 核心 +│ │ │ ├── context.go — 相关性上下文 +│ │ │ └── stages.go — StageHost +│ │ └── api/ +│ │ └── provider.go — LLM Provider +│ ├── events/ +│ │ └── bus.go — 系统事件总线 +│ ├── plugin/ +│ │ └── sdk/ +│ │ ├── api.go — PluginAPI +│ │ └── bus.go — 插件 EventBus +│ ├── memory/ — 三层记忆 +│ ├── knowledge/ — 知识库 +│ ├── tracker/ — 变更追踪 +│ ├── supervisor/ — 守护进程 +│ └── plugins/ — 插件实现 +│ ├── webui/ — HTTP API + 仪表盘 +│ ├── onebot/ — QQ 通道 +│ └── outputbus/ — 输出通道管理 +├── docs/ +│ └── ARCHITECTURE.md — 架构文档 +├── DESIGN.md +├── PLAN.md +└── README.md +``` + +## 设计原则 + +1. **核心零 IO** — Core 不依赖任何插件、设备、通道实现 +2. **三通道标准** — 所有插件通过 Tool/Stage/Event 与核心交互 +3. **增量迁移** — 每阶段保持向后兼容，旧组件与新 SDK 并行运行 +4. **测试覆盖** — 每阶段提交前确保全部测试通过 diff --git a/README.md b/README.md index cd0ae1f..c964f41 100644 --- a/README.md +++ b/README.md @@ -1,26 +1,29 @@ # HomeAgent -单二进制 24/7 智能家庭管家。基于 NextAgent 认知解耦架构 + TrulyMEM 自主图记忆。 +单二进制 24/7 智能管家。**核心零 IO**，所有输入输出通过插件，插件通过三通道与核心交互：工具、阶段钩子、事件订阅。 ## 架构 ``` - IO 抽象层（唯一输入路径） - ┌─────────────────────────────────────────────────────┐ - │ Microphone Camera GPIO HTTP OneBot-QQ Plugins │ - │ 所有外部输入 → InputEvent → inputCh │ - └──────────────────────┬──────────────────────────────┘ - │ - ┌──────────────────────▼──────────────────────────────┐ - │ Agent Core（编排器） │ - │ 三层记忆注入 → Provider.Chat() → 工具执行 → 输出 │ - └──────────────────────┬──────────────────────────────┘ - │ - ┌──────────────────────▼──────────────────────────────┐ - │ API 抽象层（唯一输出路径） │ - │ Provider: OpenAI / Ollama / Lua 适配 │ - │ DeepSeek v4 flash（默认） │ - └─────────────────────────────────────────────────────┘ + 外部（QQ / HTTP / 硬件...） + │ 通过插件注册 + ▼ + IOManager.InjectInput() + │ + ▼ + ┌─────────────────────────────────────────────────┐ + │ Agent Core │ + │ │ + │ on_input → Context → Memory Recall │ + │ → pre_action → [LLM ↔ 工具循环] → before_output │ + │ → output_send → after_output │ + │ │ + │ 内置：三层记忆 + 知识库 + Child Agent │ + └─────────────────────────────────────────────────┘ + │ + ├── Stage Pipeline（7 阶段，插件可拦截/改写） + ├── Tool System（插件注册工具给 LLM） + └── Event Bus（插件订阅系统事件） ``` ## 快速开始 @@ -32,62 +35,49 @@ make build 依赖：Go 1.19+、CGo（go-sqlite3）。 -## 记忆体系（三层） +## 阶段管道 -| 层 | 存储 | 容量 | 裁剪策略 | +``` +on_input → pre_action → post_action ↔ before_toolcall/after_toolcall → before_output → after_output + ↑_______________| + 循环 +``` + +| 阶段 | 插件可做 | +|---|---| +| `on_input` | 鉴权、拉黑、改写、短路 | +| `pre_action` | 注入 context 消息 | +| `post_action` | 审查/改写 LLM 输出、增删工具 | +| `before_toolcall` | 拒绝、改参、审计 | +| `after_toolcall` | 脱敏、改写结果 | +| `before_output` | 改写最终文本、加格式 | +| `after_output` | 记录/统计 | + +## 三层记忆 + +| 层 | 存储 | 容量 | 裁剪 | |---|---|---|---| -| 上下文 | 内存 | 30 条 | TF-IDF 相关性排序，淘汰→文档 | -| 文档 | JSON + TF-IDF 向量 | 无上限 | 72h 冷访问→图数据库 | -| 图 | SQLite 三元组 | 无上限 | 定期重整+同义合并 | +| Context | 内存 TF-IDF | 30 条 | 余弦相似度排序→文档 | +| Document | JSON + 向量索引 | ∞ | 72h 冷→图 | +| Graph | SQLite 三元组 | ∞ | 定期重整+同义合并 | -## 插件系统（OpenClaw 兼容） +## 插件三通道 -插件 = 容器，内嵌 IO 通道为组件。`plugins/` 目录热插拔，agent 通过 `plgreload` 工具控制重载。 - -``` -plugins/ -├── qq/ # OneBot QQ 通道插件 -│ ├── SKILL.md # 技能描述 + IO 端口声明 -│ └── skill.json # 元数据 + WS 连接配置 -``` - -### 示例 QQ 插件 `plugins/qq/SKILL.md` - -```markdown -# QQ 通知插件 -io_type: io -io_input_route: qq -io_output_route: qq -io_output_caps: text,file,image - -## qq_send_private_msg -发送 QQ 私聊消息 -- user_id: 目标 QQ 号 -- message: 消息内容（支持 CQ 码） -``` +| 通道 | 方向 | 用途 | +|---|---|---| +| `RegisterTool` | 插件→LLM | Agent 调用插件功能 | +| `RegisterStage` | 核心→插件 | 干预消息处理流 | +| `Subscribe/Publish` | 双向 | 审计/日志/通知 | ## 核心命令 ```bash -make build # 编译主二进制 -make run # 编译并启动（数据 /tmp/homeagent） -make install # 安装到系统 -make test # 运行测试 +make build # 编译 +make run # 编译+启动 +make test # 测试 +make install # 系统安装 ``` -## 配置 +## 完整文档 -`/etc/homeagent/config.yaml`（默认 `config/config.yaml`）： - -```yaml -daemon: - listen_addr: ":8080" - data_dir: "/var/lib/homeagent" -llm: - model: "deepseek-v4-flash" - base_url: "https://api.deepseek.com/v1" -``` - -## 许可证 - -MIT +详见 [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md)。 diff --git a/cmd/homed/main.go b/cmd/homed/main.go index cf2777d..077045c 100644 --- a/cmd/homed/main.go +++ b/cmd/homed/main.go @@ -3,7 +3,6 @@ package main import ( "flag" "log" - "net/http" "os" "os/signal" "path/filepath" @@ -16,6 +15,7 @@ import ( agentPkg "gitcode.com/JianFeeeee/HomeAgent/internal/agent" "gitcode.com/JianFeeeee/HomeAgent/internal/api" "gitcode.com/JianFeeeee/HomeAgent/config" + "gitcode.com/JianFeeeee/HomeAgent/internal/events" "gitcode.com/JianFeeeee/HomeAgent/internal/knowledge" luapkg "gitcode.com/JianFeeeee/HomeAgent/internal/lua" "gitcode.com/JianFeeeee/HomeAgent/internal/memory" @@ -227,6 +227,15 @@ func main() { log.Printf("[homed] knowledge store active with %d items", len(ks.List())) } + // === Event Bus (系统事件总线) === + evBus := events.NewBus() + log.Printf("[homed] event bus initialized") + + // === Stage Host (阶段管道编排) === + stageHost := agentCore.NewStageHost() + stageHost.SyncFromRegistry(pluginReg) + log.Printf("[homed] stage host initialized with %d plugin sdks", pluginReg.SDKPluginCount()) + // === Single Agent Core === agent := agentCore.New(agentCore.AgentConfig{ ID: "main", @@ -259,8 +268,11 @@ func main() { DocStore: docStore, Knowledge: ks, Personality: personality, - PluginReg: pluginReg, - PluginDir: filepath.Join(cfg.Daemon.DataDir, "plugins"), + PluginReg: pluginReg, + PluginDir: filepath.Join(cfg.Daemon.DataDir, "plugins"), + ContextSavePath: filepath.Join(cfg.Daemon.DataDir, "memory", "context.json"), + StageHost: stageHost, + EventBus: evBus, }) agent.Start() defer agent.Stop() @@ -271,23 +283,14 @@ func main() { log.Printf("[homed] main agent started, model=%s base=%s", cfg.LLM.Model, cfg.LLM.BaseURL) - // === HTTP API === - handler := api.NewHandler(sup, memDB, skMgr, luaVM, cfg, iom, textMem, ks, trk) - - mux := http.NewServeMux() - handler.RegisterRoutes(mux) - - server := &http.Server{ - Addr: cfg.Daemon.ListenAddr, - Handler: mux, - } - - go func() { - log.Printf("[homed] HTTP API listening on %s", cfg.Daemon.ListenAddr) - if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { - log.Fatalf("http server: %v", err) - } - }() + // === Built-in HTTP API & WebUI Plugin === + webui := api.NewWebUIPlugin( + "webui", cfg.Daemon.ListenAddr, + sup, memDB, skMgr, luaVM, cfg, iom, textMem, ks, trk, + ) + iom.RegisterDevice(webui) + webui.Start() + defer webui.Stop() sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) @@ -298,7 +301,6 @@ func main() { trk.Stop() } sup.Shutdown() - server.Close() log.Printf("[homed] stopped") } diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index ac0dc05..378c8e6 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -1,284 +1,540 @@ -# HomeAgent 架构参考 +# HomeAgent 架构设计 v4 -> 基于 NextAgent 认知解耦架构 + TrulyMEM 自主图记忆 + OneBot 协议。 +## 一、核心理念 -## 一、分层架构 +24 小时陪伴用户的智能管家。**单会话·单 Agent**，身份不漂移。 + +### 设计原则 +- **核心零 IO** — Core 没有任何硬编码 IO 能力，所有 IO 来自插件 +- **所有输出是工具调用** — Agent 必须显式调用 `output_send` 才能通信，推理不自动路由 +- **所有 LLM 调用走 Provider 接口** — 不直连 API +- **DeepSeek v4 flash** 为默认 LLM，thinking 模式关闭 +- **人格固定**（personal.md），记忆分层管理防止性格突变 +- **知识独立于记忆**，agent 主动学习 +- **插件 = 三通道**：工具、阶段钩子、事件订阅 + +--- + +## 二、核心域 vs 插件域 ``` -┌──────────────────────────────────────────────────────────┐ -│ IO 抽象层（唯一输入路径） │ -│ Device(Mic/Speaker/Camera/GPIO/OneBot-QQ/PluginDevice) │ -│ 所有外部输入 → InputEvent → inputCh │ -│ 输出通道: 能力声明(text/file/image/audio/structured) │ -│ 路由表: 输入源 → 默认输出通道 │ -└──────────────────────────┬───────────────────────────────┘ - │ -┌──────────────────────────▼───────────────────────────────┐ -│ Agent 操作层（核心编排器） │ -│ eventLoop() consume inputCh │ -│ ├─ RelevanceContext（TF-IDF 相关性管理） │ -│ ├─ 工具循环（Provider.Chat → tool_calls → Execute → ...) │ -│ ├─ 记忆索引注入（图索引 + 文档摘要） │ -│ └─ 心跳蒸馏（30min：向量同步 + 冷归档 + 同义合并） │ -└──────────────────────────┬───────────────────────────────┘ - │ -┌──────────────────────────▼───────────────────────────────┐ -│ API 抽象层（唯一输出路径） │ -│ Provider.Chat() → DeepSeek API / OpenAI / Ollama │ -│ LuaAdapter 做请求/响应格式转换 │ -└──────────────────────────────────────────────────────────┘ +┌──────────────────────────────────────────────────────────────────┐ +│ 核心域 (Core Domain) │ +│ │ +│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌───────────────────┐ │ +│ │ Provider │ │ Memory │ │Knowledge │ │ Pipeline Stages │ │ +│ │ (LLM) │ │ (T/D/G) │ │ (TF-IDF) │ │ 编排器 │ │ +│ └──────────┘ └──────────┘ └──────────┘ └───────────────────┘ │ +│ ┌──────────┐ ┌──────────┐ ┌──────────────────────────────────┐ │ +│ │ Relevance│ │ Context │ │ Plugin Host │ │ +│ │ Tf-Idf │ │ Persist │ │ (调用钩子 + 路由工具) │ │ +│ └──────────┘ └──────────┘ └──────────────────────────────────┘ │ +│ 核心无任何 IO 能力 │ +├──────────────────────────────────────────────────────────────────┤ +│ 边界 (Plugin API) │ +│ ┌──────────────────────────────────────────────────────────────┐ │ +│ │ RegisterTool(name, handler) ← 插件注册工具给 LLM │ │ +│ │ RegisterStage(stage, handler) ← 插件挂入消息处理阶段 │ │ +│ │ Subscribe(eventType, handler) ← 插件订阅系统事件 │ │ +│ │ Publish(event) → 插件发布事件 │ │ +│ └──────────────────────────────────────────────────────────────┘ │ +├──────────────────────────────────────────────────────────────────┤ +│ 插件域 (Plugin Domain) │ +│ │ +│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │ +│ │ WebUI │ │ QQ │ │OutputBus │ │ 未来插件: …… │ │ +│ │ HTTP/WS │ │ OneBot │ │通道管理 │ │ │ │ +│ └──────────┘ └──────────┘ └──────────┘ └──────────────────┘ │ +│ 所有 IO 都在这里 │ +└──────────────────────────────────────────────────────────────────┘ ``` -**三条核心规则：** -1. **所有外部输入** → 必须通过 `IOManager.InjectInput()` / `InjectText()` 注入 -2. **所有 LLM 调用** → 必须通过 `Provider.Chat()` 发出 -3. **Agent 不直接操作记忆系统**，只发射 `memory_candidate` 事件，由 Memory Pipeline 异步消费 +### 核心职责 +- LLM 调用编排（Provider → Agent 工具循环） +- 三层记忆管理（Context → Document → Graph） +- 知识库维护（Knowledge Store） +- 阶段管道编排（Stage Pipeline） +- Context 相关性管理（TF-IDF 余弦相似度） +- 心跳蒸馏 + 图重整 -## 二、子系统详解 +### 插件职责 +- 接收外部输入（WebSocket、HTTP、硬件等） +- 提供输出能力（文本发送、文件传输等） +- 干预消息处理流（阶段钩子） +- 观察系统状态（事件订阅） -### 2.1 IO 抽象层（唯一输入路径） +--- -所有外部输入必须通过此层进入系统。 +## 三、消息处理阶段管道 ``` -Device(Microphone) ─┐ -Device(OneBot-QQ) ─┤ -Device(Plugin) ─┤──→ IOManager → InputEvent → inputCh → Agent -Device(GPIO) ─┤ -HTTP API ─┘ + ┌──────────────────────────────────────────┐ + │ on_input │ + │ 消息到达，Agent 未做任何处理 │ + │ └→ 插件可鉴权/拉黑/改写/短路回复 │ + └──────────┬───────────────────────────────┘ + │ 通过 + ┌──────────▼───────────────────────────────┐ + │ 内部：Context Append + Memory Recall │ + │ + Context 组装 │ + └──────────┬───────────────────────────────┘ + │ 就绪 + ┌──────────▼───────────────────────────────┐ + │ pre_action │ + │ 上下文已就绪，即将调用 LLM │ + │ └→ 插件可注入 system 消息 / 修改 context │ + └──────────┬───────────────────────────────┘ + │ LLM 调用 + ┌──────────▼───────────────────────────────┐ + │ post_action │ + │ LLM 返回文本 + 工具调用列表 │ + │ └→ 插件可审查/修改文本、增删工具调用 │ + └──────────┬───────────────────────────────┘ + │ 判断有无工具调用 + ╱─────────────┴─────────────╲ + 有工具调用 无工具调用 + │ │ + ┌──────────▼──────────────┐ │ + │ before_toolcall │ │ + │ 即将执行某个工具调用 │ │ + │ └→ 插件可拒绝/放行/ │ │ + │ 修改参数/审计 │ │ + └──────────┬──────────────┘ │ + │ 执行工具 │ + ┌──────────▼──────────────┐ │ + │ after_toolcall │ │ + │ 工具执行完毕，准备喂回 │ │ + │ └→ 插件可脱敏/改写结果 │ │ + └──────────┬──────────────┘ │ + │ 回到 post_action 继续循环 │ + └──────────────────────────────┘ + │ + ┌──────────────────────────┘ + ▼ + ┌──────────────────────────────────────────┐ + │ before_output │ + │ 最终文本就绪，即将调用 output_send │ + │ └→ 插件可改写回复/添加格式/适配渠道 │ + └──────────────────┬───────────────────────┘ + │ output_send 调用 + ┌──────────────────▼───────────────────────┐ + │ after_output │ + │ 输出完成 │ + │ └→ 记录/统计/清理资源 │ + └──────────────────────────────────────────┘ ``` -**核心类型：** +### 7 个阶段总表 -| 类型 | 说明 | -|------|------| -| `InputEvent` | Source + Type + Payload — 所有外部输入的标准化格式 | -| `OutputEvent` | Target + Type + Payload + OutputChannel — 输出路由 | -| `Device` | 接口：Name/Type/Description/Tools/Execute/Start/Stop/OutputCapabilities | -| `DeviceType` | Input / Output / IO | -| `ToolDef` | Name + Description + Parameters + Handler — 与 LLM 函数调用同构 | -| `OutputCapability` | 位掩码：text, file, image, audio, structured | +| 阶段 | 触发时机 | 插件读写权限 | 典型用途 | +|---|---|---|---| +| `on_input` | 消息到 Agent，零处理 | 可读写 `raw_message`，可设置 `response` 短路 | 黑名单、限流、自定义指令前缀 | +| `pre_action` | Memory+Context 就绪，LLM 调用前 | 可读写 `context_messages`（追加/修改） | 注入 RAG 结果、插入时政 context | +| `post_action` | LLM 返回文本 + 工具调用列表 | 可读写 `llm_text`、`tool_calls`、`context_messages` | 敏感词过滤、强制 redirect 工具 | +| `before_toolcall` | 单个工具调用执行前 | 可读写 `tool_call.name`、`tool_call.args`，设置 `deny=true` 拒绝 | 审计高危操作、OS 命令白名单 | +| `after_toolcall` | 单个工具执行完毕 | 可读写 `tool_result` | 脱敏数据库结果、排序搜索结果 | +| `before_output` | 最终文本就绪，output_send 前 | 可读写 `final_text`，可设置 `skip_output=false` | 添加表情/at 前缀、多平台格式适配 | +| `after_output` | output_send 已调用 | 只读 `final_text` | 统计日志、触发后续流程 | -**内置设备：** +### 循环规则 -| 设备 | 方向 | 工具 | -|------|------|------| -| Microphone | Input | capture — 录音 | -| Speaker | Output | speak — 语音播放 | -| Camera | Input | capture 拍照 | -| OneBot QQ | IO | send_private/send_group/get_group_member_info/get_group_list | -| PluginDevice | IO | 插件声明工具 | -| GPIO | IO | gpio_write/gpio_read | +`post_action → before_toolcall → after_toolcall → 回到 post_action` 构成**内循环**。Agent 在以下条件退出循环进入 `before_output`： +- LLM 返回纯文本（无工具调用） +- `before_toolcall` 拒绝所有剩余工具且 LLM 无可执行工具 +- 循环超过 `max_tool_rounds` 上限 -**输出通道能力校验：** -- 每个 Device 声明 `OutputCapabilities()` → 位掩码 -- `output_send` 工具发送前校验通道是否支持文本 -- `output_list_channels` 只列出有输出能力的通道 - -### 2.2 API 抽象层（唯一输出路径） +### 短路规则 +每个阶段插件都可设置 `ctx.Response`，一旦设置管道立即短路到 `after_output`： ``` -Agent Core → Provider.Chat() - │ - ┌───────┴───────┐ - ▼ ▼ - OpenAIProvider LuaAdapter - (DeepSeek API) (格式转换) +on_input → ctx.Response = "hello" → 跳过后面的所有阶段 → after_output ``` -| 实现 | 说明 | -|------|------| -| `OpenAIProvider` | 标准 OpenAI API 格式，DeepSeek v4 flash 默认 | -| `LuaAdaptedProvider` | 通过 Lua 脚本转换请求/响应的适配 wrapper | +--- -Lua 适配器位于 `{dataDir}/adapters/*.lua`，每个适配器返回 `name` + `transform_request` + `transform_response`。 - -### 2.3 Agent 操作层（核心编排器） +## 四、记忆体系（三层递进） ``` -eventLoop() → select on inputCh +输入消息 │ ▼ -handleInput(evt) → processTextInput(input) - │ - ├─ 1. RelevanceContext.Append(input) - ├─ 2. 构建 prompt: personal.md + 图索引 + 文档摘要 + 上下文 + 工具 - ├─ 3. 工具循环（最多 10 轮） - │ LLM → tool_calls → Execute → 结果注入 → 下一轮 - ├─ 4. 追加响应到上下文 - ├─ 5. RelevanceContext.Prune() → 低分事件→文档记忆归档 - ├─ 6. OutputEvent → 输出通道 - └─ 7. memory_candidate → TextMemory + Distiller → GraphDB +┌─────────────────────────────────────────────────────┐ +│ Layer 1: Context (RelevanceContext) │ +│ 内存中维护最近 topK 条事件，TF-IDF 评分，JSON 持久化 │ +│ 每次 Append/Prune → save() 防崩溃丢数据 │ +│ keep=30 条活跃，多余 → 归档到 Document │ +└────────────────────┬────────────────────────────────┘ + │ Prune 时 + ▼ +┌─────────────────────────────────────────────────────┐ +│ Layer 2: Document (document.Store) │ +│ 文件系统 JSON + TF-IDF 向量索引 │ +│ 冷文档（72h 未访问 + access ≤ 2）→ 蒸馏到 Graph │ +│ 也可以由用户主动 commit（doc_commit 工具） │ +└────────────────────┬────────────────────────────────┘ + │ reorg 心跳 + ▼ +┌─────────────────────────────────────────────────────┐ +│ Layer 3: Graph (GraphDB + Indexer) │ +│ SQLite: entities + relations │ +│ Entity: name, type, mention_count │ +│ Relation: source → target, relation_type, confidence│ +│ 搜索: 关键词 → 向量搜索实体 → BFS 遍历邻居 │ +│ 蒸馏: 原始记录 → Distiller → 三元组提交 │ +└─────────────────────────────────────────────────────┘ ``` -**人格注入：** personal.md 加载一次，固定在 system prompt 最前，永不漂移。 - -**工具分发：** -``` -executeToolCall(tc) - ├─ memory_* → executeMemoryTool - ├─ knowledge_* → executeKnowledgeTool - ├─ doc_* → executeDocTool - ├─ output_* → executeOutputChannel/Send/ListChannels - └─ 其他 → io.ExecuteTool → Device.Execute -``` - -### 2.4 记忆系统 - -三层分级，自顶向下逐渐持久化、抽象化： +### 数据流关系 ``` -Layer 1: 上下文（RelevanceContext） - 内存环形缓冲，TF-IDF 余弦相似度排序 - 每次响应后保留 top 30，低分→文档记忆 - -Layer 2: 文档记忆（document.Store） - JSON 文件 + TF-IDF 向量索引（字符 bigram） - 冷文档（72h 未访问 + ≤2 次）→ 图数据库 - -Layer 3: 图数据库（memory.GraphDB） - SQLite 三元组（实体-关系-实体） - 只注入索引（实体名+类型+提及次数）到 prompt - 定期重整：向量同步 + bigram Jaccard 同义合并 +Context 修剪 → Document 归档 → reorg 心跳 → Graph 消化 + ↑ + Distiller (原始记录 → 三元组) ``` -### 2.5 知识系统 +### 工具入口（Agent 暴露给 LLM） + +- `memory_recall(query)` → 从 Graph 召回 +- `memory_commit(triples)` → 写入 Graph +- `memory_introspect()` → 查看统计 +- `doc_query(query)` → 从 Document 搜索 +- `doc_commit(title, content)` → 写入 Document + +--- + +## 五、知识体系 独立于记忆，agent 主动学习。 ``` -knowledge/{category}/content.md +knowledge// + content.md + +knowledge.Store + └─ TF-IDF 向量索引 (character bigram) + └─ 独立于 memory 的 vector.Store 实例 + └─ Start() 时扫描目录训练索引 + └─ Add(name, content) 时增量更新 +``` + +### 工具入口 + +- `knowledge_search(query)` → 向量搜索 +- `knowledge_create(name, content)` → 新增 +- `knowledge_list()` → 列出所有 + +### 为什么独立于 memory？ + +- Memory 是 LLM 的"对话记忆"——谁说过什么、上下文 +- Knowledge 是 LLM 的"知识库"——外部注入的固定知识 +- 两者 TF-IDF 索引实例隔离，不互相污染 + +--- + +## 六、三通道插件交互 + +``` +插件 ──→ 核心 核心 ──→ 插件 +────────────────────────────────────────────────── +RegisterTool(name, fn) ──→ buildToolDefs() + executeToolCall() → fn + （Tracker 自动包裹 Pre/PostAction） + +RegisterStage(stage, fn) ──→ runStage() 在对应阶段调用 fn(ctx) + 返回后检查 ctx.Response 决定是否短路 + +Subscribe(eventType, fn) ──→ Publish(event) + 所有订阅者收到（观察型） +``` + +### 通道对比 + +| 通道 | 方向 | 用途 | 可否拦截 | +|---|---|---|---| +| **工具** (RegisterTool) | 插件→核心→LLM | LLM 主动调用插件功能 | 否 | +| **阶段** (RegisterStage) | 核心→插件 | 核心触发插件干预消息流 | 是（response 短路） | +| **事件** (Subscribe/Publish) | 双方向 | 审计/日志/状态通知 | 否 | + +--- + +## 七、Agent 内部完整流程 + +``` +processTextInput(input) + │ + ├── on_input stage ────────────── 插件可拦截/改写 + │ + ├── context.Append(input) + ├── context.Prune(input) → 归档到 Document + ├── buildMemoryContext() → Indexer.BuildContext → Graph Recall + │ + ├── pre_action stage ──────────── 插件可注入 context + │ + ├── [循环] process(input) + │ ├── buildSystemPrompt (人格+记忆+技能+上下文) + │ ├── buildToolDefs (内置工具 + 插件工具) + │ ├── provider.Chat() → LLM + │ │ + │ ├── post_action stage ─────── 插件可见 LLM 输出 + 工具列表 + │ │ + │ ├── 有工具调用？ + │ │ ├── 每个工具: + │ │ │ ├── before_toolcall stage ── 插件可拒绝/改参 + │ │ │ ├── Tracker.PreAction + │ │ │ ├── executeToolCall() ──── 路由到插件或内置 + │ │ │ ├── Tracker.PostAction + │ │ │ └── after_toolcall stage ── 插件可改结果 + │ │ └── → 回到 post_action (继续循环) + │ │ + │ └── 无工具调用 → 退出循环 + │ + ├── context.Append(response) + ├── before_output stage ───────── 插件可改写最终文本 + ├── Publish(agent_output event) + ├── output_send (调用插件注册的 output 工具) + │ + └── after_output stage ────────── 插件只读，做统计/日志 +``` + +--- + +## 八、记忆整理（心跳 LLM 驱动消歧） + +图数据库在长期运行中会积累**同义实体**（如「张三」与「张先生」指同一人）和**矛盾关系**。心跳流程如下： + +### 流程 + +``` +心跳 tick (30min) + │ + ├── distillContext() — 蒸馏上下文 + ├── syncGraphToDocs() — 图→文档 + │ + └── reorgGraph() + ├── Indexer.Sync() — 图→向量（自动） + ├── DocStore.Reindex() — 文档重建索引（自动） + ├── 冷文档→图归化 — 将冷文档归档为图三元组（自动） + │ + └── 实体冲突检测 → 发现相似实体对 + │ 如：「张三」(person, 5次) vs 「张先生」(person, 3次) 相似度 0.75 + │ + ▼ + enqueueConsolidationTask() + │ 通过 IO 层注入 Agent 输入队列 + │ channel = "_consolidation_"（内部通道，不对外输出） + ▼ + Agent 处理 (processConsolidation) + │ 如同普通用户消息，走完整 LLM 工具循环 + │ 但输出仅写记忆，不发外部通道 + ▼ + LLM 决策： + ├─ 判断为同一实体 → 调用 memory_merge 合并 + │ → "已将「张先生」合并到「张三」，3 条关系已重定向" + ├─ 判断为不同实体 → 回复"跳过" + └─ 不确定 → 回复"待定，需更多上下文" +``` + +### 关键设计 + +| 特性 | 说明 | +|---|---| +| **启发式检测，LLM 决策** | bigram Jaccard 仅做候选筛选（低门槛 0.5），LLM 做最终判断 | +| **走 IO 输入队列** | 不阻塞心跳，不抢占用户输入，享受完整 Agent 上下文 | +| **`_consolidation_` 通道** | 内部专用通道，输出只写记忆层，不被外部插件路由 | +| **`memory_merge` 工具** | LLM 通过此工具执行合并，自动重定向关系 + 累积 mention_count | +| **异步非阻塞** | 整理任务排队在 inputCh 尾部，Agent 按序处理，不影响用户体验 | + +### 类比 + +类似人类睡眠时大脑的海马体回放——白天经历的记忆在休息时被自发整理、关联、去重。HomeAgent 的心跳就是它的"睡眠周期"，而 LLM 的参与相当于前额叶皮层执行语义判断。类比： + +```diff +- 人类: 白天经历 → 海马体暂存 → 睡眠 → 前额叶整理 → 长期记忆 ++ Agent: 用户交互 → Context缓存 → 心跳 → LLM 消歧 → GraphDB 存储 +``` + +--- + +## 九、Child Agent + +不走阶段管道，独立轻量 Agent： + +``` +spawn_child(task) → 新建轻量 Agent + ├── 独立 system prompt（仅有任务描述） + ├── 仅 output_send 工具 + ├── 无 persistent memory + ├── 无 Graph/Document 访问 + ├── 上限 5 轮工具循环 + └── 销毁时返回结果文本 +``` + +--- + +## 十、SDK API 定义 + +### PluginAPI (`internal/plugin/sdk/api.go`) + +```go +type PluginAPI struct { + Name string + Version string +} + +func NewPluginAPI(name, version string, bus EventBus, mem MemoryAPI, know KnowledgeAPI) *PluginAPI + +// 三通道 +func (p *PluginAPI) RegisterTool(name string, handler ToolHandler) error +func (p *PluginAPI) RegisterStage(stage Stage, handler StageHandler) +func (p *PluginAPI) Subscribe(eventType EventType, handler EventHandler) +func (p *PluginAPI) Publish(evt *Event) + +// 访问子系统的快捷方式 +func (p *PluginAPI) Memory() MemoryAPI +func (p *PluginAPI) Knowledge() KnowledgeAPI +``` + +### 阶段上下文 (`StageContext`) + +```go +type StageContext struct { + RawMessage string // 当前输入（可改写 on_input） + UserID string + GroupID string + ContextMsgs []map[string]interface{} // 可注入的消息 + LLMText string // LLM 返回文本（可改写 post_action） + ToolCalls []ToolCall // 工具调用列表（可增删 post_action/before_toolcall） + ToolResults []ToolResult // 工具执行结果（可改写 after_toolcall） + FinalText string // 最终输出文本（可改写 before_output） + Response *string // 设置后短路管道 + Phase Stage // 当前阶段 + Memory []MemItem // 召回的记忆 + Extra map[string]interface{} // 扩展字段 +} +``` + +--- + +## 十一、事件系统 + +### 事件类型 + +| 类型 | 发布时机 | 用途 | +|---|---|---| +| `raw_input` | 消息到达 Agent | 记录输入日志 | +| `agent_output` | 最终输出发送后 | 记录输出日志 | +| `tool_call` | 每个工具调用完成 | 审计工具调用 | +| `reasoning` | LLM 推理文本 | 展示推理过程 | +| `system` | 系统状态变更 | 健康检查、插件变更 | + +### Event Bus (`internal/events/bus.go`) + +```go +type Bus struct{} +func NewBus() *Bus +func (b *Bus) Publish(event *Event) +func (b *Bus) Subscribe(eventType EventType, handler Handler) func() +``` + +--- + +## 十二、数据流全景 + +``` +外部 (QQ/HTTP/硬件) + │ 通过插件 + ▼ +IOManager.InjectInput() → inputCh │ ▼ -TF-IDF 向量索引（字符 bigram） +Agent.eventLoop() → handleInput → processTextInput │ - ▼ -工具: knowledge_search / knowledge_create / knowledge_list + ├── 1. on_input stage（插件可拦截） + ├── 2. Context.Append + ├── 3. Memory Recall (Indexer → Graph) + ├── 4. pre_action stage（插件可注入） + ├── 5. 工具循环 (最多 10 轮) + │ LLM → post_action → [before_toolcall → 执行 → after_toolcall] → LLM ... + ├── 6. Context.Append(response) + ├── 7. Prune（不相关 → Document） + ├── 8. before_output stage（插件可改写） + ├── 9. EmitOutput (通过 output_send 到对应通道) + ├── 10. after_output stage（插件只读） + └── 11. memory_candidate → TextMemory + Distiller → GraphDB + +心跳（30min）: + ├── distillContext() + ├── syncGraphToDocs() + └── reorgGraph() + ├── Indexer.Sync() — 图→向量（自动） + ├── DocStore.Reindex() — 文档向量重建（自动） + ├── 冷文档→图归化（自动） + └── 实体冲突检测 → 走 LLM 消歧（详见第八章） ``` -### 2.6 插件系统 +--- -OpenClaw SKILL.md 兼容。 +## 十三、代码结构 ``` -Plugin（容器） - ├─ 元数据: name, version, author - ├─ IOConfig（可选）: 声明 IO 端口 - ├─ Device（可选）: 原生 Go 设备 - └─ Tools: LLM 工具定义 - -Registry: - ├─ NativeFactory: "qq" → onebot.NewDevice - ├─ SetIOManager: 绑定 IO 管理器 - └─ Reload(): 原子化重载 -``` - -**热插拔流程：** -1. 扫描 `plugins/` 目录 -2. 原生工厂优先，无工厂则 LoadSKILL.md -3. 新设备 Start（预先启动） -4. `IOManager.AtomicSwapDevices()` 原子替换设备表 + 路由表 -5. 旧设备 Stop（后台 goroutine） - -### 2.7 OneBot QQ 通道 - -``` -OneBot 前端（go-cqhttp/Lagrange） - │ Reverse WebSocket - ▼ -OneBot Client（internal/onebot/） - ├─ 事件: message/notice/request → IO InputEvent - ├─ 动作: send_private_msg / send_group_msg / get_group_info / ... - ├─ 自动重连（指数退避 1s→30s） - └─ 事件 handler 注入 IO 层 - │ - ▼ -Device（internal/onebot/device.go） - ├─ Tools: qq_send_private_msg, qq_send_group_msg, etc. - ├─ Execute → Client.SendAction - └─ Start → Connect + 注册事件 handler -``` - -### 2.8 变更追踪器 - -Overlayfs 文件变更追踪： - -``` -PreAction(tool) → 记录文件 hash -PostAction(tool) → diff → ChangeSet{ID, Tool, Files[]SHA256} -Rollback() → 用 overlayfs 下层恢复 -``` - -健康检查失败 → `tracker.Rollback()`。 - -### 2.9 Supervisor 守护进程 - -``` -Daemon: - ├─ SetTracker() → 绑定变更追踪器 - ├─ RegisterAgent("main") → 注册主 agent - └─ 健康检查周期 → LLM API 可达性 - └─ 连续失败 → tracker.Rollback() -``` - -## 三、配置 - -```yaml -daemon: - listen_addr: ":8080" - data_dir: "/var/lib/homeagent" - heartbeat_interval: 15s - check_interval: 30s -llm: - provider: "openai" - model: "deepseek-v4-flash" - base_url: "https://api.deepseek.com/v1" - api_key: "${DEEPSEEK_API_KEY}" - temperature: 0.7 - max_tokens: 4096 -``` - -## 四、数据目录 - -``` -{dataDir}/ +cmd/homed/main.go — 入口：组装所有子系统 +internal/ +├── agent/ +│ ├── core/ +│ │ ├── agent.go — Agent 核心：事件循环、工具循环、心跳 +│ │ ├── context.go — RelevanceContext：TF-IDF 上下文管理 +│ │ └── stages.go — StageHost：阶段管道编排 +│ ├── api/ +│ │ └── provider.go — Provider 接口 + DeepSeek/Ollama 实现 +│ ├── io/ +│ │ └── channel.go — IOManager + Device 接口（过渡期保留） +│ └── personal.go — 人格加载 +├── api/ +│ ├── handler.go — HTTP API 端点 +│ └── plugin.go — WebUI Device 包装 +├── events/ +│ └── bus.go — 系统事件总线 (Publish/Subscribe) ├── memory/ -│ ├── graph.db # SQLite 图数据库 -│ ├── text/ # JSONL 文本记忆 -│ ├── documents/ # JSON 文档记忆 -│ └── raw/ # 原始记录（蒸馏后删除） -├── knowledge/ # 知识库 -│ └── {name}/content.md -├── plugins/ # 插件 -│ └── {name}/ -│ ├── SKILL.md -│ └── skill.json -├── skills/ # 技能 -├── adapters/ # Lua 适配器 -├── personal/ # 人格 -│ └── personal.md -├── changesets/ # 变更追踪记录 -└── snapshots/ # 快照 +│ ├── graph.go — SQLite 图数据库 +│ ├── indexer.go — 图索引器 +│ ├── vector/store.go — TF-IDF 向量存储 +│ ├── document/doc.go — 文档记忆 +│ ├── text/text.go — 文本记忆（JSONL） +│ └── pipeline/ — 蒸馏器 +├── knowledge/ +│ └── knowledge.go — 知识系统 +├── plugin/ +│ ├── plugin.go — 插件注册表 + 旧 Device 兼容层 +│ └── sdk/ +│ ├── api.go — PluginAPI 定义 +│ └── bus.go — 插件内部 EventBus 接口 +├── onebot/ — OneBot V11 QQ 协议实现 +├── tracker/ — 变更追踪 (overlayfs) +├── supervisor/ — 守护进程 +├── skill/ — 技能管理器 +├── lua/ — Lua 适配器 +├── network/ — 网络监控 +├── container/ — 容器管理 +├── snapshot/ — 快照 +├── embed/ — 嵌入 +└── tokenizer/ — 分词器 +config/ +├── config.go — 配置加载 +└── config.yaml +pkg/types/ — 类型定义 +docs/ +└── ARCHITECTURE.md — 本架构文档 ``` -## 五、HTTP API +--- -| 方法 | 路径 | 说明 | -|------|------|------| -| POST | `/v1/chat/completions` | OpenAI 兼容对话 | -| GET | `/api/v1/knowledge` | 知识查询/列表 | -| POST | `/api/v1/knowledge` | 创建知识 | -| DELETE | `/api/v1/knowledge?name=` | 删除知识 | -| GET | `/api/v1/status` | 系统状态 | -| GET | `/api/v1/config` | 配置查看 | +## 十四、与旧设计 (v3) 的关键区别 -## 六、构建与部署 - -```bash -make build # 编译主二进制 -make run # 编译 + 启动（数据 /tmp/homeagent） -make install # 安装到系统 -make test # 运行测试 -``` - -依赖：Go 1.19+ (CGo enabled for go-sqlite3)。单二进制部署。 +| 维度 | v3 | v4 | +|------|-----|-----| +| 插件交互 | Device 接口 + IOManager 路由 | 三通道：Tool/Stage/Event | +| 消息流编辑 | 无（纯事件推送） | 阶段管道 7 个 hook 点 | +| Event Bus | 无 | `internal/events/bus.go` | +| SDK | 无 | `internal/plugin/sdk/` | +| 核心 IO | IOManager `EmitOutput` 直出 | 全部走 `output_send` 工具 | +| 插件工具路由 | IOManager `ExecuteTool` 链 | StageHost + Registry 双层路由 | diff --git a/internal/agent/core/agent.go b/internal/agent/core/agent.go index 7815253..ad353e2 100644 --- a/internal/agent/core/agent.go +++ b/internal/agent/core/agent.go @@ -11,10 +11,12 @@ import ( agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" agentPkg "gitcode.com/JianFeeeee/HomeAgent/internal/agent" agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" + "gitcode.com/JianFeeeee/HomeAgent/internal/events" "gitcode.com/JianFeeeee/HomeAgent/internal/knowledge" "gitcode.com/JianFeeeee/HomeAgent/internal/memory" "gitcode.com/JianFeeeee/HomeAgent/internal/memory/document" "gitcode.com/JianFeeeee/HomeAgent/internal/plugin" + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk" "gitcode.com/JianFeeeee/HomeAgent/internal/skill" "gitcode.com/JianFeeeee/HomeAgent/internal/tracker" "gitcode.com/JianFeeeee/HomeAgent/pkg/types" @@ -59,6 +61,10 @@ type Agent struct { // 当前请求的输出通道（mutex 保护，process() 内独占） currentOutputChannel string + + // 阶段管道：插件消息流编辑 + stageHost *StageHost + eventBus *events.Bus } type AgentConfig struct { @@ -78,7 +84,10 @@ type AgentConfig struct { PluginReg *plugin.Registry PluginDir string DistillInterval time.Duration - MaxContextSize int // 活跃上下文最大条数，超出按相关性裁剪 + MaxContextSize int // 活跃上下文最大条数，超出按相关性裁剪 + ContextSavePath string // 上下文持久化路径，空则不持久化 + StageHost *StageHost + EventBus *events.Bus } func New(cfg AgentConfig) *Agent { @@ -100,7 +109,7 @@ func New(cfg AgentConfig) *Agent { indexer: cfg.Indexer, skills: cfg.Skills, tracker: cfg.Tracker, - context: NewRelevanceContext(), + context: NewRelevanceContext(cfg.ContextSavePath), systemPrompt: cfg.SystemPrompt, ctx: ctx, cancel: cancel, @@ -112,6 +121,8 @@ func New(cfg AgentConfig) *Agent { pluginDir: cfg.PluginDir, distillInterval: cfg.DistillInterval, maxContextSize: cfg.MaxContextSize, + stageHost: cfg.StageHost, + eventBus: cfg.EventBus, } } @@ -168,13 +179,31 @@ func (a *Agent) processTextInput(evt *agentIO.InputEvent, input string) { a.currentOutputChannel = evt.Source } + // 记忆整理任务：不路由到外部输出通道 + if evt.OutputChannel == "_consolidation_" { + a.processConsolidation(input) + return + } + + // === Stage: on_input — 消息到达，插件可拦截 === + stageCtx := a.stageCtxFromInput(input, evt.Source, "") + a.publishEvent(events.EventRawInput, map[string]interface{}{ + "content": input, + "source": evt.Source, + }) + if a.runStage(sdk.StageOnInput, stageCtx) { + a.emitResponse(evt, *stageCtx.Response) + return + } + input = stageCtx.RawMessage + a.context.Append(ContextEvent{ Timestamp: start, Source: evt.Source, Input: input, }) - response, toolsUsed, err := a.process(input) + response, toolsUsed, err := a.process(input, stageCtx) if err != nil { log.Printf("[agent] process error: %v", err) resp := fmt.Sprintf("处理错误: %v", err) @@ -206,6 +235,14 @@ func (a *Agent) processTextInput(evt *agentIO.InputEvent, input string) { } func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) { + // === Stage: before_output — 最终文本就绪，插件可改写 === + stageCtx := &sdk.StageContext{ + FinalText: response, + Phase: sdk.StageBeforeOutput, + } + a.runStage(sdk.StageBeforeOutput, stageCtx) + response = stageCtx.FinalText + // 读取当前输出通道（可能已被 AI 通过 output_set_channel 切换） ch := a.currentOutputChannel if ch == "" { @@ -230,10 +267,19 @@ func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) { OutputChannel: ch, } } + + // === Stage: after_output — 输出完成，插件只读 === + a.publishEvent(events.EventAgentOutput, map[string]interface{}{ + "content": response, + "channel": ch, + "source": evt.Source, + }) + stageCtx.Phase = sdk.StageAfterOutput + a.runStage(sdk.StageAfterOutput, stageCtx) } -// process — 内部处理，带工具循环 -func (a *Agent) process(input string) (response string, toolsUsed []string, err error) { +// process — 内部处理，带工具循环和阶段管道 +func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response string, toolsUsed []string, err error) { a.mu.Lock() defer a.mu.Unlock() @@ -248,6 +294,20 @@ func (a *Agent) process(input string) (response string, toolsUsed []string, err a.personality != nil && a.personality.Content != "", a.docStoreSize()) + // === Stage: pre_action — 上下文就绪，即将调用 LLM === + if a.runStage(sdk.StagePreAction, stageCtx) { + return *stageCtx.Response, toolsUsed, nil + } + if len(stageCtx.ContextMsgs) > 0 { + for _, m := range stageCtx.ContextMsgs { + role, _ := m["role"].(string) + content, _ := m["content"].(string) + if role != "" { + msgs = append(msgs, agentAPI.Message{Role: role, Content: content}) + } + } + } + for turn := 0; turn < a.maxTurns; turn++ { req := &agentAPI.CompletionRequest{ Messages: msgs, @@ -264,6 +324,15 @@ func (a *Agent) process(input string) (response string, toolsUsed []string, err return "", toolsUsed, fmt.Errorf("provider: %w", err) } + // === Stage: post_action — LLM 返回，插件可审查/修改 === + stageCtx.LLMText = resp.Content + stageCtx.ToolCalls = convertToolCalls(resp.ToolCalls) + if a.runStage(sdk.StagePostAction, stageCtx) { + return *stageCtx.Response, toolsUsed, nil + } + resp.Content = stageCtx.LLMText + resp.ToolCalls = convertBackToolCalls(stageCtx.ToolCalls) + if len(resp.ToolCalls) == 0 { return resp.Content, toolsUsed, nil } @@ -271,16 +340,74 @@ func (a *Agent) process(input string) (response string, toolsUsed []string, err for _, tc := range resp.ToolCalls { toolsUsed = append(toolsUsed, tc.Name) log.Printf("[agent] executing tool: %s (id=%s)", tc.Name, tc.ID) + + // === Stage: before_toolcall — 插件可拒绝/改参 === + sdkTC := sdk.ToolCall{ID: tc.ID, Name: tc.Name, Arguments: tc.Arguments} + stageCtx.ToolCalls = []sdk.ToolCall{sdkTC} + stageCtx.ToolResults = nil + if a.runStage(sdk.StageBeforeToolcall, stageCtx) { + result := fmt.Sprintf("工具 %s 已被插件拒绝", tc.Name) + msgs = append(msgs, agentAPI.Message{Role: "assistant", Content: resp.Content, ToolCalls: []agentAPI.ToolCall{tc}}) + msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result}) + a.publishEvent(events.EventToolCall, map[string]interface{}{ + "tool": tc.Name, + "args": tc.Arguments, + "result": result, + "status": "denied", + }) + continue + } + tc.Arguments = stageCtx.ToolCalls[0].Arguments + result := a.executeToolCall(tc) log.Printf("[agent] tool %s result: %s", tc.Name, truncateStr(result, 100)) + + // === Stage: after_toolcall — 插件可改结果 === + stageCtx.ToolResults = []sdk.ToolResult{{CallID: tc.ID, Name: tc.Name, Success: true, Result: result}} + a.runStage(sdk.StageAfterToolcall, stageCtx) + if len(stageCtx.ToolResults) > 0 { + if r, ok := stageCtx.ToolResults[0].Result.(string); ok { + result = r + } + } + msgs = append(msgs, agentAPI.Message{Role: "assistant", Content: resp.Content, ToolCalls: []agentAPI.ToolCall{tc}}) msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result}) + + a.publishEvent(events.EventToolCall, map[string]interface{}{ + "tool": tc.Name, + "args": tc.Arguments, + "result": result, + "status": "ok", + }) } } return "", toolsUsed, fmt.Errorf("tool execution exceeded %d turns", a.maxTurns) } +func convertToolCalls(tcs []agentAPI.ToolCall) []sdk.ToolCall { + if tcs == nil { + return nil + } + result := make([]sdk.ToolCall, len(tcs)) + for i, tc := range tcs { + result[i] = sdk.ToolCall{ID: tc.ID, Name: tc.Name, Arguments: tc.Arguments} + } + return result +} + +func convertBackToolCalls(tcs []sdk.ToolCall) []agentAPI.ToolCall { + if tcs == nil { + return nil + } + result := make([]agentAPI.ToolCall, len(tcs)) + for i, tc := range tcs { + result[i] = agentAPI.ToolCall{ID: tc.ID, Name: tc.Name, Arguments: tc.Arguments} + } + return result +} + func (a *Agent) docStoreSize() int { if a.docStore == nil { return 0 @@ -322,6 +449,15 @@ func (a *Agent) executeToolCall(tc agentAPI.ToolCall) string { return a.executeOutputListChannels() case tc.Name == "plgreload": return a.executePluginReload() + case tc.Name == "spawn_child": + return a.executeSpawnChild(tc) + } + + // 插件工具（通过 SDK RegisterTool 注册） + if a.stageHost != nil { + if result, err := a.stageHost.ExecuteTool(tc.Name, tc.Arguments); err == nil { + return fmt.Sprintf("%v", result) + } } if a.tracker != nil { @@ -414,7 +550,19 @@ func (a *Agent) executeMemoryTool(tc agentAPI.ToolCall) string { return fmt.Sprintf("记忆统计: %v", stats) case "memory_document_query": - return a.executeDocTool(tc) + return a.executeDocTool(tc) + + case "memory_merge": + source, _ := tc.Arguments["source"].(string) + target, _ := tc.Arguments["target"].(string) + if source == "" || target == "" { + return "source 和 target 不能为空" + } + count, err := a.memory.MergeEntities(source, target) + if err != nil { + return fmt.Sprintf("合并失败: %v", err) + } + return fmt.Sprintf("已将「%s」合并到「%s」，%d 条关系已重定向", source, target, count) default: return fmt.Sprintf("未知的记忆工具: %s", tc.Name) @@ -600,12 +748,45 @@ func (a *Agent) buildToolDefs() []interface{} { } } + // 插件注册的工具（通过 SDK RegisterTool） + if a.stageHost != nil { + for _, td := range a.stageHost.GetToolDefs() { + tools = append(tools, map[string]interface{}{ + "type": "function", + "function": map[string]interface{}{ + "name": td.Name, + "description": td.Description, + "parameters": td.Parameters, + }, + }) + } + } + if a.indexer != nil { for _, td := range a.indexer.GetToolDefinitions() { tools = append(tools, td) } } + // 实体合并工具（心跳检测到冲突时 LLM 使用） + if a.memory != nil { + tools = append(tools, map[string]interface{}{ + "type": "function", + "function": map[string]interface{}{ + "name": "memory_merge", + "description": "合并两个同义实体：将所有关系从 source 重定向到 target，source 标记为 merged。仅在有明确证据时使用。", + "parameters": map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "source": map[string]interface{}{"type": "string", "description": "被合并的实体名（合并后消失）"}, + "target": map[string]interface{}{"type": "string", "description": "保留的实体名"}, + }, + "required": []string{"source", "target"}, + }, + }, + }) + } + // 知识库工具 if a.knowledge != nil { tools = append(tools, map[string]interface{}{ @@ -709,6 +890,25 @@ func (a *Agent) buildToolDefs() []interface{} { }) } + // 子任务工具 + tools = append(tools, map[string]interface{}{ + "type": "function", + "function": map[string]interface{}{ + "name": "spawn_child", + "description": "创建一个子 Agent 执行独立任务。子 Agent 使用传统上下文（无持久记忆），任务完成即销毁。适用于需要多步推理但不需要写入长期记忆的场景，例如：计算、分析、生成报告草稿等。", + "parameters": map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "task": map[string]interface{}{ + "type": "string", + "description": "要子 Agent 完成的任务描述。请描述清晰、完整，包含所有必要背景。", + }, + }, + "required": []string{"task"}, + }, + }, + }) + // 输出通道工具 tools = append(tools, map[string]interface{}{ "type": "function", @@ -764,6 +964,20 @@ func (a *Agent) buildToolDefs() []interface{} { return tools } +// ConsolidationTask 心跳检测到的记忆整理任务，通过 IO 发送给 Agent 让 LLM 决策 +type ConsolidationTask struct { + Type string `json:"type"` // "entity_merge", "relation_conflict", "doc_archival" + Reason string `json:"reason"` // 人类可读的描述 + Data interface{} `json:"data"` // 任务相关数据 +} + +// enqueueConsolidationTask 将记忆整理任务注入到 Agent 输入队列 +func (a *Agent) enqueueConsolidationTask(task ConsolidationTask) { + msg := fmt.Sprintf("【记忆整理任务】\n类型: %s\n说明: %s", task.Type, task.Reason) + a.io.InjectTextTo("system", "_consolidation_", msg) + log.Printf("[agent] enqueued consolidation task: %s", task.Reason) +} + // distillLoop — 定期心跳：上下文→文档 + 图→文档 + 图重整 func (a *Agent) distillLoop() { if a.docStore == nil && a.memory == nil { @@ -896,34 +1110,85 @@ func (a *Agent) reorgGraph() { } } - // 4. 实体向量同义合并 + // 4. 实体同义冲突检测 → 交由 LLM 决策 result, err := a.memory.Recall(nil, nil, 1, "") if err != nil || result == nil || len(result.Entities) < 2 { return } - merged := 0 + candidates := 0 for i := 0; i < len(result.Entities); i++ { for j := i + 1; j < len(result.Entities); j++ { - if isSimilarName(result.Entities[i].Name, result.Entities[j].Name) { - if result.Entities[i].MentionCount >= result.Entities[j].MentionCount { - log.Printf("[agent] reorg: merging '%s' → '%s'", result.Entities[j].Name, result.Entities[i].Name) - } else { - log.Printf("[agent] reorg: merging '%s' → '%s'", result.Entities[i].Name, result.Entities[j].Name) - } - merged++ + sim := entitySimilarity(result.Entities[i].Name, result.Entities[j].Name) + if sim > 0.5 { + candidates++ + a.enqueueConsolidationTask(ConsolidationTask{ + Type: "entity_merge", + Reason: fmt.Sprintf( + "实体「%s」(类型:%s, 提及%d次) 与「%s」(类型:%s, 提及%d次) 相似度 %.0f%%，可能指代同一事物，请判断是否需要合并", + result.Entities[i].Name, result.Entities[i].Type, result.Entities[i].MentionCount, + result.Entities[j].Name, result.Entities[j].Type, result.Entities[j].MentionCount, + sim*100, + ), + Data: map[string]interface{}{ + "entity_a": result.Entities[i].Name, + "entity_a_type": result.Entities[i].Type, + "entity_a_mentions": result.Entities[i].MentionCount, + "entity_b": result.Entities[j].Name, + "entity_b_type": result.Entities[j].Type, + "entity_b_mentions": result.Entities[j].MentionCount, + "similarity": sim, + }, + }) } } } - if merged > 0 { - log.Printf("[agent] graph reorg: merged %d similar entities", merged) + if candidates > 0 { + log.Printf("[agent] graph reorg: %d merge candidates sent for LLM decision", candidates) } else { - log.Printf("[agent] graph reorg: no merges needed") + log.Printf("[agent] graph reorg: no similar entities found") } } -// isSimilarName — 使用字符 bigram Jaccard 相似度判断实体名是否同义 +// entitySimilarity 计算两个实体名的相似度（字符 bigram Jaccard） +func entitySimilarity(a, b string) float64 { + if a == "" || b == "" { + return 0 + } + if a == b { + return 1.0 + } + runesA, runesB := []rune(a), []rune(b) + if len(runesA) < 2 || len(runesB) < 2 { + if len(runesA) == len(runesB) && len(runesA) == 1 { + if runesA[0] == runesB[0] { + return 1.0 + } + } + return 0 + } + + setA := make(map[string]bool) + for i := 0; i < len(runesA)-1; i++ { + setA[string(runesA[i:i+2])] = true + } + + intersect := 0 + for i := 0; i < len(runesB)-1; i++ { + if setA[string(runesB[i:i+2])] { + intersect++ + } + } + + union := len(setA) + len(runesB) - 1 - intersect + if union <= 0 { + return 0 + } + + return float64(intersect) / float64(union) +} + // docToTriples 将文档转为图记忆三元组 func docToTriples(doc *document.Doc) []memory.Triple { var triples []memory.Triple @@ -964,36 +1229,6 @@ func docToTriples(doc *document.Doc) []memory.Triple { return triples } -func isSimilarName(a, b string) bool { - if a == b { - return false // 自带跳过 - } - runesA, runesB := []rune(a), []rune(b) - if len(runesA) < 2 || len(runesB) < 2 { - return false - } - - setA := make(map[string]bool) - for i := 0; i < len(runesA)-1; i++ { - setA[string(runesA[i:i+2])] = true - } - - intersect := 0 - for i := 0; i < len(runesB)-1; i++ { - if setA[string(runesB[i:i+2])] { - intersect++ - } - } - - union := len(setA) + len(runesB) - 1 - intersect - if union <= 0 { - return false - } - - jaccard := float64(intersect) / float64(union) - return jaccard > 0.5 -} - func (a *Agent) emitMemoryCandidate(source, input, response string, toolsUsed []string) { a.io.EmitOutput("memory", "memory_candidate", map[string]interface{}{ "source": source, @@ -1007,6 +1242,33 @@ func (a *Agent) emitMemoryCandidate(source, input, response string, toolsUsed [] // executeOutputChannelTool — AI 切换当前请求的输出通道 // 在 process() 内调用，mutex 保护，只有一个请求在执行 +// processConsolidation 处理后台记忆整理任务（不发外部输出） +func (a *Agent) processConsolidation(input string) { + start := time.Now() + a.currentOutputChannel = "_consolidation_" + a.context.Append(ContextEvent{ + Timestamp: start, + Source: "system", + Input: input, + }) + response, toolsUsed, err := a.process(input, &sdk.StageContext{RawMessage: input}) + if err != nil { + log.Printf("[agent] consolidation error: %v", err) + return + } + a.context.Append(ContextEvent{ + Timestamp: time.Now(), + Source: "agent", + Input: input, + Response: response, + ToolsUsed: toolsUsed, + }) + _ = a.context.Prune(response, a.maxContextSize, a.docStore) + // 只写入记忆，不发外部输出 + a.emitMemoryCandidate("system", input, response, toolsUsed) + log.Printf("[agent] consolidation done (%dms, tools=%v)", time.Since(start).Milliseconds(), toolsUsed) +} + func (a *Agent) executeOutputChannelTool(tc agentAPI.ToolCall) string { channel, _ := tc.Arguments["channel"].(string) if channel == "" { @@ -1077,6 +1339,118 @@ func (a *Agent) executePluginReload() string { return msg } +// executeSpawnChild 创建子 Agent 执行独立任务 +// 子 Agent 使用传统上下文（单轮对话），无持久记忆，任务完即销毁 +func (a *Agent) executeSpawnChild(tc agentAPI.ToolCall) string { + task, _ := tc.Arguments["task"].(string) + if task == "" { + return "请提供 task 参数" + } + + sysPrompt := fmt.Sprintf(`你是 HomeAgent 的子任务助手。 +请完成以下任务。完成即可，无需保留记忆或查询历史。 +任务: %s`, task) + + msgs := []agentAPI.Message{ + {Role: "system", Content: sysPrompt}, + {Role: "user", Content: task}, + } + + // 子 Agent 无特殊工具，只保留基础 tool 定义（无记忆/知识/文档工具） + childTools := []interface{}{ + map[string]interface{}{ + "type": "function", + "function": map[string]interface{}{ + "name": "output_send", + "description": "通过指定输出通道发送消息", + "parameters": map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "channel": map[string]interface{}{"type": "string", "description": "输出通道"}, + "content": map[string]interface{}{"type": "string", "description": "消息内容"}, + }, + "required": []string{"channel", "content"}, + }, + }, + }, + } + + for turn := 0; turn < 5; turn++ { + req := &agentAPI.CompletionRequest{ + Messages: msgs, + MaxTokens: 4096, + Tools: childTools, + ToolChoice: "auto", + ExtraBody: map[string]interface{}{ + "thinking": map[string]interface{}{"type": "disabled"}, + }, + } + + resp, err := a.provider.Chat(a.ctx, req) + if err != nil { + return fmt.Sprintf("子 Agent 执行失败: %v", err) + } + + if len(resp.ToolCalls) == 0 { + return resp.Content + } + + for _, ct := range resp.ToolCalls { + var result string + if ct.Name == "output_send" { + channel, _ := ct.Arguments["channel"].(string) + content, _ := ct.Arguments["content"].(string) + if channel != "" && content != "" { + a.io.EmitTextTo("child_agent", channel, content) + result = fmt.Sprintf("已通过 [%s] 通道发送", channel) + } else { + result = "channel 和 content 不能为空" + } + } else { + result = fmt.Sprintf("子 Agent 无法调用工具 %s", ct.Name) + } + msgs = append(msgs, agentAPI.Message{Role: "assistant", Content: resp.Content, ToolCalls: []agentAPI.ToolCall{ct}}) + msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: ct.ID, Content: result}) + } + } + + return "子 Agent 执行超时（超过 5 轮）" +} + +// runStage — 运行阶段管道，若插件 Response 被设置则返回 true（短路） +func (a *Agent) runStage(stage sdk.Stage, ctx *sdk.StageContext) bool { + if a.stageHost == nil { + return false + } + ctx.Phase = stage + a.stageHost.RunStage(stage, ctx) + return ctx.Response != nil +} + +// publishEvent — 发布系统事件 +func (a *Agent) publishEvent(evtType events.EventType, payload map[string]interface{}) { + if a.eventBus == nil { + return + } + a.eventBus.Publish(&events.Event{ + Type: evtType, + Source: string(a.id), + Payload: payload, + Timestamp: time.Now().Unix(), + }) +} + +// stageCtxFromInput — 根据输入构建阶段上下文 +func (a *Agent) stageCtxFromInput(input, userID, groupID string) *sdk.StageContext { + return &sdk.StageContext{ + RawMessage: input, + UserID: userID, + GroupID: groupID, + Phase: sdk.StageOnInput, + Extra: make(map[string]interface{}), + } +} + func getFloat(m map[string]interface{}, key string) float64 { if v, ok := m[key]; ok { switch n := v.(type) { diff --git a/internal/agent/core/agent_functions_test.go b/internal/agent/core/agent_functions_test.go index 55cf4d6..0765020 100644 --- a/internal/agent/core/agent_functions_test.go +++ b/internal/agent/core/agent_functions_test.go @@ -6,21 +6,21 @@ import ( "gitcode.com/JianFeeeee/HomeAgent/internal/memory/document" ) -func TestIsSimilarName(t *testing.T) { +func TestEntitySimilarity(t *testing.T) { tests := []struct { a, b string - want bool + want float64 }{ - {"张三", "张三四", false}, - {"", "", false}, - {"a", "b", false}, - {"张三", "李四", false}, - {"张三", "张三", false}, + {"", "", 0}, // empty → 0 + {"a", "b", 0}, // single char → 0 + {"张三", "张三", 1.0}, // identical → 1.0 + {"张三", "李四", 0}, // no common bigrams + {"iPhone", "iPhone 15", 0.625}, // partial overlap } for _, tt := range tests { - got := isSimilarName(tt.a, tt.b) + got := entitySimilarity(tt.a, tt.b) if got != tt.want { - t.Errorf("isSimilarName(%q, %q) = %v, want %v", tt.a, tt.b, got, tt.want) + t.Errorf("entitySimilarity(%q, %q) = %.3f, want %.3f", tt.a, tt.b, got, tt.want) } } } diff --git a/internal/agent/core/context.go b/internal/agent/core/context.go index ef87754..1324f53 100644 --- a/internal/agent/core/context.go +++ b/internal/agent/core/context.go @@ -1,7 +1,10 @@ package core import ( + "encoding/json" "fmt" + "os" + "path/filepath" "sort" "strings" "sync" @@ -13,26 +16,63 @@ import ( // ContextEvent — 单条上下文事件 type ContextEvent struct { - Timestamp time.Time `json:"timestamp"` - Source string `json:"source"` - Input string `json:"input"` - Response string `json:"response,omitempty"` - ToolsUsed []string `json:"tools_used,omitempty"` - Vector vector.Vector `json:"-"` // 缓存向量，避免重复计算 + Timestamp time.Time `json:"timestamp"` + Source string `json:"source"` + Input string `json:"input"` + Response string `json:"response,omitempty"` + ToolsUsed []string `json:"tools_used,omitempty"` + Vector vector.Vector `json:"-"` // 缓存向量，避免重复计算 } // RelevanceContext — 基于相关性的上下文管理，非固定阈值 type RelevanceContext struct { - mu sync.Mutex - events []*ContextEvent - veczer *vector.TFIDFVectorizer - trained bool + mu sync.Mutex + events []*ContextEvent + veczer *vector.TFIDFVectorizer + trained bool + savePath string // 持久化路径，空则不持久化 } -func NewRelevanceContext() *RelevanceContext { - return &RelevanceContext{ - veczer: vector.NewTFIDFVectorizer(2), +func NewRelevanceContext(savePath string) *RelevanceContext { + rc := &RelevanceContext{ + veczer: vector.NewTFIDFVectorizer(2), + savePath: savePath, } + if savePath != "" { + rc.load() + } + return rc +} + +// load 从文件恢复上下文事件 +func (c *RelevanceContext) load() { + data, err := os.ReadFile(c.savePath) + if err != nil { + return + } + var events []*ContextEvent + if err := json.Unmarshal(data, &events); err != nil { + return + } + for _, evt := range events { + evt.Vector = c.veczer.Vectorize(evt.Input + " " + evt.Response) + } + c.events = events +} + +// Save 持久化上下文事件到文件 +func (c *RelevanceContext) Save() error { + if c.savePath == "" { + return nil + } + if err := os.MkdirAll(filepath.Dir(c.savePath), 0755); err != nil { + return err + } + data, err := json.Marshal(c.events) + if err != nil { + return err + } + return os.WriteFile(c.savePath, data, 0644) } func (c *RelevanceContext) Append(evt ContextEvent) { @@ -44,6 +84,20 @@ func (c *RelevanceContext) Append(evt ContextEvent) { // 增量训练向量化器 c.trained = false + + c.save() +} + +// save 无锁版本，Append/Prune 内部持有锁时调用 +func (c *RelevanceContext) save() error { + if c.savePath == "" { + return nil + } + data, err := json.Marshal(c.events) + if err != nil { + return err + } + return os.WriteFile(c.savePath, data, 0644) } // Prune — 基于当前输入计算每条上下文的相关性，归档最不相关的 @@ -114,6 +168,8 @@ func (c *RelevanceContext) Prune(currentInput string, topK int, docStore *docume } } + c.save() + return archived } diff --git a/internal/agent/core/context_test.go b/internal/agent/core/context_test.go index 567d8e1..f8a139e 100644 --- a/internal/agent/core/context_test.go +++ b/internal/agent/core/context_test.go @@ -6,7 +6,7 @@ import ( ) func TestContextAppendAndLen(t *testing.T) { - ctx := NewRelevanceContext() + ctx := NewRelevanceContext("") if ctx.Len() != 0 { t.Errorf("new context should be empty, got %d", ctx.Len()) } @@ -18,7 +18,7 @@ func TestContextAppendAndLen(t *testing.T) { } func TestContextRecent(t *testing.T) { - ctx := NewRelevanceContext() + ctx := NewRelevanceContext("") ctx.Append(ContextEvent{Timestamp: time.Now(), Source: "user", Input: "a"}) ctx.Append(ContextEvent{Timestamp: time.Now(), Source: "user", Input: "b"}) ctx.Append(ContextEvent{Timestamp: time.Now(), Source: "user", Input: "c"}) @@ -33,7 +33,7 @@ func TestContextRecent(t *testing.T) { } func TestContextFormat(t *testing.T) { - ctx := NewRelevanceContext() + ctx := NewRelevanceContext("") f := ctx.Format() if f != "" { t.Errorf("empty context should format to empty string, got %q", f) @@ -54,7 +54,7 @@ func TestContextFormat(t *testing.T) { } func TestContextPruneKeepsTopK(t *testing.T) { - ctx := NewRelevanceContext() + ctx := NewRelevanceContext("") for i := 0; i < 10; i++ { ctx.Append(ContextEvent{ Timestamp: time.Now(), @@ -80,7 +80,7 @@ func TestContextPruneKeepsTopK(t *testing.T) { } func TestContextPruneWithDocStore(t *testing.T) { - ctx := NewRelevanceContext() + ctx := NewRelevanceContext("") for i := 0; i < 15; i++ { ctx.Append(ContextEvent{ Timestamp: time.Now(), @@ -97,7 +97,7 @@ func TestContextPruneWithDocStore(t *testing.T) { } func TestContextAppendAfterPrune(t *testing.T) { - ctx := NewRelevanceContext() + ctx := NewRelevanceContext("") for i := 0; i < 10; i++ { ctx.Append(ContextEvent{ Timestamp: time.Now(), diff --git a/internal/agent/core/stages.go b/internal/agent/core/stages.go new file mode 100644 index 0000000..6b5d391 --- /dev/null +++ b/internal/agent/core/stages.go @@ -0,0 +1,74 @@ +package core + +import ( + "fmt" + + "gitcode.com/JianFeeeee/HomeAgent/internal/plugin" + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk" +) + +type StageHost struct { + plugins []*sdk.PluginAPI + toolDefs []sdk.ToolDef + tools map[string]sdk.ToolHandler +} + +func NewStageHost() *StageHost { + return &StageHost{ + tools: make(map[string]sdk.ToolHandler), + } +} + +func (h *StageHost) RegisterPlugin(api *sdk.PluginAPI) { + h.plugins = append(h.plugins, api) + for name, handler := range api.Tools() { + h.tools[name] = handler + h.toolDefs = append(h.toolDefs, sdk.ToolDef{Name: name}) + } +} + +// SyncFromRegistry 从插件注册表同步 SDK 插件 +func (h *StageHost) SyncFromRegistry(reg *plugin.Registry) { + if reg == nil { + return + } + for _, td := range reg.GetAllSDKToolDefs() { + h.toolDefs = append(h.toolDefs, td) + } +} + +func (h *StageHost) GetToolDefs() []sdk.ToolDef { + return h.toolDefs +} + +func (h *StageHost) ExecuteTool(name string, args map[string]interface{}) (interface{}, error) { + if handler, ok := h.tools[name]; ok { + return handler(args) + } + return nil, fmt.Errorf("tool %s not found in any plugin", name) +} + +func (h *StageHost) RunStage(stage sdk.Stage, ctx *sdk.StageContext) { + for _, p := range h.plugins { + for _, handler := range p.StageHandlers(stage) { + if err := handler(ctx); err != nil { + return + } + if ctx.Response != nil { + return + } + } + } +} + +func (h *StageHost) RunStageAll(stage sdk.Stage, ctx *sdk.StageContext) { + for _, p := range h.plugins { + for _, handler := range p.StageHandlers(stage) { + handler(ctx) + } + } +} + +func (h *StageHost) PluginCount() int { + return len(h.plugins) +} diff --git a/internal/agent/core/stages_test.go b/internal/agent/core/stages_test.go new file mode 100644 index 0000000..7eb6227 --- /dev/null +++ b/internal/agent/core/stages_test.go @@ -0,0 +1,183 @@ +package core + +import ( + "testing" + + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk" +) + +func TestStageHostRegisterPlugin(t *testing.T) { + host := NewStageHost() + api := sdk.NewPluginAPI("test", "1.0.0", nil, nil, nil) + + api.RegisterTool("test_tool", func(args map[string]interface{}) (interface{}, error) { + return "ok", nil + }) + + host.RegisterPlugin(api) + + if host.PluginCount() != 1 { + t.Errorf("expected 1 plugin, got %d", host.PluginCount()) + } + + defs := host.GetToolDefs() + if len(defs) != 1 { + t.Errorf("expected 1 tool def, got %d", len(defs)) + } + if defs[0].Name != "test_tool" { + t.Errorf("expected test_tool, got %s", defs[0].Name) + } +} + +func TestStageHostExecuteTool(t *testing.T) { + host := NewStageHost() + api := sdk.NewPluginAPI("test", "1.0.0", nil, nil, nil) + + api.RegisterTool("hello", func(args map[string]interface{}) (interface{}, error) { + return "world", nil + }) + + host.RegisterPlugin(api) + + result, err := host.ExecuteTool("hello", nil) + if err != nil { + t.Fatalf("execute: %v", err) + } + if result.(string) != "world" { + t.Errorf("expected world, got %v", result) + } + + _, err = host.ExecuteTool("nonexistent", nil) + if err == nil { + t.Error("expected error for nonexistent tool") + } +} + +func TestStageHostRunStage(t *testing.T) { + host := NewStageHost() + api := sdk.NewPluginAPI("test", "1.0.0", nil, nil, nil) + + var called bool + api.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error { + called = true + return nil + }) + + host.RegisterPlugin(api) + + ctx := &sdk.StageContext{RawMessage: "hello"} + host.RunStage(sdk.StageOnInput, ctx) + + if !called { + t.Error("stage handler was not called") + } +} + +func TestStageHostRunStageShortCircuit(t *testing.T) { + host := NewStageHost() + + api1 := sdk.NewPluginAPI("p1", "1.0.0", nil, nil, nil) + api1.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error { + resp := "short-circuited" + ctx.Response = &resp + return nil + }) + + var api2called bool + api2 := sdk.NewPluginAPI("p2", "1.0.0", nil, nil, nil) + api2.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error { + api2called = true + return nil + }) + + host.RegisterPlugin(api1) + host.RegisterPlugin(api2) + + ctx := &sdk.StageContext{RawMessage: "hello"} + host.RunStage(sdk.StageOnInput, ctx) + + if ctx.Response == nil || *ctx.Response != "short-circuited" { + t.Errorf("expected short-circuited, got %v", ctx.Response) + } + if api2called { + t.Error("api2 should not have been called after short circuit") + } +} + +func TestStageHostRunStageAll(t *testing.T) { + host := NewStageHost() + + count := 0 + api1 := sdk.NewPluginAPI("p1", "1.0.0", nil, nil, nil) + api1.RegisterStage(sdk.StageAfterOutput, func(ctx *sdk.StageContext) error { + count++ + return nil + }) + + api2 := sdk.NewPluginAPI("p2", "1.0.0", nil, nil, nil) + api2.RegisterStage(sdk.StageAfterOutput, func(ctx *sdk.StageContext) error { + count++ + return nil + }) + + host.RegisterPlugin(api1) + host.RegisterPlugin(api2) + + host.RunStageAll(sdk.StageAfterOutput, &sdk.StageContext{}) + + if count != 2 { + t.Errorf("expected 2 handlers called, got %d", count) + } +} + +func TestStageHostMultiplePlugins(t *testing.T) { + host := NewStageHost() + + p1 := sdk.NewPluginAPI("p1", "1.0.0", nil, nil, nil) + p1.RegisterTool("tool1", func(args map[string]interface{}) (interface{}, error) { + return "from_p1", nil + }) + + p2 := sdk.NewPluginAPI("p2", "1.0.0", nil, nil, nil) + p2.RegisterTool("tool2", func(args map[string]interface{}) (interface{}, error) { + return "from_p2", nil + }) + + host.RegisterPlugin(p1) + host.RegisterPlugin(p2) + + if host.PluginCount() != 2 { + t.Errorf("expected 2 plugins, got %d", host.PluginCount()) + } + + r1, _ := host.ExecuteTool("tool1", nil) + if r1.(string) != "from_p1" { + t.Errorf("expected from_p1, got %v", r1) + } + + r2, _ := host.ExecuteTool("tool2", nil) + if r2.(string) != "from_p2" { + t.Errorf("expected from_p2, got %v", r2) + } +} + +func TestStageHostEmpty(t *testing.T) { + host := NewStageHost() + + if host.PluginCount() != 0 { + t.Errorf("expected 0 plugins, got %d", host.PluginCount()) + } + + defs := host.GetToolDefs() + if len(defs) != 0 { + t.Errorf("expected 0 tool defs, got %d", len(defs)) + } + + _, err := host.ExecuteTool("anything", nil) + if err == nil { + t.Error("expected error on empty host") + } + + // RunStage on empty host should not panic + host.RunStage(sdk.StageOnInput, &sdk.StageContext{}) +} diff --git a/internal/api/plugin.go b/internal/api/plugin.go new file mode 100644 index 0000000..6e7c3a9 --- /dev/null +++ b/internal/api/plugin.go @@ -0,0 +1,68 @@ +package api + +import ( + "log" + "net/http" + + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" + "gitcode.com/JianFeeeee/HomeAgent/internal/knowledge" + luaVM "gitcode.com/JianFeeeee/HomeAgent/internal/lua" + "gitcode.com/JianFeeeee/HomeAgent/internal/memory" + "gitcode.com/JianFeeeee/HomeAgent/internal/memory/text" + "gitcode.com/JianFeeeee/HomeAgent/internal/skill" + "gitcode.com/JianFeeeee/HomeAgent/internal/supervisor" + "gitcode.com/JianFeeeee/HomeAgent/internal/tracker" + "gitcode.com/JianFeeeee/HomeAgent/pkg/types" +) + +// Plugin 将 HTTP API + WebUI 包装为 IO Device +// 作为 HomeAgent 自带的默认 IO 通道插件 +type Plugin struct { + name string + handler *Handler + server *http.Server + addr string + mux *http.ServeMux +} + +func NewWebUIPlugin(name, addr string, sup *supervisor.Daemon, mem *memory.GraphDB, sk *skill.Manager, + lua *luaVM.VM, cfg *types.Config, iom *agentIO.IOManager, tm *text.Memory, ks *knowledge.Store, tr *tracker.Tracker) *Plugin { + + h := NewHandler(sup, mem, sk, lua, cfg, iom, tm, ks, tr) + mux := http.NewServeMux() + h.RegisterRoutes(mux) + + return &Plugin{ + name: name, + handler: h, + addr: addr, + mux: mux, + } +} + +func (p *Plugin) Name() string { return p.name } +func (p *Plugin) Type() agentIO.DeviceType { return agentIO.DeviceIO } +func (p *Plugin) Description() string { return "HTTP API & Web Dashboard" } +func (p *Plugin) OutputCapabilities() agentIO.OutputCapability { return agentIO.CapText | agentIO.CapStructured } +func (p *Plugin) Tools() []agentIO.ToolDef { return nil } +func (p *Plugin) Execute(tool string, args map[string]interface{}) (interface{}, error) { + return nil, nil +} + +func (p *Plugin) Start() error { + p.server = &http.Server{Addr: p.addr, Handler: p.mux} + go func() { + log.Printf("[webui] HTTP server listening on %s", p.addr) + if err := p.server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Printf("[webui] server error: %v", err) + } + }() + return nil +} + +func (p *Plugin) Stop() error { + if p.server != nil { + return p.server.Close() + } + return nil +} diff --git a/internal/events/bus.go b/internal/events/bus.go new file mode 100644 index 0000000..635cf5e --- /dev/null +++ b/internal/events/bus.go @@ -0,0 +1,69 @@ +package events + +import ( + "fmt" + "sync" +) + +type EventType string + +const ( + EventRawInput EventType = "raw_input" + EventAgentOutput EventType = "agent_output" + EventToolCall EventType = "tool_call" + EventReasoning EventType = "reasoning" + EventSystem EventType = "system" + EventAll EventType = "*" +) + +type Event struct { + Type EventType `json:"type"` + Source string `json:"source"` + Payload map[string]interface{} `json:"payload"` + Timestamp int64 `json:"timestamp"` +} + +type Handler func(event *Event) + +type Bus struct { + mu sync.RWMutex + subs map[EventType][]Handler +} + +func NewBus() *Bus { + return &Bus{ + subs: make(map[EventType][]Handler), + } +} + +func (b *Bus) Publish(evt *Event) { + b.mu.RLock() + allHandlers := b.subs[EventAll] + typeHandlers := b.subs[evt.Type] + b.mu.RUnlock() + + for _, h := range allHandlers { + h(evt) + } + for _, h := range typeHandlers { + h(evt) + } +} + +func (b *Bus) Subscribe(eventType EventType, handler Handler) func() { + b.mu.Lock() + b.subs[eventType] = append(b.subs[eventType], handler) + b.mu.Unlock() + + return func() { + b.mu.Lock() + defer b.mu.Unlock() + list := b.subs[eventType] + for i, h := range list { + if fmt.Sprintf("%p", h) == fmt.Sprintf("%p", handler) { + b.subs[eventType] = append(list[:i], list[i+1:]...) + break + } + } + } +} diff --git a/internal/events/bus_test.go b/internal/events/bus_test.go new file mode 100644 index 0000000..e7adb8f --- /dev/null +++ b/internal/events/bus_test.go @@ -0,0 +1,102 @@ +package events + +import ( + "sync/atomic" + "testing" + "time" +) + +func TestBusPublishSubscribe(t *testing.T) { + bus := NewBus() + var count int32 + + bus.Subscribe(EventRawInput, func(evt *Event) { + atomic.AddInt32(&count, 1) + }) + + bus.Publish(&Event{ + Type: EventRawInput, + Source: "test", + Payload: map[string]interface{}{"content": "hello"}, + }) + + if c := atomic.LoadInt32(&count); c != 1 { + t.Errorf("expected 1, got %d", c) + } +} + +func TestBusWildcard(t *testing.T) { + bus := NewBus() + var count int32 + + bus.Subscribe(EventAll, func(evt *Event) { + atomic.AddInt32(&count, 1) + }) + + bus.Publish(&Event{Type: EventRawInput, Source: "test"}) + bus.Publish(&Event{Type: EventToolCall, Source: "test"}) + + if c := atomic.LoadInt32(&count); c != 2 { + t.Errorf("expected 2, got %d", c) + } +} + +func TestBusUnsubscribe(t *testing.T) { + bus := NewBus() + var count int32 + + handler := func(evt *Event) { + atomic.AddInt32(&count, 1) + } + unsub := bus.Subscribe(EventRawInput, handler) + + bus.Publish(&Event{Type: EventRawInput, Source: "test"}) + unsub() + bus.Publish(&Event{Type: EventRawInput, Source: "test"}) + + if c := atomic.LoadInt32(&count); c != 1 { + t.Errorf("expected 1 after unsub, got %d", c) + } +} + +func TestBusNoMatch(t *testing.T) { + bus := NewBus() + var count int32 + + bus.Subscribe(EventRawInput, func(evt *Event) { + atomic.AddInt32(&count, 1) + }) + + bus.Publish(&Event{Type: EventAgentOutput, Source: "test"}) + + if c := atomic.LoadInt32(&count); c != 0 { + t.Errorf("expected 0, got %d", c) + } +} + +func TestBusConcurrent(t *testing.T) { + bus := NewBus() + var count int32 + + bus.Subscribe(EventAll, func(evt *Event) { + atomic.AddInt32(&count, 1) + }) + + done := make(chan struct{}) + go func() { + for i := 0; i < 100; i++ { + bus.Publish(&Event{Type: EventRawInput, Source: "test"}) + } + close(done) + }() + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("timeout") + } + + if c := atomic.LoadInt32(&count); c != 100 { + t.Errorf("expected 100, got %d", c) + } +} diff --git a/internal/memory/graph.go b/internal/memory/graph.go index 3825030..6ede294 100644 --- a/internal/memory/graph.go +++ b/internal/memory/graph.go @@ -520,6 +520,98 @@ func (g *GraphDB) Introspect() (map[string]interface{}, error) { }, nil } +// MergeEntities 合并两个实体：将 sourceName 的所有信息合并到 targetName +// 1. sourceName 的所有关系重新指向 targetName +// 2. targetName 的 mention_count 增加 sourceName 的计数 +// 3. sourceName 标记为 merged +// 返回 (关系的重定向数, error) +func (g *GraphDB) MergeEntities(sourceName, targetName string) (int, error) { + g.mu.Lock() + defer g.mu.Unlock() + + tx, err := g.db.Begin() + if err != nil { + return 0, err + } + defer tx.Rollback() + + var sourceID, targetID int64 + var sourceCount, targetCount int + + err = tx.QueryRow("SELECT id, mention_count FROM entities WHERE name = ?", sourceName).Scan(&sourceID, &sourceCount) + if err != nil { + return 0, fmt.Errorf("source entity '%s' not found: %w", sourceName, err) + } + err = tx.QueryRow("SELECT id, mention_count FROM entities WHERE name = ?", targetName).Scan(&targetID, &targetCount) + if err != nil { + return 0, fmt.Errorf("target entity '%s' not found: %w", targetName, err) + } + + if sourceID == targetID { + return 0, fmt.Errorf("cannot merge entity with itself") + } + + // 重定向 source → target 的关系（作为 source） + res, err := tx.Exec( + `UPDATE relations SET source_id = ?, updated_at = CURRENT_TIMESTAMP + WHERE source_id = ? AND status = 'active'`, + targetID, sourceID, + ) + if err != nil { + return 0, err + } + redirectedSource, _ := res.RowsAffected() + + // 重定向 source → target 的关系（作为 target） + res, err = tx.Exec( + `UPDATE relations SET target_id = ?, updated_at = CURRENT_TIMESTAMP + WHERE target_id = ? AND status = 'active'`, + targetID, sourceID, + ) + if err != nil { + return 0, err + } + redirectedTarget, _ := res.RowsAffected() + + // 删除可能产生的自引用关系 + _, err = tx.Exec( + `DELETE FROM relations + WHERE source_id = target_id AND source_id = ?`, + targetID, + ) + if err != nil { + return 0, err + } + + // 更新 target 的 mention_count + _, err = tx.Exec( + `UPDATE entities SET mention_count = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?`, + targetCount+sourceCount, targetID, + ) + if err != nil { + return 0, err + } + + // 标记 source 为 merged（改名避免 UNIQUE 冲突） + _, err = tx.Exec( + `UPDATE entities SET name = ? || '@merged_' || ?, + mention_count = 0, + updated_at = CURRENT_TIMESTAMP + WHERE id = ?`, + sourceName, time.Now().Format("20060102150405"), sourceID, + ) + if err != nil { + return 0, err + } + + if err := tx.Commit(); err != nil { + return 0, err + } + + total := int(redirectedSource + redirectedTarget) + return total, nil +} + func (g *GraphDB) Archive(days int) (int, error) { g.mu.Lock() defer g.mu.Unlock() diff --git a/internal/memory/graph_test.go b/internal/memory/graph_test.go index 4283755..6512d5b 100644 --- a/internal/memory/graph_test.go +++ b/internal/memory/graph_test.go @@ -214,6 +214,95 @@ func TestIntrospectHotspots(t *testing.T) { } } +func TestMergeEntities(t *testing.T) { + g := newTestGraph(t) + defer os.Remove(g.dbPath) + defer g.Close() + + g.Commit([]Triple{ + {Subject: "张三", Relation: "喜欢", Object: "编程"}, + {Subject: "张三", Relation: "居住", Object: "北京"}, + }, "session", 0) + + g.Commit([]Triple{ + {Subject: "张先生", Relation: "工作", Object: "字节跳动"}, + }, "session", 0) + + // 合并前：两个实体各有关联 + stats, _ := g.Introspect() + if stats["entity_count"].(int) != 5 { + t.Fatalf("expected 5 entities (张三, 编程, 北京, 张先生, 字节跳动), got %d", stats["entity_count"]) + } + + // 先增加张先生的 mention_count + g.Commit([]Triple{ + {Subject: "张先生", Relation: "喜欢", Object: "Go"}, + }, "session", 0) + + n, err := g.MergeEntities("张先生", "张三") + if err != nil { + t.Fatal(err) + } + if n < 2 { + t.Errorf("expected at least 2 redirected relations, got %d", n) + } + + // source 应被改名 + result, err := g.Recall(nil, []string{"张先生"}, 1, "") + if err != nil { + t.Fatal(err) + } + if len(result.Entities) > 0 { + t.Error("张先生 should be merged and hidden") + } + + // target 的 mention_count 应合并 + // 验证 target 还存在（seedEntities 精确查找） + result2, err := g.Recall([]string{"张三"}, nil, 1, "") + if err != nil { + t.Fatal(err) + } + found := false + for _, e := range result2.Entities { + if e.Name == "张三" { + found = true + if e.MentionCount < 2 { + t.Errorf("expected 张三 mention_count >= 2 after merge, got %d", e.MentionCount) + } + break + } + } + if !found { + t.Error("张三 should still exist after merge") + } +} + +func TestMergeEntitiesSelf(t *testing.T) { + g := newTestGraph(t) + defer os.Remove(g.dbPath) + defer g.Close() + + g.Commit([]Triple{ + {Subject: "张三", Relation: "喜欢", Object: "编程"}, + }, "session", 0) + + _, err := g.MergeEntities("张三", "张三") + if err == nil { + t.Error("expected error when merging entity with itself") + } +} + +func TestMergeEntitiesNonexistent(t *testing.T) { + g := newTestGraph(t) + defer os.Remove(g.dbPath) + defer g.Close() + + _, err := g.MergeEntities("不存在", "张三") + if err == nil { + t.Error("expected error for nonexistent source") + } +} + func TestPlaceholders(t *testing.T) { if placeholders(0) != "NULL" { t.Errorf("expected NULL for n=0, got %s", placeholders(0)) diff --git a/internal/plugin/plugin.go b/internal/plugin/plugin.go index 6b00d05..741a3e9 100644 --- a/internal/plugin/plugin.go +++ b/internal/plugin/plugin.go @@ -12,6 +12,7 @@ import ( "time" agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk" ) type PluginType string @@ -147,15 +148,90 @@ type Registry struct { plugins map[string]Plugin ioMgr *agentIO.IOManager factories map[string]NativeFactory // 名称匹配的插件使用原生实现 + sdkAPIs map[string]*sdkAPI // SDK 插件 API 实例 +} + +type sdkAPI struct { + api *sdk.PluginAPI + tools map[string]sdk.ToolHandler + stages map[sdk.Stage][]sdk.StageHandler } func NewRegistry() *Registry { return &Registry{ plugins: make(map[string]Plugin), factories: make(map[string]NativeFactory), + sdkAPIs: make(map[string]*sdkAPI), } } +// RegisterPluginAPI 注册一个 SDK 插件 API 实例 +func (r *Registry) RegisterPluginAPI(api *sdk.PluginAPI) error { + r.mu.Lock() + defer r.mu.Unlock() + if _, ok := r.sdkAPIs[api.Name]; ok { + return fmt.Errorf("sdk api %s already registered", api.Name) + } + r.sdkAPIs[api.Name] = &sdkAPI{ + api: api, + tools: api.Tools(), + stages: make(map[sdk.Stage][]sdk.StageHandler), + } + for stage := range sdk.AllStages() { + if handlers := api.StageHandlers(stage); len(handlers) > 0 { + r.sdkAPIs[api.Name].stages[stage] = handlers + } + } + log.Printf("[plugin] registered SDK plugin: %s (tools=%d, stages=%d)", + api.Name, len(api.Tools()), len(r.sdkAPIs[api.Name].stages)) + return nil +} + +// GetAllSDKToolDefs 收集所有 SDK 插件的工具定义 +func (r *Registry) GetAllSDKToolDefs() []sdk.ToolDef { + r.mu.RLock() + defer r.mu.RUnlock() + var defs []sdk.ToolDef + for _, sa := range r.sdkAPIs { + for name := range sa.tools { + defs = append(defs, sdk.ToolDef{Name: name}) + } + } + return defs +} + +// ExecuteSDKTool 执行 SDK 插件工具 +func (r *Registry) ExecuteSDKTool(name string, args map[string]interface{}) (interface{}, error) { + r.mu.RLock() + defer r.mu.RUnlock() + for _, sa := range r.sdkAPIs { + if handler, ok := sa.tools[name]; ok { + return handler(args) + } + } + return nil, fmt.Errorf("sdk tool %s not found", name) +} + +// GetStageHandlers 获取所有 SDK 插件在指定阶段的处理器 +func (r *Registry) GetStageHandlers(stage sdk.Stage) []sdk.StageHandler { + r.mu.RLock() + defer r.mu.RUnlock() + var handlers []sdk.StageHandler + for _, sa := range r.sdkAPIs { + if h, ok := sa.stages[stage]; ok { + handlers = append(handlers, h...) + } + } + return handlers +} + +// SDKPluginCount 返回已注册的 SDK 插件数量 +func (r *Registry) SDKPluginCount() int { + r.mu.RLock() + defer r.mu.RUnlock() + return len(r.sdkAPIs) +} + // RegisterNative 注册内置原生插件工厂。当从 plugins/ 加载插件时， // 如果插件名称匹配已注册的工厂，优先使用原生设备注册。 // 例如: r.RegisterNative("qq", onebot.NewDeviceFactory) diff --git a/internal/plugin/sdk/api.go b/internal/plugin/sdk/api.go new file mode 100644 index 0000000..f8d393c --- /dev/null +++ b/internal/plugin/sdk/api.go @@ -0,0 +1,169 @@ +package sdk + +import "fmt" + +type Stage string + +const ( + StageOnInput Stage = "on_input" + StagePreAction Stage = "pre_action" + StagePostAction Stage = "post_action" + StageBeforeToolcall Stage = "before_toolcall" + StageAfterToolcall Stage = "after_toolcall" + StageBeforeOutput Stage = "before_output" + StageAfterOutput Stage = "after_output" +) + +type EventType string + +const ( + EventRawInput EventType = "raw_input" + EventAgentOutput EventType = "agent_output" + EventToolCall EventType = "tool_call" + EventReasoning EventType = "reasoning" + EventSystem EventType = "system" + EventAll EventType = "*" +) + +type Event struct { + Type EventType `json:"type"` + Source string `json:"source"` + Payload map[string]interface{} `json:"payload"` + Timestamp int64 `json:"timestamp"` +} + +type MemItem struct { + Content string `json:"content"` + Score float64 `json:"score"` + Source string `json:"source"` +} + +type StageContext struct { + RawMessage string + UserID string + GroupID string + ContextMsgs []map[string]interface{} + LLMText string + ToolCalls []ToolCall + ToolResults []ToolResult + FinalText string + Response *string + Phase Stage + Memory []MemItem + Extra map[string]interface{} +} + +type ToolCall struct { + ID string `json:"id"` + Name string `json:"name"` + Arguments map[string]interface{} `json:"arguments"` +} + +type ToolResult struct { + CallID string `json:"call_id"` + Name string `json:"name"` + Success bool `json:"success"` + Result interface{} `json:"result"` +} + +type ToolDef struct { + Name string `json:"name"` + Description string `json:"description"` + Parameters map[string]interface{} `json:"parameters"` +} + +type MemoryAPI interface { + Recall(query string, topK int) ([]MemItem, error) + Commit(triples []map[string]string) error + Introspect() (map[string]interface{}, error) +} + +type KnowledgeAPI interface { + Search(query string, topK int) ([]MemItem, error) + Create(name, content string) error + List() ([]string, error) +} + +type EventHandler func(event *Event) +type StageHandler func(ctx *StageContext) error +type ToolHandler func(args map[string]interface{}) (interface{}, error) + +type PluginAPI struct { + Name string + Version string + + tools map[string]ToolHandler + stages map[Stage][]StageHandler + events map[EventType][]EventHandler + eventBus EventBus + memAPI MemoryAPI + knowAPI KnowledgeAPI +} + +func NewPluginAPI(name, version string, bus EventBus, mem MemoryAPI, know KnowledgeAPI) *PluginAPI { + return &PluginAPI{ + Name: name, + Version: version, + tools: make(map[string]ToolHandler), + stages: make(map[Stage][]StageHandler), + events: make(map[EventType][]EventHandler), + eventBus: bus, + memAPI: mem, + knowAPI: know, + } +} + +func (p *PluginAPI) RegisterTool(name string, handler ToolHandler) error { + if _, ok := p.tools[name]; ok { + return fmt.Errorf("tool %s already registered by plugin %s", name, p.Name) + } + p.tools[name] = handler + if p.eventBus != nil { + p.eventBus.Publish(&Event{ + Type: EventSystem, + Source: p.Name, + Payload: map[string]interface{}{"action": "register_tool", "tool": name}, + }) + } + return nil +} + +func (p *PluginAPI) RegisterStage(stage Stage, handler StageHandler) { + p.stages[stage] = append(p.stages[stage], handler) +} + +func (p *PluginAPI) Subscribe(eventType EventType, handler EventHandler) { + p.events[eventType] = append(p.events[eventType], handler) + if p.eventBus != nil { + p.eventBus.Subscribe(eventType, handler) + } +} + +func (p *PluginAPI) Publish(evt *Event) { + if p.eventBus != nil { + p.eventBus.Publish(evt) + } +} + +func (p *PluginAPI) Tools() map[string]ToolHandler { + return p.tools +} + +func (p *PluginAPI) StageHandlers(stage Stage) []StageHandler { + return p.stages[stage] +} + +func (p *PluginAPI) Memory() MemoryAPI { return p.memAPI } +func (p *PluginAPI) Knowledge() KnowledgeAPI { return p.knowAPI } + +func AllStages() map[Stage]bool { + return map[Stage]bool{ + StageOnInput: true, + StagePreAction: true, + StagePostAction: true, + StageBeforeToolcall: true, + StageAfterToolcall: true, + StageBeforeOutput: true, + StageAfterOutput: true, + } +} diff --git a/internal/plugin/sdk/bus.go b/internal/plugin/sdk/bus.go new file mode 100644 index 0000000..573f140 --- /dev/null +++ b/internal/plugin/sdk/bus.go @@ -0,0 +1,42 @@ +package sdk + +import "fmt" + +type EventBus interface { + Publish(event *Event) + Subscribe(eventType EventType, handler EventHandler) func() +} + +type InProcessBus struct { + subs map[EventType][]EventHandler +} + +func NewInProcessBus() *InProcessBus { + return &InProcessBus{ + subs: make(map[EventType][]EventHandler), + } +} + +func (b *InProcessBus) Publish(evt *Event) { + for _, h := range b.subs[EventAll] { + h(evt) + } + if evt.Type != EventAll { + for _, h := range b.subs[evt.Type] { + h(evt) + } + } +} + +func (b *InProcessBus) Subscribe(eventType EventType, handler EventHandler) func() { + b.subs[eventType] = append(b.subs[eventType], handler) + return func() { + list := b.subs[eventType] + for i, h := range list { + if fmt.Sprintf("%p", h) == fmt.Sprintf("%p", handler) { + b.subs[eventType] = append(list[:i], list[i+1:]...) + break + } + } + } +} diff --git a/internal/plugin/sdk/bus_test.go b/internal/plugin/sdk/bus_test.go new file mode 100644 index 0000000..0ef0e04 --- /dev/null +++ b/internal/plugin/sdk/bus_test.go @@ -0,0 +1,133 @@ +package sdk + +import ( + "testing" +) + +func TestInProcessBus(t *testing.T) { + bus := NewInProcessBus() + var called bool + + bus.Subscribe(EventRawInput, func(evt *Event) { + called = true + if evt.Source != "test" { + t.Errorf("expected source test, got %s", evt.Source) + } + }) + + bus.Publish(&Event{ + Type: EventRawInput, + Source: "test", + }) + + if !called { + t.Error("handler was not called") + } +} + +func TestInProcessBusWildcard(t *testing.T) { + bus := NewInProcessBus() + count := 0 + + bus.Subscribe(EventAll, func(evt *Event) { + count++ + }) + + bus.Publish(&Event{Type: EventRawInput, Source: "s1"}) + bus.Publish(&Event{Type: EventToolCall, Source: "s2"}) + + if count != 2 { + t.Errorf("expected 2, got %d", count) + } +} + +func TestPluginAPI(t *testing.T) { + bus := NewInProcessBus() + api := NewPluginAPI("test", "1.0.0", bus, nil, nil) + + if api.Name != "test" { + t.Errorf("expected test, got %s", api.Name) + } + if api.Version != "1.0.0" { + t.Errorf("expected 1.0.0, got %s", api.Version) + } + + var stageCalled bool + api.RegisterStage(StageOnInput, func(ctx *StageContext) error { + stageCalled = true + if ctx.RawMessage != "hello" { + t.Errorf("expected hello, got %s", ctx.RawMessage) + } + return nil + }) + + ctx := &StageContext{RawMessage: "hello"} + for _, handler := range api.StageHandlers(StageOnInput) { + handler(ctx) + } + + if !stageCalled { + t.Error("stage handler was not called") + } +} + +func TestPluginAPITool(t *testing.T) { + api := NewPluginAPI("test", "1.0.0", nil, nil, nil) + + err := api.RegisterTool("test_tool", func(args map[string]interface{}) (interface{}, error) { + return "ok", nil + }) + if err != nil { + t.Fatalf("register tool: %v", err) + } + + if _, ok := api.Tools()["test_tool"]; !ok { + t.Error("tool not found") + } + + // duplicate registration should fail + err = api.RegisterTool("test_tool", func(args map[string]interface{}) (interface{}, error) { + return "ok", nil + }) + if err == nil { + t.Error("expected error on duplicate tool registration") + } +} + +func TestPluginAPIStageShortCircuit(t *testing.T) { + api := NewPluginAPI("test", "1.0.0", nil, nil, nil) + + api.RegisterStage(StageOnInput, func(ctx *StageContext) error { + resp := "intercepted" + ctx.Response = &resp + return nil + }) + + ctx := &StageContext{RawMessage: "hello"} + handlers := api.StageHandlers(StageOnInput) + if len(handlers) != 1 { + t.Fatalf("expected 1 handler, got %d", len(handlers)) + } + handlers[0](ctx) + + if ctx.Response == nil || *ctx.Response != "intercepted" { + t.Errorf("expected intercepted, got %v", ctx.Response) + } +} + +func TestAllStages(t *testing.T) { + stages := AllStages() + expected := []Stage{ + StageOnInput, StagePreAction, StagePostAction, + StageBeforeToolcall, StageAfterToolcall, + StageBeforeOutput, StageAfterOutput, + } + for _, s := range expected { + if !stages[s] { + t.Errorf("missing stage: %s", s) + } + } + if len(stages) != len(expected) { + t.Errorf("expected %d stages, got %d", len(expected), len(stages)) + } +}