From 94bcc00554423eb9e4056dd33ceff1b07263e9ec Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 30 Jul 2026 09:56:37 +0200 Subject: [PATCH] fix(ai): route Azure OpenAI agent steps through the Responses API (#10404) * fix(ai): route Azure OpenAI agent steps through the Responses API * docs: note why azure responses routing is per-provider * fix(ai): fall back to chat/completions when an azure endpoint rejects responses * refactor(ai): retry endpoint and stream_options rejections in one loop * fix(ai): keep a rejected request shape dropped across agent iterations * perf(ai): remember which deployments reject the responses route * fix(ai): only remember a route rejection the fallback resolved * fix(ai): keep web search on azure and require the deployment be named to reroute * fix(ai): remember an unserved route only on a 404 * docs: correct the reroute-flag and fallback-hook comments --- backend/windmill-ai/src/providers/mod.rs | 70 +++++++- backend/windmill-ai/src/providers/openai.rs | 4 + backend/windmill-ai/src/query_builder.rs | 8 + backend/windmill-worker/src/ai_executor.rs | 181 ++++++++++++-------- 4 files changed, 188 insertions(+), 75 deletions(-) diff --git a/backend/windmill-ai/src/providers/mod.rs b/backend/windmill-ai/src/providers/mod.rs index 9bfb47a36b..094cc71490 100644 --- a/backend/windmill-ai/src/providers/mod.rs +++ b/backend/windmill-ai/src/providers/mod.rs @@ -6,6 +6,10 @@ pub mod openai; pub mod openrouter; pub mod other; +use std::time::{Duration, Instant}; + +use windmill_common::cache::Cache; + use crate::{ ai_providers::{AIPlatform, AIProvider}, credentials::ProviderCredentials, @@ -29,7 +33,13 @@ pub fn create_query_builder( ) -> Box { match credentials.provider { AIProvider::GoogleAI => Box::new(GoogleAIQueryBuilder::new(credentials.platform.clone())), - AIProvider::OpenAI => Box::new(OpenAIQueryBuilder::new(credentials.provider.clone())), + // Azure OpenAI serves the same Responses API as OpenAI under `/openai/v1`, and + // newer deployments reject a reasoning effort combined with function tools on + // `/chat/completions`. This cannot be decided per model: on Azure the model name + // is a user-chosen deployment name, which says nothing about the model behind it. + AIProvider::OpenAI | AIProvider::AzureOpenAI => { + Box::new(OpenAIQueryBuilder::new(credentials.provider.clone())) + } AIProvider::Anthropic => Box::new(AnthropicQueryBuilder::new( credentials.provider.clone(), credentials.platform.clone(), @@ -42,6 +52,62 @@ pub fn create_query_builder( } } +/// The builder for the same resource on the OpenAI-compatible `/chat/completions` +/// surface, for a provider whose preferred surface it turned out not to serve +/// (`QueryBuilder::supports_chat_completions_fallback`). +pub fn create_chat_completions_query_builder( + credentials: &ProviderCredentials, +) -> Box { + Box::new(OtherQueryBuilder::new(credentials.provider.clone())) +} + +lazy_static::lazy_static! { + /// Deployments that turned out not to serve the endpoint their provider prefers, + /// so the steps after the one that found out start on `/chat/completions` rather + /// than paying a rejected call each to learn the same thing. Keyed by endpoint and + /// deployment name, because Azure's Responses API support varies by both. + /// + /// Only rejections are remembered: a surface that answers costs nothing to keep + /// using. The entry expires because a deployment can gain Responses support + /// (Azure rolls it out per model and region) without anything here changing. + static ref CHAT_COMPLETIONS_ONLY: Cache<(String, String), Instant> = Cache::new(500); +} + +const CHAT_COMPLETIONS_ONLY_TTL: Duration = Duration::from_secs(3600); + +/// Whether this deployment is known not to serve the endpoint its provider prefers. +pub fn is_chat_completions_only(base_url: &str, model: &str) -> bool { + CHAT_COMPLETIONS_ONLY + .get(&(base_url.to_string(), model.to_string())) + .is_some_and(|learned_at| learned_at.elapsed() < CHAT_COMPLETIONS_ONLY_TTL) +} + +/// Record that this deployment rejected the endpoint its provider prefers. +pub fn remember_chat_completions_only(base_url: &str, model: &str) { + CHAT_COMPLETIONS_ONLY.insert((base_url.to_string(), model.to_string()), Instant::now()); +} + +#[cfg(test)] +mod chat_completions_only_tests { + use super::*; + + /// Azure's Responses API support varies by deployment, so what one deployment of a + /// resource rejected says nothing about the next one. + #[test] + fn a_rejection_is_remembered_per_deployment() { + let base_url = "https://rejection-per-deployment.openai.azure.com/openai"; + + remember_chat_completions_only(base_url, "legacy-deployment"); + + assert!(is_chat_completions_only(base_url, "legacy-deployment")); + assert!(!is_chat_completions_only(base_url, "gpt-5-deployment")); + assert!(!is_chat_completions_only( + "https://other.openai.azure.com/openai", + "legacy-deployment" + )); + } +} + /// The proxy (workspace/instance AI settings) and the query builder (AI agent step) /// each derive the endpoint and the credential header for the same resource. They /// must agree, or a resource authenticates in one and 401s in the other. @@ -139,7 +205,7 @@ mod parity_tests { AIProvider::AzureOpenAI, "https://example.openai.azure.com/openai", "gpt-4o", - "chat/completions", + "responses", ), case( AIProvider::OpenAI, diff --git a/backend/windmill-ai/src/providers/openai.rs b/backend/windmill-ai/src/providers/openai.rs index f33c5728ea..fd28281864 100644 --- a/backend/windmill-ai/src/providers/openai.rs +++ b/backend/windmill-ai/src/providers/openai.rs @@ -508,6 +508,10 @@ impl QueryBuilder for OpenAIQueryBuilder { build_openai_compatible_proxy_request(args) } + fn supports_chat_completions_fallback(&self, base_url: &str) -> bool { + self.provider_kind.is_azure(base_url) + } + async fn parse_streaming_response( &self, response: reqwest::Response, diff --git a/backend/windmill-ai/src/query_builder.rs b/backend/windmill-ai/src/query_builder.rs index 79fca4bd7d..5a4cecac87 100644 --- a/backend/windmill-ai/src/query_builder.rs +++ b/backend/windmill-ai/src/query_builder.rs @@ -77,6 +77,14 @@ pub trait QueryBuilder: Send + Sync { false } + /// Whether a rejected request may be retried on the `/chat/completions` surface of + /// the same resource. Azure serves the Responses API for only a subset of models and + /// regions, and an Azure model name is a user-chosen deployment name, so such a + /// resource can only find out by being asked. + fn supports_chat_completions_fallback(&self, _base_url: &str) -> bool { + false + } + /// Build a provider-specific request from an OpenAI-compatible proxy request. fn build_proxy_request(&self, args: &ProxyBuildArgs<'_>) -> Result { Err(Error::BadRequest(format!( diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 2c078e3dd4..ec470fb966 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -23,7 +23,10 @@ use crate::ai::tools::McpClientStub as McpClient; use windmill_ai::{ ai_providers::AIProvider, image_handler::upload_image_to_s3, - providers::create_query_builder, + providers::{ + create_chat_completions_query_builder, create_query_builder, is_chat_completions_only, + remember_chat_completions_only, + }, proxy::{ common_outbound_headers, needs_unavailable_oauth_exchange, retain_effective_credentials, }, @@ -834,7 +837,15 @@ pub async fn run_agent( let api_key = credentials.api_key.as_deref().unwrap_or(""); // Create the query builder for the provider - let query_builder = create_query_builder(&credentials, args.provider.get_model()); + let mut query_builder = create_query_builder(&credentials, args.provider.get_model()); + if query_builder.supports_chat_completions_fallback(base_url) + && is_chat_completions_only(base_url, args.provider.get_model()) + { + query_builder = create_chat_completions_query_builder(&credentials); + } + // Both outlive the iteration that discovers them: a request shape or a route the + // endpoint rejected once stays rejected for the whole step. + let mut include_usage = true; // Initialize messages let mut messages = @@ -1149,20 +1160,13 @@ pub async fn run_agent( has_websearch, }; - let request_body = query_builder - .build_request(&build_args, client, &job.workspace_id) - .await?; - - let endpoint = - query_builder.get_endpoint(base_url, args.provider.get_model(), output_type); - let auth_headers = query_builder.get_auth_headers(api_key, base_url, output_type); // A worker cannot run the client credentials exchange, so an OAuth resource // has no token here: the request would carry an empty credential and come // back 401. if needs_unavailable_oauth_exchange( &credentials, args.provider.resource.token_url.as_deref(), - &auth_headers, + &query_builder.get_auth_headers(api_key, base_url, output_type), ) { return Err(Error::ExecutionErr(format!( "The {:?} resource authenticates with OAuth, which AI agent steps do not \ @@ -1171,7 +1175,6 @@ pub async fn run_agent( credentials.provider ))); } - let auth_headers = retain_effective_credentials(&credentials, auth_headers); let timeout = resolve_job_timeout(conn, &job.workspace_id, job.id, job.timeout) .await @@ -1185,75 +1188,107 @@ pub async fn run_agent( let pinned_ai_client = pinned_ai_client_for(base_url).await?; // Helper to build HTTP request with headers - let build_http_request = |body: String| { - let mut req = pinned_ai_client - .post(&endpoint) - .timeout(timeout) - .header("Content-Type", "application/json"); + let build_http_request = + |endpoint: &str, auth_headers: &[(&'static str, String)], body: String| { + let mut req = pinned_ai_client + .post(endpoint) + .timeout(timeout) + .header("Content-Type", "application/json"); - for (header_name, header_value) in &auth_headers { - req = req.header(*header_name, header_value.clone()); - } + for (header_name, header_value) in auth_headers { + req = req.header(*header_name, header_value.clone()); + } - for (header_name, header_value) in &trailing_headers { - req = req.header(header_name.as_str(), header_value.as_str()); - } + for (header_name, header_value) in &trailing_headers { + req = req.header(header_name.as_str(), header_value.as_str()); + } - req.body(body) - }; + req.body(body) + }; - let resp = build_http_request(request_body.clone()) - .send() - .await - .map_err(|e| Error::internal_err(format!("Failed to call API: {}", e)))?; + // An endpoint can reject the request shape rather than the model: + // `stream_options`, which not every OpenAI-compatible provider accepts, and + // the route itself, when an Azure resource is outside the Responses API's + // model/region matrix. Each is retried once with that part dropped. + // Set where the route is found to be absent, and read once the fallback has + // answered: a rejection it did not resolve says nothing about the deployment. + let mut rerouted_by_a_route_rejection = false; + let resp = loop { + let request_body = if include_usage { + query_builder + .build_request(&build_args, client, &job.workspace_id) + .await? + } else { + query_builder + .build_request_without_usage(&build_args, client, &job.workspace_id) + .await? + }; + let endpoint = + query_builder.get_endpoint(base_url, args.provider.get_model(), output_type); + let auth_headers = retain_effective_credentials( + &credentials, + query_builder.get_auth_headers(api_key, base_url, output_type), + ); - // Check if request failed and we should retry without stream_options - let resp = match resp.error_for_status_ref() { - Ok(_) => resp, - Err(e) => { - let status = resp.status(); - let text = resp - .text() - .await - .unwrap_or_else(|_| "".to_string()); + let resp = build_http_request(&endpoint, &auth_headers, request_body) + .send() + .await + .map_err(|e| Error::internal_err(format!("Failed to call API: {}", e)))?; - // Retry without stream_options if provider supports it and error suggests incompatibility - // Common error patterns: 400 Bad Request with mentions of stream_options or include_usage - let should_retry = query_builder.supports_retry_without_usage() - && status.as_u16() == 400 - && (text.contains("stream_options") - || text.contains("include_usage") - || text.contains("Additional properties are not allowed")); - - if should_retry { - tracing::info!( - "Retrying request without stream_options due to provider incompatibility" - ); - - let retry_body = query_builder - .build_request_without_usage(&build_args, client, &job.workspace_id) - .await?; - - let retry_resp = - build_http_request(retry_body).send().await.map_err(|e| { - Error::internal_err(format!("Failed to call API on retry: {}", e)) - })?; - - match retry_resp.error_for_status_ref() { - Ok(_) => retry_resp, - Err(retry_e) => { - let retry_text = retry_resp - .text() - .await - .unwrap_or_else(|_| "".to_string()); - return Err(Error::internal_err(format!( - "API error on retry: {} - {}", - retry_e, retry_text - ))); - } + match resp.error_for_status_ref() { + Ok(_) => { + if rerouted_by_a_route_rejection { + remember_chat_completions_only(base_url, args.provider.get_model()); + } + break resp; + } + Err(e) => { + let status = resp.status(); + let text = resp + .text() + .await + .unwrap_or_else(|_| "".to_string()); + + // Common error patterns: 400 Bad Request with mentions of stream_options or include_usage + let rejects_usage_tracking = include_usage + && query_builder.supports_retry_without_usage() + && status.as_u16() == 400 + && (text.contains("stream_options") + || text.contains("include_usage") + || text.contains("Additional properties are not allowed")); + + // Only the first call of the step may re-route: an endpoint that + // does not serve this API rejects that one already, whereas a + // rejection once the conversation is under way is about the + // conversation (context length, content filter, tool schema). + let route_unserved = i == 0 + && query_builder.supports_chat_completions_fallback(base_url) + && matches!(status.as_u16(), 400 | 404) + && *output_type == OutputType::Text; + + if rejects_usage_tracking { + tracing::info!( + "Retrying request without stream_options due to provider incompatibility" + ); + include_usage = false; + } else if route_unserved { + tracing::info!( + "Endpoint rejected the request ({}), falling back to chat/completions", + status + ); + // Only a 404 says the route is absent. A 400 is ambiguous — + // a deployment that does serve the route rejects tool + // schemas, blocked hosted tools and filtered content the + // same way — so it re-routes this step and nothing more. + rerouted_by_a_route_rejection = status.as_u16() == 404; + query_builder = create_chat_completions_query_builder(&credentials); + include_usage = true; + } else { + return Err(Error::internal_err(format!( + "API error calling {}: {} - {}", + endpoint, e, text + ))); } - } else { - return Err(Error::internal_err(format!("API error: {} - {}", e, text))); } } };