calute.cortex.orchestration.cortex

calute.cortex.orchestration.cortex#

Main Cortex orchestration module for multi-agent collaboration.

This module provides the core Cortex orchestrator class that coordinates multiple AI agents to accomplish complex tasks through various execution strategies. It serves as the central hub for managing agent lifecycles, task execution, memory integration, and output collection.

The Cortex orchestrator supports multiple process types:

  • SEQUENTIAL: Tasks are executed one after another, with context passing

  • PARALLEL: Independent tasks run concurrently for faster execution

  • HIERARCHICAL: A manager agent delegates tasks to worker agents

  • CONSENSUS: All agents contribute to each task, then reach agreement

  • PLANNED: An AI planner creates an optimized execution plan

Key Components:

  • Cortex: Main orchestrator class managing agents, tasks, and execution

  • CortexOutput: Structured output container with execution metadata

  • MemoryConfig: TypedDict for configuring memory system parameters

Typical usage example:
>>> from calute.cortex import Cortex, CortexAgent, CortexTask
>>> from calute.llms import OpenAILLM
>>>
>>> llm = OpenAILLM(api_key="your-key")
>>> agent = CortexAgent(role="Writer", goal="Write content", backstory="Expert writer")
>>> task = CortexTask(description="Write an article", expected_output="Article text", agent=agent)
>>> cortex = Cortex(agents=[agent], tasks=[task], llm=llm)
>>> result = cortex.kickoff()
class calute.cortex.orchestration.cortex.Cortex(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', parallel_max_workers: int | None = None)[source]#

Bases: object

Main orchestrator for multi-agent collaboration and task execution.

Cortex is the central coordination hub that manages multiple AI agents working together on complex tasks. It handles task distribution, execution flow, memory management, and result collection across different execution strategies.

The orchestrator supports five distinct process types:

  • SEQUENTIAL: Execute tasks one after another, passing context between them

  • PARALLEL: Run independent tasks concurrently for improved performance

  • HIERARCHICAL: Use a manager agent to delegate and review worker tasks

  • CONSENSUS: Gather contributions from all agents and synthesize agreement

  • PLANNED: Use AI-powered planning to optimize task execution order

agents#

List of CortexAgent instances available for task execution.

tasks#

List of CortexTask instances to be executed.

process#

The ProcessType determining execution strategy.

manager_agent#

Optional agent for hierarchical process management.

verbose#

Whether to enable detailed logging output.

max_iterations#

Maximum retry attempts for failed task executions.

reinvoke_after_function#

Whether to reinvoke LLM after tool execution.

enable_calute_memory#

Whether to enable Calute’s internal memory system.

cortex_name#

Display name for the Cortex instance in logs.

cortex_memory#

CortexMemory instance for shared memory across agents.

memory#

MemoryStore instance for basic memory operations.

memory_type#

Type of memory to use for context management.

task_outputs#

List of CortexTaskOutput from completed tasks.

logger#

Logger instance for verbose output.

template_engine#

PromptTemplate instance for generating prompts.

planner#

CortexPlanner instance for PLANNED process type (if applicable).

llm#

BaseLLM instance for language model interactions.

calute#

Calute instance managing agent registrations and execution.

Example

>>> from calute.cortex import Cortex, CortexAgent, CortexTask, ProcessType
>>> from calute.llms import OpenAILLM
>>>
>>> llm = OpenAILLM(api_key="your-key")
>>> writer = CortexAgent(
...     role="Content Writer",
...     goal="Create engaging content",
...     backstory="Professional writer with 10 years experience"
... )
>>> editor = CortexAgent(
...     role="Editor",
...     goal="Polish and improve content",
...     backstory="Detail-oriented editor"
... )
>>> write_task = CortexTask(
...     description="Write an article about AI",
...     expected_output="Draft article",
...     agent=writer
... )
>>> edit_task = CortexTask(
...     description="Edit and improve the article",
...     expected_output="Polished article",
...     agent=editor,
...     context=True  # Uses output from previous task
... )
>>> cortex = Cortex(
...     agents=[writer, editor],
...     tasks=[write_task, edit_task],
...     llm=llm,
...     process=ProcessType.SEQUENTIAL
... )
>>> result = cortex.kickoff()
>>> print(result.raw_output)
clear_all_memory() → None[source]#

Clear all Cortex memory including short-term, long-term, and entity memories.

Warning

This operation is irreversible. All stored memories will be permanently deleted.

clear_short_term_memory() → None[source]#

Clear the Cortex’s short-term memory while preserving long-term memories.

Useful for starting fresh between sessions while retaining important learned information.

create_ui() → Any[source]#

Create and launch a user interface for the Cortex.

Launches an interactive UI application that allows users to interact with the Cortex through a graphical interface.

Returns

The UI application instance from calute.ui.

classmethod from_task_creator(tasks: list[calute.cortex.orchestration.task.CortexTask], llm: calute.llms.base.BaseLLM | None = None, agents: list[calute.cortex.agents.agent.CortexAgent] | None = None, 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[source]#

Create a Cortex instance from tasks with auto-detected agents and LLM.

Factory method that automatically extracts agents from the provided tasks and infers the LLM from the first task’s agent. Useful when tasks have already been configured with their assigned agents.

Parameters

  • tasks – List of CortexTask instances with agents already assigned.

  • llm – Optional BaseLLM instance. If None, extracted from first task’s agent.

  • agents – Optional agent list. If None, extracted from all tasks.

  • process – Execution strategy to use. Defaults to SEQUENTIAL.

  • manager_agent – Optional agent for hierarchical delegation.

  • memory_type – Type of memory for context management.

  • verbose – Enable detailed logging. Defaults to True.

  • max_iterations – Maximum retry attempts. Defaults to 10.

  • model – Default model identifier. Defaults to “gpt-4”.

  • memory – Pre-configured CortexMemory instance.

  • memory_config – Configuration dict for memory systems.

  • reinvoke_after_function – Reinvoke LLM after tool calls. Defaults to True.

  • enable_calute_memory – Enable Calute’s memory. Defaults to False.

Returns

A configured Cortex instance with name “AutoCortex”.

Example

>>> from calute.cortex import Cortex, CortexTask, TaskCreator
>>> creator = TaskCreator(llm=llm, agents=[agent1, agent2])
>>> tasks = creator.create_tasks("Build a website")
>>> cortex = Cortex.from_task_creator(tasks)
>>> result = cortex.kickoff()
get_memory_summary() → str[source]#

Get a human-readable summary of the Cortex’s memory state.

Returns

A formatted string summarizing short-term, long-term, and entity memories currently stored in the CortexMemory system.

kickoff(inputs: dict[str, Any] | None = None, use_streaming: bool = False, stream_callback: collections.abc.Callable[[Any], None] | None = None, streamer_buffer: calute.core.streamer_buffer.StreamerBuffer | None = None, log_process: bool = False) → calute.cortex.orchestration.cortex.CortexOutput | tuple[calute.core.streamer_buffer.StreamerBuffer, threading.Thread][source]#

Execute the Cortex’s tasks according to the configured process type.

Main entry point for executing all registered tasks. Dispatches to the appropriate execution strategy (sequential, parallel, hierarchical, consensus, or planned) based on self.process. Optionally interpolates template variables and supports streaming execution.

Parameters

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

  • use_streaming – If True, executes tasks in a background thread with real-time streaming output. Defaults to False (blocking execution).

  • stream_callback – Optional callback function invoked with each stream chunk during execution. If log_process is True and no callback is provided, a default logging callback is used.

  • streamer_buffer – Optional pre-existing StreamerBuffer for collecting streaming output. A new buffer is created if None.

  • log_process – If True, enables logging of the execution process via a default stream callback when no custom callback is provided. Defaults to False.

Returns

A CortexOutput containing the final result string, individual task outputs, and total execution time. If use_streaming=True: A tuple of (StreamerBuffer, Thread) for asynchronous consumption of streaming output.

Return type

If use_streaming=False

Raises

  • ValueError – If an unknown process type is configured.

  • Exception – Re-raises any exceptions from task execution.

Side Effects:

  • Interpolates inputs into agents and tasks if inputs is provided.

  • Populates self.task_outputs with results from each task.

  • Saves a cortex-level decision to memory upon completion.

save_memory(persistence_path: str | None = None) → None[source]#

Save the Cortex’s memory to disk for later retrieval.

Parameters

persistence_path – Optional file path for saving. If provided, updates the storage path before saving.

class calute.cortex.orchestration.cortex.CortexOutput(raw_output: str, task_outputs: list[calute.cortex.orchestration.task.CortexTaskOutput], execution_time: float)[source]#

Bases: object

Structured output container from Cortex execution.

Contains the final output from a Cortex execution along with metadata about each individual task’s output, timing information, and agent assignments. Provides convenient methods for accessing and serializing the execution results.

raw_output#

The final output string from the last completed task.

Type

str

task_outputs#

List of CortexTaskOutput instances for each completed task.

Type

list[calute.cortex.orchestration.task.CortexTaskOutput]

execution_time#

Total execution time in seconds for the entire workflow.

Type

float

Example

>>> result = cortex.kickoff()
>>> print(result.raw_output)
>>> print(f"Completed in {result.execution_time:.2f}s")
>>> for task_output in result.task_outputs:
...     print(f"{task_output.agent.role}: {task_output.output[:100]}...")
execution_time: float#
raw_output: str#
task_outputs: list[calute.cortex.orchestration.task.CortexTaskOutput]#
to_dict() → dict[source]#

Convert the output to a serializable dictionary format.

Creates a dictionary representation of the execution results suitable for JSON serialization or API responses.

Returns

Dictionary containing raw_output, task_outputs with task descriptions, outputs, agent roles, timestamps, and the total execution_time.

class calute.cortex.orchestration.cortex.MemoryConfig[source]#

Bases: TypedDict

Configuration dictionary for the Cortex memory system.

This TypedDict defines the configuration options for both CortexMemory and Calute’s internal memory systems. All fields are optional and will use sensible defaults if not specified.

max_short_term#

Maximum entries in Calute’s short-term memory. Defaults to 100.

Type

int

max_working#

Maximum entries in Calute’s working memory. Defaults to 10.

Type

int

max_long_term#

Maximum entries in Calute’s long-term memory. Defaults to 1000.

Type

int

enable_short_term#

Enable CortexMemory short-term storage. Defaults to True.

Type

bool

enable_long_term#

Enable CortexMemory long-term storage. Defaults to True.

Type

bool

enable_entity#

Enable entity tracking in CortexMemory. Defaults to True.

Type

bool

enable_user#

Enable user-specific memory in CortexMemory. Defaults to False.

Type

bool

persistence_path#

File path for persisting memory to disk. Defaults to None.

Type

str | None

short_term_capacity#

Maximum entries in CortexMemory short-term. Defaults to 50.

Type

int

long_term_capacity#

Maximum entries in CortexMemory long-term. Defaults to 5000.

Type

int

Example

>>> config: MemoryConfig = {
...     "enable_short_term": True,
...     "enable_long_term": True,
...     "short_term_capacity": 100,
...     "persistence_path": "./memory.db"
... }
enable_entity: bool#
enable_long_term: bool#
enable_short_term: bool#
enable_user: bool#
long_term_capacity: int#
max_long_term: int#
max_short_term: int#
max_working: int#
persistence_path: str | None#
short_term_capacity: int#
Contents