ingestor

params.cfg

@@ -1,5 +1,4 @@
 [hf_endpoints]
-# NOTE: The actual token should be set via the HF_TOKEN environment variable.
 embedding_endpoint_url = https://f4veaarnbmqjhve9.eu-west-1.aws.endpoints.huggingface.cloud
 reranker_endpoint_url = https://whikfgijnuog8fjv.eu-west-1.aws.endpoints.huggingface.cloud
 
@@ -8,7 +7,6 @@ reranker_endpoint_url = https://whikfgijnuog8fjv.eu-west-1.aws.endpoints.hugging
 # for native just give url
 mode = native
 url = https://de438521-e2dd-43d9-b41b-b2e18299a2c0.europe-west3-0.gcp.cloud.qdrant.io:6333
-# NOTE: The API key should be set via QDRANT_API_KEY environment variable.
 port = 443
 collection = allreports
 
@@ -25,3 +23,12 @@ INFERENCE_PROVIDER = novita
 ORGANIZATION = GIZ
 CONTEXT_META_FIELDS = filename,project_id,document_source
 TITLE_META_FIELDS = filename,page
+
+
+[ingestor]
+# Size of each text chunk in characters
+chunk_size = 700
+# Overlap between consecutive chunks in characters
+chunk_overlap = 50
+# Text separators for splitting, comma-separated (order of preference)
+separators = \n\n,\n,. ,! ,? , ,

src/components/generator/generator_orchestrator.py

@@ -205,7 +205,6 @@ class Generator:
             error_msg = "Query cannot be empty"
             return {"error": error_msg} if chatui_format else f"Error: {error_msg}"
         logger.info(f"Generating answer for query: {query[:50]}")
-        logger.info(f"CHATUI format is {chatui_format}")
 
         try:
             # 1. Process Context
@@ -248,7 +247,6 @@ class Generator:
                 yield f"Error: {error_msg}"
             return
         logger.info(f"Generating streaming answer for query: {query[:50]}")
-        logger.info(f"CHATUI format is {chatui_format}")
         if conversation_context:
             logger.info(f"Using conversation context: {len(conversation_context)} chars")
 
@@ -278,10 +276,10 @@ class Generator:
             if chatui_format and processed_results:
                 cited_numbers = parse_citations(cleaned_response)
                 cited_sources = extract_sources(processed_results, cited_numbers)
-                sources = create_sources_list(cited_sources, 
+                sources = create_sources_list(cited_sources,
                             title_metadata_fields=self.title_metadata_fields,
                             link_metadata_field=self.link_metadata_field)
-                logging.debug(f"Sorces recieved: {sources}")
+                logger.info(f"Sources received: {sources}")
                 yield {"event": "sources", "data": {"sources": sources}}
 
             # Send END event for ChatUI format

src/components/ingestor/ingestor.py

@@ -0,0 +1,130 @@
+"""
+Document Ingestor Module
+
+Processes PDF and DOCX files, extracting text and chunking for RAG pipelines.
+Adapted from the original microservice architecture for integration into ChaBo.
+"""
+import os
+import logging
+import re
+from io import BytesIO
+from typing import Tuple, Dict, Any
+
+import PyPDF2
+from docx import Document as DocxDocument
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+
+from components.utils import get_config_value, getconfig
+
+logger = logging.getLogger(__name__)
+
+
+def extract_text_from_pdf_bytes(file_content: bytes) -> Tuple[str, Dict[str, Any]]:
+    """Extract text from PDF bytes (in memory)"""
+    try:
+        pdf_reader = PyPDF2.PdfReader(BytesIO(file_content))
+        text = ""
+        metadata = {"total_pages": len(pdf_reader.pages)}
+
+        for page_num, page in enumerate(pdf_reader.pages):
+            page_text = page.extract_text()
+            text += f"\n--- Page {page_num + 1} ---\n{page_text}"
+
+        return text, metadata
+    except Exception as e:
+        logger.error(f"PDF extraction error: {str(e)}")
+        raise Exception(f"Failed to extract text from PDF: {str(e)}")
+
+
+def extract_text_from_docx_bytes(file_content: bytes) -> Tuple[str, Dict[str, Any]]:
+    """Extract text from DOCX bytes (in memory)"""
+    try:
+        doc = DocxDocument(BytesIO(file_content))
+        text = ""
+        metadata = {"total_paragraphs": 0}
+
+        for paragraph in doc.paragraphs:
+            if paragraph.text.strip():
+                text += f"{paragraph.text}\n"
+                metadata["total_paragraphs"] += 1
+
+        return text, metadata
+    except Exception as e:
+        logger.error(f"DOCX extraction error: {str(e)}")
+        raise Exception(f"Failed to extract text from DOCX: {str(e)}")
+
+
+def clean_and_chunk_text(text: str, config) -> str:
+    """Clean text and split into chunks, returning formatted context"""
+    # Basic text cleaning
+    text = re.sub(r'\n+', '\n', text)
+    text = re.sub(r'\s+', ' ', text)
+    text = text.strip()
+
+    # Get chunking parameters from config
+    chunk_size = config.getint('ingestor', 'chunk_size', fallback=700)
+    chunk_overlap = config.getint('ingestor', 'chunk_overlap', fallback=50)
+    separators_str = config.get('ingestor', 'separators', fallback=r'\n\n,\n,. ,! ,? , ,')
+    separators = [s.strip() for s in separators_str.split(',')]
+
+    # Split text into chunks
+    text_splitter = RecursiveCharacterTextSplitter(
+        chunk_size=chunk_size,
+        chunk_overlap=chunk_overlap,
+        length_function=len,
+        separators=separators,
+        is_separator_regex=False
+    )
+
+    chunks = text_splitter.split_text(text)
+
+    # Create formatted context with chunk markers
+    context_parts = []
+    for i, chunk_text in enumerate(chunks):
+        context_parts.append(f"[Chunk {i+1}]: {chunk_text}")
+
+    return "\n\n".join(context_parts)
+
+
+def process_document(file_content: bytes, filename: str) -> str:
+    """
+    Main document processing function - processes file and returns chunked context.
+
+    Args:
+        file_content: Raw bytes of the uploaded file
+        filename: Name of the file (used to determine file type)
+
+    Returns:
+        Formatted chunked context string ready for RAG pipeline
+
+    Raises:
+        ValueError: If file type is unsupported
+        Exception: If processing fails
+    """
+    try:
+        # Load config
+        config = getconfig("params.cfg")
+
+        # Extract text based on file type (in memory)
+        file_extension = os.path.splitext(filename)[1].lower()
+
+        if file_extension == '.pdf':
+            text, extraction_metadata = extract_text_from_pdf_bytes(file_content)
+        elif file_extension == '.docx':
+            text, extraction_metadata = extract_text_from_docx_bytes(file_content)
+        else:
+            raise ValueError(f"Unsupported file type: {file_extension}")
+
+        # Clean and chunk text
+        context = clean_and_chunk_text(text, config)
+
+        logger.info(
+            f"Successfully processed document {filename}: "
+            f"{len(text)} characters, {extraction_metadata}"
+        )
+
+        return context
+
+    except Exception as e:
+        logger.error(f"Document processing failed for {filename}: {str(e)}")
+        raise Exception(f"Processing failed: {str(e)}")

src/components/orchestration/nodes.py

@@ -2,14 +2,15 @@
 LangGraph orchestration nodes for retrieval and generation
 
 NEEDS TO BE UPDATED
-""" 
+"""
 import logging
 logger = logging.getLogger(__name__)
 from datetime import datetime
 import json
-from typing import TYPE_CHECKING, Dict, Any, List
+from typing import TYPE_CHECKING
 from langchain_core.documents import Document
 from .telemetry import extract_retriever_telemetry
+from components.ingestor.ingestor_main import process_document
 
 # Assuming these Type definitions are available from state.py and retriever_orchestrator.py
 if TYPE_CHECKING:
@@ -90,6 +91,16 @@ async def generate_node_streaming(state: "GraphState", generator: "Generator", *
     query = state.get("query")
     raw_docs = state.get("raw_documents", [])
     metadata = state.get("metadata", {})
+    ingestor_context = state.get("ingestor_context")
+
+    # If we have ingestor_context, prepend it to raw_docs as a Document
+    if ingestor_context:
+        ingestor_doc = Document(
+            page_content=ingestor_context,
+            metadata={"source": "uploaded_file", "filename": state.get("filename", "unknown")}
+        )
+        raw_docs = [ingestor_doc] + raw_docs
+        logger.info(f"Including ingestor context ({len(ingestor_context)} chars) with retrieved docs")
 
     accumulated_text = ""
     logger.info(f"Generation: {query[:50]}... ({len(raw_docs)} docs)")
@@ -118,7 +129,6 @@ async def generate_node_streaming(state: "GraphState", generator: "Generator", *
         })
 
         logger.info(f"Streaming complete in {duration:.2f}s. Length: {len(accumulated_text)}")
-        logger.debug(f"Final answer: {accumulated_text}")
 
     except Exception as e:
         duration = (datetime.now() - start_time).total_seconds()
@@ -131,6 +141,57 @@ async def generate_node_streaming(state: "GraphState", generator: "Generator", *
         writer({"event": "error", "data": {"error": str(e)}})
 
 
+async def ingest_node(state: 'GraphState') -> 'GraphState':
+    """
+    Node to process uploaded documents (PDF, DOCX) and extract chunked context.
+    Only runs if file_content and filename are present in state.
+    """
+    start_time = datetime.now()
+
+    file_content = state.get("file_content")
+    filename = state.get("filename")
+    metadata = state.get("metadata", {})
+
+    # Skip if no file uploaded
+    if not file_content or not filename:
+        logger.info("No file to ingest, skipping ingest_node")
+        return {}
+
+    logger.info(f"Ingesting document: {filename}")
+
+    try:
+        # Process document and get chunked context
+        ingestor_context = process_document(file_content, filename)
+
+        duration = (datetime.now() - start_time).total_seconds()
+
+        metadata.update({
+            "ingest_duration": duration,
+            "ingest_success": True,
+            "ingested_filename": filename,
+            "ingestor_context_length": len(ingestor_context)
+        })
+
+        logger.info(f"Document ingested successfully in {duration:.2f}s")
+
+        return {
+            "ingestor_context": ingestor_context,
+            "metadata": metadata
+        }
+
+    except Exception as e:
+        duration = (datetime.now() - start_time).total_seconds()
+        logger.error(f"Document ingestion failed: {str(e)}", exc_info=True)
+
+        metadata.update({
+            "ingest_duration": duration,
+            "ingest_success": False,
+            "ingest_error": str(e)
+        })
+
+        return {"ingestor_context": "", "metadata": metadata}
+
+
 
 # from .state import GraphState
 

src/components/orchestration/ui_adapters.py

@@ -18,7 +18,9 @@ async def process_query_streaming(
     sources_filter: str = "",
     subtype_filter: str = "",
     year_filter: str = "",
-    conversation_context: str = None
+    conversation_context: str = None,
+    file_content: bytes = None,
+    filename: str = None
 ):
     """
     Process a query through the LangGraph workflow with streaming.
@@ -46,6 +48,11 @@ async def process_query_streaming(
         "metadata_filters": metadata_filters
     }
 
+    # Add file content if present
+    if file_content and filename:
+        initial_state["file_content"] = file_content
+        initial_state["filename"] = filename
+
     try:
         async for output in compiled_graph.astream(initial_state, stream_mode="custom"):
             if output.get("event") == "data":
@@ -65,12 +72,7 @@ async def process_query_streaming(
 
 async def chatui_adapter(data, compiled_graph, max_turns: int = 3, max_chars: int = 8000):
     """Text-only adapter for ChatUI with structured message support"""
-    logger.info("=" * 80)
-    logger.info("CHATUI_ADAPTER CALLED - Request reached adapter function")
-    logger.info(f"Data type: {type(data)}")
-    logger.info(f"Data repr: {repr(data)[:500]}")
-    logger.info(f"Data dict: {data.__dict__ if hasattr(data, '__dict__') else 'N/A'}")
-    logger.info("=" * 80)
+    logger.debug(f"ChatUI adapter called with data type: {type(data)}")
 
     try:
         # Handle both dict and object access patterns
@@ -83,10 +85,6 @@ async def chatui_adapter(data, compiled_graph, max_turns: int = 3, max_chars: in
             messages_value = getattr(data, 'messages', None)
             preprompt_value = getattr(data, 'preprompt', None)
 
-        logger.info(f"Extracted - text: {text_value[:100] if text_value else 'None'}")
-        logger.info(f"Extracted - messages count: {len(messages_value) if messages_value else 0}")
-        logger.info(f"Extracted - preprompt: {preprompt_value[:100] if preprompt_value else 'None'}")
-
         # Convert dict messages to objects if needed
         messages = []
         if messages_value:
@@ -99,19 +97,21 @@ async def chatui_adapter(data, compiled_graph, max_turns: int = 3, max_chars: in
                 else:
                     messages.append(msg)
 
-        logger.info(f"Context: {messages}")
         # Extract latest user query
         user_messages = [msg for msg in messages if msg.role == 'user']
         query = user_messages[-1].content if user_messages else text_value
 
-        # Log conversation context
-        logger.info(f"Processing query: {query[:50] if query else 'empty'}...")
-        logger.info(f"Total messages in conversation: {len(messages)}")
-        logger.info(f"User messages: {len(user_messages)}, Assistant messages: {len([m for m in messages if m.role == 'assistant'])}")
+        # Conversation metadata (troubleshooting purposes)
+        msg_metadata = {
+            'total': len(messages),
+            'user': len(user_messages),
+            'assistant': len([m for m in messages if m.role == 'assistant']),
+            'msg_lengths': [len(m.content) for m in messages]
+        }
+        logger.info(f"Processing query: {query[:20]}... | Conversation: {msg_metadata}")
 
         # Build conversation context for generation (last N turns)
         conversation_context = build_conversation_context(messages, max_turns=max_turns, max_chars=max_chars)
-        logger.info(f"Context: {conversation_context}")
         full_response = ""
         sources_collected = None
 
@@ -173,8 +173,6 @@ async def chatui_file_adapter(data, compiled_graph, max_turns: int = 3, max_char
         # Extract query - prefer structured messages
         conversation_context = None
         if messages_value and len(messages_value) > 0:
-            logger.info("Using structured messages")
-
             # Convert dict messages to objects
             messages = []
             for msg in messages_value:
@@ -189,9 +187,14 @@ async def chatui_file_adapter(data, compiled_graph, max_turns: int = 3, max_char
             user_messages = [msg for msg in messages if msg.role == 'user']
             query = user_messages[-1].content if user_messages else text_value
 
-            logger.info(f"Processing query: {query[:50] if query else 'empty'}...")
-            logger.info(f"Total messages in conversation: {len(messages)}")
-            logger.info(f"User messages: {len(user_messages)}, Assistant messages: {len([m for m in messages if m.role == 'assistant'])}")
+            # Conversation metadata (troubleshooting purposes)
+            msg_metadata = {
+                'total': len(messages),
+                'user': len(user_messages),
+                'assistant': len([m for m in messages if m.role == 'assistant']),
+                'msg_lengths': [len(m.content) for m in messages]
+            }
+            logger.info(f"Processing query with file: {query[:20]}... | Conversation: {msg_metadata}")
 
             conversation_context = build_conversation_context(messages, max_turns=max_turns, max_chars=max_chars)
         else:
@@ -219,12 +222,14 @@ async def chatui_file_adapter(data, compiled_graph, max_turns: int = 3, max_char
         async for result in process_query_streaming(
             compiled_graph=compiled_graph,
             query=query,
-            file_upload=None,  # File handling can be extended
+            file_upload=None,
             reports_filter="",
             sources_filter="",
             subtype_filter="",
             year_filter="",
-            conversation_context=conversation_context
+            conversation_context=conversation_context,
+            file_content=file_content,
+            filename=filename
         ):
             if isinstance(result, dict):
                 result_type = result.get("type", "data")

src/components/orchestration/workflow.py

@@ -6,7 +6,7 @@ from functools import partial
 from langgraph.graph import StateGraph, START, END
 
 from .state import GraphState
-from .nodes import retrieve_node, generate_node_streaming
+from .nodes import retrieve_node, generate_node_streaming, ingest_node
 
 logger = logging.getLogger(__name__)
 
@@ -17,20 +17,22 @@ def build_workflow(retriever_instance, generator_instance):
     """
     workflow = StateGraph(GraphState)
 
-    # Inject services into nodes
+    # Inject services into nodes (ingest_node doesn't need dependency injection)
     r_node = partial(retrieve_node, retriever=retriever_instance)
     g_node = partial(generate_node_streaming, generator=generator_instance)
 
     # Add nodes
+    workflow.add_node("ingest", ingest_node)
     workflow.add_node("retrieve", r_node)
     workflow.add_node("generate", g_node)
 
-    # Define edges
-    workflow.add_edge(START, "retrieve")
+    # Define edges: ingest -> retrieve -> generate
+    workflow.add_edge(START, "ingest")
+    workflow.add_edge("ingest", "retrieve")
     workflow.add_edge("retrieve", "generate")
     workflow.add_edge("generate", END)
 
     compiled_graph = workflow.compile()
 
-    logger.info("LangGraph workflow compiled successfully")
+    logger.info("LangGraph workflow compiled successfully with ingest node")
     return compiled_graph

src/components/retriever/retriever_orchestrator.py

@@ -110,7 +110,7 @@ class ChaBoHFEndpointRetriever(BaseRetriever):
             client = self._get_qdrant_client()
             
             if self.qdrant_mode.lower() == 'native':
-                logger.info(f"Performing Sync Native Qdrant search. Collection: {self.qdrant_collection}, TOP_K:{self.initial_k}")
+                logger.debug(f"Sync Native Qdrant search: collection={self.qdrant_collection}, k={self.initial_k}")
                 search_result = client.query_points(
                     collection_name=self.qdrant_collection,
                     query=query_vector,
@@ -131,7 +131,7 @@ class ChaBoHFEndpointRetriever(BaseRetriever):
                 } for hit in search_result.points]
 
             elif self.qdrant_mode.lower() == 'gradio':
-                logger.info(f"Performing Sync Gradio Qdrant search. Collection: {self.qdrant_collection}, TOP_K:{self.initial_k}")
+                logger.debug(f"Sync Gradio Qdrant search: collection={self.qdrant_collection}, k={self.initial_k}")
                 return client.predict(
                     query_vector_json=query_vector, 
                     collection_name=self.qdrant_collection,
@@ -152,8 +152,8 @@ class ChaBoHFEndpointRetriever(BaseRetriever):
             client = await self._aget_qdrant_client()
 
             if self.qdrant_mode.lower() == 'native':
-                logger.info(f"Performing Async Qdrant search. Collection: {self.qdrant_collection}, TOP_K:{self.initial_k}")
-                
+                logger.debug(f"Async Native Qdrant search: collection={self.qdrant_collection}, k={self.initial_k}")
+
                 search_result = await client.query_points(
                     collection_name=self.qdrant_collection,
                     query=query_vector,
@@ -170,7 +170,7 @@ class ChaBoHFEndpointRetriever(BaseRetriever):
                 } for hit in search_result.points]
 
             elif self.qdrant_mode.lower() == 'gradio':
-                logger.info(f"Performing Async Gradio Qdrant search. Collection: {self.qdrant_collection}, TOP_K:{self.initial_k}")
+                logger.debug(f"Async Gradio Qdrant search: collection={self.qdrant_collection}, k={self.initial_k}")
                 loop = asyncio.get_running_loop()
             
                 # Use run_in_executor to make the synchronous .predict() awaitable
@@ -344,7 +344,12 @@ def create_retriever_from_config(config_file: str = "params.cfg"):
     config = getconfig(config_file)
 
     hf_token = os.getenv("HF_TOKEN")
+    if not hf_token:
+        raise ValueError("HF_TOKEN environment variable is required but not set")
+
     qdrant_api_key = os.getenv("QDRANT_API_KEY")
+    if not qdrant_api_key:
+        raise ValueError("QDRANT_API_KEY environment variable is required but not set")
 
     config_map = {
         "embedding_endpoint_url": ("hf_endpoints", "embedding_endpoint_url", "EMBEDDING_ENDPOINT_URL"),
@@ -361,17 +366,19 @@ def create_retriever_from_config(config_file: str = "params.cfg"):
 
     retriever_config_kwargs = {
         "hf_token": hf_token,
-        "qdrant_api_key": qdrant_api_key,
+        "qdrant_api_key": qdrant_api_key
     }
 
     for key, params in config_map.items():
+
         section, option, env_var = params[:3]
         fallback = params[3] if len(params) > 3 else None
         value = get_config_value(config, section, option, env_var, fallback)
-
+        
         if key in ['initial_k', 'top_k', 'qdrant_port']:
             value = int(value)
 
+            
         retriever_config_kwargs[key] = value
 
     logger.info(f"Configuration loaded. Qdrant Mode: {retriever_config_kwargs['qdrant_mode']}")

src/main.py

@@ -84,12 +84,8 @@ add_routes(
     enable_public_trace_link_endpoint=True,
 )
 
-# Log schema info
-logger.info("=" * 80)
-logger.info("LangServe Routes Configured")
-logger.info(f"ChatUIInput schema: {ChatUIInput.model_json_schema()}")
-logger.info(f"ChatUIFileInput schema: {ChatUIFileInput.model_json_schema()}")
-logger.info("=" * 80)
+logger.debug(f"ChatUIInput schema: {ChatUIInput.model_json_schema()}")
+logger.debug(f"ChatUIFileInput schema: {ChatUIFileInput.model_json_schema()}")
 
 #----------------------------------------