data_juicer_agents.tools.apply

Apply tools.

class data_juicer_agents.tools.apply.ApplyRecipeInput(*, plan_path: str, dry_run: bool = False, timeout: Annotated[int, Ge(ge=1)] = 300, confirm: bool = False)

基类：BaseModel

plan_path: str

dry_run: bool

timeout: int

confirm: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class data_juicer_agents.tools.apply.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])

基类：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.ApplyUseCase

基类：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]