File size: 6,825 Bytes
20c251e | 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 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | from __future__ import annotations
from dovla_cil.data.schema import ActionChunk, FailureInfo, RewardInfo, StructuredEffect
from dovla_cil.tasks.library import ToyTaskLibrary
from dovla_cil.vlm.annotation import VLMFailureAnnotator
from dovla_cil.vlm.client import VLMClient
def test_mock_vlm_annotation_works(monkeypatch, tmp_path) -> None:
monkeypatch.setenv("OPENCLAUDE_MOCK", "1")
task = ToyTaskLibrary().get_by_id("toy_pick_object_among_distractors")
local_failure = _failure("wrong_target")
annotator = VLMFailureAnnotator(
client=VLMClient(api_key="test-secret", model="mock-model"),
cache_path=tmp_path / "annotations.json",
)
annotated = annotator.annotate_failure(
task=task,
instruction=task.instruction,
action=_action(candidate_type="wrong_target", target="blue_mug"),
effect=_effect(),
reward=_reward(success=False),
local_failure=local_failure,
)
assert annotated.type == "wrong_target"
assert "Mock VLM explanation" in (annotated.language_explanation or "")
metadata = annotated.metadata["semantic_annotation"]["vlm_annotation"]
assert metadata["source"] == "vlm"
assert metadata["suggested_failure_type"] == "wrong_target"
assert "test-secret" not in str(annotated.to_dict())
def test_annotation_cache_hit_avoids_client_call(tmp_path) -> None:
task = ToyTaskLibrary().get_by_id("toy_lift_can")
client = CountingAnnotationClient()
annotator = VLMFailureAnnotator(client=client, cache_path=tmp_path / "cache.json")
kwargs = {
"task": task,
"instruction": task.instruction,
"action": _action(candidate_type="no_motion", target="can"),
"effect": _effect(),
"reward": _reward(success=False),
"local_failure": _failure("no_motion"),
}
first = annotator.annotate_failure(**kwargs)
second = annotator.annotate_failure(**kwargs)
assert client.calls == 1
assert first.language_explanation == second.language_explanation
assert second.metadata["semantic_annotation"]["vlm_annotation"]["cache_hit"] is True
def test_invalid_vlm_json_falls_back_to_local_explanation(tmp_path) -> None:
task = ToyTaskLibrary().get_by_id("toy_lift_can")
local_failure = _failure("no_motion")
annotator = VLMFailureAnnotator(client=InvalidAnnotationClient(), cache_path=tmp_path / "cache.json")
annotated = annotator.annotate_failure(
task=task,
instruction=task.instruction,
action=_action(candidate_type="noop", target="can"),
effect=_effect(),
reward=_reward(success=False),
local_failure=local_failure,
)
assert annotated.type == local_failure.type
assert annotated.language_explanation == local_failure.language_explanation
metadata = annotated.metadata["semantic_annotation"]["vlm_annotation"]
assert metadata["source"] == "local_fallback"
assert metadata["fallback_error"]
def test_vlm_annotation_cannot_change_reward_or_local_failure_type(tmp_path) -> None:
task = ToyTaskLibrary().get_by_id("toy_pick_object_among_distractors")
reward = _reward(success=False)
local_failure = _failure("wrong_target")
annotator = VLMFailureAnnotator(client=OverridingAnnotationClient(), cache_path=tmp_path / "cache.json")
annotated = annotator.annotate_failure(
task=task,
instruction=task.instruction,
action=_action(candidate_type="wrong_target", target="blue_mug"),
effect=_effect(),
reward=reward,
local_failure=local_failure,
)
assert reward.progress == 0.25
assert reward.success is False
assert reward.terminal_success is False
assert annotated.type == "wrong_target"
assert annotated.metadata["semantic_annotation"]["suggested_failure_type"] == "success"
class CountingAnnotationClient:
def __init__(self) -> None:
self.calls = 0
def chat_json(self, system: str, user: str, schema_hint=None) -> dict:
del system, user, schema_hint
self.calls += 1
return {
"failure_type": "no_motion",
"explanation": "The action produced no task-relevant motion.",
"avoidance_hint": "Choose an action that contacts the can.",
"confidence": 0.9,
}
class InvalidAnnotationClient:
def chat_json(self, system: str, user: str, schema_hint=None) -> dict:
del system, user, schema_hint
return {"failure_type": "no_motion", "confidence": "high"}
class OverridingAnnotationClient:
def chat_json(self, system: str, user: str, schema_hint=None) -> dict:
del system, user, schema_hint
return {
"failure_type": "success",
"explanation": "The VLM incorrectly claims this succeeded.",
"avoidance_hint": "No hint.",
"confidence": 1.0,
}
def _action(*, candidate_type: str, target: str) -> ActionChunk:
return ActionChunk(
representation="semantic",
values=[{"command": "grasp", "object": target}],
skill_type="grasp",
metadata={
"candidate_type": candidate_type,
"intended_target": target,
"intended_relation": "grasped",
"difficulty": 0.3,
},
)
def _effect() -> StructuredEffect:
before = {
"objects": {
"red_mug": {"position": [0.0, 0.0, 0.03], "grasped": False, "lifted": False},
"blue_mug": {"position": [0.4, 0.0, 0.03], "grasped": False, "lifted": False},
},
"robot": {"eef_position": [0.0, 0.0, 0.2], "gripper": "open", "held_object": None},
}
after = {
"objects": {
"red_mug": {"position": [0.0, 0.0, 0.03], "grasped": False, "lifted": False},
"blue_mug": {"position": [0.6, 0.0, 0.03], "grasped": False, "lifted": False},
},
"robot": {"eef_position": [0.4, 0.0, 0.2], "gripper": "closed", "held_object": None},
}
return StructuredEffect(
object_pose_delta={"blue_mug": [0.2, 0.0, 0.0]},
contact_events=[{"type": "touch", "object": "blue_mug"}],
relation_before={"grasped(red_mug)": False},
relation_after={"grasped(red_mug)": False},
grasp_success=False,
moved_objects=["blue_mug"],
symbolic_before=before,
symbolic_after=after,
)
def _reward(*, success: bool) -> RewardInfo:
return RewardInfo(
progress=1.0 if success else 0.25,
success=success,
terminal_success=success,
dense_components={"partial": 0.25},
)
def _failure(failure_type: str) -> FailureInfo:
return FailureInfo(
type=failure_type,
symbolic_reason=f"local {failure_type}",
language_explanation=f"Local explanation for {failure_type}.",
)
|