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(|tool| {
314 Ok(Tool::Function(FunctionTool {
315 name: tool.name.clone(),
316 description: Some(tool.description.clone()),
317 parameters: Some(tool.parameters.clone()),
318 strict: Some(false),
319 defer_loading: None,
320 }))
321 })
322 .collect()
323}
324
325fn build_wire_request(model: &str, context: &Context) -> Result<Value> {
329 let mut body = serde_json::to_value(build_response_request(model, context)?)?;
330 if let Some(effort) = context.reasoning_effort() {
331 body["reasoning"]["effort"] = effort.as_str().into();
332 }
333 Ok(body)
334}
335
336#[cfg(test)]
337mod tests {
338 use super::*;
339 use crate::AssistantReasoning;
340 use crate::ReasoningEffort;
341 use crate::ToolCallRequest;
342 use crate::providers::test_capture_server::CaptureServer;
343 use crate::types::IsoString;
344
345 #[test]
346 fn test_build_request_simple_user_message() {
347 let context = Context::new(
348 vec![ChatMessage::User { content: vec![ContentBlock::text("Hello")], timestamp: IsoString::now() }],
349 vec![],
350 );
351
352 let req = build_response_request("gpt-4.1", &context).unwrap();
353 assert_eq!(req.model, Some("gpt-4.1".to_string()));
354 assert!(req.instructions.is_none());
355 assert!(req.tools.is_none());
356 assert!(req.reasoning.is_none());
357
358 let json = serde_json::to_value(&req).unwrap();
359 assert_eq!(json["input"][0]["role"], "user");
360 assert_eq!(json["input"][0]["content"][0]["text"], "Hello");
361 }
362
363 #[test]
364 fn test_build_request_with_system_message() {
365 let context = Context::new(
366 vec![
367 ChatMessage::System { content: "You are helpful.".to_string(), timestamp: IsoString::now() },
368 ChatMessage::User { content: vec![ContentBlock::text("Hi")], timestamp: IsoString::now() },
369 ],
370 vec![],
371 );
372
373 let req = build_response_request("gpt-4.1", &context).unwrap();
374 assert_eq!(req.instructions, Some("You are helpful.".to_string()));
375
376 let json = serde_json::to_value(&req).unwrap();
377 let items = json["input"].as_array().unwrap();
378 assert_eq!(items.len(), 1);
379 assert_eq!(items[0]["role"], "user");
380 }
381
382 #[test]
383 fn test_build_request_with_tool_calls() {
384 let context = Context::new(
385 vec![
386 ChatMessage::User { content: vec![ContentBlock::text("Search for rust")], timestamp: IsoString::now() },
387 ChatMessage::Assistant {
388 content: String::new(),
389 reasoning: AssistantReasoning::default(),
390 timestamp: IsoString::now(),
391 tool_calls: vec![ToolCallRequest {
392 id: "call_1".to_string(),
393 name: "search".to_string(),
394 arguments: r#"{"q":"rust"}"#.to_string(),
395 }],
396 },
397 ChatMessage::ToolCallResult(Ok(crate::ToolCallResult {
398 id: "call_1".to_string(),
399 name: "search".to_string(),
400 arguments: r#"{"q":"rust"}"#.to_string(),
401 result: "Found results".to_string(),
402 })),
403 ],
404 vec![ToolDefinition::new("search", "Search", serde_json::json!({ "type": "object" }))],
405 );
406
407 let req = build_response_request("gpt-4.1", &context).unwrap();
408 let json = serde_json::to_value(&req).unwrap();
409
410 let items = json["input"].as_array().unwrap();
411 assert_eq!(items[0]["role"], "user");
412 assert_eq!(items[1]["type"], "function_call");
413 assert_eq!(items[1]["call_id"], "call_1");
414 assert_eq!(items[2]["type"], "function_call_output");
415 assert_eq!(items[2]["call_id"], "call_1");
416 assert_eq!(items[2]["output"], "Found results");
417
418 assert!(req.tools.is_some());
419 let tools_json = serde_json::to_value(&req.tools).unwrap();
420 assert_eq!(tools_json[0]["type"], "function");
421 assert_eq!(tools_json[0]["name"], "search");
422 }
423
424 #[test]
425 fn test_build_wire_request_maps_every_reasoning_effort() {
426 for effort in ReasoningEffort::all() {
427 let mut context = Context::new(
428 vec![ChatMessage::User { content: vec![ContentBlock::text("Think")], timestamp: IsoString::now() }],
429 vec![],
430 );
431 context.set_reasoning_effort(Some(*effort));
432
433 let body = build_wire_request("gpt-5.6", &context).unwrap();
434 assert_eq!(body["reasoning"]["effort"], effort.as_str());
435 assert_eq!(body["reasoning"]["summary"], "auto");
436 }
437 }
438
439 #[test]
440 fn test_build_wire_request_omits_reasoning_without_effort() {
441 let context = Context::new(
442 vec![ChatMessage::User { content: vec![ContentBlock::text("Hi")], timestamp: IsoString::now() }],
443 vec![],
444 );
445
446 let body = build_wire_request("gpt-4.1", &context).unwrap();
447 assert!(body["reasoning"].is_null());
448 }
449
450 #[tokio::test]
451 async fn stream_response_sends_max_effort_on_the_wire() {
452 let mut server = CaptureServer::start().await;
453 let connection = ProviderConnectionConfig {
454 base_url: Some(server.base_url.clone()),
455 auth_mode: ProviderAuthMode::None,
456 ..Default::default()
457 };
458 let provider = OpenAiProvider::from_env_with_connection(connection).await.unwrap().with_model("gpt-5.6");
459 let mut context = Context::new(
460 vec![ChatMessage::User { content: vec![ContentBlock::text("Think harder")], timestamp: IsoString::now() }],
461 vec![],
462 );
463 context.set_reasoning_effort(Some(ReasoningEffort::Max));
464
465 let responses = provider.stream_response(&context).collect::<Vec<_>>().await;
466 let captured = server.captured().await;
467
468 assert!(responses.iter().all(Result::is_ok), "{responses:?}");
469 assert_eq!(captured.body["reasoning"]["effort"], "max");
470 assert_eq!(captured.body["model"], "gpt-5.6");
471 assert_eq!(captured.body["stream"], true);
472 }
473
474 #[test]
475 fn test_build_request_applies_model_settings() {
476 let mut context = Context::new(
477 vec![ChatMessage::User { content: vec![ContentBlock::text("Hello")], timestamp: IsoString::now() }],
478 vec![],
479 );
480 context.set_model_settings(crate::ModelSettings {
481 temperature: Some(0.0),
482 top_p: Some(0.5),
483 max_tokens: Some(128),
484 });
485
486 let req = build_response_request("gpt-4.1", &context).unwrap();
487 assert_eq!(req.temperature, Some(0.0));
488 assert_eq!(req.top_p, Some(0.5));
489 assert_eq!(req.max_output_tokens, Some(128));
490 }
491
492 #[test]
493 fn test_build_request_with_audio_returns_unsupported_content() {
494 let context = Context::new(
495 vec![ChatMessage::User {
496 content: vec![ContentBlock::Audio { data: "YXVkaW8=".to_string(), mime_type: "audio/wav".to_string() }],
497 timestamp: IsoString::now(),
498 }],
499 vec![],
500 );
501
502 assert!(matches!(build_response_request("gpt-4.1", &context), Err(LlmError::UnsupportedContent(_))));
503 }
504
505 #[test]
506 fn test_map_tools_valid() {
507 let tools = vec![ToolDefinition::new(
508 "read_file",
509 "Read a file",
510 serde_json::from_str(r#"{"type":"object","properties":{"path":{"type":"string"}}}"#).unwrap(),
511 )];
512
513 let result = map_tools(&tools).unwrap();
514 assert_eq!(result.len(), 1);
515
516 let json = serde_json::to_value(&result[0]).unwrap();
517 assert_eq!(json["type"], "function");
518 assert_eq!(json["name"], "read_file");
519 }
520
521 #[test]
522 fn test_provider_display_name() {
523 let config = AetherOpenAiConfig::new(OpenAIConfig::new().with_api_key("test"), ProviderAuthMode::Default);
524 let provider = OpenAiProvider { client: Client::with_config(config), model: "gpt-4.1".to_string() };
525 assert_eq!(provider.display_name(), "OpenAI (gpt-4.1)");
526 }
527}