From e379e498b5c2f7363ae600126d6e08b76c7bc175 Mon Sep 17 00:00:00 2001 From: wunianxiao Date: Wed, 3 Jun 2026 11:40:41 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E5=AE=88=E6=8A=A4/=E8=B0=83=E5=BA=A6?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E5=88=86=E7=A6=BB=E5=AE=9E=E7=8E=B0=E8=AE=A1?= =?UTF-8?q?=E5=88=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DAEMON_WORKER_PLAN.md | 440 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 440 insertions(+) create mode 100644 DAEMON_WORKER_PLAN.md diff --git a/DAEMON_WORKER_PLAN.md b/DAEMON_WORKER_PLAN.md new file mode 100644 index 0000000..e017d1d --- /dev/null +++ b/DAEMON_WORKER_PLAN.md @@ -0,0 +1,440 @@ +# iAs 守护进程 / 调度进程分离 — 实现计划 + +> 生成于 2026-06-02，供新会话参考当前架构状态。 + +--- + +## 目标 + +``` +ias daemon (常驻) ias worker (短命，每条消息 spawn) +┌──────────────────────┐ ┌──────────────────────┐ +│ WeChat 长轮询 │ │ LLM 对话 │ +│ 消息收发 │ Unix │ 工具调用 │ +│ 审批匹配 │◄─Domain──►│ 无 DB 连接 │ +│ DB 持有 / 上下文预载 │ Socket │ 无状态（纯计算） │ +│ 消息队列（按用户串行） │ │ stdin 读取任务/环境 │ +└──────────────────────┘ └──────────────────────┘ +``` + +- **daemon** 稳定、不更新。持有 WeChat 长连接和数据库。 +- **worker** 每次收到消息 spawn，完成后销毁。代码更新 → 编译 → 下一条消息自动用新 worker。 +- **IPC**：Unix Domain Socket，长度前缀帧 + JSON 消息。 + +--- + +## 当前架构（供参考） + +``` +src/ +├── main.rs CLI 入口 + 监听循环 + 工具执行器 +├── cli.rs clap 子命令 (login/listen/send/whoami/usage/service) +├── db/ +│ ├── mod.rs Database 连接池 +│ └── models.rs CRUD (auth/chat/usage/summary/memories/approvals) +├── llm/ +│ ├── types.rs Message / Role / Usage / StreamChunk +│ ├── provider.rs create_provider / parse_chat_chunk +│ ├── deepseek.rs DeepSeek 流式客户端 +│ └── conversation.rs Conversation (chat + chat_with_tools 工具循环) +├── context/ +│ ├── types.rs ChatSession +│ ├── builder.rs token budget 构建 + 摘要触发 +│ └── tools.rs MemoryStore +├── tools/ +│ ├── types.rs RiskLevel / SkillSpec / SkillResult / ExecutionContext +│ ├── builtin.rs BuiltinRegistry (get_current_datetime / manage_memos / query_weather) +│ ├── weather.rs QWeather 客户端 (JWT + reqwest) +│ └── approval.rs ApprovalManager (确认码 + oneshot 通道) +└── wechat/ + ├── types.rs iLink API 协议类型 + └── client.rs HTTP 客户端 (login/poll/send/notify) +``` + +--- + +## 分阶段实现 + +### Phase 1: 协议层 — IPC 帧格式 + +**目标**：实现与业务无关的 `send_frame` / `recv_frame`。 + +**新文件**：`src/ipc.rs` + +```rust +// 长度前缀帧 +// [4 字节 u32 BE: payload_len] +// [N 字节: JSON payload] + +pub async fn send_frame(stream: &mut UnixStream, msg: &serde_json::Value) -> Result<()> +pub async fn recv_frame(stream: &mut UnixStream) -> Result +``` + +**测试**： +```bash +cargo test ipc +``` + +--- + +### Phase 2: Worker — 无状态执行进程 + +**目标**：`ias worker` 子命令，接收 task JSON，执行 LLM，输出结果帧。 + +**新文件**：`src/worker.rs` + +```rust +// ias worker --sock /tmp/ias_daemon.sock +pub async fn run(sock_path: &str) -> Result<()> { + let mut stream = UnixStream::connect(sock_path).await?; + + // 1. 读 task + let task: TaskFrame = recv_frame(&mut stream).await?; + + // 2. 注入环境变量（从 task.env 字段） + for (k, v) in &task.env { std::env::set_var(k, v); } + + // 3. 构建 Conversation（复用现有代码） + let mut conv = Conversation::new(config)?; + conv.session().load_history(&task.history); + conv.session().load_memories(&task.memories); + conv.session().load_summaries(&task.summaries); + + // 4. 注册工具执行器（复用现有代码） + conv.set_tool_executor(build_executor()); + + // 5. LLM 对话 + 工具循环 + let (reply, used_tools, usage) = conv.chat_with_tools(&task.msg.text).await?; + + // 6. 输出结果帧 + send_frame(&mut stream, &OutputFrame::usage(&usage)).await?; + send_frame(&mut stream, &OutputFrame::reply(&reply)).await?; + send_frame(&mut stream, &OutputFrame::bye()).await?; + + Ok(()) +} +``` + +**改动**： +- `cli.rs`：新增 `Worker { sock_path: String }` 子命令 +- `main.rs`：`Commands::Worker { sock_path } => worker::run(&sock_path).await` + +**不依赖 daemon，可独立测试**： +```bash +# 终端 1：模拟 daemon +socat UNIX-LISTEN:/tmp/ias.sock,fork EXEC:'echo done' + +# 终端 2：运行 worker +ias worker --sock /tmp/ias.sock +``` + +--- + +### Phase 3: Daemon — 常驻进程 + 消息队列 + +**目标**：`ias daemon` 子命令，持有 WeChat + DB，spawn worker。 + +**新文件**：`src/daemon.rs` + +```rust +pub async fn run(db: Arc, client: WeChatClient) -> ! { + let sock_path = "/tmp/ias_daemon.sock"; + let _ = std::fs::remove_file(sock_path); + let listener = UnixListener::bind(sock_path)?; + + let queue = MessageQueue::new(); + + loop { + tokio::select! { + // WeChat 消息 + msg = recv_message(&client) => { + if is_approval_reply(&msg) { + handle_approval(&msg).await; + } else { + queue.enqueue(msg); + process_queue(&queue, &db).await; + } + } + // Worker 连接（worker 启动后 connect） + (stream, _) = listener.accept() => { + let user_id = queue.pop_waiting(); + spawn_process(&user_id, stream, &db).await; + } + } + } +} +``` + +**消息队列设计**（`daemon.rs` 内部）： +```rust +struct MessageQueue { + // 每个用户最多一个活跃 worker + active: HashSet, + // 等待队列 + pending: HashMap>, + // 准备就绪待处理的用户 + waiting: VecDeque, +} +``` + +**Daemon 处理一个消息的完整流程**： +```rust +async fn spawn_process(user_id, stream, db) { + // 1. 预载上下文（从 DB） + let history = db::models::list_recent_chat_records(&db.pool, &user_id, 20).await?; + let memories = db.memory_store.read_for(&user_id).await; + let summaries = db::models::load_summaries(&db.pool, 5).await?; + + // 2. 构建 env（从当前进程环境变量） + let env = HashMap::from([ + ("DEEPSEEK_API_KEY", std::env::var("DEEPSEEK_API_KEY")?), + ("DEEPSEEK_MODEL", std::env::var("DEEPSEEK_MODEL")?), + ("QWEATHER_API_HOST", std::env::var("QWEATHER_API_HOST")?), + // ... + ]); + + // 3. 发送 task 帧 + send_frame(&mut stream, &TaskFrame { user_id, msg, history, memories, summaries, env }).await?; + + // 4. 逐帧处理 worker 输出 + loop { + match recv_frame(&mut stream).await? { + OutputFrame::Chunk { text } => { + // 暂存，等最终 reply 一起发送 + reply_buffer.push_str(&text); + } + OutputFrame::DbWrite { table, row } => { + execute_db_write(&db, table, row).await; + } + OutputFrame::Reply { text } => { + client.send_text(&user_id, &text, &ctx_token).await?; + // 发送成功后记录聊天 + } + OutputFrame::NeedApproval { tool, code, message } => { + approval_manager.create(&user_id, &tool, &code).await?; + client.send_text(&user_id, &message, None).await?; + // Worker 退出。审批完成后再 spawn 新 worker。 + break; + } + OutputFrame::Bye => break, + } + } +} +``` + +**改动**： +- `cli.rs`：新增 `Daemon` 子命令（替代 `Service`） +- `main.rs`：`Commands::Daemon => daemon::run(db, client).await` + +--- + +### Phase 4: 审批跨进程 + +**目标**：worker 遇 High 风险工具 → 输出 `need_approval` 帧 → daemon 处理 → 重新 spawn worker。 + +**流程**： + +``` +Worker Daemon + │ │ + ├─ LLM 请求 manage_memos │ + ├─ 遇 High 风险 │ + ├─ write(need_approval) ──────→├─ 读 need_approval + ├─ write(bye) ────────────────→├─ 创建审批记录到 DB + │ ├─ 发确认消息到微信 + │ (Worker 退出) │ + │ ├─ 等待用户回复... + │ ├─ 收到确认码 → 匹配 + │ │ + │ (新 Worker spawn) │ + │◄───── write(task + approved)─├─ task 帧中带 approved_tool + ├─ LLM 继续执行 │ + ├─ write(reply) ──────────────→├─ 发最终回复 +``` + +**task 帧扩展**： +```json +{ + "user_id": "...", + "msg": {...}, + "history": [...], + "approved_tool": "manage_memos", // ← 新增 + "env": {...} +} +``` + +--- + +### Phase 5: 流式输出 + +**目标**：LLM 每生成一个 token，通过 UDS 帧实时推送给 daemon，daemon 可选实时推用户。 + +**当前行为**：等全部生成完 → 一次发送。 +**流式行为**：逐 chunk 发送 → daemon 攒到换行或完成 → 发送微信。 + +```json +{"type":"chunk","text":"北"} +{"type":"chunk","text":"京今"} +{"type":"chunk","text":"天晴"} +// 可攒多个 chunk 再发一帧以减少帧开销 +{"type":"chunk","text":"北京今天晴天，25°C"} +{"type":"reply","text":"北京今天晴天，25°C，微风。"} +``` + +微信不原生支持流式推送，所以收益有限。作为最后阶段可选项。 + +--- + +## 消息协议完整定义 + +### TaskFrame (Daemon → Worker) + +```json +{ + "type": "task", + "user_id": "wxid_abc", + "msg": { + "from": "wxid_abc", + "text": "北京天气", + "account_id": "bot_123", + "context_token": "token_xyz" + }, + "history": [ + {"role": "user", "content": "你好"}, + {"role": "assistant", "content": "你好！有什么可以帮你？"} + ], + "memories": ["用户叫张伟", "住在北京"], + "summaries": ["之前讨论过天气偏好..."], + "approved_tool": null, + "env": { + "DEEPSEEK_API_KEY": "sk-xxx", + "DEEPSEEK_MODEL": "deepseek-v4-flash", + "QWEATHER_API_HOST": "ky5ctpp742.re.qweatherapi.com", + "QWEATHER_JWT_KEY_ID": "KF5AJT3JE9", + "QWEATHER_JWT_PROJECT_ID": "4E2DWXQAVM", + "QWEATHER_JWT_PRIVATE_KEY_FILE": "/home/xiao/project/iAs/qweather/ed25519-private.pem" + } +} +``` + +### OutputFrame (Worker → Daemon) + +```json +{"type": "chunk", "text": "..."} + +{"type": "db_write", "table": "chat_records", "row": {...}} +{"type": "db_write", "table": "llm_usage", "row": {...}} + +{"type": "reply", "text": "最终回复文本"} + +{"type": "need_approval", "tool": "manage_memos", "code": "A1B2C3", + "message": "⚠️ 确认码 A1B2C3，回复确认码继续。"} + +{"type": "bye"} +``` + +### DB Write 表定义 + +| table | row 字段 | +|-------|---------| +| `chat_records` | `direction`, `user_id`, `account_id`, `text`, `source`, `context_token`, `message_id` | +| `llm_usage` | `user_id`, `model`, `provider`, `prompt_tokens`, `completion_tokens`, `cache_hit_tokens`, `cache_miss_tokens` | +| `user_memories` | `user_id`, `content` | + +--- + +## 文件清单 + +| 文件 | Phase | 说明 | +|------|-------|------| +| `src/ipc.rs` | 1 | `UnixStream` + 长度前缀帧 + `send_frame`/`recv_frame` | +| `src/worker.rs` | 2 | Worker 主逻辑：读 task → LLM → 写结果帧 | +| `src/daemon.rs` | 3 | Daemon 主逻辑：WeChat 轮询 + 消息队列 + spawn worker | +| `src/cli.rs` | 1-3 | 新增 `Daemon`、`Worker` 子命令，保留 `Login/Send/Usage` | +| `src/main.rs` | 1-3 | 命令路由，移除 `Service`（由 `Daemon` 替代） | +| `Cargo.toml` | 1 | 新增 `tokio/io-util`（如果有） | + +**不改动的文件**： +- `src/llm/` — 全部复用 +- `src/tools/` — 全部复用 +- `src/db/` — 全部复用（仅 daemon 侧调用） +- `src/wechat/` — 全部复用（仅 daemon 侧调用） +- `src/context/` — 需小改（ChatSession 支持从历史加载） + +--- + +## 测试方法 + +```bash +# 1. 启动 daemon +ias daemon --sock /tmp/ias.sock & + +# 2. 手动测试 worker（不通过 daemon） +echo '{"type":"task","user_id":"test","msg":{"text":"北京天气"},...}' \ + | socat - UNIX-CONNECT:/tmp/ias.sock + +# 3. 完整流程（模拟微信消息） +ias send wxid_test "北京天气" +# daemon 收到 → spawn worker → 回复 +``` + +--- + +## 关键实现细节 + +### ChatSession 从预载数据初始化 + +当前 `ChatSession` 在收到第一条消息时通过 `load_recent_messages` 从 DB 加载历史。Worker 模式需要从 task 帧直接从内存加载： + +```rust +// 在 ChatSession 或 WorkerContext 中新增 +pub fn load_from_history(&mut self, history: &[HistoryEntry]) { + for entry in history { + match entry.role { + "user" => self.add_user(&entry.content), + "assistant" => self.add_assistant(&entry.content), + _ => {} + } + } +} +``` + +### Worker 中的工具执行器 + +Worker 中注册的工具执行器**不连接 DB**： + +```rust +// read_memories / write_memory → 返回消息让 daemon 处理 +// 在 worker 上下文中，只返回 db_write 帧 +fn build_worker_executor() -> ToolExecutor { + Arc::new(move |name, args| { + Box::pin(async move { + match name { + "read_memories" => { + // memories 已在 task 帧中预载 → 直接返回 + Ok(format!("长期记忆: {:?}", loaded_memories)) + } + "write_memory" => { + // 返回 db_write 帧，由 daemon 执行 + Ok(json!({"type":"db_write","table":"user_memories","row":{...}}).to_string()) + } + // 其他 builtin 工具直接执行 + _ => builtin::execute(name, args) + } + }) + }) +} +``` + +--- + +## 工作量估计 + +| Phase | 内容 | 估计 | +|-------|------|------| +| 1 | IPC 帧格式 | 30 min | +| 2 | Worker 独立运行 | 1 h | +| 3 | Daemon + 消息队列 | 2 h | +| 4 | 审批跨进程 | 1 h | +| 5 | 流式输出 | 30 min（可选） | +| **合计** | Phase 1-4 | **~5 h** |