Skip to main content
Version: 2.0

aixplain.v2.agent

Agent module for aiXplain v2 SDK.

ConversationMessage Objects​

class ConversationMessage(TypedDict)

[view_source]

Type definition for a conversation message in agent history.

Attributes:

  • role - The role of the message sender, either 'user' or 'assistant'
  • content - The text content of the message

validate_history​

def validate_history(history: List[Dict[str, Any]]) -> bool

[view_source]

Validates conversation history for agent sessions.

This function ensures that the history is properly formatted for agent conversations, with each message containing the required 'role' and 'content' fields and proper types.

Arguments:

  • history - List of message dictionaries to validate

Returns:

  • bool - True if validation passes

Raises:

  • ValueError - If validation fails with detailed error messages

Example:

history = [ ... {"role": "user", "content": "Hello"}, ... {"role": "assistant", "content": "Hi there!"} ... ] validate_history(history) # Returns True

OutputFormat Objects​

class OutputFormat(str, Enum)

[view_source]

Output format options for agent responses.

AgentRunParams Objects​

class AgentRunParams(BaseRunParams)

[view_source]

Parameters for running an agent.

Attributes:

  • sessionId - Session ID for conversation continuity
  • query - The query to run
  • variables - Variables to replace {{variable}} placeholders in instructions and description. The backend performs the actual substitution.
  • allowHistoryAndSessionId - Allow both history and session ID
  • tasks - List of tasks for the agent
  • prompt - Custom prompt override
  • history - Conversation history
  • executionParams - Execution parameters (maxTokens, etc.)
  • criteria - Criteria for evaluation
  • evolve - Evolution parameters
  • inspectors - Inspector configurations
  • runResponseGeneration - Whether to run response generation. Defaults to True.
  • progress_format - Display format - "status" (single line) or "logs" (timeline). If None (default), progress tracking is disabled.
  • progress_verbosity - Detail level - 1 (minimal), 2 (thoughts), 3 (full I/O)
  • progress_truncate - Whether to truncate long text in progress display

AgentResponseData Objects​

@dataclass_json

@dataclass
class AgentResponseData()

[view_source]

Data structure for agent response.

AgentRunResult Objects​

@dataclass_json

@dataclass
class AgentRunResult(Result)

[view_source]

Result from running an agent.

data​

Override type from base class

debug​

def debug(prompt: Optional[str] = None,
          execution_id: Optional[str] = None,
          **kwargs: Any) -> "DebugResult"

[view_source]

Debug this agent response using the Debugger meta-agent.

This is a convenience method for quickly analyzing agent responses to identify issues, errors, or areas for improvement.

Note: This method requires the AgentRunResult to have been created through an Aixplain client context. If you have a standalone result, use the Debugger directly: aix.Debugger().debug_response(result)

Arguments:

  • prompt - Optional custom prompt to guide the debugging analysis.
  • Examples - "Why did it take so long?", "Focus on error handling"
  • execution_id - Optional execution ID (poll ID) for the run. If not provided, it will be extracted from the response's request_id or poll URL. This allows the debugger to fetch additional logs and information.
  • **kwargs - Additional parameters to pass to the debugger.

Returns:

  • DebugResult - The debugging analysis result.

Raises:

  • ValueError - If no client context is available for debugging.

Example:

agent = aix.Agent.get("my_agent_id") response = agent.run("Hello!") debug_result = response.debug() # Uses default prompt debug_result = response.debug("Why did it take so long?") # Custom prompt debug_result = response.debug(execution_id="abc-123") # With explicit ID print(debug_result.analysis)

Task Objects​

@dataclass_json

@dataclass
class Task()

[view_source]

A task definition for agent workflows.

__post_init__​

def __post_init__() -> None

[view_source]

Initialize task dependencies after dataclass creation.

Agent Objects​

@dataclass_json

@dataclass(repr=False)
class Agent(BaseResource, SearchResourceMixin[BaseSearchParams, "Agent"],
            GetResourceMixin[BaseGetParams,
                             "Agent"], DeleteResourceMixin[BaseDeleteParams,
                                                           "Agent"],
            RunnableResourceMixin[AgentRunParams, AgentRunResult])

[view_source]

Agent resource class.

__post_init__​

def __post_init__() -> None

[view_source]

Initialize agent after dataclass creation.

mark_as_deleted​

def mark_as_deleted() -> None

[view_source]

Mark the agent as deleted by setting status to DELETED and calling parent method.

before_run​

def before_run(*args: Any,
               **kwargs: Unpack[AgentRunParams]) -> Optional[AgentRunResult]

[view_source]

Hook called before running the agent to validate and prepare state.

on_poll​

def on_poll(response: AgentRunResult,
            **kwargs: Unpack[AgentRunParams]) -> None

[view_source]

Hook called after each poll to update progress display.

Arguments:

  • response - The poll response containing progress information
  • **kwargs - Run parameters

after_run​

def after_run(result: Union[AgentRunResult, Exception], *args: Any,
              **kwargs: Unpack[AgentRunParams]) -> Optional[AgentRunResult]

[view_source]

Hook called after running the agent for result transformation.

run​

def run(*args: Any, **kwargs: Unpack[AgentRunParams]) -> AgentRunResult

[view_source]

Run the agent with optional progress display.

Arguments:

  • *args - Positional arguments (first arg is treated as query)
  • query - The query to run
  • progress_format - Display format - "status" or "logs". If None (default), progress tracking is disabled.
  • progress_verbosity - Detail level 1-3 (default: 1)
  • progress_truncate - Truncate long text (default: True)
  • **kwargs - Additional run parameters
  • *args - Positional arguments (first arg is treated as query)
  • query - The query to run
  • progress_format - Display format - "status" or "logs". If None (default), progress tracking is disabled.
  • progress_verbosity - Detail level 1-3 (default: 1)
  • progress_truncate - Truncate long text (default: True)
  • **kwargs - Additional run parameters

Returns:

  • AgentRunResult - The result of the agent execution

run_async​

def run_async(*args: Any, **kwargs: Unpack[AgentRunParams]) -> AgentRunResult

[view_source]

Run the agent asynchronously.

Arguments:

  • *args - Positional arguments (first arg is treated as query)
  • query - The query to run
  • **kwargs - Additional run parameters

Returns:

  • AgentRunResult - The result of the agent execution

save​

def save(*args: Any, **kwargs: Any) -> "Agent"

[view_source]

Save the agent with dependency management.

This method extends the base save functionality to handle saving of dependent child components before the agent itself is saved.

Arguments:

  • *args - Positional arguments passed to parent save method.
  • save_subcomponents - bool - If True, recursively save all unsaved child components (default: False)
  • as_draft - bool - If True, save agent as draft status (default: False)
  • **kwargs - Other attributes to set before saving

Returns:

  • Agent - The saved agent instance

Raises:

  • ValueError - If child components are not saved and save_subcomponents is False

before_save​

def before_save(*args: Any, **kwargs: Any) -> Optional[dict]

[view_source]

Callback to be called before the resource is saved.

Handles status transitions based on save type.

after_clone​

def after_clone(result: Union["Agent", Exception],
                **kwargs: Any) -> Optional["Agent"]

[view_source]

Callback called after the agent is cloned.

Sets the cloned agent's status to DRAFT.

@classmethod
def search(cls: type["Agent"],
           query: Optional[str] = None,
           **kwargs: Unpack[BaseSearchParams]) -> "Page[Agent]"

[view_source]

Search agents with optional query and filtering.

Arguments:

  • query - Optional search query string
  • **kwargs - Additional search parameters (ownership, status, etc.)

Returns:

Page of agents matching the search criteria

build_save_payload​

def build_save_payload(**kwargs: Any) -> dict

[view_source]

Build the payload for the save action.

build_run_payload​

def build_run_payload(**kwargs: Unpack[AgentRunParams]) -> dict

[view_source]

Build the payload for the run action.

generate_session_id​

def generate_session_id(
        history: Optional[List[ConversationMessage]] = None) -> str

[view_source]

Generate a unique session ID for agent conversations.

This method creates a unique session identifier based on the agent ID and current timestamp. If conversation history is provided, it attempts to initialize the session on the server to enable context-aware conversations.

Arguments:

  • history Optional[List[Dict]], optional - Previous conversation history. Each dict should contain 'role' (either 'user' or 'assistant') and 'content' keys. Defaults to None.

Returns:

  • str - A unique session identifier in the format "{agent_id}_{timestamp}".

Raises:

  • ValueError - If the history format is invalid.

Example:

agent = Agent.get("my_agent_id") session_id = agent.generate_session_id()

Or with history

history = [ ... {"role": "user", "content": "Hello"}, ... {"role": "assistant", "content": "Hi there!"} ... ] session_id = agent.generate_session_id(history=history)