init

.gitignore

+.env

.idea/.gitignore

+# Default ignored files
+/shelf/
+/workspace.xml
+# Editor-based HTTP Client requests
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

.idea/PythonProject.iml

+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$">
+      <excludeFolder url="file://$MODULE_DIR$/.venv" />
+    </content>
+    <orderEntry type="jdk" jdkName="Python 3.12 (PythonProject)" jdkType="Python SDK" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>
\ No newline at end of file

.idea/inspectionProfiles/profiles_settings.xml

+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>
\ No newline at end of file

.idea/misc.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Black">
+    <option name="sdkName" value="Python 3.12 (PythonProject)" />
+  </component>
+</project>
\ No newline at end of file

.idea/modules.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/PythonProject.iml" filepath="$PROJECT_DIR$/.idea/PythonProject.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

.idea/vcs.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>
\ No newline at end of file

agents/agent_decision.py

+import os
+from typing import Dict, Optional, Union, List, TypedDict, Any, Literal
+from dotenv import load_dotenv
+from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.output_parsers import JsonOutputParser
+from langgraph.graph import MessagesState, StateGraph, END
+import json
+from config import Config
+from agents.rag_agent import ResearchAssistantRAG
+from langgraph.checkpoint.memory import MemorySaver
+
+load_dotenv()
+
+config = Config()
+
+memory = MemorySaver()
+
+thread_config = {"configurable": {"thread_id": "1"}}
+
+class AgentConfig:
+    DECISION_MODEL = "gpt-4o"
+    CONFIDENCE_THRESHOLD = 0.85
+
+    DECISION_SYSTEM_PROMPT = """
+    Du bist ein intelligentes akademisches Routingsystem, das Benutzeranfragen an den jeweils passenden spezialisierten Agenten weiterleitet. 
+    Deine Aufgabe ist es, die Anfrage des Benutzers zu analysieren und zu bestimmen, welcher Agent am besten geeignet ist – 
+    basierend auf dem Inhalt der Anfrage und dem Gesprächskontext.
+    
+    Verfügbare Agenten:
+    1. CONVERSATION_AGENT – Für allgemeine Konversation, Begrüßungen oder nicht-akademische Fragen.
+    2. KEYWORD_SEARCH_AGENT – Für gezielte Stichwortsuchen innerhalb der Dokumentensammlung.
+    3. RAG_AGENT – Für Fragen zu spezifischen akademischen Inhalten, die aus den bereitgestellten Dokumenten beantwortet werden können.
+    4. SUMMARIZATION_AGENT – Für Zusammenfassungen von Dokumenten oder bestimmten Abschnitten.
+    5. CLASSIFICATION_AGENT – Für die Erkennung des Dokumententyps (z. B. Bachelorarbeit, Artikel) und der Hauptthemen (z. B. Informatik, Medizin).
+    6. USER_MEMORY_AGENT – Interner Agent zur Verwaltung von Benutzerhistorie und Personalisierung.
+    
+    Triff deine Entscheidung anhand der folgenden Richtlinien:
+    - Wenn die Anfrage eine Begrüßung, allgemeine Frage oder unklare Formulierung enthält → wähle CONVERSATION_AGENT.
+    - Wenn die Anfrage eine gezielte Suche nach einem Begriff oder Schlüsselwort ist → wähle KEYWORD_SEARCH_AGENT.
+    - Wenn der Benutzer eine inhaltliche Frage zu einem Thema aus den Dokumenten stellt → wähle RAG_AGENT.
+    - Wenn eine Zusammenfassung eines Dokuments oder Abschnitts gewünscht ist → wähle SUMMARIZATION_AGENT.
+    - Wenn nach Art oder Thema eines Dokuments gefragt wird → wähle CLASSIFICATION_AGENT.
+    - Wenn die Absicht nicht eindeutig ist, wähle sicherheitshalber CONVERSATION_AGENT.
+    
+    Du musst deine Antwort im JSON-Format mit folgender Struktur liefern:
+    {
+      "agent": "AGENT_NAME",
+      "reasoning": "Schrittweise Begründung für die Wahl dieses Agenten",
+      "confidence": 0.95  // Wert zwischen 0.0 und 1.0, der das Vertrauen in die Entscheidung angibt
+    }
+    """
+
+class AgentDecision(TypedDict):
+    agent: Literal[
+        "CONVERSATION_AGENT",
+        "KEYWORD_SEARCH_AGENT",
+        "RAG_AGENT",
+        "SUMMARIZATION_AGENT",
+        "CLASSIFICATION_AGENT",
+        "USER_MEMORY_AGENT"
+    ]
+    reasoning: str
+    confidence: float
+
+class AgentState(MessagesState):
+    """State maintained across the workflow."""
+    messages: List[BaseMessage]
+    agent_name: Optional[str]
+    current_input: Optional[Union[str, Dict]]
+    has_image: bool
+    image_type: Optional[str]
+    output: Optional[str]
+    needs_human_validation: bool
+    retrieval_confidence: float
+    bypass_routing: bool
+    insufficient_info: bool
+
+
+# class AgentDecision(TypedDict):
+#     agent: str
+#     reasoning: str
+#     confidence: float
+
+
+def create_agent_graph():
+    def run_rag_agent(state: AgentState) -> AgentState:
+        print(f"Selected agent: RAG_AGENT")
+
+        rag_agent = ResearchAssistantRAG(config)
+
+        messages: List[BaseMessage] = state["messages"]
+        query: Union[str, Dict] = state["current_input"]
+        rag_context_limit: int = config.rag.context_limit
+
+        recent_context: str = ""
+        for msg in messages[-rag_context_limit:]:
+            if isinstance(msg, HumanMessage):
+                # print("######### DEBUG 1:", msg)
+                recent_context += f"User: {msg.content}\n"
+            elif isinstance(msg, AIMessage):
+                # print("######### DEBUG 2:", msg)
+                recent_context += f"Assistant: {msg.content}\n"
+
+        # Call pipeline RAG
+        response: Dict[str, Any] = rag_agent.process_query(query, chat_history=recent_context)
+
+        ### CHECKPOINT
+        print("Final response from RAG agent:", response)
+
+        retrieval_confidence: float = response.get("confidence", 0.0)
+
+        print(f"Retrieval Confidence: {retrieval_confidence}")
+        print(f"Sources: {len(response['sources'])}")
+
+        insufficient_info = False
+        response_content = response["response"]
+
+        if hasattr(response_content, 'content'):
+            response_text = response_content.content
+        else:
+            response_text = response_content
+
+        print(f"Response text type: {type(response_text)}")
+        print(f"Response text preview: {response_text[:100]}...")
+
+        if isinstance(response_text, str) and (
+                "I don't have enough information to answer this question based on the provided context" in response_text or
+                "I don't have enough information" in response_text or
+                "don't have enough information" in response_text.lower() or
+                "not enough information" in response_text.lower() or
+                "insufficient information" in response_text.lower() or
+                "cannot answer" in response_text.lower() or
+                "unable to answer" in response_text.lower()
+        ):
+            print("RAG response indicates insufficient information")
+            print(f"Response text that triggered insufficient_info: {response_text[:100]}...")
+            insufficient_info = True
+
+        print(f"Insufficient info flag set to: {insufficient_info}")
+
+        # Store RAG output ONLY if confidence is high
+        if retrieval_confidence >= config.rag.min_retrieval_confidence:
+            temp_output = response["response"]
+        else:
+            temp_output = response["response"]
+
+        return {
+            **state,
+            "output": temp_output,
+            "needs_human_validation": False,
+            "retrieval_confidence": retrieval_confidence,
+            "agent_name": "RAG_AGENT",
+            "insufficient_info": insufficient_info
+        }
+
+    graph = StateGraph(AgentState)
+
+    graph.add_node("RAG_AGENT", run_rag_agent)
+
+    graph.set_entry_point("RAG_AGENT")
+
+    graph.add_edge("RAG_AGENT", END)
+
+    return graph.compile(checkpointer=memory)
+
+
+def init_agent_state() -> AgentState:
+    """Initialize the agent state with default values."""
+    return {
+        "messages": [],
+        "agent_name": None,
+        "current_input": None,
+        "has_image": False,
+        "image_type": None,
+        "output": None,
+        "needs_human_validation": False,
+        "retrieval_confidence": 0.0,
+        "bypass_routing": True,
+        "insufficient_info": False
+    }
+
+
+def process_query(query: Union[str, Dict], conversation_history: List[BaseMessage] = None) -> Dict[str, Any]:
+    graph = create_agent_graph()
+
+    state = init_agent_state()
+    # if conversation_history:
+    #     state["messages"] = conversation_history
+
+    input_text = query if isinstance(query, str) else query.get("text", "")
+
+    state["current_input"] = query
+    state["messages"] = [HumanMessage(content=input_text)]
+
+    # result = graph.invoke(state, thread_config)
+    state = graph.invoke(state, thread_config)
+
+    output = state.get("output")
+    if hasattr(output, "content"):
+        response_text = output.content
+    else:
+        response_text = str(output)
+    # # Keep history to reasonable size (ANOTHER OPTION: summarize and store before truncating history)
+    # if len(result["messages"]) > config.max_conversation_history:  # Keep last config.max_conversation_history messages
+    #     result["messages"] = result["messages"][-config.max_conversation_history:]
+    #
+    # # visualize conversation history in console
+    # for m in result["messages"]:
+    #     m.pretty_print()
+
+    return {
+        "response": response_text,
+        "agent_name": state.get("agent_name"),
+        "messages": state.get("messages", [])
+    }
\ No newline at end of file

agents/rag_agent/__init__.py

+import logging
+import time
+from typing import List, Dict, Any, Optional
+import uuid
+from pathlib import Path
+import json
+from .vector_store import QdrantRetriever
+from .document_processor import DocumentProcessor
+from .query_processor import QueryProcessor
+from .response_generator import ResponseGenerator
+from .data_ingestion import DataIngestion
+
+class ResearchAssistantRAG:
+
+    def __init__(self, config):
+        self.config = config
+        self._initialize()
+
+
+    def _initialize(self):
+        try:
+            self.logger = logging.getLogger(f"{self.__module__}")
+            self.logger.info("Initializing Research Assistant RAG system")
+
+            self.llm = self.config.rag.llm
+            self.logger.info(f"Using LLM: {type(self.llm).__name__}")
+
+            self.embedding_model = self.config.rag.embedding_model
+            self.logger.info(f"Using embedding model: {type(self.embedding_model).__name__}")
+
+            self.query_processor = QueryProcessor(self.config, self.embedding_model)
+
+            self.document_processor = DocumentProcessor(self.config, self.embedding_model)
+
+            self.retriever = QdrantRetriever(self.config)
+
+            self.response_generator = ResponseGenerator(self.config, self.llm)
+
+            total_docs = self.retriever.count_documents()
+            self.logger.info(f"Vector store contains {total_docs} documents")
+            if total_docs == 0:
+                self.logger.warning("No documents in vector store. Results may be limited.")
+
+            self.top_k = getattr(self.config.rag, "top_k", 5)
+            self.similarity_threshold = getattr(self.config.rag, "similarity_threshold", 0.0)
+
+            self.data_ingestion = DataIngestion()
+
+            self.logger.info("Research Assistant RAG system successfully initialized")
+
+        except Exception as e:
+            self.logger.error(f"Initialization error: {e}")
+            import traceback
+            self.logger.error(traceback.format_exc())
+            raise
+
+
+    def process_query(self, query: str, chat_history: Optional[List[Dict[str, str]]] = None) -> Dict[str, Any]:
+        self.logger.info(f"RAG Agent processing query: {query}")
+
+        result = self.query(query, chat_history)
+
+        return result
+
+
+    def ingest_documents(self, documents: List[Dict[str, Any]]) -> Dict[str, Any]:
+        self.logger.info(f"Ingesting {len(documents)} documents")
+        start_time = time.time()
+
+        print(f"documents: {documents}")
+
+        try:
+            processed_dir = Path("data/processed")
+            processed_dir.mkdir(exist_ok=True, parents=True)
+            all_chunks: List[Dict[str, Any]] = []
+            chunk_ids: List[str] = []
+
+            for document in documents:
+                content = document.get("content", "")
+                metadata = document.get("metadata", {})
+
+                if "id" not in metadata:
+                    metadata["id"] = str(uuid.uuid4())
+
+                chunks_from_file = self.document_processor.process_document(content, metadata)
+
+                if chunks_from_file:
+                    all_chunks.extend(chunks_from_file)
+
+                    for chunk in chunks_from_file:
+                        chunk_ids.append(chunk["id"])
+                        doc_path = processed_dir / f"{chunk['id']}.json"
+                        with open(doc_path, 'w', encoding='utf-8') as f:
+                            json_safe_chunk = self._ensure_json_serializable(chunk)
+                            json.dump(json_safe_chunk, f, indent=2)
+
+            if not all_chunks:
+                return {
+                    "success": False,
+                    "error": "No documents were successfully processed",
+                    "processing_time": time.time() - start_time
+                }
+
+            # Here don’t recompute the embedding again, simply take it from all_chunks
+            upsert_batch: List[Dict[str, Any]] = []
+            for chunk in all_chunks:
+                chunk_record = {
+                    "id": chunk["id"],
+                    "content": chunk["content"],
+                    "embedding": chunk["embedding"],
+                    "metadata": chunk["metadata"]
+                }
+                upsert_batch.append(chunk_record)
+
+            self.retriever.upsert_chunks(upsert_batch)
+
+            return {
+                "success": True,
+                "documents_ingested": len(documents),
+                "chunks_processed": len(all_chunks),
+                # todo: change to "chunk_ids": chunk_ids
+                "document_ids": chunk_ids,
+                "processing_time": time.time() - start_time
+            }
+
+        except Exception as e:
+            self.logger.error(f"Error ingesting documents: {e}")
+            import traceback
+            self.logger.error(traceback.format_exc())
+
+            return {
+                "success": False,
+                "error": str(e),
+                "documents_processed": 0,
+                "chunks_created": 0,
+                "chunks_inserted": 0,
+                "processing_time": time.time() - start_time
+            }
+
+    def _ensure_json_serializable(self, obj):
+        if isinstance(obj, dict):
+            return {k: self._ensure_json_serializable(v) for k, v in obj.items()}
+        elif isinstance(obj, list):
+            return [self._ensure_json_serializable(item) for item in obj]
+        elif isinstance(obj, (str, int, float, bool, type(None))):
+            return obj
+        else:
+            return str(obj)
+
+
+    def ingest_file(self, file_path: str) -> Dict[str, Any]:
+        start_time = time.time()
+        self.logger.info(f"Ingesting file: {file_path}")
+
+        try:
+            ingestion_result = self.data_ingestion.ingest_file(file_path)
+
+            if not ingestion_result["success"]:
+                return {
+                    "success": False,
+                    "error": ingestion_result.get("error", "Unknown error during file ingestion"),
+                    "processing_time": time.time() - start_time
+                }
+
+            documents = []
+            if "document" in ingestion_result:
+                documents = [ingestion_result["document"]]
+            elif "documents" in ingestion_result:
+                documents = ingestion_result["documents"]
+
+            if documents:
+                return self.ingest_documents(documents)
+            else:
+                return {
+                    "success": False,
+                    "error": "No valid documents found in file",
+                    "processing_time": time.time() - start_time
+                }
+
+        except Exception as e:
+            self.logger.error(f"Error ingesting file: {e}")
+            return {
+                "success": False,
+                "error": str(e),
+                "processing_time": time.time() - start_time
+            }
+
+
+    def ingest_directory(self, directory_path: str, file_extension: Optional[str] = None) -> Dict[str, Any]:
+        pass
+
+
+    def _retrieve_documents(self, query_embedding: List[float], filters: Dict[str, Any], query: str):
+        self.logger.info(f"Retrieving documents with filters: {filters}")
+
+        retrieved_docs = self.retriever.retrieve(
+            query_vector=query_embedding,
+            filters=filters,
+            query_text=query
+        )
+
+        self.logger.info(f"Retrieved {len(retrieved_docs)} documents")
+        return retrieved_docs
+
+
+    def query(self, query: str, chat_history: Optional[List[Dict[str, str]]] = None) -> Dict[str, Any]:
+        self.logger.info(f"Processing query: {query}")
+        start_time = time.time()
+
+        try:
+            query_embedding, filters = self.query_processor.process_query(query)
+
+            # TODO: Enhance retrieval by incorporating chat history context into the query embedding or retrieval strategy
+            retrieved_docs = self._retrieve_documents(query_embedding, filters, query)
+
+            if self.similarity_threshold > 0:
+                retrieved_docs = [doc for doc in retrieved_docs if doc.get('score', 0) >= self.similarity_threshold]
+                self.logger.info(f"After similarity threshold: {len(retrieved_docs)} documents")
+
+            ###CHECKPOINT 1c.
+
+            response = self.response_generator.generate_response(
+                query=query,
+                retrieved_docs=retrieved_docs,
+                chat_history=chat_history
+            )
+
+            processing_time = time.time() - start_time
+            response["processing_time"] = processing_time
+
+            return response
+
+        except Exception as e:
+            self.logger.error(f"Error processing query: {e}")
+            import traceback
+            self.logger.error(traceback.format_exc())
+
+            return {
+                "response": f"I encountered an error while processing your query: {str(e)}",
+                "sources": [],
+                "confidence": 0.0,
+                "processing_time": time.time() - start_time
+            }
\ No newline at end of file

agents/rag_agent/data_ingestion.py

+import os
+import json
+import logging
+from pathlib import Path
+import pandas as pd
+from typing import List, Dict, Any, Optional, Union
+import os
+os.environ["PATH"] += os.pathsep + "/opt/homebrew/bin"
+from unstructured.partition.pdf import partition_pdf
+from unstructured.chunking.title import chunk_by_title
+import json
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+class DataIngestion:
+
+    def __init__(self):
+
+        self.stats = {
+            "files_processed": 0,
+            "documents_ingested": 0,
+            "errors": 0
+        }
+
+        logger.info("DataIngestion initialized")
+
+    def ingest_directory(self, directory_path: str, file_extension: Optional[str] = None) -> Dict[str, Any]:
+
+        logger.info(f"Processing directory: {directory_path}")
+
+        try:
+            directory = Path(directory_path)
+            if not directory.exists() or not directory.is_dir():
+                raise ValueError(f"Directory does not exist: {directory_path}")
+
+            if file_extension:
+                files = list(directory.glob(f"*{file_extension}"))
+            else:
+                files = [f for f in directory.iterdir() if f.is_file()]
+
+            logger.info(f"Found {len(files)} files to process")
+
+            for file_path in files:
+                try:
+                    self.ingest_file(str(file_path))
+                    self.stats["files_processed"] += 1
+                except Exception as e:
+                    logger.error(f"Error processing file {file_path}: {e}")
+                    self.stats["errors"] += 1
+
+            return self.stats
+
+        except Exception as e:
+            logger.error(f"Error processing directory: {e}")
+            return self.stats
+
+    def ingest_file(self, file_path: str) -> Dict[str, Any]:
+        print(f"Processing file (file_path): {file_path}")
+        logger.info(f"Processing file: {file_path}")
+
+        file_path = Path(file_path)
+        if not file_path.exists():
+            raise FileNotFoundError(f"File not found: {file_path}")
+
+        if file_path.suffix.lower() == '.txt':
+            return self._ingest_text_file(file_path)
+        elif file_path.suffix.lower() == '.pdf':
+            return self._ingest_pdf_file(file_path)
+        else:
+            logger.warning(f"Unsupported file format: {file_path.suffix}")
+            return {"success": False, "error": "Unsupported file format"}
+
+    def _ingest_text_file(self, file_path: Path) -> Dict[str, Any]:
+
+        try:
+            with open(file_path, 'r', encoding='utf-8') as f:
+                content = f.read()
+
+            metadata = {
+                "source": file_path.name,
+                "file_type": "txt"
+            }
+
+            document = {
+                "content": content,
+                "metadata": metadata
+            }
+
+            logger.info(f"Successfully ingested text file: {file_path}")
+            self.stats["documents_ingested"] += 1
+
+            return {"success": True, "document": document}
+
+        except Exception as e:
+            logger.error(f"Error ingesting text file: {e}")
+            return {"success": False, "error": str(e)}
+
+
+    def _ingest_pdf_file(self, file_path: Path) -> Dict[str, Any]:
+
+        logger.info(f"Processing PDF with: {file_path}")
+
+        try:
+            elements = partition_pdf(
+                file_path,
+                extract_images_in_pdf=False,
+                extract_tables=True,
+                infer_table_structure=True,
+                languages=["de"],
+                ocr_languages = "deu"
+                # chunking_strategy="by_title"
+            )
+
+            ##
+            with open("debug_elements.txt", "w", encoding="utf-8") as f:
+                for i, element in enumerate(elements):
+                    f.write(f"\n=== Element {i} ===\n")
+                    f.write(f"Category: {getattr(element, 'category', 'N/A')}\n")
+                    f.write(f"Text: {str(element)}\n")
+                    f.write(f"Metadata: {getattr(element, 'metadata', {})}\n")
+            ##
+
+            content_parts = []
+            tables = []
+            images = []
+            metadata_parts = {}
+
+            for element in elements:
+                if hasattr(element, "category"):
+                    element_text = str(element)
+                    print(f"Element text: {element_text}")
+                    element_category = element.category
+                    print(f"Element category: {element_category}")
+
+                    if element_category == "Title":
+                        content_parts.append(f"\n## {element_text}\n")
+                    elif element_category == "NarrativeText":
+                        content_parts.append(element_text)
+                    elif element_category == "ListItem":
+                        content_parts.append(f"- {element_text}")
+                    elif element_category == "Table":
+                        formatted_table_text = self._format_table_as_markdown(element_text)
+                        content_parts.append(f"\n{formatted_table_text}\n")
+
+                        if hasattr(element, "metadata") and element.metadata:
+                            table_metadata = element.metadata.__dict__ if hasattr(element.metadata,
+                                                                                  "__dict__") else element.metadata
+                            tables.append({
+                                "text": formatted_table_text,
+                                "raw_text": element_text,
+                                "metadata": table_metadata
+                            })
+                    elif element_category == "Image":
+                        content_parts.append(f"\n[IMAGE: {element_text}]\n")
+                        if hasattr(element, "metadata") and element.metadata:
+                            images.append({
+                                "text": element_text,
+                                "metadata": element.metadata.__dict__ if hasattr(element.metadata,
+                                                                                 "__dict__") else element.metadata
+                            })
+                    else:
+                        content_parts.append(element_text)
+
+                    if hasattr(element, "metadata") and element.metadata:
+                        metadata = element.metadata.__dict__ if hasattr(element.metadata,
+                                                                        "__dict__") else element.metadata
+                        for key, value in metadata.items():
+                            if key not in metadata_parts:
+                                metadata_parts[key] = value
+
+            content = "\n".join(content_parts)
+
+            metadata = {
+                "source": file_path.name,
+                "file_type": "pdf",
+                "has_tables": len(tables) > 0,
+                "table_count": len(tables),
+                "has_images": len(images) > 0,
+                "image_count": len(images)
+            }
+
+            metadata.update(metadata_parts)
+
+            document = {
+                "content": content,
+                "metadata": metadata,
+                "tables": tables,
+                "images": images,
+                "elements": [{"category": getattr(e, "category", "Unknown"), "text": str(e)} for e in elements]
+            }
+
+            logger.info(f"Successfully ingested PDF file using unstructured.io: {file_path}")
+            self.stats["documents_ingested"] += 1
+
+            return {"success": True, "document": document}
+
+        except Exception as e:
+            logger.error(f"Error ingesting PDF file with unstructured.io: {e}")
+            return {"success": False, "error": str(e)}
+
+    def _format_table_as_markdown(self, table_text: str) -> str:
+        try:
+            lines = table_text.strip().split('\n')
+            if not lines:
+                return table_text
+
+            if '|' in table_text:
+                rows = [line.strip() for line in lines]
+
+                if not rows[0].startswith('|'):
+                    rows[0] = '| ' + rows[0] + ' |'
+
+                if len(rows) > 1 and not rows[1].startswith('|---'):
+                    cols = rows[0].count('|') - 1
+                    separator = '|' + '|'.join(['---' for _ in range(cols)]) + '|'
+                    rows.insert(1, separator)
+
+                for i in range(len(rows)):
+                    if i > 1 and not rows[i].startswith('|'):
+                        rows[i] = '| ' + rows[i] + ' |'
+
+                return '\n'.join(rows)
+            else:
+                rows = []
+                for line in lines:
+                    if line.strip():
+                        rows.append(line.split())
+
+                if not rows:
+                    return table_text
+
+                num_cols = len(rows[0]) if rows else 0
+                if num_cols == 0:
+                    return table_text
+
+                for i in range(len(rows)):
+                    while len(rows[i]) < num_cols:
+                        rows[i].append('')
+                    rows[i] = rows[i][:num_cols]
+
+                header = '| ' + ' | '.join(rows[0]) + ' |'
+                separator = '|' + '|'.join(['---' for _ in range(num_cols)]) + '|'
+
+                formatted_rows = [header, separator]
+                for row in rows[1:]:
+                    formatted_rows.append('| ' + ' | '.join(row) + ' |')
+
+                return '\n'.join(formatted_rows)
+        except Exception as e:
+            logger.warning(f"Error formatting table as markdown: {e}")
+            return f"Table:\n{table_text}"
+

agents/rag_agent/document_processor.py

+import re
+import uuid
+import logging
+from typing import List, Dict, Any, Optional, Tuple, Union
+import os
+from pathlib import Path
+import hashlib
+from datetime import datetime
+import nltk
+from nltk.tokenize import sent_tokenize
+import json
+
+try:
+    nltk.data.find('tokenizers/punkt')
+except LookupError:
+    nltk.download('punkt', quiet=True)
+
+
+def make_serializable(metadata: Dict[str, Any]) -> Dict[str, Any]:
+    serializable_metadata = {}
+    for key, value in metadata.items():
+        try:
+            json.dumps(value)  # Test if value is serializable
+            serializable_metadata[key] = value
+        except TypeError:
+            # Handle non-serializable objects
+            if isinstance(value, Header):
+                serializable_metadata[key] = str(value)  # Convert Header to string
+            else:
+                serializable_metadata[key] = str(value)  # Convert other non-serializable objects to string
+    return serializable_metadata
+
+
+class DocumentProcessor:
+    def __init__(self, config, embedding_model):
+        self.logger = logging.getLogger(__name__)
+        self.embedding_model = embedding_model
+        self.chunk_size = config.rag.chunk_size
+        self.chunk_overlap = config.rag.chunk_overlap
+        self.processed_docs_dir = Path(config.rag.processed_docs_dir)
+        self.processed_docs_dir.mkdir(parents=True, exist_ok=True)
+        self.chunking_strategy = getattr(config.rag, "chunking_strategy", "sliding_window")
+
+    def process_document(self, content: str, metadata: Dict[str, Any]) -> List[Dict[str, Any]]:
+        try:
+            doc_id_base = hashlib.md5(content.encode()).hexdigest()
+            doc_id = str(uuid.UUID(doc_id_base[:32]))
+
+            enhanced_metadata = metadata.copy()
+            enhanced_metadata['processing_timestamp'] = datetime.now().isoformat()
+
+            if self.chunking_strategy == "semantic":
+                chunks = self._create_semantic_chunks(content)
+            else:
+                chunks = self._create_sliding_window_chunks(content)
+
+            processed_chunks = []
+            for i, chunk_data in enumerate(chunks):
+                if isinstance(chunk_data, tuple):
+                    chunk_text = chunk_data[0]
+                else:
+                    chunk_text = chunk_data
+
+                chunk_id = str(uuid.UUID(doc_id_base[:24] + f"{i:08}"))
+                embedding = self.embedding_model.embed_documents([chunk_text])[0]
+
+                chunk_metadata = enhanced_metadata.copy()
+                chunk_metadata["chunk_number"] = i
+                chunk_metadata["total_chunks"] = len(chunks)
+                chunk_metadata["chunking_strategy"] = self.chunking_strategy
+                chunk_metadata["word_count"] = len(chunk_text.split())
+
+                processed_chunks.append({
+                    "id": chunk_id,
+                    "content": chunk_text,
+                    "embedding": embedding,
+                    "metadata": make_serializable(chunk_metadata)
+                })
+
+            return processed_chunks
+
+        except Exception as e:
+            self.logger.error(f"Error processing document: {e}")
+            raise
+
+    def _create_semantic_chunks(self, text: str) -> List[str]:
+        paragraphs = re.split(r'\n\s*\n', text)
+        chunks = []
+
+        for para in paragraphs:
+            if not para.strip():
+                continue
+            if len(para.split()) > self.chunk_size:
+                sentences = sent_tokenize(para)
+                current_chunk = []
+                current_length = 0
+                for sentence in sentences:
+                    length = len(sentence.split())
+                    if current_length + length > self.chunk_size and current_chunk:
+                        chunks.append(" ".join(current_chunk))
+                        current_chunk = []
+                        current_length = 0
+                    current_chunk.append(sentence)
+                    current_length += length
+                if current_chunk:
+                    chunks.append(" ".join(current_chunk))
+            else:
+                chunks.append(para.strip())
+
+        return chunks
+
+    def _create_sliding_window_chunks(self, text: str) -> List[str]:
+        sentences = sent_tokenize(text)
+        chunks = []
+
+        if len(sentences) <= 3:
+            return [text]
+
+        stride = max(1, (self.chunk_size - self.chunk_overlap) // 20)  # Approximate 20 words/sentence
+
+        for i in range(0, len(sentences), stride):
+            window_size = min(i + max(3, self.chunk_size // 20), len(sentences))
+            window_text = " ".join(sentences[i:window_size])
+            chunks.append(window_text)
+
+        return chunks
+
+
+class Header:
+    def __init__(self, text):
+        self.text = text
+
+    def __str__(self):
+        return self.text
\ No newline at end of file