82 lines
3 KiB
Python
82 lines
3 KiB
Python
|
"""
|
||
|
|
agent.py — Vault Tagger + Embeddings Agent
|
||
|
|
|
||
|
|
Job:
|
||
|
|
Run the vault-tagger LLM (Qwen2.5-7B via Ollama) over new or changed vault documents,
|
||
|
|
update their frontmatter with corrected tags/type/entities, then re-embed them into
|
||
|
|
Qdrant via the existing bge-m3 TEI service.
|
||
|
|
|
||
|
|
Pipeline per document:
|
||
|
|
1. Read doc + current frontmatter
|
||
|
|
2. Load vocab: topic_categories from config.yaml + entity lexicon from vocab.json
|
||
|
|
3. Call vault-tagger via Ollama /api/generate (JSON mode, temp 0.1)
|
||
|
|
4. Parse JSON response; validate fields against vocabulary
|
||
|
|
5. If confidence >= threshold: apply tags/type to frontmatter (auto_apply)
|
||
|
|
Else: flag in changelog, do not modify file
|
||
|
|
6. Re-embed via TEI bge-m3 and upsert into Qdrant (collection: vault_docs)
|
||
|
|
7. Update related wikilinks in frontmatter.related if embedding similarity > 0.85
|
||
|
|
8. Append changelog entry (file, old tags, new tags, confidence, timestamp)
|
||
|
|
|
||
|
|
State tracking:
|
||
|
|
Maintains engine/.last_sweep (ISO timestamp) to process only docs modified since last run.
|
||
|
|
Pass --full to reprocess all docs.
|
||
|
|
|
||
|
|
Implemented in: Step 5
|
||
|
|
"""
|
||
|
|
|
||
|
|
# TODO (Step 5): imports — pathlib, json, yaml, httpx or requests, datetime, argparse, logging
|
||
|
|
|
||
|
|
|
||
|
|
def load_config(config_path: str) -> dict:
|
||
|
|
"""Load and return parsed config.yaml."""
|
||
|
|
raise NotImplementedError("implemented in step 5")
|
||
|
|
|
||
|
|
|
||
|
|
def load_vocab(engine_dir: str) -> dict:
|
||
|
|
"""Load topic_categories from config + entity lexicon from vocab.json."""
|
||
|
|
raise NotImplementedError("implemented in step 5")
|
||
|
|
|
||
|
|
|
||
|
|
def get_changed_docs(vault_dir: str, since: str) -> list:
|
||
|
|
"""Return list of .md paths modified after `since` (ISO timestamp)."""
|
||
|
|
raise NotImplementedError("implemented in step 5")
|
||
|
|
|
||
|
|
|
||
|
|
def call_tagger(doc_text: str, vocab: dict, config: dict) -> dict:
|
||
|
|
"""
|
||
|
|
POST to Ollama /api/generate with vault-tagger model.
|
||
|
|
Returns parsed JSON response dict.
|
||
|
|
"""
|
||
|
|
raise NotImplementedError("implemented in step 5")
|
||
|
|
|
||
|
|
|
||
|
|
def embed_document(doc_text: str, tei_endpoint: str) -> list[float]:
|
||
|
|
"""POST to TEI bge-m3 endpoint; return embedding vector."""
|
||
|
|
raise NotImplementedError("implemented in step 5")
|
||
|
|
|
||
|
|
|
||
|
|
def upsert_qdrant(doc_id: str, vector: list[float], payload: dict, config: dict) -> None:
|
||
|
|
"""Upsert a document vector + metadata into Qdrant vault_docs collection."""
|
||
|
|
raise NotImplementedError("implemented in step 5")
|
||
|
|
|
||
|
|
|
||
|
|
def find_related(doc_id: str, vector: list[float], config: dict, threshold: float = 0.85) -> list[str]:
|
||
|
|
"""Query Qdrant for nearest neighbours above threshold; return doc ids."""
|
||
|
|
raise NotImplementedError("implemented in step 5")
|
||
|
|
|
||
|
|
|
||
|
|
def apply_tagger_result(path, result: dict, config: dict) -> dict:
|
||
|
|
"""
|
||
|
|
Write tagger output back to doc frontmatter if confidence >= threshold.
|
||
|
|
Returns summary of changes made.
|
||
|
|
"""
|
||
|
|
raise NotImplementedError("implemented in step 5")
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
"""Entry point. Parse args, load state, process changed docs, update state."""
|
||
|
|
raise NotImplementedError("implemented in step 5")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|