From 01592278288504dd7667514476871a549070cfa6 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Tue, 18 Aug 2026 06:55:21 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E7=AC=AC=E4=BA=8C=E6=89=B9=E6=80=A7?= =?UTF-8?q?=E8=83=BD=E4=BF=AE=E5=A4=8D=E2=80=94=E2=80=94blocking=20?= =?UTF-8?q?=E7=BA=BF=E7=A8=8B=E6=B1=A0=E9=9A=94=E7=A6=BB=E3=80=81HTTP=20?= =?UTF-8?q?=E5=AE=A2=E6=88=B7=E7=AB=AF=E5=A4=8D=E7=94=A8=E3=80=81=E5=89=8D?= =?UTF-8?q?=E7=AB=AF=20memo=20=E4=B8=8E=E6=B5=81=E5=BC=8F=E8=8A=82?= =?UTF-8?q?=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 同步阻塞操作（附件处理、历史加载、scheduler/memory_search 的 SQLite 调用） 移入 spawn_blocking，避免占用 async worker - LLM Provider reqwest::Client 按超时配置缓存复用，减少 TLS/连接开销 - agent loop：图片过滤加廉价预判避免全量深拷贝；请求克隆改借用；工具定义 Arc 化 - 定向 COUNT/LIMIT 1 查询替代全量加载计数（wait_coordinator、task session 重建） - 前端：面板/侧栏/聊天组件 memo 化；merged_tool 对象按值复用缓存； 流式 delta rAF 节流批量 flush；useMemo 缓存分组排序结果 --- src/agent/agent_loop.rs | 52 ++++- src/agent/context_compressor.rs | 4 +- src/command/handlers/list_sessions.rs | 8 +- src/command/handlers/load_task_messages.rs | 12 +- src/command/handlers/save_session.rs | 20 +- src/gateway/memory_maintenance.rs | 4 +- src/gateway/wait_coordinator.rs | 5 +- src/gateway/ws.rs | 178 ++++++++++++------ src/providers/anthropic.rs | 12 +- src/providers/mod.rs | 35 ++++ src/providers/openai.rs | 21 +-- src/providers/traits.rs | 9 +- src/scheduler/mod.rs | 146 ++++++++++---- src/storage/mod.rs | 46 +++++ src/tools/memory_search.rs | 46 ++++- src/topic_description.rs | 2 +- tests/test_integration.rs | 4 +- tests/test_tool_calling.rs | 14 +- web/src/App.tsx | 35 +++- web/src/components/Chat/ChatContainer.tsx | 8 +- web/src/components/Chat/ExpertSelector.tsx | 7 +- web/src/components/Chat/MessageInput.tsx | 8 +- web/src/components/Chat/MessageList.tsx | 8 +- web/src/components/Chat/ModelSelector.tsx | 7 +- web/src/components/Panel/MemoryPanel.tsx | 42 +++-- web/src/components/Panel/SkillList.tsx | 49 +++-- web/src/components/Panel/TodoPanel.tsx | 16 +- .../components/Panel/TopicTokenStatsPanel.tsx | 9 +- .../components/Sidebar/SchedulerJobList.tsx | 12 +- web/src/components/Sidebar/TopicList.tsx | 8 +- 30 files changed, 581 insertions(+), 246 deletions(-) diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index fb0c6cf..a4a9429 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -258,6 +258,43 @@ fn filter_images_by_age_and_count( // 消息列表顺序：[old, ..., new]，所以末尾是最新的 let msg_count = messages.len(); + // 廉价预判：完整复刻下方保留逻辑，仅判断是否"确有图片会被移除"。 + // 有图片但无需过滤时（常见情形）直接返回借用，避免每轮 LLM 迭代 + // 对整个历史做全量深拷贝。 + let mut dry_run_kept = 0usize; + for (idx, message) in messages.iter().enumerate().rev() { + let age_from_end = msg_count.saturating_sub(idx).saturating_sub(1); + let exceeds_age_limit = max_age_rounds > 0 && age_from_end >= max_age_rounds; + let image_count_in_msg = message + .media_refs + .iter() + .filter(|p| supported_image_mime_type(p).is_some()) + .count(); + if image_count_in_msg == 0 { + continue; + } + if exceeds_age_limit { + // 该消息的图片全部会被过滤 + return filter_images_by_age_and_count_inner(messages, max_age_rounds, max_images); + } + let can_keep = std::cmp::min(image_count_in_msg, max_images.saturating_sub(dry_run_kept)); + if can_keep < image_count_in_msg { + // 超出数量上限的图片会被过滤 + return filter_images_by_age_and_count_inner(messages, max_age_rounds, max_images); + } + dry_run_kept += can_keep; + } + Cow::Borrowed(messages) +} + +/// `filter_images_by_age_and_count` 的重建实现：仅在预判确认有图片需要过滤时调用。 +fn filter_images_by_age_and_count_inner( + messages: &[ChatMessage], + max_age_rounds: usize, + max_images: usize, +) -> Cow<'_, [ChatMessage]> { + let msg_count = messages.len(); + // 先从后向前遍历，计算每条消息应该保留多少张图片 // 使用 Vec 存储每条消息应该保留的图片数量 let mut images_to_keep_per_msg: Vec = vec![0; msg_count]; @@ -1132,7 +1169,8 @@ impl AgentLoop { let tools = if tool_defs.is_empty() { None } else { - Some(tool_defs) + // Arc 共享：process() 内多轮迭代只读复用，避免每轮深拷贝全部工具定义 + Some(std::sync::Arc::new(tool_defs)) }; // 工具 token 估算在循环外算一次（tool_defs 在 process() 期间不变）， // 避免每轮 serde_json::to_string 全量序列化工具定义。 @@ -1247,7 +1285,7 @@ impl AgentLoop { self.emit_live_tool_call_message(cancel.final_response.clone()).await; return Ok(cancel); } - result = self.provider.chat_with_streaming(request.clone(), stream_callback.clone()) => { + result = self.provider.chat_with_streaming(&request, stream_callback.clone()) => { llm_result = result; } } @@ -1255,7 +1293,7 @@ impl AgentLoop { } else { llm_result = self .provider - .chat_with_streaming(request.clone(), stream_callback) + .chat_with_streaming(&request, stream_callback) .await; } @@ -1636,7 +1674,7 @@ impl AgentLoop { &self, messages: &[ChatMessage], system_prompt_context: Option<&SystemPromptContext>, - tools: Option>, + tools: Option>>, tools_tokens: usize, ) -> ChatCompletionRequest { let filtered_messages = filter_images_by_age_and_count( @@ -1827,12 +1865,12 @@ impl AgentLoop { self.emit_live_tool_call_message(cancel.final_response.clone()).await; return cancel; } - result = self.provider.chat(request.clone()) => { + result = self.provider.chat(&request) => { final_result = result; } } } else { - final_result = self.provider.chat(request.clone()).await; + final_result = self.provider.chat(&request).await; } match final_result { @@ -3320,7 +3358,7 @@ mod tests { impl LLMProvider for MockProvider { async fn chat( &self, - _request: ChatCompletionRequest, + _request: &ChatCompletionRequest, ) -> Result> { let mut responses = self.responses.lock().unwrap(); if responses.is_empty() { diff --git a/src/agent/context_compressor.rs b/src/agent/context_compressor.rs index f1e8d6f..7a530c6 100644 --- a/src/agent/context_compressor.rs +++ b/src/agent/context_compressor.rs @@ -458,7 +458,7 @@ OLDER SEGMENT (events from earlier in the session): }; let response = provider - .chat(request) + .chat(&request) .await .map_err(|e| AgentError::LlmError(e.to_string()))?; Ok(response.content) @@ -540,7 +540,7 @@ OLDER SEGMENT (events from earlier in the session): }; let response = provider - .chat(request) + .chat(&request) .await .map_err(|e| AgentError::LlmError(e.to_string()))?; Ok(response.content) diff --git a/src/command/handlers/list_sessions.rs b/src/command/handlers/list_sessions.rs index cc112d3..795a38d 100644 --- a/src/command/handlers/list_sessions.rs +++ b/src/command/handlers/list_sessions.rs @@ -73,11 +73,9 @@ async fn handle_list_sessions( let is_current = topic.id == current_topic_id; let marker = if is_current { " *" } else { "" }; - // 使用辅助方法获取消息数量 - let msg_count = handler - .store - .get_topic_message_count(&topic.id) - .unwrap_or(0); + // 使用 topics 表预计算列，避免循环内 N+1 次 COUNT 查询 + // （message_count 与 list_topics 的展示口径一致） + let msg_count = topic.message_count; lines.push(format!( "{}. {}{} ({})", diff --git a/src/command/handlers/load_task_messages.rs b/src/command/handlers/load_task_messages.rs index f14e37e..b07326c 100644 --- a/src/command/handlers/load_task_messages.rs +++ b/src/command/handlers/load_task_messages.rs @@ -141,15 +141,15 @@ fn reconstruct_task_from_db( store: &SessionStore, task_id: &str, ) -> Result, CommandError> { - let sessions = store - .find_sessions_by_id_suffix(&format!(":{}", task_id)) + let record = store + .find_first_session_by_id_suffix(&format!(":{}", task_id)) .map_err(|e| CommandError::new("DB_ERROR", e.to_string()))?; - if sessions.is_empty() { - return Ok(None); - } + let record = match record { + Some(record) => record, + None => return Ok(None), + }; - let record = &sessions[0]; let session_id = record.id.clone(); // Extract parent_session_id from session_id: "sub:{parent}:task:{uuid}" diff --git a/src/command/handlers/save_session.rs b/src/command/handlers/save_session.rs index 365cb89..8b37f7a 100644 --- a/src/command/handlers/save_session.rs +++ b/src/command/handlers/save_session.rs @@ -33,7 +33,7 @@ pub async fn save_session_to_file( store: &SessionStore, task_repository: Option<&dyn TaskRepository>, system_prompt_provider: &dyn SystemPromptProvider, -) -> Result { +) -> Result<(PathBuf, usize), String> { // 获取会话记录 let record = store .get_session(session_id) @@ -83,7 +83,8 @@ pub async fn save_session_to_file( // 写入文件 std::fs::write(&output_path, markdown).map_err(|e| format!("Failed to write file: {}", e))?; - Ok(output_path) + // 返回已加载的消息数，调用方无需为此再次全量加载消息 + Ok((output_path, messages.len())) } /// 保存会话命令处理器 @@ -187,8 +188,8 @@ async fn handle_save_session( } } - // 调用公共函数 - let output_path = save_session_to_file( + // 调用公共函数（返回路径与已加载消息数，避免二次全量加载只为计数） + let (output_path, message_count) = save_session_to_file( session_id, filepath, include_all, @@ -200,15 +201,6 @@ async fn handle_save_session( .await .map_err(|e| CommandError::new("SAVE_ERROR", e))?; - // 根据 include_all 获取消息数量 - let message_count = if include_all { - handler.store.load_all_messages(session_id) - } else { - handler.store.load_messages(session_id) - } - .map_err(|e| CommandError::new("LOAD_MESSAGES_ERROR", e.to_string()))? - .len(); - Ok(CommandResponse::success(ctx.request_id) .with_message( MessageKind::Notification, @@ -716,7 +708,7 @@ impl InChatCommandHandler for SaveSessionInChatHandler { // 返回成功或失败消息 match result { - Ok(output_path) => { + Ok((output_path, _message_count)) => { let msg = format!( "Session saved to: {}", output_path.display().to_string().replace('\\', "/") diff --git a/src/gateway/memory_maintenance.rs b/src/gateway/memory_maintenance.rs index 87fe271..bb47c37 100644 --- a/src/gateway/memory_maintenance.rs +++ b/src/gateway/memory_maintenance.rs @@ -192,7 +192,7 @@ impl MemoryMaintenanceService { .chain(std::iter::once(None)) .enumerate() { - let response = match provider.chat(request.clone()).await { + let response = match provider.chat(&request).await { Ok(success) => success, Err(err) => { let error_text = err.to_string(); @@ -313,7 +313,7 @@ impl MemoryMaintenanceService { .chain(std::iter::once(None)) .enumerate() { - match provider.chat(request.clone()).await { + match provider.chat(&request).await { Ok(success) => { response = Some(success); break; diff --git a/src/gateway/wait_coordinator.rs b/src/gateway/wait_coordinator.rs index eb84e50..abadbcd 100644 --- a/src/gateway/wait_coordinator.rs +++ b/src/gateway/wait_coordinator.rs @@ -117,11 +117,10 @@ impl WaitCoordinator for SessionWaitCoordinator { }; // 3.5. 记录等待前的用户消息数量（用于 wakeup 后提取新注入的消息） - // 直接从 SQLite 读取，不持有任何锁 + // 直接从 SQLite 读取（定向 COUNT，避免全量加载消息体），不持有任何锁 let user_msg_count_before = self .store - .load_messages_for_topic(&self.topic_id, None) - .map(|msgs| msgs.iter().filter(|m| m.role == "user").count()) + .count_user_messages_for_topic(&self.topic_id) .unwrap_or(0); // 4. 释放 serial_lock（取出 guard 并 drop） diff --git a/src/gateway/ws.rs b/src/gateway/ws.rs index 2e24409..3ee2d60 100644 --- a/src/gateway/ws.rs +++ b/src/gateway/ws.rs @@ -186,23 +186,43 @@ async fn handle_socket(ws: WebSocket, state: Arc) { let store = state.session_manager.store(); // 1. 查询 websocket 和 cli 两个通道的 Sessions（兼容旧版本 cli 通道创建的会话） - let mut websocket_sessions = store.list_sessions("websocket", false).unwrap_or_default(); - let cli_channel_sessions = store.list_sessions("cli", false).unwrap_or_default(); - websocket_sessions.extend(cli_channel_sessions); - websocket_sessions.sort_by_key(|s| -(s.last_active_at)); + // SQLite 同步查询移入 blocking 线程池，避免连接建立时阻塞 async worker。 + let store_for_init = store.clone(); + let cli_sessions_for_init = cli_sessions.clone(); + let (mut websocket_sessions, initial_result) = tokio::task::spawn_blocking(move || { + let mut websocket_sessions = store_for_init + .list_sessions("websocket", false) + .unwrap_or_default(); + let cli_channel_sessions = store_for_init + .list_sessions("cli", false) + .unwrap_or_default(); + websocket_sessions.extend(cli_channel_sessions); + websocket_sessions.sort_by_key(|s| -(s.last_active_at)); - // 2. 如果没有，自动创建一个默认 Session - let initial_record = if websocket_sessions.is_empty() { - match cli_sessions.create_with_channel("websocket", Some("默认会话")) { - Ok(record) => record, - Err(e) => { - tracing::error!(error = %e, "Failed to create initial WebSocket session"); - return; + // 2. 如果没有，自动创建一个默认 Session + let initial_record = if websocket_sessions.is_empty() { + match cli_sessions_for_init.create_with_channel("websocket", Some("默认会话")) { + Ok(record) => Some(record), + Err(e) => { + tracing::error!(error = %e, "Failed to create initial WebSocket session"); + None + } } - } - } else { - // 使用最新的 Session - websocket_sessions[0].clone() + } else { + // 使用最新的 Session + Some(websocket_sessions[0].clone()) + }; + (websocket_sessions, initial_record) + }) + .await + .unwrap_or_else(|e| { + tracing::error!(error = %e, "WebSocket session init task failed"); + (Vec::new(), None) + }); + + let initial_record = match initial_result { + Some(record) => record, + None => return, }; let runtime_session_id = uuid::Uuid::new_v4().to_string(); @@ -414,7 +434,14 @@ async fn handle_inbound( .await; // Process attachments: save base64 content to local files and build MediaItems with correct paths - let media = process_attachments_with_base64(attachments)?; + // base64 解码 + 同步写盘是 CPU/IO 密集操作（单条消息最大约 67MB）， + // 移入 blocking 线程池，避免长时间霸占 async worker。 + let media = + tokio::task::spawn_blocking(move || process_attachments_with_base64(attachments)) + .await + .map_err(|e| { + AgentError::Other(format!("Attachment processing task failed: {}", e)) + })??; state .bus @@ -720,24 +747,31 @@ async fn handle_inbound( let _ = sender.send(WsOutbound::MemoryList { memories }).await; } - // 记忆 CRUD 后自动刷新列表 - if response.metadata.get("memory_updated").map(|v| v.as_str()) == Some("true") - && let Ok(records) = - store.list_memories_for_scope("user", crate::storage::GLOBAL_SCOPE_KEY) - { - let memories: Vec = records - .into_iter() - .filter(|m| m.namespace != "_meta") - .map(|m| crate::protocol::MemorySummary { - id: m.id, - namespace: m.namespace, - memory_key: m.memory_key, - content: m.content, - created_at: m.created_at, - updated_at: m.updated_at, - }) - .collect(); - let _ = sender.send(WsOutbound::MemoryList { memories }).await; + // 记忆 CRUD 后自动刷新列表（SQLite 同步查询移入 blocking 线程池） + if response.metadata.get("memory_updated").map(|v| v.as_str()) == Some("true") { + let store_bg = store.clone(); + let records = tokio::task::spawn_blocking(move || { + store_bg.list_memories_for_scope("user", crate::storage::GLOBAL_SCOPE_KEY) + }) + .await; + if let Err(e) = &records { + tracing::warn!(error = %e, "Memory list task failed"); + } + if let Ok(Ok(records)) = records { + let memories: Vec = records + .into_iter() + .filter(|m| m.namespace != "_meta") + .map(|m| crate::protocol::MemorySummary { + id: m.id, + namespace: m.namespace, + memory_key: m.memory_key, + content: m.content, + created_at: m.created_at, + updated_at: m.updated_at, + }) + .collect(); + let _ = sender.send(WsOutbound::MemoryList { memories }).await; + } } // 处理加载聊天消息请求 @@ -828,14 +862,19 @@ async fn send_topic_history( sender: &mpsc::Sender, task_repository: &Arc, ) -> Result<(), Box> { - // 加载话题消息，按 session_id 过滤，避免混入子智能体消息 - let mut messages = store.load_messages_for_topic_full(topic_id, Some(session_id))?; - - // 对账 running 占位：DB 中的 task tool_result 永远保持 spawn 时的 running 状态 - // （实时完成信号只更新前端内存与 pending_subagents 表），若不替换， - // 刷新/切话题后前端卡片会永远显示"运行中"。与 Session::reconcile_running_placeholders - // 同语义，仅改发送副本，不落库。 - reconcile_running_in_messages(&mut messages, store, topic_id); + // 加载话题消息，按 session_id 过滤，避免混入子智能体消息。 + // SQLite 同步加载 + running 占位对账移入 blocking 线程池，避免阻塞 async worker。 + let store_bg = store.clone(); + let topic_id_bg = topic_id.to_string(); + let session_id_bg = session_id.to_string(); + let messages = tokio::task::spawn_blocking(move || { + let mut messages = + store_bg.load_messages_for_topic_full(&topic_id_bg, Some(&session_id_bg))?; + reconcile_running_in_messages(&mut messages, &store_bg, &topic_id_bg); + Ok::<_, crate::storage::StorageError>(messages) + }) + .await + .map_err(|e| format!("Topic history load task failed: {}", e))??; tracing::info!(topic_id = %topic_id, message_count = messages.len(), "Sending topic history"); @@ -850,11 +889,19 @@ async fn send_topic_history( } } - // 将消息转换为 WsOutbound 并发送 - for msg in messages { - for outbound in chat_message_to_ws_outbound(&msg) { - let _ = sender.send(outbound).await; - } + // 将消息转换为 WsOutbound 并发送。 + // 转换过程对每条媒体引用做同步文件读取 + base64 编码（CPU/IO 密集）， + // 整体移入 blocking 线程池一次性产出，避免阻塞 async worker。 + let outbound_batches: Vec> = tokio::task::spawn_blocking(move || { + messages + .iter() + .map(chat_message_to_ws_outbound) + .collect::>() + }) + .await + .map_err(|e| format!("Topic history convert task failed: {}", e))?; + for outbound in outbound_batches.into_iter().flatten() { + let _ = sender.send(outbound).await; } // 查询该话题下所有子智能体任务，补发 TaskStarted 事件 @@ -963,7 +1010,12 @@ async fn send_task_messages( subagent_task_id: Option, task_repository: Option<&Arc>, ) -> Result<(), Box> { - let messages = store.load_messages(session_id)?; + // SQLite 同步加载移入 blocking 线程池，避免阻塞 async worker。 + let store_bg = store.clone(); + let session_id_bg = session_id.to_string(); + let messages = tokio::task::spawn_blocking(move || store_bg.load_messages(&session_id_bg)) + .await + .map_err(|e| format!("Task messages load task failed: {}", e))??; tracing::info!(session_id = %session_id, message_count = messages.len(), "Sending task messages"); @@ -978,16 +1030,26 @@ async fn send_task_messages( } } - for msg in messages { - let mut outbounds = chat_message_to_ws_outbound(&msg); - if let Some(ref task_id) = subagent_task_id { - for ob in &mut outbounds { - set_subagent_task_id(ob, task_id); - } - } - for outbound in outbounds { - let _ = sender.send(outbound).await; - } + // 转换（含媒体文件同步读取 + base64 编码）移入 blocking 线程池一次性产出 + let subagent_task_id_bg = subagent_task_id.clone(); + let outbound_batches: Vec> = tokio::task::spawn_blocking(move || { + messages + .iter() + .map(|msg| { + let mut outbounds = chat_message_to_ws_outbound(msg); + if let Some(ref task_id) = subagent_task_id_bg { + for ob in &mut outbounds { + set_subagent_task_id(ob, task_id); + } + } + outbounds + }) + .collect::>() + }) + .await + .map_err(|e| format!("Task messages convert task failed: {}", e))?; + for outbound in outbound_batches.into_iter().flatten() { + let _ = sender.send(outbound).await; } // 补发子任务（孙智能体）的 TaskStarted 事件 diff --git a/src/providers/anthropic.rs b/src/providers/anthropic.rs index e595820..74c61b4 100644 --- a/src/providers/anthropic.rs +++ b/src/providers/anthropic.rs @@ -3,7 +3,6 @@ use reqwest::Client; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::OnceLock; -use std::time::Duration; use super::traits::Usage; use super::{ChatCompletionRequest, ChatCompletionResponse, LLMProvider, Tool, ToolCall}; @@ -144,10 +143,9 @@ impl AnthropicProvider { max_tokens: Option, model_extra: HashMap, ) -> Self { - let client = Client::builder() - .timeout(Duration::from_secs(llm_timeout_secs)) - .build() - .unwrap_or_else(|_| Client::new()); + // 复用按超时配置的共享 client（TLS 上下文 + 连接池）， + // 避免每条消息重建 Provider 时重复构造。 + let client = crate::providers::shared_llm_http_client(llm_timeout_secs); // 兼容带末尾斜杠的 base_url，避免 format!("{}/v1/messages", base_url) 产生双斜杠 let base_url = base_url.trim_end_matches('/').to_string(); @@ -258,7 +256,7 @@ impl LLMProvider for AnthropicProvider { #[tracing::instrument(skip(self, request), fields(provider = %self.name, model = %self.model_id))] async fn chat( &self, - request: ChatCompletionRequest, + request: &ChatCompletionRequest, ) -> Result> { let url = format!("{}/v1/messages", self.base_url); let max_tokens = request.max_tokens.or(self.max_tokens).unwrap_or(8192); @@ -271,7 +269,7 @@ impl LLMProvider for AnthropicProvider { "Anthropic: sending chat completion request" ); - let tools = request.tools.map(|tools| { + let tools = request.tools.as_ref().map(|tools| { tools .iter() .map(|t: &Tool| AnthropicTool { diff --git a/src/providers/mod.rs b/src/providers/mod.rs index 79782ec..ef4fd34 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -12,6 +12,41 @@ pub use traits::{ StreamCallback, StreamDelta, Usage, }; +/// 共享 HTTP client 缓存：按超时配置复用 reqwest::Client。 +/// +/// reqwest::Client 持有 TLS 上下文与连接池，构造成本高；每条消息重建 +/// Provider 会导致每次 LLM 请求都无法复用 keep-alive 连接。 +/// api_key / base_url / extra_headers 均为 per-request 应用（不影响 client +/// 构造），因此缓存键只需超时值，不同 provider/模型可安全共享同一 client。 +use std::collections::HashMap; +use std::sync::LazyLock; +use std::sync::Mutex; + +static SHARED_HTTP_CLIENTS: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +/// 缓存条目上限：超时配置的取值种类极少（来自模型配置的 llm_timeout_secs）， +/// 超过上限说明配置在频繁变动，直接清空重建（代价仅为缓存失效）。 +const SHARED_HTTP_CLIENT_CACHE_CAP: usize = 16; + +pub(crate) fn shared_llm_http_client(timeout_secs: u64) -> reqwest::Client { + let mut cache = SHARED_HTTP_CLIENTS + .lock() + .unwrap_or_else(|e| e.into_inner()); + if let Some(client) = cache.get(&timeout_secs) { + return client.clone(); + } + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(timeout_secs)) + .build() + .unwrap_or_else(|_| reqwest::Client::new()); + if cache.len() >= SHARED_HTTP_CLIENT_CACHE_CAP { + cache.clear(); + } + cache.insert(timeout_secs, client.clone()); + client +} + pub fn create_provider( config: ProviderRuntimeConfig, ) -> Result, ProviderError> { diff --git a/src/providers/openai.rs b/src/providers/openai.rs index ce53bc4..8b80ff7 100644 --- a/src/providers/openai.rs +++ b/src/providers/openai.rs @@ -5,7 +5,6 @@ use serde::Deserialize; use serde_json::{Value, json}; use std::collections::BTreeMap; use std::collections::HashMap; -use std::time::Duration; use super::traits::{StreamCallback, StreamDelta, Usage}; use super::{ChatCompletionRequest, ChatCompletionResponse, LLMProvider, ToolCall}; @@ -278,10 +277,9 @@ impl OpenAIProvider { max_tokens: Option, model_extra: HashMap, ) -> Self { - let client = Client::builder() - .timeout(Duration::from_secs(llm_timeout_secs)) - .build() - .unwrap_or_else(|_| Client::new()); + // 复用按超时配置的共享 client（TLS 上下文 + 连接池）， + // 避免每条消息重建 Provider 时重复构造。 + let client = crate::providers::shared_llm_http_client(llm_timeout_secs); // 兼容带末尾斜杠的 base_url（如 https://opencode.ai/zen/go/v1/）， // 否则 format!("{}/chat/completions", base_url) 会产生双斜杠导致 404 @@ -1103,12 +1101,12 @@ impl OpenAIUsage { impl LLMProvider for OpenAIProvider { async fn chat( &self, - request: ChatCompletionRequest, + request: &ChatCompletionRequest, ) -> Result> { // 检查是否启用流式输出 if self.is_streaming_enabled() { // 优先尝试流式输出（无回调） - match self.chat_streaming_internal(&request, None).await { + match self.chat_streaming_internal(request, None).await { Ok(response) => return Ok(response), Err(e) => { tracing::debug!( @@ -1127,7 +1125,7 @@ impl LLMProvider for OpenAIProvider { // 非流式回退实现 let url = format!("{}/chat/completions", self.base_url); - let body = self.build_request_body(&request); + let body = self.build_request_body(request); // Debug: Log LLM request summary (only in debug builds) #[cfg(debug_assertions)] @@ -1266,14 +1264,11 @@ impl LLMProvider for OpenAIProvider { async fn chat_with_streaming( &self, - request: ChatCompletionRequest, + request: &ChatCompletionRequest, callback: StreamCallback, ) -> Result> { if self.is_streaming_enabled() { - match self - .chat_streaming_internal(&request, Some(&callback)) - .await - { + match self.chat_streaming_internal(request, Some(&callback)).await { Ok(response) => return Ok(response), Err(e) => { tracing::debug!( diff --git a/src/providers/traits.rs b/src/providers/traits.rs index 37f9439..cc7c136 100644 --- a/src/providers/traits.rs +++ b/src/providers/traits.rs @@ -117,7 +117,10 @@ pub struct ChatCompletionRequest { pub messages: Vec, pub temperature: Option, pub max_tokens: Option, - pub tools: Option>, + /// 工具定义在单次 process() 内跨多轮 LLM 迭代只读复用， + /// 用 Arc 共享避免每轮深拷贝（含完整 JSON schema，可达数十 KB）。 + /// serde 对 Arc 透明序列化，线上请求格式不变。 + pub tools: Option>>, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -158,7 +161,7 @@ pub type StreamCallback = Arc; pub trait LLMProvider: Send + Sync { async fn chat( &self, - request: ChatCompletionRequest, + request: &ChatCompletionRequest, ) -> Result>; /// 带流式回调的 chat：每收到一个 SSE delta 就调用 callback。 @@ -166,7 +169,7 @@ pub trait LLMProvider: Send + Sync { /// 默认实现忽略 callback，直接调用 chat()。 async fn chat_with_streaming( &self, - request: ChatCompletionRequest, + request: &ChatCompletionRequest, _callback: StreamCallback, ) -> Result> { self.chat(request).await diff --git a/src/scheduler/mod.rs b/src/scheduler/mod.rs index 9738262..a1630d1 100644 --- a/src/scheduler/mod.rs +++ b/src/scheduler/mod.rs @@ -107,7 +107,7 @@ impl Scheduler { return; } - if let Err(error) = self.sync_config_jobs() { + if let Err(error) = self.sync_config_jobs().await { tracing::error!(error = %error, "Failed to sync scheduler config jobs"); } @@ -137,25 +137,41 @@ impl Scheduler { } } - fn sync_config_jobs(&self) -> anyhow::Result<()> { + /// 同步配置中的 job 定义到 DB。整体移入 spawn_blocking： + /// 方法内是同步 SQLite 调用（含锁等待），直接在 async worker 上执行 + /// 会阻塞同 worker 的其他任务。 + async fn sync_config_jobs(&self) -> anyhow::Result<()> { let now = Utc::now(); - for job in self.config.effective_jobs(&crate::config::TimeConfig { - timezone: self.timezone.name().to_string(), - }) { - let runtime = - RuntimeJob::from_config(&job, now, self.config.misfire_policy, self.timezone)?; - let mut upsert = runtime.to_upsert(); - if let Some(existing) = self.jobs.get_scheduler_job(&runtime.id)? { - preserve_persisted_runtime(&mut upsert, &existing); + let config = self.config.clone(); + let jobs = self.jobs.clone(); + let timezone = self.timezone; + let misfire_policy = config.misfire_policy; + tokio::task::spawn_blocking(move || -> anyhow::Result<()> { + for job in config.effective_jobs(&crate::config::TimeConfig { + timezone: timezone.name().to_string(), + }) { + let runtime = RuntimeJob::from_config(&job, now, misfire_policy, timezone)?; + let mut upsert = runtime.to_upsert(); + if let Some(existing) = jobs.get_scheduler_job(&runtime.id)? { + preserve_persisted_runtime(&mut upsert, &existing); + } + jobs.upsert_scheduler_job(&upsert)?; } - self.jobs.upsert_scheduler_job(&upsert)?; - } - Ok(()) + Ok(()) + }) + .await + .map_err(|e| anyhow::anyhow!("scheduler config sync task failed: {e}"))? } async fn recover_interrupted_jobs(&self) -> anyhow::Result { let now = Utc::now(); - let running_jobs = self.jobs.list_running_scheduler_jobs()?; + let jobs_repo = self.jobs.clone(); + let running_jobs = + tokio::task::spawn_blocking(move || jobs_repo.list_running_scheduler_jobs()) + .await + .map_err(|e| { + anyhow::anyhow!("scheduler interrupted jobs load task failed: {e}") + })??; let mut recovered_count = 0; for record in running_jobs { @@ -192,17 +208,19 @@ impl Scheduler { other => other, }; - self.jobs.update_scheduler_job_runtime( - &record.id, + Self::persist_job_runtime_off_worker( + self.jobs.clone(), + record.id.clone(), SchedulerJobState::Scheduled, Some(SchedulerJobStatus::Error), - Some(&error_msg), + Some(error_msg), record.run_count, record.last_fired_at, next_fire_at, record.paused_at, record.completed_at, - )?; + ) + .await?; recovered_count += 1; tracing::info!( @@ -226,7 +244,10 @@ impl Scheduler { async fn process_tick(&self) -> anyhow::Result<()> { let now = Utc::now(); - let jobs = self.jobs.list_scheduler_jobs(true)?; + let jobs_repo = self.jobs.clone(); + let jobs = tokio::task::spawn_blocking(move || jobs_repo.list_scheduler_jobs(true)) + .await + .map_err(|e| anyhow::anyhow!("scheduler jobs load task failed: {e}"))??; for record in jobs { let Some(job) = @@ -236,17 +257,19 @@ impl Scheduler { }; if record.next_fire_at.is_none() && job.next_fire_at.is_some() { - self.jobs.update_scheduler_job_runtime( - &job.id, + Self::persist_job_runtime_off_worker( + self.jobs.clone(), + job.id.clone(), job.state.clone(), job.last_status.clone(), - job.last_error.as_deref(), + job.last_error.clone(), job.run_count, job.last_fired_at, job.next_fire_at, job.paused_at, job.completed_at, - )?; + ) + .await?; } if !job.is_due(now) { @@ -265,17 +288,19 @@ impl Scheduler { continue; }; - self.jobs.update_scheduler_job_runtime( - &job.id, + Self::persist_job_runtime_off_worker( + self.jobs.clone(), + job.id.clone(), SchedulerJobState::Running, job.last_status.clone(), - job.last_error.as_deref(), + job.last_error.clone(), job.run_count, job.last_fired_at, job.next_fire_at, job.paused_at, job.completed_at, - )?; + ) + .await?; // 执行与事后状态写入移入后台任务：tick 循环只做派发， // 长耗时任务（agent_task 可能长达数分钟）不再串行阻塞其他 job 的触发。 @@ -314,17 +339,20 @@ impl Scheduler { timezone, ) { Ok(()) => { - if let Err(error) = jobs_repo.update_scheduler_job_runtime( - &job.id, + if let Err(error) = Scheduler::persist_job_runtime_off_worker( + jobs_repo.clone(), + job.id.clone(), job.state.clone(), status, - job.last_error.as_deref(), + job.last_error.clone(), job.run_count, job.last_fired_at, job.next_fire_at, job.paused_at, job.completed_at, - ) { + ) + .await + { tracing::error!( job_id = %job.id, error = %error, @@ -339,17 +367,20 @@ impl Scheduler { error = %error, "Failed to compute post-execution scheduler state, resetting to Scheduled" ); - if let Err(update_error) = jobs_repo.update_scheduler_job_runtime( - &job.id, + if let Err(update_error) = Scheduler::persist_job_runtime_off_worker( + jobs_repo, + job.id.clone(), SchedulerJobState::Scheduled, Some(SchedulerJobStatus::Error), - Some(&error.to_string()), + Some(error.to_string()), job.run_count, job.last_fired_at, job.next_fire_at, job.paused_at, job.completed_at, - ) { + ) + .await + { tracing::error!( job_id = %job.id, error = %update_error, @@ -379,6 +410,39 @@ impl Scheduler { .await } + /// 将 job 运行时状态写入移入 spawn_blocking： + /// update_scheduler_job_runtime 是同步 SQLite 调用，DB 锁竞争时 + /// 的等待时间不可控，直接 await 点外执行会阻塞 tokio async worker。 + async fn persist_job_runtime_off_worker( + jobs: Arc, + job_id: String, + state: SchedulerJobState, + last_status: Option, + last_error: Option, + run_count: i64, + last_fired_at: Option, + next_fire_at: Option, + paused_at: Option, + completed_at: Option, + ) -> anyhow::Result<()> { + tokio::task::spawn_blocking(move || { + jobs.update_scheduler_job_runtime( + &job_id, + state, + last_status, + last_error.as_deref(), + run_count, + last_fired_at, + next_fire_at, + paused_at, + completed_at, + ) + }) + .await + .map_err(|e| anyhow::anyhow!("scheduler job state persist task failed: {e}"))? + .map_err(anyhow::Error::from) + } + /// job 执行主体：不依赖 &self，便于移入 tokio::spawn 的后台任务。 async fn execute_job_inner( bus: &Arc, @@ -1601,8 +1665,8 @@ mod tests { assert_eq!(saved.state, SchedulerJobState::Scheduled); } - #[test] - fn sync_config_jobs_persists_builtin_memory_maintenance_job() { + #[tokio::test] + async fn sync_config_jobs_persists_builtin_memory_maintenance_job() { let store = Arc::new(SessionStore::in_memory().unwrap()); let (agent_task_executor, maintenance_service) = test_scheduler_services(); @@ -1615,7 +1679,7 @@ mod tests { maintenance_service, ); - scheduler.sync_config_jobs().unwrap(); + scheduler.sync_config_jobs().await.unwrap(); let saved = store .get_scheduler_job(BUILTIN_MEMORY_MAINTENANCE_JOB_ID) @@ -1647,8 +1711,8 @@ mod tests { assert!(saved.next_fire_at.is_some()); } - #[test] - fn sync_config_jobs_preserves_persisted_next_fire_at_for_matching_jobs() { + #[tokio::test] + async fn sync_config_jobs_preserves_persisted_next_fire_at_for_matching_jobs() { let store = Arc::new(SessionStore::in_memory().unwrap()); let persisted_next_fire_at = 1_700_000_300_000; let config_job = SchedulerJobConfig { @@ -1735,7 +1799,7 @@ mod tests { maintenance_service, ); - scheduler.sync_config_jobs().unwrap(); + scheduler.sync_config_jobs().await.unwrap(); let saved = store.get_scheduler_job("agent.heartbeat").unwrap().unwrap(); diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 1a810aa..4c98659 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -388,6 +388,37 @@ impl SessionStore { Ok(sessions) } + /// 按 id 后缀查找最近活跃的一条 session（`LIMIT 1` 定向查询）。 + /// + /// 调用方只需要"最新一条"时使用本方法，避免 LIKE 匹配多行时 + /// 全部反序列化后仅取 `[0]`。 + pub fn find_first_session_by_id_suffix( + &self, + suffix: &str, + ) -> Result, StorageError> { + let conn = self.pool.get()?; + let pattern = format!("%{}", suffix); + conn.query_row( + " + SELECT id, title, channel_name, chat_id, summary, + created_at, updated_at, last_active_at, + archived_at, deleted_at, message_count, + user_turn_count, agent_prompt_reinjection_count + FROM sessions + WHERE id LIKE ?1 AND deleted_at IS NULL + ORDER BY last_active_at DESC + LIMIT 1 + ", + params![pattern], + map_session_record, + ) + .map(Some) + .or_else(|e| match e { + rusqlite::Error::QueryReturnedNoRows => Ok(None), + other => Err(other.into()), + }) + } + pub fn list_sessions( &self, channel_name: &str, @@ -1846,6 +1877,21 @@ impl SessionStore { Ok(count as usize) } + /// 按 topic 统计 user 角色消息数（定向 `COUNT(*)`）。 + /// + /// 与 `get_topic_message_count` 同理：数据库侧计数， + /// 避免为得到数量而全量加载消息体（content、tool_calls_json 等大字段）。 + /// 过滤条件与 `load_messages_for_topic(_, None)` 一致（排除已压缩消息）。 + pub fn count_user_messages_for_topic(&self, topic_id: &str) -> Result { + let conn = self.pool.get()?; + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM messages WHERE topic_id = ?1 AND role = 'user' AND is_compacted = 0", + params![topic_id], + |row| row.get(0), + )?; + Ok(count as usize) + } + pub fn load_all_messages(&self, session_id: &str) -> Result, StorageError> { let conn = self.pool.get()?; load_messages_after(&conn, session_id, 0) diff --git a/src/tools/memory_search.rs b/src/tools/memory_search.rs index 8e1e257..919ee03 100644 --- a/src/tools/memory_search.rs +++ b/src/tools/memory_search.rs @@ -90,9 +90,15 @@ impl Tool for MemorySearchTool { let payload = match action { "list" => { let limit = extract_u64(&args, "limit").unwrap_or(10) as usize; - let memories = self - .memories - .list_memories("user", &scope_key, namespace, limit)?; + // 同步 SQLite 查询移入 spawn_blocking，避免阻塞 tokio worker + let memories_repo = self.memories.clone(); + let scope = scope_key.clone(); + let ns = namespace.map(str::to_string); + let memories = tokio::task::spawn_blocking(move || { + memories_repo.list_memories("user", &scope, ns.as_deref(), limit) + }) + .await + .map_err(|e| anyhow::anyhow!("memory list task failed: {e}"))??; json!({ "count": memories.len(), "memories": memories.into_iter().map(memory_to_json).collect::>() @@ -138,9 +144,22 @@ impl Tool for MemorySearchTool { return Ok(error_result("Missing required parameter: queries")); } let limit = extract_u64(&args, "limit").unwrap_or(10) as usize; - let memories = self - .memories - .search_memories_any("user", &scope_key, &queries, namespace, limit)?; + let memories_repo = self.memories.clone(); + let scope = scope_key.clone(); + let ns = namespace.map(str::to_string); + let query_terms = queries.clone(); + // 同步 SQLite 多关键词查询（LIKE 扫描）移入 spawn_blocking + let memories = tokio::task::spawn_blocking(move || { + memories_repo.search_memories_any( + "user", + &scope, + &query_terms, + ns.as_deref(), + limit, + ) + }) + .await + .map_err(|e| anyhow::anyhow!("memory search task failed: {e}"))??; json!({ "queries": queries, "count": memories.len(), @@ -157,10 +176,17 @@ impl Tool for MemorySearchTool { None => return Ok(error_result("Missing required parameter: key")), }; - match self - .memories - .get_memory("user", &scope_key, namespace, key)? - { + let memories_repo = self.memories.clone(); + let scope = scope_key.clone(); + let ns = namespace.to_string(); + let memory_key = key.to_string(); + // 同步 SQLite 查询移入 spawn_blocking + let memory = tokio::task::spawn_blocking(move || { + memories_repo.get_memory("user", &scope, &ns, &memory_key) + }) + .await + .map_err(|e| anyhow::anyhow!("memory get task failed: {e}"))??; + match memory { Some(memory) => memory_to_json(memory), None => { return Ok(error_result(&format!( diff --git a/src/topic_description.rs b/src/topic_description.rs index 8cb1eee..9a0d0ba 100644 --- a/src/topic_description.rs +++ b/src/topic_description.rs @@ -15,7 +15,7 @@ pub async fn generate_topic_description( tools: None, }; - let response = provider.chat(request).await?; + let response = provider.chat(&request).await?; let description = response.content.trim().to_string(); if description.is_empty() { diff --git a/tests/test_integration.rs b/tests/test_integration.rs index f4a0758..3ab6cf7 100644 --- a/tests/test_integration.rs +++ b/tests/test_integration.rs @@ -65,7 +65,7 @@ async fn test_openai_simple_completion() { let config = load_config().expect("Please configure tests/test.env with valid API keys"); let provider = create_provider(to_runtime_config(config)).expect("Failed to create provider"); - let response = provider.chat(create_request("Say 'ok'")).await.unwrap(); + let response = provider.chat(&create_request("Say 'ok'")).await.unwrap(); assert!(!response.id.is_empty()); assert!(!response.content.is_empty()); @@ -91,7 +91,7 @@ async fn test_openai_conversation() { tools: None, }; - let response = provider.chat(request).await.unwrap(); + let response = provider.chat(&request).await.unwrap(); assert!(response.content.to_lowercase().contains("alice")); } diff --git a/tests/test_tool_calling.rs b/tests/test_tool_calling.rs index 44eda80..5eea5a5 100644 --- a/tests/test_tool_calling.rs +++ b/tests/test_tool_calling.rs @@ -83,10 +83,10 @@ async fn test_openai_tool_call() { messages: vec![Message::user("What is the weather in Tokyo?")], temperature: Some(0.0), max_tokens: Some(200), - tools: Some(vec![make_weather_tool()]), + tools: Some(std::sync::Arc::new(vec![make_weather_tool()])), }; - let response = provider.chat(request).await.unwrap(); + let response = provider.chat(&request).await.unwrap(); // Should have tool calls assert!( @@ -112,10 +112,10 @@ async fn test_openai_tool_call_with_manual_execution() { messages: vec![Message::user("What is the weather in Tokyo?")], temperature: Some(0.0), max_tokens: Some(200), - tools: Some(vec![make_weather_tool()]), + tools: Some(std::sync::Arc::new(vec![make_weather_tool()])), }; - let response1 = provider.chat(request1).await.unwrap(); + let response1 = provider.chat(&request1).await.unwrap(); let tool_call = response1.tool_calls.first().expect("Expected tool call"); assert_eq!(tool_call.name, "get_weather"); @@ -127,10 +127,10 @@ async fn test_openai_tool_call_with_manual_execution() { ], temperature: Some(0.0), max_tokens: Some(200), - tools: Some(vec![make_weather_tool()]), + tools: Some(std::sync::Arc::new(vec![make_weather_tool()])), }; - let response2 = provider.chat(request2).await.unwrap(); + let response2 = provider.chat(&request2).await.unwrap(); // Should have a response assert!(!response2.content.is_empty() || !response2.tool_calls.is_empty()); @@ -150,7 +150,7 @@ async fn test_openai_no_tool_when_not_provided() { tools: None, }; - let response = provider.chat(request).await.unwrap(); + let response = provider.chat(&request).await.unwrap(); // Should NOT have tool calls assert!(response.tool_calls.is_empty()); diff --git a/web/src/App.tsx b/web/src/App.tsx index 7b5e2f5..eec70b2 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -645,6 +645,11 @@ function App() { [selectedSessionId, selectSession], ); + // merged_tool 对象复用缓存：流式期间 messages 每帧换引用，但绝大多数 + // tool_call/tool_result 输入未变；按值相等复用上次对象，使 MessageBubble + // 的 memo 浅比较命中，避免所有工具气泡每帧重渲染（含 ReactMarkdown 重解析）。 + const mergedToolCacheRef = useRef(new Map()); + const chatMessages = useMemo(() => { const result: ChatMessage[] = []; const toolCallIndex = new Map(); @@ -683,6 +688,28 @@ function App() { } } + // 按决定性字段值相等复用上次的 merged_tool 对象（渲染结果相同 → 引用可复用） + const cache = mergedToolCacheRef.current; + const nextCache = new Map(); + for (let i = 0; i < result.length; i++) { + const cur = result[i]; + if (cur.type !== 'merged_tool') continue; + const key = cur.toolCallId || cur.id; + const prev = cache.get(key); + if ( + prev && + prev.id === cur.id && + prev.status === cur.status && + prev.callContent === cur.callContent && + prev.resultContent === cur.resultContent && + prev.durationMs === cur.durationMs + ) { + result[i] = prev; + } + nextCache.set(key, result[i]); + } + mergedToolCacheRef.current = nextCache; + // 过滤无实质内容的 merged_tool：result 到达后才显示保留；calling/pending 有 callContent 也保留 return result.filter((msg) => { if (msg.type !== 'merged_tool') return true; @@ -691,12 +718,14 @@ function App() { }); }, [messages]); - // 视图标识：用于 MessageList 保存/恢复每个视图的滚动位置 + // 视图标识：用于 MessageList 保存/恢复每个视图的滚动位置。 + // 依赖 subAgentTaskId（原始值）而非 subAgentView 对象，避免子代理流式 + // 期间每帧无效重算（对象每帧换引用但 taskId 不变）。 const viewKey = useMemo(() => { if (schedulerView) return `scheduler:${schedulerView.jobId}`; - if (subAgentView) return `subagent:${subAgentView.taskId}`; + if (subAgentTaskId) return `subagent:${subAgentTaskId}`; return `main:${selectedTopic ?? ''}`; - }, [schedulerView, subAgentView, selectedTopic]); + }, [schedulerView, subAgentTaskId, selectedTopic]); return (
diff --git a/web/src/components/Chat/ChatContainer.tsx b/web/src/components/Chat/ChatContainer.tsx index 94a2c35..0e21df6 100644 --- a/web/src/components/Chat/ChatContainer.tsx +++ b/web/src/components/Chat/ChatContainer.tsx @@ -1,4 +1,4 @@ -import { useState, useCallback } from 'react'; +import { useState, useCallback, memo } from 'react'; import { MessageList } from './MessageList'; import { MessageInput } from './MessageInput'; import { ExpertSelector } from './ExpertSelector'; @@ -29,7 +29,9 @@ interface ChatContainerProps { topicId?: string | null; } -export function ChatContainer({ +// memo：props 除 messages 外全部稳定（useCallback/原始值）， +// App 因非消息类 state（侧栏折叠、主题等）重渲染时跳过整个聊天子树。 +export const ChatContainer = memo(function ChatContainer({ messages, isLoading, isReadOnly = false, @@ -136,4 +138,4 @@ export function ChatContainer({
); -} +}); diff --git a/web/src/components/Chat/ExpertSelector.tsx b/web/src/components/Chat/ExpertSelector.tsx index 454eca2..e1febf9 100644 --- a/web/src/components/Chat/ExpertSelector.tsx +++ b/web/src/components/Chat/ExpertSelector.tsx @@ -1,4 +1,4 @@ -import { useState, useEffect, useRef, useCallback } from 'react'; +import { useState, useEffect, useRef, useCallback, memo } from 'react'; import { UserCheck, ChevronDown, Loader2, Settings, Check } from 'lucide-react'; import { getSelectedExpert, selectExpert, listExperts } from '../../api/experts'; @@ -24,7 +24,8 @@ interface ExpertSelectorProps { settingsClosedTick?: number; } -export function ExpertSelector({ +// memo：props 全部稳定（useCallback/原始值），流式期间跳过重渲染 +export const ExpertSelector = memo(function ExpertSelector({ sessionId, onManageExperts, onSelectionChange, @@ -278,4 +279,4 @@ export function ExpertSelector({ {error && {error}} ); -} +}); diff --git a/web/src/components/Chat/MessageInput.tsx b/web/src/components/Chat/MessageInput.tsx index 96f8286..df6f1f9 100644 --- a/web/src/components/Chat/MessageInput.tsx +++ b/web/src/components/Chat/MessageInput.tsx @@ -10,7 +10,7 @@ import { MusicIcon, VideoIcon, } from 'lucide-react'; -import { useState, useRef, useEffect } from 'react'; +import { useState, useRef, useEffect, memo } from 'react'; import type { Attachment } from '../../types/protocol'; const MAX_FILE_SIZE = 50 * 1024 * 1024; // 50MB @@ -77,7 +77,9 @@ function getMediaType(mimeType: string): string { return 'file'; } -export function MessageInput({ +// memo：props 全部稳定（回调 useCallback、对象 prop 来自父组件 state）， +// 流式期间父组件每帧重渲染时完全跳过输入区子树。 +export const MessageInput = memo(function MessageInput({ onSend, onStop, disabled = false, @@ -528,4 +530,4 @@ export function MessageInput({ ); -} +}); diff --git a/web/src/components/Chat/MessageList.tsx b/web/src/components/Chat/MessageList.tsx index 6b7fbd8..8572a53 100644 --- a/web/src/components/Chat/MessageList.tsx +++ b/web/src/components/Chat/MessageList.tsx @@ -1,4 +1,4 @@ -import { useEffect, useLayoutEffect, useRef, useState, useCallback, useMemo } from 'react'; +import { useEffect, useLayoutEffect, useRef, useState, useCallback, useMemo, memo } from 'react'; import { useVirtualizer } from '@tanstack/react-virtual'; import { MessageBubble } from './MessageBubble'; import type { ChatMessage } from '../../types/protocol'; @@ -16,7 +16,9 @@ interface MessageListProps { effectiveModel?: { provider: string; model: string } | null; } -export function MessageList({ +// memo：props 与 ChatContainer 同源（messages 变化时才需重渲染）， +// 阻断父组件非消息类重渲染向虚拟化列表的传导。 +export const MessageList = memo(function MessageList({ messages, onNavigateToSubAgent, showThinking = true, @@ -385,4 +387,4 @@ export function MessageList({ )} ); -} +}); diff --git a/web/src/components/Chat/ModelSelector.tsx b/web/src/components/Chat/ModelSelector.tsx index 3e6198c..922bb44 100644 --- a/web/src/components/Chat/ModelSelector.tsx +++ b/web/src/components/Chat/ModelSelector.tsx @@ -1,4 +1,4 @@ -import { useState, useEffect, useRef, useCallback } from 'react'; +import { useState, useEffect, useRef, useCallback, memo } from 'react'; import { Cpu, ChevronDown, Loader2, Check } from 'lucide-react'; import { listModelOptions, @@ -19,7 +19,8 @@ interface ModelSelectorProps { onSelectionChange?: (effective: { provider: string; model: string; overridden: boolean }) => void; } -export function ModelSelector({ +// memo：props 全部稳定（useCallback/原始值），流式期间跳过重渲染 +export const ModelSelector = memo(function ModelSelector({ sessionId, topicId, settingsClosedTick, @@ -284,4 +285,4 @@ export function ModelSelector({ {error && {error}} ); -} +}); diff --git a/web/src/components/Panel/MemoryPanel.tsx b/web/src/components/Panel/MemoryPanel.tsx index 4b3c3dc..766bcfa 100644 --- a/web/src/components/Panel/MemoryPanel.tsx +++ b/web/src/components/Panel/MemoryPanel.tsx @@ -1,4 +1,4 @@ -import { useState } from 'react'; +import { useState, useMemo, memo } from 'react'; import { Brain, User, @@ -310,7 +310,9 @@ function SectionHeader({ /* ── main component ────────────────────────────────────── */ -export function MemoryPanel({ +// memo：props 全部稳定（memories 仅在刷新时换引用、回调均 useCallback）， +// 主视图流式期间 App 每帧重渲染时跳过面板重渲染与分组/排序重算。 +export const MemoryPanel = memo(function MemoryPanel({ memories, onRefresh, onClose, @@ -342,22 +344,26 @@ export function MemoryPanel({ }); }; - const grouped = new Map(); - for (const m of memories) { - const l = grouped.get(m.namespace) || []; - l.push(m); - grouped.set(m.namespace, l); - } + // memories 引用未变时跳过分组/排序重算（流式期间 App 每帧重渲染） + const { grouped, sorted } = useMemo(() => { + const grouped = new Map(); + for (const m of memories) { + const l = grouped.get(m.namespace) || []; + l.push(m); + grouped.set(m.namespace, l); + } - const order = ['user', 'semantic', 'episodic', 'skill', 'environment', 'reflection', 'other']; - const sorted = Array.from(grouped.keys()).sort((a, b) => { - const ai = order.indexOf(a); - const bi = order.indexOf(b); - if (ai !== -1 && bi !== -1) return ai - bi; - if (ai !== -1) return -1; - if (bi !== -1) return 1; - return a.localeCompare(b); - }); + const order = ['user', 'semantic', 'episodic', 'skill', 'environment', 'reflection', 'other']; + const sorted = Array.from(grouped.keys()).sort((a, b) => { + const ai = order.indexOf(a); + const bi = order.indexOf(b); + if (ai !== -1 && bi !== -1) return ai - bi; + if (ai !== -1) return -1; + if (bi !== -1) return 1; + return a.localeCompare(b); + }); + return { grouped, sorted }; + }, [memories]); const handleCreate = (ns: string, key: string, content: string) => { sendCommand(onCreateMemory(ns, key, content)); @@ -469,4 +475,4 @@ export function MemoryPanel({ )} ); -} +}); diff --git a/web/src/components/Panel/SkillList.tsx b/web/src/components/Panel/SkillList.tsx index 5c05f09..42a9e12 100644 --- a/web/src/components/Panel/SkillList.tsx +++ b/web/src/components/Panel/SkillList.tsx @@ -1,4 +1,4 @@ -import { useState } from 'react'; +import { useState, useMemo, memo } from 'react'; import { Package, User, @@ -100,7 +100,9 @@ function SkillCard({ skill, config }: { skill: SkillSummary; config: SourceConfi /* ── main component ────────────────────────────────────── */ -export function SkillList({ skills, onRefresh }: SkillListProps) { +// memo：props 全部稳定（skills 仅在刷新时换引用、onRefresh 为 useCallback）， +// 主视图流式期间 App 每帧重渲染时跳过面板重渲染与分组/排序重算。 +export const SkillList = memo(function SkillList({ skills, onRefresh }: SkillListProps) { const [collapsed, setCollapsed] = useState>(() => { try { const s = localStorage.getItem('picobot-skill-collapsed'); @@ -123,22 +125,33 @@ export function SkillList({ skills, onRefresh }: SkillListProps) { }); }; - const grouped = new Map(); - for (const s of skills) { - const l = grouped.get(s.source) || []; - l.push(s); - grouped.set(s.source, l); - } + // skills 引用未变时跳过分组/排序重算（流式期间 App 每帧重渲染） + const { grouped, sorted } = useMemo(() => { + const grouped = new Map(); + for (const s of skills) { + const l = grouped.get(s.source) || []; + l.push(s); + grouped.set(s.source, l); + } - const order = ['user', 'useragent', 'useropenclaw', 'project', 'projectagent', 'projectopenclaw']; - const sorted = Array.from(grouped.keys()).sort((a, b) => { - const ai = order.indexOf(a); - const bi = order.indexOf(b); - if (ai !== -1 && bi !== -1) return ai - bi; - if (ai !== -1) return -1; - if (bi !== -1) return 1; - return a.localeCompare(b); - }); + const order = [ + 'user', + 'useragent', + 'useropenclaw', + 'project', + 'projectagent', + 'projectopenclaw', + ]; + const sorted = Array.from(grouped.keys()).sort((a, b) => { + const ai = order.indexOf(a); + const bi = order.indexOf(b); + if (ai !== -1 && bi !== -1) return ai - bi; + if (ai !== -1) return -1; + if (bi !== -1) return 1; + return a.localeCompare(b); + }); + return { grouped, sorted }; + }, [skills]); return (
@@ -209,4 +222,4 @@ export function SkillList({ skills, onRefresh }: SkillListProps) { )}
); -} +}); diff --git a/web/src/components/Panel/TodoPanel.tsx b/web/src/components/Panel/TodoPanel.tsx index 32e0eb0..d9ac6cc 100644 --- a/web/src/components/Panel/TodoPanel.tsx +++ b/web/src/components/Panel/TodoPanel.tsx @@ -1,4 +1,4 @@ -import { useState, useCallback, useEffect, useRef } from 'react'; +import { useState, useCallback, useEffect, useRef, useMemo, memo } from 'react'; import { ClipboardList, ChevronDown, RefreshCw } from 'lucide-react'; import type { TodoItemSummary, Command } from '../../types/protocol'; @@ -53,7 +53,14 @@ function PulseDot() { /* ── TodoPanel ────────────────────────────────────────── */ -export function TodoPanel({ todos, requestTodoList, sendCommand, onTodoClick }: TodoPanelProps) { +// memo：props 全部稳定（todos 仅在 todo_list 消息到达时换引用、回调均 useCallback）， +// 主视图流式期间 App 每帧重渲染时跳过面板重渲染与分组重算。 +export const TodoPanel = memo(function TodoPanel({ + todos, + requestTodoList, + sendCommand, + onTodoClick, +}: TodoPanelProps) { const [collapsedGroups, setCollapsedGroups] = useState>( () => new Set(['completed', 'cancelled']), ); @@ -75,7 +82,8 @@ export function TodoPanel({ todos, requestTodoList, sendCommand, onTodoClick }: prevTodoIdsRef.current = newIds; }, [todos]); - const grouped = groupTodos(todos); + // todos 引用未变时跳过分组重算（流式期间 App 每帧重渲染） + const grouped = useMemo(() => groupTodos(todos), [todos]); const inProgressCount = grouped.get('in_progress')?.length ?? 0; const totalCount = todos.length; @@ -180,4 +188,4 @@ export function TodoPanel({ todos, requestTodoList, sendCommand, onTodoClick }: ); -} +}); diff --git a/web/src/components/Panel/TopicTokenStatsPanel.tsx b/web/src/components/Panel/TopicTokenStatsPanel.tsx index 8d46edc..69f79d5 100644 --- a/web/src/components/Panel/TopicTokenStatsPanel.tsx +++ b/web/src/components/Panel/TopicTokenStatsPanel.tsx @@ -1,3 +1,4 @@ +import { memo } from 'react'; import { Coins } from 'lucide-react'; import type { TopicTokenStats } from '../../types/protocol'; import { @@ -12,7 +13,11 @@ interface TopicTokenStatsPanelProps { tokenStats?: TopicTokenStats | null; } -export function TopicTokenStatsPanel({ tokenStats }: TopicTokenStatsPanelProps) { +// memo：tokenStats 引用稳定（仅 topic 刷新时换引用）， +// 流式期间 App 每帧重渲染时跳过面板重渲染。 +export const TopicTokenStatsPanel = memo(function TopicTokenStatsPanel({ + tokenStats, +}: TopicTokenStatsPanelProps) { // 无数据态 if (!tokenStats || tokenStats.total_tokens === 0) { return ( @@ -126,4 +131,4 @@ export function TopicTokenStatsPanel({ tokenStats }: TopicTokenStatsPanelProps) )} ); -} +}); diff --git a/web/src/components/Sidebar/SchedulerJobList.tsx b/web/src/components/Sidebar/SchedulerJobList.tsx index 3c1eb31..b842c0c 100644 --- a/web/src/components/Sidebar/SchedulerJobList.tsx +++ b/web/src/components/Sidebar/SchedulerJobList.tsx @@ -1,3 +1,4 @@ +import { memo } from 'react'; import { Clock, RefreshCw, ChevronRight, Check, X, Minus } from 'lucide-react'; import type { SchedulerJobSummary, SchedulerJobSessionLookup } from '../../types/protocol'; @@ -92,7 +93,14 @@ function lastStatusIcon(lastStatus: string | undefined) { } } -export function SchedulerJobList({ jobs, onRefresh, onViewJob, sessionId }: SchedulerJobListProps) { +// memo：props 全部稳定（jobs 仅在刷新时换引用、回调均 useCallback）， +// 阻断 App 重渲染向定时任务列表的无效传导。 +export const SchedulerJobList = memo(function SchedulerJobList({ + jobs, + onRefresh, + onViewJob, + sessionId, +}: SchedulerJobListProps) { return (
{/* Header */} @@ -228,4 +236,4 @@ export function SchedulerJobList({ jobs, onRefresh, onViewJob, sessionId }: Sche
); -} +}); diff --git a/web/src/components/Sidebar/TopicList.tsx b/web/src/components/Sidebar/TopicList.tsx index d62620c..76100b3 100644 --- a/web/src/components/Sidebar/TopicList.tsx +++ b/web/src/components/Sidebar/TopicList.tsx @@ -1,4 +1,4 @@ -import { useState, useEffect, useMemo, useRef, useCallback } from 'react'; +import { useState, useEffect, useMemo, useRef, useCallback, memo } from 'react'; import { Plus, MessageSquare, @@ -43,7 +43,9 @@ function formatTime(timestamp: number): string { } } -export function TopicList({ +// memo：props 全部稳定（topics 仅在 topic_list 消息到达时换引用、回调均 useCallback）， +// 流式期间 App 每帧重渲染时跳过话题列表重渲染与分页重算。 +export const TopicList = memo(function TopicList({ sessionId, topics, currentTopicId, @@ -357,4 +359,4 @@ export function TopicList({ )} ); -} +});