1use std::collections::HashMap;
2
3use async_openai::Client;
4use async_openai::config::OpenAIConfig;
5use async_openai::types::responses::{
6 CreateResponse, EasyInputContent, EasyInputMessage, FunctionCallOutput, FunctionCallOutputItemParam, FunctionTool,
7 FunctionToolCall, ImageDetail, IncludeEnum, InputContent, InputImageContent, InputItem, InputParam,
8 InputTextContent, Item, MessageType, OutputItem, Reasoning, ReasoningSummary, ResponseStreamEvent, ResponseUsage,
9 Role, Tool,
10};
11use serde_json::Value;
12use tokio_stream::StreamExt;
13use tracing::{debug, error};
14
15use crate::provider::get_context_window;
16use crate::providers::openai_compatible::AetherOpenAiConfig;
17use crate::{
18 ChatMessage, ContentBlock, Context, LlmError, LlmModel, LlmResponse, LlmResponseStream, ProviderAuthMode,
19 ProviderConnectionConfig, ProviderFactory, Result, StopReason, StreamingModelProvider, TokenUsage, ToolDefinition,
20};
21
22impl From<ResponseUsage> for TokenUsage {
23 fn from(usage: ResponseUsage) -> Self {
24 TokenUsage {
25 input_tokens: usage.input_tokens,
26 output_tokens: usage.output_tokens,
27 cache_read_tokens: Some(usage.input_tokens_details.cached_tokens),
28 reasoning_tokens: Some(usage.output_tokens_details.reasoning_tokens),
29 ..TokenUsage::default()
30 }
31 }
32}
33
34pub(crate) fn map_user_content_for_responses(parts: &[ContentBlock]) -> Result<EasyInputContent> {
35 let mut items = Vec::with_capacity(parts.len());
36 for p in parts {
37 match p {
38 ContentBlock::Text { text } => {
39 items.push(InputContent::InputText(InputTextContent { text: text.clone() }));
40 }
41 ContentBlock::Image { .. } => {
42 items.push(InputContent::InputImage(InputImageContent {
43 detail: ImageDetail::Auto,
44 file_id: None,
45 image_url: Some(p.as_data_uri().unwrap()),
46 }));
47 }
48 ContentBlock::Audio { .. } => {
49 return Err(LlmError::UnsupportedContent("OpenAI Responses does not support audio input".into()));
50 }
51 }
52 }
53 Ok(EasyInputContent::ContentList(items))
54}
55
56pub struct OpenAiProvider {
57 client: Client<AetherOpenAiConfig>,
58 model: String,
59}
60
61impl ProviderFactory for OpenAiProvider {
62 async fn from_env() -> Result<Self> {
63 Self::from_env_with_connection(ProviderConnectionConfig::default()).await
64 }
65
66 async fn from_env_with_connection(connection: ProviderConnectionConfig) -> Result<Self> {
67 let api_key = match connection.auth_mode {
68 ProviderAuthMode::Default => {
69 std::env::var("OPENAI_API_KEY").map_err(|_| LlmError::MissingApiKey("OPENAI_API_KEY".to_string()))?
70 }
71 ProviderAuthMode::None => String::new(),
72 };
73
74 let mut config = OpenAIConfig::new().with_api_key(api_key);
75 if let Some(base_url) = connection.base_url {
76 config = config.with_api_base(base_url);
77 }
78 let config = AetherOpenAiConfig::new(config, connection.auth_mode);
79
80 Ok(Self { client: Client::with_config(config), model: "gpt-4.1".to_string() })
81 }
82
83 fn with_model(mut self, model: &str) -> Self {
84 if !model.is_empty() {
85 self.model = model.to_string();
86 }
87 self
88 }
89}
90
91impl StreamingModelProvider for OpenAiProvider {
92 fn stream_response(&self, context: &Context) -> LlmResponseStream {
93 let client = self.client.clone();
94 let model = self.model.clone();
95 let request = match build_wire_request(&model, context) {
96 Ok(req) => req,
97 Err(e) => return Box::pin(async_stream::stream! { yield Err(e); }),
98 };
99
100 Box::pin(async_stream::stream! {
101 debug!("Starting OpenAI Responses API stream for model: {model}");
102
103 let stream = match client.responses().create_stream_byot::<Value, ResponseStreamEvent>(request).await {
104 Ok(s) => s,
105 Err(e) => {
106 error!("Failed to create OpenAI Responses stream: {e:?}");
107 yield Err(LlmError::ApiRequest(e.to_string()));
108 return;
109 }
110 };
111
112 let mut stream = Box::pin(stream);
113 let mut fn_calls: HashMap<String, (String, String)> = HashMap::new();
114 let mut started = false;
115
116 while let Some(result) = stream.next().await {
117 match result {
118 Ok(event) => {
119 for response in process_event(event, &mut fn_calls, &mut started) {
120 yield response;
121 }
122 }
123 Err(e) => {
124 yield Err(LlmError::ApiError(e.to_string()));
125 break;
126 }
127 }
128 }
129
130 if !started {
131 yield Ok(LlmResponse::done());
132 }
133 })
134 }
135
136 fn display_name(&self) -> String {
137 format!("OpenAI ({})", self.model)
138 }
139
140 fn context_window(&self) -> Option<u32> {
141 get_context_window("openai", &self.model)
142 }
143
144 fn model(&self) -> Option<LlmModel> {
145 format!("openai:{}", self.model).parse().ok()
146 }
147}
148
149fn process_event(
150 event: ResponseStreamEvent,
151 fn_calls: &mut HashMap<String, (String, String)>,
152 started: &mut bool,
153) -> Vec<Result<LlmResponse>> {
154 match event {
155 ResponseStreamEvent::ResponseCreated(e) => {
156 *started = true;
157 vec![Ok(LlmResponse::start(&e.response.id))]
158 }
159 ResponseStreamEvent::ResponseOutputTextDelta(e) if !e.delta.is_empty() => {
160 vec![Ok(LlmResponse::text(&e.delta))]
161 }
162 ResponseStreamEvent::ResponseReasoningSummaryTextDelta(e) if !e.delta.is_empty() => {
163 vec![Ok(LlmResponse::reasoning(&e.delta))]
164 }
165 ResponseStreamEvent::ResponseOutputItemAdded(e) => {
166 if let OutputItem::FunctionCall(fc) = e.item {
167 let item_id = fc.id.clone().unwrap_or_default();
168 fn_calls.insert(item_id, (fc.call_id.clone(), fc.name.clone()));
169 vec![Ok(LlmResponse::tool_request_start(&fc.call_id, &fc.name))]
170 } else {
171 vec![]
172 }
173 }
174 ResponseStreamEvent::ResponseFunctionCallArgumentsDelta(e) => {
175 if let Some((call_id, _)) = fn_calls.get(&e.item_id) {
176 vec![Ok(LlmResponse::tool_request_arg(call_id, &e.delta))]
177 } else {
178 vec![]
179 }
180 }
181 ResponseStreamEvent::ResponseFunctionCallArgumentsDone(e) => {
182 if let Some((call_id, name)) = fn_calls.remove(&e.item_id) {
183 let name = e.name.unwrap_or(name);
184 vec![Ok(LlmResponse::tool_request_complete(&call_id, &name, &e.arguments))]
185 } else {
186 vec![]
187 }
188 }
189 ResponseStreamEvent::ResponseCompleted(e) => {
190 let mut results = Vec::new();
191 if let Some(usage) = e.response.usage {
192 results.push(Ok(LlmResponse::Usage { tokens: usage.into() }));
193 }
194 results.push(Ok(LlmResponse::done_with_stop_reason(StopReason::EndTurn)));
195 results
196 }
197 ResponseStreamEvent::ResponseFailed(e) => {
198 let msg = e.response.error.map_or_else(|| "Unknown error".to_string(), |err| err.message);
199 vec![Err(LlmError::ApiError(msg))]
200 }
201 ResponseStreamEvent::ResponseIncomplete(_) => {
202 vec![Ok(LlmResponse::done_with_stop_reason(StopReason::Length))]
203 }
204 ResponseStreamEvent::ResponseError(e) => {
205 vec![Err(LlmError::ApiError(e.message))]
206 }
207 _ => vec![],
208 }
209}
210
211fn build_response_request(model: &str, context: &Context) -> Result<CreateResponse> {
212 let mut instructions: Option<String> = None;
213 let mut items: Vec<InputItem> = Vec::new();
214
215 for msg in context.messages() {
216 match msg {
217 ChatMessage::System { content, .. } => {
218 instructions = Some(content.clone());
219 }
220 ChatMessage::User { content, .. } => {
221 items.push(InputItem::EasyMessage(EasyInputMessage {
222 r#type: MessageType::Message,
223 role: Role::User,
224 content: map_user_content_for_responses(content)?,
225 phase: None,
226 }));
227 }
228 ChatMessage::Assistant { content, tool_calls, .. } => {
229 if !content.is_empty() {
230 items.push(InputItem::EasyMessage(EasyInputMessage {
231 r#type: MessageType::Message,
232 role: Role::Assistant,
233 content: EasyInputContent::Text(content.clone()),
234 phase: None,
235 }));
236 }
237 for tc in tool_calls {
238 items.push(InputItem::Item(Item::FunctionCall(FunctionToolCall {
239 call_id: tc.id.clone(),
240 name: tc.name.clone(),
241 arguments: tc.arguments.clone(),
242 namespace: None,
243 id: None,
244 status: None,
245 })));
246 }
247 }
248 ChatMessage::ToolCallResult(result) => {
249 let (call_id, output) = match result {
250 Ok(r) => (r.id.clone(), r.result.clone()),
251 Err(e) => (e.id.clone(), e.error.clone()),
252 };
253 items.push(InputItem::Item(Item::FunctionCallOutput(FunctionCallOutputItemParam {
254 call_id,
255 output: FunctionCallOutput::Text(output),
256 id: None,
257 status: None,
258 })));
259 }
260 ChatMessage::Summary { content, .. } => {
261 items.push(InputItem::EasyMessage(EasyInputMessage {
262 r#type: MessageType::Message,
263 role: Role::User,
264 content: EasyInputContent::Text(format!("[Previous conversation handoff]\n\n{content}")),
265 phase: None,
266 }));
267 }
268 ChatMessage::Error { .. } => {}
269 }
270 }
271
272 let tools = map_tools(context.tools())?;
273
274 let reasoning =
275 context.reasoning_effort().map(|_| Reasoning { effort: None, summary: Some(ReasoningSummary::Auto) });
276
277 let settings = context.model_settings();
278
279 Ok(CreateResponse {
280 model: Some(model.to_string()),
281 input: InputParam::Items(items),
282 instructions,
283 tools: if tools.is_empty() { None } else { Some(tools) },
284 reasoning,
285 stream: Some(true),
286 include: Some(vec![IncludeEnum::ReasoningEncryptedContent]),
287 store: Some(false),
288 background: None,
289 conversation: None,
290 max_output_tokens: settings.max_tokens,
291 metadata: None,
292 parallel_tool_calls: None,
293 previous_response_id: None,
294 prompt: None,
295 service_tier: None,
296 stream_options: None,
297 temperature: settings.temperature,
298 text: None,
299 tool_choice: None,
300 top_p: settings.top_p,
301 truncation: None,
302 prompt_cache_key: None,
303 safety_identifier: None,
304 max_tool_calls: None,
305 prompt_cache_retention: None,
306 top_logprobs: None,
307 })
308}
309
310fn map_tools(tools: &[ToolDefinition]) -> Result<Vec<Tool>> {
311 tools
312 .iter()
313 .map(|t| {
314 let parameters: serde_json::Value = serde_json::from_str(&t.parameters)
315 .map_err(|e| LlmError::ToolParameterParsing { tool_name: t.name.clone(), error: e.to_string() })?;
316
317 Ok(Tool::Function(FunctionTool {
318 name: t.name.clone(),
319 description: Some(t.description.clone()),
320 parameters: Some(parameters),
321 strict: Some(false),
322 defer_loading: None,
323 }))
324 })
325 .collect()
326}
327
328fn build_wire_request(model: &str, context: &Context) -> Result<Value> {
332 let mut body = serde_json::to_value(build_response_request(model, context)?)?;
333 if let Some(effort) = context.reasoning_effort() {
334 body["reasoning"]["effort"] = effort.as_str().into();
335 }
336 Ok(body)
337}
338
339#[cfg(test)]
340mod tests {
341 use super::*;
342 use crate::AssistantReasoning;
343 use crate::ReasoningEffort;
344 use crate::ToolCallRequest;
345 use crate::providers::test_capture_server::CaptureServer;
346 use crate::types::IsoString;
347
348 #[test]
349 fn test_build_request_simple_user_message() {
350 let context = Context::new(
351 vec![ChatMessage::User { content: vec![ContentBlock::text("Hello")], timestamp: IsoString::now() }],
352 vec![],
353 );
354
355 let req = build_response_request("gpt-4.1", &context).unwrap();
356 assert_eq!(req.model, Some("gpt-4.1".to_string()));
357 assert!(req.instructions.is_none());
358 assert!(req.tools.is_none());
359 assert!(req.reasoning.is_none());
360
361 let json = serde_json::to_value(&req).unwrap();
362 assert_eq!(json["input"][0]["role"], "user");
363 assert_eq!(json["input"][0]["content"][0]["text"], "Hello");
364 }
365
366 #[test]
367 fn test_build_request_with_system_message() {
368 let context = Context::new(
369 vec![
370 ChatMessage::System { content: "You are helpful.".to_string(), timestamp: IsoString::now() },
371 ChatMessage::User { content: vec![ContentBlock::text("Hi")], timestamp: IsoString::now() },
372 ],
373 vec![],
374 );
375
376 let req = build_response_request("gpt-4.1", &context).unwrap();
377 assert_eq!(req.instructions, Some("You are helpful.".to_string()));
378
379 let json = serde_json::to_value(&req).unwrap();
380 let items = json["input"].as_array().unwrap();
381 assert_eq!(items.len(), 1);
382 assert_eq!(items[0]["role"], "user");
383 }
384
385 #[test]
386 fn test_build_request_with_tool_calls() {
387 let context = Context::new(
388 vec![
389 ChatMessage::User { content: vec![ContentBlock::text("Search for rust")], timestamp: IsoString::now() },
390 ChatMessage::Assistant {
391 content: String::new(),
392 reasoning: AssistantReasoning::default(),
393 timestamp: IsoString::now(),
394 tool_calls: vec![ToolCallRequest {
395 id: "call_1".to_string(),
396 name: "search".to_string(),
397 arguments: r#"{"q":"rust"}"#.to_string(),
398 }],
399 },
400 ChatMessage::ToolCallResult(Ok(crate::ToolCallResult {
401 id: "call_1".to_string(),
402 name: "search".to_string(),
403 arguments: r#"{"q":"rust"}"#.to_string(),
404 result: "Found results".to_string(),
405 })),
406 ],
407 vec![ToolDefinition::new("search", "Search", r#"{"type":"object"}"#)],
408 );
409
410 let req = build_response_request("gpt-4.1", &context).unwrap();
411 let json = serde_json::to_value(&req).unwrap();
412
413 let items = json["input"].as_array().unwrap();
414 assert_eq!(items[0]["role"], "user");
415 assert_eq!(items[1]["type"], "function_call");
416 assert_eq!(items[1]["call_id"], "call_1");
417 assert_eq!(items[2]["type"], "function_call_output");
418 assert_eq!(items[2]["call_id"], "call_1");
419 assert_eq!(items[2]["output"], "Found results");
420
421 assert!(req.tools.is_some());
422 let tools_json = serde_json::to_value(&req.tools).unwrap();
423 assert_eq!(tools_json[0]["type"], "function");
424 assert_eq!(tools_json[0]["name"], "search");
425 }
426
427 #[test]
428 fn test_build_wire_request_maps_every_reasoning_effort() {
429 for effort in ReasoningEffort::all() {
430 let mut context = Context::new(
431 vec![ChatMessage::User { content: vec![ContentBlock::text("Think")], timestamp: IsoString::now() }],
432 vec![],
433 );
434 context.set_reasoning_effort(Some(*effort));
435
436 let body = build_wire_request("gpt-5.6", &context).unwrap();
437 assert_eq!(body["reasoning"]["effort"], effort.as_str());
438 assert_eq!(body["reasoning"]["summary"], "auto");
439 }
440 }
441
442 #[test]
443 fn test_build_wire_request_omits_reasoning_without_effort() {
444 let context = Context::new(
445 vec![ChatMessage::User { content: vec![ContentBlock::text("Hi")], timestamp: IsoString::now() }],
446 vec![],
447 );
448
449 let body = build_wire_request("gpt-4.1", &context).unwrap();
450 assert!(body["reasoning"].is_null());
451 }
452
453 #[tokio::test]
454 async fn stream_response_sends_max_effort_on_the_wire() {
455 let mut server = CaptureServer::start().await;
456 let connection = ProviderConnectionConfig {
457 base_url: Some(server.base_url.clone()),
458 auth_mode: ProviderAuthMode::None,
459 ..Default::default()
460 };
461 let provider = OpenAiProvider::from_env_with_connection(connection).await.unwrap().with_model("gpt-5.6");
462 let mut context = Context::new(
463 vec![ChatMessage::User { content: vec![ContentBlock::text("Think harder")], timestamp: IsoString::now() }],
464 vec![],
465 );
466 context.set_reasoning_effort(Some(ReasoningEffort::Max));
467
468 let responses = provider.stream_response(&context).collect::<Vec<_>>().await;
469 let captured = server.captured().await;
470
471 assert!(responses.iter().all(Result::is_ok), "{responses:?}");
472 assert_eq!(captured.body["reasoning"]["effort"], "max");
473 assert_eq!(captured.body["model"], "gpt-5.6");
474 assert_eq!(captured.body["stream"], true);
475 }
476
477 #[test]
478 fn test_build_request_applies_model_settings() {
479 let mut context = Context::new(
480 vec![ChatMessage::User { content: vec![ContentBlock::text("Hello")], timestamp: IsoString::now() }],
481 vec![],
482 );
483 context.set_model_settings(crate::ModelSettings {
484 temperature: Some(0.0),
485 top_p: Some(0.5),
486 max_tokens: Some(128),
487 });
488
489 let req = build_response_request("gpt-4.1", &context).unwrap();
490 assert_eq!(req.temperature, Some(0.0));
491 assert_eq!(req.top_p, Some(0.5));
492 assert_eq!(req.max_output_tokens, Some(128));
493 }
494
495 #[test]
496 fn test_build_request_with_audio_returns_unsupported_content() {
497 let context = Context::new(
498 vec![ChatMessage::User {
499 content: vec![ContentBlock::Audio { data: "YXVkaW8=".to_string(), mime_type: "audio/wav".to_string() }],
500 timestamp: IsoString::now(),
501 }],
502 vec![],
503 );
504
505 assert!(matches!(build_response_request("gpt-4.1", &context), Err(LlmError::UnsupportedContent(_))));
506 }
507
508 #[test]
509 fn test_map_tools_valid() {
510 let tools = vec![ToolDefinition::new(
511 "read_file",
512 "Read a file",
513 r#"{"type":"object","properties":{"path":{"type":"string"}}}"#,
514 )];
515
516 let result = map_tools(&tools).unwrap();
517 assert_eq!(result.len(), 1);
518
519 let json = serde_json::to_value(&result[0]).unwrap();
520 assert_eq!(json["type"], "function");
521 assert_eq!(json["name"], "read_file");
522 }
523
524 #[test]
525 fn test_map_tools_invalid_json() {
526 let tools = vec![ToolDefinition::new("broken", "Broken", "not json{")];
527
528 let result = map_tools(&tools);
529 assert!(result.is_err());
530 match result.unwrap_err() {
531 LlmError::ToolParameterParsing { tool_name, .. } => {
532 assert_eq!(tool_name, "broken");
533 }
534 other => panic!("Expected ToolParameterParsing, got: {other}"),
535 }
536 }
537
538 #[test]
539 fn test_provider_display_name() {
540 let config = AetherOpenAiConfig::new(OpenAIConfig::new().with_api_key("test"), ProviderAuthMode::Default);
541 let provider = OpenAiProvider { client: Client::with_config(config), model: "gpt-4.1".to_string() };
542 assert_eq!(provider.display_name(), "OpenAI (gpt-4.1)");
543 }
544}