125 lines
3.7 KiB
Python
125 lines
3.7 KiB
Python
"""Typed analysis packet contracts for analysis-style replies."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Annotated, Any, Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
|
|
|
|
|
|
AnalysisFocusStance = Literal[
|
|
"better_fit",
|
|
"not_fit",
|
|
"anchor",
|
|
"blocked",
|
|
"dependent",
|
|
"relevant",
|
|
]
|
|
|
|
|
|
class AnalysisFocusObject(BaseModel):
|
|
model_config = ConfigDict(extra="ignore")
|
|
|
|
name: str
|
|
kind: str = "unknown"
|
|
stance: AnalysisFocusStance = "relevant"
|
|
reason: str
|
|
|
|
|
|
class AnalysisPacketBase(BaseModel):
|
|
model_config = ConfigDict(extra="ignore")
|
|
|
|
packet_version: str = "v1"
|
|
packet_type: str
|
|
conclusion: str
|
|
focus_objects: list[AnalysisFocusObject] = Field(default_factory=list, max_length=2)
|
|
next_step: str | None = None
|
|
|
|
|
|
class ProjectFitAnalysisPacket(AnalysisPacketBase):
|
|
packet_type: Literal["project_fit_analysis"] = "project_fit_analysis"
|
|
fit_verdict: Literal["fit_now", "fit_later", "not_fit_yet", "uncertain"]
|
|
current_subject: str | None = None
|
|
candidate_position: str | None = None
|
|
rejected_position: str | None = None
|
|
preferred_layer: str | None = None
|
|
rejected_layer: str | None = None
|
|
reason_chain: list[str] = Field(default_factory=list, max_length=3)
|
|
missing_mapping: str
|
|
|
|
|
|
class ExternalImpactAnalysisPacket(AnalysisPacketBase):
|
|
packet_type: Literal["external_impact_analysis"] = "external_impact_analysis"
|
|
retrieval_status: Literal["evidence_found", "verifier_blocked", "no_relevant_results", "retrieval_failed"]
|
|
impact_status: Literal["direct_match", "indirect_only", "no_direct_impact"]
|
|
current_subject: str | None = None
|
|
reason_chain: list[str] = Field(default_factory=list, max_length=3)
|
|
anchor_objects: list[str] = Field(default_factory=list)
|
|
blocked_reason_or_gap: str
|
|
|
|
|
|
class InternalExplainAnalysisPacket(AnalysisPacketBase):
|
|
packet_type: Literal["internal_explain_analysis"] = "internal_explain_analysis"
|
|
answer_contract: str | None = None
|
|
current_subject: str | None = None
|
|
reason_chain: list[str] = Field(default_factory=list, max_length=3)
|
|
resolved_target: str | None = None
|
|
evidence_gap_or_uncertainty: str | None = None
|
|
|
|
|
|
class FollowupExplainAnalysisPacket(AnalysisPacketBase):
|
|
packet_type: Literal["followup_explain_analysis"] = "followup_explain_analysis"
|
|
anchor_kind: str | None = None
|
|
continuation_kind: str | None = None
|
|
clarification_state: str | None = None
|
|
missing_mapping: str | None = None
|
|
|
|
|
|
AnalysisPacket = Annotated[
|
|
ProjectFitAnalysisPacket
|
|
| ExternalImpactAnalysisPacket
|
|
| InternalExplainAnalysisPacket
|
|
| FollowupExplainAnalysisPacket,
|
|
Field(discriminator="packet_type"),
|
|
]
|
|
|
|
_ANALYSIS_PACKET_ADAPTER = TypeAdapter(AnalysisPacket)
|
|
|
|
|
|
def coerce_analysis_packet(value: Any) -> AnalysisPacket | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(
|
|
value,
|
|
(
|
|
ProjectFitAnalysisPacket,
|
|
ExternalImpactAnalysisPacket,
|
|
InternalExplainAnalysisPacket,
|
|
FollowupExplainAnalysisPacket,
|
|
),
|
|
):
|
|
return value
|
|
if isinstance(value, dict):
|
|
try:
|
|
return _ANALYSIS_PACKET_ADAPTER.validate_python(value)
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def normalize_analysis_packet_payload(value: Any) -> dict[str, Any]:
|
|
packet = coerce_analysis_packet(value)
|
|
return packet.model_dump() if packet is not None else {}
|
|
|
|
|
|
__all__ = [
|
|
"AnalysisFocusObject",
|
|
"AnalysisPacket",
|
|
"ExternalImpactAnalysisPacket",
|
|
"FollowupExplainAnalysisPacket",
|
|
"InternalExplainAnalysisPacket",
|
|
"ProjectFitAnalysisPacket",
|
|
"coerce_analysis_packet",
|
|
"normalize_analysis_packet_payload",
|
|
]
|