Skip to main content

Version: v1.0

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

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

Agent Objects

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

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-5.4 (69b7e5f1b2fe44704ab0e7d0).
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).
tasks List[AgentTask] - List of tasks the Agent can perform.
backend_url str - URL endpoint for the backend API.
api_key str - Authentication key for API access.
cost Dict, optional - Pricing information for using the Agent. Defaults to None.
is_valid bool - Whether the Agent's configuration is valid.
output_format OutputFormat - default output format for agent responses.
expected_output 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 = "69b7e5f1b2fe44704ab0e7d0",
             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

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-5.4 (69b7e5f1b2fe44704ab0e7d0).
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.
cost Optional[Dict], optional - Pricing information. Defaults to None.
status AssetStatus, optional - Current status of the Agent. Defaults to AssetStatus.DRAFT.
tasks List[AgentTask], optional - List of tasks the Agent can perform. Defaults to empty list.
workflow_tasks List[WorkflowTask], optional - List of workflow tasks the Agent can execute. Defaults to empty list.
output_format OutputFormat, optional - default output format for agent responses. Defaults to OutputFormat.TEXT.
expected_output Union[BaseModel, Text, dict], optional - expected output. Defaults to None.
**additional_info - Additional configuration parameters.

validate

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

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

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.

poll

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

Override poll to normalize progress data from camelCase to snake_case.

Arguments:

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

Returns:

AgentResponse - Response with normalized progress data.

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 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 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 = 4096,
        max_iterations: int = 5,
        trace_request: bool = False,
        progress_verbosity: Optional[str] = "compact",
        run_response_generation: bool = False,
        **kwargs) -> AgentResponse

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.
max_iterations int, optional - maximum number of iterations between the agent and the tools. Defaults to 10.
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".
run_response_generation bool, optional - Whether to run response generation. Defaults to False.
**kwargs - Additional keyword arguments.

Returns:

Dict - 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,
              run_response_generation: bool = False) -> AgentResponse

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.
expected_output Union[BaseModel, Text, dict], optional - expected output. Defaults to None.
evolve Union[Dict[str, Any], EvolveParam, None], optional - evolve the 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.
run_response_generation bool, optional - Whether to run response generation. Defaults to False.

Returns:

dict - polling URL in response

to_dict

def to_dict() -> Dict

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"

Create an Agent instance from a dictionary representation.

Arguments:

data - Dictionary containing Agent parameters

Returns:

Agent instance

delete

def delete() -> None

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

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

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

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

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

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