File size: 1,760 Bytes
98bde72 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 | """Strict proposal schema for language-model compiler integration."""
from typing import Literal
from pydantic import Field, model_validator
from .schema import Record
from .compiled import Node,Plan
class Obligation(Record):
id: str
endpoint: str
unit: str
context_ids: list[str]
support: Literal['calibrated','proxy']
threshold: float
direction: Literal['ge','le']
scale: float=Field(gt=0)
threshold_source: str
worker: str
class Hypothesis(Record):
id: str
mechanism: str
evidence_ids: list[str]
opposing_evidence_ids: list[str]
counterexample: str
obligation_ids: list[str]
class PlanNode(Record):
id: str
tool: str
revision: str
inputs: list[str]
outputs: list[str]
cost: int=Field(ge=1)
attempts: int=Field(ge=1,le=3)
class PlanProposal(Record):
task_hash: str
evidence_snapshot_hash: str
hypotheses: list[Hypothesis]=Field(max_length=3)
obligations: list[Obligation]
nodes: list[PlanNode]
required: list[str]=Field(min_length=1)
budget: int=Field(ge=0)
seed: int
diagnostics: list[str]
@model_validator(mode='after')
def references(self):
ids={o.id for o in self.obligations}
if len(ids)!=len(self.obligations):raise ValueError('duplicate obligation')
for h in self.hypotheses:
if not h.evidence_ids or not set(h.obligation_ids)<=ids:raise ValueError('incomplete hypothesis')
return self
def execution_plan(self):
if self.diagnostics:raise ValueError('unresolved proposal diagnostics')
return Plan(tuple(Node(n.id,n.tool,n.revision,tuple(n.inputs),tuple(n.outputs),n.cost,n.attempts) for n in self.nodes),tuple(self.required),self.budget,self.seed)
|