deerflow-code/offline-backend-20260512/backend/app/report_collaboration/contracts/artifacts.py
2026-09-07 18:24:55 +08:00

141 lines
4.1 KiB
Python

"""Typed artifacts. Nodes complete only after a validated artifact is stored."""
from __future__ import annotations
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
from app.report_collaboration.contracts.enums import ArtifactType, ArtifactValidationStatus, ReviewVerdict
class CollaborationArtifact(BaseModel):
"""User-visible artifact card. ``summary`` never carries the body."""
model_config = ConfigDict(extra="forbid")
id: str
run_id: str
node_id: str
attempt: int
artifact_type: ArtifactType | str
schema_version: int = 1
validation_status: ArtifactValidationStatus
summary: str | None = None
created_at: str
class CitationEntry(BaseModel):
model_config = ConfigDict(extra="forbid")
source_id: str
url: str | None = None
document_id: str | None = None
title: str
published_at: str | None = None
retrieved_at: str | None = None
excerpt: str
credibility: float | None = Field(default=None, ge=0, le=1)
class EvidenceClaim(BaseModel):
model_config = ConfigDict(extra="forbid")
claim_id: str
text: str
kind: Literal["fact", "inference"] = "fact"
source_ids: list[str] = Field(default_factory=list)
excerpt: str | None = None
published_at: str | None = None
retrieved_at: str | None = None
scope: str | None = None
unit: str | None = None
credibility: float | None = Field(default=None, ge=0, le=1)
class EvidenceBundle(BaseModel):
model_config = ConfigDict(extra="forbid")
schema_version: int = 1
claims: list[EvidenceClaim] = Field(default_factory=list)
sources: list[CitationEntry] = Field(default_factory=list)
conflicts: list[str] = Field(default_factory=list)
coverage_gaps: list[str] = Field(default_factory=list)
class VerificationResult(BaseModel):
model_config = ConfigDict(extra="forbid")
schema_version: int = 1
claim_id: str
supporting_source_ids: list[str] = Field(default_factory=list)
opposing_source_ids: list[str] = Field(default_factory=list)
conflicts: list[str] = Field(default_factory=list)
freshness: str | None = None
independent_sources: bool = False
usable_conclusion: str
verdict: Literal["supported", "contested", "unsupported", "stale"] = "supported"
class AngleAnalysis(BaseModel):
model_config = ConfigDict(extra="forbid")
schema_version: int = 1
angle: str
thesis: str
evidence_ids: list[str] = Field(default_factory=list)
facts: list[str] = Field(default_factory=list)
inferences: list[str] = Field(default_factory=list)
assumptions: list[str] = Field(default_factory=list)
limitations: list[str] = Field(default_factory=list)
gaps: list[str] = Field(default_factory=list)
class SynthesisOutline(BaseModel):
model_config = ConfigDict(extra="forbid")
schema_version: int = 1
argument_spine: str
sections: list[dict[str, Any]] = Field(default_factory=list)
conclusions: list[str] = Field(default_factory=list)
evidence_allocation: dict[str, list[str]] = Field(default_factory=dict)
class ReportSectionDraft(BaseModel):
model_config = ConfigDict(extra="forbid")
schema_version: int = 1
section_id: str
title: str
markdown: str
evidence_ids: list[str] = Field(default_factory=list)
claim_citation_map: dict[str, list[str]] = Field(default_factory=dict)
class ReviewIssue(BaseModel):
model_config = ConfigDict(extra="forbid")
severity: str
target: str
requirement: str
evidence: str | None = None
class ReviewDecision(BaseModel):
model_config = ConfigDict(extra="forbid")
schema_version: int = 1
verdict: ReviewVerdict
issues: list[ReviewIssue] = Field(default_factory=list)
class CoverageMatrix(BaseModel):
model_config = ConfigDict(extra="forbid")
schema_version: int = 1
required_angles: list[str] = Field(default_factory=list)
covered_angles: list[str] = Field(default_factory=list)
required_questions: list[str] = Field(default_factory=list)
covered_questions: list[str] = Field(default_factory=list)
gaps: list[str] = Field(default_factory=list)