init test

.gitignore

@@ -0,0 +1,2 @@
+.env
+*.DS_Store

Dockerfile

@@ -0,0 +1,23 @@
+# -------- base image --------
+FROM python:3.11-slim
+
+ENV PYTHONUNBUFFERED=1 \
+    OMP_NUM_THREADS=1 \
+    TOKENIZERS_PARALLELISM=false 
+    #GRADIO_MCP_SERVER=True         
+
+# -------- install deps --------
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# -------- copy source --------
+COPY app ./app
+COPY params.cfg .
+COPY .env* ./
+
+# Ports:
+#  • 7860 → Gradio UI (HF Spaces standard)
+EXPOSE 7860
+
+CMD ["python", "-m", "app.main"]

README.md

@@ -7,4 +7,4 @@ sdk: docker
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+

app/main.py

@@ -0,0 +1,84 @@
+import gradio as gr
+from .retriever import retrieve_context
+
+# ---------------------------------------------------------------------
+# Gradio Interface with MCP support
+# ---------------------------------------------------------------------
+def retriever_interface(query, reports_filter="", sources_filter="", subtype_filter="", year_filter=""):
+    """
+    Wrapper function for gradio interface to handle optional filter parameters
+    """
+    # Parse filter inputs (convert empty strings to None or lists)
+    reports = [r.strip() for r in reports_filter.split(",") if r.strip()] if reports_filter else []
+    sources = sources_filter.strip() if sources_filter else None
+    subtype = subtype_filter.strip() if subtype_filter else None
+    year = [y.strip() for y in year_filter.split(",") if y.strip()] if year_filter else None
+    
+    # Call retriever function
+    results = retrieve_context(
+        query=query,
+        reports=reports,
+        sources=sources,
+        subtype=subtype,
+        year=year
+    )
+    
+    # Format results for display
+    formatted_results = []
+    for i, doc in enumerate(results, 1):
+        metadata_str = ", ".join([f"{k}: {v}" for k, v in doc.get("metadata", {}).items()])
+        formatted_results.append(f"=== Result {i} ===\nContent: {doc['page_content']}\nMetadata: {metadata_str}\n")
+    
+    return "\n".join(formatted_results)
+
+ui = gr.Interface(
+    fn=retriever_interface,
+    inputs=[
+        gr.Textbox(
+            label="Query", 
+            lines=2, 
+            placeholder="Enter your search query here",
+            info="The query to search for in the vector database"
+        ),
+        gr.Textbox(
+            label="Reports Filter (optional)", 
+            lines=1, 
+            placeholder="report1.pdf, report2.pdf",
+            info="Comma-separated list of specific report filenames to search within (leave empty for all)"
+        ),
+        gr.Textbox(
+            label="Sources Filter (optional)", 
+            lines=1, 
+            placeholder="annual_report",
+            info="Filter by document source type (leave empty for all)"
+        ),
+        gr.Textbox(
+            label="Subtype Filter (optional)", 
+            lines=1, 
+            placeholder="financial",
+            info="Filter by document subtype (leave empty for all)"
+        ),
+        gr.Textbox(
+            label="Year Filter (optional)", 
+            lines=1, 
+            placeholder="2023, 2024",
+            info="Comma-separated list of years to filter by (leave empty for all)"
+        ),
+    ],
+    outputs=gr.Textbox(
+        label="Retrieved Context",
+        lines=10,
+        show_copy_button=True
+    ),
+    title="RAG Retrieval Service UI",
+    description="Retrieves semantically similar documents from vector database. Intended for use in RAG pipelines as an MCP server.",
+)
+
+# Launch with MCP server enabled
+if __name__ == "__main__":
+    ui.launch(
+        server_name="0.0.0.0",
+        server_port=7861,  # Different port from reranker
+        mcp_server=True,
+        show_error=True
+    )

app/retriever.py

@@ -0,0 +1,174 @@
+from typing import List, Dict, Any, Optional
+from qdrant_client.http import models as rest
+from langchain.schema import Document
+from .utils import getconfig
+import logging
+
+# Load configuration
+config = getconfig("params.cfg")
+
+# Retriever settings from config
+RETRIEVER_TOP_K = int(config.get("retriever", "TOP_K"))
+SCORE_THRESHOLD = float(config.get("retriever", "SCORE_THRESHOLD"))
+
+def create_filter(
+    reports: List[str] = None,
+    sources: str = None,
+    subtype: str = None,
+    year: List[str] = None
+) -> Optional[rest.Filter]:
+    """
+    Create a Qdrant filter based on metadata criteria.
+    
+    Args:
+        reports: List of specific report filenames to filter by
+        sources: Source type to filter by
+        subtype: Document subtype to filter by
+        year: List of years to filter by
+    
+    Returns:
+        Qdrant Filter object or None if no filters specified
+    """
+    if not any([reports, sources, subtype, year]):
+        return None
+    
+    conditions = []
+    
+    if reports and len(reports) > 0:
+        logging.info(f"Defining filter for reports: {reports}")
+        conditions.append(
+            rest.FieldCondition(
+                key="metadata.filename",
+                match=rest.MatchAny(any=reports)
+            )
+        )
+    else:
+        if sources:
+            logging.info(f"Defining filter for sources: {sources}")
+            conditions.append(
+                rest.FieldCondition(
+                    key="metadata.source",
+                    match=rest.MatchValue(value=sources)
+                )
+            )
+        
+        if subtype:
+            logging.info(f"Defining filter for subtype: {subtype}")
+            conditions.append(
+                rest.FieldCondition(
+                    key="metadata.subtype",
+                    match=rest.MatchValue(value=subtype)
+                )
+            )
+        
+        if year and len(year) > 0:
+            logging.info(f"Defining filter for years: {year}")
+            conditions.append(
+                rest.FieldCondition(
+                    key="metadata.year",
+                    match=rest.MatchAny(any=year)
+                )
+            )
+    
+    if conditions:
+        return rest.Filter(must=conditions)
+    return None
+
+def get_vectorstore():
+    """
+    Initialize and return the vectorstore connection.
+    This function should be implemented based on your specific vectorstore setup.
+    
+    Returns:
+        Vectorstore instance (e.g., Qdrant, Pinecone, etc.)
+    """
+    # TODO: Implement based on your external vector database
+    # Example for Qdrant:
+    # from langchain_community.vectorstores import Qdrant
+    # from qdrant_client import QdrantClient
+    # 
+    # client = QdrantClient(
+    #     host=config.get("vectorstore", "HOST"),
+    #     port=config.get("vectorstore", "PORT"),
+    #     api_key=config.get("vectorstore", "API_KEY", fallback=None)
+    # )
+    # 
+    # vectorstore = Qdrant(
+    #     client=client,
+    #     collection_name=config.get("vectorstore", "COLLECTION_NAME"),
+    #     embeddings=your_embedding_model  # You'll need to configure this
+    # )
+    # 
+    # return vectorstore
+    
+    raise NotImplementedError("Please implement vectorstore connection based on your setup")
+
+def retrieve_context(
+    query: str,
+    reports: List[str] = None,
+    sources: str = None,
+    subtype: str = None,
+    year: List[str] = None,
+    top_k: int = None
+) -> List[Dict[str, Any]]:
+    """
+    Retrieve semantically similar documents from the vector database.
+    
+    Args:
+        query: The search query
+        reports: List of specific report filenames to search within
+        sources: Source type to filter by
+        subtype: Document subtype to filter by
+        year: List of years to filter by
+        top_k: Number of results to return (defaults to config value)
+    
+    Returns:
+        List of dictionaries with 'page_content' and 'metadata' keys
+    """
+    try:
+        # Get vectorstore instance
+        vectorstore = get_vectorstore()
+        
+        # Create metadata filter
+        filter_obj = create_filter(
+            reports=reports or [],
+            sources=sources,
+            subtype=subtype,
+            year=year or []
+        )
+        
+        # Set up search parameters
+        k = top_k or RETRIEVER_TOP_K
+        search_kwargs = {
+            "score_threshold": SCORE_THRESHOLD,
+            "k": k
+        }
+        
+        if filter_obj:
+            search_kwargs["filter"] = filter_obj
+        
+        # Create retriever
+        retriever = vectorstore.as_retriever(
+            search_type="similarity_score_threshold",
+            search_kwargs=search_kwargs
+        )
+        
+        # Perform retrieval
+        retrieved_docs: List[Document] = retriever.invoke(query)
+        
+        logging.info(f"Retrieved {len(retrieved_docs)} documents for query: {query[:50]}...")
+        
+        # Convert to dictionary format
+        results = [
+            {
+                "page_content": doc.page_content,
+                "metadata": doc.metadata
+            }
+            for doc in retrieved_docs
+        ]
+        
+        return results
+        
+    except Exception as e:
+        logging.error(f"Error during retrieval: {str(e)}")
+        raise e

app/utils.py

@@ -0,0 +1,16 @@
+import configparser
+import logging
+
+def getconfig(configfile_path: str):
+    """
+    Read the config file
+    Params
+    ----------------
+    configfile_path: file path of .cfg file
+    """
+    config = configparser.ConfigParser()
+    try:
+        config.read_file(open(configfile_path))
+        return config
+    except:
+        logging.warning("config file not found")

params.cfg

@@ -0,0 +1,14 @@
+[retriever]
+TOP_K = 10
+SCORE_THRESHOLD = 0.6
+
+[vectorstore]
+TYPE = qdrant
+HOST = localhost
+PORT = 6333
+COLLECTION_NAME = "auditqa"
+# API_KEY = your_api_key_if_needed
+
+[embeddings]
+MODEL_NAME = BAAI/bge-m3
+# DEVICE = cpu

requirements.txt

@@ -0,0 +1,5 @@
+gradio
+langchain
+langchain-community
+qdrant-client
+sentence-transformers