Create rag_email_assistant_haystack_2_pydantic_ai_gradio_modular_2025_baseline.py

rag_email_assistant_haystack_2_pydantic_ai_gradio_modular_2025_baseline.py

@@ -0,0 +1,558 @@
+# Project layout (place files as shown)
+# ├── app/
+# │   ├── __init__.py
+# │   ├── config.py
+# │   ├── logging_setup.py
+# │   ├── models.py
+# │   ├── utils/
+# │   │   ├── __init__.py
+# │   │   └── markdown_loader.py
+# │   ├── retriever/
+# │   │   ├── __init__.py
+# │   │   ├── indexer.py
+# │   │   └── pipeline.py
+# │   ├── agents/
+# │   │   ├── __init__.py
+# │   │   ├── llm_client.py
+# │   │   ├── intent_extractor.py
+# │   │   ├── composer.py
+# │   │   └── fact_checker.py
+# │   ├── gradio_app.py
+# │   └── main.py
+# ├── requirements.txt
+# └── README.md
+
+# ===========================
+# requirements.txt
+# ===========================
+# Pin reasonably recent, stable versions (2025 best practices: uv/pip-tools recommended for locking)
+haystack-ai==2.0.1
+opensearch-py==2.6.0
+sentence-transformers==3.1.1
+pydantic==2.8.2
+pydantic-ai==0.0.10
+fastapi==0.115.0
+uvicorn==0.30.6
+httpx==0.27.2
+structlog==24.1.0
+gradio==4.44.0
+markdown-it-py==3.0.0
+mdurl==0.1.2
+python-dotenv==1.0.1
+# optional (CPU fallback for reranker)
+transformers==4.44.2
+accelerate==0.34.2
+
+# ===========================
+# app/__init__.py
+# ===========================
+from __future__ import annotations
+
+__all__ = [
+    "config", "logging_setup", "models",
+]
+
+# ===========================
+# app/logging_setup.py
+# ===========================
+from __future__ import annotations
+import logging
+import structlog
+
+_DEF_LEVEL = logging.INFO
+
+def setup_logging(level: int = _DEF_LEVEL) -> None:
+    """Structured logging; call early in main."""
+    logging.basicConfig(level=level, format="%(message)s")
+    structlog.configure(
+        processors=[
+            structlog.processors.TimeStamper(fmt="iso"),
+            structlog.processors.add_log_level,
+            structlog.processors.StackInfoRenderer(),
+            structlog.processors.format_exc_info,
+            structlog.processors.JSONRenderer()
+        ],
+        logger_factory=structlog.stdlib.LoggerFactory(),
+        wrapper_class=structlog.stdlib.BoundLogger,
+        cache_logger_on_first_use=True,
+    )
+
+# ===========================
+# app/config.py
+# ===========================
+from __future__ import annotations
+from pydantic import BaseModel, Field
+from pydantic_settings import BaseSettings
+from typing import Optional
+
+class OpenSearchSettings(BaseModel):
+    host: str = Field(default="localhost")
+    port: int = Field(default=9200)
+    scheme: str = Field(default="http")
+    index_name: str = Field(default="policies-v1")
+    embedding_dim: int = Field(default=1024)
+
+class ModelSettings(BaseModel):
+    embedding_model: str = Field(default="intfloat/multilingual-e5-large-instruct")
+    reranker_model: str = Field(default="BAAI/bge-reranker-v2-m3")
+    # LLM endpoint: use OpenAI-compatible endpoint or local server
+    llm_base_url: str = Field(default="http://localhost:8001/v1")
+    llm_api_key: Optional[str] = Field(default=None)
+    llm_model: str = Field(default="openai/gpt-oss-20b")
+
+class AppSettings(BaseSettings):
+    env: str = Field(default="dev")  # dev|prod|space
+    os: OpenSearchSettings = Field(default_factory=OpenSearchSettings)
+    models: ModelSettings = Field(default_factory=ModelSettings)
+    # retrieval knobs
+    bm25_k: int = Field(default=16)
+    dense_k: int = Field(default=16)
+    rerank_k: int = Field(default=5)
+    # chunking
+    prose_split_length: int = Field(default=12)  # ~350 tokens (sentence units)
+    prose_overlap: int = Field(default=2)
+
+    class Config:
+        env_nested_delimiter = "__"
+        env_prefix = "RAG_"  # e.g., RAG_MODELS__LLM_BASE_URL
+
+settings = AppSettings()  # read from env automatically
+
+# ===========================
+# app/models.py
+# ===========================
+from __future__ import annotations
+from typing import List, Dict, Literal
+from pydantic import BaseModel, Field
+
+class StudentQuery(BaseModel):
+    intent: str
+    questions: List[str]
+    language: Literal["de", "en", "fr", "it"] = "de"
+    entities: Dict[str, str] = Field(default_factory=dict)  # {"semester": "HS"}
+
+class Evidence(BaseModel):
+    passage: str
+    section_path: str
+    doc_title: str
+    score: float | None = None
+    doc_id: str | None = None
+
+class EmailDraft(BaseModel):
+    body: str
+    citations: List[Evidence] = Field(default_factory=list)
+    warnings: List[str] = Field(default_factory=list)
+
+# ===========================
+# app/utils/__init__.py
+# ===========================
+
+# ===========================
+# app/utils/markdown_loader.py
+# ===========================
+from __future__ import annotations
+from typing import Iterable, List
+from haystack import Document
+from markdown_it import MarkdownIt
+
+md = MarkdownIt()
+
+_DEF_LANG = "de"
+
+def _serialize_table(tokens: list) -> str:
+    # Very simple table serializer (improve as needed)
+    rows: List[List[str]] = []
+    curr: List[str] = []
+    for t in tokens:
+        if t.type.endswith("_open"):
+            curr = []
+        elif t.type.endswith("_close"):
+            if curr:
+                rows.append(curr)
+        elif t.type == "inline":
+            curr.append(t.content.strip())
+    lines = []
+    for r in rows:
+        if len(r) >= 2:
+            lines.append(f"{r[0]}: {' | '.join(r[1:])}")
+        elif r:
+            lines.append(r[0])
+    return "\n".join(lines)
+
+def load_markdown_to_documents(text: str, title: str, section_root: str | None = None, lang: str = _DEF_LANG) -> Iterable[Document]:
+    tokens = md.parse(text)
+    section = section_root or title
+    buff: List[str] = []
+    path_stack: List[str] = [section]
+
+    def flush_paragraph():
+        nonlocal buff
+        if buff:
+            yield Document(content="\n".join(buff), meta={"title": title, "section_path": ">".join(path_stack), "lang": lang, "block_type": "prose"})
+            buff = []
+
+    i = 0
+    while i < len(tokens):
+        t = tokens[i]
+        if t.type.endswith("heading_open"):
+            # flush current paragraph
+            yield from flush_paragraph()
+            # next inline has the text
+            h_text = tokens[i+1].content.strip()
+            # adjust stack
+            # naive: always attach under root
+            path_stack = [section, h_text]
+            i += 3
+            continue
+        if t.type == "paragraph_open":
+            # collect until paragraph_close
+            i += 1
+            while tokens[i].type != "paragraph_close":
+                if tokens[i].type == "inline":
+                    buff.append(tokens[i].content)
+                i += 1
+            # close handled by flush at next event
+        elif t.type == "table_open":
+            # parse whole table block
+            j = i + 1
+            table_tokens = []
+            depth = 1
+            while j < len(tokens) and depth > 0:
+                if tokens[j].type == "table_open":
+                    depth += 1
+                elif tokens[j].type == "table_close":
+                    depth -= 1
+                table_tokens.append(tokens[j])
+                j += 1
+            table_text = _serialize_table(table_tokens)
+            yield Document(content=table_text, meta={"title": title, "section_path": ">".join(path_stack), "lang": lang, "block_type": "table"})
+            i = j
+            continue
+        i += 1
+    # flush remaining
+    yield from flush_paragraph()
+
+# ===========================
+# app/retriever/__init__.py
+# ===========================
+
+# ===========================
+# app/retriever/indexer.py
+# ===========================
+from __future__ import annotations
+from typing import Iterable
+from haystack.document_stores import OpenSearchDocumentStore
+from haystack.components.preprocessors import DocumentSplitter
+from haystack.components.embedders import SentenceTransformersDocumentEmbedder
+from haystack.components.writers import DocumentWriter
+from haystack import Document
+from app.config import settings
+
+_splitter = DocumentSplitter(
+    split_by="sentence", split_length=settings.prose_split_length,
+    split_overlap=settings.prose_overlap, respect_sentence_boundary=True
+)
+
+_embedder = SentenceTransformersDocumentEmbedder(
+    model=settings.models.embedding_model,
+    normalize_embeddings=True,
+    prompt="passage: "
+)
+
+def build_docstore() -> OpenSearchDocumentStore:
+    return OpenSearchDocumentStore(
+        index=settings.os.index_name,
+        hosts=[{"host": settings.os.host, "port": settings.os.port, "scheme": settings.os.scheme}],
+        embedding_dim=settings.os.embedding_dim,
+    )
+
+def index_documents(docs: Iterable[Document]) -> int:
+    store = build_docstore()
+    writer = DocumentWriter(document_store=store)
+    # Split prose chunks only; keep tables as-is (block_type metadata guides behavior)
+    out_docs = []
+    for d in docs:
+        if d.meta.get("block_type") == "prose":
+            out_docs.extend(_splitter.run(documents=[d])["documents"])
+        else:
+            out_docs.append(d)
+    # Embed
+    embedded = _embedder.run(documents=out_docs)["documents"]
+    # Persist
+    writer.run(documents=embedded)
+    return len(embedded)
+
+# ===========================
+# app/retriever/pipeline.py
+# ===========================
+from __future__ import annotations
+from typing import Dict, List
+from haystack.document_stores import OpenSearchDocumentStore
+from haystack.components.retrievers import OpenSearchBM25Retriever, OpenSearchEmbeddingRetriever, RRF
+from haystack.components.rankers import TransformersCrossEncoderRanker
+from haystack import Pipeline, Document
+from app.config import settings
+
+_store: OpenSearchDocumentStore | None = None
+
+
+def _store_or_new() -> OpenSearchDocumentStore:
+    global _store
+    if _store is None:
+        _store = OpenSearchDocumentStore(
+            index=settings.os.index_name,
+            hosts=[{"host": settings.os.host, "port": settings.os.port, "scheme": settings.os.scheme}],
+            embedding_dim=settings.os.embedding_dim,
+        )
+    return _store
+
+_bm25 = OpenSearchBM25Retriever(document_store=_store_or_new(), top_k=settings.bm25_k)
+_dense = OpenSearchEmbeddingRetriever(document_store=_store_or_new(), top_k=settings.dense_k)
+_fuser = RRF()
+_reranker = TransformersCrossEncoderRanker(model=settings.models.reranker_model, top_k=settings.rerank_k)
+
+_pipe = Pipeline()
+_pipe.add_component("bm25", _bm25)
+_pipe.add_component("dense", _dense)
+_pipe.add_component("fuse", _fuser)
+_pipe.add_component("rerank", _reranker)
+_pipe.connect("bm25", "fuse")
+_pipe.connect("dense", "fuse")
+_pipe.connect("fuse", "rerank")
+
+
+def retrieve(query_text: str, filters: Dict | None = None) -> List[Document]:
+    q_dense = "query: " + query_text  # E5 query prefix
+    out = _pipe.run({
+        "bm25": {"query": query_text, "filters": filters},
+        "dense": {"query": q_dense, "filters": filters},
+    })
+    return out["rerank"]["documents"]
+
+# ===========================
+# app/agents/__init__.py
+# ===========================
+
+# ===========================
+# app/agents/llm_client.py
+# ===========================
+from __future__ import annotations
+from typing import Any, Dict
+import httpx
+from app.config import settings
+
+class LLMClient:
+    """Minimal OpenAI-compatible client with timeouts & retries."""
+    def __init__(self, base_url: str | None = None, api_key: str | None = None, model: str | None = None) -> None:
+        self.base_url = base_url or settings.models.llm_base_url
+        self.api_key = api_key or settings.models.llm_api_key or "sk-void"
+        self.model = model or settings.models.llm_model
+        self._client = httpx.Client(base_url=self.base_url, timeout=30.0)
+
+    def chat(self, messages: list[dict], response_format: Dict[str, Any] | None = None) -> dict:
+        payload: Dict[str, Any] = {"model": self.model, "messages": messages}
+        if response_format:
+            payload["response_format"] = response_format
+        r = self._client.post("/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json=payload)
+        r.raise_for_status()
+        return r.json()
+
+# ===========================
+# app/agents/intent_extractor.py
+# ===========================
+from __future__ import annotations
+from typing import Any
+from pydantic_ai import Agent
+from pydantic import BaseModel, Field
+from app.models import StudentQuery
+from app.agents.llm_client import LLMClient
+
+class _StudentQuerySchema(BaseModel):
+    intent: str
+    questions: list[str]
+    language: str = Field(pattern="^(de|en|fr|it)$")
+    entities: dict = Field(default_factory=dict)
+
+_client = LLMClient()
+
+intent_agent = Agent(
+    _StudentQuerySchema,
+    system_prompt=(
+        "You are a university admin triage assistant. Extract intent, a list of explicit questions,"
+        " language code (de/en/fr/it), and simple entities (e.g., semester=HS/FS, program)."
+        " Return only fields in the schema."
+    ),
+    model_client="openai",  # pydantic-ai maps to OpenAI-compatible; configured via env
+)
+
+
+def extract(email_text: str) -> StudentQuery:
+    res = intent_agent.run_sync(email_text)  # returns validated pydantic
+    return StudentQuery(**res.model_dump())
+
+# ===========================
+# app/agents/composer.py
+# ===========================
+from __future__ import annotations
+from typing import List
+from pydantic_ai import Agent
+from app.models import StudentQuery, EmailDraft, Evidence
+
+composer_agent = Agent(
+    EmailDraft,
+    system_prompt=(
+        "You draft clear, courteous, and policy-grounded emails for university admin staff.\n"
+        "Use the provided evidence only; do not invent rules. Add short citations (title + section_path).\n"
+        "Return a single text body suitable to copy-paste, plus citations and warnings if evidence is weak."
+    ),
+    model_client="openai",
+)
+
+
+def compose(query: StudentQuery, evidences: List[Evidence]) -> EmailDraft:
+    # Convert evidences to a readable context block
+    ctx = "\n\n".join(
+        f"[{i+1}] {e.doc_title} > {e.section_path}\n{e.passage}" for i, e in enumerate(evidences)
+    )
+    user = (
+        f"LANG={query.language}\n"
+        f"INTENT={query.intent}\n"
+        f"QUESTIONS={query.questions}\n"
+        f"ENTITIES={query.entities}\n\n"
+        f"EVIDENCE:\n{ctx}"
+    )
+    res = composer_agent.run_sync(user)
+    return EmailDraft(**res.model_dump())
+
+# ===========================
+# app/agents/fact_checker.py
+# ===========================
+from __future__ import annotations
+from typing import List
+from pydantic_ai import Agent
+from app.models import EmailDraft, Evidence
+
+checker_agent = Agent(
+    EmailDraft,
+    system_prompt=(
+        "You verify the draft email is fully supported by the evidence.\n"
+        "Add warnings for any claims lacking backing text; suggest placeholders instead of guessing."
+    ),
+    model_client="openai",
+)
+
+
+def fact_check(draft: EmailDraft, evidences: List[Evidence]) -> EmailDraft:
+    ctx = "\n\n".join(
+        f"[{i+1}] {e.doc_title} > {e.section_path}\n{e.passage}" for i, e in enumerate(evidences)
+    )
+    user = f"DRAFT:\n{draft.body}\n\nEVIDENCE:\n{ctx}"
+    res = checker_agent.run_sync(user)
+    return EmailDraft(**res.model_dump())
+
+# ===========================
+# app/gradio_app.py
+# ===========================
+from __future__ import annotations
+import gradio as gr
+from typing import List
+from app.agents import intent_extractor, composer, fact_checker
+from app.retriever.pipeline import retrieve
+from app.models import StudentQuery, Evidence, EmailDraft
+
+_DEF_PLACEHOLDER = "Fügen Sie hier die Studenten-E-Mail ein / Paste the student email here..."
+
+
+def _to_evidence(documents) -> List[Evidence]:
+    evs: List[Evidence] = []
+    for d in documents:
+        evs.append(Evidence(
+            passage=d.content,
+            section_path=d.meta.get("section_path", ""),
+            doc_title=d.meta.get("title", ""),
+            score=d.score,
+            doc_id=d.id,
+        ))
+    return evs
+
+
+def answer(email_text: str) -> tuple[str, str]:
+    if not email_text.strip():
+        return "", ""
+    q: StudentQuery = intent_extractor.extract(email_text)
+    docs = []
+    for question in q.questions or [email_text]:
+        docs.extend(retrieve(question, filters={"lang": [q.language]}))
+    # deduplicate while keeping top scores
+    seen = {}
+    for d in docs:
+        if d.id not in seen or d.score > seen[d.id].score:
+            seen[d.id] = d
+    top_docs = sorted(seen.values(), key=lambda x: x.score or 0.0, reverse=True)[:8]
+    evs = _to_evidence(top_docs)
+    draft: EmailDraft = composer.compose(q, evs)
+    checked: EmailDraft = fact_checker.fact_check(draft, evs)
+
+    # Advanced panel content
+    adv = []
+    for i, e in enumerate(evs, start=1):
+        adv.append(f"### {i}. {e.doc_title} › {e.section_path}\nScore: {e.score:.3f}\n\n{e.passage}")
+    advanced_md = "\n\n".join(adv)
+    return checked.body, advanced_md
+
+
+def build_interface() -> gr.Blocks:
+    with gr.Blocks(theme=gr.themes.Soft()) as demo:
+        gr.Markdown("# 📬 Staff Assist – RAG Email Drafts (Haystack 2 + PydanticAI)")
+        with gr.Row():
+            email_in = gr.Textbox(lines=12, placeholder=_DEF_PLACEHOLDER, label="Student Email")
+        with gr.Row():
+            draft_out = gr.Textbox(lines=14, label="Draft Reply (Editable)")
+        with gr.Accordion("Advanced: Retrieved Evidence (chunks & sections)", open=False):
+            advanced = gr.Markdown()
+        submit = gr.Button("Generate Draft", variant="primary")
+        submit.click(answer, inputs=[email_in], outputs=[draft_out, advanced])
+    return demo
+
+# ===========================
+# app/main.py
+# ===========================
+from __future__ import annotations
+from app.logging_setup import setup_logging
+from app.gradio_app import build_interface
+
+if __name__ == "__main__":
+    setup_logging()
+    ui = build_interface()
+    ui.launch(server_name="0.0.0.0", server_port=7860)
+
+# ===========================
+# README.md (excerpt)
+# ===========================
+# RAG Email Assistant – Haystack 2 + PydanticAI + Gradio
+
+## Quick start (dev)
+1. Run OpenSearch locally (or point to your cluster). Create index with k-NN enabled (dimension 1024).  
+2. Set env vars (see `RAG_*` in `app/config.py`).  
+3. Index your Markdown: use `load_markdown_to_documents` + `index_documents` from a small script.  
+4. `python -m app.main`
+
+## Hugging Face Spaces notes
+- Spaces (CPU) can host the **Gradio UI**; but OpenSearch must be reachable via network.  
+- If you need a self-contained demo, swap to a local FAISS `InMemoryDocumentStore` in `pipeline.py` (feature flag) and reduce models to CPU-friendly ones.
+
+## Best practices baked in (2025)
+- Strict typing and Pydantic v2 models for agent IO.  
+- OpenAI-compatible LLM client abstraction (swap endpoints/models without code changes).  
+- Heading-aware chunking, table preservation, hybrid retrieval + cross-encoder rerank.  
+- Structured logging (JSON) via `structlog`.  
+- Env-driven settings with nested prefixes (Twelve-Factor).  
+- Safe defaults (normalized E5 embeddings, E5 query prefix, dedup of docs).  
+- Clear separation: ingestion/indexing vs. serving.  
+
+## Where to extend
+- Add caching for embeddings & retrieval; add RAG evaluation notebook (Recall@k, groundedness).  
+- Add multilingual tone/style templates in `composer.py` based on `query.language`.  
+- Add policy/version metadata and link anchors per chunk for clickable citations in UI.  
+- Add guardrails (regex) to block sharing internal links when emailing students.  
+- Add DSPy for prompt/pipeline optimization once you have labeled email pairs.