Version: 1.0

aixplain.modules.agent

Agent module for aiXplain SDK.

This module provides the Agent class and related functionality for creating and managing AI agents that can execute tasks using various tools and models.

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: May 16th 2024 Description: Agentification Class

Agent Objects

class Agent(Model, DeployableMixin[Union[Tool, DeployableTool]])

[view_source]

An advanced AI system that performs tasks using specialized tools from the aiXplain marketplace.

This class represents an AI agent that can understand natural language instructions, use various tools and models, and execute complex tasks. It combines a large language model (LLM) with specialized tools to provide comprehensive task-solving capabilities.

Attributes:

id Text - ID of the Agent.
name Text - Name of the Agent.
tools List[Union[Tool, Model]] - Collection of tools and models the Agent can use.
description Text, optional - Detailed description of the Agent's capabilities. Defaults to "".
instructions Text - System instructions/prompt defining the Agent's behavior.
llm_id Text - ID of the large language model. Defaults to GPT-4o (6646261c6eb563165658bbb1).
llm Optional[LLM] - The LLM instance used by the Agent.
supplier Text - The provider/creator of the Agent.
version Text - Version identifier of the Agent.
status AssetStatus - Current status of the Agent (DRAFT or ONBOARDED).
name0 List[AgentTask] - List of tasks the Agent can perform.
name1 str - URL endpoint for the backend API.
name2 str - Authentication key for API access.
name3 Dict, optional - Pricing information for using the Agent. Defaults to None.
name4 bool - Whether the Agent's configuration is valid.
name3 Dict, optional - model price. Defaults to None.
name6 OutputFormat - default output format for agent responses.
name7 Union[BaseModel, Text, dict], optional - expected output. Defaults to None.

init

def __init__(id: Text,
             name: Text,
             description: Text,
             instructions: Optional[Text] = None,
             tools: List[Union[Tool, Model]] = [],
             llm_id: Text = "6646261c6eb563165658bbb1",
             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,
             tasks: List[AgentTask] = [],
             workflow_tasks: List[WorkflowTask] = [],
             output_format: OutputFormat = OutputFormat.TEXT,
             expected_output: Optional[Union[BaseModel, Text, dict]] = None,
             **additional_info) -> None

[view_source]

Initialize a new Agent instance.

Arguments:

id Text - ID of the Agent.
name Text - Name of the Agent.
description Text - Detailed description of the Agent's capabilities.
instructions Optional[Text], optional - System instructions/prompt defining the Agent's behavior. Defaults to None.
tools List[Union[Tool, Model]], optional - Collection of tools and models the Agent can use. Defaults to empty list.
llm_id Text, optional - ID of the large language model. Defaults to GPT-4o (6646261c6eb563165658bbb1).
llm Optional[LLM], optional - The LLM instance to use. If provided, takes precedence over llm_id. Defaults to None.
api_key Optional[Text], optional - Authentication key for API access. Defaults to config.TEAM_API_KEY.
supplier Union[Dict, Text, Supplier, int], optional - The provider/creator of the Agent. Defaults to "aiXplain".
version Optional[Text], optional - Version identifier. Defaults to None.
name0 Optional[Dict], optional - Pricing information. Defaults to None.
name1 AssetStatus, optional - Current status of the Agent. Defaults to AssetStatus.DRAFT.
name2 List[AgentTask], optional - List of tasks the Agent can perform. Defaults to empty list.
name3 List[WorkflowTask], optional - List of workflow tasks the Agent can execute. Defaults to empty list.
name4 OutputFormat, optional - default output format for agent responses. Defaults to OutputFormat.TEXT.
name5 Union[BaseModel, Text, dict], optional - expected output. Defaults to None.
name6 - Additional configuration parameters.

validate

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

[view_source]

Validate the Agent's configuration and mark its validity status.

This method runs all validation checks and updates the is_valid flag. If validation fails, it can either raise an exception or log warnings.

Arguments:

raise_exception bool, optional - Whether to raise exceptions on validation failures. If False, failures are logged as warnings. Defaults to False.

Returns:

bool - True if validation passed, False otherwise.

Raises:

Exception - If validation fails and raise_exception is True.

generate_session_id

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

[view_source]

Generate a unique session ID for agent conversations.

Arguments:

history list, optional - Previous conversation history. Defaults to None.

Returns:

str - A unique session identifier based on timestamp and random components.

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 = 4096,
        max_iterations: int = 5,
        output_format: Optional[OutputFormat] = None,
        expected_output: Optional[Union[BaseModel, Text, dict]] = None,
        trace_request: bool = False) -> AgentResponse

[view_source]

Runs an agent call.

Arguments:

data Optional[Union[Dict, Text]], optional - data to be processed by the agent. Defaults to None.
query Optional[Text], optional - query to be processed by the 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 agent. Defaults to 2048.
query0 int, optional - maximum number of iterations between the agent and the tools. Defaults to 10.
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.
query3 bool, optional - return the request id for tracing the request. Defaults to False.

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 = 5,
              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 an agent call.

Arguments:

data Optional[Union[Dict, Text]], optional - data to be processed by the agent. Defaults to None.
query Optional[Text], optional - query to be processed by the 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 agent. Defaults to 2048.
max_iterations int, optional - maximum number of iterations between the agent and the tools. Defaults to 10.
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.
output_format ResponseFormat, optional - response format. Defaults to TEXT.
query2 Union[Dict[str, Any], EvolveParam, None], optional - evolve the agent configuration. Can be a dictionary, EvolveParam instance, or None.
query3 bool, optional - return the request id for tracing the request. Defaults to False.

Returns:

query4 - polling URL in response

to_dict

def to_dict() -> Dict

[view_source]

Convert the Agent instance to a dictionary representation.

Returns:

Dict - Dictionary containing the agent's configuration and metadata.

from_dict

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

[view_source]

Create an Agent instance from a dictionary representation.

Arguments:

data - Dictionary containing Agent parameters

Returns:

Agent instance

delete

def delete() -> None

[view_source]

Delete this Agent from the aiXplain platform.

This method attempts to delete the Agent. The operation will fail if the Agent is being used by any team agents.

Raises:

Exception - If deletion fails, with detailed error messages for different failure scenarios:
- Agent is in use by accessible team agents (lists team agent IDs)
- Agent is in use by inaccessible team agents
- Other deletion errors (with HTTP status code)

update

def update() -> None

[view_source]

Update the Agent's configuration on the aiXplain platform.

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

Raises:

Exception - If validation fails or if there are errors during the update.
DeprecationWarning - This method is deprecated, use save() instead.

Notes:

This method is deprecated and will be removed in a future version. Please use save() instead.

save

def save() -> None

[view_source]

Save the Agent's current configuration to the aiXplain platform.

This method validates and saves any changes made to the Agent's configuration. It is the preferred method for updating an Agent's settings.

Raises:

Exception - If validation fails or if there are errors during the save operation.

repr

def __repr__() -> str

[view_source]

Return a string representation of the Agent.

Returns:

str - A string in the format "Agent: <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 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 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.

Agent Objects​

__init__​

validate​

generate_session_id​

run​

run_async​

to_dict​

from_dict​

delete​

update​

save​

__repr__​

evolve_async​

evolve​

Agent Objects

init

validate

generate_session_id

run

run_async

to_dict

from_dict

delete

update

save

repr

evolve_async

evolve