147 lines
5.1 KiB
Python
147 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from agents.models import Observation, VerificationVerdict
|
|
from runtime.executor import StepResult
|
|
from runtime.planner import PlannedStep
|
|
from semantic.llm_client import AnthropicSemanticClient, EnrichmentUnavailable
|
|
|
|
VERIFIER_SYSTEM_PROMPT = """You are a verification agent. Your job is to determine whether a planned step achieved its intended effect.
|
|
|
|
You will receive:
|
|
1. A pre-step observation of the device state
|
|
2. A post-step observation of the device state
|
|
3. The planned step (action, description, expected outcome)
|
|
4. The step result (whether the tool call succeeded mechanically)
|
|
|
|
Analyze whether the intended effect of the step actually occurred by comparing the pre and post observations against the step's stated intent.
|
|
|
|
Respond with JSON: {"result": "achieved" or "not_achieved", "reasoning": "explanation"}"""
|
|
|
|
VERIFIER_JSON_SCHEMA: dict[str, Any] = {
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"required": ["result", "reasoning"],
|
|
"properties": {
|
|
"result": {"type": "string", "enum": ["achieved", "not_achieved"]},
|
|
"reasoning": {"type": "string", "minLength": 1},
|
|
},
|
|
}
|
|
|
|
|
|
class Verifier:
|
|
def __init__(self, *, client: AnthropicSemanticClient | None = None) -> None:
|
|
self._client = client
|
|
|
|
def verify(
|
|
self,
|
|
*,
|
|
pre_observation: Observation,
|
|
post_observation: Observation,
|
|
planned_step: PlannedStep,
|
|
step_result: StepResult,
|
|
) -> VerificationVerdict:
|
|
if not step_result.success:
|
|
return VerificationVerdict(
|
|
result="not_achieved",
|
|
reasoning=f"Step failed mechanically: {step_result.error}",
|
|
)
|
|
|
|
if self._client is None:
|
|
return self._fallback_verify(
|
|
pre_observation=pre_observation,
|
|
post_observation=post_observation,
|
|
planned_step=planned_step,
|
|
step_result=step_result,
|
|
)
|
|
|
|
try:
|
|
return self._llm_verify(
|
|
pre_observation=pre_observation,
|
|
post_observation=post_observation,
|
|
planned_step=planned_step,
|
|
step_result=step_result,
|
|
)
|
|
except EnrichmentUnavailable:
|
|
return self._fallback_verify(
|
|
pre_observation=pre_observation,
|
|
post_observation=post_observation,
|
|
planned_step=planned_step,
|
|
step_result=step_result,
|
|
)
|
|
|
|
def _llm_verify(
|
|
self,
|
|
*,
|
|
pre_observation: Observation,
|
|
post_observation: Observation,
|
|
planned_step: PlannedStep,
|
|
step_result: StepResult,
|
|
) -> VerificationVerdict:
|
|
user_prompt = json.dumps(
|
|
{
|
|
"pre_observation": pre_observation.to_dict(),
|
|
"post_observation": post_observation.to_dict(),
|
|
"planned_step": {
|
|
"action": planned_step.action,
|
|
"description": planned_step.description,
|
|
"args": planned_step.args,
|
|
"expected_text": planned_step.expected_text,
|
|
},
|
|
"step_result": step_result.to_dict(),
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
response = self._client._create_message( # type: ignore[union-attr]
|
|
{"prompt": user_prompt, "system": VERIFIER_SYSTEM_PROMPT},
|
|
timeout=10.0,
|
|
)
|
|
payload = _extract_verdict(response)
|
|
return VerificationVerdict.from_dict(payload)
|
|
|
|
def _fallback_verify(
|
|
self,
|
|
*,
|
|
pre_observation: Observation,
|
|
post_observation: Observation,
|
|
planned_step: PlannedStep,
|
|
step_result: StepResult,
|
|
) -> VerificationVerdict:
|
|
if pre_observation.scene_summary == post_observation.scene_summary:
|
|
return VerificationVerdict(
|
|
result="not_achieved",
|
|
reasoning="Scene unchanged after step execution (fallback heuristic).",
|
|
)
|
|
return VerificationVerdict(
|
|
result="achieved",
|
|
reasoning="Scene changed after step execution (fallback heuristic).",
|
|
)
|
|
|
|
|
|
def _extract_verdict(response: Any) -> dict[str, Any]:
|
|
if isinstance(response, dict) and "result" in response:
|
|
return response
|
|
content = getattr(response, "content", None)
|
|
if isinstance(content, list):
|
|
for block in content:
|
|
text = getattr(block, "text", None)
|
|
if isinstance(text, str):
|
|
try:
|
|
decoded = json.loads(text)
|
|
if isinstance(decoded, dict) and "result" in decoded:
|
|
return decoded
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if isinstance(block, dict) and "result" in block:
|
|
return block
|
|
if isinstance(content, str):
|
|
try:
|
|
decoded = json.loads(content)
|
|
if isinstance(decoded, dict) and "result" in decoded:
|
|
return decoded
|
|
except json.JSONDecodeError:
|
|
pass
|
|
raise ValueError("could not extract verification verdict from LLM response")
|