| """ |
| knowledge_api.py — Public interface for the Knowledge Agent |
| ============================================================ |
| This module is the integration layer for the Master Orchestrator and any |
| other agent that needs to query your personal document store. |
| |
| Usage from another agent: |
| from knowledge_agent.knowledge_api import query_knowledge, index_docs |
| |
| It is intentionally stateless at the call level — the KnowledgeIndexer |
| caches the ChromaDB connection internally for performance. |
| """ |
|
|
| from typing import List, Dict, Optional |
| from indexer import KnowledgeIndexer |
| from llm import generate_answer, stream_answer |
|
|
| |
| |
| _indexer: Optional[KnowledgeIndexer] = None |
|
|
|
|
| def _get_indexer() -> KnowledgeIndexer: |
| global _indexer |
| if _indexer is None: |
| _indexer = KnowledgeIndexer() |
| return _indexer |
|
|
|
|
| |
|
|
| def query_knowledge( |
| question: str, |
| top_k: int = 5, |
| return_sources: bool = True, |
| ) -> Dict: |
| """ |
| PRIMARY ENTRY POINT for the Master Orchestrator and other agents. |
| |
| Ask a question against your indexed personal documents. |
| |
| Args: |
| question: Natural language question. |
| top_k: Number of chunks to retrieve from ChromaDB. |
| return_sources: If True, include source filenames in the response. |
| |
| Returns: |
| { |
| "answer": str, # LLM-generated, grounded answer |
| "sources": [str, ...], # source filenames (if return_sources=True) |
| "chunks_used": int, # number of relevant chunks found |
| "question": str, # echo back for logging |
| } |
| |
| Example: |
| >>> result = query_knowledge("What is our refund policy?") |
| >>> print(result["answer"]) |
| """ |
| indexer = _get_indexer() |
| chunks = indexer.query(question, top_k=top_k) |
|
|
| if not chunks: |
| return { |
| "answer": "No documents have been indexed yet. " |
| "Please run `index_docs()` first.", |
| "sources": [], |
| "chunks_used": 0, |
| "question": question, |
| } |
|
|
| result = generate_answer(question, chunks) |
| result["question"] = question |
|
|
| if not return_sources: |
| result.pop("sources", None) |
|
|
| return result |
|
|
|
|
| def index_docs(docs_dir: str = "./documents", force: bool = False): |
| """ |
| Index (or re-index) all supported documents in docs_dir. |
| Called by the CLI, the web UI, and can be called by the Orchestrator |
| to trigger a refresh after new documents are added. |
| |
| Args: |
| docs_dir: Path to the folder containing your documents. |
| force: Re-index everything even if files haven't changed. |
| """ |
| indexer = _get_indexer() |
| indexer.index_directory(docs_dir=docs_dir, force=force) |
|
|
|
|
| def index_single(path: str): |
| """Index a single file immediately (called by file-watcher or API upload).""" |
| indexer = _get_indexer() |
| indexer.index_single_file(path) |
|
|
|
|
| def get_knowledge_stats() -> Dict: |
| """Return stats about the current knowledge base (for health checks).""" |
| return _get_indexer().get_stats() |
|
|
|
|
| def list_indexed_docs() -> List[Dict]: |
| """List all documents currently indexed.""" |
| return _get_indexer().list_documents() |
|
|
|
|
| def delete_doc(filename: str): |
| """Remove a document and all its chunks from the index.""" |
| _get_indexer().delete_document(filename) |
|
|
|
|
| def stream_knowledge_answer(question: str, top_k: int = 5): |
| """ |
| Streaming variant — yields tokens one by one. |
| Used by the FastAPI /stream endpoint for the web UI. |
| """ |
| indexer = _get_indexer() |
| chunks = indexer.query(question, top_k=top_k) |
| if not chunks: |
| yield "No documents indexed yet." |
| return |
| yield from stream_answer(question, chunks) |
