From ad7779ff338c6427883d99bff5d362441f6a0b97 Mon Sep 17 00:00:00 2001 From: skywhite1024 <129768272+skywhite1024@users.noreply.github.com> Date: Thu, 27 Aug 2026 14:42:07 +0800 Subject: [PATCH] feat(task-engine): add semantic task frontend --- .../embodichain.gen_sim.task_engine.rst | 6 + embodichain/gen_sim/task_engine/__init__.py | 10 + embodichain/gen_sim/task_engine/agent.py | 289 ++++++++++++++++++ tests/gen_sim/task_engine/test_agent.py | 242 +++++++++++++++ 4 files changed, 547 insertions(+) create mode 100644 embodichain/gen_sim/task_engine/agent.py create mode 100644 tests/gen_sim/task_engine/test_agent.py diff --git a/docs/source/api_reference/embodichain/embodichain.gen_sim.task_engine.rst b/docs/source/api_reference/embodichain/embodichain.gen_sim.task_engine.rst index b62262a93..85b5d0ba2 100644 --- a/docs/source/api_reference/embodichain/embodichain.gen_sim.task_engine.rst +++ b/docs/source/api_reference/embodichain/embodichain.gen_sim.task_engine.rst @@ -11,6 +11,12 @@ Package exports .. automodule:: embodichain.gen_sim.task_engine :members: +Semantic frontend +----------------- + +.. automodule:: embodichain.gen_sim.task_engine.agent + :members: + Contracts --------- diff --git a/embodichain/gen_sim/task_engine/__init__.py b/embodichain/gen_sim/task_engine/__init__.py index bb142aaa1..669fa2216 100644 --- a/embodichain/gen_sim/task_engine/__init__.py +++ b/embodichain/gen_sim/task_engine/__init__.py @@ -18,6 +18,12 @@ from __future__ import annotations +from .agent import ( + TaskAgent, + TaskGenerationError, + derive_scene_request, + derive_success_spec, +) from .contracts import ( SCENE_REQUEST_SCHEMA, SUCCESS_SPEC_SCHEMA, @@ -68,11 +74,15 @@ "TASK_DRAFT_SCHEMA", "TERMINAL_BEHAVIORS", "TRANSPORT_DIRECTIONS", + "TaskAgent", "TaskCandidate", "TaskCandidateSet", "TaskContract", "TaskDraft", + "TaskGenerationError", "canonical_hash", + "derive_scene_request", + "derive_success_spec", "interpret_instruction_draft", "task_contract", "task_success_type", diff --git a/embodichain/gen_sim/task_engine/agent.py b/embodichain/gen_sim/task_engine/agent.py new file mode 100644 index 000000000..1d183e9d2 --- /dev/null +++ b/embodichain/gen_sim/task_engine/agent.py @@ -0,0 +1,289 @@ +# ---------------------------------------------------------------------------- +# Copyright (c) 2021-2026 DexForce Technology Co., Ltd. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ---------------------------------------------------------------------------- + +"""Scene-independent semantic candidate generation for Task Engine.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from concurrent.futures import ThreadPoolExecutor, as_completed +from copy import deepcopy +from dataclasses import dataclass +from typing import Any + +from .contracts import ( + SCENE_REQUEST_SCHEMA, + SUCCESS_SPEC_SCHEMA, + TASK_CANDIDATE_SET_SCHEMA, + TASK_DRAFT_SCHEMA, + TaskCandidate, + TaskCandidateSet, + canonical_hash, + validate_task_candidate, + validate_task_candidate_set, +) +from .interpretation import ( + InstructionCaller, + InstructionDraftResult, + interpret_instruction_draft, + validate_instruction_intent, +) +from .ontology import TASK_CONTRACTS, task_success_type + +__all__ = [ + "TaskAgent", + "TaskGenerationError", + "derive_scene_request", + "derive_success_spec", +] + +DraftInterpreter = Callable[..., InstructionDraftResult] + + +class TaskGenerationError(ValueError): + """Raised when every independently generated candidate fails validation.""" + + +@dataclass(frozen=True) +class _CandidateAttempt: + index: int + result: InstructionDraftResult | None = None + error: str = "" + + +class TaskAgent: + """Generate, validate, normalize, and vote on independent task drafts.""" + + def __init__( + self, + *, + caller: InstructionCaller | None = None, + interpreter: DraftInterpreter = interpret_instruction_draft, + ) -> None: + self._caller = caller + self._interpreter = interpreter + + def generate( + self, + task_id: str, + instruction: str, + model: str | None = None, + candidate_count: int = 3, + ) -> TaskCandidateSet: + """Generate candidates concurrently and retain votes after deduplication.""" + normalized_task_id = str(task_id).strip() + normalized_instruction = str(instruction).strip() + if not normalized_task_id or not normalized_instruction: + raise ValueError("task_id and instruction must be non-empty.") + if ( + isinstance(candidate_count, bool) + or not isinstance(candidate_count, int) + or candidate_count < 1 + ): + raise ValueError("candidate_count must be a positive integer.") + + attempts: list[_CandidateAttempt] = [] + with ThreadPoolExecutor( + max_workers=candidate_count, + thread_name_prefix="task-agent", + ) as executor: + futures = { + executor.submit( + self._interpreter, + normalized_instruction, + model=model, + caller=self._caller, + ): index + for index in range(candidate_count) + } + for future in as_completed(futures): + index = futures[future] + try: + attempts.append( + _CandidateAttempt(index=index, result=future.result()) + ) + except Exception as error: # Each candidate is an isolated vote. + attempts.append( + _CandidateAttempt( + index=index, + error=f"candidate_{index + 1:02d}: {type(error).__name__}: {error}", + ) + ) + attempts.sort(key=lambda item: item.index) + errors = [item.error for item in attempts if item.result is None] + unique: dict[str, TaskCandidate] = {} + valid_response_count = 0 + for attempt in attempts: + if attempt.result is None: + continue + assert attempt.result is not None + try: + canonical_intent = _canonicalize_intent(attempt.result.intent) + draft = { + "schema_version": TASK_DRAFT_SCHEMA, + "task_id": normalized_task_id, + "instruction": normalized_instruction, + "steps": canonical_intent["steps"], + } + semantic_hash = canonical_hash(draft["steps"]) + candidate_id = f"candidate_{len(unique) + 1:02d}" + candidate = validate_task_candidate( + { + "candidate_id": candidate_id, + "draft": draft, + "scene_request": derive_scene_request(draft), + "success_spec": derive_success_spec(draft), + "semantic_hash": semantic_hash, + "vote_count": 1, + "attempts": attempt.result.attempts, + "normalizations": deepcopy(list(attempt.result.normalizations)), + } + ) + existing = unique.get(semantic_hash) + if existing is not None: + existing["vote_count"] += 1 + existing["attempts"] = max( + existing["attempts"], attempt.result.attempts + ) + existing["normalizations"].extend(candidate["normalizations"]) + else: + unique[semantic_hash] = candidate + valid_response_count += 1 + except Exception as error: # Post-processing failures stay candidate-local. + errors.append( + f"candidate_{attempt.index + 1:02d}: " + f"{type(error).__name__}: {error}" + ) + + if not unique: + raise TaskGenerationError( + "All Task Agent candidates failed validation: " + "; ".join(errors) + ) + + return validate_task_candidate_set( + { + "schema_version": TASK_CANDIDATE_SET_SCHEMA, + "task_id": normalized_task_id, + "instruction": normalized_instruction, + "candidates": list(unique.values()), + "requested_candidate_count": candidate_count, + "valid_response_count": valid_response_count, + "errors": errors, + } + ) + + +def derive_scene_request(draft: Mapping[str, Any]) -> dict[str, Any]: + """Derive structural scene constraints without classifying reference text.""" + from .contracts import validate_scene_request, validate_task_draft + + normalized = validate_task_draft(draft) + references: list[dict[str, Any]] = [] + for step in normalized["steps"]: + task_type = str(step["task_type"]) + contract = TASK_CONTRACTS[task_type] + for role in ("object", "target"): + selector = step[role] + if selector["kind"] != "scene_ref": + continue + if role == "object": + structure = contract.source_structure + affordances = sorted(contract.scene_affordances) + initial_state = {"orientation": "fallen"} if task_type == "E2" else {} + attributes: dict[str, Any] = {} + else: + structure = _target_structure(task_type, str(step["relation"])) + affordances = _target_affordances(task_type, str(step["relation"])) + initial_state = {} + attributes = {} + references.append( + { + "reference_id": f"{step['id']}.{role}", + "step_id": step["id"], + "role": role, + "reference": selector["reference"], + "quantifier": selector["quantifier"], + "count": selector["count"], + "source_structure": structure, + "affordances": affordances, + "initial_state": initial_state, + "attributes": attributes, + } + ) + if not references: + raise ValueError("A TaskDraft must contain at least one scene_ref selector.") + return validate_scene_request( + { + "schema_version": SCENE_REQUEST_SCHEMA, + "task_id": normalized["task_id"], + "references": references, + } + ) + + +def derive_success_spec(draft: Mapping[str, Any]) -> dict[str, Any]: + """Derive every success term exclusively from the E-task ontology.""" + from .contracts import validate_success_spec, validate_task_draft + + normalized = validate_task_draft(draft) + return validate_success_spec( + { + "schema_version": SUCCESS_SPEC_SCHEMA, + "task_id": normalized["task_id"], + "op": "all", + "terms": [ + { + "step_id": step["id"], + "type": task_success_type(step["task_type"], step), + } + for step in normalized["steps"] + ], + }, + draft=normalized, + ) + + +def _canonicalize_intent(intent: Mapping[str, Any]) -> dict[str, Any]: + """Remove arbitrary model step IDs while preserving the explicit DAG order.""" + normalized = validate_instruction_intent(intent) + id_map = { + step["id"]: f"step_{index + 1:02d}" + for index, step in enumerate(normalized["steps"]) + } + steps = deepcopy(normalized["steps"]) + for step in steps: + old_id = step["id"] + step["id"] = id_map[old_id] + step["depends_on"] = [id_map[item] for item in step["depends_on"]] + for selector_name in ("object", "target"): + selector = step[selector_name] + if selector["kind"] == "step_result": + selector["step_id"] = id_map[selector["step_id"]] + return validate_instruction_intent({"steps": steps}) + + +def _target_affordances(task_type: str, relation: str) -> list[str]: + if task_type == "E3" or relation == "inside": + return ["container"] + return [] + + +def _target_structure(task_type: str, relation: str) -> str: + if relation == "on": + return "physical_entity" + if task_type == "E3" or relation == "inside": + return "rigid_object" + return "spatial_reference" diff --git a/tests/gen_sim/task_engine/test_agent.py b/tests/gen_sim/task_engine/test_agent.py new file mode 100644 index 000000000..3e20d5e58 --- /dev/null +++ b/tests/gen_sim/task_engine/test_agent.py @@ -0,0 +1,242 @@ +# ---------------------------------------------------------------------------- +# Copyright (c) 2021-2026 DexForce Technology Co., Ltd. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ---------------------------------------------------------------------------- + +from __future__ import annotations + +from copy import deepcopy +import threading +from time import sleep + +import pytest + +from embodichain.gen_sim.task_engine.contracts import ( + SUCCESS_SPEC_SCHEMA, + TASK_DRAFT_SCHEMA, + validate_success_spec, + validate_task_candidate, + validate_task_draft, +) +from embodichain.gen_sim.task_engine.agent import ( + TaskAgent, + TaskGenerationError, + derive_scene_request, + derive_success_spec, +) +from embodichain.gen_sim.task_engine.interpretation import InstructionDraftResult + +_TEST_INSTRUCTION = "test-instruction" + + +def _selector(kind="none", *, step_id="", reference="", quantifier="one", count=0): + return { + "kind": kind, + "step_id": step_id, + "reference": reference, + "quantifier": quantifier, + "count": count, + } + + +def _step(step_id="orient", reference="purple can"): + return { + "id": step_id, + "task_type": "E2", + "object": _selector("scene_ref", reference=reference), + "target": _selector(), + "relation": "none", + "required_arm": "auto", + "transfer_arm": "none", + "receive_arm": "none", + "orientation_goal": "upright", + "target_state": "none", + "target_setting": 0, + "layout": "none", + "axis": "none", + "direction": "none", + "terminal_behavior": "none", + "depends_on": [], + } + + +def _result(step): + return InstructionDraftResult( + intent={"steps": [deepcopy(step)]}, + model="injected_caller", + attempts=1, + latency_seconds=0.01, + normalizations=(), + ) + + +def test_task_agent_generates_concurrently_deduplicates_and_counts_votes(): + barrier = threading.Barrier(3) + lock = threading.Lock() + assigned = 0 + + def interpreter(_instruction, **_kwargs): + nonlocal assigned + with lock: + index = assigned + assigned += 1 + barrier.wait(timeout=2) + sleep(0.01) + if index < 2: + return _result(_step(step_id=f"arbitrary_{index}")) + return _result(_step(step_id="different", reference="orange can")) + + result = TaskAgent(interpreter=interpreter).generate("task", _TEST_INSTRUCTION) + + assert result["requested_candidate_count"] == 3 + assert result["valid_response_count"] == 3 + assert len(result["candidates"]) == 2 + assert sorted(item["vote_count"] for item in result["candidates"]) == [1, 2] + assert {item["draft"]["steps"][0]["id"] for item in result["candidates"]} == { + "step_01" + } + + +def test_scene_request_and_success_are_deterministic_contract_derivations(): + draft = { + "schema_version": TASK_DRAFT_SCHEMA, + "task_id": "upright", + "instruction": _TEST_INSTRUCTION, + "steps": [_step(reference="all cans")], + } + draft["steps"][0]["object"].update(quantifier="all") + + request = derive_scene_request(draft) + success = derive_success_spec(draft) + + assert request["references"] == [ + { + "reference_id": "orient.object", + "step_id": "orient", + "role": "object", + "reference": "all cans", + "quantifier": "all", + "count": 0, + "source_structure": "rigid_object", + "affordances": ["graspable", "orientable"], + "initial_state": {"orientation": "fallen"}, + "attributes": {}, + } + ] + assert success["terms"] == [{"step_id": "orient", "type": "object_upright"}] + + +@pytest.mark.parametrize( + ("relation", "expected_structure", "expected_affordances"), + [ + ("on", "physical_entity", []), + ("inside", "rigid_object", ["container"]), + ("behind", "spatial_reference", []), + ("front_of", "spatial_reference", []), + ("left_of", "spatial_reference", []), + ("right_of", "spatial_reference", []), + ], +) +def test_target_requirements_describe_capabilities_not_concrete_roles( + relation: str, + expected_structure: str, + expected_affordances: list[str], +) -> None: + step = _step(step_id="place", reference="green can") + step.update( + task_type="E1", + target=_selector("scene_ref", reference="red can"), + relation=relation, + orientation_goal="preserve", + ) + draft = { + "schema_version": TASK_DRAFT_SCHEMA, + "task_id": "stack", + "instruction": _TEST_INSTRUCTION, + "steps": [step], + } + + request = derive_scene_request(draft) + + target = next( + reference + for reference in request["references"] + if reference["role"] == "target" + ) + assert target["source_structure"] == expected_structure + assert target["affordances"] == expected_affordances + + +def test_draft_rejects_grounded_fields_and_task_agent_fails_closed(): + draft = { + "schema_version": TASK_DRAFT_SCHEMA, + "task_id": "bad", + "instruction": "bad", + "steps": [_step()], + } + draft["steps"][0]["object"]["uid"] = "scene_uid" + with pytest.raises(ValueError, match="forbidden|exactly fields"): + validate_task_draft(draft) + + def invalid(_instruction, **_kwargs): + raise ValueError("invalid draft after repair") + + with pytest.raises(TaskGenerationError, match="All Task Agent candidates"): + TaskAgent(interpreter=invalid).generate("bad", "bad") + + +def test_task_candidate_rejects_scene_constraints_not_derived_from_draft(): + candidate = TaskAgent( + interpreter=lambda *_args, **_kwargs: _result(_step()) + ).generate("upright", _TEST_INSTRUCTION, candidate_count=1)["candidates"][0] + candidate["scene_request"]["references"][0]["affordances"] = [] + + with pytest.raises(ValueError, match="derived exactly"): + validate_task_candidate(candidate) + + +def test_success_spec_rejects_types_outside_task_ontology(): + with pytest.raises(ValueError, match="must be one of"): + validate_success_spec( + { + "schema_version": SUCCESS_SPEC_SCHEMA, + "task_id": "bad_success", + "op": "all", + "terms": [{"step_id": "step_01", "type": "looks_good"}], + } + ) + + +def test_task_agent_isolates_invalid_interpreter_results(): + lock = threading.Lock() + calls = 0 + + def interpreter(_instruction, **_kwargs): + nonlocal calls + with lock: + index = calls + calls += 1 + if index == 0: + invalid = _step() + invalid["object"]["uid"] = "red_can" + return _result(invalid) + return _result(_step()) + + result = TaskAgent(interpreter=interpreter).generate( + "upright", _TEST_INSTRUCTION, candidate_count=2 + ) + + assert result["valid_response_count"] == 1 + assert len(result["errors"]) == 1 + assert len(result["candidates"]) == 1