feat(stage10): implement vision integration
This commit is contained in:
NSCT Agent
282
src/nsct/models/vision.py
Normal file
282
src/nsct/models/vision.py
Normal file
@@ -0,0 +1,282 @@
|
||||
"""Pydantic v2 schemas — Vision Evidence Extraction (Stage 10).
|
||||
|
||||
Qwen2.5-VL-3B extrahiert strukturierte Informationen aus Bildern:
|
||||
Diagramme, Screenshots, Infografiken, PDF-Layouts.
|
||||
Provenance-Pflicht: jede visuelle Evidenz ist quellenverknüpft.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from enum import Enum
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Enums
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class VisionCaptureType(str, Enum):
|
||||
"""Klassifikation der visuellen Erfassung."""
|
||||
|
||||
RAW_IMAGE = "raw_image"
|
||||
DIAGRAM = "diagram"
|
||||
CHART = "chart"
|
||||
SCREENSHOT = "screenshot"
|
||||
INFOGRAPHIC = "infographic"
|
||||
PDF_LAYOUT = "pdf_layout"
|
||||
|
||||
|
||||
class VisionConfidence(str, Enum):
|
||||
"""Confidence-Stufe der visuellen Extraktion."""
|
||||
|
||||
HIGH = "high"
|
||||
MEDIUM = "medium"
|
||||
LOW = "low"
|
||||
UNCERTAIN = "uncertain"
|
||||
|
||||
|
||||
class EvidenceLevel(str, Enum):
|
||||
"""Evidenz-Level einer visuellen Evidenz."""
|
||||
|
||||
HIGH = "high"
|
||||
MEDIUM = "medium"
|
||||
LOW = "low"
|
||||
UNCERTAIN = "uncertain"
|
||||
|
||||
|
||||
class VisionEntityCategory(str, Enum):
|
||||
"""Kategorie einer erkannten Entity aus einem Bild."""
|
||||
|
||||
DATE = "date"
|
||||
PERSON = "person"
|
||||
ORGANIZATION = "organization"
|
||||
LOCATION = "location"
|
||||
NUMBER = "number"
|
||||
STATISTIC = "statistic"
|
||||
GRAPH_ELEMENT = "graph_element"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# VisionCaptureSchema — einzelne visuell extrahierte Evidenz
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class VisionCaptureSchema(BaseModel):
|
||||
"""Einzelne visuell extrahierte Evidenz.
|
||||
|
||||
Felder:
|
||||
capture_type: Art der visuellen Erfassung
|
||||
image_data_url: Data-URL oder base64-codiertes Bild
|
||||
extracted_text: Vom Vision-Modell extrahierter Text
|
||||
entities: Erkannte Entities (Personen, Zahlen, etc.)
|
||||
confidence: Confidence der Extraktion
|
||||
evidence_level: Evidenz-Level
|
||||
source_id: Quelle, von der das Bild stammt
|
||||
source_url: URL der Quelle (Provenance)
|
||||
metadata: Zusätzliche Metadaten
|
||||
"""
|
||||
|
||||
capture_type: VisionCaptureType = Field(
|
||||
...,
|
||||
description="Art der visuellen Erfassung (raw_image, diagram, chart, screenshot, infographic, pdf_layout).",
|
||||
)
|
||||
image_data_url: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="Data-URL (data:image/...) oder base64-codiertes Bild.",
|
||||
)
|
||||
extracted_text: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="Vom Vision-Modell extrahierter Textinhalt.",
|
||||
)
|
||||
entities: list[dict[str, Any]] = Field(
|
||||
default_factory=list,
|
||||
description="Erkannte Entities aus dem Bild.",
|
||||
)
|
||||
confidence: float = Field(
|
||||
default=0.5,
|
||||
ge=0.0,
|
||||
le=1.0,
|
||||
description="Confidence-Wert der Extraktion (0-1).",
|
||||
)
|
||||
confidence_label: VisionConfidence = Field(
|
||||
default=VisionConfidence.MEDIUM,
|
||||
description="Confidence-Stufe als Label.",
|
||||
)
|
||||
evidence_level: EvidenceLevel = Field(
|
||||
default=EvidenceLevel.MEDIUM,
|
||||
description="Evidenz-Level der visuellen Evidenz.",
|
||||
)
|
||||
source_id: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="UUID der Quelle (source_id) zur Provenance.",
|
||||
)
|
||||
source_url: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="URL der Quelle (Provenance).",
|
||||
)
|
||||
metadata: dict[str, Any] = Field(
|
||||
default_factory=dict,
|
||||
description="Zusätzliche Metadaten (z.B. model_used, processing_time).",
|
||||
)
|
||||
|
||||
@field_validator("extracted_text")
|
||||
@classmethod
|
||||
def extracted_text_not_empty(cls, v: str) -> str:
|
||||
if not v.strip():
|
||||
raise ValueError("extracted_text darf nicht nur aus Whitespaces bestehen")
|
||||
return v
|
||||
|
||||
@field_validator("image_data_url")
|
||||
@classmethod
|
||||
def image_data_url_not_empty(cls, v: str) -> str:
|
||||
if not v.strip():
|
||||
raise ValueError("image_data_url darf nicht leer sein")
|
||||
return v
|
||||
|
||||
@field_validator("source_url")
|
||||
@classmethod
|
||||
def source_url_not_empty(cls, v: str) -> str:
|
||||
if not v.strip():
|
||||
raise ValueError("source_url darf nicht leer sein")
|
||||
return v
|
||||
|
||||
model_config = {"frozen": True}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# VisionReportSchema — Zusammenfassung aller visuellen Evidenzen
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class VisionReportSchema(BaseModel):
|
||||
"""Zusammenfassung aller visuellen Evidenzen für einen Research-Run.
|
||||
|
||||
Felder:
|
||||
research_run_id: UUID des Research-Runs
|
||||
total_captures: Anzahl der visuellen Evidenzen
|
||||
captures: Liste aller visuellen Evidenzen
|
||||
entity_summary: Zusammenfassung aller erkannten Entities
|
||||
summary_text: Kurze Zusammenfassung des visuellen Contents
|
||||
generation_timestamp: Zeitstempel der Generierung
|
||||
"""
|
||||
|
||||
research_run_id: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="UUID des Research-Runs.",
|
||||
)
|
||||
total_captures: int = Field(
|
||||
...,
|
||||
ge=0,
|
||||
description="Anzahl der visuellen Evidenzen in diesem Report.",
|
||||
)
|
||||
captures: list[VisionCaptureSchema] = Field(
|
||||
default_factory=list,
|
||||
description="Liste aller visuellen Evidenzen.",
|
||||
)
|
||||
entity_summary: dict[str, Any] = Field(
|
||||
default_factory=dict,
|
||||
description="Zusammenfassung aller erkannten Entities.",
|
||||
)
|
||||
summary_text: str = Field(
|
||||
default="",
|
||||
description="Kurze Zusammenfassung des visuellen Contents.",
|
||||
)
|
||||
generation_timestamp: datetime = Field(
|
||||
default_factory=lambda: datetime.now(timezone.utc),
|
||||
description="Zeitstempel der Generierung.",
|
||||
)
|
||||
|
||||
@field_validator("research_run_id")
|
||||
@classmethod
|
||||
def research_run_id_not_empty(cls, v: str) -> str:
|
||||
if not v.strip():
|
||||
raise ValueError("research_run_id darf nicht leer sein")
|
||||
return v
|
||||
|
||||
@field_validator("summary_text")
|
||||
@classmethod
|
||||
def summary_not_political(cls, v: str) -> str:
|
||||
import re
|
||||
|
||||
forbidden = re.compile(
|
||||
r"(sollte\s+(Regierung|Bundesregierung)\s+(handeln|unterstützen)|"
|
||||
r"muss\s+(geändert|eingesetzt|gestürzt))",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
if forbidden.search(v):
|
||||
raise ValueError("VisionReport darf keine politische Empfehlung enthalten")
|
||||
return v
|
||||
|
||||
model_config = {"frozen": True}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# VisionRequestSchema — API-Request für die Vision-Extraktion
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class VisionRequestSchema(BaseModel):
|
||||
"""API-Request zum Extrahieren visueller Evidenzen.
|
||||
|
||||
Felder:
|
||||
research_run_id: UUID des Research-Runs
|
||||
source_id: Quelle, von der das Bild stammt
|
||||
source_url: URL der Quelle (Provenance)
|
||||
image_data: Base64-codiertes Bild oder Data-URL
|
||||
capture_type: Art der visuellen Erfassung
|
||||
prompt: Optionaler Prompt an das Vision-Modell
|
||||
"""
|
||||
|
||||
research_run_id: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="UUID des Research-Runs.",
|
||||
)
|
||||
source_id: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="UUID der Quelle (source_id).",
|
||||
)
|
||||
source_url: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="URL der Quelle (Provenance).",
|
||||
)
|
||||
image_data: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="Base64-codiertes Bild oder Data-URL (data:image/...).",
|
||||
)
|
||||
capture_type: VisionCaptureType = Field(
|
||||
default=VisionCaptureType.RAW_IMAGE,
|
||||
description="Art der visuellen Erfassung.",
|
||||
)
|
||||
prompt: str | None = Field(
|
||||
default=None,
|
||||
description="Optionaler Prompt an das Vision-Modell für die Extraktion.",
|
||||
)
|
||||
|
||||
@field_validator("image_data")
|
||||
@classmethod
|
||||
def image_data_not_empty(cls, v: str) -> str:
|
||||
if not v.strip():
|
||||
raise ValueError("image_data darf nicht leer sein")
|
||||
return v
|
||||
|
||||
@field_validator("source_url")
|
||||
@classmethod
|
||||
def source_url_not_empty(cls, v: str) -> str:
|
||||
if not v.strip():
|
||||
raise ValueError("source_url darf nicht leer sein")
|
||||
return v
|
||||
|
||||
model_config = {"frozen": True}
|
||||
Reference in New Issue
Block a user