mirror of
https://github.com/meta-llama/llama-stack.git
synced 2025-12-05 02:17:31 +00:00
fix(mongodb): update protocol compliance and add graceful connection failure handling
- Changed insert_chunks and query_chunks parameter from vector_db_id to vector_store_id - Updated method names: register_vector_db -> register_vector_store, unregister_vector_db -> unregister_vector_store - Updated types: VectorDB -> VectorStore, VectorDBsProtocolPrivate -> VectorStoresProtocolPrivate, VectorDBWithIndex -> VectorStoreWithIndex - Added support for individual connection parameters (host, port, username, password) with precedence over connection_string - Changed kvstore config to use KVStoreReference with kvstore_impl for initialization - Added graceful connection failure handling with clean warning messages - Tests now skip gracefully when MongoDB is not running instead of erroring
This commit is contained in:
parent
c4ee3dcb35
commit
83276d4aaa
5 changed files with 187 additions and 77 deletions
|
|
@ -19,10 +19,26 @@ class MongoDBVectorIOConfig(BaseModel):
|
|||
This provider connects to MongoDB Atlas and uses Vector Search for RAG operations.
|
||||
"""
|
||||
|
||||
# MongoDB Atlas connection details
|
||||
# MongoDB connection details - either connection_string or individual parameters
|
||||
connection_string: str | None = Field(
|
||||
default=None,
|
||||
description="MongoDB Atlas connection string (e.g., mongodb+srv://user:pass@cluster.mongodb.net/)",
|
||||
description="MongoDB connection string (e.g., mongodb://user:pass@localhost:27017/ or mongodb+srv://user:pass@cluster.mongodb.net/)",
|
||||
)
|
||||
host: str | None = Field(
|
||||
default=None,
|
||||
description="MongoDB host (used if connection_string is not provided)",
|
||||
)
|
||||
port: int | None = Field(
|
||||
default=None,
|
||||
description="MongoDB port (used if connection_string is not provided)",
|
||||
)
|
||||
username: str | None = Field(
|
||||
default=None,
|
||||
description="MongoDB username (used if connection_string is not provided)",
|
||||
)
|
||||
password: str | None = Field(
|
||||
default=None,
|
||||
description="MongoDB password (used if connection_string is not provided)",
|
||||
)
|
||||
database_name: str = Field(default="llama_stack", description="Database name to use for vector collections")
|
||||
|
||||
|
|
@ -43,16 +59,44 @@ class MongoDBVectorIOConfig(BaseModel):
|
|||
description="Config for KV store backend for metadata storage", default=None
|
||||
)
|
||||
|
||||
def get_connection_string(self) -> str | None:
|
||||
"""Build connection string from individual parameters if not provided directly.
|
||||
|
||||
If both connection_string and individual parameters (host/port) are provided,
|
||||
individual parameters take precedence to allow test environment overrides.
|
||||
"""
|
||||
# Prioritize individual connection parameters over connection_string
|
||||
# This allows test environments to override with MONGODB_HOST/PORT/etc
|
||||
if self.host and self.port:
|
||||
auth_part = ""
|
||||
if self.username and self.password:
|
||||
auth_part = f"{self.username}:{self.password}@"
|
||||
return f"mongodb://{auth_part}{self.host}:{self.port}/"
|
||||
|
||||
# Fall back to connection_string if provided
|
||||
if self.connection_string:
|
||||
return self.connection_string
|
||||
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def sample_run_config(
|
||||
cls,
|
||||
__distro_dir__: str,
|
||||
connection_string: str = "${env.MONGODB_CONNECTION_STRING:=}",
|
||||
host: str = "${env.MONGODB_HOST:=localhost}",
|
||||
port: str = "${env.MONGODB_PORT:=27017}",
|
||||
username: str = "${env.MONGODB_USERNAME:=}",
|
||||
password: str = "${env.MONGODB_PASSWORD:=}",
|
||||
database_name: str = "${env.MONGODB_DATABASE_NAME:=llama_stack}",
|
||||
**kwargs: Any,
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"connection_string": connection_string,
|
||||
"host": host,
|
||||
"port": port,
|
||||
"username": username,
|
||||
"password": password,
|
||||
"database_name": database_name,
|
||||
"index_name": "${env.MONGODB_INDEX_NAME:=vector_index}",
|
||||
"path_field": "${env.MONGODB_PATH_FIELD:=embedding}",
|
||||
|
|
|
|||
|
|
@ -420,18 +420,21 @@ class MongoDBVectorIOAdapter(OpenAIVectorStoreMixin, VectorIO, VectorStoresProto
|
|||
if self.config.persistence:
|
||||
self.kvstore = await kvstore_impl(self.config.persistence)
|
||||
|
||||
# Skip MongoDB connection if no connection string provided
|
||||
# Get connection string from config (either direct or built from parameters)
|
||||
connection_string = self.config.get_connection_string()
|
||||
|
||||
# Skip MongoDB connection if no connection parameters provided
|
||||
# This allows other providers to work without MongoDB credentials
|
||||
if not self.config.connection_string:
|
||||
if not connection_string:
|
||||
logger.warning(
|
||||
"MongoDB connection_string not provided. "
|
||||
"MongoDB connection parameters not provided. "
|
||||
"MongoDB vector store will not be available until credentials are configured."
|
||||
)
|
||||
return
|
||||
|
||||
# Connect to MongoDB with optimized settings for RAG
|
||||
self.client = MongoClient(
|
||||
self.config.connection_string,
|
||||
connection_string,
|
||||
server_api=ServerApi("1"),
|
||||
maxPoolSize=self.config.max_pool_size,
|
||||
serverSelectionTimeoutMS=self.config.timeout_ms,
|
||||
|
|
@ -441,8 +444,22 @@ class MongoDBVectorIOAdapter(OpenAIVectorStoreMixin, VectorIO, VectorStoresProto
|
|||
)
|
||||
|
||||
# Test connection
|
||||
self.client.admin.command("ping")
|
||||
logger.info("Successfully connected to MongoDB Atlas for RAG")
|
||||
try:
|
||||
self.client.admin.command("ping")
|
||||
logger.info("Successfully connected to MongoDB Atlas for RAG")
|
||||
except Exception as conn_error:
|
||||
# Extract just the basic error type without the full traceback
|
||||
error_type = type(conn_error).__name__
|
||||
logger.warning(
|
||||
f"MongoDB connection failed ({error_type}). "
|
||||
"MongoDB vector store will not be available. "
|
||||
f"Attempted to connect to: {self.config.host or 'connection_string'}:{self.config.port or '(from connection_string)'}"
|
||||
)
|
||||
# Close the client and clear it
|
||||
if self.client:
|
||||
self.client.close()
|
||||
self.client = None
|
||||
return
|
||||
|
||||
# Get database
|
||||
self.database = self.client[self.config.database_name]
|
||||
|
|
@ -457,7 +474,12 @@ class MongoDBVectorIOAdapter(OpenAIVectorStoreMixin, VectorIO, VectorStoresProto
|
|||
|
||||
except Exception as e:
|
||||
logger.exception("Failed to initialize MongoDB Atlas Vector IO adapter for RAG")
|
||||
raise RuntimeError("Failed to initialize MongoDB Atlas Vector IO adapter for RAG") from e
|
||||
# Close the client if it was created
|
||||
if self.client:
|
||||
self.client.close()
|
||||
self.client = None
|
||||
# Log warning instead of raising to allow tests to skip gracefully
|
||||
logger.warning(f"MongoDB initialization failed: {e}. MongoDB vector store will not be available.")
|
||||
|
||||
async def shutdown(self) -> None:
|
||||
"""Shutdown MongoDB connection."""
|
||||
|
|
@ -525,22 +547,22 @@ class MongoDBVectorIOAdapter(OpenAIVectorStoreMixin, VectorIO, VectorStoresProto
|
|||
|
||||
async def insert_chunks(
|
||||
self,
|
||||
vector_db_id: str,
|
||||
vector_store_id: str,
|
||||
chunks: list[Chunk],
|
||||
ttl_seconds: int | None = None,
|
||||
) -> None:
|
||||
"""Insert chunks into the vector database optimized for RAG."""
|
||||
vector_db_with_index = await self._get_vector_db_index(vector_db_id)
|
||||
vector_db_with_index = await self._get_vector_db_index(vector_store_id)
|
||||
await vector_db_with_index.insert_chunks(chunks)
|
||||
|
||||
async def query_chunks(
|
||||
self,
|
||||
vector_db_id: str,
|
||||
vector_store_id: str,
|
||||
query: InterleavedContent,
|
||||
params: dict[str, Any] | None = None,
|
||||
) -> QueryChunksResponse:
|
||||
"""Query chunks from the vector database optimized for RAG context retrieval."""
|
||||
vector_db_with_index = await self._get_vector_db_index(vector_db_id)
|
||||
vector_db_with_index = await self._get_vector_db_index(vector_store_id)
|
||||
return await vector_db_with_index.query_chunks(query, params)
|
||||
|
||||
async def delete_chunks(self, store_id: str, chunks_for_deletion: list[ChunkForDeletion]) -> None:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue