data_juicer_agents.tools.apply.apply_recipe.logic module¶

Tool-level APIs for deterministic Data-Juicer execution.

class data_juicer_agents.tools.apply.apply_recipe.logic.ApplyResult(execution_id: str, plan_id: str, start_time: str, end_time: str, duration_seconds: float, model_info: Dict[str, str], generated_recipe_path: str, command: str, status: str, artifacts: Dict[str, str], error_type: str, error_message: str, retry_level: str, next_actions: List[str])[source]¶

Bases: object

Execution summary for one plan apply run.

execution_id: str¶

plan_id: str¶

start_time: str¶

end_time: str¶

duration_seconds: float¶

model_info: Dict[str, str]¶

generated_recipe_path: str¶

command: str¶

status: str¶

artifacts: Dict[str, str]¶

error_type: str¶

error_message: str¶

retry_level: str¶

next_actions: List[str]¶

static new_id() → str[source]¶

to_dict() → Dict[str, object][source]¶

__init__(execution_id: str, plan_id: str, start_time: str, end_time: str, duration_seconds: float, model_info: Dict[str, str], generated_recipe_path: str, command: str, status: str, artifacts: Dict[str, str], error_type: str, error_message: str, retry_level: str, next_actions: List[str]) → None¶

class data_juicer_agents.tools.apply.apply_recipe.logic.ApplyUseCase[source]¶

Bases: object

Execute validated plans and return execution summaries.

execute(plan_payload: Dict[str, Any], runtime_dir: Path, dry_run: bool = False, timeout_seconds: int = 300, command_override: str | Iterable[str] | None = None, cancel_check: Callable[[], bool] | None = None) → Tuple[ApplyResult, int, str, str][source]¶