Base Agent Module

Base module for Django agents

exception django_agents.base.AgentException[source]

Bases: Exception

Base exception for agents

exception django_agents.base.AgentValidationError[source]

Bases: AgentException

Validation error for the execution context

exception django_agents.base.AgentExecutionError[source]

Bases: AgentException

Error raised when the agent execution fails

class django_agents.base.BaseAgent[source]

Bases: ABC

Base class for every Django agent.

All agents must inherit from this class and implement the execute() method.

name

Unique agent name

Type:: str

description

Agent description

Type:: str

version

Agent version

Type:: str

enabled

Whether the agent is enabled

Type:: bool

name: str = 'base_agent'

description: str = 'Base agent'

version: str = '1.0.0'

enabled: bool = True

__init__()[source]: Initializes the agent

abstractmethod execute(context: Dict[str, Any]) → Dict[str, Any][source]

Executes the main agent logic

Parameters:: context – Dictionary containing the execution context
Returns:: Execution result as a dictionary
Raises:: AgentExecutionError – If something goes wrong during execution

validate_context(context: Dict[str, Any]) → bool[source]

Validates the provided context

Parameters:: context – Context to validate
Returns:: True when the context is valid
Raises:: AgentValidationError – If the context is not valid

pre_execute(context: Dict[str, Any]) → Dict[str, Any][source]

Hook executed before the main execution

Parameters:: context – Execution context
Returns:: Updated or original context

post_execute(result: Dict[str, Any], context: Dict[str, Any]) → Dict[str, Any][source]

Hook executed after the main execution

Parameters:

result – Execution result
context – Execution context

Returns:

Updated or original result

run(context: Dict[str, Any] | None = None) → Dict[str, Any][source]

Main entry point to run the agent

Parameters:

context – Execution context

Returns:

Execution result

Raises:

AgentValidationError – If the context is invalid
AgentExecutionError – If an error occurs during execution

log_info(message: str)[source]: Log an info message

log_warning(message: str)[source]: Log a warning message

log_error(message: str)[source]: Log an error message

log_debug(message: str)[source]: Log a debug message

get_stats() → Dict[str, Any][source]

Returns agent statistics

Returns:: Dictionary containing agent statistics

reset()[source]: Reset the agent state

class django_agents.base.AsyncAgent[source]

Bases: BaseAgent

Base class for asynchronous agents.

Extends BaseAgent and adds support for asynchronous execution through Celery or other task queues.

async_enabled: bool = True

queue_name: str = 'default'

priority: int = 5

async async_execute(context: Dict[str, Any]) → Dict[str, Any][source]

Asynchronous version of execute()

Parameters:: context – Execution context
Returns:: Execution result

enqueue(context: Dict[str, Any] | None = None)[source]

Queue the agent for asynchronous execution

Parameters:: context – Execution context
Returns:: Task ID or task reference

class django_agents.base.WorkflowAgent[source]

Bases: BaseAgent

Agent specialized in managing workflows.

A workflow is a sequence of agents executed in a defined order.

__init__()[source]: Initializes the agent

add_step(agent: BaseAgent, config: Dict[str, Any] | None = None)[source]

Add a step to the workflow

Parameters:

agent – Agent to add
config – Optional configuration for the step

execute(context: Dict[str, Any]) → Dict[str, Any][source]

Execute the workflow

Parameters:: context – Initial context
Returns:: Workflow result

BaseAgent

class django_agents.base.BaseAgent[source]

Bases: ABC

Base class for every Django agent.

All agents must inherit from this class and implement the execute() method.

name

Unique agent name

Type:: str

description

Agent description

Type:: str

version

Agent version

Type:: str

enabled

Whether the agent is enabled

Type:: bool

name: str = 'base_agent'

description: str = 'Base agent'

version: str = '1.0.0'

enabled: bool = True

__init__()[source]: Initializes the agent

abstractmethod execute(context: Dict[str, Any]) → Dict[str, Any][source]

Executes the main agent logic

Parameters:: context – Dictionary containing the execution context
Returns:: Execution result as a dictionary
Raises:: AgentExecutionError – If something goes wrong during execution

validate_context(context: Dict[str, Any]) → bool[source]

Validates the provided context

Parameters:: context – Context to validate
Returns:: True when the context is valid
Raises:: AgentValidationError – If the context is not valid

pre_execute(context: Dict[str, Any]) → Dict[str, Any][source]

Hook executed before the main execution

Parameters:: context – Execution context
Returns:: Updated or original context

post_execute(result: Dict[str, Any], context: Dict[str, Any]) → Dict[str, Any][source]

Hook executed after the main execution

Parameters:

result – Execution result
context – Execution context

Returns:

Updated or original result

run(context: Dict[str, Any] | None = None) → Dict[str, Any][source]

Main entry point to run the agent

Parameters:

context – Execution context

Returns:

Execution result

Raises:

AgentValidationError – If the context is invalid
AgentExecutionError – If an error occurs during execution

log_info(message: str)[source]: Log an info message

log_warning(message: str)[source]: Log a warning message

log_error(message: str)[source]: Log an error message

log_debug(message: str)[source]: Log a debug message

get_stats() → Dict[str, Any][source]

Returns agent statistics

Returns:: Dictionary containing agent statistics

reset()[source]: Reset the agent state

__repr__() → str[source]: Return repr(self).

__str__() → str[source]: Return str(self).

AsyncAgent

class django_agents.base.AsyncAgent[source]

Bases: BaseAgent

Base class for asynchronous agents.

Extends BaseAgent and adds support for asynchronous execution through Celery or other task queues.

async_enabled: bool = True

queue_name: str = 'default'

priority: int = 5

async async_execute(context: Dict[str, Any]) → Dict[str, Any][source]

Asynchronous version of execute()

Parameters:: context – Execution context
Returns:: Execution result

enqueue(context: Dict[str, Any] | None = None)[source]

Queue the agent for asynchronous execution

Parameters:: context – Execution context
Returns:: Task ID or task reference

WorkflowAgent

class django_agents.base.WorkflowAgent[source]

Bases: BaseAgent

Agent specialized in managing workflows.

A workflow is a sequence of agents executed in a defined order.

__init__()[source]: Initializes the agent

add_step(agent: BaseAgent, config: Dict[str, Any] | None = None)[source]

Add a step to the workflow

Parameters:

agent – Agent to add
config – Optional configuration for the step

execute(context: Dict[str, Any]) → Dict[str, Any][source]

Execute the workflow

Parameters:: context – Initial context
Returns:: Workflow result

Exceptions

exception django_agents.base.AgentException[source]

Bases: Exception

Base exception for agents

exception django_agents.base.AgentValidationError[source]

Bases: AgentException

Validation error for the execution context

exception django_agents.base.AgentExecutionError[source]

Bases: AgentException

Error raised when the agent execution fails