CAPEX initiatives above USD 10,000 require CFO approval per Policy A.
+Monthly financial statements are archived in SharePoint > Finance > Reporting.
+ROI calculations follow the formula (Net Benefit - Cost) / Cost over three years.
+ + diff --git a/data/web_samples/policy_portal.html b/data/web_samples/policy_portal.html new file mode 100644 index 0000000..f67a213 --- /dev/null +++ b/data/web_samples/policy_portal.html @@ -0,0 +1,19 @@ + + + + +Policy A ensures all strategic sourcing over USD 10,000 is reviewed by the CFO.
+Quarterly supplier audits must be uploaded to SharePoint by the 5th business day.
+Include a compliance matrix referencing Policy A approval checkpoints.
+Risk mitigation plans must map to supplier dependencies and delivery schedules.
+Submit clarifications within 48 hours to maintain audit traceability.
+ + diff --git a/main.py b/main.py new file mode 100644 index 0000000..6e020f0 --- /dev/null +++ b/main.py @@ -0,0 +1,46 @@ +"""FastAPI application entrypoint for the enterprise RAG chatbot.""" +from __future__ import annotations + +import logging +import os +from functools import lru_cache + +os.environ.setdefault("ANONYMIZED_TELEMETRY", "False") + +try: + import chromadb.telemetry.product.posthog as _chroma_ph_module + + _chroma_ph_module.posthog.capture = lambda *_a, **_kw: None # type: ignore[attr-defined] +except Exception: + pass + +from fastapi import FastAPI +from fastapi.responses import JSONResponse + +from routers import chatbot_router, crawler_router, evaluation_router +from utils.config import Settings, get_settings + +LOGGER = logging.getLogger(__name__) + + +@lru_cache() +def create_app() -> FastAPI: + settings = get_settings() + app = FastAPI( + title=settings.app_name, + version="0.1.0", + summary="Enterprise-ready Retrieval-Augmented Generation chatbot platform.", + ) + + app.include_router(chatbot_router) + app.include_router(crawler_router) + app.include_router(evaluation_router) + + @app.get("/health", tags=["health"]) + async def health_check() -> JSONResponse: + return JSONResponse({"status": "ok", "app": settings.app_name}) + + return app + + +app = create_app() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..882383c --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,51 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "enterprise-chatbot" +version = "0.1.0" +description = "Enterprise-ready RAG chatbot with evaluation and data ingestion capabilities" +authors = [{ name = "Fairoza Amira" }] +readme = "README.md" +requires-python = ">=3.10,<3.11" +dependencies = [ + "fastapi==0.109.2", + "uvicorn[standard]==0.25.0", + "langchain==0.1.20", + "langchain-community==0.0.38", + "langchainhub==0.1.17", + "langchain-experimental==0.0.55", + "langchain-openai==0.1.7", + "chromadb==0.5.0", + "pydantic==2.6.4", + "pydantic-settings==2.2.1", + "python-dotenv==1.0.1", + "deepeval==0.20.63", + "pandas==2.2.1", + "numpy==1.26.4", + "sentence-transformers==2.3.1", + "pypdf==3.17.4", + "beautifulsoup4==4.12.3", + "requests==2.31.0" +] + +[project.optional-dependencies] +dev = [ + "pytest==7.4.4", + "pytest-asyncio==0.21.1", + "httpx==0.25.2" +] + +[dependency-groups] +dev = [ + "pytest==7.4.4", + "pytest-asyncio==0.21.1", + "httpx==0.25.2" +] + +[tool.setuptools.packages.find] +include = ["sample_chatbot*"] + +[tool.setuptools.package-data] +sample_chatbot = ["py.typed"] diff --git a/routers/__init__.py b/routers/__init__.py new file mode 100644 index 0000000..0c56f41 --- /dev/null +++ b/routers/__init__.py @@ -0,0 +1,11 @@ +"""Routers package exports.""" + +from .chatbot_router import router as chatbot_router +from .crawler_router import router as crawler_router +from .evaluation_router import router as evaluation_router + +__all__ = [ + "chatbot_router", + "crawler_router", + "evaluation_router", +] diff --git a/routers/chatbot_router.py b/routers/chatbot_router.py new file mode 100644 index 0000000..8e605cb --- /dev/null +++ b/routers/chatbot_router.py @@ -0,0 +1,16 @@ +"""API router for chatbot interactions.""" +from __future__ import annotations + +from fastapi import APIRouter + +from controllers import handle_chatbot_request +from utils.schemas import ChatbotRequest, ChatbotResponse + +router = APIRouter(prefix="/api/chatbot", tags=["chatbot"]) + + +@router.post("", response_model=ChatbotResponse, summary="Query the enterprise chatbot") +async def chatbot_endpoint(payload: ChatbotRequest) -> ChatbotResponse: + """Return an answer plus citations for the supplied query.""" + + return handle_chatbot_request(payload) diff --git a/routers/crawler_router.py b/routers/crawler_router.py new file mode 100644 index 0000000..2ac0a6b --- /dev/null +++ b/routers/crawler_router.py @@ -0,0 +1,16 @@ +"""API router for data crawling operations.""" +from __future__ import annotations + +from fastapi import APIRouter + +from controllers import handle_crawl_request +from utils.schemas import CrawlRequest, CrawlResponse + +router = APIRouter(prefix="/api/crawler", tags=["crawler"]) + + +@router.post("", response_model=CrawlResponse, summary="Ingest data into the knowledge base") +async def crawler_endpoint(payload: CrawlRequest) -> CrawlResponse: + """Trigger the data ingestion workflow.""" + + return handle_crawl_request(payload) diff --git a/routers/evaluation_router.py b/routers/evaluation_router.py new file mode 100644 index 0000000..9dbb724 --- /dev/null +++ b/routers/evaluation_router.py @@ -0,0 +1,16 @@ +"""API router for chatbot evaluation.""" +from __future__ import annotations + +from fastapi import APIRouter + +from controllers import handle_evaluation_request +from utils.schemas import EvaluationRequest, EvaluationResponse + +router = APIRouter(prefix="/api/evaluation", tags=["evaluation"]) + + +@router.post("", response_model=EvaluationResponse, summary="Evaluate chatbot answer quality") +async def evaluation_endpoint(payload: EvaluationRequest) -> EvaluationResponse: + """Return evaluation metrics for a chatbot answer.""" + + return handle_evaluation_request(payload) diff --git a/scripts/start-with-ollama.sh b/scripts/start-with-ollama.sh new file mode 100644 index 0000000..cb8e14d --- /dev/null +++ b/scripts/start-with-ollama.sh @@ -0,0 +1,41 @@ +#!/usr/bin/env sh +set -eu + +OLLAMA_PID="" + +cleanup() { + if [ -n "$OLLAMA_PID" ]; then + kill "$OLLAMA_PID" >/dev/null 2>&1 || true + fi +} + +trap cleanup EXIT INT TERM + +LLM_PROVIDER="${CHATBOT_LLM_PROVIDER:-placeholder}" +LLM_MODEL="${CHATBOT_LLM_MODEL:-llama3.2}" +OLLAMA_BASE_URL="${CHATBOT_OLLAMA_BASE_URL:-http://127.0.0.1:11434}" + +if [ "$LLM_PROVIDER" = "ollama" ]; then + echo "Starting Ollama server..." + ollama serve >/tmp/ollama.log 2>&1 & + OLLAMA_PID=$! + + echo "Waiting for Ollama to become ready..." + READY=0 + for i in $(seq 1 30); do + if curl -fsS "${OLLAMA_BASE_URL%/}/api/tags" >/dev/null 2>&1; then + READY=1 + break + fi + sleep 1 + done + + if [ "$READY" -eq 1 ]; then + echo "Pulling model: $LLM_MODEL" + ollama pull "$LLM_MODEL" + else + echo "Warning: Ollama did not become ready in time; continuing without model pull." + fi +fi + +exec uvicorn main:app --host 0.0.0.0 --port 8000 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..e7661a9 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from main import create_app +from utils.config import get_settings + + +@pytest.fixture(autouse=True) +def reset_settings(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + data_dir = tmp_path / "data" + qa_dir = tmp_path / "qa" + chroma_dir = tmp_path / "chroma" + data_dir.mkdir(parents=True, exist_ok=True) + qa_dir.mkdir(parents=True, exist_ok=True) + chroma_dir.mkdir(parents=True, exist_ok=True) + + monkeypatch.setenv("CHATBOT_DATA_DIRECTORY", str(data_dir)) + monkeypatch.setenv("CHATBOT_QA_CACHE_DIRECTORY", str(qa_dir)) + monkeypatch.setenv("CHATBOT_CHROMA_PERSIST_DIRECTORY", str(chroma_dir)) + monkeypatch.setenv("CHATBOT_APP_ENV", "test") + + # Clear cached settings and app to load new environment variables per test. + get_settings.cache_clear() + create_app.cache_clear() + + +@pytest.fixture() +def client() -> TestClient: + app = create_app() + return TestClient(app) + + +def _seed_sample_documents(data_dir: Path) -> None: + knowledge_dir = data_dir / "knowledge" + knowledge_dir.mkdir(parents=True, exist_ok=True) + (knowledge_dir / "policy.txt").write_text( + "Procurement must follow policy A. Financial approvals require CFO sign-off.", + encoding="utf-8", + ) + + tender_dir = data_dir / "tender" + tender_dir.mkdir(parents=True, exist_ok=True) + (tender_dir / "tender_notes.txt").write_text( + "Tender drafting should highlight compliance with policy A and include risk mitigation steps.", + encoding="utf-8", + ) + + +def test_health_endpoint(client: TestClient) -> None: + response = client.get("/health") + assert response.status_code == 200 + payload = response.json() + assert payload["status"] == "ok" + + +def test_crawler_and_chatbot_flow(client: TestClient, tmp_path: Path) -> None: + settings = get_settings() + _seed_sample_documents(settings.data_directory) + + crawl_payload = { + "refresh_index": True, + "max_qas_per_document": 1, + "website_urls": ["https://example.com/mock"], + } + crawl_response = client.post("/api/crawler", json=crawl_payload) + assert crawl_response.status_code == 200 + crawl_body = crawl_response.json() + assert crawl_body["documents_indexed"] >= 2 + assert crawl_body["qa_pairs_generated"] >= 2 + assert "timeTaken" in crawl_body + assert "apiCost" in crawl_body + + chatbot_payload = {"query": "What does policy A require?", "domain": "knowledge", "top_k": 3} + chatbot_response = client.post("/api/chatbot", json=chatbot_payload) + assert chatbot_response.status_code == 200 + chatbot_body = chatbot_response.json() + assert chatbot_body["answer"] + assert isinstance(chatbot_body["citations"], list) + assert "timeTaken" in chatbot_body + assert "apiCost" in chatbot_body + + +def test_evaluation_endpoint(client: TestClient) -> None: + evaluation_payload = { + "question": "What is policy A?", + "answer": "Policy A ensures compliance with procurement rules.", + "references": [{"context": "Policy A covers procurement compliance."}], + } + response = client.post("/api/evaluation", json=evaluation_payload) + assert response.status_code == 200 + body = response.json() + assert body["average_score"] >= 0 + assert any(metric["metric"] == "ReferenceOverlap" for metric in body["metrics"]) + assert "timeTaken" in body + assert "apiCost" in body diff --git a/utils/__init__.py b/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils/config.py b/utils/config.py new file mode 100644 index 0000000..52485b5 --- /dev/null +++ b/utils/config.py @@ -0,0 +1,72 @@ +"""Configuration utilities for the enterprise chatbot project.""" +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path +from typing import Optional + +from pydantic_settings import BaseSettings +from pydantic import Field + + +class Settings(BaseSettings): + """Application settings loaded from environment variables or .env.""" + + app_name: str = "Enterprise RAG Chatbot" + app_env: str = Field(default="development", description="Application environment.") + # Placeholder API key for demonstration purposes. + llm_api_key: Optional[str] = Field( + default=None, + description="API key for the primary language model provider.", + ) + llm_model: str = Field(default="gpt5-mini", description="Default LLM model identifier.") + llm_provider: str = Field( + default="placeholder", + description="LLM provider to use: placeholder, openai, or ollama.", + ) + ollama_base_url: str = Field( + default="http://127.0.0.1:11434", + description="Base URL for a local Ollama server.", + ) + ollama_timeout_seconds: float = Field( + default=300.0, + description="HTTP timeout for Ollama generation requests.", + ) + llm_cost_per_1k_tokens: float = Field( + default=0.0, + description="Estimated USD cost per 1,000 tokens for the configured LLM.", + ) + + enable_deepeval: bool = Field( + default=True, + description="Toggle to enable deepeval-based metrics when configuration allows.", + ) + + chroma_persist_directory: Path = Field( + default=Path(".chroma_store"), + description="Directory where Chroma vector store is persisted.", + ) + data_directory: Path = Field( + default=Path("data"), + description="Root directory for local document ingestion.", + ) + qa_cache_directory: Path = Field( + default=Path("generated_qa"), + description="Directory where generated Q&A pairs are stored.", + ) + + class Config: + env_prefix = "CHATBOT_" + env_file = ".env" + case_sensitive = False + + +@lru_cache() +def get_settings() -> Settings: + """Return a cached instance of application settings.""" + + settings = Settings() + settings.chroma_persist_directory.mkdir(parents=True, exist_ok=True) + settings.data_directory.mkdir(parents=True, exist_ok=True) + settings.qa_cache_directory.mkdir(parents=True, exist_ok=True) + return settings diff --git a/utils/costs.py b/utils/costs.py new file mode 100644 index 0000000..8485b68 --- /dev/null +++ b/utils/costs.py @@ -0,0 +1,37 @@ +"""Cost estimation utilities for API usage.""" +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class CostEstimate: + """Represents an estimated token usage and cost.""" + + input_tokens: int + output_tokens: int + total_tokens: int + estimated_cost_usd: float + + +def _estimate_tokens(text: str) -> int: + """Approximate token count from a text blob using a heuristic.""" + + if not text: + return 0 + return max(1, len(text) // 4) + + +def estimate_cost(prompt: str, completion: str, rate_per_1k: float = 0.0) -> CostEstimate: + """Return token usage and rough cost given rate per thousand tokens.""" + + input_tokens = _estimate_tokens(prompt) + output_tokens = _estimate_tokens(completion) + total_tokens = input_tokens + output_tokens + estimated_cost_usd = (total_tokens / 1000) * rate_per_1k + return CostEstimate( + input_tokens=input_tokens, + output_tokens=output_tokens, + total_tokens=total_tokens, + estimated_cost_usd=round(estimated_cost_usd, 6), + ) diff --git a/utils/data_ingestion.py b/utils/data_ingestion.py new file mode 100644 index 0000000..cda332f --- /dev/null +++ b/utils/data_ingestion.py @@ -0,0 +1,351 @@ +"""Data ingestion and QA generation utilities for the RAG system.""" +from __future__ import annotations + +import json +import logging +import shutil +from dataclasses import dataclass +from datetime import datetime, timezone +from email import policy +from email.parser import BytesParser +from pathlib import Path +from typing import Iterable, List, Sequence + +import pandas as pd +import requests +from bs4 import BeautifulSoup +from langchain.docstore.document import Document +from langchain.text_splitter import RecursiveCharacterTextSplitter +from pypdf import PdfReader + +from .config import Settings, get_settings +from .identifiers import generate_request_id +from .llm import PlaceholderLLM, get_llm +from .schemas import ChatDomain +from .vectorstore import upsert_documents + +LOGGER = logging.getLogger(__name__) + + +@dataclass +class QAPair: + """Represents a generated question/answer pair.""" + + question: str + answer: str + source: str + domain: str + + +@dataclass +class CrawlSummary: + """Summary returned after a data ingestion run.""" + + documents_indexed: int + qa_pairs_generated: int + qa_output_path: Path | None + + +class DataCrawler: + """Crawls local data sources, optional sample websites, and generates QA pairs.""" + + SUPPORTED_SUFFIXES = { + ".txt", + ".md", + ".csv", + ".json", + ".pdf", + ".eml", + ".log", + } + + SAMPLE_WEBSITE_FILES = { + "knowledge": "policy_portal.html", + "tender": "tender_hub.html", + "finance": "finance_updates.html", + } + + def __init__(self, settings: Settings | None = None) -> None: + self.settings = settings or get_settings() + self.llm = get_llm() + self.splitter = RecursiveCharacterTextSplitter(chunk_size=900, chunk_overlap=120) + + def run( + self, + refresh_index: bool, + max_qas_per_document: int, + website_urls: Sequence[str] | None = None, + ) -> CrawlSummary: + """Execute the end-to-end ingestion workflow.""" + + LOGGER.info("Starting crawl: refresh_index=%s, max_qas=%s", refresh_index, max_qas_per_document) + if refresh_index: + self._reset_vector_store() + + raw_documents = list(self._load_documents()) + raw_documents.extend(self._load_sample_websites()) + raw_documents.extend(self._fetch_external_websites(website_urls or [])) + LOGGER.info("Loaded %d raw documents from %s", len(raw_documents), self.settings.data_directory) + + chunked_documents = self._split_documents(raw_documents) + LOGGER.info("Chunked documents into %d segments", len(chunked_documents)) + + qa_pairs = self._generate_qa_pairs(raw_documents, max_qas_per_document) + qa_documents = self._qa_pairs_to_documents(qa_pairs) + + total_upserted = upsert_documents([*chunked_documents, *qa_documents]) + LOGGER.info("Persisted %d documents into the vector store", total_upserted) + + qa_output_path = self._persist_qa_pairs(qa_pairs) if qa_pairs else None + + return CrawlSummary( + documents_indexed=len(raw_documents), + qa_pairs_generated=len(qa_pairs), + qa_output_path=qa_output_path, + ) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + def _reset_vector_store(self) -> None: + """Remove the existing vector store persistence layer.""" + + persist_directory = self.settings.chroma_persist_directory + if persist_directory.exists(): + LOGGER.info("Removing existing Chroma directory at %s", persist_directory) + shutil.rmtree(persist_directory) + persist_directory.mkdir(parents=True, exist_ok=True) + + def _load_documents(self) -> Iterable[Document]: + """Yield documents from the configured data directory.""" + + data_dir = self.settings.data_directory + if not data_dir.exists(): + LOGGER.warning("Data directory %s does not exist; no documents loaded.", data_dir) + return [] + + for path in sorted(data_dir.rglob("*")): + if not path.is_file() or path.suffix.lower() not in self.SUPPORTED_SUFFIXES: + continue + yield from self._load_file(path) + + def _infer_domain(self, path: Path) -> str: + """Infer the domain from the file path.""" + + try: + relative_parts = path.relative_to(self.settings.data_directory).parts + except ValueError: + relative_parts = () + + if relative_parts: + candidate = relative_parts[0].lower() + if candidate in {domain.value for domain in ChatDomain}: + return candidate + return ChatDomain.knowledge.value + + def _build_metadata(self, path: Path, file_type: str) -> dict[str, str]: + domain = self._infer_domain(path) + return { + "id": generate_request_id(), + "source": str(path), + "file_name": path.name, + "file_type": file_type, + "domain": domain, + "ingested_at": datetime.now(timezone.utc).isoformat(), + } + + def _load_file(self, path: Path) -> List[Document]: + suffix = path.suffix.lower() + metadata = self._build_metadata(path, file_type=suffix.strip(".")) + + try: + if suffix in {".txt", ".md", ".log"}: + content = path.read_text(encoding="utf-8", errors="ignore") + return [Document(page_content=content, metadata=metadata)] + if suffix == ".json": + data = json.loads(path.read_text(encoding="utf-8", errors="ignore")) + serialized = json.dumps(data, indent=2, ensure_ascii=False) + return [Document(page_content=serialized, metadata=metadata)] + if suffix == ".csv": + df = pd.read_csv(path) + table_text = df.to_markdown(index=False) + return [Document(page_content=table_text, metadata=metadata)] + if suffix == ".pdf": + reader = PdfReader(str(path)) + documents: List[Document] = [] + for page_number, page in enumerate(reader.pages, start=1): + text = page.extract_text() or "" + page_metadata = {**metadata, "page": page_number} + documents.append(Document(page_content=text, metadata=page_metadata)) + return documents + if suffix == ".eml": + with path.open("rb") as email_file: + message = BytesParser(policy=policy.default).parse(email_file) + body = message.get_body(preferencelist=("plain", "html")) + content = body.get_content() if body else "" + headers = "\n".join(f"{k}: {v}" for k, v in message.items()) + full_text = f"{headers}\n\n{content}" + return [Document(page_content=full_text, metadata=metadata)] + except Exception as exc: # pragma: no cover - ingestion failures are logged + LOGGER.exception("Failed to load file %s: %s", path, exc) + return [] + + LOGGER.debug("Skipping unsupported file: %s", path) + return [] + + def _load_sample_websites(self) -> List[Document]: + """Fetch sample web pages to demonstrate external data ingestion.""" + + documents: List[Document] = [] + sample_directory = self.settings.data_directory / "web_samples" + sample_directory.mkdir(parents=True, exist_ok=True) + + for domain, file_name in self.SAMPLE_WEBSITE_FILES.items(): + html_path = sample_directory / file_name + if not html_path.exists(): + LOGGER.debug("Sample website file missing for domain %s: %s", domain, html_path) + continue + + try: + html_text = html_path.read_text(encoding="utf-8") + except OSError as exc: + LOGGER.warning("Unable to read sample website %s: %s", html_path, exc) + continue + + soup = BeautifulSoup(html_text, "html.parser") + text = soup.get_text(separator="\n") + if not text.strip(): + continue + + metadata = { + "id": generate_request_id(), + "source": str(html_path.resolve()), + "file_name": file_name, + "file_type": "html", + "domain": domain, + "ingested_at": datetime.now(timezone.utc).isoformat(), + "type": "web_sample", + } + documents.append(Document(page_content=text, metadata=metadata)) + return documents + + def _fetch_external_websites(self, urls: Sequence[str]) -> List[Document]: + documents: List[Document] = [] + if not urls: + return documents + + for url in urls: + try: + LOGGER.info("Fetching external website: %s", url) + response = requests.get(url, timeout=10) + response.raise_for_status() + except requests.RequestException as exc: + LOGGER.warning("Skipping website %s due to error: %s", url, exc) + continue + + soup = BeautifulSoup(response.text, "html.parser") + text = soup.get_text(separator="\n") + if not text.strip(): + LOGGER.debug("Website %s yielded empty content; skipped", url) + continue + + metadata = { + "id": generate_request_id(), + "source": url, + "file_name": url, + "file_type": "html", + "domain": ChatDomain.knowledge.value, + "ingested_at": datetime.now(timezone.utc).isoformat(), + "type": "web_fetch", + } + documents.append(Document(page_content=text, metadata=metadata)) + return documents + + def _split_documents(self, documents: Iterable[Document]) -> List[Document]: + chunks: List[Document] = [] + for document in documents: + splits = self.splitter.split_documents([document]) + for idx, split in enumerate(splits): + split.metadata.setdefault("chunk_index", idx) + split.metadata.setdefault("parent_source", document.metadata.get("source")) + split.metadata.setdefault("domain", document.metadata.get("domain")) + split.metadata.setdefault("type", "chunk") + chunks.extend(splits) + return chunks + + def _generate_qa_pairs(self, documents: Iterable[Document], max_pairs: int) -> List[QAPair]: + if max_pairs <= 0: + return [] + + qa_pairs: List[QAPair] = [] + for document in documents: + content = document.page_content.strip() + if not content: + continue + segments = [segment.strip() for segment in content.split("\n\n") if len(segment.strip()) > 40] + if not segments: + segments = [content] + + for index, segment in enumerate(segments[:max_pairs]): + question = self._craft_question(document, index) + answer = self._craft_answer(segment) + qa_pairs.append( + QAPair( + question=question, + answer=answer, + source=document.metadata.get("source", "unknown"), + domain=document.metadata.get("domain", ChatDomain.knowledge.value), + ) + ) + + return qa_pairs + + def _craft_question(self, document: Document, index: int) -> str: + domain = document.metadata.get("domain", ChatDomain.knowledge.value) + source_name = document.metadata.get("file_name", "document") + return ( + f"What are the key insights from section {index + 1} of {source_name} " + f"within the {domain} knowledge base?" + ) + + def _craft_answer(self, segment: str) -> str: + if isinstance(self.llm, PlaceholderLLM): + return segment[:600] + + try: + prompt = ( + "You are generating knowledge-base Q&A pairs. Summarize the following segment " + "into one concise answer limited to 5 sentences. Segment:\n\n{segment}" + ) + response = self.llm.predict(prompt.format(segment=segment)) + return response.strip() + except Exception as exc: # pragma: no cover - LLM failures defer to heuristic answer + LOGGER.warning("LLM QA generation failed; using heuristic answer: %s", exc) + return segment[:600] + + def _qa_pairs_to_documents(self, qa_pairs: Iterable[QAPair]) -> List[Document]: + documents: List[Document] = [] + for pair in qa_pairs: + metadata = { + "source": pair.source, + "domain": pair.domain, + "type": "generated_qa", + "qa_id": generate_request_id(), + } + content = f"Question: {pair.question}\nAnswer: {pair.answer}" + documents.append(Document(page_content=content, metadata=metadata)) + return documents + + def _persist_qa_pairs(self, qa_pairs: Iterable[QAPair]) -> Path | None: + pairs = list(qa_pairs) + if not pairs: + return None + + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + output_path = self.settings.qa_cache_directory / f"qa_pairs_{timestamp}.json" + payload = [pair.__dict__ for pair in pairs] + output_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") + return output_path + + +__all__ = ["DataCrawler", "QAPair", "CrawlSummary"] diff --git a/utils/embeddings.py b/utils/embeddings.py new file mode 100644 index 0000000..0ae447c --- /dev/null +++ b/utils/embeddings.py @@ -0,0 +1,40 @@ +"""Embedding utilities for the RAG system.""" +from __future__ import annotations + +import logging +from functools import lru_cache +from typing import Iterable, List + +from langchain.embeddings.base import Embeddings +from langchain_community.embeddings import HuggingFaceEmbeddings + +LOGGER = logging.getLogger(__name__) + +DEFAULT_MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2" + + +class FallbackEmbeddings(Embeddings): + """Deterministic embedding generator that avoids external downloads.""" + + def _embed(self, text: str) -> List[float]: + vector = [0.0] * 32 + for index, char in enumerate(text.encode("utf-8")): + vector[index % len(vector)] += (char % 31) / 100.0 + return vector + + def embed_documents(self, texts: Iterable[str]) -> List[List[float]]: # type: ignore[override] + return [self._embed(text) for text in texts] + + def embed_query(self, text: str) -> List[float]: # type: ignore[override] + return self._embed(text) + + +@lru_cache() +def get_embeddings() -> Embeddings: + """Return a cached embedding model instance.""" + + try: + return HuggingFaceEmbeddings(model_name=DEFAULT_MODEL_NAME) + except Exception as exc: # pragma: no cover - optional runtime failure path + LOGGER.warning("Falling back to deterministic embeddings due to error: %s", exc) + return FallbackEmbeddings() diff --git a/utils/evaluation.py b/utils/evaluation.py new file mode 100644 index 0000000..3985f97 --- /dev/null +++ b/utils/evaluation.py @@ -0,0 +1,138 @@ +"""Evaluation utilities leveraging deepeval with graceful degradation.""" +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import List + +from .config import get_settings +from .schemas import EvaluationMetricResult, EvaluationReference + +LOGGER = logging.getLogger(__name__) + +try: # pragma: no cover - optional dependency branch + from deepeval.dataset import EvaluationSample + from deepeval.evaluate import evaluate + from deepeval.metrics import ( + AnswerRelevancyMetric, + ContextualPrecisionMetric, + FaithfulnessMetric, + ) + + HAS_DEEPEVAL = True +except Exception as exc: # pragma: no cover - offline fallback + LOGGER.warning("deepeval is unavailable: %s", exc) + HAS_DEEPEVAL = False + + +@dataclass +class EvaluationSummary: + """Aggregated evaluation metrics and average score.""" + + metrics: List[EvaluationMetricResult] + average_score: float + + +class EvaluationService: + """High-level evaluator for chatbot answers.""" + + def __init__(self) -> None: + self.settings = get_settings() + + def evaluate(self, question: str, answer: str, references: List[EvaluationReference]) -> EvaluationSummary: + if HAS_DEEPEVAL and self.settings.enable_deepeval and self.settings.llm_api_key: + return self._evaluate_with_deepeval(question, answer, references) + LOGGER.info("Falling back to heuristic evaluation; deepeval unavailable or not configured.") + return self._evaluate_with_heuristics(question, answer, references) + + # ------------------------------------------------------------------ + # Deepeval-backed evaluation + # ------------------------------------------------------------------ + def _evaluate_with_deepeval( + self, question: str, answer: str, references: List[EvaluationReference] + ) -> EvaluationSummary: + contexts = [ref.context for ref in references if ref.context.strip()] + if not contexts: + contexts = ["No reference context supplied."] + + samples = [ + EvaluationSample( + input=question, + actual_output=answer, + expected_output="\n\n".join(contexts), + context=contexts, + ) + ] + + metrics = [ + FaithfulnessMetric(model=self.settings.llm_model, api_key=self.settings.llm_api_key or ""), + ContextualPrecisionMetric(model=self.settings.llm_model, api_key=self.settings.llm_api_key or ""), + AnswerRelevancyMetric(model=self.settings.llm_model, api_key=self.settings.llm_api_key or ""), + ] + + try: + evaluation_result = evaluate(samples=samples, metrics=metrics) + except Exception as exc: # pragma: no cover - runtime fallback + LOGGER.warning("deepeval execution failed; switching to heuristics: %s", exc) + return self._evaluate_with_heuristics(question, answer, references) + + metric_results: List[EvaluationMetricResult] = [] + total_score = 0.0 + for metric in evaluation_result.results[0].metrics: # type: ignore[attr-defined] + score = getattr(metric, "score", 0.0) + passed = getattr(metric, "passed", bool(score >= 0.5)) + feedback = getattr(metric, "reason", None) + total_score += score + metric_results.append( + EvaluationMetricResult( + metric=getattr(metric, "name", metric.__class__.__name__), + score=score, + passed=passed, + feedback=feedback, + ) + ) + + average_score = total_score / max(len(metric_results), 1) + return EvaluationSummary(metrics=metric_results, average_score=round(average_score, 4)) + + # ------------------------------------------------------------------ + # Heuristic fallback + # ------------------------------------------------------------------ + def _evaluate_with_heuristics( + self, question: str, answer: str, references: List[EvaluationReference] + ) -> EvaluationSummary: + answer_lower = answer.lower() + reference_text = " ".join(ref.context.lower() for ref in references) + overlap_ratio = self._calculate_overlap(answer_lower, reference_text) + question_terms = set(token for token in question.lower().split() if len(token) > 3) + answer_terms = set(token for token in answer_lower.split()) + coverage_ratio = len(question_terms & answer_terms) / max(len(question_terms), 1) + + metric_results = [ + EvaluationMetricResult( + metric="ReferenceOverlap", + score=round(overlap_ratio, 4), + passed=overlap_ratio >= 0.45, + feedback="Checks for hallucination via reference overlap.", + ), + EvaluationMetricResult( + metric="QuestionCoverage", + score=round(coverage_ratio, 4), + passed=coverage_ratio >= 0.4, + feedback="Ensures answer addresses major terms in the question.", + ), + ] + + average_score = sum(metric.score for metric in metric_results) / len(metric_results) + return EvaluationSummary(metrics=metric_results, average_score=round(average_score, 4)) + + @staticmethod + def _calculate_overlap(answer: str, reference: str) -> float: + if not answer.strip() or not reference.strip(): + return 0.0 + answer_tokens = set(answer.split()) + reference_tokens = set(reference.split()) + return len(answer_tokens & reference_tokens) / max(len(answer_tokens), 1) + + +__all__ = ["EvaluationService", "EvaluationSummary"] diff --git a/utils/identifiers.py b/utils/identifiers.py new file mode 100644 index 0000000..798af5c --- /dev/null +++ b/utils/identifiers.py @@ -0,0 +1,10 @@ +"""Helpers for generating request identifiers.""" +from __future__ import annotations + +import uuid + + +def generate_request_id() -> str: + """Return a unique request identifier.""" + + return f"req_{uuid.uuid4().hex}" diff --git a/utils/llm.py b/utils/llm.py new file mode 100644 index 0000000..258f032 --- /dev/null +++ b/utils/llm.py @@ -0,0 +1,114 @@ +"""LLM utilities including a placeholder implementation for offline development.""" +from __future__ import annotations + +import logging +from typing import Any, List, Optional + +import requests + +from langchain.callbacks.manager import CallbackManagerForLLMRun +from langchain.llms.base import LLM + +from .config import get_settings + +LOGGER = logging.getLogger(__name__) + + +class PlaceholderLLM(LLM): + """A minimalistic LLM stub used when no external model is configured.""" + + model_name: str = "placeholder-gpt5-mini" + + @property + def _llm_type(self) -> str: # type: ignore[override] + return "placeholder" + + def _call( # type: ignore[override] + self, + prompt: str, + stop: Optional[List[str]] = None, + run_manager: Optional[CallbackManagerForLLMRun] = None, + **kwargs: Any, + ) -> str: + if stop: + LOGGER.debug("Stop tokens provided to PlaceholderLLM but are ignored: %s", stop) + LOGGER.info("Generating placeholder response for prompt of length %s", len(prompt)) + return ( + "[Placeholder LLM Response]\n" + "This environment is configured without external LLM access.\n" + "Prompt summary: " + f"{prompt[:200]}{'...' if len(prompt) > 200 else ''}" + ) + + +class OllamaLLM(LLM): + """Lightweight Ollama client for local open-source model inference.""" + + model_name: str = "llama3.2" + base_url: str = "http://127.0.0.1:11434" + timeout_seconds: float = 300.0 + + @property + def _llm_type(self) -> str: # type: ignore[override] + return "ollama" + + def _call( # type: ignore[override] + self, + prompt: str, + stop: Optional[List[str]] = None, + run_manager: Optional[CallbackManagerForLLMRun] = None, + **kwargs: Any, + ) -> str: + if stop: + LOGGER.debug("Stop tokens provided to OllamaLLM but are ignored: %s", stop) + url = f"{self.base_url.rstrip('/')}/api/generate" + payload = { + "model": self.model_name, + "prompt": prompt, + "stream": False, + } + try: + response = requests.post(url, json=payload, timeout=self.timeout_seconds) + response.raise_for_status() + data = response.json() + return str(data.get("response", "")).strip() + except Exception as exc: + LOGGER.warning("Ollama request failed (%s). Falling back to placeholder output.", exc) + return ( + "[Ollama Unavailable - Placeholder Response]\n" + "Could not reach local Ollama server.\n" + "Prompt summary: " + f"{prompt[:200]}{'...' if len(prompt) > 200 else ''}" + ) + + +def get_llm() -> LLM: + """Return an LLM instance based on configuration.""" + + settings = get_settings() + provider = settings.llm_provider.lower().strip() + + if provider == "ollama": + LOGGER.info("Using Ollama provider at %s with model %s", settings.ollama_base_url, settings.llm_model) + return OllamaLLM( + model_name=settings.llm_model, + base_url=settings.ollama_base_url, + timeout_seconds=settings.ollama_timeout_seconds, + ) + + if settings.llm_api_key: + try: + from langchain_openai import ChatOpenAI + + return ChatOpenAI( + model=settings.llm_model, + openai_api_key=settings.llm_api_key, + temperature=0.3, + ) + except ImportError as exc: # pragma: no cover - optional dependency + LOGGER.warning( + "langchain-openai is not installed; falling back to placeholder LLM: %s", + exc, + ) + LOGGER.info("Using placeholder LLM (no external provider configured).") + return PlaceholderLLM() diff --git a/utils/rag.py b/utils/rag.py new file mode 100644 index 0000000..4ec378a --- /dev/null +++ b/utils/rag.py @@ -0,0 +1,102 @@ +"""Core Retrieval-Augmented Generation pipeline orchestration.""" +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import List + +from langchain_core.documents import Document +from langchain_core.prompts import PromptTemplate + +from .config import get_settings +from .costs import estimate_cost +from .identifiers import generate_request_id +from .llm import get_llm +from .schemas import ChatbotResponse, Citation, ChatDomain, CostBreakdown +from .timing import track_duration +from .vectorstore import get_vector_store + +LOGGER = logging.getLogger(__name__) + +ANSWER_PROMPT = PromptTemplate.from_template( + ( + "You are an enterprise compliance and procurement assistant. " + "You must strictly answer using the given context.\n\n" + "Context:\n{context}\n\n" + "User question: {question}\n" + "Guidelines:\n" + "- Quote relevant facts from the context and cite their source IDs.\n" + "- If the answer is unavailable, respond with 'I'm unable to locate that information in the current knowledge base.'\n" + "- Maintain professional, concise tone." + ) +) + + +@dataclass +class RetrievedContext: + """Wrapper containing retrieved documents and the generated answer.""" + + answer: str + documents: List[Document] + + +class RAGPipeline: + """High-level interface for answering questions using RAG.""" + + def __init__(self) -> None: + self.settings = get_settings() + self.llm = get_llm() + + def _build_chain(self, domain: ChatDomain, top_k: int): + vector_store = get_vector_store() + # Retrieve a larger candidate set, then apply domain filtering in-process. + candidate_k = max(top_k * 4, top_k) + retriever = vector_store.as_retriever(search_kwargs={"k": candidate_k}) + return retriever + + def answer(self, query: str, domain: ChatDomain, top_k: int) -> ChatbotResponse: + with track_duration() as elapsed_ms: + retriever = self._build_chain(domain, top_k) + LOGGER.info("Retrieving documents for query: %s", query) + candidates = retriever.invoke(query) + filtered_documents = [doc for doc in candidates if doc.metadata.get("domain") == domain.value] + documents = (filtered_documents or candidates)[:top_k] + context = "\n\n".join(doc.page_content for doc in documents if doc.page_content) + prompt_text = ANSWER_PROMPT.format(context=context, question=query) + LOGGER.info("Invoking LLM for query: %s", query) + answer = self.llm.invoke(prompt_text) + # Ensure answer is string + if hasattr(answer, "content"): + answer = answer.content + else: + answer = str(answer) + + citations = [ + Citation( + source=doc.metadata.get("source", "unknown"), + snippet=(doc.page_content or "")[:400], + score=doc.metadata.get("score"), + ) + for doc in documents + ] + cost_breakdown = CostBreakdown() + api_cost = None + if self.settings.llm_api_key: + estimate = estimate_cost(query, answer, rate_per_1k=self.settings.llm_cost_per_1k_tokens) + cost_breakdown = CostBreakdown(**estimate.__dict__) + api_cost = estimate.estimated_cost_usd + + response = ChatbotResponse( + id=generate_request_id(), + created_at=datetime.now(timezone.utc), + time_taken_ms=elapsed_ms(), + api_cost=api_cost, + cost=cost_breakdown, + answer=answer.strip(), + citations=citations, + ) + return response + + +__all__ = ["RAGPipeline"] diff --git a/utils/schemas.py b/utils/schemas.py new file mode 100644 index 0000000..ec8e7ee --- /dev/null +++ b/utils/schemas.py @@ -0,0 +1,122 @@ +"""Pydantic schemas shared across controllers and routers.""" +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from typing import List, Optional + +from pydantic import BaseModel, Field, ConfigDict + + +class ChatDomain(str, Enum): + """Supported knowledge domains for the chatbot.""" + + knowledge = "knowledge" + tender = "tender" + finance = "finance" + + +class Citation(BaseModel): + """Represents contextual evidence backing a chatbot answer.""" + + source: str + snippet: str + score: Optional[float] = None + + +class CostBreakdown(BaseModel): + """Tracks token usage and monetary cost of an interaction.""" + + input_tokens: int = 0 + output_tokens: int = 0 + total_tokens: int = 0 + estimated_cost_usd: float = 0.0 + + +class BaseAPIResponse(BaseModel): + """Common metadata returned by all API endpoints.""" + + model_config = ConfigDict(populate_by_name=True) + + id: str = Field(..., description="Primary key identifier for the response") + created_at: datetime = Field( + ..., + alias="createdAt", + description="UTC timestamp when the response was generated", + ) + time_taken_ms: float = Field( + ..., + alias="timeTaken", + description="Processing time in milliseconds", + ) + api_cost: Optional[float] = Field( + default=None, + alias="apiCost", + description="Estimated USD cost billed by the LLM provider.", + ) + cost: CostBreakdown = Field(default_factory=CostBreakdown) + + +class ChatbotRequest(BaseModel): + """Incoming request payload for chatbot interactions.""" + + query: str = Field(..., min_length=3, description="User's question or instruction") + domain: ChatDomain = Field(default=ChatDomain.knowledge) + top_k: int = Field(default=4, ge=1, le=10) + + +class ChatbotResponse(BaseAPIResponse): + """Response payload for chatbot interactions.""" + + answer: str + citations: List[Citation] = Field(default_factory=list) + + +class CrawlRequest(BaseModel): + """Incoming payload for the data crawler endpoint.""" + + refresh_index: bool = Field(default=True, description="Whether to rebuild the index from scratch") + max_qas_per_document: int = Field(default=2, ge=0, le=10) + website_urls: List[str] = Field( + default_factory=list, + description="Optional external website URLs to crawl into the knowledge base.", + ) + + +class CrawlResponse(BaseAPIResponse): + """Response payload describing ingestion results.""" + + documents_indexed: int + qa_pairs_generated: int + qa_output_path: Optional[str] = None + + +class EvaluationReference(BaseModel): + """Reference materials for evaluation.""" + + context: str + source: Optional[str] = None + + +class EvaluationRequest(BaseModel): + """Incoming payload for evaluation endpoint.""" + + question: str + answer: str + references: List[EvaluationReference] + + +class EvaluationMetricResult(BaseModel): + """Individual metric outcome from evaluation.""" + + metric: str + score: float + passed: bool + feedback: Optional[str] = None + + +class EvaluationResponse(BaseAPIResponse): + """Aggregated evaluation result.""" + + metrics: List[EvaluationMetricResult] + average_score: float diff --git a/utils/timing.py b/utils/timing.py new file mode 100644 index 0000000..accca1b --- /dev/null +++ b/utils/timing.py @@ -0,0 +1,18 @@ +"""Utilities for measuring execution duration.""" +from __future__ import annotations + +import time +from contextlib import contextmanager +from typing import Generator + + +@contextmanager +def track_duration() -> Generator[callable, None, None]: + """Context manager that yields a callable returning elapsed milliseconds.""" + + start = time.perf_counter() + + def _elapsed_ms() -> float: + return (time.perf_counter() - start) * 1000 + + yield _elapsed_ms diff --git a/utils/vectorstore.py b/utils/vectorstore.py new file mode 100644 index 0000000..27644e1 --- /dev/null +++ b/utils/vectorstore.py @@ -0,0 +1,37 @@ +"""Vector store utilities for persistence and retrieval.""" +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Iterable, List + +from langchain.docstore.document import Document +from langchain_community.vectorstores import Chroma + +from .config import get_settings +from .embeddings import get_embeddings + +LOGGER = logging.getLogger(__name__) + + +def get_vector_store(persist_directory: Path | None = None) -> Chroma: + """Return a Chroma vector store instance with persistent storage.""" + + settings = get_settings() + directory = persist_directory or settings.chroma_persist_directory + directory.mkdir(parents=True, exist_ok=True) + LOGGER.debug("Using Chroma persist directory: %s", directory) + return Chroma(persist_directory=str(directory), embedding_function=get_embeddings()) + + +def upsert_documents(documents: Iterable[Document]) -> int: + """Upsert documents into the vector store and return the number of records.""" + + docs: List[Document] = list(documents) + if not docs: + LOGGER.info("No documents provided for upsert.") + return 0 + LOGGER.info("Upserting %d documents into Chroma store", len(docs)) + vector_store = get_vector_store() + vector_store.add_documents(docs) + return len(docs)