async_openai_compat/
client.rs

1use std::pin::Pin;
2
3use bytes::Bytes;
4use futures::{stream::StreamExt, Stream};
5use reqwest::multipart::Form;
6use reqwest_eventsource::{Error, Event, EventSource, RequestBuilderExt};
7use serde::{de::DeserializeOwned, Serialize};
8
9use crate::{
10    config::{Config, OpenAIConfig},
11    error::{map_deserialization_error, ApiError, OpenAIError, WrappedError},
12    file::Files,
13    image::Images,
14    moderation::Moderations,
15    traits::AsyncTryFrom,
16    Assistants, Audio, AuditLogs, Batches, Chat, Completions, Embeddings, FineTuning, Invites,
17    Models, Projects, Responses, Threads, Uploads, Users, VectorStores,
18};
19
20#[derive(Debug, Clone, Default)]
21/// Client is a container for config, backoff and http_client
22/// used to make API calls.
23pub struct Client<C: Config> {
24    http_client: reqwest::Client,
25    config: C,
26    backoff: backoff::ExponentialBackoff,
27}
28
29impl Client<OpenAIConfig> {
30    /// Client with default [OpenAIConfig]
31    pub fn new() -> Self {
32        Self::default()
33    }
34}
35
36impl<C: Config> Client<C> {
37    /// Create client with a custom HTTP client, OpenAI config, and backoff.
38    pub fn build(
39        http_client: reqwest::Client,
40        config: C,
41        backoff: backoff::ExponentialBackoff,
42    ) -> Self {
43        Self {
44            http_client,
45            config,
46            backoff,
47        }
48    }
49
50    /// Create client with [OpenAIConfig] or [crate::config::AzureConfig]
51    pub fn with_config(config: C) -> Self {
52        Self {
53            http_client: reqwest::Client::new(),
54            config,
55            backoff: Default::default(),
56        }
57    }
58
59    /// Provide your own [client] to make HTTP requests with.
60    ///
61    /// [client]: reqwest::Client
62    pub fn with_http_client(mut self, http_client: reqwest::Client) -> Self {
63        self.http_client = http_client;
64        self
65    }
66
67    /// Exponential backoff for retrying [rate limited](https://platform.openai.com/docs/guides/rate-limits) requests.
68    pub fn with_backoff(mut self, backoff: backoff::ExponentialBackoff) -> Self {
69        self.backoff = backoff;
70        self
71    }
72
73    // API groups
74
75    /// To call [Models] group related APIs using this client.
76    pub fn models(&self) -> Models<C> {
77        Models::new(self)
78    }
79
80    /// To call [Completions] group related APIs using this client.
81    pub fn completions(&self) -> Completions<C> {
82        Completions::new(self)
83    }
84
85    /// To call [Chat] group related APIs using this client.
86    pub fn chat(&self) -> Chat<C> {
87        Chat::new(self)
88    }
89
90    /// To call [Images] group related APIs using this client.
91    pub fn images(&self) -> Images<C> {
92        Images::new(self)
93    }
94
95    /// To call [Moderations] group related APIs using this client.
96    pub fn moderations(&self) -> Moderations<C> {
97        Moderations::new(self)
98    }
99
100    /// To call [Files] group related APIs using this client.
101    pub fn files(&self) -> Files<C> {
102        Files::new(self)
103    }
104
105    /// To call [Uploads] group related APIs using this client.
106    pub fn uploads(&self) -> Uploads<C> {
107        Uploads::new(self)
108    }
109
110    /// To call [FineTuning] group related APIs using this client.
111    pub fn fine_tuning(&self) -> FineTuning<C> {
112        FineTuning::new(self)
113    }
114
115    /// To call [Embeddings] group related APIs using this client.
116    pub fn embeddings(&self) -> Embeddings<C> {
117        Embeddings::new(self)
118    }
119
120    /// To call [Audio] group related APIs using this client.
121    pub fn audio(&self) -> Audio<C> {
122        Audio::new(self)
123    }
124
125    /// To call [Assistants] group related APIs using this client.
126    pub fn assistants(&self) -> Assistants<C> {
127        Assistants::new(self)
128    }
129
130    /// To call [Threads] group related APIs using this client.
131    pub fn threads(&self) -> Threads<C> {
132        Threads::new(self)
133    }
134
135    /// To call [VectorStores] group related APIs using this client.
136    pub fn vector_stores(&self) -> VectorStores<C> {
137        VectorStores::new(self)
138    }
139
140    /// To call [Batches] group related APIs using this client.
141    pub fn batches(&self) -> Batches<C> {
142        Batches::new(self)
143    }
144
145    /// To call [AuditLogs] group related APIs using this client.
146    pub fn audit_logs(&self) -> AuditLogs<C> {
147        AuditLogs::new(self)
148    }
149
150    /// To call [Invites] group related APIs using this client.
151    pub fn invites(&self) -> Invites<C> {
152        Invites::new(self)
153    }
154
155    /// To call [Users] group related APIs using this client.
156    pub fn users(&self) -> Users<C> {
157        Users::new(self)
158    }
159
160    /// To call [Projects] group related APIs using this client.
161    pub fn projects(&self) -> Projects<C> {
162        Projects::new(self)
163    }
164
165    /// To call [Responses] group related APIs using this client.
166    pub fn responses(&self) -> Responses<C> {
167        Responses::new(self)
168    }
169
170    pub fn config(&self) -> &C {
171        &self.config
172    }
173
174    /// Make a GET request to {path} and deserialize the response body
175    pub(crate) async fn get<O>(&self, path: &str) -> Result<O, OpenAIError>
176    where
177        O: DeserializeOwned,
178    {
179        let request_maker = || async {
180            Ok(self
181                .http_client
182                .get(self.config.url(path))
183                .query(&self.config.query())
184                .headers(self.config.headers())
185                .build()?)
186        };
187
188        self.execute(request_maker).await
189    }
190
191    /// Make a GET request to {path} with given Query and deserialize the response body
192    pub(crate) async fn get_with_query<Q, O>(&self, path: &str, query: &Q) -> Result<O, OpenAIError>
193    where
194        O: DeserializeOwned,
195        Q: Serialize + ?Sized,
196    {
197        let request_maker = || async {
198            Ok(self
199                .http_client
200                .get(self.config.url(path))
201                .query(&self.config.query())
202                .query(query)
203                .headers(self.config.headers())
204                .build()?)
205        };
206
207        self.execute(request_maker).await
208    }
209
210    /// Make a DELETE request to {path} and deserialize the response body
211    pub(crate) async fn delete<O>(&self, path: &str) -> Result<O, OpenAIError>
212    where
213        O: DeserializeOwned,
214    {
215        let request_maker = || async {
216            Ok(self
217                .http_client
218                .delete(self.config.url(path))
219                .query(&self.config.query())
220                .headers(self.config.headers())
221                .build()?)
222        };
223
224        self.execute(request_maker).await
225    }
226
227    /// Make a GET request to {path} and return the response body
228    pub(crate) async fn get_raw(&self, path: &str) -> Result<Bytes, OpenAIError> {
229        let request_maker = || async {
230            Ok(self
231                .http_client
232                .get(self.config.url(path))
233                .query(&self.config.query())
234                .headers(self.config.headers())
235                .build()?)
236        };
237
238        self.execute_raw(request_maker).await
239    }
240
241    /// Make a POST request to {path} and return the response body
242    pub(crate) async fn post_raw<I>(&self, path: &str, request: I) -> Result<Bytes, OpenAIError>
243    where
244        I: Serialize,
245    {
246        let request_maker = || async {
247            Ok(self
248                .http_client
249                .post(self.config.url(path))
250                .query(&self.config.query())
251                .headers(self.config.headers())
252                .json(&request)
253                .build()?)
254        };
255
256        self.execute_raw(request_maker).await
257    }
258
259    /// Make a POST request to {path} and deserialize the response body
260    pub(crate) async fn post<I, O>(&self, path: &str, request: I) -> Result<O, OpenAIError>
261    where
262        I: Serialize,
263        O: DeserializeOwned,
264    {
265        let request_maker = || async {
266            Ok(self
267                .http_client
268                .post(self.config.url(path))
269                .query(&self.config.query())
270                .headers(self.config.headers())
271                .json(&request)
272                .build()?)
273        };
274
275        self.execute(request_maker).await
276    }
277
278    /// POST a form at {path} and return the response body
279    pub(crate) async fn post_form_raw<F>(&self, path: &str, form: F) -> Result<Bytes, OpenAIError>
280    where
281        Form: AsyncTryFrom<F, Error = OpenAIError>,
282        F: Clone,
283    {
284        let request_maker = || async {
285            Ok(self
286                .http_client
287                .post(self.config.url(path))
288                .query(&self.config.query())
289                .headers(self.config.headers())
290                .multipart(<Form as AsyncTryFrom<F>>::try_from(form.clone()).await?)
291                .build()?)
292        };
293
294        self.execute_raw(request_maker).await
295    }
296
297    /// POST a form at {path} and deserialize the response body
298    pub(crate) async fn post_form<O, F>(&self, path: &str, form: F) -> Result<O, OpenAIError>
299    where
300        O: DeserializeOwned,
301        Form: AsyncTryFrom<F, Error = OpenAIError>,
302        F: Clone,
303    {
304        let request_maker = || async {
305            Ok(self
306                .http_client
307                .post(self.config.url(path))
308                .query(&self.config.query())
309                .headers(self.config.headers())
310                .multipart(<Form as AsyncTryFrom<F>>::try_from(form.clone()).await?)
311                .build()?)
312        };
313
314        self.execute(request_maker).await
315    }
316
317    /// Execute a HTTP request and retry on rate limit
318    ///
319    /// request_maker serves one purpose: to be able to create request again
320    /// to retry API call after getting rate limited. request_maker is async because
321    /// reqwest::multipart::Form is created by async calls to read files for uploads.
322    async fn execute_raw<M, Fut>(&self, request_maker: M) -> Result<Bytes, OpenAIError>
323    where
324        M: Fn() -> Fut,
325        Fut: core::future::Future<Output = Result<reqwest::Request, OpenAIError>>,
326    {
327        let client = self.http_client.clone();
328
329        backoff::future::retry(self.backoff.clone(), || async {
330            let request = request_maker().await.map_err(backoff::Error::Permanent)?;
331            let response = client
332                .execute(request)
333                .await
334                .map_err(OpenAIError::Reqwest)
335                .map_err(backoff::Error::Permanent)?;
336
337            let status = response.status();
338            let bytes = response
339                .bytes()
340                .await
341                .map_err(OpenAIError::Reqwest)
342                .map_err(backoff::Error::Permanent)?;
343
344            if status.is_server_error() {
345                // OpenAI does not guarantee server errors are returned as JSON so we cannot deserialize them.
346                let message: String = String::from_utf8_lossy(&bytes).into_owned();
347                tracing::warn!("Server error: {status} - {message}");
348                return Err(backoff::Error::Transient {
349                    err: OpenAIError::ApiError(ApiError {
350                        message,
351                        r#type: None,
352                        param: None,
353                        code: None,
354                    }),
355                    retry_after: None,
356                });
357            }
358
359            // Deserialize response body from either error object or actual response object
360            if !status.is_success() {
361                let wrapped_error: WrappedError = serde_json::from_slice(bytes.as_ref())
362                    .map_err(|e| map_deserialization_error(e, bytes.as_ref()))
363                    .map_err(backoff::Error::Permanent)?;
364
365                if status.as_u16() == 429
366                    // API returns 429 also when:
367                    // "You exceeded your current quota, please check your plan and billing details."
368                    && wrapped_error.error.r#type != Some("insufficient_quota".to_string())
369                {
370                    // Rate limited retry...
371                    tracing::warn!("Rate limited: {}", wrapped_error.error.message);
372                    return Err(backoff::Error::Transient {
373                        err: OpenAIError::ApiError(wrapped_error.error),
374                        retry_after: None,
375                    });
376                } else {
377                    return Err(backoff::Error::Permanent(OpenAIError::ApiError(
378                        wrapped_error.error,
379                    )));
380                }
381            }
382
383            Ok(bytes)
384        })
385        .await
386    }
387
388    /// Execute a HTTP request and retry on rate limit
389    ///
390    /// request_maker serves one purpose: to be able to create request again
391    /// to retry API call after getting rate limited. request_maker is async because
392    /// reqwest::multipart::Form is created by async calls to read files for uploads.
393    async fn execute<O, M, Fut>(&self, request_maker: M) -> Result<O, OpenAIError>
394    where
395        O: DeserializeOwned,
396        M: Fn() -> Fut,
397        Fut: core::future::Future<Output = Result<reqwest::Request, OpenAIError>>,
398    {
399        let bytes = self.execute_raw(request_maker).await?;
400
401        let response: O = serde_json::from_slice(bytes.as_ref())
402            .map_err(|e| map_deserialization_error(e, bytes.as_ref()))?;
403
404        Ok(response)
405    }
406
407    /// Make HTTP POST request to receive SSE
408    pub(crate) async fn post_stream<I, O>(
409        &self,
410        path: &str,
411        request: I,
412    ) -> Pin<Box<dyn Stream<Item = Result<O, OpenAIError>> + Send>>
413    where
414        I: Serialize,
415        O: DeserializeOwned + std::marker::Send + 'static,
416    {
417        let event_source = self
418            .http_client
419            .post(self.config.url(path))
420            .query(&self.config.query())
421            .headers(self.config.headers())
422            .json(&request)
423            .eventsource()
424            .unwrap();
425
426        stream(event_source).await
427    }
428
429    pub(crate) async fn post_stream_mapped_raw_events<I, O>(
430        &self,
431        path: &str,
432        request: I,
433        event_mapper: impl Fn(eventsource_stream::Event) -> Result<O, OpenAIError> + Send + 'static,
434    ) -> Pin<Box<dyn Stream<Item = Result<O, OpenAIError>> + Send>>
435    where
436        I: Serialize,
437        O: DeserializeOwned + std::marker::Send + 'static,
438    {
439        let event_source = self
440            .http_client
441            .post(self.config.url(path))
442            .query(&self.config.query())
443            .headers(self.config.headers())
444            .json(&request)
445            .eventsource()
446            .unwrap();
447
448        stream_mapped_raw_events(event_source, event_mapper).await
449    }
450
451    /// Make HTTP GET request to receive SSE
452    pub(crate) async fn _get_stream<Q, O>(
453        &self,
454        path: &str,
455        query: &Q,
456    ) -> Pin<Box<dyn Stream<Item = Result<O, OpenAIError>> + Send>>
457    where
458        Q: Serialize + ?Sized,
459        O: DeserializeOwned + std::marker::Send + 'static,
460    {
461        let event_source = self
462            .http_client
463            .get(self.config.url(path))
464            .query(query)
465            .query(&self.config.query())
466            .headers(self.config.headers())
467            .eventsource()
468            .unwrap();
469
470        stream(event_source).await
471    }
472}
473
474async fn handle_eventsource_error(e: Error) -> Result<(), OpenAIError> {
475    let error_text = e.to_string();
476    if let Error::InvalidStatusCode(_code, response) = e {
477        if let Ok(text) = response.text().await {
478            let api_error = serde_json::from_str::<WrappedError>(&text);
479            if let Ok(e) = api_error {
480                return Err(OpenAIError::ApiError(e.error));
481            }
482        }
483    }
484
485    Err(OpenAIError::StreamError(error_text))
486}
487
488/// Request which responds with SSE.
489/// [server-sent events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format)
490pub(crate) async fn stream<O>(
491    mut event_source: EventSource,
492) -> Pin<Box<dyn Stream<Item = Result<O, OpenAIError>> + Send>>
493where
494    O: DeserializeOwned + std::marker::Send + 'static,
495{
496    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
497
498    tokio::spawn(async move {
499        while let Some(ev) = event_source.next().await {
500            match ev {
501                Err(e) => {
502                    if let Err(e) = handle_eventsource_error(e).await {
503                        if let Err(_e) = tx.send(Err(e)) {
504                            // rx dropped
505                            break;
506                        }
507                    }
508                }
509                Ok(event) => match event {
510                    Event::Message(message) => {
511                        if message.data == "[DONE]" {
512                            break;
513                        }
514
515                        let response = match serde_json::from_str::<O>(&message.data) {
516                            Err(e) => Err(map_deserialization_error(e, message.data.as_bytes())),
517                            Ok(output) => Ok(output),
518                        };
519
520                        if let Err(_e) = tx.send(response) {
521                            // rx dropped
522                            break;
523                        }
524                    }
525                    Event::Open => continue,
526                },
527            }
528        }
529
530        event_source.close();
531    });
532
533    Box::pin(tokio_stream::wrappers::UnboundedReceiverStream::new(rx))
534}
535
536pub(crate) async fn stream_mapped_raw_events<O>(
537    mut event_source: EventSource,
538    event_mapper: impl Fn(eventsource_stream::Event) -> Result<O, OpenAIError> + Send + 'static,
539) -> Pin<Box<dyn Stream<Item = Result<O, OpenAIError>> + Send>>
540where
541    O: DeserializeOwned + std::marker::Send + 'static,
542{
543    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
544
545    tokio::spawn(async move {
546        while let Some(ev) = event_source.next().await {
547            match ev {
548                Err(e) => {
549                    if let Err(_e) = tx.send(Err(OpenAIError::StreamError(e.to_string()))) {
550                        // rx dropped
551                        break;
552                    }
553                }
554                Ok(event) => match event {
555                    Event::Message(message) => {
556                        let mut done = false;
557
558                        if message.data == "[DONE]" {
559                            done = true;
560                        }
561
562                        let response = event_mapper(message);
563
564                        if let Err(_e) = tx.send(response) {
565                            // rx dropped
566                            break;
567                        }
568
569                        if done {
570                            break;
571                        }
572                    }
573                    Event::Open => continue,
574                },
575            }
576        }
577
578        event_source.close();
579    });
580
581    Box::pin(tokio_stream::wrappers::UnboundedReceiverStream::new(rx))
582}
async_openai_compat/client.rs

async_openai_compat/
client.rs
