feat(providers): add streaming support for Google Gemini provider (#6191)

Signed-off-by: rabi <ramishra@redhat.com>
2026-01-08 00:47:12 +05:30
parent f7fe570b55
commit a4c7a746a0
3 changed files with 586 additions and 129 deletions
@@ -3,11 +3,12 @@ use crate::providers::base::Usage;
 use crate::providers::errors::ProviderError;
 use crate::providers::utils::{is_valid_function_name, sanitize_function_name};
 use anyhow::Result;
-use rand::{distributions::Alphanumeric, Rng};
 use rmcp::model::{
    object, AnnotateAble, CallToolRequestParam, ErrorCode, ErrorData, RawContent, Role, Tool,
 };
+use serde::Serialize;
 use std::borrow::Cow;
+use uuid::Uuid;

 use crate::conversation::message::{Message, MessageContent, ProviderMetadata};
 use serde_json::{json, Map, Value};
@@ -306,87 +307,125 @@ pub fn process_map(map: &Map<String, Value>, parent_key: Option<&str>) -> Value
    Value::Object(filtered_map)
 }

+#[derive(Clone, Copy)]
+enum SignedTextHandling {
+    SkipSignedText,
+    SignedTextAsThinking,
+    SignedTextAsRegularText,
+}
+
+pub fn process_response_part(
+    part: &Value,
+    last_signature: &mut Option<String>,
+) -> Option<MessageContent> {
+    // For streaming: skip text with signatures (matches Anthropic/OpenAI behavior)
+    process_response_part_impl(part, last_signature, SignedTextHandling::SkipSignedText)
+}
+
+fn process_response_part_non_streaming(
+    part: &Value,
+    last_signature: &mut Option<String>,
+    has_function_calls: bool,
+) -> Option<MessageContent> {
+    // For non-streaming: signed text is thinking only if there are function calls
+    let handling = if has_function_calls {
+        SignedTextHandling::SignedTextAsThinking
+    } else {
+        SignedTextHandling::SignedTextAsRegularText
+    };
+    process_response_part_impl(part, last_signature, handling)
+}
+
+fn process_response_part_impl(
+    part: &Value,
+    last_signature: &mut Option<String>,
+    signed_text_handling: SignedTextHandling,
+) -> Option<MessageContent> {
+    let signature = part.get(THOUGHT_SIGNATURE_KEY).and_then(|v| v.as_str());
+
+    if let Some(sig) = signature {
+        *last_signature = Some(sig.to_string());
+    }
+
+    let text_value = part.get("text");
+    if let Some(text) = text_value.and_then(|v| v.as_str()) {
+        if text.is_empty() {
+            return None;
+        }
+        match (signature, signed_text_handling) {
+            (Some(_), SignedTextHandling::SkipSignedText) => None,
+            (Some(sig), SignedTextHandling::SignedTextAsThinking) => {
+                Some(MessageContent::thinking(text.to_string(), sig.to_string()))
+            }
+            _ => Some(MessageContent::text(text.to_string())),
+        }
+    } else if text_value.is_some() {
+        tracing::warn!(
+            "Google response part has 'text' field but it's not a string: {:?}",
+            text_value
+        );
+        None
+    } else if let Some(function_call) = part.get("functionCall") {
+        let id = Uuid::new_v4().to_string();
+        let name = function_call["name"].as_str().unwrap_or_default();
+
+        if !is_valid_function_name(name) {
+            let error = ErrorData {
+                code: ErrorCode::INVALID_REQUEST,
+                message: Cow::from(format!(
+                    "The provided function name '{}' had invalid characters, it must match this regex [a-zA-Z0-9_-]+",
+                    name
+                )),
+                data: None,
+            };
+            Some(MessageContent::tool_request(id, Err(error)))
+        } else {
+            let arguments = function_call
+                .get("args")
+                .map(|params| object(params.clone()));
+            let effective_signature = signature.or(last_signature.as_deref());
+            let metadata = effective_signature.map(metadata_with_signature);
+
+            Some(MessageContent::tool_request_with_metadata(
+                id,
+                Ok(CallToolRequestParam {
+                    name: name.to_string().into(),
+                    arguments,
+                }),
+                metadata.as_ref(),
+            ))
+        }
+    } else {
+        None
+    }
+}
+
 pub fn response_to_message(response: Value) -> Result<Message> {
-    let mut content = Vec::new();
-    let binding = vec![];
-    let candidates: &Vec<Value> = response
-        .get("candidates")
-        .and_then(|v| v.as_array())
-        .unwrap_or(&binding);
-    let candidate = candidates.first();
    let role = Role::Assistant;
    let created = chrono::Utc::now().timestamp();
-    if candidate.is_none() {
-        return Ok(Message::new(role, created, content));
-    }
-    let candidate = candidate.unwrap();
-    let parts = candidate
-        .get("content")
-        .and_then(|content| content.get("parts"))
-        .and_then(|parts| parts.as_array())
-        .unwrap_or(&binding);

-    // Track the last seen thought signature to use as fallback for function calls without one
-    // This handles cases where Google's API returns multiple function calls but only includes
-    // thoughtSignature on some of them
-    let mut last_signature: Option<String> = None;
+    let parts = response
+        .get("candidates")
+        .and_then(|v| v.as_array())
+        .and_then(|c| c.first())
+        .and_then(|c| c.get("content"))
+        .and_then(|c| c.get("parts"))
+        .and_then(|p| p.as_array());
+
+    let Some(parts) = parts else {
+        return Ok(Message::new(role, created, Vec::new()));
+    };

    let has_function_calls = parts.iter().any(|p| p.get("functionCall").is_some());

+    let mut content = Vec::new();
+    let mut last_signature: Option<String> = None;
+
    for part in parts {
-        let signature = part
-            .get(THOUGHT_SIGNATURE_KEY)
-            .and_then(|v| v.as_str())
-            .map(|s| s.to_string());
-
-        if signature.is_some() {
-            last_signature = signature.clone();
-        }
-
-        if let Some(text) = part.get("text").and_then(|v| v.as_str()) {
-            // Text is "thinking" only if:
-            // 1. It has a signature AND
-            // 2. The response also contains function calls (meaning this is reasoning before acting)
-            // If there are no function calls, this is the final response and should be shown
-            if let (Some(sig), true) = (&signature, has_function_calls) {
-                content.push(MessageContent::thinking(text.to_string(), sig.clone()));
-            } else {
-                content.push(MessageContent::text(text.to_string()));
-            }
-        } else if let Some(function_call) = part.get("functionCall") {
-            let id: String = rand::thread_rng()
-                .sample_iter(&Alphanumeric)
-                .take(8)
-                .map(char::from)
-                .collect();
-            let name = function_call["name"]
-                .as_str()
-                .unwrap_or_default()
-                .to_string();
-            if !is_valid_function_name(&name) {
-                let error = ErrorData {
-                    code: ErrorCode::INVALID_REQUEST,
-                    message: Cow::from(format!(
-                        "The provided function name '{}' had invalid characters, it must match this regex [a-zA-Z0-9_-]+",
-                        name
-                    )),
-                    data: None,
-                };
-                content.push(MessageContent::tool_request(id, Err(error)));
-            } else {
-                let parameters = function_call.get("args");
-                let arguments = parameters.map(|params| object(params.clone()));
-                let effective_signature = signature.as_deref().or(last_signature.as_deref());
-                let metadata = effective_signature.map(metadata_with_signature);
-                content.push(MessageContent::tool_request_with_metadata(
-                    id,
-                    Ok(CallToolRequestParam {
-                        name: name.into(),
-                        arguments,
-                    }),
-                    metadata.as_ref(),
-                ));
-            }
+        if let Some(msg_content) =
+            process_response_part_non_streaming(part, &mut last_signature, has_function_calls)
+        {
+            content.push(msg_content);
        }
    }
    Ok(Message::new(role, created, content))
@@ -418,37 +457,197 @@ pub fn get_usage(data: &Value) -> Result<Usage> {
    }
 }

-/// Create a complete request payload for Google's API
+pub fn response_to_streaming_message<S>(
+    mut stream: S,
+) -> impl futures::Stream<
+    Item = anyhow::Result<(
+        Option<Message>,
+        Option<crate::providers::base::ProviderUsage>,
+    )>,
+> + 'static
+where
+    S: futures::Stream<Item = anyhow::Result<String>> + Unpin + Send + 'static,
+{
+    use async_stream::try_stream;
+    use futures::StreamExt;
+
+    try_stream! {
+        let mut final_usage: Option<crate::providers::base::ProviderUsage> = None;
+        let mut last_signature: Option<String> = None;
+        let stream_id = Uuid::new_v4().to_string();
+        let mut incomplete_data: Option<String> = None;
+
+        while let Some(line_result) = stream.next().await {
+            let line = line_result?;
+
+            if line.trim().is_empty() {
+                continue;
+            }
+
+            let data_part = if line.starts_with("data: ") {
+                line.strip_prefix("data: ").unwrap()
+            } else if line.starts_with("event:") || line.starts_with("id:") || line.starts_with("retry:") {
+                continue;
+            } else if incomplete_data.is_some() {
+                &line
+            } else {
+                continue;
+            };
+
+            if data_part.trim() == "[DONE]" {
+                break;
+            }
+
+            let chunk: Value = if let Some(ref mut incomplete) = incomplete_data {
+                incomplete.push_str(data_part);
+                match serde_json::from_str(incomplete) {
+                    Ok(v) => {
+                        incomplete_data = None;
+                        v
+                    }
+                    Err(e) => {
+                        if e.is_eof() {
+                            continue;
+                        }
+                        tracing::warn!("Failed to parse streaming chunk: {}", e);
+                        incomplete_data = None;
+                        continue;
+                    }
+                }
+            } else {
+                match serde_json::from_str(data_part) {
+                    Ok(v) => v,
+                    Err(e) => {
+                        if e.is_eof() {
+                            incomplete_data = Some(data_part.to_string());
+                            continue;
+                        }
+                        tracing::warn!("Failed to parse streaming chunk: {}", e);
+                        continue;
+                    }
+                }
+            };
+
+            if let Some(error) = chunk.get("error") {
+                let message = error
+                    .get("message")
+                    .and_then(|m| m.as_str())
+                    .unwrap_or("Unknown error");
+                let status = error
+                    .get("status")
+                    .and_then(|s| s.as_str())
+                    .unwrap_or("UNKNOWN");
+                Err(anyhow::anyhow!("Google API error ({}): {}", status, message))?;
+            }
+
+            if let Ok(usage) = get_usage(&chunk) {
+                if usage.input_tokens.is_some() || usage.output_tokens.is_some() {
+                    let model = chunk.get("modelVersion")
+                        .and_then(|v| v.as_str())
+                        .unwrap_or("unknown")
+                        .to_string();
+                    final_usage = Some(crate::providers::base::ProviderUsage::new(model, usage));
+                }
+            }
+
+            let parts = chunk
+                .get("candidates")
+                .and_then(|v| v.as_array())
+                .and_then(|c| c.first())
+                .and_then(|c| c.get("content"))
+                .and_then(|c| c.get("parts"))
+                .and_then(|p| p.as_array());
+
+            if let Some(parts) = parts {
+                for part in parts {
+                    if let Some(content) = process_response_part(part, &mut last_signature) {
+                        let message = Message::new(
+                            Role::Assistant,
+                            chrono::Utc::now().timestamp(),
+                            vec![content],
+                        ).with_id(stream_id.clone());
+                        yield (Some(message), None);
+                    }
+                }
+            }
+        }
+
+        if let Some(usage) = final_usage {
+            yield (None, Some(usage));
+        }
+    }
+}
+
+#[derive(Serialize)]
+struct TextPart<'a> {
+    text: &'a str,
+}
+
+#[derive(Serialize)]
+struct SystemInstruction<'a> {
+    parts: [TextPart<'a>; 1],
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct ToolsWrapper {
+    function_declarations: Vec<Value>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct GenerationConfig {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    temperature: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    max_output_tokens: Option<i32>,
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "camelCase")]
+struct GoogleRequest<'a> {
+    system_instruction: SystemInstruction<'a>,
+    contents: Vec<Value>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    tools: Option<ToolsWrapper>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    generation_config: Option<GenerationConfig>,
+}
+
 pub fn create_request(
    model_config: &ModelConfig,
    system: &str,
    messages: &[Message],
    tools: &[Tool],
 ) -> Result<Value> {
-    let mut payload = Map::new();
-    payload.insert(
-        "system_instruction".to_string(),
-        json!({"parts": [{"text": system}]}),
-    );
-    payload.insert("contents".to_string(), json!(format_messages(messages)));
-    if !tools.is_empty() {
-        payload.insert(
-            "tools".to_string(),
-            json!({"functionDeclarations": format_tools(tools)}),
-        );
-    }
-    let mut generation_config = Map::new();
-    if let Some(temp) = model_config.temperature {
-        generation_config.insert("temperature".to_string(), json!(temp as f64));
-    }
-    if let Some(tokens) = model_config.max_tokens {
-        generation_config.insert("maxOutputTokens".to_string(), json!(tokens));
-    }
-    if !generation_config.is_empty() {
-        payload.insert("generationConfig".to_string(), json!(generation_config));
-    }
+    let tools_wrapper = if tools.is_empty() {
+        None
+    } else {
+        Some(ToolsWrapper {
+            function_declarations: format_tools(tools),
+        })
+    };

-    Ok(json!(payload))
+    let generation_config =
+        if model_config.temperature.is_some() || model_config.max_tokens.is_some() {
+            Some(GenerationConfig {
+                temperature: model_config.temperature.map(|t| t as f64),
+                max_output_tokens: model_config.max_tokens,
+            })
+        } else {
+            None
+        };
+
+    let request = GoogleRequest {
+        system_instruction: SystemInstruction {
+            parts: [TextPart { text: system }],
+        },
+        contents: format_messages(messages),
+        tools: tools_wrapper,
+        generation_config,
+    };
+
+    Ok(serde_json::to_value(request)?)
 }

 #[cfg(test)]
@@ -1033,12 +1232,201 @@ mod tests {
        assert_eq!(google_out[0]["parts"][0]["thoughtSignature"], SIG);
        assert_eq!(google_out[1]["parts"][0]["thoughtSignature"], SIG);

-        let final_response =
+        // Text-only response WITH signature but WITHOUT function calls should be regular text
+        // (per original behavior: thinking is only when reasoning before tool calls)
+        let final_response_with_sig =
            google_response(vec![json!({"text": "Done!", "thoughtSignature": SIG})]);
-        let final_native = response_to_message(final_response).unwrap();
+        let final_native_with_sig = response_to_message(final_response_with_sig).unwrap();
        assert!(
-            final_native.content[0].as_text().is_some(),
-            "Text-only = final answer"
+            final_native_with_sig.content[0].as_text().is_some(),
+            "Text with signature but no function calls should be regular text (final response)"
+        );
+
+        let final_response_no_sig = google_response(vec![json!({"text": "Done!"})]);
+        let final_native_no_sig = response_to_message(final_response_no_sig).unwrap();
+        assert!(
+            final_native_no_sig.content[0].as_text().is_some(),
+            "Text without signature is regular text"
        );
    }
+
+    const GOOGLE_TEXT_STREAM: &str = concat!(
+        r#"data: {"candidates": [{"content": {"role": "model", "#,
+        r#""parts": [{"text": "Hello"}]}}]}"#,
+        "\n",
+        r#"data: {"candidates": [{"content": {"role": "model", "#,
+        r#""parts": [{"text": " world"}]}}]}"#,
+        "\n",
+        r#"data: {"candidates": [{"content": {"role": "model", "#,
+        r#""parts": [{"text": "!"}]}}], "#,
+        r#""usageMetadata": {"promptTokenCount": 10, "#,
+        r#""candidatesTokenCount": 3, "totalTokenCount": 13}}"#
+    );
+
+    const GOOGLE_FUNCTION_STREAM: &str = concat!(
+        r#"data: {"candidates": [{"content": {"role": "model", "#,
+        r#""parts": [{"functionCall": {"name": "test_tool", "#,
+        r#""args": {"param": "value"}}}]}}], "#,
+        r#""usageMetadata": {"promptTokenCount": 5, "#,
+        r#""candidatesTokenCount": 2, "totalTokenCount": 7}}"#
+    );
+
+    #[tokio::test]
+    async fn test_streaming_text_response() {
+        use futures::StreamExt;
+
+        let lines: Vec<Result<String, anyhow::Error>> = GOOGLE_TEXT_STREAM
+            .lines()
+            .map(|l| Ok(l.to_string()))
+            .collect();
+        let stream = Box::pin(futures::stream::iter(lines));
+        let mut message_stream = std::pin::pin!(response_to_streaming_message(stream));
+
+        let mut text_parts = Vec::new();
+        let mut message_ids: Vec<Option<String>> = Vec::new();
+        let mut final_usage = None;
+
+        while let Some(result) = message_stream.next().await {
+            let (message, usage) = result.unwrap();
+            if let Some(msg) = message {
+                message_ids.push(msg.id.clone());
+                if let Some(MessageContent::Text(text)) = msg.content.first() {
+                    text_parts.push(text.text.clone());
+                }
+            }
+            if usage.is_some() {
+                final_usage = usage;
+            }
+        }
+
+        assert_eq!(text_parts, vec!["Hello", " world", "!"]);
+        let usage = final_usage.unwrap();
+        assert_eq!(usage.usage.input_tokens, Some(10));
+        assert_eq!(usage.usage.output_tokens, Some(3));
+
+        // Verify all streaming messages have consistent IDs for UI aggregation
+        assert!(
+            message_ids.iter().all(|id| id.is_some()),
+            "All streaming messages should have an ID"
+        );
+        let first_id = message_ids.first().unwrap();
+        assert!(
+            message_ids.iter().all(|id| id == first_id),
+            "All streaming messages should have the same ID"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_streaming_function_call() {
+        use futures::StreamExt;
+
+        let lines: Vec<Result<String, anyhow::Error>> = GOOGLE_FUNCTION_STREAM
+            .lines()
+            .map(|l| Ok(l.to_string()))
+            .collect();
+        let stream = Box::pin(futures::stream::iter(lines));
+        let mut message_stream = std::pin::pin!(response_to_streaming_message(stream));
+
+        let mut tool_calls = Vec::new();
+
+        while let Some(result) = message_stream.next().await {
+            let (message, _usage) = result.unwrap();
+            if let Some(msg) = message {
+                if let Some(MessageContent::ToolRequest(req)) = msg.content.first() {
+                    if let Ok(tool_call) = &req.tool_call {
+                        tool_calls.push(tool_call.name.to_string());
+                    }
+                }
+            }
+        }
+
+        assert_eq!(tool_calls, vec!["test_tool"]);
+    }
+
+    #[tokio::test]
+    async fn test_streaming_error_response() {
+        use futures::StreamExt;
+
+        let error_stream = concat!(
+            r#"data: {"error": {"code": 400, "#,
+            r#""message": "Invalid request", "status": "INVALID_ARGUMENT"}}"#
+        );
+        let lines: Vec<Result<String, anyhow::Error>> =
+            error_stream.lines().map(|l| Ok(l.to_string())).collect();
+        let stream = Box::pin(futures::stream::iter(lines));
+        let mut message_stream = std::pin::pin!(response_to_streaming_message(stream));
+
+        let result = message_stream.next().await;
+        assert!(result.is_some());
+        let err = result.unwrap();
+        assert!(err.is_err());
+        let error_msg = err.unwrap_err().to_string();
+        assert!(error_msg.contains("INVALID_ARGUMENT"));
+        assert!(error_msg.contains("Invalid request"));
+    }
+
+    #[tokio::test]
+    async fn test_streaming_with_sse_event_lines() {
+        use futures::StreamExt;
+
+        // SSE format can include event: lines which should be skipped
+        let sse_stream = r#"event: message
+data: {"candidates": [{"content": {"role": "model", "parts": [{"text": "Hello"}]}}]}
+
+event: message
+data: {"candidates": [{"content": {"role": "model", "parts": [{"text": " world"}]}}]}
+
+data: [DONE]"#;
+        let lines: Vec<Result<String, anyhow::Error>> =
+            sse_stream.lines().map(|l| Ok(l.to_string())).collect();
+        let stream = Box::pin(futures::stream::iter(lines));
+        let mut message_stream = std::pin::pin!(response_to_streaming_message(stream));
+
+        let mut text_parts = Vec::new();
+
+        while let Some(result) = message_stream.next().await {
+            let (message, _usage) = result.unwrap();
+            if let Some(msg) = message {
+                if let Some(MessageContent::Text(text)) = msg.content.first() {
+                    text_parts.push(text.text.clone());
+                }
+            }
+        }
+
+        assert_eq!(text_parts, vec!["Hello", " world"]);
+    }
+
+    #[tokio::test]
+    async fn test_streaming_handles_done_signal() {
+        use futures::StreamExt;
+
+        let stream_with_done = concat!(
+            r#"data: {"candidates": [{"content": {"role": "model", "#,
+            r#""parts": [{"text": "Complete"}]}}]}"#,
+            "\n",
+            "data: [DONE]\n",
+            r#"data: {"candidates": [{"content": {"role": "model", "#,
+            r#""parts": [{"text": "Should not appear"}]}}]}"#
+        );
+        let lines: Vec<Result<String, anyhow::Error>> = stream_with_done
+            .lines()
+            .map(|l| Ok(l.to_string()))
+            .collect();
+        let stream = Box::pin(futures::stream::iter(lines));
+        let mut message_stream = std::pin::pin!(response_to_streaming_message(stream));
+
+        let mut text_parts = Vec::new();
+
+        while let Some(result) = message_stream.next().await {
+            let (message, _usage) = result.unwrap();
+            if let Some(msg) = message {
+                if let Some(MessageContent::Text(text)) = msg.content.first() {
+                    text_parts.push(text.text.clone());
+                }
+            }
+        }
+
+        // Only "Complete" should be captured, stream should stop at [DONE]
+        assert_eq!(text_parts, vec!["Complete"]);
+    }
 }
@@ -1,16 +1,28 @@
 use super::api_client::{ApiClient, AuthMethod};
+use super::base::MessageStream;
 use super::errors::ProviderError;
 use super::retry::ProviderRetry;
-use super::utils::{handle_response_google_compat, unescape_json_values, RequestLog};
+use super::utils::{
+    handle_response_google_compat, handle_status_openai_compat, unescape_json_values, RequestLog,
+};
 use crate::conversation::message::Message;

 use crate::model::ModelConfig;
 use crate::providers::base::{ConfigKey, Provider, ProviderMetadata, ProviderUsage};
-use crate::providers::formats::google::{create_request, get_usage, response_to_message};
+use crate::providers::formats::google::{
+    create_request, get_usage, response_to_message, response_to_streaming_message,
+};
 use anyhow::Result;
+use async_stream::try_stream;
 use async_trait::async_trait;
+use futures::TryStreamExt;
 use rmcp::model::Tool;
 use serde_json::Value;
+use std::io;
+use tokio::pin;
+use tokio_stream::StreamExt;
+use tokio_util::codec::{FramedRead, LinesCodec};
+use tokio_util::io::StreamReader;

 pub const GOOGLE_API_HOST: &str = "https://generativelanguage.googleapis.com";
 pub const GOOGLE_DEFAULT_MODEL: &str = "gemini-2.5-pro";
@@ -84,6 +96,16 @@ impl GoogleProvider {
        let response = self.api_client.response_post(&path, payload).await?;
        handle_response_google_compat(response).await
    }
+
+    async fn post_stream(
+        &self,
+        model_name: &str,
+        payload: &Value,
+    ) -> Result<reqwest::Response, ProviderError> {
+        let path = format!("v1beta/models/{}:streamGenerateContent?alt=sse", model_name);
+        let response = self.api_client.response_post(&path, payload).await?;
+        handle_status_openai_compat(response).await
+    }
 }

 #[async_trait]
@@ -126,10 +148,7 @@ impl Provider for GoogleProvider {
        let mut log = RequestLog::start(model_config, &payload)?;

        let response = self
-            .with_retry(|| async {
-                let payload_clone = payload.clone();
-                self.post(&model_config.model_name, &payload_clone).await
-            })
+            .with_retry(|| async { self.post(&model_config.model_name, &payload).await })
            .await?;

        let message = response_to_message(unescape_json_values(&response))?;
@@ -158,4 +177,45 @@ impl Provider for GoogleProvider {
        models.sort();
        Ok(Some(models))
    }
+
+    fn supports_streaming(&self) -> bool {
+        true
+    }
+
+    async fn stream(
+        &self,
+        system: &str,
+        messages: &[Message],
+        tools: &[Tool],
+    ) -> Result<MessageStream, ProviderError> {
+        let payload = create_request(&self.model, system, messages, tools)?;
+        let mut log = RequestLog::start(&self.model, &payload)?;
+
+        let response = self
+            .with_retry(|| async { self.post_stream(&self.model.model_name, &payload).await })
+            .await
+            .inspect_err(|e| {
+                let _ = log.error(e);
+            })?;
+
+        let stream = response.bytes_stream().map_err(io::Error::other);
+
+        Ok(Box::pin(try_stream! {
+            let stream_reader = StreamReader::new(stream);
+            let framed = FramedRead::new(stream_reader, LinesCodec::new())
+                .map_err(anyhow::Error::from);
+
+            let message_stream = response_to_streaming_message(framed);
+            pin!(message_stream);
+            while let Some(message) = message_stream.next().await {
+                let (message, usage) = message.map_err(|e|
+                    ProviderError::RequestFailed(format!("Stream decode error: {}", e))
+                )?;
+                if message.is_some() || usage.is_some() {
+                    log.write(&message, usage.as_ref().map(|f| f.usage).as_ref())?;
+                }
+                yield (message, usage);
+            }
+        }))
+    }
 }
@@ -12,12 +12,13 @@ use regex::Regex;
 use reqwest::{Response, StatusCode};
 use rmcp::model::{AnnotateAble, ImageContent, RawImageContent};
 use serde::{Deserialize, Serialize};
-use serde_json::{json, Map, Value};
+use serde_json::{json, Value};
 use std::fmt::Display;
 use std::fs::File;
 use std::io;
 use std::io::{BufWriter, Read, Write};
 use std::path::{Path, PathBuf};
+use std::sync::OnceLock;
 use std::time::Duration;
 use tokio::pin;
 use tokio_stream::StreamExt;
@@ -324,12 +325,14 @@ pub async fn handle_response_google_compat(response: Response) -> Result<Value,
 }

 pub fn sanitize_function_name(name: &str) -> String {
-    let re = Regex::new(r"[^a-zA-Z0-9_-]").unwrap();
+    static RE: OnceLock<Regex> = OnceLock::new();
+    let re = RE.get_or_init(|| Regex::new(r"[^a-zA-Z0-9_-]").unwrap());
    re.replace_all(name, "_").to_string()
 }

 pub fn is_valid_function_name(name: &str) -> bool {
-    let re = Regex::new(r"^[a-zA-Z0-9_-]+$").unwrap();
+    static RE: OnceLock<Regex> = OnceLock::new();
+    let re = RE.get_or_init(|| Regex::new(r"^[a-zA-Z0-9_-]+$").unwrap());
    re.is_match(name)
 }

@@ -435,31 +438,37 @@ pub fn load_image_file(path: &str) -> Result<ImageContent, ProviderError> {
 }

 pub fn unescape_json_values(value: &Value) -> Value {
+    let mut cloned = value.clone();
+    unescape_json_values_in_place(&mut cloned);
+    cloned
+}
+
+fn unescape_json_values_in_place(value: &mut Value) {
    match value {
        Value::Object(map) => {
-            let new_map: Map<String, Value> = map
-                .iter()
-                .map(|(k, v)| (k.clone(), unescape_json_values(v))) // Process each value
-                .collect();
-            Value::Object(new_map)
+            for v in map.values_mut() {
+                unescape_json_values_in_place(v);
+            }
        }
        Value::Array(arr) => {
-            let new_array: Vec<Value> = arr.iter().map(unescape_json_values).collect();
-            Value::Array(new_array)
+            for v in arr.iter_mut() {
+                unescape_json_values_in_place(v);
+            }
        }
        Value::String(s) => {
-            let unescaped = s
-                .replace("\\\\n", "\n")
-                .replace("\\\\t", "\t")
-                .replace("\\\\r", "\r")
-                .replace("\\\\\"", "\"")
-                .replace("\\n", "\n")
-                .replace("\\t", "\t")
-                .replace("\\r", "\r")
-                .replace("\\\"", "\"");
-            Value::String(unescaped)
+            if s.contains('\\') {
+                *s = s
+                    .replace("\\\\n", "\n")
+                    .replace("\\\\t", "\t")
+                    .replace("\\\\r", "\r")
+                    .replace("\\\\\"", "\"")
+                    .replace("\\n", "\n")
+                    .replace("\\t", "\t")
+                    .replace("\\r", "\r")
+                    .replace("\\\"", "\"");
+            }
        }
-        _ => value.clone(),
+        _ => {}
    }
 }