137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
app/core/qdrant_points.py
|
|
|
|
Zweck
|
|
- Gemeinsame Helfer zum Erzeugen von Qdrant-Points für Notes, Chunks und Edges.
|
|
- Abwärtskompatibel zu altem Edge-Payload-Schema aus edges.py:
|
|
- alt: {'edge_type','src_id','dst_id', ...}
|
|
- neu: {'kind','source_id','target_id', ...}
|
|
|
|
Version
|
|
- 1.3 (2025-09-08)
|
|
|
|
Änderungen (ggü. 1.2)
|
|
- points_for_edges() akzeptiert jetzt beide Edge-Schemata.
|
|
- Normalisiert alte Felder auf 'kind' / 'source_id' / 'target_id' und schreibt eine
|
|
stabile 'edge_id' zurück in die Payload.
|
|
- Verhindert, dass mehrere Edges dieselbe Point-ID erhalten (Root Cause deiner 1-Edge-Sammlung).
|
|
|
|
Aufruf / Verwendung
|
|
- Wird von Import-/Backfill-Skripten via:
|
|
from app.core.qdrant_points import points_for_note, points_for_chunks, points_for_edges, upsert_batch
|
|
eingebunden. Keine CLI.
|
|
|
|
Hinweise
|
|
- Edges bekommen absichtlich einen 1D-Dummy-Vektor [0.0], damit Qdrant das Objekt akzeptiert.
|
|
- Die Point-IDs werden deterministisch aus stabilen Strings (UUIDv5) abgeleitet.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import uuid
|
|
from typing import List, Tuple
|
|
from qdrant_client.http import models as rest
|
|
|
|
|
|
def _names(prefix: str) -> Tuple[str, str, str]:
|
|
return f"{prefix}_notes", f"{prefix}_chunks", f"{prefix}_edges"
|
|
|
|
|
|
def _to_uuid(stable_key: str) -> str:
|
|
"""Stabile UUIDv5 aus einem String-Key (deterministisch)."""
|
|
return str(uuid.uuid5(uuid.NAMESPACE_URL, stable_key))
|
|
|
|
|
|
def points_for_note(
|
|
prefix: str,
|
|
note_payload: dict,
|
|
note_vec: List[float] | None,
|
|
dim: int,
|
|
) -> Tuple[str, List[rest.PointStruct]]:
|
|
"""Notes-Collection: falls kein Note-Embedding -> Nullvektor der Länge dim."""
|
|
notes_col, _, _ = _names(prefix)
|
|
vector = note_vec if note_vec is not None else [0.0] * int(dim)
|
|
raw_note_id = note_payload.get("note_id") or note_payload.get("id") or "missing-note-id"
|
|
point_id = _to_uuid(raw_note_id)
|
|
pt = rest.PointStruct(id=point_id, vector=vector, payload=note_payload)
|
|
return notes_col, [pt]
|
|
|
|
|
|
def points_for_chunks(
|
|
prefix: str,
|
|
chunk_payloads: List[dict],
|
|
vectors: List[List[float]],
|
|
) -> Tuple[str, List[rest.PointStruct]]:
|
|
"""
|
|
Chunks-Collection: erwartet pro Chunk einen Vektor.
|
|
Robustheit:
|
|
- Fehlt 'chunk_id', nutze 'id', sonst baue '${note_id}#${i}' (1-basiert).
|
|
- Schreibe die abgeleitete ID zurück in die Payload (pl['chunk_id']).
|
|
"""
|
|
_, chunks_col, _ = _names(prefix)
|
|
points: List[rest.PointStruct] = []
|
|
for i, (pl, vec) in enumerate(zip(chunk_payloads, vectors), start=1):
|
|
chunk_id = pl.get("chunk_id") or pl.get("id")
|
|
if not chunk_id:
|
|
note_id = pl.get("note_id") or pl.get("parent_note_id") or "missing-note"
|
|
chunk_id = f"{note_id}#{i}"
|
|
pl["chunk_id"] = chunk_id
|
|
point_id = _to_uuid(chunk_id)
|
|
points.append(rest.PointStruct(id=point_id, vector=vec, payload=pl))
|
|
return chunks_col, points
|
|
|
|
|
|
def _normalize_edge_payload(pl: dict) -> dict:
|
|
"""
|
|
Sorgt für kompatible Feldnamen.
|
|
akzeptiert:
|
|
- neu: kind, source_id, target_id, seq?
|
|
- alt: edge_type, src_id, dst_id, order?/index?
|
|
schreibt zurück: kind, source_id, target_id, seq?
|
|
"""
|
|
# bereits neu?
|
|
kind = pl.get("kind") or pl.get("edge_type") or "edge"
|
|
source_id = pl.get("source_id") or pl.get("src_id") or "unknown-src"
|
|
target_id = pl.get("target_id") or pl.get("dst_id") or "unknown-tgt"
|
|
seq = pl.get("seq") or pl.get("order") or pl.get("index")
|
|
|
|
# in Payload zurückschreiben (ohne alte Felder zu entfernen → maximal kompatibel)
|
|
pl.setdefault("kind", kind)
|
|
pl.setdefault("source_id", source_id)
|
|
pl.setdefault("target_id", target_id)
|
|
if seq is not None and "seq" not in pl:
|
|
pl["seq"] = seq
|
|
return pl
|
|
|
|
|
|
def points_for_edges(prefix: str, edge_payloads: List[dict]) -> Tuple[str, List[rest.PointStruct]]:
|
|
"""
|
|
Edges-Collection mit 1D-Dummy-Vektor.
|
|
- Akzeptiert sowohl neues als auch altes Edge-Schema (siehe _normalize_edge_payload).
|
|
- Fehlt 'edge_id', wird sie stabil aus (kind, source_id, target_id, seq) konstruiert.
|
|
"""
|
|
_, _, edges_col = _names(prefix)
|
|
points: List[rest.PointStruct] = []
|
|
for raw in edge_payloads:
|
|
pl = _normalize_edge_payload(raw)
|
|
|
|
edge_id = pl.get("edge_id")
|
|
if not edge_id:
|
|
kind = pl.get("kind", "edge")
|
|
s = pl.get("source_id", "unknown-src")
|
|
t = pl.get("target_id", "unknown-tgt")
|
|
seq = pl.get("seq") or ""
|
|
edge_id = f"{kind}:{s}->{t}#{seq}"
|
|
pl["edge_id"] = edge_id
|
|
|
|
point_id = _to_uuid(edge_id)
|
|
points.append(rest.PointStruct(id=point_id, vector=[0.0], payload=pl))
|
|
return edges_col, points
|
|
|
|
|
|
def upsert_batch(client, collection: str, points: List[rest.PointStruct]) -> None:
|
|
if not points:
|
|
return
|
|
client.upsert(collection_name=collection, points=points, wait=True)
|