openai_api_dispatch/executor/openai/
mod.rs

1use std::{env, iter};
2
3use alloc::{string::String, sync::Arc};
4use async_openai::{
5    Client,
6    config::OpenAIConfig,
7    types::chat::{
8        ChatChoice, ChatCompletionRequestAssistantMessage,
9        ChatCompletionRequestAssistantMessageContent, ChatCompletionRequestMessage,
10        ChatCompletionRequestUserMessage, ChatCompletionRequestUserMessageContent,
11        ChatCompletionResponseMessage, CreateChatCompletionRequestArgs, ResponseFormat,
12        ResponseFormatJsonSchema,
13    },
14};
15
16use crate::{
17    executor::Executor,
18    task::{Interaction, Response, Task, TaskChat, TaskType},
19    utils,
20};
21
22#[cfg(test)]
23mod tests;
24
25#[derive(Debug, Clone)]
26/// Executes non-streaming Chat Completions and returns the first choice's text.
27///
28/// Task models override the configured default. Missing models, malformed chat
29/// data, and request/API failures return `Err`; missing choices or text produce
30/// an unsuccessful [`Response`]. Reported usage is total tokens, or zero if absent.
31///
32/// Sends the optional system instruction, prior user/assistant turns, and the
33/// current prompt as a user message, in that order.
34///
35/// # Current limitations
36///
37/// - System entries in input history are discarded; use [`TaskChat::system`]
38///   for a persistent system instruction.
39/// - Returned history currently duplicates the latest user prompt: it is kept
40///   from the request messages and appended again before the assistant reply.
41/// - Schemas request strict JSON output named `response`; returned text is not
42///   locally parsed or validated against the schema.
43/// - `max_tokens` is cast from `u64` to `u32` without range checking. Payload
44///   metadata is preserved in the task but is not sent to the model.
45pub struct ExecutorAsyncOpenai {
46    client: Client<OpenAIConfig>,
47    default_model: Arc<Option<String>>,
48}
49
50impl ExecutorAsyncOpenai {
51    /// Creates a client without contacting the endpoint.
52    ///
53    /// Reads `OPENAI_API_URL` (default `http://127.0.0.1:8000/v1`) and snapshots
54    /// `OPENAI_API_DEFAULT_MODEL`. Authentication uses `async-openai`'s default
55    /// configuration. Endpoint/model validity is checked only when executing.
56    pub fn from_env_or_default() -> Self {
57        let default_model = utils::get_default_model();
58        let url = env::var("OPENAI_API_URL").unwrap_or_else(|_| "http://127.0.0.1:8000/v1".into());
59        let config = OpenAIConfig::new().with_api_base(&url);
60        let client = Client::with_config(config);
61
62        tracing::info!("openai api connected to `{url}`");
63
64        Self {
65            client,
66            default_model,
67        }
68    }
69
70    async fn execute_chat(
71        &self,
72        model: String,
73        task: Task,
74        chat: TaskChat,
75    ) -> anyhow::Result<Response> {
76        let TaskChat {
77            system,
78            history,
79            schema,
80            prompt,
81        } = chat;
82
83        let history = history
84            .as_deref()
85            .unwrap_or(&[])
86            .iter()
87            .filter_map(|h| match h {
88                Interaction::System(_) => None,
89                Interaction::Assistant(m) => {
90                    Some(ChatCompletionRequestMessage::Assistant(m.as_str().into()))
91                }
92                Interaction::User(m) => Some(ChatCompletionRequestMessage::User(m.as_str().into())),
93            })
94            .map(Some);
95
96        let history: Vec<_> = iter::once(
97            system
98                .as_ref()
99                .map(|s| ChatCompletionRequestMessage::System(s.as_str().into())),
100        )
101        .chain(history)
102        .chain(iter::once(Some(ChatCompletionRequestMessage::User(
103            prompt.as_str().into(),
104        ))))
105        .flatten()
106        .collect();
107
108        let mut request = CreateChatCompletionRequestArgs::default();
109
110        request.model(&model);
111        request.messages(history.clone());
112
113        if let Some(t) = task.max_tokens {
114            request.max_tokens(t as u32);
115        }
116
117        if let Some(schema) = schema.as_ref().cloned() {
118            request.response_format(ResponseFormat::JsonSchema {
119                json_schema: ResponseFormatJsonSchema {
120                    name: "response".to_string(),
121                    description: None,
122                    schema,
123                    strict: Some(true),
124                },
125            });
126        }
127
128        let request = request.build()?;
129
130        tracing::debug!("task `{}` submitting", task.id);
131
132        let response = self.client.chat().create(request).await?;
133        let tokens = response
134            .usage
135            .as_ref()
136            .map(|u| u.total_tokens as u64)
137            .unwrap_or(0);
138
139        tracing::debug!("task `{}` returned; consumed `{tokens}` tokens", task.id);
140
141        let response = match response.choices.first() {
142            Some(r) => r,
143            None => {
144                tracing::debug!("task `{}` error; no response provided", task.id);
145                return Ok(Response::error(task, tokens, model, "no response provided"));
146            }
147        };
148
149        match &response {
150            ChatChoice {
151                message:
152                    ChatCompletionResponseMessage {
153                        content: None,
154                        refusal: r,
155                        ..
156                    },
157                ..
158            } => {
159                let r = r
160                    .as_ref()
161                    .map(|r| r.as_str())
162                    .unwrap_or("no reason provided");
163                let m = format!("response refused: `{r}`");
164                tracing::debug!("task `{}` error; `{m}`", task.id,);
165
166                Ok(Response::error(task, tokens, model, m))
167            }
168
169            ChatChoice {
170                message:
171                    ChatCompletionResponseMessage {
172                        content: Some(m), ..
173                    },
174                ..
175            } => {
176                let history: Vec<_> =
177                    history
178                        .into_iter()
179                        .filter_map(|h| match h {
180                            ChatCompletionRequestMessage::User(
181                                ChatCompletionRequestUserMessage {
182                                    content: ChatCompletionRequestUserMessageContent::Text(m),
183                                    ..
184                                },
185                            ) => Some(Interaction::User(m)),
186                            ChatCompletionRequestMessage::Assistant(
187                                ChatCompletionRequestAssistantMessage {
188                                    content:
189                                        Some(ChatCompletionRequestAssistantMessageContent::Text(m)),
190                                    ..
191                                },
192                            ) => Some(Interaction::Assistant(m)),
193                            _ => {
194                                tracing::debug!("skipping model reply {h:?}");
195                                None
196                            }
197                        })
198                        .chain(iter::once(Interaction::Assistant(m.clone())))
199                        .collect();
200
201                tracing::debug!("task `{}` served", task.id);
202
203                Ok(Response::success(task, tokens, model, m).with_history(history))
204            }
205        }
206    }
207}
208
209impl Executor for ExecutorAsyncOpenai {
210    fn default_model(&self) -> &Arc<Option<String>> {
211        &self.default_model
212    }
213
214    async fn execute(&self, task: Task) -> anyhow::Result<Response> {
215        tracing::debug!("received task id `{}`", task.id);
216
217        let model = task.model(self.default_model())?;
218
219        tracing::debug!("task `{}` using model `{model}`", task.id);
220
221        match &task.task_type {
222            TaskType::Chat => {
223                let chat = task.try_to_chat()?;
224
225                self.execute_chat(model, task, chat).await
226            }
227        }
228    }
229}
openai_api_dispatch/executor/openai/mod.rs

openai_api_dispatch/executor/openai/
mod.rs
