diff --git a/.env b/.env new file mode 100644 index 0000000000000000000000000000000000000000..24793b7a0ff5a6708e0716f85d008aaf8a30ae3c --- /dev/null +++ b/.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" \ No newline at end of file diff --git a/.env.example b/.env.example new file mode 100644 index 0000000000000000000000000000000000000000..1e63d7cc570adc39537744d70a32d96add74b341 --- /dev/null +++ b/.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 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000000000000000000000000000000000000..d4fe15c5eb98c7211244544a0b6dc9e8010c9170 --- /dev/null +++ b/.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/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000000000000000000000000000000000000..206d43196fffc6cf85fb86202cdec55b583ed47e --- /dev/null +++ b/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"] diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/app/__pycache__/__init__.cpython-312.pyc b/app/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..91dbb4b80a019488a2e73a900c743f52a50baa9b Binary files /dev/null and b/app/__pycache__/__init__.cpython-312.pyc differ diff --git a/app/__pycache__/config.cpython-312.pyc b/app/__pycache__/config.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..cdc3f88edd3c945774680124179df1c7ba65126d Binary files /dev/null and b/app/__pycache__/config.cpython-312.pyc differ diff --git a/app/__pycache__/main.cpython-312.pyc b/app/__pycache__/main.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..5a91db78b3e45175ca2305d98368719c5bd377d7 Binary files /dev/null and b/app/__pycache__/main.cpython-312.pyc differ diff --git a/app/agent/__init__.py b/app/agent/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/app/agent/__pycache__/__init__.cpython-312.pyc b/app/agent/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..3ef06f6f864c0f5b5465d8b4b3390d2ebc2866c6 Binary files /dev/null and b/app/agent/__pycache__/__init__.cpython-312.pyc differ diff --git a/app/agent/__pycache__/graph.cpython-312.pyc b/app/agent/__pycache__/graph.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..9055466c593da22dffb7c6a6641f79e0e613eb14 Binary files /dev/null and b/app/agent/__pycache__/graph.cpython-312.pyc differ diff --git a/app/agent/__pycache__/llm.cpython-312.pyc b/app/agent/__pycache__/llm.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..d83879a2d23921cfb90b8febc1d335070f77783f Binary files /dev/null and b/app/agent/__pycache__/llm.cpython-312.pyc differ diff --git a/app/agent/__pycache__/prompts.cpython-312.pyc b/app/agent/__pycache__/prompts.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..9ddb0203c971ef2d35590a07d81784ad889ecfae Binary files /dev/null and b/app/agent/__pycache__/prompts.cpython-312.pyc differ diff --git a/app/agent/__pycache__/state.cpython-312.pyc b/app/agent/__pycache__/state.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..44b11ac72ed17917e76ccbf2bba16160e1fa478c Binary files /dev/null and b/app/agent/__pycache__/state.cpython-312.pyc differ diff --git a/app/agent/graph.py b/app/agent/graph.py new file mode 100644 index 0000000000000000000000000000000000000000..a2b4338c0428e89f0a158bef6700f41548c1bacc --- /dev/null +++ b/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} diff --git a/app/agent/llm.py b/app/agent/llm.py new file mode 100644 index 0000000000000000000000000000000000000000..fd006e754292c6bfc52e7634a3201b2923478e3b --- /dev/null +++ b/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 "" \ No newline at end of file diff --git a/app/agent/prompts.py b/app/agent/prompts.py new file mode 100644 index 0000000000000000000000000000000000000000..e562f4cc287f26b10a467ce359a0f3c7092b74ba --- /dev/null +++ b/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. +""" diff --git a/app/agent/state.py b/app/agent/state.py new file mode 100644 index 0000000000000000000000000000000000000000..58bd17aa7696554a9a33d0f2e66a925ccbcdcffe --- /dev/null +++ b/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] \ No newline at end of file diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000000000000000000000000000000000000..f53d8d2118991788baa38e2edc7e5572150fbc92 --- /dev/null +++ b/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() diff --git a/app/db/__init__.py b/app/db/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/app/db/__pycache__/__init__.cpython-312.pyc b/app/db/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..1e7ed181eaba509d6cdd4e9513a8cd54bce7f4d7 Binary files /dev/null and b/app/db/__pycache__/__init__.cpython-312.pyc differ diff --git a/app/db/__pycache__/neo4j.cpython-312.pyc b/app/db/__pycache__/neo4j.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..8d2baab2d5366773733066f89aeb4823e4308a04 Binary files /dev/null and b/app/db/__pycache__/neo4j.cpython-312.pyc differ diff --git a/app/db/__pycache__/postgres.cpython-312.pyc b/app/db/__pycache__/postgres.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..446323b8dfdc39736227e333a4f9d9c58c636aff Binary files /dev/null and b/app/db/__pycache__/postgres.cpython-312.pyc differ diff --git a/app/db/neo4j.py b/app/db/neo4j.py new file mode 100644 index 0000000000000000000000000000000000000000..fb4c2367c060864933c4ba464f2d9544bfe36b34 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/app/db/postgres.py b/app/db/postgres.py new file mode 100644 index 0000000000000000000000000000000000000000..e6a377ad62c507628d7da950c6a34caa170c2246 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/app/deps.py b/app/deps.py new file mode 100644 index 0000000000000000000000000000000000000000..2eaa06348130ca05704c45f3158db9f7399af4b4 --- /dev/null +++ b/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)] \ No newline at end of file diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000000000000000000000000000000000000..efa237c9627b82a7831f2659af96c94c0c3fd4f8 --- /dev/null +++ b/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"], + ) diff --git a/app/memory/__init__.py b/app/memory/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/app/memory/__pycache__/__init__.cpython-312.pyc b/app/memory/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..2bb1173135c24f9e63a497c6b0102e407e2af2d4 Binary files /dev/null and b/app/memory/__pycache__/__init__.cpython-312.pyc differ diff --git a/app/memory/__pycache__/history_trimmer.cpython-312.pyc b/app/memory/__pycache__/history_trimmer.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..fd9ca5f724f471f5732d9061b15580eea0861eb4 Binary files /dev/null and b/app/memory/__pycache__/history_trimmer.cpython-312.pyc differ diff --git a/app/memory/__pycache__/redis_checkpointer.cpython-312.pyc b/app/memory/__pycache__/redis_checkpointer.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..b7fb410a7bd752859597c8640b14f35e39a4aff5 Binary files /dev/null and b/app/memory/__pycache__/redis_checkpointer.cpython-312.pyc differ diff --git a/app/memory/history_trimmer.py b/app/memory/history_trimmer.py new file mode 100644 index 0000000000000000000000000000000000000000..8fe87c11d8377787b3b7a61aed9a0255973c9f2b --- /dev/null +++ b/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]}) \ No newline at end of file diff --git a/app/memory/redis_checkpointer.py b/app/memory/redis_checkpointer.py new file mode 100644 index 0000000000000000000000000000000000000000..53234f84a61240bd8fe6db93525c7b13ebba0cf2 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/app/schemas/__pycache__/__init__.cpython-312.pyc b/app/schemas/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..ec4e0f52e0387a0d8db47b15491b1b697d5c146c Binary files /dev/null and b/app/schemas/__pycache__/__init__.cpython-312.pyc differ diff --git a/app/schemas/__pycache__/request.cpython-312.pyc b/app/schemas/__pycache__/request.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..ca642016571e82fd20dc21698a1927c501193ed2 Binary files /dev/null and b/app/schemas/__pycache__/request.cpython-312.pyc differ diff --git a/app/schemas/__pycache__/response.cpython-312.pyc b/app/schemas/__pycache__/response.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..2b6594871b60bfaad026da819f388e3ea84fab0e Binary files /dev/null and b/app/schemas/__pycache__/response.cpython-312.pyc differ diff --git a/app/schemas/request.py b/app/schemas/request.py new file mode 100644 index 0000000000000000000000000000000000000000..22eedd9ceff68dab37815880a6234fcd67d1dece --- /dev/null +++ b/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 \ No newline at end of file diff --git a/app/schemas/response.py b/app/schemas/response.py new file mode 100644 index 0000000000000000000000000000000000000000..fd5968902dbb144bd9f562392eaf7623f0b99b58 --- /dev/null +++ b/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] = [] diff --git a/app/services/__pycache__/nlp_client.cpython-312.pyc b/app/services/__pycache__/nlp_client.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..445e30cea97d97c67d6b7dbc198ea06eab119573 Binary files /dev/null and b/app/services/__pycache__/nlp_client.cpython-312.pyc differ diff --git a/app/services/nlp_client.py b/app/services/nlp_client.py new file mode 100644 index 0000000000000000000000000000000000000000..937ab04f2b52be23799d905037bcaa6cbe3f3326 --- /dev/null +++ b/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") \ No newline at end of file diff --git a/app/tools/__init__.py b/app/tools/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/app/tools/__pycache__/__init__.cpython-312.pyc b/app/tools/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..9e4741c899aa551119f896fea1f16442991b21ba Binary files /dev/null and b/app/tools/__pycache__/__init__.cpython-312.pyc differ diff --git a/app/tools/__pycache__/article_tool.cpython-312.pyc b/app/tools/__pycache__/article_tool.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..c42edce9d6fb392da1c6670a5d6666eb402d5bf4 Binary files /dev/null and b/app/tools/__pycache__/article_tool.cpython-312.pyc differ diff --git a/app/tools/__pycache__/graph_tool.cpython-312.pyc b/app/tools/__pycache__/graph_tool.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..ac2d34ce97671e8b1efb3e22a3e7f073da01b6b7 Binary files /dev/null and b/app/tools/__pycache__/graph_tool.cpython-312.pyc differ diff --git a/app/tools/__pycache__/sql_tool.cpython-312.pyc b/app/tools/__pycache__/sql_tool.cpython-312.pyc new file mode 100644 index 0000000000000000000000000000000000000000..fd9d40dbb092ba97776a9628ee439e1ec003f693 Binary files /dev/null and b/app/tools/__pycache__/sql_tool.cpython-312.pyc differ diff --git a/app/tools/article_tool.py b/app/tools/article_tool.py new file mode 100644 index 0000000000000000000000000000000000000000..ab2fd5a757f48f033bba984e2678df327c182c8c --- /dev/null +++ b/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"]}], + } diff --git a/app/tools/graph_tool.py b/app/tools/graph_tool.py new file mode 100644 index 0000000000000000000000000000000000000000..ef3ed23cc4c417a89a5f3f90b7475995e00f1b94 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/app/tools/hybrid_tool.py b/app/tools/hybrid_tool.py new file mode 100644 index 0000000000000000000000000000000000000000..58bcdfc57d0271e3c7cf200ced07b0223e8d00ae --- /dev/null +++ b/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) diff --git a/app/tools/sql_tool.py b/app/tools/sql_tool.py new file mode 100644 index 0000000000000000000000000000000000000000..82d61aa03f0dd7458334cc016845f17bdb410c3a --- /dev/null +++ b/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, + ) diff --git a/docker-compose.override.yml b/docker-compose.override.yml new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000000000000000000000000000000000000..9d3e8147687cd416ae8bed00c6ad31f388dd24fe --- /dev/null +++ b/requirements.txt @@ -0,0 +1,28 @@ +# --- Web framework --- +fastapi==0.115.6 +uvicorn[standard]==0.34.0 +pydantic==2.10.4 +pydantic-settings==2.7.1 + +# --- LLM / Agent --- +groq==0.13.1 +langgraph==0.2.62 +langchain-groq==0.2.3 +langchain-core==0.3.28 + +# --- Databases --- +sqlalchemy==2.0.36 +psycopg[binary,pool] +neo4j==5.27.0 + +# --- Memory / cache --- +redis==5.2.1 +langgraph-checkpoint-redis + +# --- Utils --- +python-dotenv==1.0.1 + +# --- Testing --- +pytest==8.3.4 +pytest-asyncio==0.25.0 +httpx==0.28.1 \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/tests/test_agent_e2e.py b/tests/test_agent_e2e.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/tests/test_graph_tool.py b/tests/test_graph_tool.py new file mode 100644 index 0000000000000000000000000000000000000000..6185e0850f7f8130a081437381116d9344b5bdcf --- /dev/null +++ b/tests/test_graph_tool.py @@ -0,0 +1,14 @@ +import asyncio +import sys + +if sys.platform == "win32": + asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) + +from app.tools.graph_tool import graph_query_tool + + +async def main(): + result = await graph_query_tool(query_type="most_connected_entities", limit=5) + print(result) + +asyncio.run(main()) diff --git a/tests/test_sql_tool.py b/tests/test_sql_tool.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391