Skip to main content

Version: v1.0

aixplain.v1.modules.team_agent

Team Agent module for aiXplain SDK.

This module provides the TeamAgent class and related functionality for creating and managing multi-agent teams that can collaborate on complex tasks.

Copyright 2024 The aiXplain SDK authors

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Author: Lucas Pavanelli and Thiago Castro Ferreira Date: August 15th 2024 Description: Team Agent Class

ContextOverflowStrategy Objects

class ContextOverflowStrategy(str, Enum)

Strategy applied when input messages exceed the model's context window.

Attributes:

TRUNCATE - Remove the oldest chat-history messages until the context fits.
SUMMARIZE - Replace the full chat history with an LLM-generated summary.

InspectorTarget Objects

class InspectorTarget(str, Enum)

Target stages for inspector validation in the team agent pipeline.

This enumeration defines the stages where inspectors can be applied to validate and ensure quality of the team agent's operation.

Attributes:

INPUT - Validates the input data before processing.
STEPS - Validates intermediate steps during processing.
OUTPUT - Validates the final output before returning.

str

def __str__()

Return the string value of the enum member.

Returns:

str - The string value associated with the enum member.

TeamAgent Objects

class TeamAgent(Model, DeployableMixin[Agent])

Advanced AI system capable of using multiple agents to perform a variety of tasks.

Attributes:

id Text - ID of the Team Agent
name Text - Name of the Team Agent
agents List[Agent] - List of agents that the Team Agent uses.
description Text, optional - description of the Team Agent. Defaults to "".
llm Optional[LLM] - Main LLM instance for the team agent.
supervisor_llm Optional[LLM] - Supervisor LLM instance for the team agent.
api_key str - The TEAM API key used for authentication.
supplier Text - Supplier of the Team Agent.
version Text - Version of the Team Agent.
cost Dict, optional - model price. Defaults to None.
status AssetStatus - Status of the Team Agent. Defaults to DRAFT.
instructions Optional[Text] - Instructions to guide the team agent.
output_format OutputFormat - Response format. Defaults to TEXT.
expected_output Optional[Union[BaseModel, Text, dict]] - Expected output format.

Deprecated Attributes:
llm_id Text - DEPRECATED. Use 'llm' parameter instead. Large language model ID.
mentalist_llm Optional[LLM] - DEPRECATED. LLM for planning.
use_mentalist bool - DEPRECATED. Whether to use Mentalist agent for pre-planning.

init

def __init__(id: Text,
             name: Text,
             agents: List[Agent] = [],
             description: Text = "",
             llm: Optional[LLM] = None,
             supervisor_llm: Optional[LLM] = None,
             api_key: Optional[Text] = config.TEAM_API_KEY,
             supplier: Union[Dict, Text, Supplier, int] = "aiXplain",
             version: Optional[Text] = None,
             cost: Optional[Dict] = None,
             status: AssetStatus = AssetStatus.DRAFT,
             instructions: Optional[Text] = None,
             output_format: OutputFormat = OutputFormat.TEXT,
             expected_output: Optional[Union[BaseModel, Text, dict]] = None,
             context_overflow_strategy: Optional[Union[ContextOverflowStrategy,
                                                       Text]] = None,
             **additional_info) -> None

Initialize a TeamAgent instance.

Arguments:

id Text - Unique identifier for the team agent.
name Text - Name of the team agent.
agents List[Agent], optional - List of agents in the team. Defaults to [].
description Text, optional - Description of the team agent. Defaults to "".
llm Optional[LLM], optional - LLM instance. Defaults to None.
supervisor_llm Optional[LLM], optional - Supervisor LLM instance. Defaults to None.
api_key Optional[Text], optional - API key. Defaults to config.TEAM_API_KEY.
supplier Union[Dict, Text, Supplier, int], optional - Supplier. Defaults to "aiXplain".
version Optional[Text], optional - Version. Defaults to None.
cost Optional[Dict], optional - Cost information. Defaults to None.
status AssetStatus, optional - Status of the team agent. Defaults to AssetStatus.DRAFT.
instructions Optional[Text], optional - Instructions for the team agent. Defaults to None.
output_format OutputFormat, optional - Output format. Defaults to OutputFormat.TEXT.
expected_output Optional[Union[BaseModel, Text, dict]], optional - Expected output format. Defaults to None. context_overflow_strategy (Optional[Union[ContextOverflowStrategy, Text]], optional): Strategy for handling context window overflow. Defaults to None (disabled).
**additional_info - Additional keyword arguments.

Deprecated Args:
llm_id Text, optional - DEPRECATED. Use 'llm' parameter instead. ID of the language model. Defaults to "69b7e5f1b2fe44704ab0e7d0".
mentalist_llm Optional[LLM], optional - DEPRECATED. Mentalist/Planner LLM instance. Defaults to None.
use_mentalist bool, optional - DEPRECATED. Whether to use mentalist/planner. Defaults to True.

generate_session_id

def generate_session_id(history: list = None) -> str

Generate a new session ID for the team agent.

Arguments:

history list, optional - Chat history to initialize the session with. Defaults to None.

Returns:

str - The generated session ID in format "{team_agent_id}_{timestamp}".

sync_poll

def sync_poll(poll_url: Text,
              name: Text = "model_process",
              wait_time: float = 0.5,
              timeout: float = 300,
              progress_verbosity: Optional[str] = "compact") -> AgentResponse

Poll the platform until team agent execution completes or times out.

Arguments:

poll_url Text - URL to poll for operation status.
name Text, optional - Identifier for the operation. Defaults to "model_process".
wait_time float, optional - Initial wait time in seconds between polls. Defaults to 0.5.
timeout float, optional - Maximum total time to poll in seconds. Defaults to 300.
progress_verbosity Optional[str], optional - Progress display mode - "full" (detailed), "compact" (brief), or None (no progress). Defaults to "compact".

Returns:

AgentResponse - The final response from the team agent execution.

run

def run(data: Optional[Union[Dict, Text]] = None,
        query: Optional[Text] = None,
        session_id: Optional[Text] = None,
        history: Optional[List[Dict]] = None,
        name: Text = "model_process",
        timeout: float = 300,
        parameters: Dict = {},
        wait_time: float = 0.5,
        content: Optional[Union[Dict[Text, Text], List[Text]]] = None,
        max_tokens: int = 2048,
        max_iterations: int = 30,
        trace_request: bool = False,
        progress_verbosity: Optional[str] = "compact",
        context_overflow_strategy: Optional[Union[ContextOverflowStrategy,
                                                  Text]] = None,
        **kwargs) -> AgentResponse

Runs a team agent call.

Arguments:

data Optional[Union[Dict, Text]], optional - data to be processed by the team agent. Defaults to None.
query Optional[Text], optional - query to be processed by the team agent. Defaults to None.
session_id Optional[Text], optional - conversation Session ID. Defaults to None.
history Optional[List[Dict]], optional - chat history (in case session ID is None). Defaults to None.
name Text, optional - ID given to a call. Defaults to "model_process".
timeout float, optional - total polling time. Defaults to 300.
parameters Dict, optional - optional parameters to the model. Defaults to "{}".
wait_time float, optional - wait time in seconds between polling calls. Defaults to 0.5.
content Union[Dict[Text, Text], List[Text]], optional - Content inputs to be processed according to the query. Defaults to None.
max_tokens int, optional - maximum number of tokens which can be generated by the agents. Defaults to 2048.
max_iterations int, optional - maximum number of iterations between the agents. Defaults to 30.
trace_request bool, optional - return the request id for tracing the request. Defaults to False.
progress_verbosity Optional[str], optional - Progress display mode - "full" (detailed), "compact" (brief), or None (no progress). Defaults to "compact". context_overflow_strategy (Optional[Union[ContextOverflowStrategy, Text]], optional): Strategy for handling context window overflow. Defaults to None.
**kwargs - Additional deprecated keyword arguments (output_format, expected_output).

Returns:

AgentResponse - parsed output from model

run_async

def run_async(
    data: Optional[Union[Dict, Text]] = None,
    query: Optional[Text] = None,
    session_id: Optional[Text] = None,
    history: Optional[List[Dict]] = None,
    name: Text = "model_process",
    parameters: Dict = {},
    content: Optional[Union[Dict[Text, Text], List[Text]]] = None,
    max_tokens: int = 2048,
    max_iterations: int = 30,
    output_format: Optional[OutputFormat] = None,
    expected_output: Optional[Union[BaseModel, Text, dict]] = None,
    evolve: Union[Dict[str, Any], EvolveParam, None] = None,
    trace_request: bool = False,
    context_overflow_strategy: Optional[Union[ContextOverflowStrategy,
                                              Text]] = None
) -> AgentResponse

Runs asynchronously a Team Agent call.

Arguments:

data Optional[Union[Dict, Text]], optional - data to be processed by the Team Agent. Defaults to None.
query Optional[Text], optional - query to be processed by the Team Agent. Defaults to None.
session_id Optional[Text], optional - conversation Session ID. Defaults to None.
history Optional[List[Dict]], optional - chat history (in case session ID is None). Defaults to None.
name Text, optional - ID given to a call. Defaults to "model_process".
parameters Dict, optional - optional parameters to the model. Defaults to "{}".
content Union[Dict[Text, Text], List[Text]], optional - Content inputs to be processed according to the query. Defaults to None.
max_tokens int, optional - maximum number of tokens which can be generated by the agents. Defaults to 2048.
max_iterations int, optional - maximum number of iterations between the agents. Defaults to 30.
output_format OutputFormat, optional - response format. If not provided, uses the format set during initialization.
expected_output Union[BaseModel, Text, dict], optional - expected output. Defaults to None.
evolve Union[Dict[str, Any], EvolveParam, None], optional - evolve the team agent configuration. Can be a dictionary, EvolveParam instance, or None.
trace_request bool, optional - return the request id for tracing the request. Defaults to False. context_overflow_strategy (Optional[Union[ContextOverflowStrategy, Text]], optional): Strategy for handling context window overflow. Defaults to None.

Returns:

AgentResponse - polling URL in response

poll

def poll(poll_url: Text, name: Text = "model_process") -> AgentResponse

Poll once for team agent execution status.

Arguments:

poll_url Text - URL to poll for status.
name Text, optional - Identifier for the operation. Defaults to "model_process".

Returns:

AgentResponse - Response containing status, data, and progress information.

delete

def delete() -> None

Deletes Team Agent.

to_dict

def to_dict() -> Dict

Convert the TeamAgent instance to a dictionary representation.

This method serializes the TeamAgent and all its components (agents, LLMs, etc.) into a dictionary format suitable for storage or transmission.

Returns:

Dict - A dictionary containing:
- id (str): The team agent's ID
- name (str): The team agent's name
- agents (List[Dict]): Serialized list of agents
- links (List): Empty list (reserved for future use)
- description (str): The team agent's description
- llmId (str): ID of the main language model
- supervisorId (str): ID of the supervisor language model
- plannerId (str): ID of the planner model (if use_mentalist)
- supplier (str): The supplier code
- version (str): The version number
- status (str): The current status
- instructions (str): The team agent's instructions

from_dict

@classmethod
def from_dict(cls, data: Dict) -> "TeamAgent"

Create a TeamAgent instance from a dictionary representation.

Arguments:

data - Dictionary containing TeamAgent parameters

Returns:

TeamAgent instance

validate

def validate(raise_exception: bool = False) -> bool

Validate the TeamAgent configuration.

This method checks the validity of the TeamAgent's configuration, including name format, LLM compatibility, and agent validity.

Arguments:

raise_exception bool, optional - If True, raises exceptions for validation failures. If False, logs warnings. Defaults to False.

Returns:

bool - True if validation succeeds, False otherwise.

Raises:

Exception - If raise_exception is True and validation fails, with details about the specific validation error.

Notes:

The team agent cannot be run until all validation issues are fixed
Name must contain only alphanumeric chars, spaces, hyphens, brackets
LLM must be a text generation model
All agents must pass their own validation

update

def update() -> None

Update the TeamAgent in the backend.

This method validates and updates the TeamAgent's configuration in the backend system. It is deprecated in favor of the save() method.

Raises:

Exception - If validation fails or if the update request fails. Specific error messages will indicate:
- Validation failures with details
- HTTP errors with status codes
- General update errors requiring admin attention

Notes:

This method is deprecated, use save() instead
Performs validation before attempting update
Requires valid team API key for authentication
Returns a new TeamAgent instance if successful

save

def save() -> None

Save the Agent.

repr

def __repr__()

Return a string representation of the TeamAgent.

Returns:

str - A string in the format "TeamAgent: <name> (id=<id>)".

evolve_async

def evolve_async(evolve_type: Union[EvolveType, str] = EvolveType.TEAM_TUNING,
                 max_successful_generations: int = 3,
                 max_failed_generation_retries: int = 3,
                 max_iterations: int = 50,
                 max_non_improving_generations: Optional[int] = 2,
                 llm: Optional[Union[Text, LLM]] = None) -> AgentResponse

Asynchronously evolve the Team Agent and return a polling URL in the AgentResponse.

Arguments:

evolve_type Union[EvolveType, str] - Type of evolution (TEAM_TUNING or INSTRUCTION_TUNING). Defaults to TEAM_TUNING.
max_successful_generations int - Maximum number of successful generations to evolve. Defaults to 3.
max_failed_generation_retries int - Maximum retry attempts for failed generations. Defaults to 3.
max_iterations int - Maximum number of iterations. Defaults to 50.
max_non_improving_generations Optional[int] - Stop condition parameter for non-improving generations. Defaults to 2, can be None.
llm Optional[Union[Text, LLM]] - LLM to use for evolution. Can be an LLM ID string or LLM object. Defaults to None.

Returns:

AgentResponse - Response containing polling URL and status.

evolve

def evolve(evolve_type: Union[EvolveType, str] = EvolveType.TEAM_TUNING,
           max_successful_generations: int = 3,
           max_failed_generation_retries: int = 3,
           max_iterations: int = 50,
           max_non_improving_generations: Optional[int] = 2,
           llm: Optional[Union[Text, LLM]] = None) -> AgentResponse

Synchronously evolve the Team Agent and poll for the result.

Arguments:

evolve_type Union[EvolveType, str] - Type of evolution (TEAM_TUNING or INSTRUCTION_TUNING). Defaults to TEAM_TUNING.
max_successful_generations int - Maximum number of successful generations to evolve. Defaults to 3.
max_failed_generation_retries int - Maximum retry attempts for failed generations. Defaults to 3.
max_iterations int - Maximum number of iterations. Defaults to 50.
max_non_improving_generations Optional[int] - Stop condition parameter for non-improving generations. Defaults to 2, can be None.
llm Optional[Union[Text, LLM]] - LLM to use for evolution. Can be an LLM ID string or LLM object. Defaults to None.

Returns:

AgentResponse - Final response from the evolution process.

ContextOverflowStrategy Objects
InspectorTarget Objects
TeamAgent Objects