diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 7bd961b..90668df 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -917,7 +917,7 @@ impl AgentLoop { "Pre-process message state before sanitize" ); } - let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut messages); + let removed = Self::sanitize_messages_for_llm(&mut messages); if removed > 0 { tracing::warn!( removed_count = removed, @@ -937,13 +937,11 @@ impl AgentLoop { // 检查取消信号 // 使用 unwrap_or(true):即使 watch channel 因异常关闭(sender drop 但未 send), // 也视为取消信号。defense-in-depth —— fail safe 而非 fail silent。 - if let Some(ref mutex) = self.cancel_token { - if mutex.lock().await.has_changed().unwrap_or(true) { - tracing::info!(iteration, "Agent execution cancelled by user"); - let cancel = Self::build_cancel_result(iteration, emitted_messages); - self.emit_live_tool_call_message(cancel.final_response.clone()).await; - return Ok(cancel); - } + if self.check_cancelled().await { + tracing::info!(iteration, "Agent execution cancelled by user"); + let cancel = Self::build_cancel_result(iteration, emitted_messages); + self.emit_live_tool_call_message(cancel.final_response.clone()).await; + return Ok(cancel); } // Build request @@ -959,8 +957,7 @@ impl AgentLoop { // This catches edge cases where compression, persistence races, // or delta message merging may have introduced orphaned sequences // that survived the initial sanitization. - let mid_loop_removed = - crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut messages); + let mid_loop_removed = Self::sanitize_messages_for_llm(&mut messages); if mid_loop_removed > 0 { tracing::warn!( iteration = iteration, @@ -969,50 +966,7 @@ impl AgentLoop { ); } - // 过滤超出轮次和数量限制的图片 - let filtered_messages = filter_images_by_age_and_count( - &messages, - self.runtime_config.max_image_age_rounds, - self.runtime_config.max_images_in_context, - ); - let image_count = count_supported_image_media_refs(&filtered_messages); - - // 构建系统提示词(统一注入 Agent 和 Skill 提示词) - let system_prompt = system_prompt_context.and_then(|ctx| { - self.system_prompt_provider - .as_ref() - .and_then(|provider| provider.build(ctx)) - }); - - let mut text_only_messages: Vec = Vec::with_capacity(filtered_messages.len() + 2); - if let Some(ref prompt) = system_prompt { - text_only_messages.push(Message::system(prompt.content.clone())); - } - text_only_messages.extend(filtered_messages.iter().map(chat_message_to_text_only_llm_message)); - - let image_tokens = image_token_budget_for_request( - &self.runtime_config, - &text_only_messages, - tools.as_ref(), - ); - let mut image_budget = ImageInlineBudget::new(image_tokens, image_count); - let mut messages_for_llm: Vec = Vec::with_capacity(filtered_messages.len() + 2); - // 使用相同的系统提示词(已构建) - if let Some(ref prompt) = system_prompt { - messages_for_llm.push(Message::system(prompt.content.clone())); - } - messages_for_llm.extend( - filtered_messages - .iter() - .map(|message| chat_message_to_llm_message(message, &mut image_budget)), - ); - - let request = ChatCompletionRequest { - messages: messages_for_llm, - temperature: None, - max_tokens: None, - tools, - }; + let request = self.build_llm_request(&messages, system_prompt_context, tools); // Set up streaming delta consumer // Pre-generate the message ID so stream deltas and the final assistant @@ -1116,23 +1070,10 @@ impl AgentLoop { // If no tool calls, this is the final response if response.tool_calls.is_empty() { - let mut assistant_message = if let Some(reasoning_content) = response.reasoning_content - { - ChatMessage::assistant_with_reasoning(response.content, reasoning_content) - } else { - ChatMessage::assistant(response.content) - }; - // Use the same ID as the stream deltas so the front-end can replace - // the streamed message with this authoritative response. - if had_streaming { - assistant_message.id = streaming_message_id; - } - emitted_messages.push(assistant_message.clone()); - self.emit_live_tool_call_message(assistant_message.clone()).await; - return Ok(AgentProcessResult { - final_response: assistant_message, - emitted_messages, - }); + let result = self.build_final_response( + response, &streaming_message_id, had_streaming, &mut emitted_messages, + ).await; + return Ok(result); } // Execute tool calls @@ -1201,57 +1142,10 @@ impl AgentLoop { self.execute_tools(&response.tool_calls).await }; - for (tool_call, result) in response.tool_calls.iter().zip(tool_results.iter()) { - // Truncate tool result if too large - let truncated_output = - truncate_tool_result(&result.output, self.runtime_config.tool_result_max_chars); - - // Record tool call and check for loops - let loop_result = loop_detector.record(&tool_call.name, &tool_call.arguments); - - match loop_result { - LoopDetectionResult::Warning(msg) => { - // Add warning and proceed - tracing::warn!( - tool = %tool_call.name, - "Loop warning: {}", - msg - ); - let tool_message = ChatMessage::tool_with_state( - tool_call.id.clone(), - tool_call.name.clone(), - format!("{}\n\n[上一条结果]\n{}", msg, truncated_output), - if result.state == ToolExecutionState::PendingUserAction { - ToolMessageState::PendingUserAction - } else { - ToolMessageState::Completed - }, - ) - .with_tool_duration(result.duration.as_millis() as u64); - messages.push(tool_message.clone()); - emitted_messages.push(tool_message.clone()); - let duration_ms = Some(result.duration.as_millis() as u64); - self.emit_tool_result(tool_message, duration_ms).await; - } - LoopDetectionResult::Ok => { - let tool_message = ChatMessage::tool_with_state( - tool_call.id.clone(), - tool_call.name.clone(), - truncated_output, - if result.state == ToolExecutionState::PendingUserAction { - ToolMessageState::PendingUserAction - } else { - ToolMessageState::Completed - }, - ) - .with_tool_duration(result.duration.as_millis() as u64); - messages.push(tool_message.clone()); - emitted_messages.push(tool_message.clone()); - let duration_ms = Some(result.duration.as_millis() as u64); - self.emit_tool_result(tool_message, duration_ms).await; - } - } - } + self.process_tool_results( + &response.tool_calls, &tool_results, &mut loop_detector, + &mut messages, &mut emitted_messages, + ).await; // Loop continues to next iteration with updated messages // PendingUserAction 工具的结果已在上方加入 messages, @@ -1264,116 +1158,8 @@ impl AgentLoop { ); } - // Max iterations reached - ask LLM for a summary based on completed work - tracing::warn!("Max iterations reached, requesting final summary from LLM"); - - // Defense: sanitize before final summary request - let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut messages); - if removed > 0 { - tracing::warn!(removed_count = removed, "Sanitized before max-iterations summary"); - } - - // Add a message asking for summary - let summary_request = ChatMessage::user( - "You have reached the maximum number of tool call iterations. \ - Please provide your best answer based on the work completed so far.", - ); - messages.push(summary_request); - - // 过滤超出轮次和数量限制的图片 - let filtered_messages = filter_images_by_age_and_count( - &messages, - self.runtime_config.max_image_age_rounds, - self.runtime_config.max_images_in_context, - ); - - // Convert messages to LLM format (使用系统提示词提供者) - let image_count = count_supported_image_media_refs(&filtered_messages); - let mut text_only_messages: Vec = Vec::with_capacity(filtered_messages.len() + 1); - if let Some(ref provider) = self.system_prompt_provider { - if let Some(ctx) = system_prompt_context { - if let Some(prompt) = provider.build(ctx) { - text_only_messages.push(Message::system(prompt.content.clone())); - } - } - } - text_only_messages.extend(filtered_messages.iter().map(chat_message_to_text_only_llm_message)); - let image_tokens = - image_token_budget_for_request(&self.runtime_config, &text_only_messages, None); - let mut image_budget = ImageInlineBudget::new(image_tokens, image_count); - let mut messages_for_llm: Vec = Vec::with_capacity(filtered_messages.len() + 1); - if let Some(ref provider) = self.system_prompt_provider { - if let Some(ctx) = system_prompt_context { - if let Some(prompt) = provider.build(ctx) { - messages_for_llm.push(Message::system(prompt.content.clone())); - } - } - } - messages_for_llm.extend( - filtered_messages - .iter() - .map(|message| chat_message_to_llm_message(message, &mut image_budget)), - ); - - let request = ChatCompletionRequest { - messages: messages_for_llm, - temperature: None, - max_tokens: None, - tools: None, // No tools in final summary call - }; - - // 最终 summary 调用也与取消信号竞速 - let final_result: Result< - crate::providers::ChatCompletionResponse, - Box, - >; - if self.cancel_token.is_some() { - tokio::select! { - _ = self.cancel_signal() => { - let cancel = Self::build_cancel_result(self.max_iterations, emitted_messages); - self.emit_live_tool_call_message(cancel.final_response.clone()).await; - return Ok(cancel); - } - result = self.provider.chat(request) => { - final_result = result; - } - } - } else { - final_result = self.provider.chat(request).await; - } - - match final_result { - Ok(response) => { - let assistant_message = if let Some(reasoning_content) = response.reasoning_content - { - ChatMessage::assistant_with_reasoning(response.content, reasoning_content) - } else { - ChatMessage::assistant(response.content) - }; - emitted_messages.push(assistant_message.clone()); - self.emit_live_tool_call_message(assistant_message.clone()).await; - Ok(AgentProcessResult { - final_response: assistant_message, - emitted_messages, - }) - } - Err(e) => { - tracing::error!( - provider = %self.provider.name(), - model = %self.provider.model_id(), - error = %e, - error_details = %format_error_chain(e.as_ref()), - "Failed to get summary from LLM" - ); - let final_message = ChatMessage::assistant(recoverable_llm_message(&e.to_string())); - emitted_messages.push(final_message.clone()); - self.emit_live_tool_call_message(final_message.clone()).await; - Ok(AgentProcessResult { - final_response: final_message, - emitted_messages, - }) - } - } + // Max iterations reached - request final summary from LLM + Ok(self.run_final_summary(&mut messages, system_prompt_context, &mut emitted_messages).await) } /// 等待取消信号。若未配置 cancel_token,永远不返回。 @@ -1389,6 +1175,217 @@ impl AgentLoop { } } + /// 净化消息中的不完整 tool_call 序列,返回移除数量。 + /// 日志由调用方负责,因为不同调用点需要不同的结构化字段。 + fn sanitize_messages_for_llm(messages: &mut Vec) -> usize { + crate::bus::message::sanitize_incomplete_tool_call_sequences(messages) + } + + /// 检查取消信号。返回 true 表示已取消。 + async fn check_cancelled(&self) -> bool { + if let Some(ref mutex) = self.cancel_token { + mutex.lock().await.has_changed().unwrap_or(true) + } else { + false + } + } + + /// 构建 LLM 请求:过滤图片、构建系统提示、转换消息格式。 + /// 统一了循环内调用和最终 summary 调用的请求构建逻辑。 + fn build_llm_request( + &self, + messages: &[ChatMessage], + system_prompt_context: Option<&SystemPromptContext>, + tools: Option>, + ) -> ChatCompletionRequest { + let filtered_messages = filter_images_by_age_and_count( + messages, + self.runtime_config.max_image_age_rounds, + self.runtime_config.max_images_in_context, + ); + let image_count = count_supported_image_media_refs(&filtered_messages); + + let system_prompt = system_prompt_context.and_then(|ctx| { + self.system_prompt_provider + .as_ref() + .and_then(|provider| provider.build(ctx)) + }); + + let mut text_only_messages: Vec = Vec::with_capacity(filtered_messages.len() + 2); + if let Some(ref prompt) = system_prompt { + text_only_messages.push(Message::system(prompt.content.clone())); + } + text_only_messages.extend(filtered_messages.iter().map(chat_message_to_text_only_llm_message)); + + let image_tokens = image_token_budget_for_request( + &self.runtime_config, + &text_only_messages, + tools.as_ref(), + ); + let mut image_budget = ImageInlineBudget::new(image_tokens, image_count); + let mut messages_for_llm: Vec = Vec::with_capacity(filtered_messages.len() + 2); + if let Some(ref prompt) = system_prompt { + messages_for_llm.push(Message::system(prompt.content.clone())); + } + messages_for_llm.extend( + filtered_messages + .iter() + .map(|message| chat_message_to_llm_message(message, &mut image_budget)), + ); + + ChatCompletionRequest { + messages: messages_for_llm, + temperature: None, + max_tokens: None, + tools, + } + } + + /// 构建 LLM 最终响应(无工具调用时)。调用方应先检查 `response.tool_calls.is_empty()`。 + async fn build_final_response( + &self, + response: crate::providers::ChatCompletionResponse, + streaming_message_id: &str, + had_streaming: bool, + emitted_messages: &mut Vec, + ) -> AgentProcessResult { + let mut assistant_message = if let Some(reasoning_content) = response.reasoning_content { + ChatMessage::assistant_with_reasoning(response.content, reasoning_content) + } else { + ChatMessage::assistant(response.content) + }; + // Use the same ID as the stream deltas so the front-end can replace + // the streamed message with this authoritative response. + if had_streaming { + assistant_message.id = streaming_message_id.to_string(); + } + emitted_messages.push(assistant_message.clone()); + self.emit_live_tool_call_message(assistant_message.clone()).await; + AgentProcessResult { + final_response: assistant_message, + emitted_messages: std::mem::take(emitted_messages), + } + } + + /// 处理工具执行结果,追加到 messages 和 emitted_messages,发出 tool_result 事件。 + /// 合并了 LoopDetectionResult::Warning 和 ::Ok 两个分支的逻辑。 + async fn process_tool_results( + &self, + tool_calls: &[ToolCall], + tool_results: &[ToolExecutionOutcome], + loop_detector: &mut LoopDetector, + messages: &mut Vec, + emitted_messages: &mut Vec, + ) { + for (tool_call, result) in tool_calls.iter().zip(tool_results.iter()) { + let truncated_output = + truncate_tool_result(&result.output, self.runtime_config.tool_result_max_chars); + + let loop_result = loop_detector.record(&tool_call.name, &tool_call.arguments); + let prefix = match loop_result { + LoopDetectionResult::Warning(msg) => { + tracing::warn!(tool = %tool_call.name, "Loop warning: {}", msg); + format!("{}\n\n[上一条结果]\n", msg) + } + LoopDetectionResult::Ok => String::new(), + }; + + let state = if result.state == ToolExecutionState::PendingUserAction { + ToolMessageState::PendingUserAction + } else { + ToolMessageState::Completed + }; + let tool_message = ChatMessage::tool_with_state( + tool_call.id.clone(), + tool_call.name.clone(), + format!("{}{}", prefix, truncated_output), + state, + ) + .with_tool_duration(result.duration.as_millis() as u64); + messages.push(tool_message.clone()); + emitted_messages.push(tool_message.clone()); + let duration_ms = Some(result.duration.as_millis() as u64); + self.emit_tool_result(tool_message, duration_ms).await; + } + } + + /// 达到最大迭代次数后,请求 LLM 生成总结。 + async fn run_final_summary( + &self, + messages: &mut Vec, + system_prompt_context: Option<&SystemPromptContext>, + emitted_messages: &mut Vec, + ) -> AgentProcessResult { + tracing::warn!("Max iterations reached, requesting final summary from LLM"); + + // Defense: sanitize before final summary request + let removed = Self::sanitize_messages_for_llm(messages); + if removed > 0 { + tracing::warn!(removed_count = removed, "Sanitized before max-iterations summary"); + } + + // Add a message asking for summary + let summary_request = ChatMessage::user( + "You have reached the maximum number of tool call iterations. \ + Please provide your best answer based on the work completed so far.", + ); + messages.push(summary_request); + + let request = self.build_llm_request(messages, system_prompt_context, None); + + // 最终 summary 调用也与取消信号竞速 + let final_result: Result< + crate::providers::ChatCompletionResponse, + Box, + >; + if self.cancel_token.is_some() { + tokio::select! { + _ = self.cancel_signal() => { + let cancel = Self::build_cancel_result(self.max_iterations, std::mem::take(emitted_messages)); + self.emit_live_tool_call_message(cancel.final_response.clone()).await; + return cancel; + } + result = self.provider.chat(request) => { + final_result = result; + } + } + } else { + final_result = self.provider.chat(request).await; + } + + match final_result { + Ok(response) => { + let assistant_message = if let Some(reasoning_content) = response.reasoning_content { + ChatMessage::assistant_with_reasoning(response.content, reasoning_content) + } else { + ChatMessage::assistant(response.content) + }; + emitted_messages.push(assistant_message.clone()); + self.emit_live_tool_call_message(assistant_message.clone()).await; + AgentProcessResult { + final_response: assistant_message, + emitted_messages: std::mem::take(emitted_messages), + } + } + Err(e) => { + tracing::error!( + provider = %self.provider.name(), + model = %self.provider.model_id(), + error = %e, + error_details = %format_error_chain(e.as_ref()), + "Failed to get summary from LLM" + ); + let final_message = ChatMessage::assistant(recoverable_llm_message(&e.to_string())); + emitted_messages.push(final_message.clone()); + self.emit_live_tool_call_message(final_message.clone()).await; + AgentProcessResult { + final_response: final_message, + emitted_messages: std::mem::take(emitted_messages), + } + } + } + } + /// 构建取消响应,包含已完成的迭代次数和已生成的消息数量。 /// 构建取消响应,将取消通知追加到 `emitted_messages` 末尾后一并返回。 /// 这样 finalize_result 会把中间消息加入内存历史,确保下一个 LLM 调用有完整上下文。