feat(storage): 历史压缩保留原消息,前端展示完整对话、LLM 用压缩摘要
将压缩流程从物理删除原消息改为标记位 is_compacted 分离两个视图: - LLM 视角(load_messages_for_topic):只看 is_compacted=0 的压缩摘要+新消息 - UI 视角(load_messages_for_topic_full):看原消息(含 is_compacted=1)+新消息,排除摘要 新增 compact_topic_history:不删除原消息,仅打标记+插入摘要, 同事务内删除旧摘要避免累积。修复 token 统计因压缩丢失 usage 的问题。
This commit is contained in:
parent
bb2774c6b5
commit
81fbee4878
@ -79,7 +79,7 @@ async fn handle_get_current_session(
|
||||
// 直读 DB 按话题加载消息(不依赖 SessionManager 内存,重启后也能正确获取历史)
|
||||
let messages = handler
|
||||
.store
|
||||
.load_messages_for_topic(topic_id, Some(&topic.session_id))
|
||||
.load_messages_for_topic_full(topic_id, Some(&topic.session_id))
|
||||
.map_err(|e| CommandError::new("LOAD_MESSAGES_ERROR", e.to_string()))?;
|
||||
|
||||
let actual_message_count = messages.len();
|
||||
|
||||
@ -254,7 +254,7 @@ async fn handle_save_topic(
|
||||
|
||||
let messages = handler
|
||||
.store
|
||||
.load_messages_for_topic(topic_id, Some(&topic_record.session_id))
|
||||
.load_messages_for_topic_full(topic_id, Some(&topic_record.session_id))
|
||||
.map_err(|e| CommandError::new("LOAD_MESSAGES_ERROR", e.to_string()))?;
|
||||
|
||||
tracing::debug!(
|
||||
|
||||
@ -52,17 +52,17 @@ pub(crate) async fn schedule_background_history_compaction(
|
||||
.compress_two_segment(&history, &provider_config)
|
||||
.await?;
|
||||
|
||||
// Replace only this topic's history in DB (not the entire session).
|
||||
// This avoids clobbering other topics' messages during compaction.
|
||||
// 保留原始消息(标记 is_compacted=1)+ 插入压缩摘要,不删除原消息,
|
||||
// 从而让前端仍能展示完整原始对话,LLM 只看压缩后的精简历史。
|
||||
store
|
||||
.replace_topic_history(&session_id, &topic_id, &compressed)
|
||||
.map_err(|e| AgentError::Other(format!("replace_topic_history error: {}", e)))?;
|
||||
.compact_topic_history(&session_id, &topic_id, &compressed)
|
||||
.map_err(|e| AgentError::Other(format!("compact_topic_history error: {}", e)))?;
|
||||
|
||||
tracing::info!(
|
||||
chat_id = %chat_id,
|
||||
topic_id = %topic_id,
|
||||
compressed_msg_count = compressed.len(),
|
||||
"Two-segment compression committed"
|
||||
"Two-segment compression committed (original messages retained)"
|
||||
);
|
||||
|
||||
session_guard.reload_topic_history(&chat_id, &topic_id)?;
|
||||
|
||||
@ -362,7 +362,7 @@ impl InboundProcessor {
|
||||
tokio::spawn(async move {
|
||||
// 从 DB 查询该 topic 的第一条用户消息作为描述生成的依据
|
||||
let first_user_message = store_clone
|
||||
.load_messages_for_topic(&topic_id_clone, None)
|
||||
.load_messages_for_topic_full(&topic_id_clone, None)
|
||||
.ok()
|
||||
.and_then(|msgs| {
|
||||
msgs.into_iter().find(|m| m.role == "user")
|
||||
|
||||
@ -786,7 +786,7 @@ async fn send_topic_history(
|
||||
task_repository: &Arc<dyn TaskRepository>,
|
||||
) -> Result<(), Box<dyn std::error::Error>> {
|
||||
// 加载话题消息,按 session_id 过滤,避免混入子智能体消息
|
||||
let messages = store.load_messages_for_topic(topic_id, Some(session_id))?;
|
||||
let messages = store.load_messages_for_topic_full(topic_id, Some(session_id))?;
|
||||
|
||||
tracing::info!(topic_id = %topic_id, message_count = messages.len(), "Sending topic history");
|
||||
|
||||
|
||||
@ -71,6 +71,16 @@ pub(super) fn ensure_messages_schema(conn: &Connection) -> Result<(), StorageErr
|
||||
)?;
|
||||
}
|
||||
|
||||
// is_compacted: 1 表示该消息是被压缩消费掉的原始消息(前端可见、LLM 不可见)。
|
||||
// 压缩摘要消息 is_compacted=0(LLM 可见),通过 system_context='history_compaction*'
|
||||
// 在前端查询中被排除。保留的原消息(system_guards / 最新 user)is_compacted=0,不重复。
|
||||
if !has_column(conn, "messages", "is_compacted")? {
|
||||
add_column_if_missing(
|
||||
conn,
|
||||
"ALTER TABLE messages ADD COLUMN is_compacted INTEGER NOT NULL DEFAULT 0",
|
||||
)?;
|
||||
}
|
||||
|
||||
// 创建 topic_id 索引(如果不存在)
|
||||
conn.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_messages_topic_seq ON messages(topic_id, seq) WHERE topic_id IS NOT NULL",
|
||||
|
||||
@ -958,6 +958,107 @@ impl SessionStore {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 压缩该 topic 的历史:保留原始消息(标记 is_compacted=1,前端可见、LLM 不可见),
|
||||
/// 并插入压缩摘要消息(is_compacted=0,LLM 可见,前端通过 system_context 过滤排除)。
|
||||
///
|
||||
/// 与 `replace_topic_history` 的区别:不删除原消息,仅打标记,从而让前端仍能展示
|
||||
/// 完整原始对话,同时 LLM 只看到压缩后的精简历史。
|
||||
///
|
||||
/// `new_messages` 是 `compress_two_segment` 的输出,包含:
|
||||
/// - 保留原样的消息(system_guards / 最新 user,保留原 ID)
|
||||
/// - 压缩摘要消息(system_context = history_compaction_*,新 ID)
|
||||
pub fn compact_topic_history(
|
||||
&self,
|
||||
session_id: &str,
|
||||
topic_id: &str,
|
||||
new_messages: &[ChatMessage],
|
||||
) -> Result<(), StorageError> {
|
||||
let mut conn = self.pool.get()?;
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
|
||||
let now = current_timestamp();
|
||||
|
||||
// 分离摘要消息与保留消息(保留消息携带原 ID,摘要消息是新构造的)
|
||||
let (summaries, preserved): (Vec<&ChatMessage>, Vec<&ChatMessage>) = new_messages
|
||||
.iter()
|
||||
.partition(|m| {
|
||||
m.system_context
|
||||
.as_deref()
|
||||
.map_or(false, |sc| sc.starts_with("history_compaction"))
|
||||
});
|
||||
|
||||
// 先删除该 topic 下已有的旧压缩摘要(system_context LIKE 'history_compaction%')。
|
||||
// 旧摘要已被新摘要替代,保留它们只会累积垃圾行(前端和 LLM 都看不到,但占存储)。
|
||||
tx.execute(
|
||||
"DELETE FROM messages \
|
||||
WHERE topic_id = ?1 AND session_id = ?2 \
|
||||
AND system_context LIKE 'history_compaction%'",
|
||||
params![topic_id, session_id],
|
||||
)?;
|
||||
|
||||
// 将该 topic 中未被保留的原消息标记为 is_compacted=1(仅更新尚未标记的行,避免重复写)。
|
||||
// 保留消息(system_guards / 最新 user)保持 is_compacted=0,不重复插入。
|
||||
let preserved_ids: Vec<String> =
|
||||
preserved.iter().map(|m| m.id.clone()).collect();
|
||||
if preserved_ids.is_empty() {
|
||||
tx.execute(
|
||||
"UPDATE messages SET is_compacted = 1 \
|
||||
WHERE topic_id = ?1 AND session_id = ?2 AND is_compacted = 0",
|
||||
params![topic_id, session_id],
|
||||
)?;
|
||||
} else {
|
||||
let placeholders = (0..preserved_ids.len())
|
||||
.map(|_| "?")
|
||||
.collect::<Vec<_>>()
|
||||
.join(",");
|
||||
let sql = format!(
|
||||
"UPDATE messages SET is_compacted = 1 \
|
||||
WHERE topic_id = ? AND session_id = ? AND is_compacted = 0 \
|
||||
AND id NOT IN ({})",
|
||||
placeholders
|
||||
);
|
||||
let mut params_vec: Vec<String> = vec![topic_id.to_string(), session_id.to_string()];
|
||||
params_vec.extend(preserved_ids.iter().cloned());
|
||||
tx.execute(&sql, rusqlite::params_from_iter(params_vec))?;
|
||||
}
|
||||
|
||||
// 插入压缩摘要消息(is_compacted=0,由列默认值保证)
|
||||
let start_seq: i64 = tx.query_row(
|
||||
"SELECT COALESCE(MAX(seq), 0) + 1 FROM messages WHERE session_id = ?1",
|
||||
params![session_id],
|
||||
|row| row.get(0),
|
||||
)?;
|
||||
for (i, message) in summaries.iter().enumerate() {
|
||||
let seq = start_seq + i as i64;
|
||||
insert_message_with_topic_seq(&tx, session_id, topic_id, seq, message)?;
|
||||
}
|
||||
|
||||
// 更新 topic / session 计数(基于该 topic 全部消息,含被压缩的原始消息)
|
||||
let topic_count: i64 = tx.query_row(
|
||||
"SELECT COUNT(*) FROM messages WHERE session_id = ?1 AND topic_id = ?2",
|
||||
params![session_id, topic_id],
|
||||
|row| row.get(0),
|
||||
)?;
|
||||
tx.execute(
|
||||
"UPDATE topics SET message_count = ?2, last_active_at = ?3, updated_at = ?3 WHERE id = ?1",
|
||||
params![topic_id, topic_count, now],
|
||||
)?;
|
||||
let (total_count, user_turn_count): (i64, i64) = tx.query_row(
|
||||
"SELECT COUNT(*), COALESCE(SUM(CASE WHEN role = 'user' THEN 1 ELSE 0 END), 0) \
|
||||
FROM messages WHERE session_id = ?1",
|
||||
params![session_id],
|
||||
|row| Ok((row.get(0)?, row.get(1)?)),
|
||||
)?;
|
||||
tx.execute(
|
||||
"UPDATE sessions SET message_count = ?2, user_turn_count = ?3, \
|
||||
updated_at = ?4, last_active_at = ?4, archived_at = NULL \
|
||||
WHERE id = ?1 AND deleted_at IS NULL",
|
||||
params![session_id, total_count, user_turn_count, now],
|
||||
)?;
|
||||
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn mark_agent_prompt_reinjected(&self, session_id: &str) -> Result<(), StorageError> {
|
||||
let now = current_timestamp();
|
||||
let conn = self.pool.get()?;
|
||||
@ -1556,6 +1657,8 @@ impl SessionStore {
|
||||
load_messages_after(&conn, session_id, 0)
|
||||
}
|
||||
|
||||
/// LLM 视角:只返回 is_compacted = 0 的消息(压缩摘要 + 未被压缩的新消息)。
|
||||
/// 被压缩消费掉的原始消息(is_compacted = 1)对 LLM 不可见,以节省 context。
|
||||
pub fn load_messages_for_topic(
|
||||
&self,
|
||||
topic_id: &str,
|
||||
@ -1563,12 +1666,56 @@ impl SessionStore {
|
||||
) -> Result<Vec<ChatMessage>, StorageError> {
|
||||
let conn = self.pool.get()?;
|
||||
|
||||
if let Some(sid) = session_id {
|
||||
let mut stmt = conn.prepare(
|
||||
"
|
||||
SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens
|
||||
FROM messages
|
||||
WHERE topic_id = ?1 AND session_id = ?2 AND is_compacted = 0
|
||||
ORDER BY seq ASC
|
||||
",
|
||||
)?;
|
||||
let rows = stmt.query_map(params![topic_id, sid], map_chat_message_row)?;
|
||||
let mut messages = Vec::new();
|
||||
for row in rows {
|
||||
messages.push(row?);
|
||||
}
|
||||
Ok(messages)
|
||||
} else {
|
||||
let mut stmt = conn.prepare(
|
||||
"
|
||||
SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens
|
||||
FROM messages
|
||||
WHERE topic_id = ?1 AND is_compacted = 0
|
||||
ORDER BY seq ASC
|
||||
",
|
||||
)?;
|
||||
let rows = stmt.query_map(params![topic_id], map_chat_message_row)?;
|
||||
let mut messages = Vec::new();
|
||||
for row in rows {
|
||||
messages.push(row?);
|
||||
}
|
||||
Ok(messages)
|
||||
}
|
||||
}
|
||||
|
||||
/// UI 视角:返回原始消息(含被压缩消费的 is_compacted=1 消息)+ 未压缩新消息,
|
||||
/// 排除压缩摘要消息(system_context LIKE 'history_compaction%')。
|
||||
/// 用于前端历史展示、/current、/save-topic、topic 描述生成等场景。
|
||||
pub fn load_messages_for_topic_full(
|
||||
&self,
|
||||
topic_id: &str,
|
||||
session_id: Option<&str>,
|
||||
) -> Result<Vec<ChatMessage>, StorageError> {
|
||||
let conn = self.pool.get()?;
|
||||
|
||||
if let Some(sid) = session_id {
|
||||
let mut stmt = conn.prepare(
|
||||
"
|
||||
SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens
|
||||
FROM messages
|
||||
WHERE topic_id = ?1 AND session_id = ?2
|
||||
AND (system_context IS NULL OR system_context NOT LIKE 'history_compaction%')
|
||||
ORDER BY seq ASC
|
||||
",
|
||||
)?;
|
||||
@ -1584,6 +1731,7 @@ impl SessionStore {
|
||||
SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json, tool_duration_ms, prompt_tokens, completion_tokens, total_tokens, context_window_tokens
|
||||
FROM messages
|
||||
WHERE topic_id = ?1
|
||||
AND (system_context IS NULL OR system_context NOT LIKE 'history_compaction%')
|
||||
ORDER BY seq ASC
|
||||
",
|
||||
)?;
|
||||
|
||||
@ -28,6 +28,14 @@ pub trait ConversationRepository: Send + Sync + 'static {
|
||||
session_id: Option<&str>,
|
||||
) -> Result<Vec<ChatMessage>, StorageError>;
|
||||
|
||||
/// UI 视角:返回原始消息(含被压缩消费的 is_compacted=1 消息)+ 未压缩新消息,
|
||||
/// 排除压缩摘要消息(system_context LIKE 'history_compaction%')。
|
||||
fn load_messages_for_topic_full(
|
||||
&self,
|
||||
topic_id: &str,
|
||||
session_id: Option<&str>,
|
||||
) -> Result<Vec<ChatMessage>, StorageError>;
|
||||
|
||||
fn append_message(&self, session_id: &str, message: &ChatMessage) -> Result<(), StorageError>;
|
||||
|
||||
fn append_message_with_topic(
|
||||
@ -74,6 +82,15 @@ pub trait ConversationRepository: Send + Sync + 'static {
|
||||
topic_id: &str,
|
||||
messages: &[ChatMessage],
|
||||
) -> Result<(), StorageError>;
|
||||
|
||||
/// 压缩该 topic 的历史:保留原始消息(标记 is_compacted=1)+ 插入压缩摘要。
|
||||
/// 不删除原消息,让前端仍能展示完整原始对话,LLM 只看压缩后的精简历史。
|
||||
fn compact_topic_history(
|
||||
&self,
|
||||
session_id: &str,
|
||||
topic_id: &str,
|
||||
new_messages: &[ChatMessage],
|
||||
) -> Result<(), StorageError>;
|
||||
}
|
||||
|
||||
pub trait PromptInjectionRepository: Send + Sync + 'static {
|
||||
@ -272,6 +289,23 @@ impl ConversationRepository for super::SessionStore {
|
||||
) -> Result<(), StorageError> {
|
||||
super::SessionStore::replace_topic_history(self, session_id, topic_id, messages)
|
||||
}
|
||||
|
||||
fn load_messages_for_topic_full(
|
||||
&self,
|
||||
topic_id: &str,
|
||||
session_id: Option<&str>,
|
||||
) -> Result<Vec<ChatMessage>, StorageError> {
|
||||
super::SessionStore::load_messages_for_topic_full(self, topic_id, session_id)
|
||||
}
|
||||
|
||||
fn compact_topic_history(
|
||||
&self,
|
||||
session_id: &str,
|
||||
topic_id: &str,
|
||||
new_messages: &[ChatMessage],
|
||||
) -> Result<(), StorageError> {
|
||||
super::SessionStore::compact_topic_history(self, session_id, topic_id, new_messages)
|
||||
}
|
||||
}
|
||||
|
||||
impl PromptInjectionRepository for super::SessionStore {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user