From 3f447668b8a746fbbfc3015803f2021148a49865 Mon Sep 17 00:00:00 2001 From: wunianxiao Date: Tue, 2 Jun 2026 14:27:35 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E7=94=A8=E6=88=B7=E9=9A=94=E7=A6=BB?= =?UTF-8?q?=E5=AE=8C=E6=95=B4=E6=80=A7=20+=20=E8=81=8A=E5=A4=A9=E5=8E=BB?= =?UTF-8?q?=E9=87=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 启动时不加载历史消息（等第一条消息确定身份后按需恢复） 2. MemoryStore 改为 HashMap>，按用户隔离缓存 3. session.current_user_id 与外部 mutex 同步 4. chat_records.message_id 加 UNIQUE 索引 + ON CONFLICT DO NOTHING --- migrations/20260601000001_init.sql | 3 ++ migrations/20260601000009_chat_dedup.sql | 3 ++ src/context/tools.rs | 63 ++++++++++-------------- src/db/models.rs | 1 + src/main.rs | 13 ++--- 5 files changed, 40 insertions(+), 43 deletions(-) create mode 100644 migrations/20260601000009_chat_dedup.sql diff --git a/migrations/20260601000001_init.sql b/migrations/20260601000001_init.sql index 9e3c22b..f8b7e61 100644 --- a/migrations/20260601000001_init.sql +++ b/migrations/20260601000001_init.sql @@ -27,3 +27,6 @@ CREATE INDEX IF NOT EXISTS idx_chat_records_user_created CREATE INDEX IF NOT EXISTS idx_chat_records_account_created ON chat_records (account_id, created_at DESC); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_records_message_id + ON chat_records (message_id) WHERE message_id != ''; diff --git a/migrations/20260601000009_chat_dedup.sql b/migrations/20260601000009_chat_dedup.sql new file mode 100644 index 0000000..d4e2e0d --- /dev/null +++ b/migrations/20260601000009_chat_dedup.sql @@ -0,0 +1,3 @@ +-- 聊天记录去重索引 +CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_records_message_id + ON chat_records (message_id) WHERE message_id != ''; diff --git a/src/context/tools.rs b/src/context/tools.rs index be6cf49..3589f9e 100644 --- a/src/context/tools.rs +++ b/src/context/tools.rs @@ -1,22 +1,22 @@ use sqlx::PgPool; +use std::collections::HashMap; use std::sync::Arc; use tokio::sync::Mutex; -/// 长期记忆管理器（内存 + PostgreSQL 双写） +/// 长期记忆管理器（内存 + PostgreSQL 双写，按用户隔离） pub struct MemoryStore { pool: Option>, - cache: Arc>>, + cache: Arc>>>, } impl MemoryStore { pub fn new(pool: Option>) -> Self { Self { pool, - cache: Arc::new(Mutex::new(Vec::new())), + cache: Arc::new(Mutex::new(HashMap::new())), } } - /// 从数据库加载记忆到缓存 pub async fn load(&self, user_id: &str) { let pool = match self.pool.as_ref() { Some(p) => p.clone(), @@ -30,37 +30,32 @@ impl MemoryStore { .fetch_all(pool.as_ref()) .await { - let mut cache = self.cache.lock().await; - *cache = rows.into_iter().map(|(c,)| c).collect(); + let mems: Vec = rows.into_iter().map(|(c,)| c).collect(); + self.cache.lock().await.insert(uid.to_string(), mems); } } - /// 读取记忆 - pub async fn read(&self) -> String { - let mems = self.cache.lock().await; - if mems.is_empty() { "暂无长期记忆".to_string() } else { - mems.iter().enumerate().map(|(i, m)| format!("{}. {}", i+1, m)).collect::>().join("\n") - } - } - - /// 按用户读取记忆（缓存按 user_id 过滤由调用方保证，这里简化为全量读取） - pub async fn read_for(&self, _user_id: &str) -> String { self.read().await } - - /// 写入记忆 - pub async fn write(&self, content: &str) -> String { self.write_for("default", content).await } - - /// 按用户写入 - pub async fn write_for(&self, user_id: &str, content: &str) -> String { - self.cache.lock().await.push(content.to_string()); + pub async fn read_for(&self, user_id: &str) -> String { let uid = if user_id.is_empty() { "default" } else { user_id }; + let cache = self.cache.lock().await; + match cache.get(uid) { + Some(m) if !m.is_empty() => m.iter().enumerate() + .map(|(i, v)| format!("{}. {}", i + 1, v)) + .collect::>() + .join("\n"), + _ => "暂无长期记忆".to_string(), + } + } + + pub async fn write_for(&self, user_id: &str, content: &str) -> String { + let uid = if user_id.is_empty() { "default" } else { user_id }; + self.cache.lock().await.entry(uid.to_string()).or_default().push(content.to_string()); if let Some(ref pool) = self.pool { - let _ = sqlx::query( - "INSERT INTO user_memories (user_id, content) VALUES ($1, $2)", - ) - .bind(uid) - .bind(content) - .execute(pool.as_ref()) - .await; + let _ = sqlx::query("INSERT INTO user_memories (user_id, content) VALUES ($1, $2)") + .bind(uid) + .bind(content) + .execute(pool.as_ref()) + .await; } "已添加长期记忆".to_string() } @@ -73,7 +68,6 @@ pub async fn read_summaries( let s = session.lock().await; let mut lines: Vec = Vec::new(); - // 内存中的摘要 for sum in &s.summaries { let reason = match sum.reason { super::types::SummaryReason::Overflow => "上下文压缩", @@ -87,7 +81,6 @@ pub async fn read_summaries( )); } - // 数据库中的历史摘要 if let Some(pool) = &s.db_pool { if let Ok(db_summaries) = crate::db::models::load_summaries(pool, 5).await { for text in db_summaries { @@ -98,9 +91,5 @@ pub async fn read_summaries( } } - if lines.is_empty() { - "暂无历史摘要".to_string() - } else { - lines.join("\n---\n") - } + if lines.is_empty() { "暂无历史摘要".to_string() } else { lines.join("\n---\n") } } diff --git a/src/db/models.rs b/src/db/models.rs index b6f365e..94911d6 100644 --- a/src/db/models.rs +++ b/src/db/models.rs @@ -70,6 +70,7 @@ pub async fn insert_chat_record( r#" INSERT INTO chat_records (direction, user_id, account_id, text, source, context_token, message_id) VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (message_id) DO NOTHING RETURNING id "#, ) diff --git a/src/main.rs b/src/main.rs index 0539f3b..dc0ee79 100644 --- a/src/main.rs +++ b/src/main.rs @@ -306,14 +306,11 @@ async fn cmd_listen( // 创建 Conversation let mut conv = match Conversation::new(cfg) { Ok(c) => { - // 设置 db_pool 并加载历史消息 + // 设置 db_pool（启动时不加载消息，等第一条消息到达后再按需恢复） if let Some(db) = database { let session = c.session(); - { - let mut s = session.lock().await; - s.db_pool = Some(Arc::new(db.pool().clone())); - s.load_recent_messages(20).await; - } + let mut s = session.lock().await; + s.db_pool = Some(Arc::new(db.pool().clone())); } info!("LLM 已初始化: {} / {}", c.provider_name(), c.model()); c @@ -541,6 +538,10 @@ async fn listen_loop( // 更新当前用户（审批流程需要） *current_user_id.lock().await = from.to_string(); + // 同步到 session（executor 从 session 读取 user_id） + if let Some(conv) = &conversation { + conv.session().lock().await.current_user_id = from.to_string(); + } // LLM 回复（异步执行，避免审批阻塞主循环） if enable_llm {