Skip to main content
Version: 1.0

aixplain.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

InspectorTarget Objects​

class InspectorTarget(str, Enum)

[view_source]

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__()

[view_source]

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])

[view_source]

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.

  • name0 AssetStatus - Status of the Team Agent. Defaults to DRAFT.

  • name1 Optional[Text] - Instructions to guide the team agent.

  • name2 OutputFormat - Response format. Defaults to TEXT.

  • name3 Optional[Union[BaseModel, Text, dict]] - Expected output format.

    Deprecated Attributes:

  • name4 Text - DEPRECATED. Use 'llm' parameter instead. Large language model ID.

  • name5 Optional[LLM] - DEPRECATED. LLM for planning.

  • name6 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,
             **additional_info) -> None

[view_source]

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.

  • name0 AssetStatus, optional - Status of the team agent. Defaults to AssetStatus.DRAFT.

  • name1 Optional[Text], optional - Instructions for the team agent. Defaults to None.

  • name2 OutputFormat, optional - Output format. Defaults to OutputFormat.TEXT.

  • name3 Optional[Union[BaseModel, Text, dict]], optional - Expected output format. Defaults to None.

  • name4 - Additional keyword arguments.

    Deprecated Args:

  • name5 Text, optional - DEPRECATED. Use 'llm' parameter instead. ID of the language model. Defaults to "6646261c6eb563165658bbb1".

  • name6 Optional[LLM], optional - DEPRECATED. Mentalist/Planner LLM instance. Defaults to None.

  • name7 bool, optional - DEPRECATED. Whether to use mentalist/planner. Defaults to True.

generate_session_id​

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

[view_source]

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

[view_source]

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",
        **kwargs) -> AgentResponse

[view_source]

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.
  • query0 int, optional - maximum number of iterations between the agents. Defaults to 30.
  • query1 bool, optional - return the request id for tracing the request. Defaults to False.
  • query2 Optional[str], optional - Progress display mode - "full" (detailed), "compact" (brief), or None (no progress). Defaults to "compact".
  • query3 - Additional deprecated keyword arguments (output_format, expected_output).

Returns:

  • query4 - 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) -> AgentResponse

[view_source]

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.
  • query0 Union[BaseModel, Text, dict], optional - expected output. Defaults to None.
  • query1 Union[Dict[str, Any], EvolveParam, None], optional - evolve the team agent configuration. Can be a dictionary, EvolveParam instance, or None.
  • query2 bool, optional - return the request id for tracing the request. Defaults to False.

Returns:

  • query3 - polling URL in response

poll​

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

[view_source]

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

[view_source]

Deletes Team Agent.

to_dict​

def to_dict() -> Dict

[view_source]

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"

[view_source]

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

[view_source]

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

[view_source]

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

[view_source]

Save the Agent.

__repr__​

def __repr__()

[view_source]

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

[view_source]

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

[view_source]

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.