fix(web): token 统计按 topic_id 聚合，避免同 session 多 topic 共享总和
batch_session_token_stats 重命名为 batch_topic_token_stats，SQL 从 GROUP BY session_id 改为 GROUP BY topic_id，修复同一 session 下多个 topic 显示相同 token 总和的问题。 子代理天然隔离：子代理消息的 topic_id 属于子代理自身的 topic，不在主 topic 列表中。topic_id IN (...) 自动排除 NULL topic_id 的旧消息。 对抗性审查补充：latest assistant 消息查询的 JOIN 条件 m.topic_id = latest.topic_id AND m.seq = latest.max_seq 可能匹配到子代理消息（seq 是 session 级递增，主/子 session 可能有相同 seq）。在外层 JOIN 添加 WHERE m.session_id NOT LIKE 'sub:%'，避免重复行导致 stats.entry(tid) 被覆盖、结果不确定。
This commit is contained in:
oudecheng
parent
235f57c9ae
commit
deab3699fe
@ -40,9 +40,10 @@ pub struct TopicSummary {
|
|||||||
|
|
||||||
/// 构建 TopicSummary 列表的公共函数。
|
/// 构建 TopicSummary 列表的公共函数。
|
||||||
///
|
///
|
||||||
/// 一次批量查询所有 topic 对应 session 的 token 统计，避免 N 次 RTT。
|
/// 一次批量查询所有 topic 的 token 统计，按 `topic_id` 聚合，
|
||||||
/// 按 `session_id` 聚合天然分离主 agent 与子 agent（子代理 session_id 形如
|
/// 避免按 session_id 聚合时同 session 下多个 topic 共享同一总和。
|
||||||
/// `sub:...`，不在此查询的主 session_id 列表中）。
|
/// 子代理天然隔离：子代理消息的 topic_id 属于子代理自身的 topic，
|
||||||
|
/// 不在主 topic 列表中。
|
||||||
///
|
///
|
||||||
/// `context_window_tokens` 来自最新 assistant 消息记录（LLM 调用时持久化），
|
/// `context_window_tokens` 来自最新 assistant 消息记录（LLM 调用时持久化），
|
||||||
/// 无需从配置链路注入，保持存储层与配置解耦。
|
/// 无需从配置链路注入，保持存储层与配置解耦。
|
||||||
@ -54,24 +55,24 @@ pub fn build_topic_summaries(
|
|||||||
return Ok(Vec::new());
|
return Ok(Vec::new());
|
||||||
}
|
}
|
||||||
|
|
||||||
// 收集所有 topic 的 session_id（去重）
|
// 收集所有 topic_id（去重）
|
||||||
let mut session_ids: Vec<String> = Vec::new();
|
let mut topic_ids: Vec<String> = Vec::new();
|
||||||
for t in &topics {
|
for t in &topics {
|
||||||
if !session_ids.contains(&t.session_id) {
|
if !topic_ids.contains(&t.id) {
|
||||||
session_ids.push(t.session_id.clone());
|
topic_ids.push(t.id.clone());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
let session_id_refs: Vec<&str> = session_ids.iter().map(|s| s.as_str()).collect();
|
let topic_id_refs: Vec<&str> = topic_ids.iter().map(|s| s.as_str()).collect();
|
||||||
|
|
||||||
// 一次批量查询 token 统计
|
// 一次批量查询 token 统计（按 topic_id 聚合）
|
||||||
let stats_map: HashMap<String, SessionTokenStats> = store
|
let stats_map: HashMap<String, SessionTokenStats> = store
|
||||||
.batch_session_token_stats(&session_id_refs)
|
.batch_topic_token_stats(&topic_id_refs)
|
||||||
.map_err(|e| CommandError::new("TOKEN_STATS_ERROR", e.to_string()))?;
|
.map_err(|e| CommandError::new("TOKEN_STATS_ERROR", e.to_string()))?;
|
||||||
|
|
||||||
let summaries = topics
|
let summaries = topics
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|t| {
|
.map(|t| {
|
||||||
let token_stats = stats_map.get(&t.session_id).map(|s| TopicTokenStats {
|
let token_stats = stats_map.get(&t.id).map(|s| TopicTokenStats {
|
||||||
prompt_tokens: s.prompt_tokens,
|
prompt_tokens: s.prompt_tokens,
|
||||||
completion_tokens: s.completion_tokens,
|
completion_tokens: s.completion_tokens,
|
||||||
total_tokens: s.total_tokens,
|
total_tokens: s.total_tokens,
|
||||||
@ -140,8 +141,6 @@ async fn handle_list_topics(
|
|||||||
.list_topics(&session_id)
|
.list_topics(&session_id)
|
||||||
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?;
|
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?;
|
||||||
|
|
||||||
// context_window_tokens 当前未从配置链路注入（前端可从已有 config 接口获取），
|
|
||||||
// 此处传 0 表示"后端不提供上限"，前端按需隐藏百分比。
|
|
||||||
let summaries = build_topic_summaries(handler.store.as_ref(), topics)?;
|
let summaries = build_topic_summaries(handler.store.as_ref(), topics)?;
|
||||||
|
|
||||||
let topics_json = serde_json::to_string(&summaries)
|
let topics_json = serde_json::to_string(&summaries)
|
||||||
|
|||||||
@ -1630,32 +1630,44 @@ impl SessionStore {
|
|||||||
.map_err(StorageError::from)
|
.map_err(StorageError::from)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 批量查询多个 session 的 token 消耗统计（cost 累计 + context 瞬时）。
|
/// 批量查询多个 topic 的 token 消耗统计（cost 累计 + context 瞬时）。
|
||||||
pub fn batch_session_token_stats(
|
///
|
||||||
|
/// 按 `topic_id` 聚合而非 `session_id`：一个 session 可包含多个 topic，
|
||||||
|
/// 若按 session_id 聚合会导致同 session 下的所有 topic 显示相同的总和。
|
||||||
|
///
|
||||||
|
/// 子代理隔离：子代理消息持久化时 session_id='sub:...'，topic_id=父 topic_id
|
||||||
|
/// （见 task::runtime PersistingEmittedMessageHandler 构造），因此不能仅靠
|
||||||
|
/// topic_id 隔离。此处用 `session_id NOT LIKE 'sub:%'` 显式排除子代理消息，
|
||||||
|
/// 与项目约定一致（session 列表同样过滤 'sub:%'）。子代理 token 不计入父 topic，
|
||||||
|
/// 保持"子代理分别计算"语义。
|
||||||
|
pub fn batch_topic_token_stats(
|
||||||
&self,
|
&self,
|
||||||
session_ids: &[&str],
|
topic_ids: &[&str],
|
||||||
) -> Result<HashMap<String, SessionTokenStats>, StorageError> {
|
) -> Result<HashMap<String, SessionTokenStats>, StorageError> {
|
||||||
if session_ids.is_empty() {
|
if topic_ids.is_empty() {
|
||||||
return Ok(HashMap::new());
|
return Ok(HashMap::new());
|
||||||
}
|
}
|
||||||
let conn = self.pool.get()?;
|
let conn = self.pool.get()?;
|
||||||
|
|
||||||
let placeholders = (0..session_ids.len())
|
let placeholders = (0..topic_ids.len())
|
||||||
.map(|i| format!("?{}", i + 1))
|
.map(|i| format!("?{}", i + 1))
|
||||||
.collect::<Vec<_>>()
|
.collect::<Vec<_>>()
|
||||||
.join(", ");
|
.join(", ");
|
||||||
|
// topic_id IN (...) 自动排除 NULL topic_id 的旧消息；
|
||||||
|
// session_id NOT LIKE 'sub:%' 排除子代理消息（其 topic_id=父 topic_id）
|
||||||
let sum_sql = format!(
|
let sum_sql = format!(
|
||||||
"SELECT session_id, \
|
"SELECT topic_id, \
|
||||||
COALESCE(SUM(prompt_tokens), 0) AS sum_prompt, \
|
COALESCE(SUM(prompt_tokens), 0) AS sum_prompt, \
|
||||||
COALESCE(SUM(completion_tokens), 0) AS sum_completion, \
|
COALESCE(SUM(completion_tokens), 0) AS sum_completion, \
|
||||||
COALESCE(SUM(total_tokens), 0) AS sum_total \
|
COALESCE(SUM(total_tokens), 0) AS sum_total \
|
||||||
FROM messages \
|
FROM messages \
|
||||||
WHERE session_id IN ({placeholders}) AND role = 'assistant' \
|
WHERE topic_id IN ({placeholders}) AND role = 'assistant' \
|
||||||
GROUP BY session_id"
|
AND session_id NOT LIKE 'sub:%' \
|
||||||
|
GROUP BY topic_id"
|
||||||
);
|
);
|
||||||
|
|
||||||
let mut stmt = conn.prepare(&sum_sql)?;
|
let mut stmt = conn.prepare(&sum_sql)?;
|
||||||
let params: Vec<&dyn rusqlite::ToSql> = session_ids
|
let params: Vec<&dyn rusqlite::ToSql> = topic_ids
|
||||||
.iter()
|
.iter()
|
||||||
.map(|s| s as &dyn rusqlite::ToSql)
|
.map(|s| s as &dyn rusqlite::ToSql)
|
||||||
.collect();
|
.collect();
|
||||||
@ -1674,26 +1686,34 @@ impl SessionStore {
|
|||||||
|
|
||||||
let mut stats: HashMap<String, SessionTokenStats> = HashMap::new();
|
let mut stats: HashMap<String, SessionTokenStats> = HashMap::new();
|
||||||
for row in sum_rows {
|
for row in sum_rows {
|
||||||
let (sid, s) = row?;
|
let (tid, s) = row?;
|
||||||
stats.insert(sid, s);
|
stats.insert(tid, s);
|
||||||
}
|
}
|
||||||
|
|
||||||
// 查找每个 session 中最新的**有 usage 数据的** assistant 消息，
|
// 查找每个 topic 中最新的**有 usage 数据的** assistant 消息，
|
||||||
// 读取其 prompt_tokens 和 context_window_tokens。
|
// 读取其 prompt_tokens 和 context_window_tokens。
|
||||||
// 过滤 prompt_tokens IS NOT NULL 确保跳过 error/cancel 消息（usage 为 NULL）。
|
// 过滤 prompt_tokens IS NOT NULL 确保跳过 error/cancel 消息（usage 为 NULL）；
|
||||||
|
// session_id NOT LIKE 'sub:%' 排除子代理消息，避免取到子代理的 context_window。
|
||||||
|
//
|
||||||
|
// 注意：seq 是 session 级递增（见 append_message_with_topic），主 session 与
|
||||||
|
// 子代理 session 各自独立计数，可能存在相同 seq。外层 WHERE 必须再次过滤
|
||||||
|
// session_id NOT LIKE 'sub:%'，否则 JOIN 会同时匹配主消息和子代理消息，
|
||||||
|
// 导致重复行并使 stats.entry(tid) 被覆盖，结果不确定。
|
||||||
let last_sql = format!(
|
let last_sql = format!(
|
||||||
"SELECT m.session_id, m.prompt_tokens, m.context_window_tokens \
|
"SELECT m.topic_id, m.prompt_tokens, m.context_window_tokens \
|
||||||
FROM messages m \
|
FROM messages m \
|
||||||
INNER JOIN ( \
|
INNER JOIN ( \
|
||||||
SELECT session_id, MAX(seq) AS max_seq \
|
SELECT topic_id, MAX(seq) AS max_seq \
|
||||||
FROM messages \
|
FROM messages \
|
||||||
WHERE session_id IN ({placeholders}) AND role = 'assistant' \
|
WHERE topic_id IN ({placeholders}) AND role = 'assistant' \
|
||||||
AND prompt_tokens IS NOT NULL \
|
AND prompt_tokens IS NOT NULL \
|
||||||
GROUP BY session_id \
|
AND session_id NOT LIKE 'sub:%' \
|
||||||
) latest ON m.session_id = latest.session_id AND m.seq = latest.max_seq"
|
GROUP BY topic_id \
|
||||||
|
) latest ON m.topic_id = latest.topic_id AND m.seq = latest.max_seq \
|
||||||
|
WHERE m.session_id NOT LIKE 'sub:%'"
|
||||||
);
|
);
|
||||||
let mut stmt2 = conn.prepare(&last_sql)?;
|
let mut stmt2 = conn.prepare(&last_sql)?;
|
||||||
let params2: Vec<&dyn rusqlite::ToSql> = session_ids
|
let params2: Vec<&dyn rusqlite::ToSql> = topic_ids
|
||||||
.iter()
|
.iter()
|
||||||
.map(|s| s as &dyn rusqlite::ToSql)
|
.map(|s| s as &dyn rusqlite::ToSql)
|
||||||
.collect();
|
.collect();
|
||||||
@ -1705,8 +1725,8 @@ impl SessionStore {
|
|||||||
))
|
))
|
||||||
})?;
|
})?;
|
||||||
for row in last_rows {
|
for row in last_rows {
|
||||||
let (sid, last_prompt, last_ctx_window) = row?;
|
let (tid, last_prompt, last_ctx_window) = row?;
|
||||||
let entry = stats.entry(sid).or_insert(SessionTokenStats {
|
let entry = stats.entry(tid).or_insert(SessionTokenStats {
|
||||||
prompt_tokens: 0,
|
prompt_tokens: 0,
|
||||||
completion_tokens: 0,
|
completion_tokens: 0,
|
||||||
total_tokens: 0,
|
total_tokens: 0,
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user