LCORE-873: Support Solr Vector I/O Provider in LCORE

lightspeed-stack.yaml

@@ -30,3 +30,7 @@ conversation_cache:
 
 authentication:
   module: "noop"
+
+
+solr:
+  offline: True

run.yaml

@@ -15,7 +15,7 @@ apis:
 benchmarks: []
 datasets: []
 image_name: starter
-# external_providers_dir: /opt/app-root/src/.llama/providers.d
+external_providers_dir: ${env.EXTERNAL_PROVIDERS_DIR}
 
 providers:
   inference:
@@ -24,7 +24,9 @@ providers:
     config:
       api_key: ${env.OPENAI_API_KEY}
       allowed_models: ["${env.E2E_OPENAI_MODEL:=gpt-4o-mini}"]
-  - config: {}
+  - config:
+      allowed_models:
+        - ${env.EMBEDDING_MODEL_DIR}
     provider_id: sentence-transformers
     provider_type: inline::sentence-transformers
   files:
@@ -56,6 +58,18 @@ providers:
     provider_id: rag-runtime
     provider_type: inline::rag-runtime
   vector_io:
+  - provider_id: solr-vector
+    provider_type: remote::solr_vector_io
+    config:
+      solr_url: http://localhost:8983/solr
+      collection_name: portal-rag
+      vector_field: chunk_vector
+      content_field: chunk
+      embedding_dimension: 384
+      embedding_model: ${env.EMBEDDING_MODEL_DIR}
+      persistence:
+        namespace: portal-rag
+        backend: kv_default
   - config: # Define the storage backend for RAG
       persistence:
         namespace: vector_io::faiss
@@ -127,7 +141,13 @@ storage:
       namespace: prompts
       backend: kv_default
 registered_resources:
-  models: []
+  models:
+  - model_id: granite-embedding-30m
+    model_type: embedding
+    provider_id: sentence-transformers
+    provider_model_id: ${env.EMBEDDING_MODEL_DIR}
+    metadata:
+      embedding_dimension: 384
   shields:
   - shield_id: llama-guard
     provider_id: llama-guard

src/app/endpoints/query.py

@@ -386,9 +386,9 @@ async def query_endpoint_handler_base(  # pylint: disable=R0914
         response = QueryResponse(
             conversation_id=conversation_id,
             response=summary.llm_response,
-            tool_calls=summary.tool_calls,
-            tool_results=summary.tool_results,
-            rag_chunks=summary.rag_chunks,
+            rag_chunks=rag_chunks_dict,
+            tool_calls=summary.tool_calls if summary.tool_calls else [],
+            tool_results=summary.tool_results if summary.tool_results else [],
             referenced_documents=referenced_documents,
             truncated=False,  # TODO: implement truncation detection
             input_tokens=token_usage.input_tokens,

src/app/endpoints/query_v2.py

@@ -59,6 +59,7 @@
 from utils.suid import normalize_conversation_id, to_llama_stack_conversation_id
 from utils.token_counter import TokenCounter
 from utils.types import RAGChunk, ToolCallSummary, ToolResultSummary, TurnSummary
+from utils.vector_search import perform_vector_search, format_rag_context_for_injection
 
 logger = logging.getLogger("app.endpoints.handlers")
 router = APIRouter(tags=["query_v1"])
@@ -364,9 +365,14 @@ async def retrieve_response(  # pylint: disable=too-many-locals,too-many-branche
     if query_request.attachments:
         validate_attachments_metadata(query_request.attachments)
 
-    # Prepare tools for responses API
+    # Prepare tools for responses API - skip RAG tools since we're doing direct vector query
     toolgroups = await prepare_tools_for_responses_api(
-        client, query_request, token, configuration, mcp_headers
+        client,
+        query_request,
+        token,
+        configuration,
+        mcp_headers=mcp_headers,
+        skip_rag_tools=True,
     )
 
     # Prepare input for Responses API
@@ -420,6 +426,18 @@ async def retrieve_response(  # pylint: disable=too-many-locals,too-many-branche
             TokenCounter(),
         )
 
+    # Extract RAG chunks from vector DB query response BEFORE calling responses API
+    _, _, doc_ids_from_chunks, rag_chunks = await perform_vector_search(
+        client, query_request, configuration
+    )
+
+    # Format RAG context for injection into user message
+    rag_context = format_rag_context_for_injection(rag_chunks)
+
+    # Inject RAG context into input text
+    if rag_context:
+        input_text = input_text + rag_context
+
     # Create OpenAI response using responses API
     create_kwargs: dict[str, Any] = {
         "input": input_text,
@@ -444,18 +462,29 @@ async def retrieve_response(  # pylint: disable=too-many-locals,too-many-branche
     llm_response = ""
     tool_calls: list[ToolCallSummary] = []
     tool_results: list[ToolResultSummary] = []
-    rag_chunks: list[RAGChunk] = []
+    response_api_rag_chunks: list[RAGChunk] = []
     for output_item in response.output:
         message_text = extract_text_from_response_output_item(output_item)
         if message_text:
             llm_response += message_text
 
-        tool_call, tool_result = _build_tool_call_summary(output_item, rag_chunks)
+        tool_call, tool_result = _build_tool_call_summary(
+            output_item, response_api_rag_chunks
+        )
         if tool_call:
             tool_calls.append(tool_call)
         if tool_result:
             tool_results.append(tool_result)
 
+    # Merge RAG chunks from direct vector query with those from responses API
+    all_rag_chunks = rag_chunks + response_api_rag_chunks
+    logger.info(
+        "Combined RAG chunks: %d from direct query + %d from responses API = %d total",
+        len(rag_chunks),
+        len(response_api_rag_chunks),
+        len(all_rag_chunks),
+    )
+
     logger.info(
         "Response processing complete - Tool calls: %d, Response length: %d chars",
         len(tool_calls),
@@ -466,11 +495,21 @@ async def retrieve_response(  # pylint: disable=too-many-locals,too-many-branche
         llm_response=llm_response,
         tool_calls=tool_calls,
         tool_results=tool_results,
-        rag_chunks=rag_chunks,
+        rag_chunks=all_rag_chunks,
     )
 
     # Extract referenced documents and token usage from Responses API response
-    referenced_documents = parse_referenced_documents_from_responses_api(response)
+    # Merge with documents from direct vector query
+    response_referenced_documents = parse_referenced_documents_from_responses_api(
+        response
+    )
+    all_referenced_documents = doc_ids_from_chunks + response_referenced_documents
+    logger.info(
+        "Combined referenced documents: %d from direct query + %d from responses API = %d total",
+        len(doc_ids_from_chunks),
+        len(response_referenced_documents),
+        len(all_referenced_documents),
+    )
     model_label = model_id.split("/", 1)[1] if "/" in model_id else model_id
     token_usage = extract_token_usage_from_responses_api(
         response, model_label, provider_id, system_prompt
@@ -485,7 +524,7 @@ async def retrieve_response(  # pylint: disable=too-many-locals,too-many-branche
     return (
         summary,
         normalize_conversation_id(conversation_id),
-        referenced_documents,
+        all_referenced_documents,
         token_usage,
     )
 
@@ -687,12 +726,15 @@ def _increment_llm_call_metric(provider: str, model: str) -> None:
         logger.warning("Failed to update LLM call metric: %s", e)
 
 
-def get_rag_tools(vector_store_ids: list[str]) -> Optional[list[dict[str, Any]]]:
+def get_rag_tools(
+    vector_store_ids: list[str], solr_params: Optional[dict[str, Any]] = None
+) -> Optional[list[dict[str, Any]]]:
     """
     Convert vector store IDs to tools format for Responses API.
 
     Args:
         vector_store_ids: List of vector store identifiers
+        solr_params: Optional Solr filtering parameters
 
     Returns:
         Optional[list[dict[str, Any]]]: List containing file_search tool configuration,
@@ -701,13 +743,16 @@ def get_rag_tools(vector_store_ids: list[str]) -> Optional[list[dict[str, Any]]]
     if not vector_store_ids:
         return None
 
-    return [
-        {
-            "type": "file_search",
-            "vector_store_ids": vector_store_ids,
-            "max_num_results": 10,
-        }
-    ]
+    tool_config = {
+        "type": "file_search",
+        "vector_store_ids": vector_store_ids,
+        "max_num_results": 10,
+    }
+
+    if solr_params:
+        tool_config["solr"] = solr_params
+
+    return [tool_config]
 
 
 def get_mcp_tools(
@@ -808,7 +853,9 @@ async def prepare_tools_for_responses_api(
     query_request: QueryRequest,
     token: str,
     config: AppConfig,
+    *,
     mcp_headers: Optional[dict[str, dict[str, str]]] = None,
+    skip_rag_tools: bool = False,
 ) -> Optional[list[dict[str, Any]]]:
     """
     Prepare tools for Responses API including RAG and MCP tools.
@@ -822,6 +869,7 @@ async def prepare_tools_for_responses_api(
         token: Authentication token for MCP tools
         config: Configuration object containing MCP server settings
         mcp_headers: Per-request headers for MCP servers
+        skip_rag_tools: If True, skip adding RAG tools (used when doing direct vector querying)
 
     Returns:
         Optional[list[dict[str, Any]]]: List of tool configurations for the
@@ -831,18 +879,32 @@ async def prepare_tools_for_responses_api(
         return None
 
     toolgroups = []
-    # Get vector stores for RAG tools - use specified ones or fetch all
-    if query_request.vector_store_ids:
-        vector_store_ids = query_request.vector_store_ids
-    else:
-        vector_store_ids = [
-            vector_store.id for vector_store in (await client.vector_stores.list()).data
-        ]
 
-    # Add RAG tools if vector stores are available
-    rag_tools = get_rag_tools(vector_store_ids)
-    if rag_tools:
-        toolgroups.extend(rag_tools)
+    # Add RAG tools if not skipped
+    if not skip_rag_tools:
+        # Get vector stores for RAG tools - use specified ones or fetch all
+        if query_request.vector_store_ids:
+            vector_store_ids = query_request.vector_store_ids
+            logger.info("Using specified vector_store_ids: %s", vector_store_ids)
+        else:
+            vector_store_ids = [
+                vector_store.id
+                for vector_store in (await client.vector_stores.list()).data
+            ]
+            logger.info("Using all available vector_store_ids: %s", vector_store_ids)
+
+        # Add RAG tools if vector stores are available
+        if vector_store_ids:
+            rag_tools = get_rag_tools(vector_store_ids)
+            if rag_tools:
+                logger.info("rag_tool are: %s", rag_tools)
+                toolgroups.extend(rag_tools)
+            else:
+                logger.info("No RAG tools configured")
+        else:
+            logger.info("No vector stores available for RAG tools")
+    else:
+        logger.info("Skipping RAG tools - using direct vector querying instead")
 
     # Add MCP server tools
     mcp_tools = get_mcp_tools(config.mcp_servers, token, mcp_headers)

src/app/endpoints/streaming_query.py

@@ -59,7 +59,6 @@
 logger = logging.getLogger("app.endpoints.handlers")
 router = APIRouter(tags=["streaming_query"])
 
-
 streaming_query_responses: dict[int | str, dict[str, Any]] = {
     200: StreamingQueryResponse.openapi_response(),
     401: UnauthorizedResponse.openapi_response(
@@ -662,6 +661,7 @@ async def streaming_query_endpoint_handler_base(  # pylint: disable=too-many-loc
             token,
             mcp_headers=mcp_headers,
         )
+
         metadata_map: dict[str, dict[str, Any]] = {}
 
         # Create context object for response generator