calute.cortex.orchestration.task_creator

Dynamic task creator agent for generating tasks from prompts.

This module provides the TaskCreator class for automatically generating structured task breakdowns from natural language objectives. It uses an LLM-powered agent to analyze objectives and produce detailed, actionable task plans that can be executed within the Cortex framework.

The module includes: - TaskCreator: Main class for generating tasks from prompts - TaskDefinition: Data class representing individual task specifications - TaskCreationPlan: Container for complete task breakdowns

Key features: - Natural language to structured task conversion - Automatic agent assignment based on task requirements - Dependency detection and ordering - Priority and importance scoring - Tool requirement identification - Background/approach context integration - Streaming support for task creation - Fallback handling for parsing failures

Typical usage example:

from calute.cortex.task_creator import TaskCreator from calute.cortex.agent import CortexAgent

creator = TaskCreator(verbose=True, model=”gpt-4”)

agents = [

CortexAgent(role=”Researcher”, goal=”Research topics”, backstory=”Expert”), CortexAgent(role=”Writer”, goal=”Write content”, backstory=”Expert”)

]

plan, tasks = creator.create_tasks_from_prompt(

prompt=”Write a research report on AI trends”, background=”Focus on recent developments”, available_agents=agents

)

for task in tasks:

print(f”Task: {task.description}”)

class calute.cortex.orchestration.task_creator.TaskCreationPlan(plan_id: str, objective: str, approach: str, tasks: list[calute.cortex.orchestration.task_creator.TaskDefinition] = <factory>, estimated_complexity: ~typing.Literal['simple', 'medium', 'complex'] = 'medium', total_tasks: int = 0, sequential: bool = True)

Bases: object

Complete task creation plan containing all task definitions.

TaskCreationPlan represents the full breakdown of an objective into structured tasks. It includes metadata about the plan itself and a collection of TaskDefinition instances.

plan_id

Unique identifier for this plan (typically based on objective hash).

Type

str

objective

The original objective or goal that was broken down.

Type

str

approach

Description of the strategy used for task decomposition.

Type

str

tasks

List of TaskDefinition instances in execution order.

Type

list[calute.cortex.orchestration.task_creator.TaskDefinition]

estimated_complexity

Overall complexity rating (‘simple’, ‘medium’, ‘complex’).

Type

Literal[‘simple’, ‘medium’, ‘complex’]

total_tasks

Current count of tasks in the plan.

Type

int

sequential

Whether tasks should be executed in sequence (True) or can be parallelized where dependencies allow (False).

Type

bool

add_task(task: TaskDefinition)

Add a task to the plan and update the task count.

Parameters

task – The TaskDefinition to add to the plan.

Side Effects:

Appends task to the tasks list and updates total_tasks.

approach: str

estimated_complexity: Literal['simple', 'medium', 'complex'] = 'medium'

get_task_by_id(task_id: int) → calute.cortex.orchestration.task_creator.TaskDefinition | None

Retrieve a task definition by its ID.

Parameters

task_id – The unique identifier of the task to find.

Returns

The matching TaskDefinition, or None if not found.

objective: str

plan_id: str

sequential: bool = True

tasks: list[calute.cortex.orchestration.task_creator.TaskDefinition]

total_tasks: int = 0

class calute.cortex.orchestration.task_creator.TaskCreator(verbose: bool = True, model: str | None = None, llm: calute.llms.base.BaseLLM | None = None, max_tasks: int = 10, auto_assign_agents: bool = True)

Bases: object

Dynamic task creator that generates structured tasks from natural language prompts.

TaskCreator uses an LLM-powered agent to analyze high-level objectives and automatically generate detailed, actionable task breakdowns. It produces structured XML output that is parsed into TaskDefinition and TaskCreationPlan objects, which can then be converted to CortexTask instances for execution.

verbose

Whether to output detailed logging during task creation.

model

Model identifier for the creator agent’s LLM.

llm

Direct BaseLLM instance for LLM operations.

max_tasks

Maximum number of tasks to include in a plan.

auto_assign_agents

Whether to suggest agent assignments based on roles.

logger

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

template_engine

PromptTemplate engine for rendering prompts.

creator_agent

Internal CortexAgent that performs task creation.

Class Attributes:

TASK_CREATION_TEMPLATE: Jinja2 template for generating task creation prompts.

Includes placeholders for objective, background, available agents, and constraints.

Example

>>> creator = TaskCreator(model="gpt-4", verbose=True)
>>> plan, tasks = creator.create_tasks_from_prompt(
...     prompt="Build a REST API for user management",
...     background="Use FastAPI with PostgreSQL",
...     available_agents=my_agents
... )
>>> print(f"Created {len(tasks)} tasks")

TASK_CREATION_TEMPLATE = '\nYou are a task creation specialist. Create a detailed set of tasks for the following objective.\n\nOBJECTIVE: {{ objective }}\n\n{% if background %}\nBACKGROUND/APPROACH:\n{{ background }}\nThis background should guide your approach to breaking down the tasks.\n{% else %}\nBACKGROUND/APPROACH: Use your best judgment to determine the optimal approach.\n{% endif %}\n\n{% if available_agents %}\nAVAILABLE AGENTS:\n{% for agent in available_agents %}\n- {{ agent.role }}: {{ agent.goal }}\n{% endfor %}\n{% endif %}\n\n{% if constraints %}\nCONSTRAINTS:\n{{ constraints }}\n{% endif %}\n\nCreate a task breakdown using the following XML format:\n\n<task_plan>\n    <objective>{{ objective }}</objective>\n    <approach>Brief description of the approach taken based on background</approach>\n    <complexity>simple|medium|complex</complexity>\n    <sequential>true|false</sequential>\n\n    <task id="1">\n        <description>Clear description of what needs to be done</description>\n        <expected_output>What the successful completion looks like</expected_output>\n        <agent_role>Optional: Suggested agent role for this task</agent_role>\n        <dependencies></dependencies>\n        <context_needed>true|false</context_needed>\n        <tools_needed>tool1,tool2</tools_needed>\n        <importance>0.1-1.0</importance>\n        <validation_required>true|false</validation_required>\n        <human_feedback>true|false</human_feedback>\n    </task>\n\n    <task id="2">\n        <description>Another task description</description>\n        <expected_output>Expected result</expected_output>\n        <agent_role>Another Agent Role</agent_role>\n        <dependencies>1</dependencies>\n        <context_needed>true</context_needed>\n        <tools_needed></tools_needed>\n        <importance>0.5</importance>\n        <validation_required>false</validation_required>\n        <human_feedback>false</human_feedback>\n    </task>\n</task_plan>\n\nINSTRUCTIONS:\n1. Break down the objective into clear, actionable tasks\n2. Each task should be self-contained but can depend on others\n3. Consider the background/approach when determining task breakdown\n4. Assign importance scores (0.1=low, 0.5=medium, 1.0=critical)\n5. Specify if tasks need context from previous tasks\n6. Identify any tools or capabilities needed\n7. Mark tasks that need validation or human feedback\n8. Create between 2-10 tasks as appropriate\n\nRespond ONLY with the XML plan, no additional text.\n'

create_and_execute(prompt: str, background: str | None, cortex: Cortex, process_type: ProcessType | None = None, use_streaming: bool = False, stream_callback: Callable[[Any], None] | None = None, log_process: bool = False) → Any | tuple[StreamerBuffer, Thread]

Create tasks from a prompt and immediately execute them using a Cortex.

Combines task creation and execution into a single operation. First generates tasks from the prompt using available agents in the Cortex, then kicks off execution of those tasks.

Parameters

• prompt – The objective or goal to accomplish.

• background – Optional approach or context for task creation.

• cortex – The Cortex instance containing agents for execution. Must have agents defined.

• process_type – Optional ProcessType to override the Cortex’s default process type for this execution.

• use_streaming – Whether to stream execution output.

• stream_callback – Optional callback for streaming chunks.

• log_process – Whether to log the execution process.

Returns

The execution result from cortex.kickoff(). If use_streaming=True: Tuple of (StreamerBuffer, Thread).

Return type

If use_streaming=False

Raises

ValueError – If the Cortex has no agents defined.

Note

The original process_type is restored after execution if overridden.

create_tasks_from_prompt(prompt: str, background: str | None = None, available_agents: list[calute.cortex.agents.agent.CortexAgent] | None = None, constraints: str | None = None, stream: bool = False, stream_callback: collections.abc.Callable[[Any], None] | None = None, streamer_buffer: calute.core.streamer_buffer.StreamerBuffer | None = None) → tuple[calute.cortex.orchestration.task_creator.TaskCreationPlan, list[calute.cortex.orchestration.task.CortexTask] | None]

Create structured tasks from a natural language prompt.

Analyzes the provided objective and generates a detailed task breakdown using the internal creator agent. Optionally assigns agents to tasks based on role matching.

Parameters

• prompt – The objective or goal to break down into tasks.

• background – Optional approach or context to guide task creation. Helps the creator understand the preferred methodology.

• available_agents – Optional list of CortexAgent instances for automatic task assignment. If provided and auto_assign_agents is True, returns CortexTask instances with agents assigned.

• constraints – Optional constraints or requirements that tasks must satisfy.

• stream – Whether to stream the LLM response during creation.

• stream_callback – Optional callback invoked with each streamed chunk.

• streamer_buffer – Optional StreamerBuffer for collecting stream output.

Returns

• TaskCreationPlan containing all parsed TaskDefinition objects

• List of CortexTask instances if available_agents provided and auto_assign_agents is True, otherwise None

Return type

Tuple of (TaskCreationPlan, list[CortexTask] | None)

Note

If XML parsing fails, a fallback single-task plan is returned.

create_ui(cortex: Cortex | None = None)

Launch a UI application for interacting with the task creator.

Creates and launches an interactive user interface that allows users to create and execute tasks visually through the TaskCreator.

Parameters

cortex – Optional Cortex instance to use for task execution. If not provided, only task creation will be available.

Returns

The launched UI application instance.

class calute.cortex.orchestration.task_creator.TaskDefinition(task_id: int, description: str, expected_output: str, agent_role: str | None = None, dependencies: list[int] = <factory>, context_needed: bool = False, tools_needed: list[str] = <factory>, importance: float = 0.5, validation_required: bool = False, human_feedback: bool = False)

Bases: object

Definition of a task to be created from an LLM-generated plan.

TaskDefinition represents the specification of a single task as parsed from the LLM’s XML task plan output. It contains all the metadata needed to create an actual CortexTask instance.

task_id

Unique integer identifier for this task within the plan.

Type

int

description

Detailed description of what the task should accomplish.

Type

str

expected_output

Description of what successful completion looks like.

Type

str

agent_role

Suggested role name for the agent to execute this task.

Type

str | None

dependencies

List of task_ids that must complete before this task.

Type

list[int]

context_needed

Whether this task requires context from previous tasks.

Type

bool

tools_needed

List of tool names that may be needed for execution.

Type

list[str]

importance

Priority score from 0.1 (low) to 1.0 (critical).

Type

float

validation_required

Whether output should be validated against a schema.

Type

bool

human_feedback

Whether to request human review of the output.

Type

bool

agent_role: str | None = None

context_needed: bool = False

dependencies: list[int]

description: str

expected_output: str

human_feedback: bool = False

importance: float = 0.5

task_id: int

tools_needed: list[str]

validation_required: bool = False