chore: Bring async-openai into repo as request starter (#2520)

Co-authored-by: Graham King <[email protected]>

Author: nachiketb-nvidia

Cargo.lock

@@ -9,6 +9,8 @@ members = [
     "lib/llm",
     "lib/runtime",
     "lib/tokens",
+    "lib/async-openai",
+    "lib/async-openai-macros",
     "lib/bindings/c",
     "lib/engines/*",
 ]
@@ -29,11 +31,11 @@ keywords = ["llm", "genai", "inference", "nvidia", "distributed", "dynamo"]
 dynamo-runtime = { path = "lib/runtime", version = "0.4.0" }
 dynamo-llm = { path = "lib/llm", version = "0.4.0" }
 dynamo-tokens = { path = "lib/tokens", version = "0.4.0" }
+dynamo-async-openai = { path = "lib/async-openai", version = "0.4.0", features = ["byot", "rustls"]}
 
 # External dependencies
 anyhow = { version = "1" }
 async-nats = { version = "0.40", features = ["service"] }
-async-openai = { version = "0.29.0", features = ["rustls", "byot"] }
 async-stream = { version = "0.3" }
 async-trait = { version = "0.1" }
 async_zmq = { version = "0.4.0" }

Cargo.toml

@@ -33,7 +33,7 @@ dynamo-engine-llamacpp = { path = "../../lib/engines/llamacpp", optional = true
 dynamo-engine-mistralrs = { path = "../../lib/engines/mistralrs", optional = true }
 
 anyhow = { workspace = true }
-async-openai = { workspace = true }
+dynamo-async-openai = { workspace = true }
 async-stream = { workspace = true }
 async-trait = { workspace = true }
 either = { workspace = true }

launch/dynamo-run/Cargo.toml

@@ -0,0 +1,27 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Based on https://github.com/64bit/async-openai/ by Himanshu Neema
+# Original Copyright (c) 2022 Himanshu Neema
+# Licensed under MIT License (see ATTRIBUTIONS-Rust.md)
+#
+# Modifications Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES.
+# Licensed under Apache 2.0
+
+[package]
+name = "async-openai-macros"
+version.workspace = true
+edition.workspace = true
+authors.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+readme.workspace = true
+
+[lib]
+proc-macro = true
+
+[dependencies]
+syn = { version = "2.0", features = ["full"] }
+quote = "1.0"
+proc-macro2 = "1.0"

lib/async-openai-macros/Cargo.toml

@@ -0,0 +1,151 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+//
+// Based on https://github.com/64bit/async-openai/ by Himanshu Neema
+// Original Copyright (c) 2022 Himanshu Neema
+// Licensed under MIT License (see ATTRIBUTIONS-Rust.md)
+//
+// Modifications Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES.
+// Licensed under Apache 2.0
+
+use proc_macro::TokenStream;
+use quote::{quote, ToTokens};
+use syn::{
+    parse::{Parse, ParseStream},
+    parse_macro_input,
+    punctuated::Punctuated,
+    token::Comma,
+    FnArg, GenericParam, Generics, ItemFn, Pat, PatType, TypeParam, WhereClause,
+};
+
+// Parse attribute arguments like #[byot(T0: Display + Debug, T1: Clone, R: Serialize)]
+struct BoundArgs {
+    bounds: Vec<(String, syn::TypeParamBound)>,
+    where_clause: Option<String>,
+    stream: bool, // Add stream flag
+}
+
+impl Parse for BoundArgs {
+    fn parse(input: ParseStream) -> syn::Result<Self> {
+        let mut bounds = Vec::new();
+        let mut where_clause = None;
+        let mut stream = false; // Default to false
+        let vars = Punctuated::<syn::MetaNameValue, Comma>::parse_terminated(input)?;
+
+        for var in vars {
+            let name = var.path.get_ident().unwrap().to_string();
+            match name.as_str() {
+                "where_clause" => {
+                    where_clause = Some(var.value.into_token_stream().to_string());
+                }
+                "stream" => {
+                    stream = var.value.into_token_stream().to_string().contains("true");
+                }
+                _ => {
+                    let bound: syn::TypeParamBound =
+                        syn::parse_str(&var.value.into_token_stream().to_string())?;
+                    bounds.push((name, bound));
+                }
+            }
+        }
+        Ok(BoundArgs {
+            bounds,
+            where_clause,
+            stream,
+        })
+    }
+}
+
+#[proc_macro_attribute]
+pub fn byot_passthrough(_args: TokenStream, item: TokenStream) -> TokenStream {
+    item
+}
+
+#[proc_macro_attribute]
+pub fn byot(args: TokenStream, item: TokenStream) -> TokenStream {
+    let bounds_args = parse_macro_input!(args as BoundArgs);
+    let input = parse_macro_input!(item as ItemFn);
+    let mut new_generics = Generics::default();
+    let mut param_count = 0;
+
+    // Process function arguments
+    let mut new_params = Vec::new();
+    let args = input
+        .sig
+        .inputs
+        .iter()
+        .map(|arg| {
+            match arg {
+                FnArg::Receiver(receiver) => receiver.to_token_stream(),
+                FnArg::Typed(PatType { pat, .. }) => {
+                    if let Pat::Ident(pat_ident) = &**pat {
+                        let generic_name = format!("T{}", param_count);
+                        let generic_ident =
+                            syn::Ident::new(&generic_name, proc_macro2::Span::call_site());
+
+                        // Create type parameter with optional bounds
+                        let mut type_param = TypeParam::from(generic_ident.clone());
+                        if let Some((_, bound)) = bounds_args
+                            .bounds
+                            .iter()
+                            .find(|(name, _)| name == &generic_name)
+                        {
+                            type_param.bounds.extend(vec![bound.clone()]);
+                        }
+
+                        new_params.push(GenericParam::Type(type_param));
+                        param_count += 1;
+                        quote! { #pat_ident: #generic_ident }
+                    } else {
+                        arg.to_token_stream()
+                    }
+                }
+            }
+        })
+        .collect::<Vec<_>>();
+
+    // Add R type parameter with optional bounds
+    let generic_r = syn::Ident::new("R", proc_macro2::Span::call_site());
+    let mut return_type_param = TypeParam::from(generic_r.clone());
+    if let Some((_, bound)) = bounds_args.bounds.iter().find(|(name, _)| name == "R") {
+        return_type_param.bounds.extend(vec![bound.clone()]);
+    }
+    new_params.push(GenericParam::Type(return_type_param));
+
+    // Add all generic parameters
+    new_generics.params.extend(new_params);
+
+    let fn_name = &input.sig.ident;
+    let byot_fn_name = syn::Ident::new(&format!("{}_byot", fn_name), fn_name.span());
+    let vis = &input.vis;
+    let block = &input.block;
+    let attrs = &input.attrs;
+    let asyncness = &input.sig.asyncness;
+
+    // Parse where clause if provided
+    let where_clause = if let Some(where_str) = bounds_args.where_clause {
+        match syn::parse_str::<WhereClause>(&format!("where {}", where_str.replace("\"", ""))) {
+            Ok(where_clause) => quote! { #where_clause },
+            Err(e) => return TokenStream::from(e.to_compile_error()),
+        }
+    } else {
+        quote! {}
+    };
+
+    // Generate return type based on stream flag
+    let return_type = if bounds_args.stream {
+        quote! { Result<::std::pin::Pin<Box<dyn ::futures::Stream<Item = Result<R, OpenAIError>> + Send>>, OpenAIError> }
+    } else {
+        quote! { Result<R, OpenAIError> }
+    };
+
+    let expanded = quote! {
+        #(#attrs)*
+        #input
+
+        #(#attrs)*
+        #vis #asyncness fn #byot_fn_name #new_generics (#(#args),*) -> #return_type #where_clause #block
+    };
+
+    expanded.into()
+}

lib/async-openai-macros/src/lib.rs

@@ -0,0 +1,70 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Based on https://github.com/64bit/async-openai/ by Himanshu Neema
+# Original Copyright (c) 2022 Himanshu Neema
+# Licensed under MIT License (see ATTRIBUTIONS-Rust.md)
+#
+# Modifications Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES.
+# Licensed under Apache 2.0
+
+[package]
+name = "dynamo-async-openai"
+version.workspace = true
+edition.workspace = true
+authors.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+readme.workspace = true
+
+[features]
+default = ["rustls"]
+# Enable rustls for TLS support
+rustls = ["reqwest/rustls-tls-native-roots"]
+# Enable rustls and webpki-roots
+rustls-webpki-roots = ["reqwest/rustls-tls-webpki-roots"]
+# Enable native-tls for TLS support
+native-tls = ["reqwest/native-tls"]
+# Remove dependency on OpenSSL
+native-tls-vendored = ["reqwest/native-tls-vendored"]
+realtime = ["dep:tokio-tungstenite"]
+# Bring your own types
+byot = []
+
+[dependencies]
+async-openai-macros = { path = "../async-openai-macros" }
+backoff = { version = "0.4.0", features = ["tokio"] }
+base64 = "0.22.1"
+futures = "0.3.31"
+rand = "0.9.0"
+reqwest = { version = "0.12.12", features = [
+  "json",
+  "stream",
+  "multipart",
+], default-features = false }
+reqwest-eventsource = "0.6.0"
+serde = { version = "1.0.217", features = ["derive", "rc"] }
+serde_json = "1.0.135"
+thiserror = "2.0.11"
+tokio = { version = "1.43.0", features = ["fs", "macros"] }
+tokio-stream = "0.1.17"
+tokio-util = { version = "0.7.13", features = ["codec", "io-util"] }
+tracing = "0.1.41"
+derive_builder = "0.20.2"
+secrecy = { version = "0.10.3", features = ["serde"] }
+bytes = "1.9.0"
+eventsource-stream = "0.2.3"
+tokio-tungstenite = { version = "0.26.1", optional = true, default-features = false }
+
+[dev-dependencies]
+tokio-test = "0.4.4"
+serde_json = "1.0"
+
+[[test]]
+name = "bring-your-own-type"
+required-features = ["byot"]
+
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = ["--cfg", "docsrs"]

lib/async-openai/Cargo.toml

@@ -0,0 +1,80 @@
+// SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+//
+// Based on https://github.com/64bit/async-openai/ by Himanshu Neema
+// Original Copyright (c) 2022 Himanshu Neema
+// Licensed under MIT License (see ATTRIBUTIONS-Rust.md)
+//
+// Modifications Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES.
+// Licensed under Apache 2.0
+
+use serde::Serialize;
+
+use crate::{
+    config::Config,
+    error::OpenAIError,
+    types::{
+        AssistantObject, CreateAssistantRequest, DeleteAssistantResponse, ListAssistantsResponse,
+        ModifyAssistantRequest,
+    },
+    Client,
+};
+
+/// Build assistants that can call models and use tools to perform tasks.
+///
+/// [Get started with the Assistants API](https://platform.openai.com/docs/assistants)
+pub struct Assistants<'c, C: Config> {
+    client: &'c Client<C>,
+}
+
+impl<'c, C: Config> Assistants<'c, C> {
+    pub fn new(client: &'c Client<C>) -> Self {
+        Self { client }
+    }
+
+    /// Create an assistant with a model and instructions.
+    #[crate::byot(T0 = serde::Serialize, R = serde::de::DeserializeOwned)]
+    pub async fn create(
+        &self,
+        request: CreateAssistantRequest,
+    ) -> Result<AssistantObject, OpenAIError> {
+        self.client.post("/assistants", request).await
+    }
+
+    /// Retrieves an assistant.
+    #[crate::byot(T0 = std::fmt::Display, R = serde::de::DeserializeOwned)]
+    pub async fn retrieve(&self, assistant_id: &str) -> Result<AssistantObject, OpenAIError> {
+        self.client
+            .get(&format!("/assistants/{assistant_id}"))
+            .await
+    }
+
+    /// Modifies an assistant.
+    #[crate::byot(T0 = std::fmt::Display, T1 = serde::Serialize, R = serde::de::DeserializeOwned)]
+    pub async fn update(
+        &self,
+        assistant_id: &str,
+        request: ModifyAssistantRequest,
+    ) -> Result<AssistantObject, OpenAIError> {
+        self.client
+            .post(&format!("/assistants/{assistant_id}"), request)
+            .await
+    }
+
+    /// Delete an assistant.
+    #[crate::byot(T0 = std::fmt::Display, R = serde::de::DeserializeOwned)]
+    pub async fn delete(&self, assistant_id: &str) -> Result<DeleteAssistantResponse, OpenAIError> {
+        self.client
+            .delete(&format!("/assistants/{assistant_id}"))
+            .await
+    }
+
+    /// Returns a list of assistants.
+    #[crate::byot(T0 = serde::Serialize, R = serde::de::DeserializeOwned)]
+    pub async fn list<Q>(&self, query: &Q) -> Result<ListAssistantsResponse, OpenAIError>
+    where
+        Q: Serialize + ?Sized,
+    {
+        self.client.get_with_query("/assistants", &query).await
+    }
+}