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])

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

to_dict() → Dict[str, object]

__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

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]