aixplain.modules.team_agent

__author__

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_id Text, optional - large language model. Defaults to GPT-4o (6646261c6eb563165658bbb1).
• 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.
• use_mentalist bool - Use Mentalist agent for pre-planning. Defaults to True.
• name0 List[Inspector] - List of inspectors that the team agent uses.
• name1 List[InspectorTarget] - List of targets where the inspectors are applied. Defaults to [InspectorTarget.STEPS].

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,
    output_format: Optional[OutputFormat] = None,
    expected_output: Optional[Union[BaseModel, Text, dict]] = None
) -> 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 OutputFormat, optional - response format. If not provided, uses the format set during initialization.
• query2 Union[BaseModel, Text, dict], optional - expected output. Defaults to None.

Returns:

• query3 - 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
) -> 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.

Returns:

• query1 - polling URL in response

delete

def delete() -> None

[view_source]

Delete Corpus service

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, inspectors, 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)
  • inspectors (List[Dict]): Serialized list of inspectors
  • inspectorTargets (List[str]): List of inspector target stages
  • supplier (str): The supplier code
  • version (str): The version number
  • status (str): The current status
  • role (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>)".