didi-lot1-ai/ai_platform/modules/didi_brain/brain_api/schemas.py

880 lines
27 KiB
Python

"""Pydantic v2 schemas — the exact contract Didi's backend parses.
These match the shape documented for the existing web-gathering module
1:1. Any field the backend checks MUST exist in the response. Where we don't
have meaningful data, we populate with safe non-null defaults (empty list,
"Global", "en", etc.) rather than leaving a field out or null.
Additive fields that are DidiBrain-specific live under `brain_meta` blocks so
an unknowing backend ignores them while a newer one can consume them.
"""
from __future__ import annotations
from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
# =============================================================================
# Request models
# =============================================================================
class SearchRequest(BaseModel):
"""POST /v1/search — simple list-style retrieval."""
model_config = ConfigDict(extra="allow")
queries: list[str] = Field(
...,
min_length=1,
description="One or more query strings. Multiple are executed in parallel.",
)
max_results: int = Field(20, ge=1, le=200)
language: str | None = Field(None, description="Hint for detected query language, optional.")
class FetchRequest(BaseModel):
"""POST /v1/fetch — retrieve full text for a list of URLs."""
model_config = ConfigDict(extra="allow")
urls: list[str] = Field(..., min_length=1)
include_html: bool = Field(False, description="Return raw HTML along with extracted text.")
class GatherRequest(BaseModel):
"""POST /v1/gather — full claim-to-evidence pipeline."""
model_config = ConfigDict(extra="allow")
claim: str = Field(..., min_length=3)
max_evidence: int = Field(15, ge=1, le=100)
include_full_text: bool = Field(True)
summarize: bool = Field(True)
score_relevance: bool = Field(True)
language_hint: str | None = Field(None)
# When true, run an independent NLI pass to decide whether each piece of
# evidence supports, contradicts, or is neutral toward the input claim.
# Adds ~2-4 seconds to the gather call; disable when latency matters more
# than classification detail.
run_nli: bool = Field(True)
# ----- Verification cache extension (agreed contract with didi-backend) -----
include_verification: bool = Field(
False,
description=(
"If true, brain attempts to look up cached verification for "
"(claim, evidence_urls, tier) and attach it under brain_meta."
),
)
tier: Literal["free", "premium"] | None = Field(
None,
description="Required when include_verification=true — isolates cache.",
)
prompt_hash: str | None = Field(
None,
description=(
"Current backend prompt hash. Brain marks entries stale when the "
"cached entry's prompt_hash differs (returns nothing)."
),
)
framework_version: str | None = Field(
None,
description=(
"Current backend framework config hash (thresholds). "
"Differs → return verification_raw so backend can recompute."
),
)
# ----- Phase B4: recency-aware retrieval (Pilon 3+4) -------------------
volatility_hint: Literal["volatile", "evolving", "stable"] | None = Field(
None,
description=(
"Caller's hint about how fast this claim's truth can change. "
"Drives recency boost in ranking and a hard recency filter for "
"volatile claims. Absent → mild defaults applied (no aggression)."
),
)
recency_window_days: int | None = Field(
None,
ge=1,
le=365,
description=(
"When volatility_hint='volatile' and this is set, hard-drop any "
"evidence older than N days. Default behavior (None): 7 days for "
"volatile, no cut for evolving/stable."
),
)
class VerificationCacheWriteRequest(BaseModel):
"""POST /v1/verification_cache — fire-and-forget push from didi-backend."""
model_config = ConfigDict(extra="allow")
claim: str = Field(..., min_length=1)
evidence_urls: list[str] = Field(..., min_length=1)
tier: Literal["free", "premium"]
model: str | None = None
prompt_hash: str = Field(..., min_length=4)
framework_version: str | None = None
schema_name: str = Field("didi-v1")
verification_processed: dict = Field(..., description="Opaque blob; stored 1:1.")
verification_raw: dict | None = Field(
None,
description=(
"Optional — lets brain hand this back when thresholds change "
"(stale_framework response), so backend can recompute status."
),
)
class VerificationCacheWriteResponse(BaseModel):
model_config = ConfigDict(extra="allow")
cached: bool
claim_hash: str
evidence_hash: str
tier: str
created_at: datetime
updated_at: datetime
expires_at: datetime
# -----------------------------------------------------------------------------
# Analysis Atom — cache for techniques + ai_tampered LLM results
# Contract: didi-backend agent-v3 reads/writes via /v1/analysis_atom/{lookup,POST,PATCH}
# -----------------------------------------------------------------------------
AtomComponent = Literal["techniques", "ai_tampered", "claims"]
AtomTier = Literal["free", "premium"]
AtomCacheTier = Literal["gold", "silver", "bronze"]
AtomStaleness = Literal["fresh", "stale_prompt", "stale_framework", "miss"]
class AnalysisAtomLookupRequest(BaseModel):
"""POST /v1/analysis_atom/lookup — find a cached LLM analysis result.
Lookup is keyed on (content_hash, component, prompt_hash). Tier is NOT
part of the key — read tier-agnostic so free users benefit from premium
cached entries.
"""
model_config = ConfigDict(extra="allow")
content_hash: str = Field(..., min_length=8, description="sha256 of normalized content")
component: AtomComponent
tier: AtomTier = Field(..., description="Caller's current tier (informational, not used for lookup key)")
prompt_hash: str = Field(..., min_length=4)
framework_version: str | None = None
class AnalysisAtomData(BaseModel):
model_config = ConfigDict(extra="allow")
atom_id: int
content_hash: str
component: AtomComponent
tier: AtomTier
prompt_hash: str
framework_version: str | None
model_used: str | None
cache_tier: AtomCacheTier
human_validated: bool
result_processed: dict
validator_user_id: str | None
validated_at: datetime | None
hit_count: int
created_at: datetime
updated_at: datetime
expires_at: datetime | None
class AnalysisAtomLookupResponse(BaseModel):
model_config = ConfigDict(extra="allow")
hit: bool
atom: AnalysisAtomData | None = None
staleness: AtomStaleness | None = None
match_type: Literal["exact", "semantic"] | None = None
class AnalysisAtomWriteRequest(BaseModel):
"""POST /v1/analysis_atom — fire-and-forget write from backend after LLM run.
Brain rejects writes for tier='free' (premium-only ingest, by design).
cache_tier is computed from llm_confidence (>= threshold → silver, else bronze).
"""
model_config = ConfigDict(extra="allow")
content_hash: str = Field(..., min_length=8)
content_preview: str | None = Field(None, max_length=500)
component: AtomComponent
tier: AtomTier
prompt_hash: str = Field(..., min_length=4)
framework_version: str | None = None
model_used: str | None = None
result_processed: dict = Field(..., description="Canonical mapped result, stored 1:1")
result_raw: dict | None = None
llm_confidence: float | None = Field(None, ge=0, le=100)
# If None (default), cache_tier is decided server-side from llm_confidence.
# Set explicitly only if caller wants to force a specific tier.
cache_tier: AtomCacheTier | None = Field(None, description="Optional override; None = decide from confidence")
class AnalysisAtomWriteResponse(BaseModel):
model_config = ConfigDict(extra="allow")
cached: bool
atom_id: int | None = None
cache_tier: AtomCacheTier | None = None
skipped_reason: str | None = None # e.g. "tier=free" or "confidence_below_threshold"
class AnalysisAtomPatchRequest(BaseModel):
"""PATCH /v1/analysis_atom/{atom_id} — promote to gold after human review."""
model_config = ConfigDict(extra="allow")
human_validated: bool = True
human_corrections: dict | None = None
validator_user_id: str | None = None
result_processed: dict | None = Field(None, description="Updated result after corrections applied")
cache_tier: AtomCacheTier = "gold"
class AnalysisAtomStatsResponse(BaseModel):
model_config = ConfigDict(extra="allow")
total_atoms: int
by_tier: dict # {gold: N, silver: N, bronze: N}
by_component: dict # {techniques: N, ai_tampered: N, claims: N}
hit_rate_24h: float | None = None
writes_24h: int
# -----------------------------------------------------------------------------
# Admin browser models (consumed by AI platform dashboard reskin)
# -----------------------------------------------------------------------------
class AnalysisAtomListItem(BaseModel):
"""Lightweight row for atom list (no result_raw / result_processed)."""
atom_id: int
content_hash: str
content_preview: str | None = None
component: str
tier: str
cache_tier: str
prompt_hash: str
framework_version: str | None = None
model_used: str | None = None
llm_confidence: float | None = None
human_validated: bool
hit_count: int
last_hit_at: datetime | None = None
created_at: datetime
updated_at: datetime
expires_at: datetime | None = None
class AnalysisAtomListResponse(BaseModel):
items: list[AnalysisAtomListItem]
total: int
page: int
page_size: int
class AnalysisAtomDetailResponse(AnalysisAtomListItem):
"""Full atom row including result payload + corrections."""
result_processed: dict
result_raw: dict | None = None
human_corrections: dict | None = None
validator_user_id: str | None = None
validated_at: datetime | None = None
class VerificationCacheListItem(BaseModel):
claim_hash: str
tier: str
model: str | None = None
prompt_hash: str
framework_version: str | None = None
schema_name: str
evidence_url_count: int
status: str | None = None
volatility: str | None = None
topic_codes: list[str] = []
created_at: datetime
updated_at: datetime
expires_at: datetime
class VerificationCacheListResponse(BaseModel):
items: list[VerificationCacheListItem]
total: int
page: int
page_size: int
class VerificationCacheDetailResponse(VerificationCacheListItem):
evidence_urls: list[str]
verification_processed: dict
verification_raw: dict | None = None
class TaxonomyInfoResponse(BaseModel):
total_tags: int
namespaces: list[str]
by_namespace: dict
class TaxonomyReloadResponse(BaseModel):
ok: bool
before: int | None = None
after: int | None = None
fetched: int | None = None
error: str | None = None
class AnalysisAtomStatsExtendedResponse(BaseModel):
model_config = ConfigDict(extra="allow")
total_atoms: int
by_tier: dict
by_component: dict
hit_rate_24h: float | None = None
hits_24h_gold: int
hits_24h_silver: int
writes_24h: int
gold_promotions_24h: int
class GenericOkResponse(BaseModel):
ok: bool
message: str | None = None
class CacheInvalidateRequest(BaseModel):
"""POST /v1/cache/invalidate — Pilon 8 mass invalidation.
At least one filter field must be set (topic_codes, entity_canonicals,
claim_pattern, or since) — empty filter is rejected to avoid accidental
"flush everything". By default gold atoms are spared; pass
``invalidate_gold=true`` to flush them too (only do this in moderator-
initiated flows).
Use ``dry_run=true`` first to count matches without modifying anything.
"""
model_config = ConfigDict(extra="forbid")
topic_codes: list[str] | None = Field(
default=None,
description="Match rows whose topic_codes overlap with this set.",
)
entity_canonicals: list[str] | None = Field(
default=None,
description=(
"Pre-normalized canonical forms ('subject predicate object' "
"lowercased). Caller computes via fact_status.canonicalize_triple."
),
)
claim_pattern: str | None = Field(
default=None,
max_length=200,
description="ILIKE pattern matched against content_preview.",
)
since: str | None = Field(
default=None,
description="ISO datetime — match rows updated at or after this time.",
)
invalidate_gold: bool = Field(
default=False,
description="If true, also expire gold (human-validated) atoms.",
)
dry_run: bool = Field(
default=False,
description="Count matches without modifying anything.",
)
actor: str | None = Field(
default=None,
max_length=120,
description="Audit-log label (e.g., 'admin:foo@bar', 'breaking_watcher').",
)
reason: str | None = Field(
default=None,
max_length=500,
description="Optional human-readable note for audit trail.",
)
class CacheInvalidateResponse(BaseModel):
"""Output of POST /v1/cache/invalidate."""
model_config = ConfigDict(extra="allow")
invalidated_atoms: int
invalidated_vcache: int
dry_run: bool
filters_applied: dict
executed_at: datetime
# =============================================================================
# Fact Status admin schemas (Phase D2)
# =============================================================================
class FactStatusItem(BaseModel):
"""One brain_fact_status row, flattened for admin browser."""
model_config = ConfigDict(extra="allow")
fact_id: int
subject: str
predicate: str
object: str
canonical_form: str
canonical_form_hash: str
current_truth: bool | None = None
current_version_id: int | None = None
current_confidence: float | None = None
last_verified_at: datetime | None = None
last_evidence_urls: list[str] = Field(default_factory=list)
volatility: str | None = None
topic_codes: list[str] = Field(default_factory=list)
next_check_at: datetime
check_interval_hours: int
moderator_locked: bool
moderator_user_id: str | None = None
moderator_notes: str | None = None
created_at: datetime
updated_at: datetime
class FactStatusListResponse(BaseModel):
items: list[FactStatusItem]
total: int
page: int
page_size: int
class FactStatusVersionItem(BaseModel):
"""One brain_fact_version row."""
model_config = ConfigDict(extra="allow")
version_id: int
fact_id: int
truth_value: bool
confidence: float | None = None
valid_from: datetime
valid_to: datetime | None = None
source_atom_ids: list[str] = Field(default_factory=list)
evidence_urls: list[str] = Field(default_factory=list)
llm_reasoning: str | None = None
created_by: str
moderator_user_id: str | None = None
notes: str | None = None
created_at: datetime
class FactStatusVersionsResponse(BaseModel):
versions: list[FactStatusVersionItem]
fact_id: int
total: int
class FactStatusPatchRequest(BaseModel):
"""PATCH /v1/fact_status/{fact_id} — moderator override.
Three orthogonal operations, all optional:
- ``set_truth``: assert TRUE/FALSE as the moderator's verdict
- ``lock``: prevent the auditor from auto-changing this fact
- ``unlock``: re-enable auditor updates
"""
model_config = ConfigDict(extra="forbid")
set_truth: bool | None = Field(
default=None,
description="Set current_truth (TRUE/FALSE). Omit to leave unchanged.",
)
confidence: float | None = Field(
default=None,
ge=0.0,
le=100.0,
description="Moderator confidence in this assertion, 0-100.",
)
evidence_urls: list[str] = Field(default_factory=list)
notes: str | None = Field(default=None, max_length=2000)
lock: bool | None = Field(
default=None,
description=(
"true → moderator_locked=true (auditor must skip). "
"false → unlock. None → leave as-is."
),
)
moderator_user_id: str = Field(
...,
min_length=1,
max_length=120,
description="Required — keycloak_id of the moderator making the change.",
)
class AuditLogItem(BaseModel):
"""One brain_audit_log row."""
model_config = ConfigDict(extra="allow")
log_id: int
action: str
target_table: str
target_id: str
actor: str | None = None
payload: dict
created_at: datetime
class AuditLogResponse(BaseModel):
items: list[AuditLogItem]
total: int
page: int
page_size: int
class CanonicalizeRequest(BaseModel):
"""POST /v1/canonicalize — Pilon 7 temporal disambiguation.
Caller (typically agent-v3) sends the raw user claim plus an optional
explicit current_date (ISO). Brain returns the rewritten claim with
relative time markers and ambiguous entities anchored. The caller hashes
the canonical form for cache lookups.
"""
model_config = ConfigDict(extra="forbid")
claim: str = Field(..., min_length=1, max_length=2000)
current_date: str | None = Field(
default=None,
description=(
"ISO date (YYYY-MM-DD) used as 'now' for relative-marker "
"resolution. Defaults to UTC today."
),
)
class CanonicalizeResponse(BaseModel):
"""Output of POST /v1/canonicalize.
On LLM failure the canonical equals the original and ``error`` is set.
Caller can still proceed (no regression).
"""
model_config = ConfigDict(extra="allow")
canonical: str
original: str
changed: bool
anchors_added: list[str] = Field(default_factory=list)
reasoning: str = ""
error: str | None = None
class ImageSearchRequest(BaseModel):
"""POST /v1/image-search — stub, returns empty list."""
model_config = ConfigDict(extra="allow")
queries: list[str] = Field(..., min_length=1)
max_results: int = Field(20, ge=1, le=200)
class IngestRequest(BaseModel):
"""POST /v1/ingest — populate brain from Didi's web-gathering results.
Body shape is intentionally liberal: we accept either a full GatherResponse
(as emitted by the web module) or a thinned envelope with just evidence[].
"""
model_config = ConfigDict(extra="allow")
claim: str | None = Field(None, description="Original query that produced this evidence.")
evidence: list[EvidenceItem] = Field(default_factory=list)
default_tags: list[str] = Field(
default_factory=list,
description="Canonical tag paths to apply to every ingested atom.",
)
run_extraction: bool = Field(
True,
description="If true, queue claim extraction on the newly ingested documents.",
)
# =============================================================================
# Shared sub-schemas
# =============================================================================
class Provenance(BaseModel):
"""Where this evidence came from and how it was produced."""
model_config = ConfigDict(extra="allow")
extraction_method: str = Field("http", description="http | browse | vision | brain")
fallback_chain: list[str] = Field(default_factory=list)
# Additive DidiBrain-specific metadata — safe to ignore if unknown.
brain_meta: "BrainEvidenceMeta | None" = None
class BrainEvidenceMeta(BaseModel):
"""Additive fields specific to DidiBrain that a consumer MAY use."""
model_config = ConfigDict(extra="allow")
parent_atom_id: str
matching_claim_atom_ids: list[str] = Field(default_factory=list)
best_claim_text: str = ""
best_claim_stance_in_source: str = "NEUTRAL"
best_claim_hash: str = ""
claim_count: int = 0
reranker_score: float = 0.0
embedding_similarity: float = 0.0
# NLI stance of the evidence AGAINST the user's query claim. Populated
# when GatherRequest.run_nli is true (default). Unlike stance_in_source,
# this is the direction the backend actually needs for disinfo verdicts.
stance_vs_query: str = "UNKNOWN" # SUPPORTS / CONTRADICTS / NEUTRAL / UNKNOWN
nli_confidence: float = 0.0 # 0..1
nli_error: str | None = None # populated on timeout / bad response
class Entities(BaseModel):
model_config = ConfigDict(extra="allow")
persons: list[str] = Field(default_factory=list)
institutions: list[str] = Field(default_factory=list)
locations: list[str] = Field(default_factory=list)
class SearchContext(BaseModel):
model_config = ConfigDict(extra="allow")
primary_country: str = "Global"
secondary_countries: list[str] = Field(default_factory=list)
entities: Entities = Field(default_factory=Entities)
detected_language: str = "en"
search_queries: list[str] = Field(default_factory=list)
class SearchResultItem(BaseModel):
model_config = ConfigDict(extra="allow")
query: str
url: str
title: str
snippet: str = ""
rank: int
site: str = ""
published_at: datetime | None = None
class StageRecord(BaseModel):
model_config = ConfigDict(extra="allow")
stage: str
success: bool
items_processed: int = 0
items_failed: int = 0
duration_ms: float = 0.0
error: str | None = None
class EvidenceStats(BaseModel):
model_config = ConfigDict(extra="allow")
input_items: int = 0
after_dedup: int = 0
output_items: int = 0
duplicates_removed: int = 0
tokens_used: int = 0
class EvidenceItem(BaseModel):
"""One piece of evidence — always at the DOCUMENT level, not chunk level."""
model_config = ConfigDict(extra="allow")
url: str
canonical_url: str | None = None
title: str
publisher: str = ""
published_at: datetime | None = None
retrieved_at: datetime
snippet: str | None = None
summary: str | None = None
full_text: str | None = None
full_text_hash: str = ""
provenance: Provenance = Field(default_factory=Provenance)
relevance_score: float = 0.0
credibility_score: float = 0.5
# =============================================================================
# Response models
# =============================================================================
class FailedUrl(BaseModel):
model_config = ConfigDict(extra="allow")
url: str
error: str
class FetchedPage(BaseModel):
model_config = ConfigDict(extra="allow")
url: str
canonical_url: str | None = None
title: str = ""
text: str = ""
text_hash: str = ""
html: str | None = None
extraction_method: str = "brain"
fallback_chain: list[str] = Field(default_factory=list)
published_at: datetime | None = None
retrieved_at: datetime
extraction_time_ms: float = 0.0
warnings: list[str] = Field(default_factory=list)
needs_fallback: bool = False
status_code: int = 200
content_type: str = "text/markdown"
class BrainMeta(BaseModel):
"""Top-level meta about the brain response — additive, safe to ignore."""
model_config = ConfigDict(extra="allow")
cache_status: Literal["HIT", "PARTIAL", "MISS"] = "MISS"
api_version: str = "v1"
implementation: str = "didibrain"
evidence_sources: int = 0
total_claim_atoms_matched: int = 0
# ---- Verification cache (populated only when request included the flag) ----
verification: dict | None = Field(
None,
description=(
"Cached verification payload when (claim, tier) is a HIT. Shape "
"is opaque — returned 1:1 as stored by backend."
),
)
verification_staleness: (
Literal[
"fresh",
"stale_framework",
"stale_prompt",
"stale_evidence", # Pilon 11: bound facts have flipped
"miss",
]
| None
) = None
verification_model: str | None = None
verification_tier: str | None = None
verification_prompt_hash: str | None = None
verification_framework_version: str | None = None
verification_cached_at: datetime | None = None
verification_expires_at: datetime | None = None
# URLs the cache was written for — surfaced so backend can detect corpus
# drift and decide whether the cached verification still applies to the
# current evidence set (e.g. compute URL overlap %).
verification_evidence_urls: list[str] | None = None
verification_evidence_hash: str | None = None
# Pilon 11: when staleness=stale_evidence, lists each entity binding whose
# current_truth in brain_fact_status contradicts what the cache assumed.
# Each item: {subject, predicate, object, canonical_form, cached_assumes,
# current_truth}.
verification_facts_invalidated: list[dict] | None = None
class SearchResponse(BaseModel):
model_config = ConfigDict(extra="allow")
request_id: str
results: list[SearchResultItem] = Field(default_factory=list)
total_results: int = 0
execution_time_ms: float = 0.0
queries_processed: int = 0
brain_meta: BrainMeta | None = None
class FetchResponse(BaseModel):
model_config = ConfigDict(extra="allow")
request_id: str
pages: list[FetchedPage] = Field(default_factory=list)
total_fetched: int = 0
total_failed: int = 0
execution_time_ms: float = 0.0
failed_urls: list[FailedUrl] = Field(default_factory=list)
brain_meta: BrainMeta | None = None
class GatherResponse(BaseModel):
model_config = ConfigDict(extra="allow")
request_id: str
claim: str
evidence: list[EvidenceItem] = Field(default_factory=list)
evidence_stats: EvidenceStats = Field(default_factory=EvidenceStats)
search_context: SearchContext = Field(default_factory=SearchContext)
search_results: list[SearchResultItem] = Field(default_factory=list)
stages: list[StageRecord] = Field(default_factory=list)
total_urls_found: int = 0
total_pages_fetched: int = 0
total_evidence_items: int = 0
execution_time_ms: float = 0.0
brain_meta: BrainMeta | None = None
class ImageSearchResponse(BaseModel):
model_config = ConfigDict(extra="allow")
request_id: str
results: list[Any] = Field(default_factory=list)
total_results: int = 0
execution_time_ms: float = 0.0
queries_processed: int = 0
brain_meta: BrainMeta | None = None
class IngestResponse(BaseModel):
model_config = ConfigDict(extra="allow")
request_id: str
accepted: int = 0
skipped_duplicate: int = 0
errors: int = 0
created_atom_ids: list[str] = Field(default_factory=list)
extraction_queued: bool = False
execution_time_ms: float = 0.0
warnings: list[str] = Field(default_factory=list)
# Rebuild forward-ref
Provenance.model_rebuild()