WIN-JHFT4D3SIVT\caoxiaozhu
aa0ef0fbea
Add Day 1 agent runtime foundations with task and event schemas, verifier support, capability metadata, graph event tracing, and regression coverage while preserving the direct execution path.
61 lines
2.3 KiB
Python
61 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.agents.schemas.task import AgentTask, TaskResult, VerificationStatus
|
|
from app.agents.state import AgentState
|
|
|
|
|
|
class VerificationVerdict(BaseModel):
|
|
status: VerificationStatus
|
|
summary: str | None = None
|
|
evidence: list[dict[str, Any]] = Field(default_factory=list)
|
|
|
|
|
|
def verify_task_result(
|
|
*,
|
|
task: AgentTask | dict[str, Any] | None = None,
|
|
result: TaskResult | dict[str, Any] | None = None,
|
|
summary: str | None = None,
|
|
evidence: list[dict[str, Any]] | None = None,
|
|
status: VerificationStatus | None = None,
|
|
) -> VerificationVerdict:
|
|
normalized_result = result.model_dump() if isinstance(result, TaskResult) else dict(result or {})
|
|
normalized_task = task.model_dump() if isinstance(task, AgentTask) else dict(task or {})
|
|
normalized_summary = summary or normalized_result.get("summary") or normalized_task.get("result_summary")
|
|
normalized_evidence = list(evidence or normalized_result.get("evidence") or normalized_task.get("evidence") or [])
|
|
|
|
if status is not None:
|
|
return VerificationVerdict(status=status, summary=normalized_summary, evidence=normalized_evidence)
|
|
|
|
if normalized_result.get("status") in {"passed", "failed", "skipped"}:
|
|
inferred_status = normalized_result["status"]
|
|
elif normalized_result.get("success") is True:
|
|
inferred_status = "passed"
|
|
elif normalized_result.get("success") is False:
|
|
inferred_status = "failed"
|
|
elif normalized_summary or normalized_evidence:
|
|
inferred_status = "skipped"
|
|
else:
|
|
inferred_status = "failed"
|
|
normalized_summary = "No verification input available."
|
|
|
|
return VerificationVerdict(
|
|
status=inferred_status,
|
|
summary=normalized_summary,
|
|
evidence=normalized_evidence,
|
|
)
|
|
|
|
|
|
def apply_verification_verdict(state: AgentState, verdict: VerificationVerdict) -> AgentState:
|
|
next_state = dict(state)
|
|
next_state["verification_status"] = verdict.status
|
|
next_state["verification_summary"] = verdict.summary
|
|
next_state["verification_evidence"] = list(verdict.evidence)
|
|
return AgentState(**next_state)
|
|
|
|
|
|
__all__ = ["VerificationVerdict", "apply_verification_verdict", "verify_task_result"]
|