Better error handling

async-openai/src/client.rs

@@ -341,7 +341,12 @@ impl<C: Config> Client<C> {
                 let message: String = String::from_utf8_lossy(&bytes).into_owned();
                 tracing::warn!("Server error: {status} - {message}");
                 return Err(backoff::Error::Transient {
-                    err: OpenAIError::ApiError(ApiError { message, r#type: None, param: None, code: None }),
+                    err: OpenAIError::ApiError(ApiError {
+                        message,
+                        r#type: None,
+                        param: None,
+                        code: None,
+                    }),
                     retry_after: None,
                 });
             }
@@ -475,7 +480,7 @@ where
         while let Some(ev) = event_source.next().await {
             match ev {
                 Err(e) => {
-                    if let Err(_e) = tx.send(Err(OpenAIError::StreamError(e.to_string()))) {
+                    if let Err(_e) = tx.send(Err(OpenAIError::StreamError(e))) {
                         // rx dropped
                         break;
                     }
@@ -520,7 +525,7 @@ where
         while let Some(ev) = event_source.next().await {
             match ev {
                 Err(e) => {
-                    if let Err(_e) = tx.send(Err(OpenAIError::StreamError(e.to_string()))) {
+                    if let Err(_e) = tx.send(Err(OpenAIError::StreamError(e))) {
                         // rx dropped
                         break;
                     }

async-openai/src/embedding.rs

@@ -64,7 +64,6 @@ impl<'c, C: Config> Embeddings<'c, C> {
 
 #[cfg(test)]
 mod tests {
-    use crate::error::OpenAIError;
     use crate::types::{CreateEmbeddingResponse, Embedding, EncodingFormat};
     use crate::{types::CreateEmbeddingRequestArgs, Client};
 

async-openai/src/error.rs

@@ -20,7 +20,7 @@ pub enum OpenAIError {
     FileReadError(String),
     /// Error on SSE streaming
     #[error("stream failed: {0}")]
-    StreamError(String),
+    StreamError(reqwest_eventsource::Error),
     /// Error from client side validation
     /// or when builder fails to build request before making API call
     #[error("invalid args: {0}")]

async-openai/src/lib.rs

@@ -185,3 +185,6 @@ pub use users::Users;
 pub use vector_store_file_batches::VectorStoreFileBatches;
 pub use vector_store_files::VectorStoreFiles;
 pub use vector_stores::VectorStores;
+
+// re-exports
+pub use reqwest_eventsource;

async-openai/src/types/assistant_stream.rs

@@ -1,4 +1,4 @@
-use std::pin::Pin;
+use std::{pin::Pin, time::Duration};
 
 use futures::Stream;
 use serde::Deserialize;
@@ -108,13 +108,22 @@ pub enum AssistantStreamEvent {
     /// Occurs when a stream ends.
     #[serde(rename = "done")]
     Done(String),
+    /// Unknown event type.
+    #[serde(rename = "unknown")]
+    Unknown {
+        event: String,
+        data: String,
+        id: String,
+        retry: Option<Duration>,
+    },
 }
 
 pub type AssistantEventStream =
     Pin<Box<dyn Stream<Item = Result<AssistantStreamEvent, OpenAIError>> + Send>>;
 
 impl TryFrom<eventsource_stream::Event> for AssistantStreamEvent {
     type Error = OpenAIError;
+
     fn try_from(value: eventsource_stream::Event) -> Result<Self, Self::Error> {
         match value.event.as_str() {
             "thread.created" => serde_json::from_str::<ThreadObject>(value.data.as_str())
@@ -206,10 +215,20 @@ impl TryFrom<eventsource_stream::Event> for AssistantStreamEvent {
                 .map_err(|e| map_deserialization_error(e, value.data.as_bytes()))
                 .map(AssistantStreamEvent::ErrorEvent),
             "done" => Ok(AssistantStreamEvent::Done(value.data)),
-
-            _ => Err(OpenAIError::StreamError(
-                "Unrecognized event: {value:?#}".into(),
-            )),
+            _ => {
+                let eventsource_stream::Event {
+                    id,
+                    event,
+                    data,
+                    retry,
+                } = value;
+                Ok(AssistantStreamEvent::Unknown {
+                    event,
+                    data,
+                    id,
+                    retry,
+                })
+            }
         }
     }
 }