From 2e4343905f0a7efeede9b2529fc9e0fcaa0e4c66 Mon Sep 17 00:00:00 2001 From: Jacob Magar Date: Sun, 12 Apr 2026 08:48:47 -0400 Subject: [PATCH] fix(noxa-68r.4): replace qdrant-client with plain reqwest REST calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit qdrant-client v1.x is gRPC-only — reqwest feature is snapshot downloads only. Rewrite QdrantStore using direct HTTP calls against Qdrant REST API (port 6333): - collection_exists: GET /collections/{name} - create_collection: PUT /collections/{name} + PUT /collections/{name}/index - upsert: PUT /collections/{name}/points?wait=true (256-batch) - delete_by_url: POST /collections/{name}/points/delete?wait=true - search: POST /collections/{name}/points/search No new dependencies — reqwest already in workspace. --- crates/noxa-rag/Cargo.toml | 3 +- crates/noxa-rag/src/config.rs | 2 +- crates/noxa-rag/src/store/qdrant.rs | 292 +++++++++++++++++----------- 3 files changed, 180 insertions(+), 117 deletions(-) diff --git a/crates/noxa-rag/Cargo.toml b/crates/noxa-rag/Cargo.toml index d7e8287..f46aeda 100644 --- a/crates/noxa-rag/Cargo.toml +++ b/crates/noxa-rag/Cargo.toml @@ -33,8 +33,7 @@ async-trait = "0.1" # HTTP client (plain reqwest — no primp patches needed for TEI/Qdrant) reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } -# Vector store — REST transport only (reqwest feature, no protoc required) -qdrant-client = { version = "1", default-features = false, features = ["reqwest"] } +# No qdrant-client crate — REST calls via plain reqwest (no protoc/gRPC dependency) # Chunking text-splitter = { version = "0.25", features = ["markdown", "tokenizers"] } diff --git a/crates/noxa-rag/src/config.rs b/crates/noxa-rag/src/config.rs index 831c454..ec8b319 100644 --- a/crates/noxa-rag/src/config.rs +++ b/crates/noxa-rag/src/config.rs @@ -58,7 +58,7 @@ pub enum EmbedProviderConfig { #[serde(tag = "type", rename_all = "snake_case")] pub enum VectorStoreConfig { Qdrant { - /// gRPC URL — port 6334. qdrant-client v1.x uses gRPC (tonic), NOT REST. + /// REST URL — port 6333 (e.g. http://127.0.0.1:53333 if port-mapped). url: String, collection: String, /// Optional API key. Override with NOXA_RAG_QDRANT_API_KEY env var. diff --git a/crates/noxa-rag/src/store/qdrant.rs b/crates/noxa-rag/src/store/qdrant.rs index d5e5b15..3a7be6b 100644 --- a/crates/noxa-rag/src/store/qdrant.rs +++ b/crates/noxa-rag/src/store/qdrant.rs @@ -1,92 +1,139 @@ use async_trait::async_trait; - -use qdrant_client::Qdrant; -use qdrant_client::qdrant::{ - Condition, CreateCollectionBuilder, CreateFieldIndexCollectionBuilder, DeletePointsBuilder, - Distance, FieldType, Filter, HnswConfigDiffBuilder, PointStruct, SearchPointsBuilder, - UpsertPointsBuilder, VectorParamsBuilder, -}; -use qdrant_client::Payload; +use serde::{Deserialize, Serialize}; +use serde_json::json; use crate::error::RagError; use crate::store::VectorStore; use crate::types::{Point, SearchResult}; +// ── REST request/response shapes ───────────────────────────────────────────── + +#[derive(Deserialize)] +struct QdrantStatus { + status: String, // "ok" on success +} + +#[derive(Deserialize)] +struct CollectionInfoResponse { + result: Option, +} + +#[derive(Serialize)] +struct UpsertRequest { + points: Vec, +} + +#[derive(Serialize)] +struct QdrantPoint { + id: String, // UUID string + vector: Vec, + payload: std::collections::HashMap, +} + +#[derive(Serialize)] +struct DeleteByFilterRequest { + filter: serde_json::Value, +} + +#[derive(Serialize)] +struct SearchRequest { + vector: Vec, + limit: usize, + with_payload: bool, + score_threshold: Option, +} + +#[derive(Deserialize)] +struct SearchResponse { + result: Vec, +} + +#[derive(Deserialize)] +struct SearchHit { + score: f32, + payload: Option>, +} + +// ── QdrantStore ─────────────────────────────────────────────────────────────── + pub struct QdrantStore { - client: Qdrant, + client: reqwest::Client, + base_url: String, // e.g. "http://127.0.0.1:53333" collection: String, - #[allow(dead_code)] uuid_namespace: uuid::Uuid, } impl QdrantStore { - /// Create a new QdrantStore. - /// - /// `url` should be the gRPC endpoint, typically `http://localhost:6334`. - /// The crate uses gRPC transport via tonic — the `reqwest` feature only - /// enables snapshot downloads, not a REST transport. pub fn new( url: &str, collection: String, api_key: Option, uuid_namespace: uuid::Uuid, ) -> Result { - let mut builder = Qdrant::from_url(url); + let mut headers = reqwest::header::HeaderMap::new(); if let Some(key) = api_key { - builder = builder.api_key(key); + headers.insert( + "api-key", + key.parse().map_err(|_| RagError::Config("invalid Qdrant api-key".into()))?, + ); } - let client = builder + let client = reqwest::Client::builder() + .default_headers(headers) .build() - .map_err(|e| RagError::Store(format!("failed to build qdrant client: {e}")))?; + .map_err(|e| RagError::Config(format!("failed to build HTTP client: {e}")))?; Ok(Self { client, + base_url: url.trim_end_matches('/').to_string(), collection, uuid_namespace, }) } - /// Check whether the collection already exists. + /// GET /collections/{name} → true if 200, false if 404. pub async fn collection_exists(&self) -> Result { - self.client - .collection_exists(&self.collection) - .await - .map_err(|e| RagError::Store(format!("collection_exists failed: {e}"))) + let url = format!("{}/collections/{}", self.base_url, self.collection); + let resp = self.client.get(&url).send().await?; + match resp.status().as_u16() { + 200 => Ok(true), + 404 => Ok(false), + s => Err(RagError::Store(format!("collection_exists: unexpected HTTP {s}"))), + } } - /// Create the collection with cosine distance, HNSW m=16/ef_construct=200, - /// and payload indexes on `url` + `domain`. + /// PUT /collections/{name} — create with Cosine/HNSW + payload indexes. pub async fn create_collection(&self, dims: usize) -> Result<(), RagError> { - let hnsw = HnswConfigDiffBuilder::default() - .m(16) - .ef_construct(200) - .build(); + let url = format!("{}/collections/{}", self.base_url, self.collection); + let body = json!({ + "vectors": { + "size": dims, + "distance": "Cosine", + "on_disk": true, + "hnsw_config": { "m": 16, "ef_construct": 200 } + }, + "on_disk_payload": true + }); - let vectors = VectorParamsBuilder::new(dims as u64, Distance::Cosine) - .on_disk(true) - .hnsw_config(hnsw); + let resp = self.client.put(&url).json(&body).send().await?; + if !resp.status().is_success() { + let text = resp.text().await.unwrap_or_default(); + return Err(RagError::Store(format!("create_collection failed: {text}"))); + } - self.client - .create_collection( - CreateCollectionBuilder::new(&self.collection) - .vectors_config(vectors) - .on_disk_payload(true), - ) - .await - .map_err(|e| RagError::Store(format!("create_collection failed: {e}")))?; - - // Payload indexes for fast filtering by url and domain. - for field in ["url", "domain"] { - self.client - .create_field_index(CreateFieldIndexCollectionBuilder::new( - &self.collection, - field, - FieldType::Keyword, - )) - .await - .map_err(|e| { - RagError::Store(format!("create_field_index({field}) failed: {e}")) - })?; + // Payload indexes for fast URL/domain filtering. + for (field, schema_type) in [("url", "keyword"), ("domain", "keyword")] { + let idx_url = format!( + "{}/collections/{}/index", + self.base_url, self.collection + ); + let idx_body = json!({ "field_name": field, "field_schema": schema_type }); + let r = self.client.put(&idx_url).json(&idx_body).send().await?; + if !r.status().is_success() { + let text = r.text().await.unwrap_or_default(); + return Err(RagError::Store(format!( + "create_field_index({field}) failed: {text}" + ))); + } } Ok(()) @@ -95,86 +142,107 @@ impl QdrantStore { #[async_trait] impl VectorStore for QdrantStore { - /// Upsert points into the collection in batches of 256. + /// PUT /collections/{name}/points?wait=true in batches of 256. async fn upsert(&self, points: Vec) -> Result<(), RagError> { + let url = format!( + "{}/collections/{}/points?wait=true", + self.base_url, self.collection + ); + for chunk in points.chunks(256) { - let qdrant_points: Vec = chunk + let qdrant_points: Vec = chunk .iter() .map(|p| { - let mut payload = Payload::new(); - payload.insert("text", p.payload.text.as_str()); - payload.insert("url", p.payload.url.as_str()); - payload.insert("domain", p.payload.domain.as_str()); - payload.insert("chunk_index", p.payload.chunk_index as i64); - payload.insert("total_chunks", p.payload.total_chunks as i64); - payload.insert("token_estimate", p.payload.token_estimate as i64); - - PointStruct::new( - p.id.to_string(), // UUID as string PointId - p.vector.clone(), + let mut payload = std::collections::HashMap::new(); + payload.insert("text".into(), json!(p.payload.text)); + payload.insert("url".into(), json!(p.payload.url)); + payload.insert("domain".into(), json!(p.payload.domain)); + payload.insert("chunk_index".into(), json!(p.payload.chunk_index)); + payload.insert("total_chunks".into(), json!(p.payload.total_chunks)); + payload.insert("token_estimate".into(), json!(p.payload.token_estimate)); + QdrantPoint { + id: p.id.to_string(), + vector: p.vector.clone(), payload, - ) + } }) .collect(); - self.client - .upsert_points( - UpsertPointsBuilder::new(&self.collection, qdrant_points).wait(true), - ) - .await - .map_err(|e| RagError::Store(format!("upsert_points failed: {e}")))?; + let resp = self + .client + .put(&url) + .json(&UpsertRequest { points: qdrant_points }) + .send() + .await?; + + if !resp.status().is_success() { + let text = resp.text().await.unwrap_or_default(); + return Err(RagError::Store(format!("upsert failed: {text}"))); + } } Ok(()) } - /// Delete all points whose `url` payload field matches the normalized URL. + /// POST /collections/{name}/points/delete?wait=true filtered by url payload. async fn delete_by_url(&self, url: &str) -> Result<(), RagError> { let normalized = normalize_url(url); + let endpoint = format!( + "{}/collections/{}/points/delete?wait=true", + self.base_url, self.collection + ); + let body = DeleteByFilterRequest { + filter: json!({ + "must": [{ "key": "url", "match": { "value": normalized } }] + }), + }; - self.client - .delete_points( - DeletePointsBuilder::new(&self.collection) - .points(Filter::must([Condition::matches( - "url", - normalized.clone(), - )])) - .wait(true), - ) - .await - .map_err(|e| RagError::Store(format!("delete_points failed: {e}")))?; + let resp = self.client.post(&endpoint).json(&body).send().await?; + if !resp.status().is_success() { + let text = resp.text().await.unwrap_or_default(); + return Err(RagError::Store(format!("delete_by_url failed: {text}"))); + } Ok(()) } - /// Search for the nearest `limit` vectors and return their payloads. + /// POST /collections/{name}/points/search async fn search(&self, vector: &[f32], limit: usize) -> Result, RagError> { - let response = self - .client - .search_points( - SearchPointsBuilder::new(&self.collection, vector.to_vec(), limit as u64) - .with_payload(true), - ) - .await - .map_err(|e| RagError::Store(format!("search_points failed: {e}")))?; + let url = format!( + "{}/collections/{}/points/search", + self.base_url, self.collection + ); + let body = SearchRequest { + vector: vector.to_vec(), + limit, + with_payload: true, + score_threshold: None, + }; + + let resp = self.client.post(&url).json(&body).send().await?; + if !resp.status().is_success() { + let text = resp.text().await.unwrap_or_default(); + return Err(RagError::Store(format!("search failed: {text}"))); + } + + let response: SearchResponse = resp.json().await?; let results = response .result .into_iter() .filter_map(|hit| { - let text = hit.get("text").as_str()?.to_string(); - let url = hit.get("url").as_str()?.to_string(); - let chunk_index = hit.get("chunk_index").as_integer().unwrap_or(0) as usize; - let token_estimate = - hit.get("token_estimate").as_integer().unwrap_or(0) as usize; - - Some(SearchResult { - text, - url, - score: hit.score, - chunk_index, - token_estimate, - }) + let payload = hit.payload?; + let text = payload.get("text")?.as_str()?.to_string(); + let url = payload.get("url")?.as_str()?.to_string(); + let chunk_index = payload + .get("chunk_index") + .and_then(|v| v.as_u64()) + .unwrap_or(0) as usize; + let token_estimate = payload + .get("token_estimate") + .and_then(|v| v.as_u64()) + .unwrap_or(0) as usize; + Some(SearchResult { text, url, score: hit.score, chunk_index, token_estimate }) }) .collect(); @@ -186,13 +254,9 @@ impl VectorStore for QdrantStore { } } -/// Normalize a URL for consistent storage and lookup: -/// - Strip fragment -/// - Strip trailing slash from path -/// - Scheme and host are already lowercased by the `url` crate +/// Strip fragment, trailing path slash, lowercase scheme+host (url crate already does the latter). fn normalize_url(url: &str) -> String { - use url::Url; - let Ok(mut parsed) = Url::parse(url) else { + let Ok(mut parsed) = url::Url::parse(url) else { return url.to_string(); }; parsed.set_fragment(None);