calute.cortex.orchestration.planner

XML-based planner agent for task orchestration.

This module provides an XML-based planning system for the Cortex framework, inspired by SmolAgent’s planning approach. It enables intelligent task decomposition by having an LLM generate structured execution plans in XML format, which can then be executed step by step.

Key features: - Automatic plan generation from natural language objectives - XML-based plan format for structured, parseable output - Dependency tracking between plan steps - Support for parallel step execution when dependencies allow - Fallback planning when XML parsing fails - Streaming support for real-time plan creation feedback

The module provides three main classes: - PlanStep: Represents a single step in an execution plan - ExecutionPlan: Complete plan with steps and metadata - CortexPlanner: The planner agent that creates and executes plans

Typical usage example:

planner = CortexPlanner(cortex_instance=cortex, verbose=True)

# Create a plan for an objective plan = planner.create_plan(

objective=”Research AI trends and write a summary report”, available_agents=[researcher, writer]

)

# Execute the plan results = planner.execute_plan(plan)

class calute.cortex.orchestration.planner.CortexPlanner(cortex_instance: Optional[Cortex] = None, verbose: bool = True, planning_model: str | None = None)

Bases: object

XML-based planner agent for task orchestration.

CortexPlanner uses an LLM-powered agent to analyze objectives and create structured execution plans in XML format. The planner understands agent capabilities and creates efficient plans with proper dependency ordering.

The planner operates in two phases: 1. Plan Creation: Generates an ExecutionPlan from a natural language objective 2. Plan Execution: Executes the plan step by step, respecting dependencies

cortex_instance

Reference to the parent Cortex instance for access to agents and execution context.

verbose

Whether to output detailed logging information.

planning_model

Optional model identifier for the planner agent.

logger

Logger instance for verbose output (None if verbose=False).

template_engine

PromptTemplate instance for generating planning prompts.

planner_agent

CortexAgent configured as a strategic planner.

Example

planner = CortexPlanner(

cortex_instance=cortex, verbose=True, planning_model=”gpt-4”

)

plan = planner.create_plan(

objective=”Build a data pipeline”, available_agents=[analyst, engineer]

)

results = planner.execute_plan(plan)

create_plan(objective: str, available_agents: list[calute.cortex.agents.agent.CortexAgent], context: str = '', streamer_buffer: calute.core.streamer_buffer.StreamerBuffer | None = None, stream_callback: collections.abc.Callable[[Any], None] | None = None) → ExecutionPlan

Create an execution plan for the given objective.

Uses the planner agent to analyze the objective and available agents, then generates a structured ExecutionPlan with properly ordered steps. Supports streaming for real-time feedback during plan creation.

Parameters

• objective – The high-level goal to create a plan for.

• available_agents – List of CortexAgent instances that can be assigned to plan steps.

• context – Optional additional context to guide plan creation.

• streamer_buffer – Optional StreamerBuffer for streaming output.

• stream_callback – Optional callback function for streaming chunks.

Returns

A complete plan with steps, dependencies, and

metadata. Returns a fallback plan if XML parsing fails.

Return type

ExecutionPlan

Raises

No exceptions are raised; failures result in fallback plans. –

execute_plan(plan: ExecutionPlan, tasks: list['CortexTask'] | None = None) → dict

Execute the plan step by step.

Iterates through plan steps, respecting dependencies, and executes each step using the assigned agent. Results from previous steps are passed as context to dependent steps.

Parameters

• plan – The ExecutionPlan to execute.

• tasks – Optional list of original CortexTask objects to provide additional context during execution.

Returns

Dictionary mapping step IDs to their execution results.

Each key is a step_id (int) and value is the result string.

Return type

dict

Raises

ValueError – If cortex_instance is not set (required for execution).

class calute.cortex.orchestration.planner.ExecutionPlan(plan_id: str, objective: str, steps: list[calute.cortex.orchestration.planner.PlanStep] = <factory>, estimated_time: float = 0.0, complexity: ~typing.Literal['low', 'medium', 'high'] = 'medium')

Bases: object

Complete execution plan with steps and metadata.

ExecutionPlan represents a complete plan for achieving an objective, consisting of multiple PlanStep instances with their dependencies. It provides methods for managing steps and determining execution order based on dependency resolution.

The plan tracks metadata like complexity and estimated time to help with resource allocation and progress estimation.

plan_id

Unique identifier for this plan, typically derived from the objective hash.

Type

str

objective

The high-level goal this plan is designed to achieve.

Type

str

steps

List of PlanStep instances that make up the plan.

Type

list[calute.cortex.orchestration.planner.PlanStep]

estimated_time

Estimated execution time in minutes. Used for planning and progress tracking.

Type

float

complexity

Plan complexity rating (“low”, “medium”, or “high”). Affects resource allocation and timeout settings.

Type

Literal[‘low’, ‘medium’, ‘high’]

Example

plan = ExecutionPlan(

plan_id=”plan_1234”, objective=”Create a comprehensive market analysis”, complexity=”high”, estimated_time=30.0

) plan.add_step(research_step) plan.add_step(analysis_step)

# Get steps ready to execute ready_steps = plan.get_next_steps(completed_steps=set())

add_step(step: PlanStep)

Add a step to the plan.

Appends a new PlanStep to the end of the steps list.

Parameters

step – The PlanStep instance to add to the plan.

complexity: Literal['low', 'medium', 'high'] = 'medium'

estimated_time: float = 0.0

get_next_steps(completed_steps: set[int]) → list[calute.cortex.orchestration.planner.PlanStep]

Get steps that can be executed next based on dependencies.

Analyzes the plan to find all steps whose dependencies have been satisfied (all dependent steps are in the completed set). This enables parallel execution of independent steps.

Parameters

completed_steps – Set of step IDs that have already completed execution successfully.

Returns

List of steps that are ready to execute. These

steps have not yet been completed and have all their dependencies satisfied. May return multiple steps if they can be executed in parallel.

Return type

list[PlanStep]

get_step(step_id: int) → calute.cortex.orchestration.planner.PlanStep | None

Get step by ID.

Searches through the plan’s steps to find one with the matching ID.

Parameters

step_id – The unique identifier of the step to retrieve.

Returns

The step with the matching ID, or None if not found.

Return type

PlanStep

objective: str

plan_id: str

steps: list[calute.cortex.orchestration.planner.PlanStep]

class calute.cortex.orchestration.planner.PlanStep(step_id: int, agent: str, action: str, arguments: dict = <factory>, dependencies: list[int] = <factory>, description: str = '')

Bases: object

A single step in the execution plan.

PlanStep represents an atomic unit of work within an ExecutionPlan. Each step is assigned to a specific agent and defines an action to perform along with any required arguments and dependencies on other steps.

step_id

Unique integer identifier for this step within the plan.

Type

int

agent

The role name of the agent assigned to execute this step.

Type

str

action

The action verb describing what the step does (e.g., “research”, “write”, “analyze”).

Type

str

arguments

Dictionary of key-value pairs providing input parameters for the action. May include references to results from previous steps.

Type

dict

dependencies

List of step IDs that must complete before this step can execute. Empty list means the step can run immediately.

Type

list[int]

description

Human-readable description of what this step accomplishes.

Type

str

Example

step = PlanStep(

step_id=2, agent=”Writer”, action=”write_draft”, arguments={“topic”: “AI trends”, “input”: “result_from_step_1”}, dependencies=[1], description=”Write initial draft based on research findings”

)

action: str

agent: str

arguments: dict

dependencies: list[int]

description: str = ''

step_id: int