calute.cortex.orchestration.dynamic

calute.cortex.orchestration.dynamic#

Dynamic prompt-based execution for Cortex framework.

This module provides dynamic task creation and execution capabilities for the Cortex framework. It enables building workflows from natural language prompts without pre-defining task structures, making it easier to create flexible and adaptive multi-agent systems.

Key features: - Dynamic task creation from natural language prompts - Automatic task chaining with context passing - Streaming execution support - Integration with TaskCreator for intelligent task decomposition - Flexible agent assignment strategies

The module provides two main classes: - DynamicTaskBuilder: Utility for creating tasks from prompts - DynamicCortex: Extended Cortex with dynamic execution capabilities

Typical usage example:

# Create dynamic cortex with agents cortex = create_dynamic_cortex(

agents=[analyst, writer], llm=my_llm, process=ProcessType.SEQUENTIAL

)

# Execute a single prompt result = cortex.execute_prompt(“Analyze the sales data”, agent=”Data Analyst”)

# Or create and execute tasks from a complex prompt result = cortex.execute_with_task_creation(

prompt=”Research AI trends and write a blog post about them”

)

class calute.cortex.orchestration.dynamic.DynamicCortex(agents: list[calute.cortex.agents.agent.CortexAgent], tasks: list[calute.cortex.orchestration.task.CortexTask], llm: BaseLLM, process: ProcessType = ProcessType.SEQUENTIAL, manager_agent: calute.cortex.agents.agent.CortexAgent | None = None, memory_type: MemoryType = MemoryType.SHORT_TERM, verbose: bool = True, max_iterations: int = 10, model: str = 'gpt-4', memory: calute.cortex.agents.memory_integration.CortexMemory | None = None, memory_config: calute.cortex.orchestration.cortex.MemoryConfig | None = None, reinvoke_after_function: bool = True, enable_calute_memory: bool = False, cortex_name: str = 'CorTex')[source]#

Bases: Cortex

Extended Cortex with dynamic prompt execution capabilities.

DynamicCortex extends the base Cortex class with methods for creating and executing tasks dynamically from natural language prompts. This enables more flexible workflows where tasks don’t need to be pre-defined, and the system can adapt to user inputs at runtime.

Key capabilities: - Execute single prompts with automatic agent selection - Create tasks from complex objectives using TaskCreator - Chain multiple prompts with automatic context passing - Stream responses for real-time feedback - Launch interactive UI for prompt execution

task_creator#

Optional TaskCreator instance for intelligent task decomposition. Initialized lazily when first needed.

Inherits all attributes from Cortex base class including:

agents, tasks, llm, process, memory, verbose, etc.

Example

cortex = DynamicCortex(

agents=[researcher, writer], tasks=[], # Can be empty initially llm=my_llm

)

# Execute a simple prompt result = cortex.execute_prompt(“Summarize this article”)

# Or use task creation for complex objectives result = cortex.execute_with_task_creation(

prompt=”Create a marketing plan for our new product”

)

create_tasks_from_prompt(prompt: str, background: str | None = None, auto_assign: bool = True, stream: bool = False, stream_callback: collections.abc.Callable[[Any], None] | None = None) → calute.cortex.orchestration.task_creator.TaskCreationPlan | tuple[calute.cortex.orchestration.task_creator.TaskCreationPlan, list[calute.cortex.orchestration.task.CortexTask]][source]#

Create tasks dynamically from a natural language prompt using TaskCreator.

Lazily initializes a TaskCreator instance and uses it to analyze the provided objective, generating a structured task breakdown. If auto_assign is True and agents are available, the created tasks are automatically assigned to appropriate agents and stored on this DynamicCortex instance.

Parameters

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

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

  • auto_assign – Whether to automatically assign agents to the generated tasks based on role matching. Defaults to True.

  • stream – Whether to stream the LLM response during task creation for real-time feedback. Defaults to False.

  • stream_callback – Optional callback function invoked with each streamed chunk during creation. Only used when stream=True.

Returns

Tuple of (TaskCreationPlan, list[CortexTask]) where tasks are

also stored on self.tasks.

Otherwise:

The TaskCreationPlan alone.

Return type

If auto_assign is True and agents produce CortexTask instances

Side Effects:

  • Initializes self.task_creator on first call.

  • Updates self.tasks with created CortexTask instances when auto_assign produces tasks.

create_ui()[source]#

Create and launch an interactive UI for prompt execution.

Launches a user interface application that allows interactive prompt execution with this DynamicCortex instance. The UI provides a convenient way to experiment with prompts and view agent responses.

Returns

The result of launch_application, typically a UI application instance or handle.

Note

Requires the calute.ui module to be available. The UI runs with this DynamicCortex instance as the execution backend.

execute_prompt(prompt: str, agent: calute.cortex.agents.agent.CortexAgent | str | None = None, stream: bool = False, stream_callback: collections.abc.Callable[[Any], None] | None = None, streamer_buffer: calute.core.streamer_buffer.StreamerBuffer | None = None) → str | tuple[calute.core.streamer_buffer.StreamerBuffer, threading.Thread][source]#

Execute a single prompt with a specific or auto-selected agent.

Creates a temporary task from the prompt, assigns it to the specified agent (or auto-selects one), and executes it. Supports both blocking and streaming execution modes.

Parameters

  • prompt – The natural language prompt to execute as a task.

  • agent – The agent to execute the prompt. Accepts: - A CortexAgent instance to use directly. - A string matching an agent’s role name (case-insensitive). - None to use the first available agent in the Cortex.

  • stream – Whether to stream the response for real-time output. Defaults to False (blocking execution).

  • stream_callback – Optional callback function invoked with each streamed chunk. Used when stream=True.

  • streamer_buffer – Optional pre-existing StreamerBuffer to use for streaming. If None and stream=True, a new buffer is created.

Returns

The agent’s response as a string. If stream=True: Tuple of (StreamerBuffer, Thread) for asynchronous consumption of the streaming response.

Return type

If stream=False

Raises

ValueError – If no matching agent is found for the given agent name or if no agents are configured.

execute_prompts(prompts: list[str] | dict[str, str], process: calute.cortex.core.enums.ProcessType | None = None, stream: bool = False, stream_callback: collections.abc.Callable[[Any], None] | None = None, streamer_buffer: calute.core.streamer_buffer.StreamerBuffer | None = None) → dict[str, str] | str | tuple[calute.core.streamer_buffer.StreamerBuffer, threading.Thread][source]#

Execute multiple prompts with automatic agent assignment.

Supports two input formats: a list of prompts (chained with automatic agent assignment) or a dictionary mapping agent roles to their prompts (explicit assignment). Optionally overrides the process type.

Parameters

  • prompts –

    Either: - list[str]: List of prompts to chain in sequence.

    Agents are assigned round-robin from self.agents. Context is passed between tasks if process is SEQUENTIAL.

    • dict[str, str]: Mapping of agent role names to prompts. Each prompt is assigned to the agent with the matching role.

  • process – Optional ProcessType override for this execution. The original process type is restored after execution. If None, uses the DynamicCortex’s default.

  • stream – Whether to stream responses for real-time output. Defaults to False.

  • stream_callback – Optional callback invoked with each streamed chunk.

  • streamer_buffer – Optional pre-existing StreamerBuffer for streaming. If None and stream=True, a new buffer is created.

Returns

  • stream=False with dict: Dict mapping agent roles to response strings.

  • stream=False with list: The final output string (raw_output).

  • stream=True: Tuple of (StreamerBuffer, Thread) for async consumption.

Return type

Depends on input format and streaming mode

Raises

ValueError – If a dict key does not match any agent’s role name.

execute_with_task_creation(prompt: str, inputs: dict[str, Any] | None = None, background: str | None = None, process: calute.cortex.core.enums.ProcessType | None = None, stream: bool = False, stream_callback: collections.abc.Callable[[Any], None] | None = None) → Any[source]#

Create tasks from a prompt and execute them immediately.

Combines task creation and execution into a single operation. First generates a structured task plan from the given prompt, then executes all tasks using the Cortex’s kickoff mechanism. Optionally overrides the default process type for this execution.

Parameters

  • prompt – The natural language objective to accomplish. Will be decomposed into tasks by the TaskCreator.

  • inputs – Optional dictionary of input values to interpolate into agent and task templates before execution.

  • background – Optional approach or context string to guide the task creation strategy.

  • process – Optional ProcessType override for this execution only. The original process type is restored after execution. If None, uses the DynamicCortex’s default process.

  • stream – Whether to stream execution output for real-time feedback. Defaults to False.

  • stream_callback – Optional callback function invoked with each streamed chunk. Only used when stream=True.

Returns

The execution result from cortex.kickoff(). Type depends on whether streaming is enabled (CortexOutput or streaming result).

Side Effects:

  • Creates and stores tasks on self.tasks.

  • Temporarily overrides self.process if process is provided.

class calute.cortex.orchestration.dynamic.DynamicTaskBuilder[source]#

Bases: object

Utility for creating tasks dynamically from prompts.

DynamicTaskBuilder provides static methods for converting natural language prompts into CortexTask objects. It supports single task creation and chaining multiple prompts into a sequence of dependent tasks.

This class is designed as a stateless utility and should be used through its static methods rather than instantiation.

Example

# Create a single task task = DynamicTaskBuilder.from_prompt(

“Analyze the quarterly report”, agent=analyst_agent

)

# Create a chain of tasks tasks = DynamicTaskBuilder.chain_prompts(

prompts=[“Research topic”, “Write draft”, “Review and edit”], agents=[researcher, writer, editor]

)

static chain_prompts(prompts: list[str], agents: list[calute.cortex.agents.agent.CortexAgent] | None = None, use_context: bool = True) → list[calute.cortex.orchestration.task.CortexTask][source]#

Create a chain of dependent tasks from a list of prompts.

Converts multiple text prompts into a sequence of CortexTask instances with optional context dependencies, where each task can receive the output of the previous task as context. Agents are assigned in a round-robin fashion if fewer agents than prompts are provided.

Parameters

  • prompts – List of natural language prompt strings, each defining one task in the chain. Tasks are created in the order given.

  • agents – Optional list of CortexAgent instances to assign to tasks. Agents are matched to tasks by index. If fewer agents than prompts, agents cycle (wrapping via modulo). If None, all tasks are created without agent assignments.

  • use_context – Whether each task should receive the output of the immediately preceding task as context. When True, a context dependency is set on the previous task. Defaults to True.

Returns

List of CortexTask instances in execution order. If use_context is True, each task (except the first) has a context dependency on the preceding task.

Example

>>> tasks = DynamicTaskBuilder.chain_prompts(
...     prompts=["Research the topic", "Write a draft", "Edit and polish"],
...     agents=[researcher, writer, editor],
...     use_context=True
... )
>>> len(tasks)
3
static from_prompt(prompt: str, agent: calute.cortex.agents.agent.CortexAgent | None = None, expected_output: str = 'Complete the requested task', tools: list | None = None, **task_kwargs) → CortexTask[source]#

Create a CortexTask dynamically from a natural language prompt.

Converts a simple text prompt into a fully configured CortexTask instance. The prompt becomes the task description, and optional parameters allow customization of agent assignment, expected output, and available tools.

Parameters

  • prompt – The natural language instruction or prompt describing what the task should accomplish. Becomes the task’s description field.

  • agent – Optional CortexAgent to assign to this task. If None, the agent can be assigned later before execution.

  • expected_output – Description of what successful task completion looks like. Defaults to “Complete the requested task”.

  • tools – Optional list of tool instances to make available for this specific task. Defaults to an empty list.

  • **task_kwargs – Additional keyword arguments passed directly to the CortexTask constructor (e.g., importance, max_retries, human_feedback).

Returns

A new CortexTask instance configured with the provided prompt as its description and the specified parameters.

Example

>>> task = DynamicTaskBuilder.from_prompt(
...     "Analyze the quarterly sales data",
...     agent=analyst_agent,
...     expected_output="Summary report with key metrics",
...     importance=0.8
... )
calute.cortex.orchestration.dynamic.create_dynamic_cortex(agents: list[calute.cortex.agents.agent.CortexAgent], llm: BaseLLM, process: ProcessType = ProcessType.SEQUENTIAL, **cortex_kwargs) → DynamicCortex[source]#

Create a DynamicCortex instance for flexible prompt-based execution.

Factory function that creates a DynamicCortex pre-configured with the given agents and LLM, starting with an empty task list. The resulting instance supports dynamic task creation and execution from natural language prompts at runtime.

Parameters

  • agents – List of CortexAgent instances to register with the Cortex. These agents will be available for task assignment and execution.

  • llm – BaseLLM instance to use for all language model interactions across agents and the orchestrator.

  • process – Default execution strategy for the Cortex. Determines how tasks are coordinated (e.g., sequential, parallel). Defaults to ProcessType.SEQUENTIAL.

  • **cortex_kwargs – Additional keyword arguments passed directly to the DynamicCortex constructor (e.g., verbose, memory_config, max_iterations, cortex_name).

Returns

A new DynamicCortex instance with an empty task list, ready for dynamic prompt execution via execute_prompt(), execute_prompts(), or execute_with_task_creation().

Example

>>> cortex = create_dynamic_cortex(
...     agents=[researcher, writer],
...     llm=my_llm,
...     process=ProcessType.SEQUENTIAL,
...     verbose=True
... )
>>> result = cortex.execute_prompt("Summarize the report")
Contents