From bb33c95c5d1668ef5d217dc0eeee609ad11df4bb Mon Sep 17 00:00:00 2001 From: Nikhil Sinha Date: Sun, 16 Aug 2026 18:40:59 +0700 Subject: [PATCH] deprecate llm api and related code --- src/handlers/http/llm.rs | 174 ------------------------ src/handlers/http/mod.rs | 1 - src/handlers/http/modal/query_server.rs | 1 - src/handlers/http/modal/server.rs | 14 +- src/otel_generator.rs | 11 +- 5 files changed, 8 insertions(+), 193 deletions(-) delete mode 100644 src/handlers/http/llm.rs diff --git a/src/handlers/http/llm.rs b/src/handlers/http/llm.rs deleted file mode 100644 index feaeff164..000000000 --- a/src/handlers/http/llm.rs +++ /dev/null @@ -1,174 +0,0 @@ -/* - * Parseable Server (C) 2022 - 2025 Parseable, Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU Affero General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License for more details. - * - * You should have received a copy of the GNU Affero General Public License - * along with this program. If not, see . - * - */ - -use actix_web::{ - HttpRequest, HttpResponse, Result, - http::{StatusCode, header::ContentType}, - web, -}; -use http::header; -use itertools::Itertools; -use reqwest; -use serde_json::{Value, json}; - -use crate::{ - parseable::{PARSEABLE, StreamNotFound}, - utils::get_tenant_id_from_request, -}; - -const OPEN_AI_URL: &str = "https://api.openai.com/v1/chat/completions"; - -// Deserialize types for OpenAI Response -#[derive(serde::Deserialize, Debug)] -struct ResponseData { - choices: Vec, -} - -#[derive(serde::Deserialize, Debug)] -struct Choice { - message: Message, -} - -#[derive(serde::Deserialize, Debug)] -struct Message { - content: String, -} - -// Request body -#[derive(serde::Deserialize, Debug)] -pub struct AiPrompt { - prompt: String, - stream: String, -} - -// Temperory type -#[derive(Debug, serde::Serialize)] -pub struct Field { - name: String, - data_type: String, -} - -impl From<&arrow_schema::Field> for Field { - fn from(field: &arrow_schema::Field) -> Self { - Self { - name: field.name().clone(), - data_type: field.data_type().to_string(), - } - } -} - -fn build_prompt(stream: &str, prompt: &str, schema_json: &str) -> String { - format!( - r#"I have a table called {stream}. -It has the columns:\n{schema_json} -Based on this schema, generate valid SQL for the query: "{prompt}" -Generate only simple SQL as output. Also add comments in SQL syntax to explain your actions. Don't output anything else. If it is not possible to generate valid SQL, output an SQL comment saying so."# - ) -} - -fn build_request_body(ai_prompt: String) -> impl serde::Serialize { - json!({ - "model": "gpt-3.5-turbo", - "messages": [{ "role": "user", "content": ai_prompt}], - "temperature": 0.7, - }) -} - -pub async fn make_llm_request( - req: HttpRequest, - body: web::Json, -) -> Result { - let api_key = match &PARSEABLE.options.open_ai_key { - Some(api_key) if api_key.len() > 3 => api_key, - _ => return Err(LLMError::InvalidAPIKey), - }; - - let stream_name = &body.stream; - let tenant_id = get_tenant_id_from_request(&req); - let schema = PARSEABLE.get_stream(stream_name, &tenant_id)?.get_schema(); - let filtered_schema = schema - .flattened_fields() - .into_iter() - .map(Field::from) - .collect_vec(); - - let schema_json = - serde_json::to_string(&filtered_schema).expect("always converted to valid json"); - - let prompt = build_prompt(stream_name, &body.prompt, &schema_json); - let body = build_request_body(prompt); - - let client = reqwest::Client::new(); - let response = client - .post(OPEN_AI_URL) - .header(header::CONTENT_TYPE, "application/json") - .bearer_auth(api_key) - .json(&body) - .send() - .await?; - - if response.status().is_success() { - let body: ResponseData = response - .json() - .await - .expect("OpenAI response is always the same"); - Ok(HttpResponse::Ok() - .content_type("application/json") - .json(&body.choices[0].message.content)) - } else { - let body: Value = response.json().await?; - let message = body - .as_object() - .and_then(|body| body.get("error")) - .and_then(|error| error.as_object()) - .and_then(|error| error.get("message")) - .map(|message| message.to_string()) - .unwrap_or_else(|| "Error from OpenAI".to_string()); - - Err(LLMError::APIError(message)) - } -} - -#[derive(Debug, thiserror::Error)] -pub enum LLMError { - #[error("Either OpenAI key was not provided or was invalid")] - InvalidAPIKey, - #[error("Failed to call OpenAI endpoint: {0}")] - FailedRequest(#[from] reqwest::Error), - #[error("{0}")] - APIError(String), - #[error("{0}")] - StreamDoesNotExist(#[from] StreamNotFound), -} - -impl actix_web::ResponseError for LLMError { - fn status_code(&self) -> StatusCode { - match self { - Self::InvalidAPIKey => StatusCode::INTERNAL_SERVER_ERROR, - Self::FailedRequest(_) => StatusCode::INTERNAL_SERVER_ERROR, - Self::APIError(_) => StatusCode::INTERNAL_SERVER_ERROR, - Self::StreamDoesNotExist(_) => StatusCode::INTERNAL_SERVER_ERROR, - } - } - - fn error_response(&self) -> actix_web::HttpResponse { - actix_web::HttpResponse::build(self.status_code()) - .insert_header(ContentType::plaintext()) - .body(self.to_string()) - } -} diff --git a/src/handlers/http/mod.rs b/src/handlers/http/mod.rs index 5716809cb..99d6d19a3 100644 --- a/src/handlers/http/mod.rs +++ b/src/handlers/http/mod.rs @@ -40,7 +40,6 @@ pub mod demo_data; pub mod health_check; pub mod ingest; mod kinesis; -pub mod llm; pub mod logstream; pub mod middleware; pub mod modal; diff --git a/src/handlers/http/modal/query_server.rs b/src/handlers/http/modal/query_server.rs index e25eb0edf..6f468be27 100644 --- a/src/handlers/http/modal/query_server.rs +++ b/src/handlers/http/modal/query_server.rs @@ -68,7 +68,6 @@ impl ParseableServer for QueryServer { .service(Server::get_users_webscope()) .service(Server::get_dashboards_webscope()) .service(Server::get_filters_webscope()) - .service(Server::get_llm_webscope()) .service(Server::get_oauth_webscope()) .service(Server::get_roles_webscope()) .service(Self::get_user_role_webscope()) diff --git a/src/handlers/http/modal/server.rs b/src/handlers/http/modal/server.rs index 3b50dfd53..5ad185a06 100644 --- a/src/handlers/http/modal/server.rs +++ b/src/handlers/http/modal/server.rs @@ -62,7 +62,7 @@ use tokio::sync::oneshot; use crate::{ handlers::http::{ - self, ingest, llm, logstream, + self, ingest, logstream, middleware::{DisAllowRootUser, RouteExt}, oidc, role, traces, }, @@ -96,7 +96,6 @@ impl ParseableServer for Server { .service(Self::get_users_webscope()) .service(Self::get_dashboards_webscope()) .service(Self::get_filters_webscope()) - .service(Self::get_llm_webscope()) .service(Self::get_oauth_webscope()) .service(Self::get_user_role_webscope()) .service(Self::get_roles_webscope()) @@ -804,17 +803,6 @@ impl Server { ) } - // get the llm webscope - pub fn get_llm_webscope() -> Scope { - web::scope("/llm").service( - web::resource("").route( - web::post() - .to(llm::make_llm_request) - .authorize(Action::QueryLLM), - ), - ) - } - // get the live check // GET "/liveness" ==> Liveness check as per https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/#define-a-liveness-command // HEAD "/liveness" diff --git a/src/otel_generator.rs b/src/otel_generator.rs index 072283a57..332609450 100644 --- a/src/otel_generator.rs +++ b/src/otel_generator.rs @@ -481,7 +481,7 @@ fn build_batch(sequence: u64) -> TelemetryBatch { kv_string("cloud.region", "us-east-1"), kv_string("http.method", method), kv_string("http.url", path), - kv_int("http.status_code", status_code), + kv_int("http.response.status_code", status_code), kv_string("trace.id", &hex::encode(&trace_id)), kv_string("span.id", &hex::encode(&root_span_id)), kv_string("net.peer.ip", &format!("10.0.{}.{}", index + 1, 10 + index)), @@ -784,7 +784,7 @@ fn build_service_trace( ), kv_string("http.method", method), kv_string("http.url", path), - kv_int("http.status_code", status_code), + kv_int("http.response.status_code", status_code), kv_string("http.scheme", "https"), kv_string("http.target", path), kv_string("http.host", &format!("{service}.internal:8080")), @@ -1147,7 +1147,10 @@ fn build_service_trace( kv_string("http.target", "/v1/charge"), kv_string("http.flavor", "HTTP/2"), kv_int("http.request_content_length", 1_024), - kv_int("http.status_code", if is_error { 503 } else { 200 }), + kv_int( + "http.response.status_code", + if is_error { 503 } else { 200 }, + ), kv_int("http.response_content_length", 2_048), kv_string("net.peer.name", "api.stripe.com"), kv_int("net.peer.port", 443), @@ -1511,7 +1514,7 @@ mod tests { "cloud.region", "http.method", "http.url", - "http.status_code", + "http.response.status_code", "trace.id", "span.id", "net.peer.ip",