diff --git a/backend/windmill-ai/src/providers/anthropic.rs b/backend/windmill-ai/src/providers/anthropic.rs index 486b08b739..038aef2c2e 100644 --- a/backend/windmill-ai/src/providers/anthropic.rs +++ b/backend/windmill-ai/src/providers/anthropic.rs @@ -767,6 +767,7 @@ impl QueryBuilder for AnthropicQueryBuilder { let AnthropicSSEParser { accumulated_content, + accumulated_reasoning, accumulated_tool_calls, events_str, annotations, @@ -790,6 +791,7 @@ impl QueryBuilder for AnthropicQueryBuilder { } else { Some(accumulated_content) }, + reasoning: (!accumulated_reasoning.is_empty()).then_some(accumulated_reasoning), tool_calls: accumulated_tool_calls.into_values().collect(), events_str: Some(events_str), annotations, diff --git a/backend/windmill-ai/src/providers/bedrock.rs b/backend/windmill-ai/src/providers/bedrock.rs index cf977f2484..adee6e5053 100644 --- a/backend/windmill-ai/src/providers/bedrock.rs +++ b/backend/windmill-ai/src/providers/bedrock.rs @@ -1188,6 +1188,13 @@ impl BedrockQueryBuilder { Some(accumulated_text) }; + // The block folded for replay is also what the reader sees as thinking. Read out + // before the block itself moves into the tool calls below. + let reasoning_text = reasoning + .as_ref() + .and_then(|r| r.reasoning_text.clone()) + .filter(|t| !t.is_empty()); + let tool_calls = streaming_tool_calls_to_openai( accumulated_tool_calls.into_values().collect(), reasoning, @@ -1195,6 +1202,7 @@ impl BedrockQueryBuilder { Ok(ParsedResponse::Text { content, + reasoning: reasoning_text, tool_calls, events_str: if events_str.is_empty() { None diff --git a/backend/windmill-ai/src/providers/google_ai.rs b/backend/windmill-ai/src/providers/google_ai.rs index 21428903fa..8320ef5661 100644 --- a/backend/windmill-ai/src/providers/google_ai.rs +++ b/backend/windmill-ai/src/providers/google_ai.rs @@ -666,6 +666,7 @@ impl QueryBuilder for GoogleAIQueryBuilder { let GeminiSSEParser { accumulated_content, + accumulated_reasoning, accumulated_tool_calls, mut events_str, stream_event_processor, @@ -698,6 +699,7 @@ impl QueryBuilder for GoogleAIQueryBuilder { } else { Some(accumulated_content) }, + reasoning: (!accumulated_reasoning.is_empty()).then_some(accumulated_reasoning), tool_calls: accumulated_tool_calls.into_values().collect(), events_str: Some(events_str), annotations, diff --git a/backend/windmill-ai/src/providers/openai.rs b/backend/windmill-ai/src/providers/openai.rs index 34185a8852..8fa65dca21 100644 --- a/backend/windmill-ai/src/providers/openai.rs +++ b/backend/windmill-ai/src/providers/openai.rs @@ -538,6 +538,9 @@ impl QueryBuilder for OpenAIQueryBuilder { } else { Some(parser.accumulated_content) }, + // The Responses stream has no reasoning-summary event in + // `OpenAIResponsesSSEEvent`, so nothing thinks out loud on this path yet. + reasoning: None, tool_calls: parser.accumulated_tool_calls.into_values().collect(), events_str: Some(parser.events_str), annotations: parser.annotations, diff --git a/backend/windmill-ai/src/providers/other.rs b/backend/windmill-ai/src/providers/other.rs index bc2d8f2992..b6040d8d79 100644 --- a/backend/windmill-ai/src/providers/other.rs +++ b/backend/windmill-ai/src/providers/other.rs @@ -251,6 +251,7 @@ impl QueryBuilder for OtherQueryBuilder { let OpenAISSEParser { accumulated_content, + accumulated_reasoning, accumulated_tool_calls, mut events_str, stream_event_processor, @@ -277,6 +278,7 @@ impl QueryBuilder for OtherQueryBuilder { } else { Some(accumulated_content) }, + reasoning: (!accumulated_reasoning.is_empty()).then_some(accumulated_reasoning), tool_calls: accumulated_tool_calls.into_values().collect(), events_str: Some(events_str), annotations: Vec::new(), diff --git a/backend/windmill-ai/src/query_builder.rs b/backend/windmill-ai/src/query_builder.rs index c8f5fbb092..60ece300a3 100644 --- a/backend/windmill-ai/src/query_builder.rs +++ b/backend/windmill-ai/src/query_builder.rs @@ -33,6 +33,8 @@ pub struct BuildRequestArgs<'a> { pub enum ParsedResponse { Text { content: Option, + /// The thinking the model streamed before the answer, when it emitted any. + reasoning: Option, tool_calls: Vec, events_str: Option, annotations: Vec, diff --git a/backend/windmill-ai/src/sse.rs b/backend/windmill-ai/src/sse.rs index 96692d0173..986642fd7e 100644 --- a/backend/windmill-ai/src/sse.rs +++ b/backend/windmill-ai/src/sse.rs @@ -135,6 +135,8 @@ pub trait SSEParser { pub struct OpenAISSEParser { pub accumulated_content: String, + /// The thinking streamed before the answer, kept so it can be stored with it. + pub accumulated_reasoning: String, pub accumulated_tool_calls: HashMap, pub events_str: String, pub stream_event_processor: Box, @@ -146,6 +148,7 @@ impl OpenAISSEParser { pub fn new(stream_event_processor: Box) -> Self { Self { accumulated_content: String::new(), + accumulated_reasoning: String::new(), accumulated_tool_calls: HashMap::new(), events_str: String::new(), stream_event_processor, @@ -175,6 +178,7 @@ impl SSEParser for OpenAISSEParser { if let Some(mut choices) = event.choices.filter(|s| !s.is_empty()) { if let Some(delta) = choices.remove(0).delta { if let Some(reasoning) = delta.reasoning_content.filter(|s| !s.is_empty()) { + self.accumulated_reasoning.push_str(&reasoning); let event = StreamingEvent::ReasoningTokenDelta { content: reasoning }; self.stream_event_processor .send(event, &mut self.events_str) @@ -353,6 +357,8 @@ enum ContentBlockState { /// Anthropic SSE Parser for streaming responses pub struct AnthropicSSEParser { pub accumulated_content: String, + /// The thinking streamed before the answer, kept so it can be stored with it. + pub accumulated_reasoning: String, pub accumulated_tool_calls: HashMap, pub events_str: String, pub stream_event_processor: Box, @@ -375,6 +381,7 @@ impl AnthropicSSEParser { pub fn new(stream_event_processor: Box) -> Self { Self { accumulated_content: String::new(), + accumulated_reasoning: String::new(), accumulated_tool_calls: HashMap::new(), events_str: String::new(), stream_event_processor, @@ -455,6 +462,7 @@ impl SSEParser for AnthropicSSEParser { .thinking .get_or_insert_with(String::new) .push_str(&thinking); + self.accumulated_reasoning.push_str(&thinking); self.stream_event_processor .send( StreamingEvent::ReasoningTokenDelta { content: thinking }, @@ -523,6 +531,7 @@ impl SSEParser for AnthropicSSEParser { .thinking .get_or_insert_with(String::new) .push_str(&thinking); + self.accumulated_reasoning.push_str(&thinking); self.stream_event_processor .send( StreamingEvent::ReasoningTokenDelta { content: thinking }, @@ -590,6 +599,8 @@ impl SSEParser for AnthropicSSEParser { /// `windmill_ai::ai_google` so the logic can be shared with the API proxy. pub struct GeminiSSEParser { pub accumulated_content: String, + /// The thinking streamed before the answer, kept so it can be stored with it. + pub accumulated_reasoning: String, pub accumulated_tool_calls: HashMap, pub events_str: String, pub stream_event_processor: Box, @@ -603,6 +614,7 @@ impl GeminiSSEParser { pub fn new(stream_event_processor: Box) -> Self { Self { accumulated_content: String::new(), + accumulated_reasoning: String::new(), accumulated_tool_calls: HashMap::new(), events_str: String::new(), stream_event_processor, @@ -621,6 +633,7 @@ impl SSEParser for GeminiSSEParser { }; if let Some(reasoning) = parsed.reasoning.filter(|s| !s.is_empty()) { + self.accumulated_reasoning.push_str(&reasoning); self.stream_event_processor .send( StreamingEvent::ReasoningTokenDelta { content: reasoning }, diff --git a/backend/windmill-ai/src/types.rs b/backend/windmill-ai/src/types.rs index 192fc6f09f..8dfb9af1c6 100644 --- a/backend/windmill-ai/src/types.rs +++ b/backend/windmill-ai/src/types.rs @@ -374,6 +374,13 @@ pub struct AIAgentResult<'a> { pub messages: Vec>, #[serde(skip_serializing_if = "Option::is_none")] pub wm_stream: Option, + /// The model's thinking across every iteration of the loop, in order, blank-line + /// separated. Present whenever the provider's parser surfaced any, whether or not + /// the step streams, so a downstream step never has to pick it out of `wm_stream`. + /// Absent when the model thought nothing, and on the OpenAI Responses path, whose + /// parser does not return reasoning yet. + #[serde(skip_serializing_if = "Option::is_none")] + pub reasoning: Option, #[serde(skip_serializing_if = "Option::is_none")] pub usage: Option, } diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 0a96f4e8bf..debfbeff39 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -1211,6 +1211,7 @@ pub async fn run_agent( *has_stream = user_wants_streaming && is_text_output; let mut final_events_str = String::new(); + let mut final_reasoning = String::new(); // Always create a StreamEventProcessor for text output (use silent mode if user doesn't want streaming) let stream_event_processor = if is_text_output { @@ -1463,6 +1464,7 @@ pub async fn run_agent( match parsed { ParsedResponse::Text { content: response_content, + reasoning: response_reasoning, tool_calls, events_str, annotations, @@ -1479,6 +1481,7 @@ pub async fn run_agent( if let Some(events_str) = events_str { final_events_str.push_str(&events_str); } + append_reasoning(&mut final_reasoning, response_reasoning.as_deref()); // Add websearch tool message if websearch was used if used_websearch { @@ -1801,6 +1804,7 @@ pub async fn run_agent( } else { None }, + reasoning: (!final_reasoning.is_empty()).then_some(final_reasoning), usage: if final_usage.as_ref().map(|u| u.is_empty()).unwrap_or(true) { None } else { @@ -1820,6 +1824,19 @@ fn streaming_requested(streaming: Option) -> bool { streaming.unwrap_or(true) } +/// Add one iteration's thinking to the step's. Every iteration thinks, and a tool-call +/// iteration's thinking is what led to the call, so the result keeps all of them in order, +/// blank-line separated, rather than only the answering turn's. +fn append_reasoning(accumulated: &mut String, reasoning: Option<&str>) { + let Some(reasoning) = reasoning.map(str::trim).filter(|r| !r.is_empty()) else { + return; + }; + if !accumulated.is_empty() { + accumulated.push_str("\n\n"); + } + accumulated.push_str(reasoning); +} + #[cfg(test)] mod tests { use super::*; @@ -1832,6 +1849,16 @@ mod tests { } } + #[test] + fn reasoning_keeps_every_iteration_in_order() { + let mut acc = String::new(); + append_reasoning(&mut acc, Some("I need both cities.\n\n")); + append_reasoning(&mut acc, None); + append_reasoning(&mut acc, Some(" ")); + append_reasoning(&mut acc, Some("Paris is closer.")); + assert_eq!(acc, "I need both cities.\n\nParis is closer."); + } + #[test] fn an_unwritten_streaming_field_streams() { assert!(streaming_requested(None));