mindnet/app/models/dto.py
2025-12-07 16:32:02 +01:00

146 lines
4.5 KiB
Python

"""
app/models/dto.py — Pydantic-Modelle (DTOs) für WP-04 Endpunkte
Zweck:
Laufzeit-Modelle für FastAPI (Requests/Responses), getrennt von JSON-Schemas.
Deckt die Graph-/Retriever-Endpunkte ab.
Enthält Erweiterungen für WP-04b (Explanation Layer).
Kompatibilität:
Python 3.12+, Pydantic 2.x, FastAPI 0.110+
Version:
0.2.0 (Update für WP-04b Explanation Layer)
Stand:
2025-12-07
Bezug:
- schemas/*.json (Speicherschema für Notes/Chunks/Edges)
- WP-04 API-Design (Query- und Graph-Endpunkte)
Nutzung:
from app.models.dto import QueryRequest, QueryResponse, GraphResponse
"""
from __future__ import annotations
from pydantic import BaseModel, Field
from typing import List, Literal, Optional, Dict, Any
EdgeKind = Literal["references", "references_at", "backlink", "next", "prev", "belongs_to", "depends_on", "related_to", "similar_to"]
class NodeDTO(BaseModel):
"""Darstellung eines Knotens (Note oder Chunk) im API-Graph."""
id: str
type: Literal["note", "chunk"]
title: Optional[str] = None
note_id: Optional[str] = None
tags: Optional[List[str]] = None
in_degree: Optional[int] = None
out_degree: Optional[int] = None
score: Optional[float] = None
section_title: Optional[str] = None
section_path: Optional[str] = None
path: Optional[str] = None
class EdgeDTO(BaseModel):
"""Darstellung einer Kante im API-Graph."""
id: str
kind: str # String statt Literal, um flexibel für Custom-Types zu bleiben
source: str
target: str
weight: float
direction: Literal["out", "in", "undirected"] = "out"
class QueryRequest(BaseModel):
"""
Request für /query:
- mode: 'semantic' | 'edge' | 'hybrid'
- query: (optional) Freitext; Embedding wird später angebunden
- query_vector: (optional) direkter Vektor (384d) für Quick-Tests ohne Embedding
- explain: (optional) Fordert detaillierte Erklärungen an (WP-04b)
"""
mode: Literal["semantic", "edge", "hybrid"] = "hybrid"
query: Optional[str] = None
query_vector: Optional[List[float]] = None
top_k: int = 10
expand: Dict = {"depth": 1, "edge_types": ["references", "belongs_to", "prev", "next", "depends_on", "related_to"]}
filters: Optional[Dict] = None
# Flags zur Steuerung der Rückgabe
ret: Dict = {"with_paths": True, "with_notes": True, "with_chunks": True}
# WP-04b: Soll eine Erklärung generiert werden?
explain: bool = False
# --- WP-04b Explanation Models ---
class ScoreBreakdown(BaseModel):
"""
Aufschlüsselung der Score-Komponenten.
Zeigt die gewichteten Beiträge zum Total Score.
"""
semantic_contribution: float = Field(..., description="W_sem * semantic_score * weight")
edge_contribution: float = Field(..., description="W_edge * edge_bonus")
centrality_contribution: float = Field(..., description="W_cent * centrality_bonus")
# Rohwerte für Transparenz
raw_semantic: float
raw_edge_bonus: float
raw_centrality: float
node_weight: float = Field(..., description="Typ-Gewicht (retriever_weight)")
class Reason(BaseModel):
"""
Ein semantischer Grund für das Ranking.
z.B. 'Verlinkt von Projekt X', 'Hohe Textähnlichkeit'.
"""
kind: Literal["semantic", "edge", "type", "centrality"]
message: str
score_impact: Optional[float] = None
details: Optional[Dict[str, Any]] = None
class Explanation(BaseModel):
"""
Container für alle Erklärungsdaten eines Treffers.
"""
breakdown: ScoreBreakdown
reasons: List[Reason]
# Optional: Pfade im Graphen, die zu diesem Treffer geführt haben
related_edges: Optional[List[EdgeDTO]] = None
# --- End Explanation Models ---
class QueryHit(BaseModel):
"""Einzelnes Trefferobjekt für /query."""
node_id: str
note_id: Optional[str]
# Flache Scores (Kompatibilität WP-04a)
semantic_score: float
edge_bonus: float
centrality_bonus: float
total_score: float
paths: Optional[List[List[Dict]]] = None
source: Optional[Dict] = None
# WP-04b: Erklärungsobjekt (nur gefüllt, wenn explain=True)
explanation: Optional[Explanation] = None
class QueryResponse(BaseModel):
"""Antwortstruktur für /query (Liste von Treffern + Telemetrie)."""
results: List[QueryHit]
used_mode: str
latency_ms: int
class GraphResponse(BaseModel):
"""Antwortstruktur für /graph/{note_id} (Nachbarschaft)."""
center_note_id: str
nodes: List[NodeDTO]
edges: List[EdgeDTO]
stats: Dict[str, int]