agents.planning_v2.base.tree_models¶
Tree-based planning models using the enhanced tree_leaf structure.
This module provides planning models that leverage the generic tree/leaf structure from haive-core for more flexible and type-safe planning.
Classes¶
Content for plan nodes (both tasks and sub-plans). |
|
Result of executing a plan node. |
|
Status for plan nodes. |
|
A concrete plan implementation using the tree structure. |
Functions¶
|
Create a plan with phases (sub-plans). |
|
Create a simple linear plan from a list of task names. |
Module Contents¶
- class agents.planning_v2.base.tree_models.PlanContent(/, **data)¶
Bases:
pydantic.BaseModel
Content for plan nodes (both tasks and sub-plans).
Create a new model by parsing and validating input data from keyword arguments.
Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.
self is explicitly positional-only to allow self as a field name.
- Parameters:
data (Any)
- class agents.planning_v2.base.tree_models.PlanResult(/, **data)¶
Bases:
pydantic.BaseModel
Result of executing a plan node.
Create a new model by parsing and validating input data from keyword arguments.
Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.
self is explicitly positional-only to allow self as a field name.
- Parameters:
data (Any)
- class agents.planning_v2.base.tree_models.PlanStatus¶
-
Status for plan nodes.
Initialize self. See help(type(self)) for accurate signature.
- class agents.planning_v2.base.tree_models.TaskPlan¶
Bases:
PlanTree
A concrete plan implementation using the tree structure.
This class extends the generic Tree to provide planning-specific functionality while maintaining type safety.
Examples
# Create a plan plan = TaskPlan(content=PlanContent(
objective=”Deploy new feature”, priority=4
))
# Add simple tasks plan.add_task(“Write tests”, priority=5) plan.add_task(“Code review”, priority=3)
# Add sub-plan deploy_plan = plan.add_subplan(“Deploy to production”) deploy_plan.add_task(“Deploy to staging”) deploy_plan.add_task(“Run smoke tests”) deploy_plan.add_task(“Deploy to prod”)
# Check status print(f”Total tasks: {plan.total_nodes}”) print(f”Progress: {plan.progress_percentage}%”)
- add_parallel_tasks(tasks)¶
Add multiple tasks that can execute in parallel.
- add_subplan(objective, description='', priority=1)¶
Add a sub-plan that can contain its own tasks.
- add_task(objective, description='', priority=1)¶
Add a simple task to the plan.
- get_current_task()¶
Get the current task to execute (first pending or in-progress).
- Return type:
Union[PlanLeaf, TaskPlan] | None
- get_tasks_by_priority(min_priority=1)¶
Get all tasks with priority >= min_priority.
- mark_current_completed(output='Done')¶
Mark the current active task as completed.
- mark_current_failed(error)¶
Mark the current active task as failed.
- agents.planning_v2.base.tree_models.create_phased_plan(objective, phases)¶
Create a plan with phases (sub-plans).
- Parameters:
- Return type:
Examples
- plan = create_phased_plan(
“Launch Product”, {
“Development”: [“Code”, “Test”, “Review”], “Deployment”: [“Stage”, “Verify”, “Prod”], “Marketing”: [“Announce”, “Demo”, “Feedback”]
}
)