331 lines
15 KiB
Python
331 lines
15 KiB
Python
"""
|
|
FILE: app/core/ingestion.py
|
|
DESCRIPTION: Haupt-Ingestion-Logik. Transformiert Markdown in den Graphen.
|
|
WP-20: Optimiert für OpenRouter (openai/gpt-oss-20b:free) als Primary.
|
|
WP-22: Fallback-Unterstützung für Google Gemini und Ollama.
|
|
FIX: Dynamische Provider-Wahl und Modell-Zuweisung für den Turbo-Modus.
|
|
VERSION: 2.11.9
|
|
STATUS: Active
|
|
"""
|
|
import os
|
|
import json
|
|
import re
|
|
import logging
|
|
import asyncio
|
|
import time
|
|
from typing import Dict, List, Optional, Tuple, Any
|
|
|
|
# Core Module Imports
|
|
from app.core.parser import (
|
|
read_markdown,
|
|
normalize_frontmatter,
|
|
validate_required_frontmatter,
|
|
extract_edges_with_context,
|
|
)
|
|
from app.core.note_payload import make_note_payload
|
|
from app.core.chunker import assemble_chunks, get_chunk_config
|
|
from app.core.chunk_payload import make_chunk_payloads
|
|
|
|
# Fallback für Edges
|
|
try:
|
|
from app.core.derive_edges import build_edges_for_note
|
|
except ImportError:
|
|
def build_edges_for_note(*args, **kwargs): return []
|
|
|
|
from app.core.qdrant import QdrantConfig, get_client, ensure_collections, ensure_payload_indexes
|
|
from app.core.qdrant_points import (
|
|
points_for_chunks,
|
|
points_for_note,
|
|
points_for_edges,
|
|
upsert_batch,
|
|
)
|
|
|
|
from app.services.embeddings_client import EmbeddingsClient
|
|
from app.services.edge_registry import registry as edge_registry
|
|
from app.services.llm_service import LLMService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# --- Helper ---
|
|
def extract_json_from_response(text: str) -> Any:
|
|
"""Extrahiert JSON-Daten, selbst wenn sie in Markdown-Blöcken stehen."""
|
|
if not text: return []
|
|
# Suche nach ```json ... ``` oder ``` ... ```
|
|
match = re.search(r"```(?:json)?\s*(.*?)\s*```", text, re.DOTALL)
|
|
clean_text = match.group(1) if match else text
|
|
try:
|
|
return json.loads(clean_text.strip())
|
|
except json.JSONDecodeError:
|
|
# Versuch: Alles vor der ersten [ und nach der letzten ] entfernen
|
|
start = clean_text.find('[')
|
|
end = clean_text.rfind(']') + 1
|
|
if start != -1 and end != 0:
|
|
try: return json.loads(clean_text[start:end])
|
|
except: pass
|
|
raise
|
|
|
|
def load_type_registry(custom_path: Optional[str] = None) -> dict:
|
|
import yaml
|
|
from app.config import get_settings
|
|
settings = get_settings()
|
|
path = custom_path or settings.MINDNET_TYPES_FILE
|
|
if not os.path.exists(path): return {}
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f: return yaml.safe_load(f) or {}
|
|
except Exception: return {}
|
|
|
|
def resolve_note_type(requested: Optional[str], reg: dict) -> str:
|
|
types = reg.get("types", {})
|
|
if requested and requested in types: return requested
|
|
return "concept"
|
|
|
|
def effective_chunk_profile_name(fm: dict, note_type: str, reg: dict) -> str:
|
|
override = fm.get("chunking_profile") or fm.get("chunk_profile")
|
|
if override and isinstance(override, str): return override
|
|
t_cfg = reg.get("types", {}).get(note_type, {})
|
|
if t_cfg:
|
|
cp = t_cfg.get("chunking_profile") or t_cfg.get("chunk_profile")
|
|
if cp: return cp
|
|
return reg.get("defaults", {}).get("chunking_profile", "sliding_standard")
|
|
|
|
def effective_retriever_weight(fm: dict, note_type: str, reg: dict) -> float:
|
|
override = fm.get("retriever_weight")
|
|
if override is not None:
|
|
try: return float(override)
|
|
except: pass
|
|
t_cfg = reg.get("types", {}).get(note_type, {})
|
|
if t_cfg and "retriever_weight" in t_cfg: return float(t_cfg["retriever_weight"])
|
|
return float(reg.get("defaults", {}).get("retriever_weight", 1.0))
|
|
|
|
|
|
class IngestionService:
|
|
def __init__(self, collection_prefix: str = None):
|
|
from app.config import get_settings
|
|
self.settings = get_settings()
|
|
|
|
self.prefix = collection_prefix or self.settings.COLLECTION_PREFIX
|
|
self.cfg = QdrantConfig.from_env()
|
|
self.cfg.prefix = self.prefix
|
|
self.client = get_client(self.cfg)
|
|
self.dim = self.settings.VECTOR_SIZE
|
|
self.registry = load_type_registry()
|
|
self.embedder = EmbeddingsClient()
|
|
self.llm = LLMService()
|
|
|
|
self.active_hash_mode = self.settings.CHANGE_DETECTION_MODE
|
|
|
|
try:
|
|
ensure_collections(self.client, self.prefix, self.dim)
|
|
ensure_payload_indexes(self.client, self.prefix)
|
|
except Exception as e:
|
|
logger.warning(f"DB init warning: {e}")
|
|
|
|
def _get_chunk_config_by_profile(self, profile_name: str, note_type: str) -> Dict[str, Any]:
|
|
"""Holt die Chunker-Parameter für ein spezifisches Profil."""
|
|
profiles = self.registry.get("chunking_profiles", {})
|
|
if profile_name in profiles:
|
|
cfg = profiles[profile_name].copy()
|
|
if "overlap" in cfg and isinstance(cfg["overlap"], list):
|
|
cfg["overlap"] = tuple(cfg["overlap"])
|
|
return cfg
|
|
return get_chunk_config(note_type)
|
|
|
|
async def _perform_smart_edge_allocation(self, text: str, note_id: str) -> List[Dict]:
|
|
"""
|
|
WP-20: Nutzt das Hybrid LLM für die semantische Kanten-Extraktion.
|
|
Bevorzugt den primär eingestellten Provider (z.B. OpenRouter).
|
|
"""
|
|
# 1. Provider & Modell Bestimmung (User-Request: OpenRouter Primary)
|
|
provider = self.settings.MINDNET_LLM_PROVIDER
|
|
|
|
if provider == "openrouter":
|
|
model = self.settings.OPENROUTER_MODEL
|
|
elif provider == "gemini":
|
|
model = self.settings.GEMINI_MODEL
|
|
else:
|
|
model = self.settings.LLM_MODEL
|
|
|
|
logger.info(f"🚀 [Ingestion] Turbo-Mode: Extracting edges for '{note_id}' using {model} on {provider}")
|
|
|
|
edge_registry.ensure_latest()
|
|
valid_types_str = ", ".join(sorted(list(edge_registry.valid_types)))
|
|
|
|
template = self.llm.get_prompt("edge_extraction", provider)
|
|
|
|
try:
|
|
# FIX: Format-Safety Block gegen KeyError: '"to"'
|
|
try:
|
|
prompt = template.format(
|
|
text=text[:6000],
|
|
note_id=note_id,
|
|
valid_types=valid_types_str
|
|
)
|
|
except KeyError as ke:
|
|
logger.error(f"❌ [Ingestion] Prompt-Template Fehler (Variable {ke} fehlt). Prüfe prompts.yaml Maskierung.")
|
|
return []
|
|
|
|
response_json = await self.llm.generate_raw_response(
|
|
prompt=prompt, priority="background", force_json=True,
|
|
provider=provider, model_override=model
|
|
)
|
|
|
|
raw_data = extract_json_from_response(response_json)
|
|
|
|
if isinstance(raw_data, dict):
|
|
for k in ["edges", "links", "results", "kanten"]:
|
|
if k in raw_data and isinstance(raw_data[k], list):
|
|
raw_data = raw_data[k]
|
|
break
|
|
|
|
if not isinstance(raw_data, list): return []
|
|
|
|
processed = []
|
|
for item in raw_data:
|
|
# FIX: Schutz vor 'str' object does not support item assignment
|
|
if isinstance(item, dict) and "to" in item:
|
|
item["provenance"] = "semantic_ai"
|
|
item["line"] = f"ai-{provider}"
|
|
processed.append(item)
|
|
elif isinstance(item, str) and ":" in item:
|
|
parts = item.split(":", 1)
|
|
processed.append({
|
|
"to": parts[1].strip(),
|
|
"kind": parts[0].strip(),
|
|
"provenance": "semantic_ai",
|
|
"line": f"ai-{provider}"
|
|
})
|
|
return processed
|
|
|
|
except Exception as e:
|
|
logger.warning(f"⚠️ [Ingestion] Smart Edge Allocation failed for {note_id}: {e}")
|
|
return []
|
|
|
|
async def process_file(
|
|
self, file_path: str, vault_root: str,
|
|
force_replace: bool = False, apply: bool = False, purge_before: bool = False,
|
|
note_scope_refs: bool = False, hash_source: str = "parsed", hash_normalize: str = "canonical"
|
|
) -> Dict[str, Any]:
|
|
"""Verarbeitet eine Markdown-Datei und schreibt sie in den Graphen."""
|
|
result = {"path": file_path, "status": "skipped", "changed": False, "error": None}
|
|
|
|
try:
|
|
parsed = read_markdown(file_path)
|
|
if not parsed: return {**result, "error": "Empty file"}
|
|
fm = normalize_frontmatter(parsed.frontmatter)
|
|
validate_required_frontmatter(fm)
|
|
except Exception as e:
|
|
return {**result, "error": f"Validation failed: {str(e)}"}
|
|
|
|
status = fm.get("status", "draft").lower().strip()
|
|
if status in ["system", "template", "archive", "hidden"]:
|
|
return {**result, "status": "skipped", "reason": f"lifecycle_{status}"}
|
|
|
|
note_type = resolve_note_type(fm.get("type"), self.registry)
|
|
fm["type"] = note_type
|
|
effective_profile = effective_chunk_profile_name(fm, note_type, self.registry)
|
|
effective_weight = effective_retriever_weight(fm, note_type, self.registry)
|
|
|
|
try:
|
|
note_pl = make_note_payload(parsed, vault_root=vault_root, hash_normalize=hash_normalize, hash_source=hash_source, file_path=file_path)
|
|
note_pl["retriever_weight"] = effective_weight
|
|
note_pl["chunk_profile"] = effective_profile
|
|
note_pl["status"] = status
|
|
note_id = note_pl["note_id"]
|
|
except Exception as e:
|
|
return {**result, "error": f"Payload failed: {str(e)}"}
|
|
|
|
old_payload = None if force_replace else self._fetch_note_payload(note_id)
|
|
check_key = f"{self.active_hash_mode}:{hash_source}:{hash_normalize}"
|
|
old_hash = (old_payload or {}).get("hashes", {}).get(check_key)
|
|
new_hash = note_pl.get("hashes", {}).get(check_key)
|
|
|
|
should_write = force_replace or (not old_payload) or (old_hash != new_hash) or any(self._artifacts_missing(note_id))
|
|
|
|
if not should_write: return {**result, "status": "unchanged", "note_id": note_id}
|
|
if not apply: return {**result, "status": "dry-run", "changed": True, "note_id": note_id}
|
|
|
|
try:
|
|
body_text = getattr(parsed, "body", "") or ""
|
|
if hasattr(edge_registry, "ensure_latest"): edge_registry.ensure_latest()
|
|
|
|
chunk_config = self._get_chunk_config_by_profile(effective_profile, note_type)
|
|
chunks = await assemble_chunks(fm["id"], body_text, fm["type"], config=chunk_config)
|
|
chunk_pls = make_chunk_payloads(fm, note_pl["path"], chunks, note_text=body_text)
|
|
|
|
vecs = await self.embedder.embed_documents([c.get("window") or c.get("text") or "" for c in chunk_pls]) if chunk_pls else []
|
|
|
|
edges = []
|
|
context = {"file": file_path, "note_id": note_id}
|
|
|
|
for e in extract_edges_with_context(parsed):
|
|
e["kind"] = edge_registry.resolve(edge_type=e["kind"], provenance="explicit", context={**context, "line": e.get("line")})
|
|
edges.append(e)
|
|
|
|
ai_edges = await self._perform_smart_edge_allocation(body_text, note_id)
|
|
for e in ai_edges:
|
|
e["kind"] = edge_registry.resolve(edge_type=e.get("kind"), provenance="semantic_ai", context={**context, "line": e.get("line")})
|
|
edges.append(e)
|
|
|
|
try:
|
|
sys_edges = build_edges_for_note(note_id, chunk_pls, note_level_references=note_pl.get("references", []), include_note_scope_refs=note_scope_refs)
|
|
except: sys_edges = build_edges_for_note(note_id, chunk_pls)
|
|
|
|
for e in sys_edges:
|
|
valid_kind = edge_registry.resolve(edge_type=e.get("kind", "belongs_to"), provenance="structure", context={**context, "line": "system"})
|
|
if valid_kind:
|
|
e["kind"] = valid_kind
|
|
edges.append(e)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Processing failed for {file_path}: {e}", exc_info=True)
|
|
return {**result, "error": f"Processing failed: {str(e)}"}
|
|
|
|
try:
|
|
if purge_before and old_payload: self._purge_artifacts(note_id)
|
|
n_name, n_pts = points_for_note(self.prefix, note_pl, None, self.dim)
|
|
upsert_batch(self.client, n_name, n_pts)
|
|
|
|
if chunk_pls and vecs:
|
|
c_name, c_pts = points_for_chunks(self.prefix, chunk_pls, vecs)
|
|
upsert_batch(self.client, c_name, c_pts)
|
|
|
|
if edges:
|
|
e_name, e_pts = points_for_edges(self.prefix, edges)
|
|
upsert_batch(self.client, e_name, e_pts)
|
|
|
|
return {"path": file_path, "status": "success", "changed": True, "note_id": note_id, "chunks_count": len(chunk_pls), "edges_count": len(edges)}
|
|
except Exception as e:
|
|
return {**result, "error": f"DB Upsert failed: {e}"}
|
|
|
|
def _fetch_note_payload(self, note_id: str) -> Optional[dict]:
|
|
from qdrant_client.http import models as rest
|
|
try:
|
|
f = rest.Filter(must=[rest.FieldCondition(key="note_id", match=rest.MatchValue(value=note_id))])
|
|
pts, _ = self.client.scroll(collection_name=f"{self.prefix}_notes", scroll_filter=f, limit=1, with_payload=True)
|
|
return pts[0].payload if pts else None
|
|
except: return None
|
|
|
|
def _artifacts_missing(self, note_id: str) -> Tuple[bool, bool]:
|
|
from qdrant_client.http import models as rest
|
|
try:
|
|
f = rest.Filter(must=[rest.FieldCondition(key="note_id", match=rest.MatchValue(value=note_id))])
|
|
c_pts, _ = self.client.scroll(collection_name=f"{self.prefix}_chunks", scroll_filter=f, limit=1)
|
|
e_pts, _ = self.client.scroll(collection_name=f"{self.prefix}_edges", scroll_filter=f, limit=1)
|
|
return (not bool(c_pts)), (not bool(e_pts))
|
|
except: return True, True
|
|
|
|
def _purge_artifacts(self, note_id: str):
|
|
from qdrant_client.http import models as rest
|
|
f = rest.Filter(must=[rest.FieldCondition(key="note_id", match=rest.MatchValue(value=note_id))])
|
|
for suffix in ["chunks", "edges"]:
|
|
try: self.client.delete(collection_name=f"{self.prefix}_{suffix}", points_selector=rest.FilterSelector(filter=f))
|
|
except: pass
|
|
|
|
async def create_from_text(self, markdown_content: str, filename: str, vault_root: str, folder: str = "00_Inbox") -> Dict[str, Any]:
|
|
target_dir = os.path.join(vault_root, folder)
|
|
os.makedirs(target_dir, exist_ok=True)
|
|
file_path = os.path.join(target_dir, filename)
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
f.write(markdown_content)
|
|
await asyncio.sleep(0.1)
|
|
return await self.process_file(file_path=file_path, vault_root=vault_root, apply=True, force_replace=True, purge_before=True) |