diff --git a/README.md b/README.md index 3db737f..4cba346 100644 --- a/README.md +++ b/README.md @@ -41,10 +41,6 @@ curl -X POST http://localhost:8000/api/v1/ask \ -H "Content-Type: application/json" \ -d '{"question":"When is the next event?"}' -curl -X POST http://localhost:8000/api/v1/ingest \ - -H "Content-Type: application/json" \ - -d '{"path":"data/announcements.json"}' - curl -X POST http://localhost:8000/api/v1/ingest/discord ``` @@ -89,7 +85,7 @@ The main goals of **mAIcro** are: │ ├── api/ # HTTP routes, schemas, error handlers │ ├── core/ # Config, logging, ingestion, providers, vector store │ └── services/ # Business logic (Q&A service) -├── data/ # Local data sources used for ingestion +├── data/ # Legacy sample data (not used by default) ├── tests/ │ ├── api/ # API route tests │ └── unit/ # Unit tests @@ -223,3 +219,4 @@ Future integrations may include: * Web dashboards * APIs for other tools * Knowledge management platforms + diff --git a/data/announcements.json b/data/announcements.json index 68ae56e..e69de29 100644 --- a/data/announcements.json +++ b/data/announcements.json @@ -1,12 +0,0 @@ -[ - { - "title": "AI Team Applications", - "content": "Applications for the AI team open on March 10.", - "date": "2026-03-10" - }, - { - "title": "FrAIday Event", - "content": "FrAIday workshop about MCP will take place this Friday.", - "date": "2026-03-14" - } -] diff --git a/src/maicro/api/routes.py b/src/maicro/api/routes.py index f33aacf..6f3126d 100644 --- a/src/maicro/api/routes.py +++ b/src/maicro/api/routes.py @@ -4,7 +4,7 @@ from fastapi import APIRouter, HTTPException -from maicro.api.schemas import AskRequest, AskResponse, IngestFileRequest, IngestResponse +from maicro.api.schemas import AskRequest, AskResponse, IngestResponse from maicro.core.config import settings from maicro.services.qa_service import ask_question @@ -37,19 +37,6 @@ async def ask(req: AskRequest): return AskResponse(question=req.question, answer=answer) -@router.post("/ingest", response_model=IngestResponse) -async def ingest_file(req: IngestFileRequest): - """Ingest documents from a local JSON file.""" - from maicro.core.ingestion import ingest_from_json - - try: - count = ingest_from_json(req.path) - except ValueError as exc: - raise HTTPException(status_code=400, detail=str(exc)) from exc - - return IngestResponse(status="ok", documents_ingested=count) - - @router.post("/ingest/discord", response_model=IngestResponse) async def ingest_discord(): """Fetch messages from Discord channels and ingest them.""" diff --git a/src/maicro/api/schemas.py b/src/maicro/api/schemas.py index 50e8028..aafb272 100644 --- a/src/maicro/api/schemas.py +++ b/src/maicro/api/schemas.py @@ -14,11 +14,6 @@ class AskResponse(BaseModel): answer: str -class IngestFileRequest(BaseModel): - source: str = "file" - path: str = "data/announcements.json" - - class IngestResponse(BaseModel): status: str documents_ingested: int diff --git a/src/maicro/cli.py b/src/maicro/cli.py index 93484cd..dc600eb 100644 --- a/src/maicro/cli.py +++ b/src/maicro/cli.py @@ -36,31 +36,20 @@ def ask_main() -> None: def ingest_main() -> None: - """Ingest data from JSON (default) or Discord from the command line.""" - from maicro.core.ingestion import ingest_from_json + """Ingest data from Discord from the command line.""" parser = argparse.ArgumentParser(prog="maicro-ingest") - parser.add_argument( - "--discord", - action="store_true", - help="Ingest from configured Discord channels instead of local JSON file.", - ) parser.add_argument( "--limit", type=int, default=200, - help="Max messages to fetch per Discord channel (used with --discord).", - ) - parser.add_argument( - "--path", - default="data/announcements.json", - help="Path to JSON file (used when --discord is not set).", + help="Max messages to fetch per Discord channel.", ) args = parser.parse_args(sys.argv[1:]) - if args.discord: - from maicro.core.ingestion import ingest_from_discord + from maicro.core.ingestion import ingest_from_discord + try: if not settings.DISCORD_BOT_TOKEN: print("Error: DISCORD_BOT_TOKEN not found in .env.") raise SystemExit(1) @@ -70,39 +59,26 @@ def ingest_main() -> None: if args.limit <= 0: print("Error: --limit must be a positive integer.") raise SystemExit(1) - - try: - result = asyncio.run(ingest_from_discord(limit_per_channel=args.limit)) - status = "partial" if result.get("errors") else "ok" + if not settings.GOOGLE_API_KEY: print( - "Discord ingestion complete. " - f"Status: {status}. Documents ingested: {result.get('total_documents', 0)}" + "Error: GOOGLE_API_KEY not found in .env. " + "Ingestion embeddings currently use Google." ) - channels = result.get("channels") or {} - errors = result.get("errors") or {} - if channels: - print(f"Per-channel counts: {channels}") - if errors: - print(f"Channel errors: {errors}") - raise SystemExit(2) - return - except KeyboardInterrupt: - print("Cancelled by user.") - raise SystemExit(130) - except Exception as exc: - print(f"Error: {exc}") - raise SystemExit(2) from exc + raise SystemExit(1) - if not settings.GOOGLE_API_KEY: + result = asyncio.run(ingest_from_discord(limit_per_channel=args.limit)) + status = "partial" if result.get("errors") else "ok" print( - "Error: GOOGLE_API_KEY not found in .env. " - "Ingestion embeddings currently use Google." + "Discord ingestion complete. " + f"Status: {status}. Documents ingested: {result.get('total_documents', 0)}" ) - raise SystemExit(1) - - try: - count = ingest_from_json(args.path) - print(f"Ingestion complete. Documents ingested: {count}") + channels = result.get("channels") or {} + errors = result.get("errors") or {} + if channels: + print(f"Per-channel counts: {channels}") + if errors: + print(f"Channel errors: {errors}") + raise SystemExit(2) except Exception as exc: print(f"Error: {exc}") raise SystemExit(2) from exc diff --git a/src/maicro/core/ingestion.py b/src/maicro/core/ingestion.py index 30aaf4c..018a6f2 100644 --- a/src/maicro/core/ingestion.py +++ b/src/maicro/core/ingestion.py @@ -2,14 +2,10 @@ Ingestion pipeline — converts raw data into LangChain Documents and pushes them into the Qdrant vector store. -Supports two sources: - 1. JSON file (existing announcements.json format) - 2. Discord messages (from the discord_fetcher) +Supports one source: + 1. Discord messages (from the discord_fetcher) """ -import json -import os - from langchain_core.documents import Document from qdrant_client.http import models as qdrant_models from qdrant_client.http.exceptions import UnexpectedResponse @@ -42,27 +38,6 @@ def _bootstrap_collection() -> None: get_vector_store.cache_clear() -def _docs_from_json_file(file_path: str) -> list[Document]: - """Load documents from a JSON file with [{title, content, date}, ...].""" - if not os.path.exists(file_path): - raise FileNotFoundError(f"Data file not found: {file_path}") - - with open(file_path, "r") as f: - data = json.load(f) - - docs: list[Document] = [] - for item in data: - content = f"{item['title']}\n{item['content']}" - metadata = { - "source": "json_file", - "file": file_path, - "date": item.get("date", ""), - "title": item["title"], - } - docs.append(Document(page_content=content, metadata=metadata)) - return docs - - def _docs_from_discord_messages( messages: list[dict], channel_id: str, @@ -155,12 +130,6 @@ def ingest_documents(documents: list[Document]) -> int: return len(documents) -def ingest_from_json(file_path: str) -> int: - """Ingest documents from a JSON file.""" - docs = _docs_from_json_file(file_path) - return ingest_documents(docs) - - async def ingest_from_discord(limit_per_channel: int = 200) -> dict: """ Fetch messages from all configured Discord channels and ingest them. diff --git a/src/maicro/services/qa_service.py b/src/maicro/services/qa_service.py index ccc410b..4a55d89 100644 --- a/src/maicro/services/qa_service.py +++ b/src/maicro/services/qa_service.py @@ -1,11 +1,8 @@ """Question-answering service built on top of retrieval-augmented generation.""" -import json import re from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeoutError from datetime import datetime, timezone -from functools import lru_cache -from pathlib import Path from langchain_core.documents import Document from langchain_core.output_parsers import StrOutputParser @@ -72,66 +69,6 @@ def _invoke_with_timeout(chain, question: str, timeout_seconds: int = _ASK_TIMEO executor.shutdown(wait=False, cancel_futures=True) -@lru_cache(maxsize=1) -def _load_local_announcements_docs() -> list[Document]: - """Load local announcements as a fallback knowledge source.""" - data_path = Path(__file__).resolve().parents[2] / "data" / "announcements.json" - if not data_path.exists(): - raise AskConfigError(f"Fallback data file not found: {data_path}") - - try: - raw = json.loads(data_path.read_text(encoding="utf-8")) - except json.JSONDecodeError as exc: - raise AskConfigError(f"Invalid JSON in fallback data file: {data_path}") from exc - - docs: list[Document] = [] - for item in raw: - if not isinstance(item, dict): - continue - title = str(item.get("title") or "Untitled") - content = str(item.get("content") or "") - date = str(item.get("date") or "unknown") - - docs.append( - Document( - page_content=f"{title}\n{content}".strip(), - metadata={ - "source": "data/announcements.json", - "date": date, - "title": title, - }, - ) - ) - - if not docs: - raise AskConfigError("Fallback data file contains no usable documents.") - - return docs - - -def _tokenize(text: str) -> set[str]: - return set(re.findall(r"[a-z0-9]+", text.lower())) - - -def _fallback_keyword_retrieve(question: str, k: int = 3) -> list[Document]: - """Very small lexical retriever used when vector embeddings are unavailable.""" - docs = _load_local_announcements_docs() - q_tokens = _tokenize(question) - - scored: list[tuple[int, Document]] = [] - for doc in docs: - doc_tokens = _tokenize(doc.page_content) - score = len(q_tokens & doc_tokens) - scored.append((score, doc)) - - scored.sort(key=lambda x: x[0], reverse=True) - top = [doc for score, doc in scored if score > 0][:k] - if top: - return top - - return [doc for _, doc in scored[:k]] - - def _format_llm_error(exc: Exception) -> str: """Convert provider errors into short messages suitable for API/CLI users.""" message = str(exc) @@ -423,11 +360,7 @@ def ask_question(question: str) -> str: vector_store = get_vector_store() retriever = vector_store.as_retriever(search_kwargs={"k": 6}) except ConfigurationError as exc: - # Allow non-Google LLM setups to run using local lexical retrieval. - if "GOOGLE_API_KEY" in str(exc): - retriever = RunnableLambda(lambda q: _fallback_keyword_retrieve(q, k=6)) - else: - raise AskConfigError(str(exc)) from exc + raise AskConfigError(str(exc)) from exc except Exception as exc: message = str(exc) lowered = message.lower() @@ -447,9 +380,10 @@ def ask_question(question: str) -> str: ) ) if is_lock or is_connection: - retriever = RunnableLambda(lambda q: _fallback_keyword_retrieve(q, k=6)) - else: - raise AskConfigError(f"Vector store initialization failed: {message}") from exc + raise AskConfigError( + "Vector store is unavailable. Start Qdrant and ingest Discord data first." + ) from exc + raise AskConfigError(f"Vector store initialization failed: {message}") from exc prompt = build_rag_prompt_template() normalized_question = _normalize_question(question) @@ -476,7 +410,6 @@ def _build_chain(model): raise AskConfigError( "Vector store is not initialized yet. " f"Qdrant collection `{settings.COLLECTION_NAME}` does not exist. " - "Ingest data first (POST /api/v1/ingest with " - '{"path":"data/announcements.json"}).' + "Ingest Discord data first (POST /api/v1/ingest/discord)." ) from exc raise AskError(_format_llm_error(exc)) from exc diff --git a/tests/unit/test_qa_service.py b/tests/unit/test_qa_service.py index be17d4b..89be575 100644 --- a/tests/unit/test_qa_service.py +++ b/tests/unit/test_qa_service.py @@ -1,7 +1,6 @@ from types import SimpleNamespace import pytest -from langchain_core.documents import Document from langchain_core.runnables import RunnableLambda from maicro.services import qa_service @@ -225,16 +224,8 @@ def test_ask_question_recency_query_returns_latest_message_when_llm_fails(monkey assert answer == latest_message -def test_ask_question_uses_keyword_fallback_when_qdrant_is_locked(monkeypatch): - fallback_calls = [] - docs = [ - Document( - page_content="Roadmap sync at 3pm", - metadata={"source": "data/announcements.json", "date": "2026-03-10"}, - ) - ] - - monkeypatch.setattr(qa_service, "get_llm", lambda: RunnableLambda(lambda _prompt: "keyword fallback answer")) +def test_ask_question_errors_when_qdrant_is_locked(monkeypatch): + monkeypatch.setattr(qa_service, "get_llm", lambda: RunnableLambda(lambda _prompt: "ignored")) monkeypatch.setattr( qa_service, "get_vector_store", @@ -242,26 +233,11 @@ def test_ask_question_uses_keyword_fallback_when_qdrant_is_locked(monkeypatch): RuntimeError("Storage already accessed by another instance of Qdrant client") ), ) - monkeypatch.setattr( - qa_service, - "_fallback_keyword_retrieve", - lambda question, k=6: fallback_calls.append((question, k)) or docs, - ) - monkeypatch.setattr( - qa_service, - "build_rag_prompt_template", - lambda: RunnableLambda(lambda data: f"{data['question']}\n{data['context']}"), - ) - monkeypatch.setattr( - qa_service, - "_invoke_with_timeout", - lambda chain, question, timeout_seconds=30: chain.invoke(question), - ) - answer = qa_service.ask_question("roadmap sync") + with pytest.raises(qa_service.AskConfigError) as excinfo: + qa_service.ask_question("roadmap sync") - assert answer == "keyword fallback answer" - assert fallback_calls == [("roadmap sync", 6)] + assert "vector store is unavailable" in str(excinfo.value).lower() def test_ask_question_missing_collection_raises_config_error(monkeypatch):