feat: completed the video processing

src/ai_models/instance.rs

@@ -4,6 +4,7 @@ use crate::models::chat::{ChatCompletionRequest, ChatCompletionResponse};
 use crate::models::completion::{CompletionRequest, CompletionResponse};
 use crate::models::embeddings::{EmbeddingsRequest, EmbeddingsResponse};
 use crate::models::image::{ImageRequest, ImageResponse};
+use crate::models::video::{VideoRequest, VideoResponse};
 use crate::providers::provider::Provider;
 use axum::http::StatusCode;
 use std::sync::Arc;
@@ -52,4 +53,12 @@ impl ModelInstance {
         payload.model = self.model_type.clone();
         self.provider.process_image(payload, &self.config).await
     }
+
+    pub async fn process_video(
+        &self,
+        mut payload: VideoRequest,
+    ) -> Result<VideoResponse, StatusCode> {
+        payload.model = self.model_type.clone();
+        self.provider.process_video(payload, &self.config).await
+    }
 }

src/config/models.rs

@@ -36,6 +36,7 @@ pub enum PipelineType {
     Embeddings,
     Audio,
     Image,
+    Video,
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone)]

src/models/mod.rs

@@ -10,3 +10,4 @@ pub mod tool_calls;
 pub mod tool_choice;
 pub mod tool_definition;
 pub mod usage;
+pub mod video;

src/models/video.rs

@@ -0,0 +1,25 @@
+use super::usage::Usage;
+use serde::{Deserialize, Serialize};
+
+#[derive(Deserialize, Serialize, Clone)]
+pub struct VideoRequest {
+    pub model: String,
+    pub file_uri: String,
+    pub mime_type: String,
+    pub instruction: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub temperature: Option<f32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub max_tokens: Option<u32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub process_audio: Option<bool>,
+}
+
+#[derive(Deserialize, Serialize, Clone)]
+pub struct VideoResponse {
+    pub id: String,
+    pub model: String,
+    pub created: i64,
+    pub content: String,
+    pub usage: Usage,
+}

src/pipelines/otel.rs

@@ -6,6 +6,7 @@ use crate::models::embeddings::{EmbeddingsInput, EmbeddingsRequest, EmbeddingsRe
 use crate::models::image::{ImageRequest, ImageResponse};
 use crate::models::streaming::ChatCompletionChunk;
 use crate::models::usage::Usage;
+use crate::models::video::{VideoRequest, VideoResponse};
 use opentelemetry::global::{BoxedSpan, ObjectSafeSpan};
 use opentelemetry::trace::{SpanKind, Status, Tracer};
 use opentelemetry::{global, KeyValue};
@@ -380,3 +381,40 @@ impl RecordSpan for ImageResponse {
         self.usage.record_span(span);
     }
 }
+
+impl RecordSpan for VideoRequest {
+    fn record_span(&self, span: &mut BoxedSpan) {
+        span.set_attribute(KeyValue::new("llm.request.type", "video"));
+        span.set_attribute(KeyValue::new(GEN_AI_REQUEST_MODEL, self.model.clone()));
+
+        span.set_attribute(KeyValue::new("video.file_uri", self.file_uri.clone()));
+        span.set_attribute(KeyValue::new("video.mime_type", self.mime_type.clone()));
+
+        span.set_attribute(KeyValue::new("gen_ai.prompt", self.instruction.clone()));
+
+        if let Some(temp) = self.temperature {
+            span.set_attribute(KeyValue::new(GEN_AI_REQUEST_TEMPERATURE, temp as f64));
+        }
+        if let Some(max_tokens) = self.max_tokens {
+            span.set_attribute(KeyValue::new(
+                "gen_ai.request.max_tokens",
+                max_tokens as i64,
+            ));
+        }
+        if let Some(process_audio) = self.process_audio {
+            span.set_attribute(KeyValue::new("video.process_audio", process_audio));
+        }
+    }
+}
+
+impl RecordSpan for VideoResponse {
+    fn record_span(&self, span: &mut BoxedSpan) {
+        span.set_attribute(KeyValue::new(GEN_AI_RESPONSE_MODEL, self.model.clone()));
+        span.set_attribute(KeyValue::new(GEN_AI_RESPONSE_ID, self.id.clone()));
+        span.set_attribute(KeyValue::new(
+            "gen_ai.completion.content",
+            self.content.clone(),
+        ));
+        self.usage.record_span(span);
+    }
+}

src/pipelines/pipeline.rs

@@ -5,6 +5,7 @@ use crate::models::completion::CompletionRequest;
 use crate::models::embeddings::EmbeddingsRequest;
 use crate::models::image::ImageRequest;
 use crate::models::streaming::ChatCompletionChunk;
+use crate::models::video::VideoRequest;
 use crate::pipelines::otel::OtelTracer;
 use crate::{
     ai_models::registry::ModelRegistry,
@@ -50,6 +51,10 @@ pub fn create_pipeline(pipeline: &Pipeline, model_registry: &ModelRegistry) -> R
                     "/image",
                     post(move |state, payload| process_image(state, payload, models)),
                 ),
+                PipelineType::Video => router.route(
+                    "/video",
+                    post(move |state, payload| process_video(state, payload, models)),
+                ),
             },
             _ => router,
         };
@@ -205,3 +210,24 @@ pub async fn process_image(
     tracer.log_error("No matching model found".to_string());
     Err(StatusCode::NOT_FOUND)
 }
+
+pub async fn process_video(
+    State(model_registry): State<Arc<ModelRegistry>>,
+    Json(payload): Json<VideoRequest>,
+    model_keys: Vec<String>,
+) -> Result<impl IntoResponse, StatusCode> {
+    let mut tracer = OtelTracer::start("video", &payload);
+
+    for model_key in model_keys {
+        let model = model_registry.get(&model_key).unwrap();
+
+        if payload.model == model.model_type {
+            let response = model.process_video(payload.clone()).await?;
+            tracer.log_success(&response);
+            return Ok(Json(response));
+        }
+    }
+
+    tracer.log_error("No matching model found".to_string());
+    Err(StatusCode::NOT_FOUND)
+}

src/providers/anthropic/provider.rs

@@ -9,6 +9,7 @@ use crate::models::chat::{ChatCompletionRequest, ChatCompletionResponse};
 use crate::models::completion::{CompletionRequest, CompletionResponse};
 use crate::models::embeddings::{EmbeddingsRequest, EmbeddingsResponse};
 use crate::models::image::{ImageRequest, ImageResponse};
+use crate::models::video::{VideoRequest, VideoResponse};
 use crate::providers::provider::Provider;
 
 pub struct AnthropicProvider {
@@ -105,4 +106,12 @@ impl Provider for AnthropicProvider {
     ) -> Result<ImageResponse, StatusCode> {
         unimplemented!()
     }
+
+    async fn process_video(
+        &self,
+        _payload: VideoRequest,
+        _model_config: &ModelConfig,
+    ) -> Result<VideoResponse, StatusCode> {
+        unimplemented!()
+    }
 }

src/providers/azure/provider.rs

@@ -10,6 +10,7 @@ use crate::models::completion::{CompletionRequest, CompletionResponse};
 use crate::models::embeddings::{EmbeddingsRequest, EmbeddingsResponse};
 use crate::models::image::{ImageRequest, ImageResponse};
 use crate::models::streaming::ChatCompletionChunk;
+use crate::models::video::{VideoRequest, VideoResponse};
 use crate::providers::provider::Provider;
 use reqwest::Client;
 pub struct AzureProvider {
@@ -195,4 +196,12 @@ impl Provider for AzureProvider {
     ) -> Result<ImageResponse, StatusCode> {
         unimplemented!()
     }
+
+    async fn process_video(
+        &self,
+        _payload: VideoRequest,
+        _model_config: &ModelConfig,
+    ) -> Result<VideoResponse, StatusCode> {
+        unimplemented!()
+    }
 }

src/providers/openai/provider.rs

@@ -6,6 +6,7 @@ use crate::models::completion::{CompletionRequest, CompletionResponse};
 use crate::models::embeddings::{EmbeddingsRequest, EmbeddingsResponse};
 use crate::models::image::{ImageRequest, ImageResponse};
 use crate::models::streaming::ChatCompletionChunk;
+use crate::models::video::{VideoRequest, VideoResponse};
 use crate::providers::provider::Provider;
 use axum::async_trait;
 use axum::http::StatusCode;
@@ -155,4 +156,12 @@ impl Provider for OpenAIProvider {
     ) -> Result<ImageResponse, StatusCode> {
         unimplemented!()
     }
+
+    async fn process_video(
+        &self,
+        _payload: VideoRequest,
+        _model_config: &ModelConfig,
+    ) -> Result<VideoResponse, StatusCode> {
+        unimplemented!()
+    }
 }

src/providers/provider.rs

@@ -7,6 +7,7 @@ use crate::models::chat::{ChatCompletionRequest, ChatCompletionResponse};
 use crate::models::completion::{CompletionRequest, CompletionResponse};
 use crate::models::embeddings::{EmbeddingsRequest, EmbeddingsResponse};
 use crate::models::image::{ImageRequest, ImageResponse};
+use crate::models::video::{VideoRequest, VideoResponse};
 
 #[async_trait]
 pub trait Provider: Send + Sync {
@@ -45,4 +46,10 @@ pub trait Provider: Send + Sync {
         payload: ImageRequest,
         model_config: &ModelConfig,
     ) -> Result<ImageResponse, StatusCode>;
+
+    async fn process_video(
+        &self,
+        payload: VideoRequest,
+        model_config: &ModelConfig,
+    ) -> Result<VideoResponse, StatusCode>;
 }

src/providers/vertexai/models.rs

@@ -258,6 +258,14 @@ pub(crate) struct VertexAIImageRequest {
     pub generation_config: Option<GenerationConfig>,
 }
 
+#[derive(Deserialize, Serialize, Clone, Debug)]
+pub(crate) struct VertexAIVideoRequest {
+    #[serde(rename = "contents")]
+    pub contents: Vec<Content>,
+    #[serde(rename = "generation_config")]
+    pub generation_config: Option<GenerationConfig>,
+}
+
 impl From<crate::models::chat::ChatCompletionRequest> for VertexAIChatCompletionRequest {
     fn from(request: crate::models::chat::ChatCompletionRequest) -> Self {
         let contents = request
@@ -603,3 +611,68 @@ impl From<VertexAIChatCompletionResponse> for crate::models::image::ImageRespons
         }
     }
 }
+
+impl From<crate::models::video::VideoRequest> for VertexAIVideoRequest {
+    fn from(request: crate::models::video::VideoRequest) -> Self {
+        VertexAIVideoRequest {
+            contents: vec![Content {
+                role: "user".to_string(),
+                parts: vec![
+                    // First part contains the video file
+                    Part {
+                        content: PartContent::File {
+                            file_data: FileData {
+                                file_uri: request.file_uri,
+                                mime_type: request.mime_type,
+                            },
+                        },
+                    },
+                    // Second part contains the instruction
+                    Part {
+                        content: PartContent::Text {
+                            text: request.instruction,
+                        },
+                    },
+                ],
+            }],
+            generation_config: Some(GenerationConfig {
+                temperature: request.temperature,
+                max_output_tokens: request.max_tokens,
+                top_p: None,
+                top_k: None,
+                candidate_count: None,
+                audio_timestamp: request.process_audio,
+            }),
+        }
+    }
+}
+
+impl From<VertexAIChatCompletionResponse> for crate::models::video::VideoResponse {
+    fn from(response: VertexAIChatCompletionResponse) -> Self {
+        crate::models::video::VideoResponse {
+            id: uuid::Uuid::new_v4().to_string(),
+            model: "gemini-1.5-pro".to_string(),
+            created: chrono::Utc::now().timestamp(),
+            content: response
+                .candidates
+                .first()
+                .and_then(
+                    |candidate| match &candidate.content.parts.first().map(|p| &p.content) {
+                        Some(PartContent::Text { text }) => Some(text.clone()),
+                        _ => None,
+                    },
+                )
+                .unwrap_or_default(),
+            usage: response
+                .usage_metadata
+                .map(|metadata| Usage {
+                    prompt_tokens: metadata.prompt_token_count as u32,
+                    completion_tokens: metadata.candidates_token_count as u32,
+                    total_tokens: metadata.total_token_count as u32,
+                    completion_tokens_details: None,
+                    prompt_tokens_details: None,
+                })
+                .unwrap_or_default(),
+        }
+    }
+}

src/providers/vertexai/provider.rs

@@ -1,7 +1,7 @@
 use super::models::{
     VertexAIAudioRequest, VertexAIAudioResponse, VertexAIChatCompletionRequest,
     VertexAIChatCompletionResponse, VertexAIEmbeddingsRequest, VertexAIEmbeddingsResponse,
-    VertexAIImageRequest, VertexAIStreamChunk,
+    VertexAIImageRequest, VertexAIStreamChunk, VertexAIVideoRequest,
 };
 use crate::config::constants::stream_buffer_size_bytes;
 use crate::config::models::{ModelConfig, Provider as ProviderConfig};
@@ -10,6 +10,7 @@ use crate::models::chat::{ChatCompletionRequest, ChatCompletionResponse};
 use crate::models::completion::{CompletionRequest, CompletionResponse};
 use crate::models::embeddings::{EmbeddingsRequest, EmbeddingsResponse};
 use crate::models::image::{ImageRequest, ImageResponse};
+use crate::models::video::{VideoRequest, VideoResponse};
 use crate::providers::provider::Provider;
 use axum::async_trait;
 use axum::http::StatusCode;
@@ -287,4 +288,23 @@ impl Provider for VertexAIProvider {
             .await?;
         Ok(response.into())
     }
+
+    async fn process_video(
+        &self,
+        payload: VideoRequest,
+        _model_config: &ModelConfig,
+    ) -> Result<VideoResponse, StatusCode> {
+        let token = self.get_token().await?;
+        let request: VertexAIVideoRequest = payload.clone().into();
+
+        let url = self
+            .construct_vertex_url(&payload.model, "generateContent")
+            .await?;
+
+        let response: VertexAIChatCompletionResponse = self
+            .make_vertex_api_call(url, token, json!(request))
+            .await?;
+
+        Ok(response.into())
+    }
 }

src/routes.rs

@@ -37,6 +37,8 @@ pub fn create_router(state: Arc<AppState>) -> Router {
             return *pipeline_idxs.get("embeddings").unwrap_or(&0);
         } else if path.contains("/image") {
             return *pipeline_idxs.get("image").unwrap_or(&0);
+        } else if path.contains("/video") {
+            return *pipeline_idxs.get("video").unwrap_or(&0);
         }
 
         // Fall back to header-based routing if needed