diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 98ae64d..e5d5527 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -333,6 +333,33 @@ flowchart TB I --> J["发送给 LLM"] ``` +### 5.3.1 `/clear` 清空上下文流程 + +用户发送 `/clear` 指令时，主动截断上下文并把本次会话归档为摘要（按会话过期处理）： + +```mermaid +flowchart TB + U(["📱 用户发送 /clear"]) --> K{"is_clear_command() 匹配?"} + K -->|"否"| N(["走正常 LLM 对话流程"]) + K -->|"是"| L["加载历史（过滤掉 /clear 本身）"] + L --> M{"历史非空?"} + M -->|"否"| R2["回复: 上下文已是空的"] + M -->|"是"| S["一次性 Conversation + summarizer"] + S --> T["trigger_clear_summary() → LLM 生成摘要"] + T --> DB[("session_summaries\nreason=clear")] + DB --> D["丢弃该用户在途会话 (sessions.retain)"] + D --> R1["回复确认 + 摘要预览"] + R2 --> END(["📱 用户收到回复"]) + R1 --> END +``` + +- **会话边界**：摘要的 `created_at` 作为边界，之后加载历史时通过 + `latest_session_boundary()` 过滤掉此前的消息（`/clear` 与 12h 空闲超时 + 共用同一套边界机制） +- **注入与否**：`/clear` 摘要**不会**注入下一次对话的 system prompt， + 只存档供 `read_summaries` 工具查询（与空闲超时摘要一致） +- **在途会话**：`/clear` 会丢弃该用户尚未完成的工具往返，避免用旧上下文继续 + --- ## 6. 文件依赖矩阵 diff --git a/migrations/20260601000016_session_summaries_user_idx.sql b/migrations/20260601000016_session_summaries_user_idx.sql new file mode 100644 index 0000000..a549a43 --- /dev/null +++ b/migrations/20260601000016_session_summaries_user_idx.sql @@ -0,0 +1,4 @@ +-- 为"按用户查询最近的会话边界（clear/timeout 摘要）"添加索引。 +-- list_recent_chat_records_for_context 会据此过滤掉已过期会话的消息。 +CREATE INDEX IF NOT EXISTS idx_session_summaries_user_created + ON session_summaries (user_id, created_at DESC); diff --git a/src/context/builder.rs b/src/context/builder.rs index c6b6a19..cf1a41c 100644 --- a/src/context/builder.rs +++ b/src/context/builder.rs @@ -260,13 +260,48 @@ pub async fn trigger_overflow_summary( /// 而是只保存到数据库，供 `read_summaries` 工具查询。 /// /// 生成摘要后会清空所有消息，checkpoint 重置为 0。 +/// 该摘要的 `created_at` 同时作为"会话边界"——之后加载历史时 +/// 不会包含此时间点之前的消息（见 `db::latest_session_boundary`）。 pub async fn trigger_idle_summary( session: &Arc>, summarizer: Option<&Summarizer>, ) { + let _ = archive_session(session, summarizer, super::types::SummaryReason::Timeout, "timeout") + .await; +} + +/// ## 触发清空上下文摘要（`/clear` 指令） +/// +/// 用户发送 `/clear` 时调用。把当前会话消息交给 LLM 生成摘要并存档， +/// 然后清空消息、重置 checkpoint，使会话"过期"。 +/// +/// 与空闲超时摘要一样： +/// * 摘要**不会**注入到下一次对话，只存档供 `read_summaries` 查询 +/// * 摘要的 `created_at` 作为会话边界，之后加载历史不再包含此前的消息 +/// +/// 返回生成的摘要文本；若会话本就没有消息，返回 `None`。 +pub async fn trigger_clear_summary( + session: &Arc>, + summarizer: Option<&Summarizer>, +) -> Option { + archive_session(session, summarizer, super::types::SummaryReason::Clear, "clear").await +} + +/// 归档当前会话：生成摘要 → 存库 → 清空消息 → 重置 checkpoint。 +/// +/// `trigger_idle_summary` 与 `trigger_clear_summary` 共用此逻辑， +/// 仅 `reason` 不同（Timeout / Clear），二者都代表"会话过期"。 +/// +/// 返回生成的摘要文本；若会话没有消息可归档，返回 `None`。 +async fn archive_session( + session: &Arc>, + summarizer: Option<&Summarizer>, + reason: super::types::SummaryReason, + reason_str: &str, +) -> Option { let mut s = session.lock().await; if s.messages.is_empty() { - return; + return None; } let summary_text = generate_summary(&s.messages, summarizer).await; @@ -275,12 +310,12 @@ pub async fn trigger_idle_summary( s.summaries.push(super::types::SummaryEntry { checkpoint: cp, text: summary_text.clone(), - reason: super::types::SummaryReason::Timeout, + reason, created_at: chrono::Utc::now(), }); let msg_count = cp as i32; - s.save_summary_to_db(&summary_text, "timeout", msg_count) + s.save_summary_to_db(&summary_text, reason_str, msg_count) .await; s.checkpoint = s.messages.len(); @@ -296,6 +331,8 @@ pub async fn trigger_idle_summary( } } } + + Some(summary_text) } /// 生成摘要：优先使用 LLM，不可用时回退到简单截断 @@ -431,4 +468,41 @@ mod tests { assert!(has_anchor); assert!(has_preface); } + + #[tokio::test] + async fn trigger_clear_summary_archives_and_clears() { + use crate::context::types::SummaryReason; + + let mut session = ChatSession::new(); + session.add_user("今天天气怎么样".to_string()); + session.add_assistant("北京今天晴，25度".to_string()); + session.add_user("明天呢".to_string()); + session.add_assistant("明天多云转小雨".to_string()); + + let arc = Arc::new(Mutex::new(session)); + let summarizer: Summarizer = Arc::new(|_prompt: String| { + Box::pin(async { Ok("摘要：讨论了北京近两天天气".to_string()) }) + }); + + let summary = trigger_clear_summary(&arc, Some(&summarizer)).await; + + assert!(summary.is_some()); + assert_eq!(summary.unwrap(), "摘要：讨论了北京近两天天气"); + let s = arc.lock().await; + assert!(s.messages.is_empty(), "清空后消息应被清空"); + assert_eq!(s.checkpoint, 0); + assert!(s + .summaries + .iter() + .any(|x| x.reason == SummaryReason::Clear)); + } + + #[tokio::test] + async fn trigger_clear_summary_empty_session_returns_none() { + let arc = Arc::new(Mutex::new(ChatSession::new())); + let summarizer: Summarizer = + Arc::new(|_p: String| Box::pin(async { Ok("x".to_string()) })); + let summary = trigger_clear_summary(&arc, Some(&summarizer)).await; + assert!(summary.is_none()); + } } diff --git a/src/context/types.rs b/src/context/types.rs index 97edb20..b165fd1 100644 --- a/src/context/types.rs +++ b/src/context/types.rs @@ -14,6 +14,7 @@ //! 2. AI 回复 → `add_assistant()` → 存入 `messages` //! 3. Token 超 budget → `trigger_overflow_summary()` → 压缩 checkpoint 前的消息 //! 4. 12h 空闲 → `trigger_idle_summary()` → 生成摘要，清空消息 +//! 5. `/clear` 指令 → `trigger_clear_summary()` → 生成摘要归档，会话过期 use crate::llm::types::Message; use chrono::{DateTime, Utc}; @@ -30,15 +31,18 @@ pub struct HistoryEntry { /// ## 摘要原因 /// -/// 区分两种触发摘要的场景： +/// 区分触发摘要的场景： /// * `Overflow` — 上下文 token 总数超过预算（token budget），触发压缩 /// * `Timeout` — 距上条用户消息超过 12 小时空闲，触发会话总结 +/// * `Clear` — 用户发送 `/clear` 主动清空上下文，触发会话过期归档 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub enum SummaryReason { /// 上下文 token 超 budget 触发 Overflow, /// 距上条用户消息超过 12 小时触发 Timeout, + /// 用户发送 `/clear` 主动清空上下文触发 + Clear, } /// ## 一条摘要记录 @@ -77,6 +81,7 @@ pub struct SummaryEntry { /// 2. AI 回复 → `add_assistant()` → 存入 `messages` /// 3. Token 超 budget → `trigger_overflow_summary()` → 压缩 checkpoint 前的消息 /// 4. 12h 空闲 → `trigger_idle_summary()` → 生成摘要，清空消息 +/// 5. `/clear` 指令 → `trigger_clear_summary()` → 生成摘要归档，会话过期 #[derive(Debug, Clone)] pub struct ChatSession { pub user_id: String, diff --git a/src/daemon.rs b/src/daemon.rs index 8276b1e..cdaac4d 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -39,6 +39,7 @@ //! 采用统一的单进程 tokio task 架构。旧版 daemon+worker 分离模式（UDS IPC）已在 v2 中移除。 use crate::channel::ChannelId; +use crate::context::builder; use crate::context::HistoryEntry; use crate::context::MemoryStore; use crate::db::Database; @@ -59,6 +60,10 @@ use tokio::sync::Mutex; use tracing::{error, info, warn}; use uuid::Uuid; +/// LLM Consumer 的会话缓存：`correlation_id → (Conversation, 创建时间, user_id)`。 +/// 额外记录 `user_id` 是为了在 `/clear` 时能按用户丢弃尚未完成的在途会话。 +type SessionMap = HashMap; + /// ## Daemon 上下文 —— 消费者之间共享的全部状态 /// /// 这个结构体被所有三个消费者（LLM / Tool / Send）通过 `Arc` 共享访问。 @@ -194,7 +199,7 @@ pub async fn run(database: &Arc, memory_store: &Arc) { sessions_for_clean .lock() .await - .retain(|_, (conv, created)| { + .retain(|_, (conv, created, _)| { let ttl_secs = if conv.has_pending_async() { 1800 } else { 300 }; now.duration_since(*created).as_secs() < ttl_secs }); @@ -270,9 +275,9 @@ fn spawn_consumers( enqueue: EnqueueHandle, ) -> ( Vec>, - Arc>>, + Arc>, ) { - let sessions: Arc>> = + let sessions: Arc> = Arc::new(Mutex::new(HashMap::new())); let mut handles = Vec::new(); @@ -312,7 +317,7 @@ async fn llm_consumer_loop( rx: &mut tokio::sync::mpsc::Receiver, ctx: &DaemonCtx, enqueue: &EnqueueHandle, - sessions: &Arc>>, + sessions: &Arc>, ) { while let Some(msg) = rx.recv().await { let user_id = msg.channel.user_id.clone(); @@ -324,6 +329,19 @@ async fn llm_consumer_loop( message_id: _, context_token, } => { + // /clear 指令：截断上下文 + 生成摘要 + 按会话过期处理 + if is_clear_command(&text) { + handle_clear_command( + ctx, + enqueue, + sessions, + &user_id, + correlation_id, + context_token, + ) + .await; + continue; + } // 加载数据 ctx.memory_store.load_if_needed(&user_id).await; let history = load_history(&ctx.db, &user_id).await; @@ -428,7 +446,7 @@ async fn llm_consumer_loop( sessions .lock() .await - .insert(correlation_id, (conv, Instant::now())); + .insert(correlation_id, (conv, Instant::now(), user_id.clone())); // 执行 LLM 对话 run_llm_round( @@ -452,7 +470,7 @@ async fn llm_consumer_loop( // 从缓存恢复会话，注入工具结果并恢复 LLM 循环 let should_resume = { let mut map = sessions.lock().await; - if let Some((conv, last_access)) = map.get_mut(&correlation_id) { + if let Some((conv, last_access, _)) = map.get_mut(&correlation_id) { conv.session().lock().await.add_tool_result( &tool_call_id, &tool_name, @@ -502,13 +520,13 @@ async fn run_llm_round( correlation_id: Uuid, context_token: Option, text: &str, - sessions: &Arc>>, + sessions: &Arc>, enqueue: &EnqueueHandle, ) { let result = { let map = sessions.lock().await; let conv = match map.get(&correlation_id) { - Some((c, _)) => c, + Some((c, _, _)) => c, None => { error!(target: "ias::err","[llm-session] conv不在缓存"); return; @@ -545,13 +563,13 @@ async fn resume_llm_round( user_id: &str, correlation_id: Uuid, context_token: Option, - sessions: &Arc>>, + sessions: &Arc>, enqueue: &EnqueueHandle, ) { let result = { let map = sessions.lock().await; let conv = match map.get(&correlation_id) { - Some((c, _)) => c, + Some((c, _, _)) => c, None => { error!(target: "ias::err","[llm-resume] conv不在缓存"); return; @@ -589,7 +607,7 @@ async fn handle_chat_result( correlation_id: Uuid, context_token: Option, result: ChatResult, - sessions: &Arc>>, + sessions: &Arc>, enqueue: &EnqueueHandle, ) { if result.has_pending_async { @@ -617,6 +635,111 @@ async fn handle_chat_result( sessions.lock().await.remove(&correlation_id); } +// ─── /clear 指令处理 ─── + +/// 判断文本是否为清空上下文指令 `/clear`（大小写不敏感，允许前后空白）。 +fn is_clear_command(text: &str) -> bool { + text.trim().eq_ignore_ascii_case("/clear") +} + +/// ## 处理 `/clear` 指令 —— 截断上下文 + 生成摘要 + 会话过期 +/// +/// 用户发送 `/clear` 时： +/// 1. 加载当前历史，用 LLM 生成会话摘要并归档（reason = clear） +/// 2. 摘要的 `created_at` 成为会话边界，之后加载历史不再包含此前的消息 +/// 3. 丢弃该用户尚未完成的在途会话（按会话过期处理） +/// 4. 回复用户清空确认（附带摘要预览） +async fn handle_clear_command( + ctx: &DaemonCtx, + enqueue: &EnqueueHandle, + sessions: &Arc>, + user_id: &str, + correlation_id: Uuid, + context_token: Option, +) { + info!(target: "ias::msg", "[clear] {} 请求清空上下文", user_id); + + // 1. 生成摘要：用一次性 Conversation 提供 summarizer + session + // 过滤掉历史里的 /clear 指令本身，避免污染摘要 + let history: Vec = load_history(&ctx.db, user_id) + .await + .into_iter() + .filter(|h| !is_clear_command(&h.content)) + .collect(); + + let summary = if history.is_empty() { + None + } else { + let cfg = ConversationConfig { + system_prompt: String::new(), + model: ctx.model.clone(), + ..Default::default() + }; + match Conversation::new(cfg) { + Ok(conv) => { + { + let s = conv.session(); + let mut s = s.lock().await; + s.db_pool = Some(Arc::new(ctx.db.pool().clone())); + s.load_from_history(&history); + s.current_user_id = user_id.to_string(); + } + let summarizer = conv.summarizer(); + builder::trigger_clear_summary(&conv.session(), Some(&summarizer)).await + } + Err(e) => { + error!(target: "ias::err", "[clear] 初始化摘要器失败: {e}"); + None + } + } + }; + + // 2. 丢弃该用户在途的会话（按会话过期处理） + sessions + .lock() + .await + .retain(|_, (_, _, uid)| uid.as_str() != user_id); + + // 3. 回复确认（附带摘要预览） + let reply = build_clear_reply(&summary); + + let msg = PipelineMessage::llm_reply_with_source( + ChannelId::wechat(user_id), + correlation_id, + &reply, + "clear", + context_token, + ); + enqueue.enqueue(msg).await; +} + +/// 根据 `trigger_clear_summary` 的返回值构建 `/clear` 回复文案。 +/// +/// * `None` — 无历史可清空 +/// * `Some("")` — 已清空但 LLM 摘要为空 +/// * `Some(text)` — 已清空，附带 300 字预览 +fn build_clear_reply(summary: &Option) -> String { + match summary { + None => "🧹 上下文已是空的，无需清空。".to_string(), + Some(text) if text.is_empty() => "🧹 已清空上下文，本次对话已归档。".to_string(), + Some(text) => { + let chars: Vec = text.chars().collect(); + let (preview, suffix) = if chars.len() > 300 { + ( + chars[..300].iter().collect::(), + "\n…（摘要已截断预览，完整内容已归档）", + ) + } else { + (text.clone(), "") + }; + format!( + "🧹 已清空上下文，本次对话已归档为摘要。\n\n📝 摘要预览：\n{}{}", + preview, suffix + ) + } + } +} + // ─── Tool Consumer ─── async fn tool_consumer_loop( @@ -964,3 +1087,43 @@ async fn load_summaries(db: &Arc, user_id: &str) -> Vec { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn clear_command_detection() { + assert!(is_clear_command("/clear")); + assert!(is_clear_command(" /clear ")); + assert!(is_clear_command("/CLEAR")); + assert!(is_clear_command("/Clear")); + assert!(is_clear_command("\t/clear\n")); + // 非精确匹配不应触发 + assert!(!is_clear_command("/clear all")); + assert!(!is_clear_command("请/clear")); + assert!(!is_clear_command("清空")); + assert!(!is_clear_command("")); + assert!(!is_clear_command("/clea")); + } + + #[test] + fn clear_reply_branches() { + // 无历史可清空 + assert_eq!(build_clear_reply(&None), "🧹 上下文已是空的，无需清空。"); + // 已清空但摘要为空（不应误报“已是空的”） + assert_eq!( + build_clear_reply(&Some(String::new())), + "🧹 已清空上下文，本次对话已归档。" + ); + // 正常摘要：附带预览 + let r = build_clear_reply(&Some("讨论了天气".to_string())); + assert!(r.starts_with("🧹 已清空上下文，本次对话已归档为摘要。")); + assert!(r.contains("讨论了天气")); + // 超长摘要：截断预览 + let long = "x".repeat(500); + let r = build_clear_reply(&Some(long)); + assert!(r.contains("摘要已截断预览")); + assert!(!r.contains(&"x".repeat(500))); + } +} diff --git a/src/db/models.rs b/src/db/models.rs index ff01f6a..e16740d 100644 --- a/src/db/models.rs +++ b/src/db/models.rs @@ -138,22 +138,31 @@ pub async fn insert_chat_record( } /// 查询最近可进入 LLM 上下文的聊天记录（排除工具等待/审批等中间提示）。 +/// +/// 会自动应用"会话边界"：最近一次 `/clear` 或 12h 空闲超时摘要之前的消息 +/// 已归档为摘要，不再进入新会话的上下文（见 [`latest_session_boundary`]）。 pub async fn list_recent_chat_records_for_context( pool: &PgPool, user_id: &str, limit: i64, ) -> Result, String> { + // 会话边界：最近一次"会话过期"摘要（/clear 或 12h 空闲超时）的时间点。 + // 边界之前的消息已归档为摘要，不再进入新会话的上下文。 + let boundary = latest_session_boundary(pool, user_id).await; + let records = sqlx::query_as::<_, ChatRecord>( r#" SELECT id, created_at, direction, user_id, account_id, text, source, context_token, message_id FROM chat_records WHERE user_id = $1 - AND source NOT IN ('llm_pending_tool', 'tool_approval') + AND source NOT IN ('llm_pending_tool', 'tool_approval', 'clear') + AND created_at > COALESCE($2, '-infinity'::timestamptz) ORDER BY created_at DESC - LIMIT $2 + LIMIT $3 "#, ) .bind(user_id) + .bind(boundary) .bind(limit) .fetch_all(pool) .await @@ -162,6 +171,32 @@ pub async fn list_recent_chat_records_for_context( Ok(records) } +/// ## 查询用户最近的"会话边界"时间点 +/// +/// 返回最近一次会话过期摘要（`/clear` 或 12h 空闲超时）的 `created_at`。 +/// 该时间点之前的聊天记录已归档为摘要，不应再进入新会话的上下文。 +/// +/// `Overflow` 摘要不构成边界（它发生在会话进行中，仅压缩早期消息）。 +pub async fn latest_session_boundary( + pool: &PgPool, + user_id: &str, +) -> Option> { + sqlx::query_as::<_, (DateTime,)>( + r#" + SELECT created_at + FROM session_summaries + WHERE user_id = $1 AND reason IN ('clear', 'timeout') + ORDER BY created_at DESC + LIMIT 1 + "#, + ) + .bind(user_id) + .fetch_optional(pool) + .await + .ok()? + .map(|(t,)| t) +} + // ─── LLM 用量 ─── /// LLM 用量聚合统计