Upload 56 files

.env

@@ -0,0 +1,29 @@
+# --- LLM ---
+GROQ_API_KEY=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
+
+# --- Internal service auth (shared secret with Django) ---
+CHAT_SERVICE_INTERNAL_TOKEN=XoR_B5BiwcEDimWeL-a_kIqe-WcmC79wMg4FnRlMQXI
+
+# --- Postgres (read-only role) ---
+POSTGRES_READONLY_URL=postgresql://chat_ro_user:strong_password@hayabusa.proxy.rlwy.net:42420/railway?sslmode=require
+
+# --- Neo4j ---
+NEO4J_URI=neo4j+s://78f4ecfc.databases.neo4j.io
+NEO4J_USER=78f4ecfc
+NEO4J_PASSWORD=0SS0p8cU5f_cwxbcF9fErEkyNp6xEuhZC_wqcEjG06A
+
+# --- Redis (reuse existing instance, separate DB index) ---
+REDIS_URL=redis://default:UzTsAxeHPgOQUKPjDSvlxkdMcRVoJalU@hayabusa.proxy.rlwy.net:56307
+
+# --- Chat session behavior ---
+CHAT_SESSION_TTL_SECONDS=86400
+CHAT_HISTORY_MAX_TURNS=10
+
+# --- App ---
+ENV=development
+LOG_LEVEL=info
+PORT=8002
+
+# --- NLP ---
+NLP_SERVICE_URL=https://AbdelazizAushar-arabic-nlp-service.hf.space
+NLP_SERVICE_TOKEN="aNs-2026-nlp-key-final"

.env.example

@@ -0,0 +1,25 @@
+# --- LLM ---
+GROQ_API_KEY=
+
+# --- Internal service auth (shared secret with Django) ---
+CHAT_SERVICE_INTERNAL_TOKEN=
+
+# --- Postgres (read-only role) ---
+POSTGRES_READONLY_URL=postgresql://chat_ro_user:CHANGEME@localhost:5432/newsdb
+
+# --- Neo4j (read-only role) ---
+NEO4J_READONLY_URI=neo4j://localhost:7687
+NEO4J_READONLY_USER=chat_ro_user
+NEO4J_READONLY_PASSWORD=CHANGEME
+
+# --- Redis (reuse existing instance, separate DB index) ---
+REDIS_URL=redis://localhost:6379/1
+
+# --- Chat session behavior ---
+CHAT_SESSION_TTL_SECONDS=86400
+CHAT_HISTORY_MAX_TURNS=10
+
+# --- App ---
+ENV=development
+LOG_LEVEL=info
+PORT=8002

.gitignore

@@ -0,0 +1,44 @@
+# # --- Environment / secrets ---
+# .env
+# .env.*
+# !.env.example
+
+# --- Python ---
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+venv/
+env/
+ENV/
+.venv/
+pip-log.txt
+pip-delete-this-directory.txt
+
+# --- Testing / coverage ---
+.pytest_cache/
+.coverage
+htmlcov/
+.tox/
+
+# --- IDE ---
+.vscode/
+.idea/
+*.swp
+
+# --- OS ---
+.DS_Store
+Thumbs.db
+
+# --- Logs ---
+*.log
+logs/
+
+# --- Docker ---
+*.pid
+
+# --- Build artifacts ---
+build/
+dist/
+*.egg-info/

Dockerfile

@@ -0,0 +1,28 @@
+FROM python:3.12-slim
+
+# Prevent Python from writing .pyc files and enable unbuffered logs
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+# System deps needed for psycopg / neo4j drivers to build
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    libpq-dev \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install Python deps first (layer caching)
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+EXPOSE 8000
+
+HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
+    CMD curl -f http://localhost:8000/health || exit 1
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

app/agent/graph.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+from app.memory.history_trimmer import maybe_trim_history
+
+import json
+from typing import Any, Optional
+
+from langchain_groq import ChatGroq
+from langgraph.prebuilt import create_react_agent
+from langgraph.checkpoint.redis.aio import AsyncRedisSaver
+
+from app.agent.prompts import SYSTEM_PROMPT
+from app.agent.state import AgentState
+from app.config import get_settings
+from app.tools.graph_tool import graph_query_tool
+from app.tools.sql_tool import sql_query_tool
+from app.tools.article_tool import get_article_detail
+from app.tools.hybrid_tool import hybrid_search_tool
+
+DEFAULT_MODEL = "llama-3.3-70b-versatile"
+
+_agent = None
+
+
+def build_agent(checkpointer: AsyncRedisSaver):
+    """Called once at startup after the checkpointer is ready."""
+    global _agent
+    settings = get_settings()
+    llm = ChatGroq(
+        api_key=settings.groq_api_key,
+        model=DEFAULT_MODEL,
+        temperature=0,
+    )
+    _agent = create_react_agent(
+        model=llm,
+        tools=[sql_query_tool, graph_query_tool, hybrid_search_tool],
+        state_schema=AgentState,
+        prompt=SYSTEM_PROMPT,
+        checkpointer=checkpointer,
+    )
+    return _agent
+
+
+def get_agent():
+    if _agent is None:
+        raise RuntimeError(
+            "Agent not initialized — call build_agent() at startup first")
+    return _agent
+
+
+def _extract_sources(messages: list) -> list[dict]:
+    sources: list[dict] = []
+    for m in messages:
+        if getattr(m, "type", None) != "tool":
+            continue
+        try:
+            payload = json.loads(m.content) if isinstance(
+                m.content, str) else m.content
+        except (json.JSONDecodeError, TypeError):
+            continue
+        refs = payload.get("source_refs") if isinstance(
+            payload, dict) else None
+        if refs:
+            sources.extend(refs)
+
+    seen: set[tuple] = set()
+    deduped: list[dict] = []
+    for s in sources:
+        key = (s.get("type"), s.get("id"))
+        if key not in seen:
+            seen.add(key)
+            deduped.append(s)
+    return deduped
+
+
+async def _enrich_sources(raw_refs: list[dict]) -> list[dict]:
+    enriched: list[dict] = []
+    for ref in raw_refs:
+        if ref.get("type") == "article" and ref.get("id") is not None:
+            detail = await get_article_detail(int(ref["id"]))
+            rows = detail.get("rows") or []
+            if rows:
+                article = rows[0]
+                enriched.append({
+                    "type": "article",
+                    "id": article["id"],
+                    "title": article.get("title"),
+                    "url": article.get("url"),
+                })
+                continue
+        enriched.append(ref)
+    return enriched
+
+
+async def run_agent(message: str, session_id: str, user_id: Optional[str]) -> dict[str, Any]:
+    agent = get_agent()
+    config = {"configurable": {"thread_id": session_id}}
+
+    result = await agent.ainvoke(
+        {
+            "messages": [{"role": "user", "content": message}],
+            "sources": [],
+            "session_id": session_id,
+            "user_id": user_id,
+        },
+        config=config,
+    )
+
+    final_message = result["messages"][-1]
+    answer = final_message.content if isinstance(
+        final_message.content, str) else str(final_message.content)
+
+    raw_sources = _extract_sources(result["messages"])
+    sources = await _enrich_sources(raw_sources)
+
+    await maybe_trim_history(agent, config)
+
+    return {"answer": answer, "sources": sources}

app/agent/llm.py

@@ -0,0 +1,30 @@
+from groq import Groq
+
+from app.config import get_settings
+
+_client: Groq | None = None
+
+DEFAULT_MODEL = "llama-3.3-70b-versatile"
+
+
+def _get_client() -> Groq:
+    global _client
+    if _client is None:
+        settings = get_settings()
+        _client = Groq(api_key=settings.groq_api_key)
+    return _client
+
+
+def ask(prompt: str, model: str = DEFAULT_MODEL) -> str:
+    """
+    Minimal single-turn call to Groq. No tools, no memory —
+    just proves the API key + connection work end-to-end.
+    """
+    client = _get_client()
+
+    response = client.chat.completions.create(
+        model=model,
+        messages=[{"role": "user", "content": prompt}],
+    )
+
+    return response.choices[0].message.content or ""

app/agent/prompts.py

@@ -0,0 +1,44 @@
+SYSTEM_PROMPT = """\
+You are the assistant for a news analysis platform. You help users understand \
+statistics and findings about analyzed news articles: fake-news verdicts, \
+topics, sentiment, propaganda techniques, hate-speech categories, dialect, \
+and entity relationships extracted from those articles.
+
+## Tool use
+- You have access to `sql_query_tool`, which runs pre-defined, parameterized \
+queries against a read-only database of analyzed articles (stats, counts, \
+breakdowns, lookups).
+- You have access to `graph_query_tool`, which runs pre-defined Cypher queries \
+against a read-only entity relationship graph (connections between entities, \
+which articles mention an entity, shared entities between articles, most \
+mentioned entities, fact-check verdicts, per-claim breakdowns, and full \
+per-article analysis profiles).
+- Use whichever tool matches the question — `sql_query_tool` for statistics/ \
+counts/breakdowns, `graph_query_tool` for relationships/connections/fact-check \
+detail. Do not answer such questions from memory.
+- Never invent numbers, article titles, verdicts, or entity names. If a tool \
+does not return the information needed to answer, say so plainly instead of \
+guessing.
+- If the user asks about "last month", "this week", "yesterday", etc., convert \
+that into concrete ISO `date_from` / `date_to` values yourself before calling \
+`sql_query_tool`, using today's date as reference.
+- If a question is ambiguous about which metric or query_type applies, pick \
+the closest matching one and state the assumption you made in your answer.
+- If a question asks about something no available tool covers (e.g. general \
+knowledge, opinions, anything outside this platform's analyzed article data), \
+say clearly that you cannot answer it rather than fabricating a response.
+
+## Citations
+- Every factual claim in your answer that is based on tool results must be \
+traceable to a specific article. Mention every `article_id` you actually used \
+to construct your answer — do not omit any, and do not include IDs you did not use.
+- Do not fabricate article IDs, titles, or entity names that were not returned \
+by a tool call. If a tool returned no matching data, say so — do not guess.
+
+## Language
+- Always respond in the same language the user wrote their message in.
+
+## Style
+- Be direct and concise. Give the number or fact requested first, then brief \
+supporting context if useful. Do not pad answers with unnecessary caveats.
+"""

app/agent/state.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from typing import Annotated, TypedDict
+
+from langchain_core.messages import BaseMessage
+from langgraph.graph.message import add_messages
+from langgraph.managed.is_last_step import RemainingSteps
+from typing_extensions import NotRequired
+
+
+class AgentState(TypedDict):
+    messages: Annotated[list[BaseMessage], add_messages]
+    sources: list[dict]
+    session_id: str
+    user_id: str | None
+    remaining_steps: NotRequired[RemainingSteps]

app/config.py

@@ -0,0 +1,47 @@
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    # --- App ---
+    env: str = "development"
+    log_level: str = "info"
+    port: int = 8002
+
+    # --- LLM ---
+    groq_api_key: str
+
+    # --- Internal service auth (shared secret with Django) ---
+    chat_service_internal_token: str = ""
+
+    # --- Postgres (read-only role) ---
+    postgres_readonly_url: str = ""
+
+    # --- Neo4j (Aura Free — no RBAC, enforcement is app-level only) ---
+    neo4j_uri: str = ""
+    neo4j_user: str = ""
+    neo4j_password: str = ""
+
+    # --- Redis ---
+    redis_url: str = ""
+
+    # --- Chat session behavior ---
+    chat_session_ttl_seconds: int = 86400
+    chat_history_max_turns: int = 10
+
+    # --- NLP EMBED ---
+    nlp_service_url: str = ""
+    nlp_service_token: str = ""
+
+
+@lru_cache
+def get_settings() -> Settings:
+    """Cached settings instance — reads .env once per process."""
+    return Settings()

app/db/neo4j.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from neo4j import AsyncGraphDatabase, AsyncDriver, AsyncTransaction
+
+from app.config import get_settings
+
+_driver: AsyncDriver | None = None
+
+
+def get_driver() -> AsyncDriver:
+    global _driver
+    if _driver is None:
+        settings = get_settings()
+        _driver = AsyncGraphDatabase.driver(
+            settings.neo4j_uri,
+            auth=(settings.neo4j_user, settings.neo4j_password),
+        )
+    return _driver
+
+
+async def verify_connectivity() -> None:
+    """Call on app startup to fail fast if Aura is unreachable."""
+    await get_driver().verify_connectivity()
+
+
+async def close_driver() -> None:
+    global _driver
+    if _driver is not None:
+        await _driver.close()
+    _driver = None
+
+
+async def run_read_query(query: str, **params) -> list[dict]:
+    """
+    Runs a Cypher query inside an explicit READ transaction.
+
+    App-level read-only enforcement (Aura Free has no RBAC/custom roles,
+    unlike the Postgres chat_ro_user). NEVER pass write Cypher
+    (CREATE/MERGE/SET/DELETE/REMOVE) into this function — query text must
+    only come from pre-written templates in tool code, never from the LLM.
+    """
+    driver = get_driver()
+
+    async def _tx_func(tx: AsyncTransaction):
+        result = await tx.run(query, **params)
+        return await result.data()
+
+    async with driver.session() as session:
+        return await session.execute_read(_tx_func)

app/db/postgres.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+
+from psycopg_pool import AsyncConnectionPool
+
+from app.config import get_settings
+
+_pool: AsyncConnectionPool | None = None
+
+
+def get_pool() -> AsyncConnectionPool:
+    global _pool
+    if _pool is None:
+        settings = get_settings()
+        _pool = AsyncConnectionPool(
+            conninfo=settings.postgres_readonly_url,
+            min_size=1,
+            max_size=10,
+            open=False,
+        )
+    return _pool
+
+
+async def open_pool() -> None:
+    """Call on app startup."""
+    pool = get_pool()
+    if pool.closed:
+        await pool.open(wait=True)
+
+
+async def close_pool() -> None:
+    """Call on app shutdown."""
+    global _pool
+    if _pool is not None and not _pool.closed:
+        await _pool.close()
+    _pool = None
+
+
+@asynccontextmanager
+async def get_readonly_connection():
+    """
+    Yields a connection with an extra read-only guard set at the
+    session level, on top of the chat_ro_user DB role.
+    """
+    pool = get_pool()
+    async with pool.connection() as conn:
+        await conn.execute("SET default_transaction_read_only = on")
+        yield conn

app/deps.py

@@ -0,0 +1,11 @@
+from typing import Annotated
+
+from fastapi import Depends
+
+from app.agent.graph import get_agent
+from app.config import Settings, get_settings
+
+SettingsDep = Annotated[Settings, Depends(get_settings)]
+
+
+AgentDep = Annotated[object, Depends(get_agent)]

app/main.py

@@ -0,0 +1,51 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from app.agent.graph import run_agent
+from app.config import get_settings
+from app.db.postgres import close_pool, open_pool
+from app.db.neo4j import verify_connectivity, close_driver
+from app.schemas.request import ChatRequest
+from app.schemas.response import ChatResponse
+from app.memory.redis_checkpointer import get_checkpointer, close_checkpointer
+from app.agent.graph import build_agent
+
+settings = get_settings()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    await open_pool()
+    await verify_connectivity()
+    checkpointer = await get_checkpointer()
+    build_agent(checkpointer)
+    yield
+    await close_pool()
+    await close_driver()
+    await close_checkpointer()
+
+app = FastAPI(title="chat-service", version="0.1.0", lifespan=lifespan)
+
+
+@app.get("/health")
+def health() -> dict:
+    return {"status": "ok"}
+
+
+@app.post("/chat", response_model=ChatResponse)
+async def chat(payload: ChatRequest) -> ChatResponse:
+    """
+    Milestone 2: routes through the LangGraph ReAct agent, which can call
+    sql_query_tool to answer questions grounded in the analyzed-articles DB.
+    """
+    result = await run_agent(
+        message=payload.message,
+        session_id=payload.session_id,
+        user_id=payload.user_id,
+    )
+    return ChatResponse(
+        session_id=payload.session_id,
+        answer=result["answer"],
+        sources=result["sources"],
+    )

app/memory/history_trimmer.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from langchain_core.messages import BaseMessage, RemoveMessage, SystemMessage
+from langchain_groq import ChatGroq
+
+from app.config import get_settings
+
+_summarizer_llm = None
+
+
+def _get_summarizer() -> ChatGroq:
+    """Separate, cheaper model just for summarization — keeps main agent's
+    token budget free and avoids competing for the same Groq quota."""
+    global _summarizer_llm
+    if _summarizer_llm is None:
+        settings = get_settings()
+        _summarizer_llm = ChatGroq(
+            api_key=settings.groq_api_key,
+            model="llama-3.1-8b-instant",
+            temperature=0,
+        )
+    return _summarizer_llm
+
+
+def _is_turn_message(m: BaseMessage) -> bool:
+    """Only human/ai messages count as a 'turn' — tool/system noise is excluded."""
+    return getattr(m, "type", None) in ("human", "ai")
+
+
+async def maybe_trim_history(agent, config: dict) -> None:
+    """
+    Keeps the last CHAT_HISTORY_MAX_TURNS human/ai messages verbatim in
+    the checkpointed state. Anything older is collapsed into a single
+    SystemMessage summary. Call this after every agent.ainvoke().
+    """
+    settings = get_settings()
+    max_turns = settings.chat_history_max_turns
+
+    state = await agent.aget_state(config)
+    messages: list[BaseMessage] = state.values.get("messages", [])
+
+    turn_messages = [m for m in messages if _is_turn_message(m)]
+    if len(turn_messages) <= max_turns:
+        return  # nothing to trim yet
+
+    cutoff_id = turn_messages[-max_turns].id
+    cutoff_index = next(i for i, m in enumerate(messages) if m.id == cutoff_id)
+    old_messages = [m for m in messages[:cutoff_index] if m.id is not None]
+
+    if not old_messages:
+        return
+
+    transcript = "\n".join(
+        f"{getattr(m, 'type', '?')}: {m.content}"
+        for m in old_messages
+        if isinstance(getattr(m, "content", None), str)
+    )
+    if not transcript.strip():
+        return
+
+    summarizer = _get_summarizer()
+    summary_resp = await summarizer.ainvoke([
+        SystemMessage(content=(
+            "Summarize this conversation history in 2-4 concise sentences. "
+            "Preserve key facts, entities, dates, and figures the user asked "
+            "about — this summary replaces the original messages as context "
+            "for future turns."
+        )),
+        SystemMessage(content=transcript),
+    ])
+
+    summary_message = SystemMessage(
+        content=f"[Earlier conversation summary]: {summary_resp.content}",
+        additional_kwargs={"is_history_summary": True},
+    )
+
+    removals = [RemoveMessage(id=m.id) for m in old_messages]
+    await agent.aupdate_state(config, {"messages": removals + [summary_message]})

app/memory/redis_checkpointer.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from langgraph.checkpoint.redis.aio import AsyncRedisSaver
+
+from app.config import get_settings
+
+_checkpointer_cm = None
+_checkpointer: AsyncRedisSaver | None = None
+
+
+async def get_checkpointer() -> AsyncRedisSaver:
+    """
+    Call once at app startup. Returns a live AsyncRedisSaver.
+    TTL is set from CHAT_SESSION_TTL_SECONDS (converted to minutes,
+    library expects minutes). refresh_on_read keeps active sessions alive.
+    """
+    global _checkpointer_cm, _checkpointer
+    if _checkpointer is None:
+        settings = get_settings()
+        ttl_minutes = max(1, settings.chat_session_ttl_seconds // 60)
+
+        _checkpointer_cm = AsyncRedisSaver.from_conn_string(
+            settings.redis_url,
+            ttl={"default_ttl": ttl_minutes, "refresh_on_read": True},
+            checkpoint_prefix="chat:checkpoint",
+            checkpoint_write_prefix="chat:checkpoint_write",
+        )
+        _checkpointer = await _checkpointer_cm.__aenter__()
+        await _checkpointer.asetup()  # creates Redis search indices, idempotent
+    return _checkpointer
+
+
+async def close_checkpointer() -> None:
+    global _checkpointer_cm, _checkpointer
+    if _checkpointer_cm is not None:
+        await _checkpointer_cm.__aexit__(None, None, None)
+    _checkpointer = None
+    _checkpointer_cm = None

app/schemas/request.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel, Field
+class ChatRequest(BaseModel):
+    session_id: str
+    message: str = Field(..., min_length=1)
+    user_id: str | None = None

app/schemas/response.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel
+
+
+class SourceRef(BaseModel):
+    type: str
+    id: int
+    title: str | None = None
+    url: str | None = None
+
+
+class ChatResponse(BaseModel):
+    session_id: str
+    answer: str
+    sources: list[SourceRef] = []

app/services/nlp_client.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+import httpx
+import numpy as np
+
+from app.config import get_settings
+
+logger = logging.getLogger(__name__)
+
+FAISS_DIM = 384  # match Django's MiniLM output dim (NewsArticle.search_vector)
+
+_NLP_MAX_RETRIES = 3
+_NLP_TIMEOUT_S = 60.0  # generous — accounts for HF Space cold start
+
+
+def _nlp_service_url() -> str:
+    settings = get_settings()
+    url = settings.nlp_service_url.rstrip("/")
+    if not url:
+        raise RuntimeError("NLP_SERVICE_URL not configured in settings")
+    return url
+
+
+def _nlp_headers() -> dict[str, str]:
+    settings = get_settings()
+    if not settings.nlp_service_token:
+        raise RuntimeError("NLP_SERVICE_TOKEN not configured in settings")
+    return {
+        "Authorization": f"Bearer {settings.nlp_service_token}",
+        "Content-Type": "application/json",
+    }
+
+
+async def _nlp_post(path: str, payload: dict) -> Any:
+    """POST to the shared NLP microservice with retry on timeout/5xx (cold-start safe)."""
+    url = f"{_nlp_service_url()}{path}"
+    headers = _nlp_headers()
+    last_exc: Exception | None = None
+
+    async with httpx.AsyncClient(timeout=_NLP_TIMEOUT_S) as client:
+        for attempt in range(1, _NLP_MAX_RETRIES + 1):
+            try:
+                resp = await client.post(url, json=payload, headers=headers)
+                resp.raise_for_status()
+                return resp.json()
+            except (httpx.TimeoutException, httpx.ConnectError) as exc:
+                last_exc = exc
+                logger.warning(
+                    "NLP service attempt %d/%d timed out: %s",
+                    attempt, _NLP_MAX_RETRIES, exc,
+                )
+                if attempt < _NLP_MAX_RETRIES:
+                    await asyncio.sleep(2 ** attempt)  # 2s, 4s
+            except httpx.HTTPStatusError as exc:
+                if exc.response.status_code < 500:
+                    raise  # 4xx — don't retry
+                last_exc = exc
+                logger.warning(
+                    "NLP service attempt %d/%d HTTP error: %s",
+                    attempt, _NLP_MAX_RETRIES, exc,
+                )
+                if attempt < _NLP_MAX_RETRIES:
+                    await asyncio.sleep(2 ** attempt)
+
+    raise RuntimeError(
+        f"NLP microservice unavailable after {_NLP_MAX_RETRIES} attempts"
+    ) from last_exc
+
+
+async def embed(texts: list[str]) -> np.ndarray:
+    """
+    Embed texts via the shared remote MiniLM endpoint (same HF Space Django uses).
+    Returns float32 array of shape (len(texts), FAISS_DIM), L2-normalised.
+    """
+    result: list[list[float]] = await _nlp_post("/embed", {"texts": texts})
+    return np.array(result, dtype="float32")

app/tools/article_tool.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from app.db.postgres import get_readonly_connection
+
+
+async def get_article_detail(article_id: int) -> dict:
+    """
+    Resolves a single article's full title/snippet/metadata for citation
+    display. Used by the agent post-processing step to enrich raw
+    article_id references into full source objects.
+
+    Returns: {"rows": [...], "source_refs": [{"type": "article", "id": ...}]}
+    (kept in the same shape as sql_tool/graph_tool for consistency, even
+    though this always returns 0 or 1 row.)
+    """
+    query = """
+        SELECT id, title, url, published_at, content
+        FROM news_articles
+        WHERE id = %s
+    """
+    async with get_readonly_connection() as conn:
+        cur = await conn.execute(query, (article_id,))
+        row = await cur.fetchone()
+
+    if row is None:
+        return {"rows": [], "source_refs": []}
+
+    columns = [desc[0] for desc in cur.description]
+    article = dict(zip(columns, row))
+
+    # Truncate content into a short snippet — don't return full article text
+    # for a citation lookup, keep responses lean
+    snippet = (article.get("content") or "")[:280]
+    if len(article.get("content") or "") > 280:
+        snippet += "..."
+
+    return {
+        "rows": [{
+            "id": article["id"],
+            "title": article["title"],
+            "url": article["url"],
+            "published_at": (
+                article["published_at"].isoformat()
+                if article.get("published_at") else None
+            ),
+            "snippet": snippet,
+        }],
+        "source_refs": [{"type": "article", "id": article["id"]}],
+    }

app/tools/graph_tool.py

@@ -0,0 +1,177 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Literal, Optional
+
+from langchain_core.tools import tool
+
+from app.db.neo4j import run_read_query
+
+QueryType = Literal[
+    "entity_connections",
+    "entity_mentions",
+    "shared_entities_between_articles",
+    "most_connected_entities",
+    "article_verdict",
+    "claims_for_article",
+    "analysis_for_article",
+]
+
+
+async def _entity_connections(entity_name, limit):
+    query = """
+        MATCH (e:Entity {name: $entity_name})-[r]-(other)
+        RETURN other, type(r) AS relationship, labels(other) AS other_labels
+        LIMIT $limit
+    """
+    rows = await run_read_query(query, entity_name=entity_name, limit=limit)
+
+    refs = []
+    for row in rows:
+        other = row.get("other", {}) or {}
+        if "entity_id" in other:
+            refs.append({"type": "entity", "id": other["entity_id"]})
+        elif other.get("article_id") is not None:
+            refs.append({"type": "article", "id": other["article_id"]})
+    return rows, refs
+
+
+async def _entity_mentions(entity_name, limit):
+    query = """
+        MATCH (a:Article)-[r:MENTIONS]->(e:Entity {name: $entity_name})
+        RETURN a.article_id AS article_id, r.confidence AS confidence
+        ORDER BY r.confidence DESC
+        LIMIT $limit
+    """
+    rows = await run_read_query(query, entity_name=entity_name, limit=limit)
+    refs = [{"type": "article", "id": r["article_id"]} for r in rows if r.get("article_id") is not None]
+    return rows, refs
+
+
+async def _shared_entities_between_articles(article_id, article_id_2, limit):
+    query = """
+        MATCH (a1:Article {article_id: $article_id})-[:MENTIONS]->(e:Entity)
+              <-[:MENTIONS]-(a2:Article {article_id: $article_id_2})
+        RETURN e.name AS name, e.type AS type, e.entity_id AS entity_id
+        LIMIT $limit
+    """
+    rows = await run_read_query(query, article_id=article_id, article_id_2=article_id_2, limit=limit)
+    refs = [{"type": "article", "id": article_id}, {"type": "article", "id": article_id_2}]
+    refs += [{"type": "entity", "id": r["entity_id"]} for r in rows if r.get("entity_id")]
+    return rows, refs
+
+
+async def _most_connected_entities(limit):
+    query = """
+        MATCH (e:Entity)<-[:MENTIONS]-(a:Article)
+        RETURN e.name AS name, e.type AS type, e.entity_id AS entity_id,
+               count(a) AS mention_count
+        ORDER BY mention_count DESC
+        LIMIT $limit
+    """
+    rows = await run_read_query(query, limit=limit)
+    refs = [{"type": "entity", "id": r["entity_id"]} for r in rows if r.get("entity_id")]
+    return rows, refs
+
+
+async def _article_verdict(article_id):
+    query = """
+        MATCH (a:Article {article_id: $article_id})-[:HAS_FINAL_VERDICT]->(v:Verdict)
+        OPTIONAL MATCH (v)-[:BASED_ON]->(e:Evidence)
+        RETURN v.verdict AS verdict, v.confidence AS confidence,
+               v.explanation AS explanation, collect(e) AS evidence
+    """
+    rows = await run_read_query(query, article_id=article_id)
+    return rows, [{"type": "article", "id": article_id}]
+
+
+async def _claims_for_article(article_id):
+    query = """
+        MATCH (a:Article {article_id: $article_id})-[:PROPOSES]->(c:Claim)
+              -[:EVALUATED_BY]->(v:Verdict)
+        RETURN c.text AS claim_text, c.claim_idx AS claim_idx,
+               v.verdict AS verdict, v.confidence AS confidence,
+               v.explanation AS explanation
+        ORDER BY c.claim_idx
+    """
+    rows = await run_read_query(query, article_id=article_id)
+    return rows, [{"type": "article", "id": article_id}]
+
+
+async def _analysis_for_article(article_id):
+    query = """
+        MATCH (a:Article {article_id: $article_id})-[r]->(an:Analysis)
+        RETURN type(r) AS relationship, an.analysis_type AS analysis_type,
+               an.value AS value, an.score AS score
+    """
+    rows = await run_read_query(query, article_id=article_id)
+    return rows, [{"type": "article", "id": article_id}]
+
+
+@tool
+async def graph_query_tool(
+    query_type: QueryType,
+    entity_name: Optional[str] = None,
+    article_id: Optional[int] = None,
+    article_id_2: Optional[int] = None,
+    limit: int = 20,
+) -> str:
+    """
+    Run a pre-defined, read-only Cypher query over the Neo4j entity/article
+    relationship graph.
+
+    query_type: which relationship query to run —
+        - entity_connections: everything connected to entity_name, any
+          relationship direction (mentions + co-mentions)
+        - entity_mentions: which articles mention entity_name
+        - shared_entities_between_articles: entities mentioned in both
+          article_id and article_id_2
+        - most_connected_entities: global ranking of most-mentioned entities
+          (no entity_name needed)
+        - article_verdict: final fake-news verdict + evidence for article_id
+        - claims_for_article: per-claim verdict breakdown for article_id
+        - analysis_for_article: full sentiment/topic/dialect/propaganda/
+          hate-speech profile for article_id
+
+    entity_name: match by Entity.name (NOT internal entity_id). Required for
+        entity_connections, entity_mentions, shared_entities_between_articles.
+    article_id / article_id_2: required for article_verdict, claims_for_article,
+        analysis_for_article, and shared_entities_between_articles (needs both).
+    limit: max rows to return, ignored for single-article lookups.
+
+    Returns a JSON string: {"rows": [...], "source_refs": [{"type": ...,
+    "id": ...}, ...]}
+    """
+    try:
+        if query_type == "entity_connections":
+            if not entity_name:
+                raise ValueError("entity_connections requires entity_name")
+            rows, refs = await _entity_connections(entity_name, limit)
+        elif query_type == "entity_mentions":
+            if not entity_name:
+                raise ValueError("entity_mentions requires entity_name")
+            rows, refs = await _entity_mentions(entity_name, limit)
+        elif query_type == "shared_entities_between_articles":
+            if article_id is None or article_id_2 is None:
+                raise ValueError("shared_entities_between_articles requires article_id and article_id_2")
+            rows, refs = await _shared_entities_between_articles(article_id, article_id_2, limit)
+        elif query_type == "most_connected_entities":
+            rows, refs = await _most_connected_entities(limit)
+        elif query_type == "article_verdict":
+            if article_id is None:
+                raise ValueError("article_verdict requires article_id")
+            rows, refs = await _article_verdict(article_id)
+        elif query_type == "claims_for_article":
+            if article_id is None:
+                raise ValueError("claims_for_article requires article_id")
+            rows, refs = await _claims_for_article(article_id)
+        elif query_type == "analysis_for_article":
+            if article_id is None:
+                raise ValueError("analysis_for_article requires article_id")
+            rows, refs = await _analysis_for_article(article_id)
+        else:
+            return json.dumps({"rows": [], "source_refs": [], "error": f"Unknown query_type: {query_type}"})
+    except ValueError as exc:
+        return json.dumps({"rows": [], "source_refs": [], "error": str(exc)})
+
+    return json.dumps({"rows": rows, "source_refs": refs}, default=str)

app/tools/hybrid_tool.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from langchain_core.tools import tool
+
+from app.db.neo4j import run_read_query
+from app.db.postgres import get_readonly_connection
+from app.services.nlp_client import embed
+
+# Must match FAISS_DIM / MiniLM output dim used by nlp_client.embed()
+EMBEDDING_DIM = 384
+
+
+def _to_vector_literal(vec) -> str:
+    """
+    Format an embedding as a pgvector literal, e.g. '[0.1,0.2,...]'.
+    No pgvector adapter is registered on the psycopg pool (see
+    app/db/postgres.py), so we build the literal ourselves and cast
+    with ::vector in SQL rather than relying on parameter binding.
+    """
+    return "[" + ",".join(f"{float(x):.8f}" for x in vec) + "]"
+
+
+async def _semantic_search(query_text: str, limit: int) -> list[dict[str, Any]]:
+    """Step 1 + 2: embed query_text, then pgvector cosine similarity search."""
+    vectors = await embed([query_text])
+    vector_literal = _to_vector_literal(vectors[0])
+
+    sql = """
+        SELECT id, title, url, published_at, content,
+               (search_vector <=> %s::vector) AS distance
+        FROM news_articles
+        WHERE search_vector IS NOT NULL
+        ORDER BY distance ASC
+        LIMIT %s
+    """
+    async with get_readonly_connection() as conn:
+        cur = await conn.execute(sql, (vector_literal, limit))
+        rows = await cur.fetchall()
+        columns = [desc[0] for desc in cur.description]
+
+    articles = []
+    for row in rows:
+        record = dict(zip(columns, row))
+        content = record.pop("content", None) or ""
+        snippet = content[:280] + ("..." if len(content) > 280 else "")
+        distance = record.pop("distance")
+        articles.append({
+            "id": record["id"],
+            "title": record["title"],
+            "url": record["url"],
+            "published_at": (
+                record["published_at"].isoformat()
+                if record.get("published_at") else None
+            ),
+            "snippet": snippet,
+            # cosine distance -> similarity; pgvector's <=> is cosine
+            # distance under the vector_cosine_ops opclass
+            "similarity": round(1 - distance, 4),
+        })
+    return articles
+
+
+async def _entities_for_article(article_id: int, limit: int) -> list[dict[str, Any]]:
+    """Step 3: pull mentioned entities from Neo4j for one article."""
+    query = """
+        MATCH (a:Article {article_id: $article_id})-[:MENTIONS]->(e:Entity)
+        RETURN e.name AS name, e.type AS type, e.entity_id AS entity_id
+        LIMIT $limit
+    """
+    rows = await run_read_query(query, article_id=article_id, limit=limit)
+    return [r for r in rows if r.get("entity_id")]
+
+
+@tool
+async def hybrid_search_tool(query_text: str, limit: int = 10) -> str:
+    """
+    Semantic search over article content, expanded with mentioned entities.
+
+    Use this when a question needs *finding* articles by meaning/topic
+    (not an exact stat or a known entity/article_id lookup) — e.g. "articles
+    about the election and who's mentioned in them", "coverage of the
+    drought", "find articles similar to X".
+
+    Steps:
+        1. Embed query_text (shared MiniLM endpoint, same as Django uses).
+        2. Cosine similarity search over news_articles.search_vector
+           (pgvector) to get the top-`limit` articles, ranked by similarity.
+        3. For each returned article, pull its mentioned entities from
+           Neo4j (:MENTIONS relationship).
+        4. Merge into one ranked result set with combined source_refs
+           (articles + entities).
+
+    query_text: free-text description of what to search for.
+    limit: max number of articles to return (entities per article are
+        capped separately and not counted against this).
+
+    Returns a JSON string: {"rows": [{"id", "title", "url", "published_at",
+    "snippet", "similarity", "entities": [...]}], "source_refs": [{"type":
+    "article"|"entity", "id": ...}, ...]}
+    """
+    try:
+        articles = await _semantic_search(query_text, limit)
+    except Exception as exc:  # noqa: BLE001 - surface to agent, don't crash turn
+        return json.dumps({"rows": [], "source_refs": [], "error": str(exc)})
+
+    source_refs: list[dict[str, Any]] = []
+    seen_entity_ids: set[Any] = set()
+
+    for article in articles:
+        source_refs.append({"type": "article", "id": article["id"]})
+        try:
+            entities = await _entities_for_article(article["id"], limit=10)
+        except Exception:  # noqa: BLE001 - one article's graph lookup failing shouldn't sink the rest
+            entities = []
+
+        article["entities"] = [
+            {"name": e.get("name"), "type": e.get("type"),
+             "entity_id": e.get("entity_id")}
+            for e in entities
+        ]
+        for e in entities:
+            entity_id = e.get("entity_id")
+            if entity_id is not None and entity_id not in seen_entity_ids:
+                seen_entity_ids.add(entity_id)
+                source_refs.append({"type": "entity", "id": entity_id})
+
+    # already ranked by similarity from the SQL ORDER BY
+    return json.dumps({"rows": articles, "source_refs": source_refs}, default=str)

app/tools/sql_tool.py

@@ -0,0 +1,322 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Literal, Optional
+
+from langchain_core.tools import tool
+from psycopg.rows import dict_row
+
+from app.db.postgres import get_readonly_connection
+
+Metric = Literal[
+    "fake_news_count",
+    "articles_by_topic",
+    "sentiment_breakdown",
+    "propaganda_count",
+    "hate_speech_count",
+    "dialect_breakdown",
+    "article_lookup",
+]
+
+
+async def _fetch(sql: str, params: list[Any]) -> list[dict[str, Any]]:
+    async with get_readonly_connection() as conn:
+        async with conn.cursor(row_factory=dict_row) as cur:
+            await cur.execute(sql, params)
+            return await cur.fetchall()
+
+
+def _date_filter(where: list[str], params: list[Any], date_col: str,
+                 date_from: Optional[str], date_to: Optional[str]) -> None:
+    if date_from:
+        where.append(f"{date_col} >= %s")
+        params.append(date_from)
+    if date_to:
+        where.append(f"{date_col} <= %s")
+        params.append(date_to)
+
+
+async def _fake_news_count(date_from, date_to, article_id, limit):
+    where: list[str] = []
+    params: list[Any] = []
+    _date_filter(where, params, "analyzed_at", date_from, date_to)
+    if article_id is not None:
+        where.append("article_id = %s")
+        params.append(article_id)
+    where_sql = f"WHERE {' AND '.join(where)}" if where else ""
+
+    breakdown_sql = f"""
+        SELECT verdict, COUNT(*) AS count
+        FROM article_fake_news
+        {where_sql}
+        GROUP BY verdict
+        ORDER BY count DESC
+    """
+    rows = await _fetch(breakdown_sql, params)
+
+    refs_sql = f"""
+        SELECT article_id
+        FROM article_fake_news
+        {where_sql}
+        ORDER BY analyzed_at DESC
+        LIMIT %s
+    """
+    refs = await _fetch(refs_sql, params + [limit])
+    return breakdown_sql, rows, refs
+
+
+async def _articles_by_topic(date_from, date_to, topic, article_id, limit):
+    where: list[str] = []
+    params: list[Any] = []
+    _date_filter(where, params, "analyzed_at", date_from, date_to)
+    if topic:
+        where.append("(primary_topic = %s OR secondary_topic = %s)")
+        params.extend([topic, topic])
+    if article_id is not None:
+        where.append("article_id = %s")
+        params.append(article_id)
+    where_sql = f"WHERE {' AND '.join(where)}" if where else ""
+
+    breakdown_sql = f"""
+        SELECT primary_topic, COUNT(*) AS count
+        FROM article_topic
+        {where_sql}
+        GROUP BY primary_topic
+        ORDER BY count DESC
+    """
+    rows = await _fetch(breakdown_sql, params)
+
+    refs_sql = f"""
+        SELECT article_id
+        FROM article_topic
+        {where_sql}
+        ORDER BY analyzed_at DESC
+        LIMIT %s
+    """
+    refs = await _fetch(refs_sql, params + [limit])
+    return breakdown_sql, rows, refs
+
+
+async def _sentiment_breakdown(date_from, date_to, sentiment, article_id, limit):
+    where: list[str] = []
+    params: list[Any] = []
+    _date_filter(where, params, "analyzed_at", date_from, date_to)
+    if sentiment:
+        where.append("primary_sentiment = %s")
+        params.append(sentiment)
+    if article_id is not None:
+        where.append("article_id = %s")
+        params.append(article_id)
+    where_sql = f"WHERE {' AND '.join(where)}" if where else ""
+
+    breakdown_sql = f"""
+        SELECT primary_sentiment, COUNT(*) AS count
+        FROM article_sentiment
+        {where_sql}
+        GROUP BY primary_sentiment
+        ORDER BY count DESC
+    """
+    rows = await _fetch(breakdown_sql, params)
+
+    refs_sql = f"""
+        SELECT article_id
+        FROM article_sentiment
+        {where_sql}
+        ORDER BY analyzed_at DESC
+        LIMIT %s
+    """
+    refs = await _fetch(refs_sql, params + [limit])
+    return breakdown_sql, rows, refs
+
+
+async def _propaganda_count(date_from, date_to, article_id, limit):
+    where: list[str] = []
+    params: list[Any] = []
+    _date_filter(where, params, "analyzed_at", date_from, date_to)
+    if article_id is not None:
+        where.append("article_id = %s")
+        params.append(article_id)
+    where_sql = f"WHERE {' AND '.join(where)}" if where else ""
+    propaganda_where = where + ["is_propaganda = true"]
+
+    count_sql = f"""
+        SELECT COUNT(*) AS count
+        FROM article_propaganda
+        WHERE {' AND '.join(propaganda_where)}
+    """
+    rows = await _fetch(count_sql, params)
+
+    technique_sql = f"""
+        SELECT technique, COUNT(*) AS count
+        FROM article_propaganda, unnest(techniques) AS technique
+        WHERE {' AND '.join(propaganda_where)}
+        GROUP BY technique
+        ORDER BY count DESC
+        LIMIT 10
+    """
+    technique_rows = await _fetch(technique_sql, params)
+    rows[0]["technique_breakdown"] = technique_rows
+
+    refs_sql = f"""
+        SELECT article_id
+        FROM article_propaganda
+        WHERE {' AND '.join(propaganda_where)}
+        ORDER BY analyzed_at DESC
+        LIMIT %s
+    """
+    refs = await _fetch(refs_sql, params + [limit])
+    return count_sql, rows, refs
+
+
+async def _hate_speech_count(date_from, date_to, category, article_id, limit):
+    where: list[str] = []
+    params: list[Any] = []
+    _date_filter(where, params, "analyzed_at", date_from, date_to)
+    if category:
+        where.append("predicted_category = %s")
+        params.append(category)
+    if article_id is not None:
+        where.append("article_id = %s")
+        params.append(article_id)
+    where_sql = f"WHERE {' AND '.join(where)}" if where else ""
+
+    breakdown_sql = f"""
+        SELECT predicted_category, COUNT(*) AS count
+        FROM article_hate_speech
+        {where_sql}
+        GROUP BY predicted_category
+        ORDER BY count DESC
+    """
+    rows = await _fetch(breakdown_sql, params)
+
+    refs_sql = f"""
+        SELECT article_id
+        FROM article_hate_speech
+        {where_sql}
+        ORDER BY analyzed_at DESC
+        LIMIT %s
+    """
+    refs = await _fetch(refs_sql, params + [limit])
+    return breakdown_sql, rows, refs
+
+
+async def _dialect_breakdown(date_from, date_to, dialect, article_id, limit):
+    where: list[str] = []
+    params: list[Any] = []
+    _date_filter(where, params, "analyzed_at", date_from, date_to)
+    if dialect:
+        where.append("detected_dialect = %s")
+        params.append(dialect)
+    if article_id is not None:
+        where.append("article_id = %s")
+        params.append(article_id)
+    where_sql = f"WHERE {' AND '.join(where)}" if where else ""
+
+    breakdown_sql = f"""
+        SELECT detected_dialect, COUNT(*) AS count
+        FROM article_dialect
+        {where_sql}
+        GROUP BY detected_dialect
+        ORDER BY count DESC
+    """
+    rows = await _fetch(breakdown_sql, params)
+
+    refs_sql = f"""
+        SELECT article_id
+        FROM article_dialect
+        {where_sql}
+        ORDER BY analyzed_at DESC
+        LIMIT %s
+    """
+    refs = await _fetch(refs_sql, params + [limit])
+    return breakdown_sql, rows, refs
+
+
+async def _article_lookup(date_from, date_to, article_id, keyword, limit):
+    where: list[str] = []
+    params: list[Any] = []
+    _date_filter(where, params, "published_at", date_from, date_to)
+    if article_id is not None:
+        where.append("id = %s")
+        params.append(article_id)
+    if keyword:
+        where.append("title ILIKE %s")
+        params.append(f"%{keyword}%")
+    where_sql = f"WHERE {' AND '.join(where)}" if where else ""
+
+    sql = f"""
+        SELECT id, title, url, published_at
+        FROM news_articles
+        {where_sql}
+        ORDER BY published_at DESC
+        LIMIT %s
+    """
+    rows = await _fetch(sql, params + [limit])
+    refs = [{"article_id": r["id"]} for r in rows]
+    return sql, rows, refs
+
+
+@tool
+async def sql_query_tool(
+    metric: Metric,
+    date_from: Optional[str] = None,
+    date_to: Optional[str] = None,
+    topic: Optional[str] = None,
+    sentiment: Optional[str] = None,
+    category: Optional[str] = None,
+    dialect: Optional[str] = None,
+    article_id: Optional[int] = None,
+    keyword: Optional[str] = None,
+    limit: int = 20,
+) -> str:
+    """
+    Run a pre-defined, read-only query over analyzed news articles.
+
+    metric: which statistic/lookup to run —
+        - fake_news_count: breakdown of fake-news verdicts
+          (SUPPORTED, REFUTED, PARTIALLY_TRUE, UNVERIFIABLE)
+        - articles_by_topic: article counts grouped by primary topic
+        - sentiment_breakdown: article counts grouped by sentiment
+          (positive, negative, neutral)
+        - propaganda_count: count of articles flagged as propaganda,
+          plus a technique frequency breakdown
+        - hate_speech_count: article counts grouped by hate-speech category
+          (none, other, origin, gender, religion)
+        - dialect_breakdown: article counts grouped by detected dialect
+        - article_lookup: look up specific articles by id or title keyword
+
+    date_from / date_to: ISO dates (YYYY-MM-DD). For all metrics except
+        article_lookup, these filter on the analysis timestamp
+        (`analyzed_at`) — i.e. when the ML model ran, not when the article
+        was published. For article_lookup they filter on `published_at`.
+    topic, sentiment, category, dialect: optional exact-match filters for
+        the corresponding metric.
+    article_id: restrict to a single article.
+    keyword: for article_lookup, a substring to search for in the title.
+    limit: max number of source article references / lookup rows to return.
+
+    Returns a JSON string: {"rows": [...], "sql_used": "...",
+    "source_refs": [{"article_id": ...}, ...]}
+    """
+    handlers = {
+        "fake_news_count": lambda: _fake_news_count(date_from, date_to, article_id, limit),
+        "articles_by_topic": lambda: _articles_by_topic(date_from, date_to, topic, article_id, limit),
+        "sentiment_breakdown": lambda: _sentiment_breakdown(date_from, date_to, sentiment, article_id, limit),
+        "propaganda_count": lambda: _propaganda_count(date_from, date_to, article_id, limit),
+        "hate_speech_count": lambda: _hate_speech_count(date_from, date_to, category, article_id, limit),
+        "dialect_breakdown": lambda: _dialect_breakdown(date_from, date_to, dialect, article_id, limit),
+        "article_lookup": lambda: _article_lookup(date_from, date_to, article_id, keyword, limit),
+    }
+
+    handler = handlers.get(metric)
+    if handler is None:
+        return json.dumps({"rows": [], "sql_used": "", "source_refs": [],
+                           "error": f"Unknown metric: {metric}"})
+
+    sql_used, rows, refs = await handler()
+    source_refs = [{"type": "article", "id": r["article_id"]} for r in refs]
+
+    return json.dumps(
+        {"rows": rows, "sql_used": sql_used.strip(), "source_refs": source_refs},
+        default=str,
+    )