[KM-467] [DED][AI] Orchestration Agent - Init

- https://bukittechnology.atlassian.net/browse/KM-467

.gitignore

@@ -0,0 +1,36 @@
+# Python-generated files
+**/__pycache__/*
+.env
+__pycache__
+agent-chat-ui
+config.yaml
+_archieved
+
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+asset_testing/
+test/users/user_accounts.csv
+.continue/
+
+# Virtual environments
+.venv
+
+# env
+.env
+.env.dev
+.env.uat
+.env.prd
+.env.example
+
+erd/
+playground/
+playground_retriever.py
+playground_chat.py
+playground_flush_cache.py
+playground_create_user.py
+API_CONTRACT.md
+context_engineering/

Dockerfile

@@ -0,0 +1,34 @@
+FROM python:3.12-slim-bookworm
+
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+
+WORKDIR /app
+
+ENV PYTHONUNBUFFERED=1 \
+    UV_COMPILE_BYTECODE=1
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    libpq-dev \
+    gcc \
+    libgomp1 \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN addgroup --system app && \
+    adduser --system --group --home /home/app app
+
+COPY pyproject.toml uv.lock ./
+RUN uv sync --frozen --no-dev
+
+# Download spaCy model required by presidio-analyzer
+RUN uv pip install https://github.com/explosion/spacy-models/releases/download/en_core_web_lg-3.8.0/en_core_web_lg-3.8.0-py3-none-any.whl
+
+COPY . .
+
+RUN chown -R app:app /app
+
+USER app
+
+EXPOSE 7860
+
+CMD ["uv", "run", "--no-sync", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,11 +1,77 @@
 ---
 title: Agentic Service Data Eyond
-emoji: 📉
-colorFrom: purple
-colorTo: purple
+emoji: 🏆
+colorFrom: red
+colorTo: blue
 sdk: docker
 pinned: false
-short_description: agentic backend service Data Eyond
 ---
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+
+
+How to run:
+`uv run --no-sync uvicorn main:app --host 0.0.0.0 --port 7860`
+
+
+Agent
+Orchestrator : intent recognition, orchestrate, and plannings
+Chatbot : have tools (retriever, and search), called by orchestrator
+
+
+APIs                        
+/api/v1/login -> login by email and password
+
+/api/v1/documents/{user_id} -> list all documents
+/api/v1/document/upload -> upload document
+/api/v1/document/delete -> delete document
+/api/v1/document/process -> extract document and ingest to vector index
+
+/api/v1/chat/stream -> talk with agent chatbot, in streaming response
+/api/v1/rooms/{user_id} -> list all room based on user id
+/api/v1/room/{room_id} -> get room based on room id
+
+
+Config
+- Agent: system prompt, guardrails
+- others config needed
+
+DB
+- using postgres as db
+- we can use pg vector from this db also
+- use redis for caching response, same question will not re-processed for 24 hour
+
+Document
+- service to manage document, upload, delete, log to db
+
+Knowledge
+- service to process document into vector, until ingestion to pg vector
+
+Middleware
+CORS:
+- allow all
+Rate limiting:
+- upload document: 10 document per menit
+Logging:
+- create clear and strutured logging for better debuging
+
+Models
+- Data models
+
+Observability
+- Langfuse traceability
+
+RAG
+- retriever service to get relevant context from pg vector
+
+storage
+- storage functionality to communicate with storage provider
+
+tools
+- tools that can be use by agent
+
+Users
+- Users management, to get user indentity based on login information.
+
+Utils
+- Other functionality

main.py

@@ -0,0 +1,70 @@
+"""Main application entry point."""
+
+from fastapi import FastAPI
+from src.middlewares.logging import configure_logging, get_logger
+from src.middlewares.cors import add_cors_middleware
+from src.middlewares.rate_limit import limiter, _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
+from src.api.v1.document import router as document_router
+from src.api.v1.chat import router as chat_router
+from src.api.v1.room import router as room_router
+from src.api.v1.users import router as users_router
+from src.api.v1.knowledge import router as knowledge_router
+from src.db.postgres.init_db import init_db
+import uvicorn
+
+# Configure logging
+configure_logging()
+logger = get_logger("main")
+
+# Create FastAPI app
+app = FastAPI(
+    title="DataEyond Agentic Service",
+    description="Multi-agent AI backend with RAG capabilities",
+    version="0.1.0"
+)
+
+# Add middleware
+add_cors_middleware(app)
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
+# Include routers
+app.include_router(users_router)
+app.include_router(document_router)
+app.include_router(knowledge_router)
+app.include_router(room_router)
+app.include_router(chat_router)
+
+
+@app.on_event("startup")
+async def startup_event():
+    """Initialize database on startup."""
+    logger.info("Starting application...")
+    await init_db()
+    logger.info("Database initialized")
+
+
+@app.get("/")
+async def root():
+    """Root endpoint."""
+    return {
+        "status": "ok",
+        "service": "DataEyond Agentic Service",
+        "version": "0.1.0"
+    }
+
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint."""
+    return {"status": "healthy"}
+
+
+if __name__ == "__main__":
+    uvicorn.run(
+        "main:app",
+        host="0.0.0.0",
+        port=7860,
+        reload=True
+    )

pyproject.toml

@@ -0,0 +1,135 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "agent-service-data-eyond"
+version = "0.1.0"
+description = "Agentic Service Data Eyond — Multi-Agent AI Backend"
+requires-python = ">=3.12,<3.13"
+
+dependencies = [
+    # --- Web Framework ---
+    "fastapi[standard]==0.115.6",
+    "uvicorn[standard]==0.32.1",
+    "python-multipart==0.0.12",
+    "starlette==0.41.3",
+    "sse-starlette==2.1.3",
+    # --- LangChain Core Ecosystem (NO LiteLLM) ---
+    "langchain==0.3.13",
+    "langchain-core==0.3.28",
+    "langchain-community==0.3.13",
+    "langchain-openai==0.2.14",
+    "langchain-postgres>=0.0.13",
+    "langgraph==0.2.60",
+    "langgraph-checkpoint-postgres==2.0.9",
+    # --- LLM / Azure OpenAI ---
+    "openai==1.58.1",
+    "tiktoken==0.8.0",
+    # --- Database ---
+    "sqlalchemy[asyncio]==2.0.36",
+    "asyncpg==0.30.0",
+    "psycopg[binary,pool]==3.2.3",
+    "pgvector==0.3.6",
+    "alembic==1.14.0",
+    # --- Azure ---
+    "azure-storage-blob==12.23.1",
+    "azure-identity==1.19.0",
+    "azure-ai-documentintelligence==1.0.0",
+    # --- Pydantic / Validation ---
+    "pydantic==2.10.3",
+    "pydantic-settings==2.7.0",
+    # --- Observability ---
+    "langfuse==2.57.4",
+    "structlog==24.4.0",
+    "prometheus-client==0.21.1",
+    # --- Security ---
+    "passlib[bcrypt]==1.7.4",
+    "cryptography==44.0.0",
+    # --- Rate Limiting ---
+    "slowapi==0.1.9",
+    "redis==5.2.1",
+    # --- Retry ---
+    "tenacity==9.0.0",
+    # --- Document Processing (for reading existing docs from blob) ---
+    "pypdf==5.1.0",
+    "python-docx==1.1.2",
+    "openpyxl==3.1.5",
+    "pandas==2.2.3",
+    # --- Chart/Visualization ---
+    "matplotlib==3.9.3",
+    "plotly==5.24.1",
+    "kaleido==0.2.1",
+    # --- MCP ---
+    "mcp==1.2.0",
+    # --- Advanced RAG ---
+    "rank-bm25==0.2.2",
+    "sentence-transformers==3.3.1",
+    # --- PII Detection (no LiteLLM) ---
+    "presidio-analyzer==2.2.355",
+    "presidio-anonymizer==2.2.355",
+    "spacy==3.8.3",
+    # --- Utilities ---
+    "httpx==0.28.1",
+    "anyio==4.7.0",
+    "python-dotenv==1.0.1",
+    "orjson==3.10.12",
+    "cachetools==5.5.0",
+    "apscheduler==3.10.4",
+    "jsonpatch>=1.33",
+    "pymongo>=4.14.0",
+    "psycopg2>=2.9.11",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest==8.3.4",
+    "pytest-asyncio==0.24.0",
+    "pytest-cov==6.0.0",
+    "httpx==0.28.1",
+    "ruff==0.8.4",
+    "mypy==1.13.0",
+    "pre-commit==4.0.1",
+]
+
+[tool.uv]
+dev-dependencies = [
+    "pytest==8.3.4",
+    "pytest-asyncio==0.24.0",
+    "pytest-cov==6.0.0",
+    "ruff==0.8.4",
+    "mypy==1.13.0",
+    "pre-commit==4.0.1",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/agent_service"]
+
+[tool.ruff]
+target-version = "py312"
+line-length = 100
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "N", "UP", "S", "B", "A", "C4", "T20"]
+ignore = [
+    "S101",   # assert statements OK in tests
+    "S105",   # hardcoded passwords — false positives in config
+    "S106",
+    "B008",   # FastAPI Depends() calls OK in function args
+]
+
+[tool.ruff.lint.per-file-ignores]
+"tests/**" = ["S101", "S105", "S106"]
+
+[tool.mypy]
+python_version = "3.12"
+strict = true
+ignore_missing_imports = true
+plugins = ["pydantic.mypy"]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]
+filterwarnings = [
+    "ignore::DeprecationWarning",
+]

run.py

@@ -0,0 +1,18 @@
+"""Entry point for running the app locally on Windows.
+
+Sets WindowsSelectorEventLoopPolicy BEFORE uvicorn creates its event loop,
+which is required for psycopg3 async mode compatibility.
+Use this instead of calling uvicorn directly on Windows:
+    uv run --no-sync python run.py
+"""
+
+import sys
+import asyncio
+
+if sys.platform == "win32":
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+import uvicorn
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host="0.0.0.0", port=7860, reload=False)

src/agents/chatbot.py

@@ -0,0 +1,75 @@
+"""Chatbot agent with RAG capabilities."""
+
+from langchain_openai import AzureChatOpenAI
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_core.output_parsers import StrOutputParser
+from src.config.settings import settings
+from src.middlewares.logging import get_logger
+from langchain_core.messages import HumanMessage, AIMessage
+
+logger = get_logger("chatbot")
+
+
+class ChatbotAgent:
+    """Chatbot agent with RAG capabilities."""
+
+    def __init__(self):
+        self.llm = AzureChatOpenAI(
+            azure_deployment=settings.azureai_deployment_name_4o,
+            openai_api_version=settings.azureai_api_version_4o,
+            azure_endpoint=settings.azureai_endpoint_url_4o,
+            api_key=settings.azureai_api_key_4o,
+            temperature=0.7
+        )
+
+        # Read system prompt
+        try:
+            with open("src/config/agents/system_prompt.md", "r") as f:
+                system_prompt = f.read()
+        except FileNotFoundError:
+            system_prompt = "You are a helpful AI assistant with access to user's uploaded documents."
+
+        # Create prompt template
+        self.prompt = ChatPromptTemplate.from_messages([
+            ("system", system_prompt),
+            MessagesPlaceholder(variable_name="messages"),
+            ("system", "Relevant documents:\n{context}")
+        ])
+
+        # Create chain
+        self.chain = self.prompt | self.llm | StrOutputParser()
+
+    async def generate_response(
+        self,
+        messages: list,
+        context: str = ""
+    ) -> str:
+        """Generate response with optional RAG context."""
+        try:
+            logger.info("Generating chatbot response")
+
+            # Generate response
+            response = await self.chain.ainvoke({
+                "messages": messages,
+                "context": context
+            })
+
+            logger.info(f"Generated response: {response[:100]}...")
+            return response
+
+        except Exception as e:
+            logger.error("Response generation failed", error=str(e))
+            raise
+
+    async def astream_response(self, messages: list, context: str = ""):
+        """Stream response tokens as they are generated."""
+        try:
+            logger.info("Streaming chatbot response")
+            async for token in self.chain.astream({"messages": messages, "context": context}):
+                yield token
+        except Exception as e:
+            logger.error("Response streaming failed", error=str(e))
+            raise
+
+
+chatbot = ChatbotAgent()

src/agents/orchestration.py

@@ -0,0 +1,74 @@
+"""Orchestrator agent for intent recognition and planning."""
+
+from langchain_openai import AzureChatOpenAI
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from src.config.settings import settings
+from src.middlewares.logging import get_logger
+from src.models.structured_output import IntentClassification
+
+logger = get_logger("orchestrator")
+
+
+class OrchestratorAgent:
+    """Orchestrator agent for intent recognition and planning."""
+
+    def __init__(self):
+        self.llm = AzureChatOpenAI(
+            azure_deployment=settings.azureai_deployment_name_4o,
+            openai_api_version=settings.azureai_api_version_4o,
+            azure_endpoint=settings.azureai_endpoint_url_4o,
+            api_key=settings.azureai_api_key_4o,
+            temperature=0
+        )
+
+        self.prompt = ChatPromptTemplate.from_messages([
+            ("system", """You are an orchestrator agent. You receive recent conversation history and the user's latest message.
+
+Your task:
+1. Determine intent: question, greeting, goodbye, or other
+2. Decide whether to search the user's documents (needs_search)
+3. If search is needed, rewrite the user's message into a STANDALONE search query that incorporates necessary context from conversation history. If the user says "tell me more" or "how many papers?", the search_query must spell out the full topic explicitly from history.
+4. If no search needed, provide a short direct_response (plain text only, no markdown formatting).
+
+Intent Routing:
+- question -> needs_search=True, search_query=<standalone rewritten query>
+- greeting -> needs_search=False, direct_response="Hello! How can I assist you today?"
+- goodbye -> needs_search=False, direct_response="Goodbye! Have a great day!"
+- other -> needs_search=True, search_query=<standalone rewritten query>
+"""),
+            MessagesPlaceholder(variable_name="history"),
+            ("user", "{message}")
+        ])
+
+        # with_structured_output uses function calling — guarantees valid schema regardless of LLM response style
+        self.chain = self.prompt | self.llm.with_structured_output(IntentClassification)
+
+    async def analyze_message(self, message: str, history: list = None) -> dict:
+        """Analyze user message and determine next actions.
+
+        Args:
+            message: The current user message.
+            history: Recent conversation as LangChain BaseMessage objects (oldest-first).
+                     Used to rewrite ambiguous follow-ups into standalone search queries.
+        """
+        try:
+            logger.info(f"Analyzing message: {message[:50]}...")
+
+            history_messages = history or []
+            result: IntentClassification = await self.chain.ainvoke({"message": message, "history": history_messages})
+
+            logger.info(f"Intent: {result.intent}, Needs search: {result.needs_search}, Search query: {result.search_query[:50] if result.search_query else ''}")
+            return result.model_dump()
+
+        except Exception as e:
+            logger.error("Message analysis failed", error=str(e))
+            # Fallback to treating everything as a question
+            return {
+                "intent": "question",
+                "needs_search": True,
+                "search_query": message,
+                "direct_response": None
+            }
+
+
+orchestrator = OrchestratorAgent()

src/api/v1/chat.py

@@ -0,0 +1,218 @@
+"""Chat endpoint with streaming support."""
+
+import asyncio
+import uuid
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+from src.db.postgres.connection import get_db
+from src.db.postgres.models import ChatMessage, MessageSource
+from src.agents.orchestration import orchestrator
+from src.agents.chatbot import chatbot
+from src.rag.retriever import retriever
+from src.db.redis.connection import get_redis
+from src.config.settings import settings
+from src.middlewares.logging import get_logger, log_execution
+from sse_starlette.sse import EventSourceResponse
+from langchain_core.messages import HumanMessage, AIMessage
+from sqlalchemy import select
+from pydantic import BaseModel
+from typing import List, Dict, Any, Optional
+import json
+
+_GREETINGS = frozenset(["hi", "hello", "hey", "halo", "hai", "hei"])
+_GOODBYES = frozenset(["bye", "goodbye", "thanks", "thank you", "terima kasih", "sampai jumpa"])
+
+
+def _fast_intent(message: str) -> Optional[dict]:
+    """Bypass LLM orchestrator for obvious greetings and farewells."""
+    lower = message.lower().strip().rstrip("!.,?")
+    if lower in _GREETINGS:
+        return {"intent": "greeting", "needs_search": False,
+                "direct_response": "Hello! How can I assist you today?", "search_query": ""}
+    if lower in _GOODBYES:
+        return {"intent": "goodbye", "needs_search": False,
+                "direct_response": "Goodbye! Have a great day!", "search_query": ""}
+    return None
+
+logger = get_logger("chat_api")
+
+router = APIRouter(prefix="/api/v1", tags=["Chat"])
+
+
+class ChatRequest(BaseModel):
+    user_id: str
+    room_id: str
+    message: str
+
+
+def _format_context(results: List[Dict[str, Any]]) -> str:
+    """Format retrieval results as context string for the LLM."""
+    lines = []
+    for result in results:
+        filename = result["metadata"].get("filename", "Unknown")
+        page = result["metadata"].get("page_label")
+        source_label = f"{filename}, p.{page}" if page else filename
+        lines.append(f"[Source: {source_label}]\n{result['content']}\n")
+    return "\n".join(lines)
+
+
+def _extract_sources(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """Extract deduplicated source references from retrieval results."""
+    seen = set()
+    sources = []
+    for result in results:
+        meta = result["metadata"]
+        key = (meta.get("document_id"), meta.get("page_label"))
+        if key not in seen:
+            seen.add(key)
+            sources.append({
+                "document_id": meta.get("document_id"),
+                "filename": meta.get("filename", "Unknown"),
+                "page_label": meta.get("page_label"),
+            })
+    return sources
+
+
+async def get_cached_response(redis, cache_key: str) -> Optional[str]:
+    cached = await redis.get(cache_key)
+    if cached:
+        return json.loads(cached)
+    return None
+
+
+async def cache_response(redis, cache_key: str, response: str):
+    await redis.setex(cache_key, 86400, json.dumps(response))
+
+
+async def load_history(db: AsyncSession, room_id: str, limit: int = 10) -> list:
+    """Load recent chat messages for a room as LangChain message objects (oldest-first)."""
+    result = await db.execute(
+        select(ChatMessage)
+        .where(ChatMessage.room_id == room_id)
+        .order_by(ChatMessage.created_at.asc())
+        .limit(limit)
+    )
+    rows = result.scalars().all()
+    return [
+        HumanMessage(content=row.content) if row.role == "user" else AIMessage(content=row.content)
+        for row in rows
+    ]
+
+
+async def save_messages(
+    db: AsyncSession,
+    room_id: str,
+    user_content: str,
+    assistant_content: str,
+    sources: Optional[List[Dict[str, Any]]] = None,
+):
+    """Persist user and assistant messages, and attach sources to the assistant message."""
+    db.add(ChatMessage(id=str(uuid.uuid4()), room_id=room_id, role="user", content=user_content))
+    assistant_id = str(uuid.uuid4())
+    db.add(ChatMessage(id=assistant_id, room_id=room_id, role="assistant", content=assistant_content))
+    for src in (sources or []):
+        page = src.get("page_label")
+        db.add(MessageSource(
+            id=str(uuid.uuid4()),
+            message_id=assistant_id,
+            document_id=src.get("document_id"),
+            filename=src.get("filename"),
+            page_label=str(page) if page is not None else None,
+        ))
+    await db.commit()
+
+
+@router.post("/chat/stream")
+@log_execution(logger)
+async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
+    """Chat endpoint with streaming response.
+
+    SSE event sequence:
+      1. sources  — JSON array of {document_id, filename, page_label}
+      2. chunk    — text fragments of the answer
+      3. done     — signals end of stream
+    """
+    redis = await get_redis()
+
+    cache_key = f"{settings.redis_prefix}chat:{request.room_id}:{request.message}"
+    cached = await get_cached_response(redis, cache_key)
+    if cached:
+        logger.info("Returning cached response")
+
+        async def stream_cached():
+            yield {"event": "sources", "data": json.dumps([])}
+            for i in range(0, len(cached), 50):
+                yield {"event": "chunk", "data": cached[i:i + 50]}
+            yield {"event": "done", "data": ""}
+
+        return EventSourceResponse(stream_cached())
+
+    try:
+        # Step 1: Fast local intent check (skips LLM for greetings/farewells)
+        intent_result = _fast_intent(request.message)
+
+        context = ""
+        sources: List[Dict[str, Any]] = []
+
+        if intent_result is None:
+            # Step 2: Launch retrieval and history loading in parallel, then run orchestrator
+            retrieval_task = asyncio.create_task(
+                retriever.retrieve(request.message, request.user_id, db)
+            )
+            history_task = asyncio.create_task(
+                load_history(db, request.room_id, limit=6)  # 6 msgs (3 pairs) for orchestrator
+            )
+            history = await history_task  # fast DB query (<100ms), done before orchestrator finishes
+            intent_result = await orchestrator.analyze_message(request.message, history)
+
+            if not intent_result.get("needs_search"):
+                retrieval_task.cancel()
+                raw_results = []
+            else:
+                search_query = intent_result.get("search_query", request.message)
+                logger.info(f"Searching for: {search_query}")
+                if search_query != request.message:
+                    retrieval_task.cancel()
+                    raw_results = await retriever.retrieve(
+                        query=search_query,
+                        user_id=request.user_id,
+                        db=db,
+                    )
+                else:
+                    raw_results = await retrieval_task
+
+            context = _format_context(raw_results)
+            sources = _extract_sources(raw_results)
+
+        # Step 3: Direct response for greetings / non-document intents
+        if intent_result.get("direct_response"):
+            response = intent_result["direct_response"]
+            await cache_response(redis, cache_key, response)
+            await save_messages(db, request.room_id, request.message, response, sources=[])
+
+            async def stream_direct():
+                yield {"event": "sources", "data": json.dumps([])}
+                yield {"event": "message", "data": response}
+
+            return EventSourceResponse(stream_direct())
+
+        # Step 4: Stream answer token-by-token as LLM generates it
+        # Load full history (10 msgs) for chatbot — richer context than the 6 used by orchestrator
+        full_history = await load_history(db, request.room_id, limit=10)
+        messages = full_history + [HumanMessage(content=request.message)]
+
+        async def stream_response():
+            full_response = ""
+            yield {"event": "sources", "data": json.dumps(sources)}
+            async for token in chatbot.astream_response(messages, context):
+                full_response += token
+                yield {"event": "chunk", "data": token}
+            yield {"event": "done", "data": ""}
+            await cache_response(redis, cache_key, full_response)
+            await save_messages(db, request.room_id, request.message, full_response, sources=sources)
+
+        return EventSourceResponse(stream_response())
+
+    except Exception as e:
+        logger.error("Chat failed", error=str(e))
+        raise HTTPException(status_code=500, detail=f"Chat failed: {str(e)}")

src/api/v1/document.py

@@ -0,0 +1,193 @@
+"""Document management API endpoints."""
+
+from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File, status
+from sqlalchemy.ext.asyncio import AsyncSession
+from src.db.postgres.connection import get_db
+from src.document.document_service import document_service
+from src.knowledge.processing_service import knowledge_processor
+from src.storage.az_blob.az_blob import blob_storage
+from src.middlewares.logging import get_logger, log_execution
+from src.middlewares.rate_limit import limiter
+from pydantic import BaseModel
+from typing import List
+
+logger = get_logger("document_api")
+
+router = APIRouter(prefix="/api/v1", tags=["Documents"])
+
+
+class DocumentResponse(BaseModel):
+    id: str
+    filename: str
+    status: str
+    file_size: int
+    file_type: str
+    created_at: str
+
+
+@router.get("/documents/{user_id}", response_model=List[DocumentResponse])
+@log_execution(logger)
+async def list_documents(
+    user_id: str,
+    db: AsyncSession = Depends(get_db)
+):
+    """List all documents for a user."""
+    documents = await document_service.get_user_documents(db, user_id)
+    return [
+        DocumentResponse(
+            id=doc.id,
+            filename=doc.filename,
+            status=doc.status,
+            file_size=doc.file_size or 0,
+            file_type=doc.file_type,
+            created_at=doc.created_at.isoformat()
+        )
+        for doc in documents
+    ]
+
+
+@router.post("/document/upload")
+@limiter.limit("10/minute")
+@log_execution(logger)
+async def upload_document(
+    request: Request,
+    file: UploadFile = File(...),
+    user_id: str = None,
+    db: AsyncSession = Depends(get_db)
+):
+    """Upload a document."""
+    if not user_id:
+        raise HTTPException(
+            status_code=400,
+            detail="user_id is required"
+        )
+
+    try:
+        # Read file content
+        content = await file.read()
+        file_size = len(content)
+
+        # Get file type
+        filename = file.filename
+        file_type = filename.split('.')[-1].lower() if '.' in filename else 'txt'
+
+        if file_type not in ['pdf', 'docx', 'txt']:
+            raise HTTPException(
+                status_code=400,
+                detail="Unsupported file type. Supported: pdf, docx, txt"
+            )
+
+        # Upload to blob storage
+        blob_name = await blob_storage.upload_file(content, filename, user_id)
+
+        # Create document record
+        document = await document_service.create_document(
+            db=db,
+            user_id=user_id,
+            filename=filename,
+            blob_name=blob_name,
+            file_size=file_size,
+            file_type=file_type
+        )
+
+        return {
+            "status": "success",
+            "message": "Document uploaded successfully",
+            "data": {
+                "id": document.id,
+                "filename": document.filename,
+                "status": document.status
+            }
+        }
+
+    except Exception as e:
+        logger.error(f"Upload failed for user {user_id}", error=str(e))
+        raise HTTPException(
+            status_code=500,
+            detail=f"Upload failed: {str(e)}"
+        )
+
+
+@router.delete("/document/delete")
+@log_execution(logger)
+async def delete_document(
+    document_id: str,
+    user_id: str,
+    db: AsyncSession = Depends(get_db)
+):
+    """Delete a document."""
+    document = await document_service.get_document(db, document_id)
+
+    if not document:
+        raise HTTPException(
+            status_code=404,
+            detail="Document not found"
+        )
+
+    if document.user_id != user_id:
+        raise HTTPException(
+            status_code=403,
+            detail="Access denied"
+        )
+
+    success = await document_service.delete_document(db, document_id)
+
+    if success:
+        return {"status": "success", "message": "Document deleted successfully"}
+    else:
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to delete document"
+        )
+
+
+@router.post("/document/process")
+@log_execution(logger)
+async def process_document(
+    document_id: str,
+    user_id: str,
+    db: AsyncSession = Depends(get_db)
+):
+    """Process document and ingest to vector index."""
+    document = await document_service.get_document(db, document_id)
+
+    if not document:
+        raise HTTPException(
+            status_code=404,
+            detail="Document not found"
+        )
+
+    if document.user_id != user_id:
+        raise HTTPException(
+            status_code=403,
+            detail="Access denied"
+        )
+
+    try:
+        # Update status to processing
+        await document_service.update_document_status(db, document_id, "processing")
+
+        # Process document
+        chunks_count = await knowledge_processor.process_document(document, db)
+
+        # Update status to completed
+        await document_service.update_document_status(db, document_id, "completed")
+
+        return {
+            "status": "success",
+            "message": "Document processed successfully",
+            "data": {
+                "document_id": document_id,
+                "chunks_processed": chunks_count
+            }
+        }
+
+    except Exception as e:
+        logger.error(f"Processing failed for document {document_id}", error=str(e))
+        await document_service.update_document_status(
+            db, document_id, "failed", str(e)
+        )
+        raise HTTPException(
+            status_code=500,
+            detail=f"Processing failed: {str(e)}"
+        )

src/api/v1/knowledge.py

@@ -0,0 +1,25 @@
+"""Knowledge base management API endpoints."""
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+from src.db.postgres.connection import get_db
+from src.middlewares.logging import get_logger, log_execution
+
+logger = get_logger("knowledge_api")
+
+router = APIRouter(prefix="/api/v1", tags=["Knowledge"])
+
+
+@router.post("/knowledge/rebuild")
+@log_execution(logger)
+async def rebuild_vector_index(
+    user_id: str,
+    db: AsyncSession = Depends(get_db)
+):
+    """Rebuild vector index for a user (admin endpoint)."""
+    # This would re-process all documents
+    # For POC, we'll skip this complexity
+    return {
+        "status": "success",
+        "message": "Vector index rebuild initiated"
+    }

src/api/v1/room.py

@@ -0,0 +1,169 @@
+"""Room management API endpoints."""
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+from src.db.postgres.connection import get_db
+from src.db.postgres.models import Room, ChatMessage, MessageSource
+from src.middlewares.logging import get_logger, log_execution
+from pydantic import BaseModel
+from typing import List, Optional
+from datetime import datetime
+import uuid
+
+logger = get_logger("room_api")
+
+router = APIRouter(prefix="/api/v1", tags=["Rooms"])
+
+
+class MessageSourceResponse(BaseModel):
+    document_id: Optional[str]
+    filename: Optional[str]
+    page_label: Optional[str]
+
+
+class ChatMessageResponse(BaseModel):
+    id: str
+    role: str
+    content: str
+    created_at: str
+    sources: List[MessageSourceResponse] = []
+
+
+class RoomResponse(BaseModel):
+    id: str
+    title: str
+    created_at: str
+    updated_at: str | None
+    messages: List[ChatMessageResponse] = []
+
+
+class CreateRoomRequest(BaseModel):
+    user_id: str
+    title: str = "New Chat"
+
+
+@router.get("/rooms/{user_id}", response_model=List[RoomResponse])
+@log_execution(logger)
+async def list_rooms(
+    user_id: str,
+    db: AsyncSession = Depends(get_db)
+):
+    """List all rooms for a user."""
+    result = await db.execute(
+        select(Room)
+        .where(Room.user_id == user_id, Room.status == "active")
+        .order_by(Room.updated_at.desc())
+    )
+    rooms = result.scalars().all()
+
+    return [
+        RoomResponse(
+            id=room.id,
+            title=room.title,
+            created_at=room.created_at.isoformat(),
+            updated_at=room.updated_at.isoformat() if room.updated_at else None
+        )
+        for room in rooms
+    ]
+
+
+@router.get("/room/{room_id}", response_model=RoomResponse)
+@log_execution(logger)
+async def get_room(
+    room_id: str,
+    db: AsyncSession = Depends(get_db)
+):
+    """Get a specific room with its chat history."""
+    result = await db.execute(
+        select(Room)
+        .where(Room.id == room_id)
+        .options(selectinload(Room.messages).selectinload(ChatMessage.sources))
+    )
+    room = result.scalars().first()
+
+    if not room:
+        raise HTTPException(
+            status_code=404,
+            detail="Room not found"
+        )
+
+    messages = sorted(room.messages, key=lambda m: m.created_at)
+
+    return RoomResponse(
+        id=room.id,
+        title=room.title,
+        created_at=room.created_at.isoformat(),
+        updated_at=room.updated_at.isoformat() if room.updated_at else None,
+        messages=[
+            ChatMessageResponse(
+                id=msg.id,
+                role=msg.role,
+                content=msg.content,
+                created_at=msg.created_at.isoformat(),
+                sources=[
+                    MessageSourceResponse(
+                        document_id=src.document_id,
+                        filename=src.filename,
+                        page_label=src.page_label,
+                    )
+                    for src in msg.sources
+                ],
+            )
+            for msg in messages
+        ]
+    )
+
+
+@router.delete("/room/{room_id}")
+@log_execution(logger)
+async def delete_room(
+    room_id: str,
+    user_id: str,
+    db: AsyncSession = Depends(get_db)
+):
+    """Soft-delete a room by setting its status to inactive."""
+    result = await db.execute(
+        select(Room).where(Room.id == room_id)
+    )
+    room = result.scalars().first()
+
+    if not room:
+        raise HTTPException(status_code=404, detail="Room not found")
+
+    if room.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Access denied")
+
+    room.status = "inactive"
+    await db.commit()
+
+    return {"status": "success", "message": "Room deleted successfully"}
+
+
+@router.post("/room/create")
+@log_execution(logger)
+async def create_room(
+    request: CreateRoomRequest,
+    db: AsyncSession = Depends(get_db)
+):
+    """Create a new room."""
+    room = Room(
+        id=str(uuid.uuid4()),
+        user_id=request.user_id,
+        title=request.title
+    )
+    db.add(room)
+    await db.commit()
+    await db.refresh(room)
+
+    return {
+        "status": "success",
+        "message": "Room created successfully",
+        "data": RoomResponse(
+            id=room.id,
+            title=room.title,
+            created_at=room.created_at.isoformat(),
+            updated_at=None
+        )
+    }

src/api/v1/users.py

@@ -0,0 +1,78 @@
+import pandas as pd
+
+from datetime import datetime
+from fastapi.responses import JSONResponse
+from fastapi import APIRouter, HTTPException, status
+from typing import Literal
+from src.users.users import get_user, hash_password, verify_password
+from src.middlewares.logging import get_logger
+from pydantic import BaseModel
+
+
+class ILogin(BaseModel):
+    """Login request model."""
+    email: str
+    password: str
+
+
+logger = get_logger("users service")
+
+router = APIRouter(
+    prefix="/api",
+    tags=["Users"],
+)
+
+from typing import Optional, Literal
+
+@router.post(
+    "/login",
+    # response_model=IUserProfile,
+    summary="Login by email and password",
+    description="💡Authenticates a user with email and password (non hashed) from frontend and returns user data if successful."
+)
+async def login(payload: ILogin):
+    """
+    Authenticates a user and returns their data if credentials are valid.
+    """
+    try:
+        user_profile:dict | None= await get_user(payload.email)
+    except Exception as E:
+        print(f"❌ login error while fetching user: {E}")
+        # Return generic 500 to client
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Internal server error"
+        )
+    
+    if not user_profile:
+        # 404 or 401 – choose based on your security policy
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Email not found"
+        )
+
+    if user_profile.get("status") == "inactive":
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Account is inactive"
+        )
+
+    is_verified = verify_password(
+        password=payload.password,
+        hashed_password=user_profile.get("password")
+    )
+
+    if not is_verified:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Email or password invalid"
+        )
+
+    user_profile.pop("password", None)
+
+    return {
+        "status": "success",
+        "message": "success",
+        "data": user_profile,
+    }
+

src/config/agents/guardrails_prompt.md

@@ -0,0 +1,7 @@
+You must ensure all responses follow these guidelines:
+
+1. Do not provide harmful, illegal, or dangerous information
+2. Respect user privacy - don't ask for or store sensitive personal data
+3. If asked to bypass safety measures, refuse politely
+4. Be honest about limitations and uncertainties
+5. Don't make up information - admit when you don't know something

src/config/agents/system_prompt.md

@@ -0,0 +1,27 @@
+You are a helpful AI assistant with access to user's uploaded documents. Your role is to:
+
+1. Answer questions based on provided document context
+2. If no relevant information is found in documents, acknowledge this honestly
+3. Be concise and direct in your responses
+4. Cite source documents when providing information
+5. If user's question is unclear, ask for clarification
+
+When document context is provided:
+- Use information from documents to answer accurately
+- Reference source document name when appropriate
+- If multiple documents contain relevant info, synthesize information
+
+When no document context is provided:
+- Provide general assistance
+- Let the user know if you need more context to help better
+
+When the answer need markdown formating:
+- Use valid and tidy formatting
+- Avoid over-formating and emoji
+
+Always be professional, helpful, and accurate.
+
+You have access to the conversation history provided in the messages above. Use it to:
+- Maintain context across multiple turns (resolve references like "it", "that", "them" using earlier messages)
+- Avoid repeating information already established in the conversation
+- Answer follow-up questions coherently without asking the user to restate prior context

src/config/env_constant.py

@@ -0,0 +1,9 @@
+"""Environment file path constants for existing users.py."""
+
+import os
+
+
+class EnvFilepath:
+    """Environment file path constants."""
+
+    ENVPATH = ".env"

src/config/settings.py

@@ -0,0 +1,67 @@
+"""Centralized configuration management using pydantic-settings."""
+
+import os
+from typing import Optional
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    """Application settings loaded from environment variables."""
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="allow",
+        case_sensitive=False,
+    )
+
+    # Database
+    postgres_connstring: str
+
+    # Redis
+    redis_url: str
+    redis_prefix: str = "dataeyond-agent-service_"
+
+    # Azure OpenAI - GPT-4o (map to .env names with double underscores)
+    azureai_api_key_4o: str = Field(alias="azureai__api_key__4o", default="")
+    azureai_endpoint_url_4o: str = Field(alias="azureai__endpoint__url__4o", default="")
+    azureai_deployment_name_4o: str = Field(alias="azureai__deployment__name__4o", default="")
+    azureai_api_version_4o: str = Field(alias="azureai__api__version__4o", default="")
+
+    # Azure OpenAI - Embeddings
+    azureai_api_key_embedding: str = Field(alias="azureai__api_key__embedding", default="")
+    azureai_endpoint_url_embedding: str = Field(alias="azureai__endpoint__url__embedding", default="")
+    azureai_deployment_name_embedding: str = Field(alias="azureai__deployment__name__embedding", default="")
+    azureai_api_version_embedding: str = Field(alias="azureai__api__version__embedding", default="")
+
+    # Azure Document Intelligence
+    azureai_docintel_endpoint: str = Field(alias="azureai__docintel__endpoint", default="")
+    azureai_docintel_key: str = Field(alias="azureai__docintel__key", default="")
+
+    # Azure Blob Storage
+    azureai_blob_sas: str = Field(alias="azureai__blob__sas", default="")
+    azureai_container_endpoint: str = Field(alias="azureai__container__endpoint", default="")
+    azureai_container_name: str = Field(alias="azureai__container__name", default="")
+    azureai_container_account_name: str = Field(alias="azureai__container__account__name", default="")
+
+    # Langfuse
+    LANGFUSE_PUBLIC_KEY: str
+    LANGFUSE_SECRET_KEY: str
+    LANGFUSE_HOST: str
+
+    # MongoDB (for users - existing)
+    emarcal_mongo_endpoint_url: str = Field(alias="emarcal__mongo__endpoint__url", default="")
+    emarcal_buma_mongo_dbname: str = Field(alias="emarcal__buma__mongo__dbname", default="")
+
+    # JWT (for users - existing)
+    emarcal_jwt_secret_key: str = Field(alias="emarcal__jwt__secret_key", default="")
+    emarcal_jwt_algorithm: str = Field(alias="emarcal__jwt__algorithm", default="HS256")
+
+    # Bcrypt salt (for users - existing)
+    emarcal_bcrypt_salt: str = Field(alias="emarcal__bcrypt__salt", default="")
+
+
+# Singleton instance
+settings = Settings()
+

src/db/postgres/connection.py

@@ -0,0 +1,52 @@
+"""Async PostgreSQL connection management."""
+
+from sqlalchemy.engine import make_url
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
+from sqlalchemy.orm import declarative_base
+from src.config.settings import settings
+
+# asyncpg doesn't support libpq query params like sslmode/channel_binding.
+# Use SQLAlchemy's URL parser to strip all query params cleanly, then pass ssl via connect_args.
+_url = make_url(settings.postgres_connstring).set(drivername="postgresql+asyncpg", query={})
+
+# Separate asyncpg engine for PGVector with prepared_statement_cache_size=0.
+# PGVector runs advisory_lock + CREATE EXTENSION as a single multi-statement string.
+# asyncpg normally uses prepared statements which reject multi-statement SQL.
+# Setting cache_size=0 forces asyncpg to use execute() instead of prepare(),
+# which supports multiple statements — no psycopg3 needed, no ProactorEventLoop issue.
+_pgvector_engine = create_async_engine(
+    _url,
+    pool_pre_ping=True,
+    connect_args={
+        "ssl": "require",
+        "prepared_statement_cache_size": 0,
+    },
+)
+
+engine = create_async_engine(
+    _url,
+    echo=False,
+    pool_pre_ping=True,
+    pool_size=5,
+    max_overflow=10,
+    connect_args={"ssl": "require"},
+)
+
+AsyncSessionLocal = async_sessionmaker(
+    engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+    autocommit=False,
+    autoflush=False
+)
+
+Base = declarative_base()
+
+
+async def get_db():
+    """Get database session dependency for FastAPI."""
+    async with AsyncSessionLocal() as session:
+        try:
+            yield session
+        finally:
+            await session.close()

src/db/postgres/init_db.py

@@ -0,0 +1,23 @@
+"""Database initialization."""
+
+from sqlalchemy import text
+from src.db.postgres.connection import engine, Base
+from src.db.postgres.models import Document, Room, ChatMessage, User, MessageSource
+
+
+async def init_db():
+    """Initialize database tables and required extensions."""
+    async with engine.begin() as conn:
+        # Create pgvector extension using two separate statements.
+        # Must NOT be combined into one string — asyncpg rejects multi-statement
+        # prepared statements (langchain_postgres bug workaround via create_extension=False).
+        await conn.execute(text("SELECT pg_advisory_xact_lock(1573678846307946496)"))
+        await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
+
+        # Create application tables
+        await conn.run_sync(Base.metadata.create_all)
+
+        # Schema migrations (idempotent — safe to run on every startup)
+        await conn.execute(text(
+            "ALTER TABLE rooms ADD COLUMN IF NOT EXISTS status VARCHAR NOT NULL DEFAULT 'active'"
+        ))

src/db/postgres/models.py

@@ -0,0 +1,83 @@
+"""SQLAlchemy database models."""
+
+from uuid import uuid4
+from sqlalchemy import Column, String, DateTime, Text, Integer, ForeignKey
+from sqlalchemy.orm import relationship
+from sqlalchemy.sql import func
+from src.db.postgres.connection import Base
+
+
+class User(Base):
+    """User model."""
+    __tablename__ = "users"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid4()))
+    fullname = Column(String, nullable=False)
+    email = Column(String, nullable=False, unique=True, index=True)
+    password = Column(String, nullable=False)  # bcrypt-hashed
+    company = Column(String)
+    company_size = Column(String)
+    function = Column(String)
+    site = Column(String)
+    role = Column(String)
+    status = Column(String, nullable=False, default="active")  # active | inactive
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+
+
+class Document(Base):
+    """Document model."""
+    __tablename__ = "documents"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid4()))
+    user_id = Column(String, nullable=False, index=True)
+    filename = Column(String, nullable=False)
+    blob_name = Column(String, nullable=False, unique=True)
+    file_size = Column(Integer)
+    file_type = Column(String)  # pdf, docx, txt, etc.
+    status = Column(String, default="uploaded")  # uploaded, processing, completed, failed
+    processed_at = Column(DateTime(timezone=True))
+    error_message = Column(Text)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+
+
+class Room(Base):
+    """Room model for chat sessions."""
+    __tablename__ = "rooms"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid4()))
+    user_id = Column(String, nullable=False, index=True)
+    title = Column(String, default="New Chat")
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
+
+    status = Column(String, nullable=False, default="active")  # active | inactive
+
+    messages = relationship("ChatMessage", back_populates="room", cascade="all, delete-orphan")
+
+
+class ChatMessage(Base):
+    """Chat message model."""
+    __tablename__ = "chat_messages"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid4()))
+    room_id = Column(String, ForeignKey("rooms.id"), nullable=False, index=True)
+    role = Column(String, nullable=False)  # user, assistant
+    content = Column(Text, nullable=False)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+
+    room = relationship("Room", back_populates="messages")
+    sources = relationship("MessageSource", back_populates="message", cascade="all, delete-orphan")
+
+
+class MessageSource(Base):
+    """Sources (RAG references) attached to an assistant message."""
+    __tablename__ = "message_sources"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid4()))
+    message_id = Column(String, ForeignKey("chat_messages.id", ondelete="CASCADE"), nullable=False, index=True)
+    document_id = Column(String)
+    filename = Column(Text)
+    page_label = Column(Text)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+
+    message = relationship("ChatMessage", back_populates="sources")

src/db/postgres/vector_store.py

@@ -0,0 +1,31 @@
+"""PGVector store setup for document embeddings."""
+
+from langchain_postgres import PGVector
+from langchain_openai import AzureOpenAIEmbeddings
+from src.config.settings import settings
+from src.db.postgres.connection import _pgvector_engine
+
+# Initialize embeddings
+embeddings = AzureOpenAIEmbeddings(
+    azure_deployment=settings.azureai_deployment_name_embedding,
+    openai_api_version=settings.azureai_api_version_embedding,
+    azure_endpoint=settings.azureai_endpoint_url_embedding,
+    api_key=settings.azureai_api_key_embedding
+)
+
+# Use psycopg3 connection string (not asyncpg engine) with async_mode=True.
+# psycopg3 supports multi-statement SQL, which PGVector needs for
+# advisory_lock + CREATE EXTENSION vector. asyncpg rejects this as a prepared statement.
+vector_store = PGVector(
+    embeddings=embeddings,
+    connection=_pgvector_engine,
+    collection_name="document_embeddings",
+    use_jsonb=True,
+    async_mode=True,
+    create_extension=False,  # Extension pre-created in init_db.py (avoids multi-statement asyncpg bug)
+)
+
+
+def get_vector_store():
+    """Get the vector store instance."""
+    return vector_store

src/db/redis/connection.py

@@ -0,0 +1,16 @@
+"""Redis connection for caching."""
+
+import redis.asyncio as redis
+from src.config.settings import settings
+
+redis_client = redis.from_url(
+    settings.redis_url,
+    encoding="utf-8",
+    decode_responses=True,
+    ssl_cert_reqs=None
+)
+
+
+async def get_redis():
+    """Get Redis client."""
+    return redis_client

src/document/document_service.py

@@ -0,0 +1,108 @@
+"""Service for managing documents."""
+
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, delete
+from src.db.postgres.models import Document
+from src.storage.az_blob.az_blob import blob_storage
+from src.middlewares.logging import get_logger
+from typing import List, Optional
+from datetime import datetime
+
+logger = get_logger("document_service")
+
+
+class DocumentService:
+    """Service for managing documents."""
+
+    async def create_document(
+        self,
+        db: AsyncSession,
+        user_id: str,
+        filename: str,
+        blob_name: str,
+        file_size: int,
+        file_type: str
+    ) -> Document:
+        """Create a new document record."""
+        import uuid
+        document = Document(
+            id=str(uuid.uuid4()),
+            user_id=user_id,
+            filename=filename,
+            blob_name=blob_name,
+            file_size=file_size,
+            file_type=file_type,
+            status="uploaded"
+        )
+        db.add(document)
+        await db.commit()
+        await db.refresh(document)
+        logger.info(f"Created document {document.id} for user {user_id}")
+        return document
+
+    async def get_user_documents(
+        self,
+        db: AsyncSession,
+        user_id: str
+    ) -> List[Document]:
+        """Get all documents for a user."""
+        result = await db.execute(
+            select(Document)
+            .where(Document.user_id == user_id)
+            .order_by(Document.created_at.desc())
+        )
+        return result.scalars().all()
+
+    async def get_document(
+        self,
+        db: AsyncSession,
+        document_id: str
+    ) -> Optional[Document]:
+        """Get a specific document."""
+        result = await db.execute(
+            select(Document).where(Document.id == document_id)
+        )
+        return result.scalars().first()
+
+    async def delete_document(
+        self,
+        db: AsyncSession,
+        document_id: str
+    ) -> bool:
+        """Delete a document (from DB and Blob storage)."""
+        document = await self.get_document(db, document_id)
+        if not document:
+            return False
+
+        # Delete from blob storage
+        await blob_storage.delete_file(document.blob_name)
+
+        # Delete from database
+        await db.execute(
+            delete(Document).where(Document.id == document_id)
+        )
+        await db.commit()
+
+        logger.info(f"Deleted document {document_id}")
+        return True
+
+    async def update_document_status(
+        self,
+        db: AsyncSession,
+        document_id: str,
+        status: str,
+        error_message: Optional[str] = None
+    ) -> Document:
+        """Update document processing status."""
+        document = await self.get_document(db, document_id)
+        if document:
+            document.status = status
+            document.processed_at = datetime.utcnow()
+            document.error_message = error_message
+            await db.commit()
+            await db.refresh(document)
+            logger.info(f"Updated document {document_id} status to {status}")
+        return document
+
+
+document_service = DocumentService()

src/knowledge/processing_service.py

@@ -0,0 +1,146 @@
+"""Service for processing documents and ingesting to vector store."""
+
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain_core.documents import Document as LangChainDocument
+from src.db.postgres.vector_store import get_vector_store
+from src.storage.az_blob.az_blob import blob_storage
+from src.db.postgres.models import Document as DBDocument
+from src.config.settings import settings
+from sqlalchemy.ext.asyncio import AsyncSession
+from src.middlewares.logging import get_logger
+from azure.ai.documentintelligence.aio import DocumentIntelligenceClient
+from azure.core.credentials import AzureKeyCredential
+from typing import List
+import pypdf
+import docx
+from io import BytesIO
+
+logger = get_logger("knowledge_processing")
+
+
+class KnowledgeProcessingService:
+    """Service for processing documents and ingesting to vector store."""
+
+    def __init__(self):
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000,
+            chunk_overlap=200,
+            length_function=len
+        )
+
+    async def process_document(self, db_doc: DBDocument, db: AsyncSession) -> int:
+        """Process document and ingest to vector store.
+
+        Returns:
+            Number of chunks ingested
+        """
+        try:
+            logger.info(f"Processing document {db_doc.id}")
+            content = await blob_storage.download_file(db_doc.blob_name)
+
+            if db_doc.file_type == "pdf":
+                documents = await self._build_pdf_documents(content, db_doc)
+            else:
+                text = self._extract_text(content, db_doc.file_type)
+                if not text.strip():
+                    raise ValueError("No text extracted from document")
+                chunks = self.text_splitter.split_text(text)
+                documents = [
+                    LangChainDocument(
+                        page_content=chunk,
+                        metadata={
+                            "document_id": db_doc.id,
+                            "user_id": db_doc.user_id,
+                            "filename": db_doc.filename,
+                            "chunk_index": i,
+                        }
+                    )
+                    for i, chunk in enumerate(chunks)
+                ]
+
+            if not documents:
+                raise ValueError("No text extracted from document")
+
+            vector_store = get_vector_store()
+            await vector_store.aadd_documents(documents)
+
+            logger.info(f"Processed {db_doc.id}: {len(documents)} chunks ingested")
+            return len(documents)
+
+        except Exception as e:
+            logger.error(f"Failed to process document {db_doc.id}", error=str(e))
+            raise
+
+    async def _build_pdf_documents(
+        self, content: bytes, db_doc: DBDocument
+    ) -> List[LangChainDocument]:
+        """Build LangChain documents from PDF with page_label metadata.
+
+        Uses Azure Document Intelligence (per-page) when credentials are present,
+        falls back to pypdf (also per-page) otherwise.
+        """
+        documents: List[LangChainDocument] = []
+
+        if settings.azureai_docintel_endpoint and settings.azureai_docintel_key:
+            async with DocumentIntelligenceClient(
+                endpoint=settings.azureai_docintel_endpoint,
+                credential=AzureKeyCredential(settings.azureai_docintel_key),
+            ) as client:
+                poller = await client.begin_analyze_document(
+                    model_id="prebuilt-read",
+                    body=BytesIO(content),
+                    content_type="application/pdf",
+                )
+                result = await poller.result()
+                logger.info(f"Azure DI extracted {len(result.pages or [])} pages")
+
+                for page in result.pages or []:
+                    page_text = "\n".join(
+                        line.content for line in (page.lines or [])
+                    )
+                    if not page_text.strip():
+                        continue
+                    for chunk in self.text_splitter.split_text(page_text):
+                        documents.append(LangChainDocument(
+                            page_content=chunk,
+                            metadata={
+                                "document_id": db_doc.id,
+                                "user_id": db_doc.user_id,
+                                "filename": db_doc.filename,
+                                "chunk_index": len(documents),
+                                "page_label": page.page_number,
+                            }
+                        ))
+        else:
+            logger.warning("Azure DI not configured, using pypdf")
+            pdf_reader = pypdf.PdfReader(BytesIO(content))
+            for page_num, page in enumerate(pdf_reader.pages, start=1):
+                page_text = page.extract_text() or ""
+                if not page_text.strip():
+                    continue
+                for chunk in self.text_splitter.split_text(page_text):
+                    documents.append(LangChainDocument(
+                        page_content=chunk,
+                        metadata={
+                            "document_id": db_doc.id,
+                            "user_id": db_doc.user_id,
+                            "filename": db_doc.filename,
+                            "chunk_index": len(documents),
+                            "page_label": page_num,
+                        }
+                    ))
+
+        return documents
+
+    def _extract_text(self, content: bytes, file_type: str) -> str:
+        """Extract text from DOCX or TXT content."""
+        if file_type == "docx":
+            doc = docx.Document(BytesIO(content))
+            return "\n".join(p.text for p in doc.paragraphs)
+        elif file_type == "txt":
+            return content.decode("utf-8")
+        else:
+            raise ValueError(f"Unsupported file type: {file_type}")
+
+
+knowledge_processor = KnowledgeProcessingService()

src/middlewares/cors.py

@@ -0,0 +1,14 @@
+"""CORS middleware configuration."""
+
+from fastapi.middleware.cors import CORSMiddleware
+
+
+def add_cors_middleware(app):
+    """Add CORS middleware to allow all origins for POC."""
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],  # For POC - allow all
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )

src/middlewares/logging.py

@@ -0,0 +1,66 @@
+"""Structured logging middleware with structlog."""
+
+import structlog
+from functools import wraps
+from typing import Callable, Any
+import time
+
+
+def configure_logging():
+    """Configure structured logging."""
+    structlog.configure(
+        processors=[
+            structlog.stdlib.filter_by_level,
+            structlog.stdlib.add_logger_name,
+            structlog.stdlib.add_log_level,
+            structlog.stdlib.PositionalArgumentsFormatter(),
+            structlog.processors.TimeStamper(fmt="iso"),
+            structlog.processors.StackInfoRenderer(),
+            structlog.processors.format_exc_info,
+            structlog.processors.UnicodeDecoder(),
+            structlog.processors.JSONRenderer()
+        ],
+        context_class=dict,
+        logger_factory=structlog.stdlib.LoggerFactory(),
+        cache_logger_on_first_use=True,
+    )
+
+
+def get_logger(name: str) -> structlog.stdlib.BoundLogger:
+    """Get a configured logger."""
+    return structlog.get_logger(name)
+
+
+def log_execution(logger: structlog.stdlib.BoundLogger):
+    """Decorator to log function execution."""
+    def decorator(func: Callable) -> Callable:
+        @wraps(func)
+        async def async_wrapper(*args, **kwargs) -> Any:
+            start_time = time.time()
+            logger.info(f"Starting {func.__name__}")
+            try:
+                result = await func(*args, **kwargs)
+                duration = time.time() - start_time
+                logger.info(f"Completed {func.__name__}", duration=duration)
+                return result
+            except Exception as e:
+                duration = time.time() - start_time
+                logger.error(f"Error in {func.__name__}", error=str(e), duration=duration)
+                raise
+
+        @wraps(func)
+        def sync_wrapper(*args, **kwargs) -> Any:
+            start_time = time.time()
+            logger.info(f"Starting {func.__name__}")
+            try:
+                result = func(*args, **kwargs)
+                duration = time.time() - start_time
+                logger.info(f"Completed {func.__name__}", duration=duration)
+                return result
+            except Exception as e:
+                duration = time.time() - start_time
+                logger.error(f"Error in {func.__name__}", error=str(e), duration=duration)
+                raise
+
+        return async_wrapper if hasattr(func, '__call__') and hasattr(func, '__code__') and func.__code__.co_flags & 0x80 else sync_wrapper
+    return decorator

src/middlewares/rate_limit.py

@@ -0,0 +1,17 @@
+"""Rate limiting middleware using slowapi."""
+
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.util import get_remote_address
+from slowapi.errors import RateLimitExceeded
+from fastapi import Request
+
+limiter = Limiter(key_func=get_remote_address)
+
+
+def get_user_id_from_request(request: Request) -> str:
+    """Extract user ID from request for rate limiting."""
+    # For document upload, use user_id if available, otherwise IP
+    user_id = request.headers.get("X-User-ID")
+    if user_id:
+        return user_id
+    return get_remote_address(request)

src/models/security.py

@@ -0,0 +1,10 @@
+"""Security models for password validation."""
+
+from pydantic import BaseModel
+
+
+class ValidatePassword(BaseModel):
+    """Password validation response."""
+    status: int
+    data: bool
+    error: str | None

src/models/states.py

@@ -0,0 +1,14 @@
+"""LangGraph state definitions for agent workflows."""
+
+from typing import TypedDict, List, Annotated, Optional
+from langgraph.graph.message import add_messages
+from langchain_core.messages import BaseMessage
+
+
+class AgentState(TypedDict):
+    """State for agent graph."""
+    messages: Annotated[List[BaseMessage], add_messages]
+    user_id: str
+    room_id: str
+    retrieved_docs: List[dict]
+    needs_search: bool

src/models/structured_output.py

@@ -0,0 +1,21 @@
+"""Structured output models for LLM."""
+
+from pydantic import BaseModel, Field
+
+
+class IntentClassification(BaseModel):
+    """Intent classification output."""
+    intent: str = Field(
+        description="The user's intent: 'question', 'greeting', 'goodbye', 'other'"
+    )
+    needs_search: bool = Field(
+        description="Whether document search is needed"
+    )
+    search_query: str = Field(
+        default="",
+        description="The query to use for document search if needed"
+    )
+    direct_response: str = Field(
+        default="",
+        description="Direct response if no search needed (for greetings, etc.)"
+    )

src/models/user_info.py

@@ -0,0 +1,15 @@
+"""User info models for existing users.py."""
+
+from pydantic import BaseModel
+
+
+class UserCreate(BaseModel):
+    """User creation model."""
+    fullname: str
+    email: str
+    password: str
+    company: str | None = None
+    company_size: str | None = None
+    function: str | None = None
+    site: str | None = None
+    role: str | None = None

src/observability/langfuse/langfuse.py

@@ -0,0 +1,29 @@
+"""Langfuse observability integration."""
+
+from langfuse import Langfuse
+from src.config.settings import settings
+from src.middlewares.logging import get_logger
+
+logger = get_logger("langfuse")
+
+
+def get_langfuse():
+    """Get Langfuse client."""
+    return Langfuse(
+        public_key=settings.LANGFUSE_PUBLIC_KEY,
+        secret_key=settings.LANGFUSE_SECRET_KEY,
+        host=settings.LANGFUSE_HOST
+    )
+
+
+def trace_chat(user_id: str, room_id: str, query: str, response: str):
+    """Trace a chat interaction."""
+    langfuse = get_langfuse()
+
+    langfuse.score(
+        name="chat_interaction",
+        value=1,  # Placeholder for quality score
+        comment="Successful chat"
+    )
+
+    langfuse.flush()

src/rag/retriever.py

@@ -0,0 +1,70 @@
+"""Service for retrieving relevant documents from vector store."""
+
+import hashlib
+import json
+from src.db.postgres.vector_store import get_vector_store
+from src.db.redis.connection import get_redis
+from sqlalchemy.ext.asyncio import AsyncSession
+from src.middlewares.logging import get_logger
+from typing import List, Dict, Any
+
+logger = get_logger("retriever")
+
+_RETRIEVAL_CACHE_TTL = 3600  # 1 hour
+
+
+class RetrieverService:
+    """Service for retrieving relevant documents."""
+
+    def __init__(self):
+        self.vector_store = get_vector_store()
+
+    async def retrieve(
+        self,
+        query: str,
+        user_id: str,
+        db: AsyncSession,
+        k: int = 5
+    ) -> List[Dict[str, Any]]:
+        """Retrieve relevant chunks for a query, scoped to the user's documents.
+
+        Returns:
+            List of dicts with keys: content, metadata
+            metadata includes: document_id, user_id, filename, chunk_index, page_label (if PDF)
+        """
+        try:
+            redis = await get_redis()
+            query_hash = hashlib.md5(query.encode()).hexdigest()
+            cache_key = f"retrieval:{user_id}:{query_hash}:{k}"
+
+            cached = await redis.get(cache_key)
+            if cached:
+                logger.info("Returning cached retrieval results")
+                return json.loads(cached)
+
+            logger.info(f"Retrieving for user {user_id}, query: {query[:50]}...")
+
+            docs = await self.vector_store.asimilarity_search(
+                query=query,
+                k=k,
+                filter={"user_id": user_id}
+            )
+
+            results = [
+                {
+                    "content": doc.page_content,
+                    "metadata": doc.metadata,
+                }
+                for doc in docs
+            ]
+
+            logger.info(f"Retrieved {len(results)} chunks")
+            await redis.setex(cache_key, _RETRIEVAL_CACHE_TTL, json.dumps(results))
+            return results
+
+        except Exception as e:
+            logger.error("Retrieval failed", error=str(e))
+            return []
+
+
+retriever = RetrieverService()

src/storage/az_blob/az_blob.py

@@ -0,0 +1,76 @@
+"""Azure Blob Storage client wrapper."""
+
+from azure.storage.blob.aio import BlobClient
+from src.config.settings import settings
+from src.middlewares.logging import get_logger
+import uuid
+
+logger = get_logger("azure_blob")
+
+
+class AzureBlobStorage:
+    """Azure Blob Storage async client wrapper."""
+
+    def __init__(self):
+        self.container_name = settings.azureai_container_name
+        self.sas_token = settings.azureai_blob_sas
+        self.account_url = settings.azureai_container_endpoint.rstrip('/')
+
+    def _get_blob_client(self, blob_name: str) -> BlobClient:
+        """Get async blob client with SAS token."""
+        sas_url = f"{self.account_url}/{self.container_name}/{blob_name}?{self.sas_token}"
+        return BlobClient.from_blob_url(sas_url)
+
+    async def upload_file(self, file_content: bytes, filename: str, user_id: str) -> str:
+        """Upload file to Azure Blob Storage.
+
+        Returns:
+            blob_name: Unique blob name in storage
+        """
+        try:
+            ext = filename.split('.')[-1] if '.' in filename else 'txt'
+            blob_name = f"{user_id}/{uuid.uuid4()}.{ext}"
+
+            async with self._get_blob_client(blob_name) as blob_client:
+                logger.info(f"Uploading file {filename} to blob {blob_name}")
+                await blob_client.upload_blob(file_content, overwrite=True)
+
+            logger.info(f"Successfully uploaded {blob_name}")
+            return blob_name
+
+        except Exception as e:
+            logger.error(f"Failed to upload file {filename}", error=str(e))
+            raise
+
+    async def download_file(self, blob_name: str) -> bytes:
+        """Download file from Azure Blob Storage."""
+        try:
+            async with self._get_blob_client(blob_name) as blob_client:
+                logger.info(f"Downloading blob {blob_name}")
+                stream = await blob_client.download_blob()
+                content = await stream.readall()
+
+            logger.info(f"Successfully downloaded {blob_name}")
+            return content
+
+        except Exception as e:
+            logger.error(f"Failed to download blob {blob_name}", error=str(e))
+            raise
+
+    async def delete_file(self, blob_name: str) -> bool:
+        """Delete file from Azure Blob Storage."""
+        try:
+            async with self._get_blob_client(blob_name) as blob_client:
+                logger.info(f"Deleting blob {blob_name}")
+                await blob_client.delete_blob()
+
+            logger.info(f"Successfully deleted {blob_name}")
+            return True
+
+        except Exception as e:
+            logger.error(f"Failed to delete blob {blob_name}", error=str(e))
+            return False
+
+
+# Singleton instance
+blob_storage = AzureBlobStorage()

src/tools/search.py

@@ -0,0 +1,46 @@
+"""Search tool for agent."""
+
+from langchain_core.tools import tool
+from src.rag.retriever import retriever
+from sqlalchemy.ext.asyncio import AsyncSession
+from src.middlewares.logging import get_logger
+
+logger = get_logger("search_tool")
+
+
+@tool
+async def search_documents(
+    query: str,
+    user_id: str,
+    db: AsyncSession,
+    num_results: int = 5
+) -> str:
+    """Search user's uploaded documents for relevant information.
+
+    Args:
+        query: The search query or question
+        user_id: The user's ID
+        db: Database session
+        num_results: Number of results to return (default: 5)
+
+    Returns:
+        Relevant document excerpts with source and page information
+    """
+    try:
+        results = await retriever.retrieve(query, user_id, db, num_results)
+
+        if not results:
+            return "No relevant information found in the documents."
+
+        formatted_results = []
+        for result in results:
+            filename = result["metadata"].get("filename", "Unknown")
+            page = result["metadata"].get("page_label")
+            source_label = f"{filename}, p.{page}" if page else filename
+            formatted_results.append(f"[Source: {source_label}]\n{result['content']}\n")
+
+        return "\n".join(formatted_results)
+
+    except Exception as e:
+        logger.error("Search failed", error=str(e))
+        return "Sorry, I encountered an error while searching the documents."