Camel.societies.workforce.workforce

Documentation Index

Fetch the complete documentation index at: https://docs.camel-ai.org/llms.txt

Use this file to discover all available pages before exploring further.

​

WorkforceState

class WorkforceState(Enum):

​

WorkforceMode

class WorkforceMode(Enum):

​

WorkforceSnapshot

class WorkforceSnapshot:

​

init

def __init__(
    self,
    main_task: Optional[Task] = None,
    pending_tasks: Optional[Deque[Task]] = None,
    completed_tasks: Optional[List[Task]] = None,
    task_dependencies: Optional[Dict[str, List[str]]] = None,
    assignees: Optional[Dict[str, str]] = None,
    current_task_index: int = 0,
    description: str = ''
):

​

Workforce

class Workforce(BaseNode):

• Coordinator Agent: Assigns tasks to workers based on their capabilities
• Task Planner Agent: Decomposes complex tasks and composes results
• Dynamic Workers: Created at runtime when tasks fail repeatedly

• description (str): Description of the workforce.
• children (Optional[List[BaseNode]], optional): List of child nodes under this node. Each child node can be a worker node or another workforce node. (default: :obj:None)
• coordinator_agent (Optional[ChatAgent], optional): A custom coordinator agent instance for task assignment and worker creation. If provided, the workforce will create a new agent using this agent’s model configuration but with the required system message and functionality. If None, a default agent will be created using DEFAULT model settings. (default: :obj:None)
• task_agent (Optional[ChatAgent], optional): A custom task planning agent instance for task decomposition and composition. If provided, the workforce will create a new agent using this agent’s model configuration but with the required system message. If None, a default agent will be created using DEFAULT model settings. (default: :obj:None)
• new_worker_agent (Optional[ChatAgent], optional): A template agent for workers created dynamically at runtime when existing workers cannot handle failed tasks. If None, workers will be created with default settings including SearchToolkit, CodeExecutionToolkit, and ThinkingToolkit. (default: :obj:None) default_model (Optional[Union[BaseModelBackend, ModelManager]], optional): Model backend or manager to use when creating default coordinator, task, or dynamic worker agents. If None, agents will be created using ModelPlatformType.DEFAULT and ModelType.DEFAULT settings. (default: :obj:None)
• graceful_shutdown_timeout (float, optional): The timeout in seconds for graceful shutdown when a task fails 3 times. During this period, the workforce remains active for debugging. Set to 0 for immediate shutdown. (default: :obj:15.0)
• task_timeout_seconds (Optional[float], optional): The timeout in seconds for waiting for tasks to be returned by workers. If None, uses the global TASK_TIMEOUT_SECONDS value (600.0 seconds). Increase this value for tasks that require more processing time. (default: :obj:None)
• share_memory (bool, optional): Whether to enable shared memory across SingleAgentWorker instances in the workforce. When enabled, all SingleAgentWorker instances, coordinator agent, and task planning agent will share their complete conversation history and function-calling trajectory, providing better context for task handoffs and continuity. Note: Currently only supports SingleAgentWorker instances; RolePlayingWorker and nested Workforce instances do not participate in memory sharing. (default: :obj:False)
• use_structured_output_handler (bool, optional): Whether to use the structured output handler instead of native structured output. When enabled, the workforce will use prompts with structured output instructions and regex extraction to parse responses. This ensures compatibility with agents that don’t reliably support native structured output. When disabled, the workforce uses the native response_format parameter. (default: :obj:True)
• callbacks (Optional[List[WorkforceCallback]], optional): A list of callback handlers to observe and record workforce lifecycle events and metrics (e.g., task creation/assignment/start/completion/ failure, worker creation/deletion, all-tasks-completed). All items must be instances of :class:WorkforceCallback, otherwise
• a: class:ValueError is raised. If none of the provided callbacks implement :class:WorkforceMetrics, a built-in :class:WorkforceLogger (implements both callback and metrics) is added automatically. If at least one provided callback
• implements: class:WorkforceMetrics, no default logger is added. (default: :obj:None)
• mode (WorkforceMode, optional): The execution mode for task processing. AUTO_DECOMPOSE mode uses intelligent recovery strategies (decompose, replan, etc.) when tasks fail. PIPELINE mode uses simple retry logic and allows failed tasks to continue the workflow, passing error information to dependent tasks. (default: :obj:WorkforceMode.AUTO_DECOMPOSE)
• failure_handling_config (Optional[Union[FailureHandlingConfig, Dict]]): Configuration for customizing failure handling behavior. Can be a FailureHandlingConfig instance or a dict with the same fields. Allows fine-grained control over which recovery strategies are enabled, maximum retry attempts, and whether to halt on max retries. The enabled_strategies field accepts both enum values and string lists like ["retry", "replan"]. If None, uses default configuration with all strategies enabled. (default: :obj:None)

​

init

def __init__(
    self,
    description: str,
    children: Optional[List[BaseNode]] = None,
    coordinator_agent: Optional[ChatAgent] = None,
    task_agent: Optional[ChatAgent] = None,
    new_worker_agent: Optional[ChatAgent] = None,
    default_model: Optional[Union[BaseModelBackend, ModelManager]] = None,
    graceful_shutdown_timeout: float = 15.0,
    share_memory: bool = False,
    use_structured_output_handler: bool = True,
    task_timeout_seconds: Optional[float] = None,
    mode: WorkforceMode = WorkforceMode.AUTO_DECOMPOSE,
    callbacks: Optional[List[WorkforceCallback]] = None,
    failure_handling_config: Optional[Union[FailureHandlingConfig, Dict[str, Any]]] = None
):

​

_initialize_callbacks

def _initialize_callbacks(self, callbacks: Optional[List[WorkforceCallback]]):

​

_notify_worker_created

def _notify_worker_created(self, worker_node: BaseNode):

​

_get_or_create_shared_context_utility

def _get_or_create_shared_context_utility(self, session_id: Optional[str] = None):

• session_id (Optional[str]): Custom session ID to use. If None, auto-generates a timestamped session ID. (default: :obj:None)

​

_get_role_identifier

def _get_role_identifier(
    self,
    worker: ChatAgent,
    description: str,
    workflow_summary: Optional['WorkflowSummary'] = None
):

• worker (ChatAgent): The worker agent to extract role from.
• description (str): Worker description to use as fallback.
• workflow_summary (Optional[WorkflowSummary]): Optional WorkflowSummary object that may contain agent_title field.

​

_validate_agent_compatibility

def _validate_agent_compatibility(self, agent: ChatAgent, agent_context: str = 'agent'):

• agent (ChatAgent): The agent to validate.
• agent_context (str): Context description for error messages.

​

_attach_pause_event_to_agent

def _attach_pause_event_to_agent(self, agent: ChatAgent):

​

_ensure_pause_event_in_kwargs

def _ensure_pause_event_in_kwargs(self, kwargs: Optional[Dict]):

​

repr

def __repr__(self):

​

set_mode

def set_mode(self, mode: WorkforceMode):

• mode (WorkforceMode): The desired execution mode. - AUTO_DECOMPOSE: Intelligent task decomposition with recovery - PIPELINE: Predefined task pipeline with simple retry logic

​

_ensure_pipeline_builder

def _ensure_pipeline_builder(self):

​

pipeline_add

def pipeline_add(
    self,
    content: Union[str, Task],
    task_id: Optional[str] = None,
    dependencies: Optional[List[str]] = None,
    additional_info: Optional[Dict[str, Any]] = None,
    auto_depend: bool = True
):

• content (Union[str, Task]): The task content string or a Task object. If a Task object is provided, task_id and additional_info parameters are ignored.
• task_id (str, optional): Unique identifier for the task. If None, a unique ID will be generated. Only used when content is a string. (default: :obj:None)
• dependencies (List[str], optional): List of task IDs that this task depends on. If None and auto_depend=True, will depend on the last added task. (default: :obj:None)
• additional_info (Dict[str, Any], optional): Additional information for the task. Only used when content is a string. (default: :obj:None)
• auto_depend (bool, optional): If True and dependencies is None, automatically depend on the last added task. (default: :obj:True)

​

add_parallel_pipeline_tasks

def add_parallel_pipeline_tasks(
    self,
    task_contents: Union[List[str], List[Task]],
    dependencies: Optional[List[str]] = None,
    task_id_prefix: str = 'parallel',
    auto_depend: bool = True
):

• task_contents (Union[List[str], List[Task]]): List of task content strings or Task objects. If Task objects are provided, task_id_prefix is ignored.
• dependencies (List[str], optional): Common dependencies for all parallel tasks. (default: :obj:None)
• task_id_prefix (str, optional): Prefix for generated task IDs. Only used when task_contents contains strings. (default: :obj:"parallel")
• auto_depend (bool, optional): If True and dependencies is None, automatically depend on the last added task. (default: :obj:True)

​

add_sync_pipeline_task

def add_sync_pipeline_task(
    self,
    content: Union[str, Task],
    wait_for: Optional[List[str]] = None,
    task_id: Optional[str] = None
):

• content (Union[str, Task]): Content of the synchronization task or a Task object. If a Task object is provided, task_id parameter is ignored.
• wait_for (List[str], optional): List of task IDs to wait for. If None, will automatically wait for the last parallel tasks. (default: :obj:None)
• task_id (str, optional): ID for the sync task. Only used when content is a string. (default: :obj:None)

​

pipeline_fork

def pipeline_fork(self, task_contents: Union[List[str], List[Task]]):

• task_contents (Union[List[str], List[Task]]): List of task content strings or Task objects for parallel execution.

​

pipeline_join

def pipeline_join(self, content: Union[str, Task], task_id: Optional[str] = None):

• content (Union[str, Task]): Content of the join/sync task or a Task object. If a Task object is provided, task_id parameter is ignored.
• task_id (str, optional): ID for the sync task. Only used when content is a string. (default: :obj:None)

​

pipeline_build

def pipeline_build(self):

​

get_pipeline_builder

def get_pipeline_builder(self):

​

set_pipeline_tasks

def set_pipeline_tasks(self, tasks: List[Task]):

• tasks (List[Task]): List of tasks with dependencies already set. The dependencies should be Task objects in the Task.dependencies attribute.

​

_collect_shared_memory

def _collect_shared_memory(self):

​

_share_memory_with_agents

def _share_memory_with_agents(self, shared_memory: Dict[str, List]):

• shared_memory (Dict[str, List]): Memory records collected from all agents to be shared.

​

_sync_shared_memory

def _sync_shared_memory(self):

​

_update_dependencies_for_decomposition

def _update_dependencies_for_decomposition(self, original_task: Task, subtasks: List[Task]):

​

_increment_in_flight_tasks

def _increment_in_flight_tasks(self, task_id: str):

​

_decrement_in_flight_tasks

def _decrement_in_flight_tasks(self, task_id: str, context: str = ''):

​

_cleanup_task_tracking

def _cleanup_task_tracking(self, task_id: str):

• task_id (str): The ID of the task to clean up.

​

_decompose_task

def _decompose_task(
    self,
    task: Task,
    stream_callback: Optional[Callable[['ChatAgentResponse'], None]] = None
):

• task (Task): The task to decompose.
• stream_callback (Callable[[ChatAgentResponse], None], optional): A callback function that receives each chunk (ChatAgentResponse) during streaming decomposition.

​

_get_available_strategies_text

def _get_available_strategies_text(self):

​

_analyze_task

def _analyze_task(self, task: Task):

• task (Task): The task to analyze
• for_failure (bool): True for failure analysis, False for quality evaluation
• error_message (Optional[str]): Error message, required when for_failure=True

​

pause

def pause(self):

​

resume

def resume(self):

​

stop_gracefully

def stop_gracefully(self):

​

stop_immediately

def stop_immediately(self):

​

skip_gracefully

def skip_gracefully(self):

​

save_snapshot

def save_snapshot(self, description: str = ''):

​

list_snapshots

def list_snapshots(self):

​

get_pending_tasks

def get_pending_tasks(self):

​

get_completed_tasks

def get_completed_tasks(self):

​

modify_task_content

def modify_task_content(self, task_id: str, new_content: str):

​

get_main_task_queue

def get_main_task_queue(self):

​

add_task

def add_task(
    self,
    content: str,
    task_id: Optional[str] = None,
    additional_info: Optional[Dict[str, Any]] = None,
    as_subtask: bool = False,
    insert_position: int = -1
):

• content (str): The content of the task.
• task_id (Optional[str], optional): Optional ID for the task. If not provided, a unique ID will be generated.
• additional_info (Optional[Dict[str, Any]], optional): Optional additional metadata for the task.
• as_subtask (bool, optional): If True, adds the task directly to the pending subtask queue. If False, adds as a main task that will be decomposed. Defaults to False.
• insert_position (int, optional): Position to insert the task in the pending queue. Only applies when as_subtask=True. Defaults to -1 (append to end).

​

add_main_task

def add_main_task(
    self,
    content: str,
    task_id: Optional[str] = None,
    additional_info: Optional[Dict[str, Any]] = None
):

• content (str): The content of the main task.
• task_id (Optional[str], optional): Optional ID for the task.
• additional_info (Optional[Dict[str, Any]], optional): Optional additional metadata.

​

add_subtask

def add_subtask(
    self,
    content: str,
    task_id: Optional[str] = None,
    additional_info: Optional[Dict[str, Any]] = None,
    insert_position: int = -1
):

• content (str): The content of the subtask.
• task_id (Optional[str], optional): Optional ID for the task.
• additional_info (Optional[Dict[str, Any]], optional): Optional additional metadata.
• insert_position (int, optional): Position to insert the task. Defaults to -1 (append to end).

​

remove_task

def remove_task(self, task_id: str):

• task_id (str): The ID of the task to remove.

​

reorder_tasks

def reorder_tasks(self, task_ids: List[str]):

​

resume_from_task

def resume_from_task(self, task_id: str):

​

restore_from_snapshot

def restore_from_snapshot(self, snapshot_index: int):

​

get_workforce_status

def get_workforce_status(self):

​

_collect_pipeline_results

def _collect_pipeline_results(self):

​

_all_pipeline_tasks_successful

def _all_pipeline_tasks_successful(self):

​

process_task

def process_task(self, task: Task):

• task (Task): The task to be processed.

​

_process_task_with_intervention

def _process_task_with_intervention(self, task: Task):

• task (Task): The task to be processed.

​

continue_from_pause

def continue_from_pause(self):

​

_start_child_node_when_paused

def _start_child_node_when_paused(self, start_coroutine: Coroutine):

• start_coroutine: The coroutine to start (e.g., worker_node.start())

​

add_single_agent_worker

def add_single_agent_worker(
    self,
    description: str,
    worker: ChatAgent,
    pool_max_size: int = DEFAULT_WORKER_POOL_SIZE,
    enable_workflow_memory: bool = False
):

• description (str): Description of the worker node.
• worker (ChatAgent): The agent to be added.
• pool_max_size (int): Maximum size of the agent pool. (default: :obj:10)
• enable_workflow_memory (bool): Whether to enable workflow memory accumulation. Set to True if you plan to call save_workflow_memories(). (default: :obj:False)

​

add_role_playing_worker

def add_role_playing_worker(
    self,
    description: str,
    assistant_role_name: str,
    user_role_name: str,
    assistant_agent_kwargs: Optional[Dict] = None,
    user_agent_kwargs: Optional[Dict] = None,
    summarize_agent_kwargs: Optional[Dict] = None,
    chat_turn_limit: int = 3
):

• description (str): Description of the node.
• assistant_role_name (str): The role name of the assistant agent.
• user_role_name (str): The role name of the user agent.
• assistant_agent_kwargs (Optional[Dict]): The keyword arguments to initialize the assistant agent in the role playing, like the model name, etc. (default: :obj:None)
• user_agent_kwargs (Optional[Dict]): The keyword arguments to initialize the user agent in the role playing, like the model name, etc. (default: :obj:None)
• summarize_agent_kwargs (Optional[Dict]): The keyword arguments to initialize the summarize agent, like the model name, etc. (default: :obj:None)
• chat_turn_limit (int): The maximum number of chat turns in the role playing. (default: :obj:3)

​

add_workforce

def add_workforce(self, workforce: Workforce):

• workforce (Workforce): The workforce node to be added.

​

reset

def reset(self):

​

save_workflow_memories

def save_workflow_memories(self):

​

load_workflow_memories

def load_workflow_memories(
    self,
    session_id: Optional[str] = None,
    worker_max_workflows: int = 3,
    coordinator_max_workflows: int = 5,
    task_agent_max_workflows: int = 3
):

• session_id (Optional[str]): Specific workforce session ID to load from. If None, searches across all sessions. (default: :obj:None)
• worker_max_workflows (int): Maximum number of workflow files to load per worker agent. (default: :obj:3)
• coordinator_max_workflows (int): Maximum number of workflow files to load for the coordinator agent. (default: :obj:5)
• task_agent_max_workflows (int): Maximum number of workflow files to load for the task planning agent. (default: :obj:3)

​

_load_management_agent_workflows

def _load_management_agent_workflows(
    self,
    coordinator_max_workflows: int,
    task_agent_max_workflows: int,
    session_id: Optional[str] = None
):

• Coordinator agent: understand task assignment patterns and worker capabilities
• Task agent: understand task decomposition patterns and successful strategies

• coordinator_max_workflows (int): Maximum number of workflow files to load for the coordinator agent.
• task_agent_max_workflows (int): Maximum number of workflow files to load for the task planning agent.
• session_id (Optional[str]): Specific session ID to load from. If None, searches across all sessions.

​

set_channel

def set_channel(self, channel: TaskChannel):

​

_get_child_nodes_info

def _get_child_nodes_info(self):

​

_get_node_info

def _get_node_info(self, node):

​

_get_single_agent_toolkit_info

def _get_single_agent_toolkit_info(self, worker: 'SingleAgentWorker'):

​

_group_tools_by_toolkit

def _group_tools_by_toolkit(self, tool_dict: dict):

​

_get_valid_worker_ids

def _get_valid_worker_ids(self):

​

_call_coordinator_for_assignment

def _call_coordinator_for_assignment(self, tasks: List[Task], invalid_ids: Optional[List[str]] = None):

• tasks (List[Task]): Tasks to assign.
• invalid_ids (List[str], optional): Invalid worker IDs from previous attempt (if any).

​

_validate_assignments

def _validate_assignments(self, assignments: List[TaskAssignment], valid_ids: Set[str]):

• assignments (List[TaskAssignment]): Assignments to validate.
• valid_ids (Set[str]): Set of valid worker IDs.

​

_update_task_dependencies_from_assignments

def _update_task_dependencies_from_assignments(self, assignments: List[TaskAssignment], tasks: List[Task]):

• assignments (List[TaskAssignment]): The task assignments containing dependency IDs.
• tasks (List[Task]): The tasks that were assigned.

​

get_workforce_log_tree

def get_workforce_log_tree(self):

​

get_workforce_kpis

def get_workforce_kpis(self):

​

dump_workforce_logs

def dump_workforce_logs(self, file_path: str):

• file_path (str): The path to the JSON file.

​

_submit_coro_to_loop

def _submit_coro_to_loop(self, coro: 'Coroutine'):

​

stop

def stop(self):

​

clone

def clone(self, with_memory: bool = False):

• with_memory (bool, optional): Whether to copy the memory (conversation history) to the new instance. If True, the new instance will have the same conversation history. If False, the new instance will have a fresh memory. (default: :obj:False)

​

to_mcp

def to_mcp(
    self,
    name: str = 'CAMEL-Workforce',
    description: str = 'A workforce system using the CAMEL AI framework for multi-agent collaboration.',
    dependencies: Optional[List[str]] = None,
    host: str = 'localhost',
    port: int = 8001
):

• name (str): Name of the MCP server. (default: :obj:CAMEL-Workforce)
• description (str): Description of the workforce. If None, a generic description is used. (default: :obj:A workforce system using the CAMEL AI framework for multi-agent collaboration.)
• dependencies (Optional[List[str]]): Additional dependencies for the MCP server. (default: :obj:None)
• host (str): Host to bind to for HTTP transport. (default: :obj:localhost)
• port (int): Port to bind to for HTTP transport. (default: :obj:8001)