sherpa_ai.agents package

In This Page:

sherpa_ai.agents package#

Overview#

The agents package provides specialized AI agents with different roles and capabilities. Each agent is designed to excel at specific types of tasks, from answering questions to planning complex workflows.

Key Components

QAAgent: Specialized in question answering tasks
MLEngineer: Expert in machine learning tasks and code generation
Physicist: Specialized in physics-related questions and calculations
Planner: Creates strategic plans for complex problems
Critic: Evaluates and provides feedback on other agents’ outputs
AgentPool: Manages multiple agents and their interactions
PersistentAgentPool: Enhanced agent pool with persistence capabilities
UserAgentManager: Multi-user agent management with preferences

Class Hierarchy#

The following diagram shows the inheritance relationships between the core agent classes:

Component Relationships#

The diagram below shows how agents interact with other components of the Sherpa AI framework:

digraph "Agent Relationships" { graph [fontname="Helvetica", fontsize=14, rankdir=LR, nodesep=0.8, ranksep=0.8]; node [fontname="Helvetica", fontsize=12, shape=box, style="filled,rounded", fillcolor="#f5f5f5", color="#336790", margin=0.3]; edge [fontname="Helvetica", fontsize=10, color="#555555"]; Agent [fillcolor="#d5f5e3", label="Agent"]; Model [fillcolor="#fadbd8", label="Model"]; Memory [fillcolor="#ebdef0", label="Memory"]; Policy [fillcolor="#fdebd0", label="Policy"]; Tools [fillcolor="#d6eaf8", label="Tools"]; Agent -> Model [label="uses"]; Agent -> Memory [label="reads/writes"]; Agent -> Policy [label="follows"]; Agent -> Tools [label="utilizes"]; }

Agent Component Relationships#

Example Usage#

Basic Agent Usage#

from sherpa_ai.agents import QAAgent, Critic
from sherpa_ai.models import SherpaBaseChatModel

# Create a QA agent
model = SherpaBaseChatModel()
qa_agent = QAAgent(model=model)

# Get a response from the QA agent
response = qa_agent.get_response("What is machine learning?")

# Create a critic to evaluate the response
critic = Critic(model=model)
evaluation = critic.evaluate(response)

print(f"Response: {response}")
print(f"Evaluation: {evaluation}")

Agent Persistence#

from sherpa_ai.agents.persistent_agent_pool import PersistentAgentPool
from sherpa_ai.agents.user_agent_manager import UserAgentManager

# Create persistent agent pool
pool = PersistentAgentPool("agents.db", "sqlite")

# Save an agent
agent_id = pool.save_agent(qa_agent, user_id="user123")

# Load agent later
loaded_agent = pool.get_agent(agent_id)

# Multi-user management
manager = UserAgentManager("user_agents.db")
manager.create_user("user123", max_agents=5)
agent_id = manager.create_agent_for_user("user123", "My Assistant", "QAAgent")

Submodules#

Module	Description
`sherpa_ai.agents.agent_pool`	Provides the AgentPool class for managing multiple agents and their interactions.
`sherpa_ai.agents.persistent_agent_pool`	Enhanced agent pool with persistence capabilities for saving and loading agents.
`sherpa_ai.agents.user_agent_manager`	Multi-user agent management with preferences and session tracking.
`sherpa_ai.agents.agent_storage`	Database schema management and migration utilities for agent persistence.
`sherpa_ai.agents.base`	Contains abstract base classes defining the agent interface and core functionality.
`sherpa_ai.agents.qa_agent`	Implements the QAAgent specialized in question answering tasks.
`sherpa_ai.agents.user`	Provides the User agent that represents human users in the system.

sherpa_ai.agents.agent_pool module#

class sherpa_ai.agents.agent_pool.AgentPool[source]#

Bases: object

A collection of agents that can be managed and accessed by name.

This class provides a centralized registry for agents, allowing them to be added, retrieved, and listed. It maintains a dictionary of agents indexed by their names for easy access.

This is the base class for in-memory agent pools. For persistent storage, use PersistentAgentPool which extends this class.

agents#

Dictionary mapping agent names to agent instances.

Type:: Dict[str, BaseAgent]

Example

>>> from sherpa_ai.agents.agent_pool import AgentPool
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> pool = AgentPool()
>>> agent = QAAgent(name="Research Assistant")
>>> pool.add_agent(agent)
>>> print("Research Assistant" in pool)
True

get_agent(agent_name)[source]#

Retrieve an agent by its name.

Parameters:: agent_name (str) – The name of the agent to retrieve.
Returns:: The agent with the specified name, or None if not found.
Return type:: Optional[BaseAgent]

Example

>>> from sherpa_ai.agents.agent_pool import AgentPool
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> pool = AgentPool()
>>> agent = QAAgent(name="Research Assistant")
>>> pool.add_agent(agent)
>>> retrieved = pool.get_agent("Research Assistant")
>>> print(retrieved.name)
Research Assistant

add_agent(agent)[source]#

Add a single agent to the pool.

Parameters:: agent (BaseAgent) – The agent to add to the pool.

Example

>>> from sherpa_ai.agents.agent_pool import AgentPool
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> pool = AgentPool()
>>> agent = QAAgent(name="Research Assistant")
>>> pool.add_agent(agent)
>>> print(len(pool.agents))
1

add_agents(agents)[source]#

Add multiple agents to the pool.

Parameters:: agents (List[BaseAgent]) – List of agents to add to the pool.

Example

>>> from sherpa_ai.agents.agent_pool import AgentPool
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> from sherpa_ai.agents.ml_engineer import MLEngineer
>>> pool = AgentPool()
>>> agents = [
...     QAAgent(name="Research Assistant"),
...     MLEngineer(name="AI Expert")
... ]
>>> pool.add_agents(agents)
>>> print(len(pool.agents))
2

get_agent_pool_description()[source]#

Create a description of all agents in the pool.

This method generates a formatted string containing the name and description of each agent in the pool, which can be used for agent planning or display.

Returns:: A formatted string describing all agents in the pool.
Return type:: str

Example

>>> from sherpa_ai.agents.agent_pool import AgentPool
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> pool = AgentPool()
>>> agent = QAAgent(name="Research Assistant")
>>> pool.add_agent(agent)
>>> description = pool.get_agent_pool_description()
>>> print("Research Assistant" in description)
True

sherpa_ai.agents.persistent_agent_pool module#

Persistent Agent Pool module for Sherpa AI.

This module provides a comprehensive agent persistence and retrieval system that supports user-specific agent management, state serialization, and both SQLite database and JSON file storage.

sherpa_ai.agents.persistent_agent_pool.handle_storage_errors(default_return=None)[source]#

Decorator to handle common storage operation errors.

Parameters:: default_return – Value to return on error (None, False, [], etc.)

class sherpa_ai.agents.persistent_agent_pool.AgentMetadata(**data)[source]#

Bases: BaseModel

Metadata for agent storage and retrieval.

agent_id#

Unique identifier for the agent.

Type:: str

user_id#

User who owns this agent.

Type:: str

agent_name#

Human-readable name of the agent.

Type:: str

agent_type#

Type/class of the agent.

Type:: str

created_at#

When the agent was created.

Type:: datetime

updated_at#

When the agent was last updated.

Type:: datetime

is_active#

Whether the agent is currently active.

Type:: bool

tags#

Tags for categorization and search.

Type:: List[str]

description#

Description of the agent’s purpose.

Type:: str

agent_id: str#

user_id: str#

agent_name: str#

agent_type: str#

created_at: datetime#

updated_at: datetime#

is_active: bool#

tags: List[str]#

description: str#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sherpa_ai.agents.persistent_agent_pool.AgentState(**data)[source]#

Bases: BaseModel

Serializable agent state for persistence.

agent_config#

Agent configuration data.

Type:: Dict[str, Any]

belief_state#

Agent’s belief state.

Type:: Dict[str, Any]

shared_memory_state#

Shared memory state.

Type:: Dict[str, Any]

execution_state#

Current execution state.

Type:: Dict[str, Any]

agent_config: Dict[str, Any]#

belief_state: Dict[str, Any]#

shared_memory_state: Dict[str, Any]#

execution_state: Dict[str, Any]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sherpa_ai.agents.persistent_agent_pool.StoredAgent(**data)[source]#

Bases: BaseModel

Complete stored agent representation.

metadata#

Agent metadata.

Type:: AgentMetadata

state#

Agent state data.

Type:: AgentState

metadata: AgentMetadata#

state: AgentState#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sherpa_ai.agents.persistent_agent_pool.PersistentAgentPool(storage_path='agent_pool.db', storage_type='sqlite')[source]#

Bases: AgentPool

Enhanced agent pool with persistence and user-specific management.

This class extends the base AgentPool with persistent storage capabilities. It provides a comprehensive agent management system that supports: - User-specific agent storage and retrieval - Agent state persistence across sessions - Pydantic validation for data integrity - Multiple storage backends (SQLite database and JSON file) - Thread-safe operations - Agent metadata and tagging

The agents are persisted to either an SQLite database or JSON file based on the storage_type parameter.

storage_path#

Path to the storage file (database or JSON).

Type:: str

storage_type#

Type of storage backend (‘sqlite’ or ‘json’).

Type:: str

_lock#

Thread safety lock.

Type:: threading.RLock

save_agent(agent, user_id='default', tags=None, overwrite=False)[source]#

Save an agent to persistent storage.

Parameters:

agent (BaseAgent) – The agent to save.
user_id (str) – User ID for the agent.
tags (List[str]) – Tags for categorization.
overwrite (bool) – Whether to overwrite existing agent with same name.

Returns:

The agent ID of the saved agent.

Return type:

str

Raises:

ValueError – If agent with same name exists and overwrite is False.

Example

>>> pool = PersistentAgentPool()
>>> agent = QAAgent(name="My Agent")
>>> agent_id = pool.save_agent(agent, user_id="user123", tags=["qa", "research"])
>>> print(agent_id)
My Agent_140123456789

get_agent(agent_id)[source]#

Retrieve an agent by its ID.

Parameters:: agent_id (str) – The agent ID to retrieve.
Returns:: The agent if found, None otherwise.
Return type:: Optional[BaseAgent]

Example

>>> pool = PersistentAgentPool()
>>> agent = pool.get_agent("My Agent_140123456789")
>>> print(agent.name if agent else "Not found")
My Agent

get_agent_by_name(agent_name, user_id='default')[source]#

Retrieve an agent by name and user ID.

Parameters:

agent_name (str) – The agent name to retrieve.
user_id (str) – The user ID who owns the agent.

Returns:

The agent if found, None otherwise.

Return type:

Optional[BaseAgent]

Example

>>> pool = PersistentAgentPool()
>>> agent = pool.get_agent_by_name("My Agent", "user123")
>>> print(agent.name if agent else "Not found")
My Agent

list_agents(user_id=None, agent_type=None, tags=None, active_only=True)[source]#

List agents with optional filtering.

Parameters:

user_id (str) – Filter by user ID.
agent_type (str) – Filter by agent type.
tags (List[str]) – Filter by tags (any match).
active_only (bool) – Only return active agents.

Returns:

List of agent metadata matching the criteria.

Return type:

List[AgentMetadata]

Example

>>> pool = PersistentAgentPool()
>>> agents = pool.list_agents(user_id="user123", tags=["qa"])
>>> for agent in agents:
...     print(f"{agent.agent_name} ({agent.agent_type})")
My QA Agent (QAAgent)

delete_agent(agent_id, soft_delete=True)[source]#

Delete an agent.

Parameters:

agent_id (str) – The agent ID to delete.
soft_delete (bool) – If True, mark as inactive; if False, permanently delete.

Returns:

True if successful, False otherwise.

Return type:

bool

Example

>>> pool = PersistentAgentPool()
>>> success = pool.delete_agent("My Agent_140123456789")
>>> print("Deleted" if success else "Failed")
Deleted

update_agent(agent_id, agent)[source]#

Update an existing agent.

Parameters:

agent_id (str) – The agent ID to update.
agent (BaseAgent) – The updated agent.

Returns:

True if successful, False otherwise.

Return type:

bool

Example

>>> pool = PersistentAgentPool()
>>> agent = QAAgent(name="Updated Agent")
>>> success = pool.update_agent("My Agent_140123456789", agent)
>>> print("Updated" if success else "Failed")
Updated

get_agent_count(user_id=None)[source]#

Get the count of agents.

Parameters:: user_id (str) – Count agents for specific user, or all if None.
Returns:: Number of active agents.
Return type:: int

Example

>>> pool = PersistentAgentPool()
>>> count = pool.get_agent_count("user123")
>>> print(f"User has {count} agents")
User has 5 agents

clear_cache()[source]#: Clear the in-memory agent cache.

reload_from_storage()[source]#

Reload all agents from storage into cache.

This method leverages the base AgentPool’s agents dictionary and repopulates it from the persistent storage.

get_cache_stats()[source]#

Get statistics about the current cache state.

Returns:: Cache statistics including size, storage type, etc.
Return type:: Dict[str, Any]

sherpa_ai.agents.user_agent_manager module#

User-specific agent management module for Sherpa AI.

This module provides user-specific agent management capabilities, including personalized agent creation, user preferences, and multi-user support.

class sherpa_ai.agents.user_agent_manager.UserPreferences(**data)[source]#

Bases: BaseModel

User preferences for agent management.

user_id#

Unique user identifier.

Type:: str

default_agent_type#

Default agent type for new agents.

Type:: str

max_agents#

Maximum number of agents per user.

Type:: int

auto_save#

Whether to automatically save agent state.

Type:: bool

preferred_llm#

Preferred LLM model.

Type:: str

notification_settings#

Notification preferences.

Type:: Dict[str, Any]

created_at#

When preferences were created.

Type:: datetime

updated_at#

When preferences were last updated.

Type:: datetime

user_id: str#

default_agent_type: str#

max_agents: int#

auto_save: bool#

preferred_llm: str#

notification_settings: Dict[str, Any]#

created_at: datetime#

updated_at: datetime#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sherpa_ai.agents.user_agent_manager.UserAgentSession(**data)[source]#

Bases: BaseModel

User agent session information.

session_id#

Unique session identifier.

Type:: str

user_id#

User who owns this session.

Type:: str

agent_id#

Agent being used in this session.

Type:: str

started_at#

When the session started.

Type:: datetime

last_activity#

Last activity timestamp.

Type:: datetime

is_active#

Whether the session is currently active.

Type:: bool

session_data#

Additional session data.

Type:: Dict[str, Any]

session_id: str#

user_id: str#

agent_id: str#

started_at: datetime#

last_activity: datetime#

is_active: bool#

session_data: Dict[str, Any]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sherpa_ai.agents.user_agent_manager.UserAgentManager(db_path='user_agents.db')[source]#

Bases: object

Manages user-specific agents and their interactions.

This class provides comprehensive user-specific agent management, including personalized agent creation, session management, and user preference handling.

agent_pool#

The persistent agent pool.

Type:: PersistentAgentPool

user_preferences#

User preferences cache.

Type:: Dict[str, UserPreferences]

active_sessions#

Active user sessions.

Type:: Dict[str, UserAgentSession]

user_preferences: Dict[str, UserPreferences]#

active_sessions: Dict[str, UserAgentSession]#

create_user(user_id, preferences=None, **kwargs)[source]#

Create a new user with default or custom preferences.

Parameters:

user_id (str) – Unique user identifier.
preferences (Optional[UserPreferences]) – Custom user preferences.
**kwargs – Additional preferences to override defaults.

Returns:

The created user preferences.

Return type:

UserPreferences

Example

>>> manager = UserAgentManager()
>>> prefs = manager.create_user("user123", max_agents=5)
>>> print(prefs.max_agents)
5

get_user_preferences(user_id)[source]#

Get user preferences.

Parameters:: user_id (str) – User identifier.
Returns:: User preferences or None if not found.
Return type:: Optional[UserPreferences]

Example

>>> manager = UserAgentManager()
>>> prefs = manager.get_user_preferences("user123")
>>> print(prefs.max_agents if prefs else "User not found")
10

update_user_preferences(user_id, **kwargs)[source]#

Update user preferences.

Parameters:

user_id (str) – User identifier.
**kwargs – Preference fields to update.

Returns:

True if successful, False otherwise.

Return type:

bool

Example

>>> manager = UserAgentManager()
>>> success = manager.update_user_preferences("user123", max_agents=20)
>>> print("Updated" if success else "Failed")
Updated

create_agent_for_user(user_id, agent_name, agent_type=None, **kwargs)[source]#

Create a new agent for a specific user.

Parameters:

user_id (str) – User identifier.
agent_name (str) – Name for the new agent.
agent_type (Optional[str]) – Type of agent to create.
**kwargs – Additional agent configuration.

Returns:

Agent ID if successful, None otherwise.

Return type:

Optional[str]

Example

>>> manager = UserAgentManager()
>>> agent_id = manager.create_agent_for_user("user123", "My Assistant", "QAAgent")
>>> print(agent_id if agent_id else "Failed")
My Assistant_140123456789

get_user_agents(user_id, active_only=True)[source]#

Get all agents for a specific user.

Parameters:

user_id (str) – User identifier.
active_only (bool) – Only return active agents.

Returns:

List of agent metadata.

Return type:

List[AgentMetadata]

Example

>>> manager = UserAgentManager()
>>> agents = manager.get_user_agents("user123")
>>> for agent in agents:
...     print(f"{agent.agent_name} ({agent.agent_type})")
My Assistant (QAAgent)

get_agent_for_user(user_id, agent_name)[source]#

Get a specific agent for a user.

Parameters:

user_id (str) – User identifier.
agent_name (str) – Agent name.

Returns:

The agent if found, None otherwise.

Return type:

Optional[BaseAgent]

Example

>>> manager = UserAgentManager()
>>> agent = manager.get_agent_for_user("user123", "My Assistant")
>>> print(agent.name if agent else "Not found")
My Assistant

start_user_session(user_id, agent_name)[source]#

Start a new user session with an agent.

Parameters:

user_id (str) – User identifier.
agent_name (str) – Agent name to use.

Returns:

Session ID if successful, None otherwise.

Return type:

Optional[str]

Example

>>> manager = UserAgentManager()
>>> session_id = manager.start_user_session("user123", "My Assistant")
>>> print(session_id if session_id else "Failed")
session_140123456789

end_user_session(session_id)[source]#

End a user session.

Parameters:: session_id (str) – Session identifier.
Returns:: True if successful, False otherwise.
Return type:: bool

Example

>>> manager = UserAgentManager()
>>> success = manager.end_user_session("session_140123456789")
>>> print("Ended" if success else "Failed")
Ended

get_user_session(session_id)[source]#

Get user session information.

Parameters:: session_id (str) – Session identifier.
Returns:: Session information or None if not found.
Return type:: Optional[UserAgentSession]

Example

>>> manager = UserAgentManager()
>>> session = manager.get_user_session("session_140123456789")
>>> print(session.user_id if session else "Not found")
user123

get_user_active_sessions(user_id)[source]#

Get all active sessions for a user.

Parameters:: user_id (str) – User identifier.
Returns:: List of active sessions.
Return type:: List[UserAgentSession]

Example

>>> manager = UserAgentManager()
>>> sessions = manager.get_user_active_sessions("user123")
>>> print(f"User has {len(sessions)} active sessions")
User has 2 active sessions

auto_save_agent_state(user_id, agent)[source]#

Automatically save agent state if user has auto-save enabled.

Parameters:

user_id (str) – User identifier.
agent (BaseAgent) – Agent to save.

Returns:

True if saved or auto-save disabled, False if failed.

Return type:

bool

Example

>>> manager = UserAgentManager()
>>> success = manager.auto_save_agent_state("user123", agent)
>>> print("Saved" if success else "Failed")
Saved

get_user_statistics(user_id)[source]#

Get statistics for a user.

Parameters:: user_id (str) – User identifier.
Returns:: User statistics.
Return type:: Dict[str, Any]

Example

>>> manager = UserAgentManager()
>>> stats = manager.get_user_statistics("user123")
>>> print(f"User has {stats['total_agents']} agents")
User has 5 agents

cleanup_inactive_sessions(max_age_hours=24)[source]#

Clean up inactive sessions older than specified hours.

Parameters:: max_age_hours (int) – Maximum age of sessions in hours.
Returns:: Number of sessions cleaned up.
Return type:: int

Example

>>> manager = UserAgentManager()
>>> cleaned = manager.cleanup_inactive_sessions(12)
>>> print(f"Cleaned up {cleaned} sessions")
Cleaned up 3 sessions

sherpa_ai.agents.agent_storage module#

Agent storage and database management module for Sherpa AI.

This module provides database schema management, migrations, and storage utilities for the persistent agent pool system.

class sherpa_ai.agents.agent_storage.DatabaseMigration(**data)[source]#

Bases: BaseModel

Database migration definition.

version#

Migration version number.

Type:: int

description#

Description of what the migration does.

Type:: str

up_sql#

SQL to apply the migration.

Type:: str

down_sql#

SQL to rollback the migration.

Type:: str

created_at#

When the migration was created.

Type:: datetime

version: int#

description: str#

up_sql: str#

down_sql: str#

created_at: datetime#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sherpa_ai.agents.agent_storage.AgentStorageManager(db_path='agent_pool.db')[source]#

Bases: object

Manages agent storage, database schema, and migrations.

This class provides comprehensive database management for agent persistence, including schema creation, migrations, and data integrity checks.

db_path#

Path to the SQLite database file.

Type:: str

migrations_table#

Name of the migrations tracking table.

Type:: str

initialize_database()[source]#

Initialize the database with the latest schema.

Returns:: True if successful, False otherwise.
Return type:: bool

Example

>>> storage = AgentStorageManager()
>>> success = storage.initialize_database()
>>> print("Database initialized" if success else "Failed")
Database initialized

get_database_info()[source]#

Get information about the database.

Returns:: Database information including tables, indexes, and statistics.
Return type:: Dict[str, Any]

Example

>>> storage = AgentStorageManager()
>>> info = storage.get_database_info()
>>> print(f"Database has {info['table_count']} tables")
Database has 4 tables

backup_database(backup_path)[source]#

Create a backup of the database.

Parameters:: backup_path (str) – Path where to save the backup.
Returns:: True if successful, False otherwise.
Return type:: bool

Example

>>> storage = AgentStorageManager()
>>> success = storage.backup_database("backup_agents.db")
>>> print("Backup created" if success else "Failed")
Backup created

restore_database(backup_path)[source]#

Restore database from backup.

Parameters:: backup_path (str) – Path to the backup file.
Returns:: True if successful, False otherwise.
Return type:: bool

Example

>>> storage = AgentStorageManager()
>>> success = storage.restore_database("backup_agents.db")
>>> print("Database restored" if success else "Failed")
Database restored

cleanup_old_data(days_to_keep=30)[source]#

Clean up old data from the database.

Parameters:: days_to_keep (int) – Number of days of data to keep.
Returns:: Number of records cleaned up by table.
Return type:: Dict[str, int]

Example

>>> storage = AgentStorageManager()
>>> cleaned = storage.cleanup_old_data(7)  # Keep 7 days
>>> print(f"Cleaned {cleaned['conversations']} old conversations")
Cleaned 150 old conversations

optimize_database()[source]#

Optimize the database for better performance.

Returns:: True if successful, False otherwise.
Return type:: bool

Example

>>> storage = AgentStorageManager()
>>> success = storage.optimize_database()
>>> print("Database optimized" if success else "Failed")
Database optimized

validate_data_integrity()[source]#

Validate data integrity in the database.

Returns:: Validation results and any issues found.
Return type:: Dict[str, Any]

Example

>>> storage = AgentStorageManager()
>>> validation = storage.validate_data_integrity()
>>> print(f"Validation: {validation['status']}")
Validation: passed

sherpa_ai.agents.base module#

class sherpa_ai.agents.base.BaseAgent(**data)[source]#

Bases: ABC, BaseModel

Base class for all agents in the Sherpa AI system.

This abstract class defines the core functionality and interface that all agents must implement. It provides methods for action selection, execution, validation, and output synthesis.

name#

The name of the agent.

Type:: str

description#

A description of the agent’s purpose and capabilities.

Type:: str

shared_memory#

Memory shared between agents for information exchange.

Type:: SharedMemory

belief#

The agent’s internal state and knowledge.

Type:: Belief

policy#

The policy that determines which actions to take.

Type:: BasePolicy

num_runs#

Number of action execution cycles to perform. Defaults to 1.

Type:: int

actions#

List of actions the agent can perform.

Type:: List[BaseAction]

validation_steps#

Number of validation attempts per validation. Defaults to 1.

Type:: int

validations#

List of output validators.

Type:: List[BaseOutputProcessor]

feedback_agent_name#

Name of the agent providing feedback. Defaults to “critic”.

Type:: str

global_regen_max#

Maximum number of output regeneration attempts. Defaults to 12.

Type:: int

llm#

Language model used for text generation.

Type:: Any

prompt_template#

Template for generating prompts.

Type:: PromptTemplate

stop_checker#

Function to determine if execution should stop.

Type:: Callable[[Belief], bool]

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> from sherpa_ai.actions.base import BaseAction
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> print(agent.name)
TestAgent

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str#

description: str#

shared_memory: SharedMemory#

belief: Belief#

policy: BasePolicy#

num_runs: int#

actions: List[BaseAction]#

validation_steps: int#

validations: List[BaseOutputProcessor]#

feedback_agent_name: str#

global_regen_max: int#

llm: BaseLanguageModel | None#

stop_checker: Callable[[Belief], bool]#

prompt_template: PromptTemplate#

abstractmethod create_actions()[source]#

Create and return the list of actions available to this agent.

This method must be implemented by all agent subclasses to define the specific actions that the agent can perform.

Returns:: List of action objects that the agent can use.
Return type:: List[BaseAction]

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> from sherpa_ai.actions.base import BaseAction
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return [MyCustomAction()]
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> actions = agent.create_actions()
>>> print(len(actions))
1

abstractmethod synthesize_output()[source]#

Generate the final output based on the agent’s actions and belief state.

This method must be implemented by all agent subclasses to define how the agent produces its final output from its internal state.

Returns:: The synthesized output string.
Return type:: str

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "This is my final answer."
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> output = agent.synthesize_output()
>>> print(output)
This is my final answer.

send_event(event, args)[source]#

Send an event to the state machine in the belief.

This method dispatches an event to the agent’s belief state machine, allowing the agent to update its internal state based on external events.

Parameters:

event (str) – The name of the event to send.
args (dict) – Arguments to pass to the event handler.

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> agent.send_event("task_start", {"task_id": "123"})

async async_send_event(event, args)[source]#

Send an event to the state machine in the belief asynchronously.

This method dispatches an event to the agent’s belief state machine asynchronously, allowing for non-blocking state updates.

Parameters:

event (str) – The name of the event to send.
args (dict) – Arguments to pass to the event handler.

Example

>>> import asyncio
>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> asyncio.run(agent.async_send_event("task_start", {"task_id": "123"}))

async async_handle_event(event)[source]#

Handle a specific type of event.

This method processes the event based on its type and updates the agent’s belief state accordingly. To create customized event handling, override this method in subclasses.

Parameters:: event (Event) – The event to handle.

agent_preparation()[source]#

Prepare the agent for execution.

This method initializes the agent’s state, observes the shared memory, and sets up the agent’s actions if they haven’t been defined yet.

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> agent.agent_preparation()

select_action()[source]#

Select the next action to execute based on the current belief state.

This method uses the agent’s policy to determine which action to take next based on the current state of the agent’s belief.

Returns:

The selected action and its arguments, or None if: no action could be selected.

Return type:

Optional[PolicyOutput]

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> action = agent.select_action()
>>> print(action is None)
True

async async_select_action()[source]#

Select the next action to execute asynchronously.

This method uses the agent’s policy to determine which action to take next based on the current state of the agent’s belief, in an asynchronous manner.

Returns:

The selected action and its arguments, or None if: no action could be selected.

Return type:

Optional[PolicyOutput]

Example

>>> import asyncio
>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> action = asyncio.run(agent.async_select_action())
>>> print(action is None)
True

async agent_finished(result)[source]#

Process the final result after all actions have been executed.

This method validates the output if validators are present, logs the result, and stores it in the shared memory.

Parameters:: result (str) – The final result to process.
Returns:: The processed result.
Return type:: str

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> final_result = agent.agent_finished("My answer")
>>> print(final_result)
My answer

run()[source]#

Run the agent synchronously.

This method executes the agent’s action selection and execution loop synchronously, returning a TaskResult with the final output.

Returns:: The result of the agent’s execution.
Return type:: TaskResult

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> result = agent.run()
>>> print(result.status)
success

async async_run()[source]#

Run the agent asynchronously.

This method executes the agent’s action selection and execution loop asynchronously, returning a TaskResult with the final output.

Returns:: The result of the agent’s execution.
Return type:: TaskResult

Example

>>> import asyncio
>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> result = asyncio.run(agent.async_run())
>>> print(result.status)
success

validation_iterator(validations, global_regen_count, all_pass, validation_is_scaped, result)[source]#

Iterate through validations and process their results.

This method processes each validation in sequence, updating the belief and regenerating output if necessary.

Parameters:

validations – List of validators to process.
global_regen_count – Current count of regeneration attempts.
all_pass – Whether all validations have passed so far.
validation_is_scaped – Whether any validation has been skipped.
result – Current result to validate.

Returns:

Updated values for global_regen_count, all_pass, validation_is_scaped, and result.

Return type:

tuple

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> count, passed, escaped, output = agent.validation_iterator(
...     validations=[],
...     global_regen_count=0,
...     all_pass=False,
...     validation_is_scaped=False,
...     result="Test output"
... )
>>> print(count)
0

validate_output()[source]#

Validate the synthesized output through a series of validation steps.

This method iterates through each validation in the ‘validations’ list, and for each validation, it performs ‘validation_steps’ attempts to synthesize output using ‘synthesize_output’ method. If the output doesn’t pass validation, feedback is incorporated into the belief system.

If a validation fails after all attempts, the error messages from the last failed validation are appended to the final result.

Returns:: The synthesized output after validation.
Return type:: str

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return []
...     def synthesize_output(self) -> str:
...         return "Validated output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> result = agent.validate_output()
>>> print(result)
Validated output

act(action, inputs)[source]#

Execute an action synchronously.

This method executes the specified action with the given inputs, handling any exceptions that may occur during execution.

Parameters:

action (BaseAction) – The action to execute.
inputs (dict) – Input parameters for the action.

Returns:

The result of the action execution,: or an exception if execution failed.

Return type:

Union[Optional[str], Exception]

Example

>>> from sherpa_ai.agents.base import BaseAgent
>>> from sherpa_ai.actions.base import BaseAction
>>> class MyAction(BaseAction):
...     def execute(self, **kwargs):
...         return "Action executed"
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return [MyAction()]
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> result = agent.act(MyAction(), {"param": "value"})
>>> print(result)
Action executed

async async_act(action, inputs)[source]#

Execute an action asynchronously.

This method executes the specified action with the given inputs asynchronously, handling any exceptions that may occur during execution.

Parameters:

action (BaseAction) – The action to execute.
inputs (dict) – Input parameters for the action.

Returns:

The result of the action execution, or None if execution failed.

Return type:

Optional[str]

Example

>>> import asyncio
>>> from sherpa_ai.agents.base import BaseAgent
>>> from sherpa_ai.actions.base import BaseAction
>>> class MyAction(BaseAction):
...     async def execute(self, **kwargs):
...         return "Action executed"
>>> class MyAgent(BaseAgent):
...     def create_actions(self) -> List[BaseAction]:
...         return [MyAction()]
...     def synthesize_output(self) -> str:
...         return "Output"
>>> agent = MyAgent(name="TestAgent", description="A test agent")
>>> result = asyncio.run(agent.async_act(MyAction(), {"param": "value"}))
>>> print(result)
Action executed

sherpa_ai.agents.qa_agent module#

class sherpa_ai.agents.qa_agent.QAAgent(*args, custom_task_description_prompt=None, custom_action_plan_prompt=None, **kwargs)[source]#

Bases: BaseAgent

A specialized agent for answering questions and providing information.

This agent is designed to handle question-answering tasks by searching for information, synthesizing responses, and validating outputs. It can optionally include citations in its responses.
Attributes:
name (str): The name of the agent, defaults to “QA Agent”. description (str): A description of the agent’s purpose and capabilities. config (AgentConfig): Configuration settings for the agent. num_runs (int): Number of action execution cycles, defaults to 3. global_regen_max (int): Maximum number of output regeneration attempts, defaults to 5. citation_enabled (bool): Whether to include citations in responses, defaults to False.

Args:
custom_task_description_prompt (Optional[str]): Custom prompt for task description. If None, uses default from prompt template. custom_action_plan_prompt (Optional[str]): Custom prompt for action plan description. If None, uses default from prompt template.

Example:
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> from sherpa_ai.config import AgentConfig
>>> agent = QAAgent(
...     name="Research Assistant",
...     config=AgentConfig(),
...     citation_enabled=True
... )
>>> print(agent.name)
Research Assistant
>>> # Using custom prompts
>>> agent = QAAgent(
...     name="Custom QA",
...     custom_task_description_prompt="You are an expert QA assistant.",
...     custom_action_plan_prompt="Plan to solve: {task}"
... )
>>> print(agent.description)
You are an expert QA assistant.

Your name is Custom QA.

name: str#

num_runs: int#

global_regen_max: int#

description: str#

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

config: AgentConfig#

citation_enabled: bool#

create_actions()[source]#

Create and return the list of actions available to this agent.

This method defines the specific actions that the QA agent can perform, including Google search for finding information.

Returns:: List of action objects that the agent can use.
Return type:: List[BaseAction]

Example

>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> agent = QAAgent()
>>> actions = agent.create_actions()
>>> print(len(actions))
1
>>> print(actions[0].__class__.__name__)
GoogleSearch

synthesize_output()[source]#

Generate the final answer based on the agent’s actions and belief state.

This method creates a SynthesizeOutput action and executes it with the current task, context, and internal history to produce a coherent response.

Returns:: The synthesized answer to the question.
Return type:: str

Example

>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> agent = QAAgent()
>>> agent.belief.current_task.content = "What is machine learning?"
>>> # In a real scenario, this would generate a response based on
>>> # the agent's actions and belief state
>>> # result = agent.synthesize_output()
>>> # print(result)

sherpa_ai.agents.user module#

class sherpa_ai.agents.user.UserAgent(**data)[source]#

Bases: BaseAgent

A specialized agent that redirects tasks to human users.

This agent serves as a bridge between the AI system and human users, allowing tasks to be delegated to human experts when automated processing is not suitable or when human input is required.

name#

The name of the agent, defaults to “User”.

Type:: str

description#

A description of the agent’s purpose, defaults to “A user agent that redirects the task to an expert”.

Type:: str

user_id#

The identifier of the human user to redirect tasks to.

Type:: str

verbose_logger#

Logger for displaying messages to the user.

Type:: Any

Example

>>> from sherpa_ai.agents.user import UserAgent
>>> agent = UserAgent(user_id="expert1")
>>> agent.belief.current_task.content = "Analyze this data"
>>> result = agent.run()
>>> print(result)
@expert1 Please complete the following task:
Analyze this data

name: str#

description: str#

create_actions()[source]#

Create an empty list of actions as this agent delegates to humans.

Returns:: An empty list since this agent doesn’t perform actions.
Return type:: List[BaseAction]

Example

>>> from sherpa_ai.agents.user import UserAgent
>>> agent = UserAgent()
>>> actions = agent.create_actions()
>>> print(len(actions))
0

synthesize_output()[source]#

Return an empty string as this agent doesn’t synthesize output.

Returns:: An empty string.
Return type:: str

Example

>>> from sherpa_ai.agents.user import UserAgent
>>> agent = UserAgent()
>>> output = agent.synthesize_output()
>>> print(output)

run()[source]#

Redirect the task to a human user and return their response.

This method displays the task to the human user and waits for their input. The user’s response is then stored in the shared memory and returned.

Returns:: The human user’s response to the task.
Return type:: str

Example

>>> from sherpa_ai.agents.user import UserAgent
>>> agent = UserAgent(user_id="expert1")
>>> agent.belief.current_task.content = "Review this code"
>>> # In a real scenario, this would prompt the user and wait for input
>>> # result = agent.run()
>>> # print(result)

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Module contents#

class sherpa_ai.agents.AgentPool[source]#

Bases: object

A collection of agents that can be managed and accessed by name.

This class provides a centralized registry for agents, allowing them to be added, retrieved, and listed. It maintains a dictionary of agents indexed by their names for easy access.

This is the base class for in-memory agent pools. For persistent storage, use PersistentAgentPool which extends this class.

agents#

Dictionary mapping agent names to agent instances.

Type:: Dict[str, BaseAgent]

Example

>>> from sherpa_ai.agents.agent_pool import AgentPool
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> pool = AgentPool()
>>> agent = QAAgent(name="Research Assistant")
>>> pool.add_agent(agent)
>>> print("Research Assistant" in pool)
True

get_agent(agent_name)[source]#

Retrieve an agent by its name.

Parameters:: agent_name (str) – The name of the agent to retrieve.
Returns:: The agent with the specified name, or None if not found.
Return type:: Optional[BaseAgent]

Example

>>> from sherpa_ai.agents.agent_pool import AgentPool
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> pool = AgentPool()
>>> agent = QAAgent(name="Research Assistant")
>>> pool.add_agent(agent)
>>> retrieved = pool.get_agent("Research Assistant")
>>> print(retrieved.name)
Research Assistant

add_agent(agent)[source]#

Add a single agent to the pool.

Parameters:: agent (BaseAgent) – The agent to add to the pool.

Example

>>> from sherpa_ai.agents.agent_pool import AgentPool
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> pool = AgentPool()
>>> agent = QAAgent(name="Research Assistant")
>>> pool.add_agent(agent)
>>> print(len(pool.agents))
1

add_agents(agents)[source]#

Add multiple agents to the pool.

Parameters:: agents (List[BaseAgent]) – List of agents to add to the pool.

Example

>>> from sherpa_ai.agents.agent_pool import AgentPool
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> from sherpa_ai.agents.ml_engineer import MLEngineer
>>> pool = AgentPool()
>>> agents = [
...     QAAgent(name="Research Assistant"),
...     MLEngineer(name="AI Expert")
... ]
>>> pool.add_agents(agents)
>>> print(len(pool.agents))
2

get_agent_pool_description()[source]#

Create a description of all agents in the pool.

This method generates a formatted string containing the name and description of each agent in the pool, which can be used for agent planning or display.

Returns:: A formatted string describing all agents in the pool.
Return type:: str

Example

>>> from sherpa_ai.agents.agent_pool import AgentPool
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> pool = AgentPool()
>>> agent = QAAgent(name="Research Assistant")
>>> pool.add_agent(agent)
>>> description = pool.get_agent_pool_description()
>>> print("Research Assistant" in description)
True

class sherpa_ai.agents.Physicist(*args, name: str = 'Physicist', description: str = None, shared_memory: ~sherpa_ai.memory.shared_memory.SharedMemory = None, belief: ~sherpa_ai.memory.belief.Belief = None, policy: ~sherpa_ai.policies.base.BasePolicy = None, num_runs: int = 3, actions: ~typing.List[~sherpa_ai.actions.base.BaseAction] = [], validation_steps: int = 1, validations: ~typing.List[~sherpa_ai.output_parsers.base.BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, llm: ~langchain_core.language_models.base.BaseLanguageModel | None = None, prompt_template: ~sherpa_ai.prompts.prompt_template_loader.PromptTemplate = <sherpa_ai.prompts.prompt_template_loader.PromptTemplate object>, stop_checker: ~typing.Callable[[~sherpa_ai.memory.belief.Belief], bool] = <function BaseAgent.<lambda>>, **kwargs)[source]#

Bases: BaseAgent

A specialized agent for answering questions about physics topics.

This agent is designed to handle questions and research tasks related to physics, including classical mechanics, quantum mechanics, relativity, and other physical theories. It can search for information and synthesize comprehensive responses.

name#

The name of the agent, defaults to “Physicist”.

Type:: str

description#

A description of the agent’s purpose and capabilities.

Type:: str

num_runs#

Number of action execution cycles, defaults to 3.

Type:: int

Example

>>> from sherpa_ai.agents.physicist import Physicist
>>> agent = Physicist(name="Quantum Expert")
>>> print(agent.name)
Quantum Expert

name: str#

num_runs: int#

description: str#

create_actions()[source]#

Create and return the list of actions available to this agent.

This method defines the specific actions that the Physicist agent can perform, including deliberation and Google search for finding information.

Returns:: List of action objects that the agent can use.
Return type:: List[BaseAction]

Example

>>> from sherpa_ai.agents.physicist import Physicist
>>> agent = Physicist()
>>> actions = agent.create_actions()
>>> print(len(actions))
2
>>> print([action.__class__.__name__ for action in actions])
['Deliberation', 'GoogleSearch']

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

synthesize_output()[source]#

Generate the final answer based on the agent’s actions and belief state.

This method creates a SynthesizeOutput action and executes it with the current task, context, and internal history to produce a coherent response.

Returns:: The synthesized answer to the physics question.
Return type:: str

Example

>>> from sherpa_ai.agents.physicist import Physicist
>>> agent = Physicist()
>>> agent.belief.current_task.content = "Explain quantum entanglement"
>>> # In a real scenario, this would generate a response based on
>>> # the agent's actions and belief state
>>> # result = agent.synthesize_output()
>>> # print(result)

class sherpa_ai.agents.MLEngineer(*args, name: str = 'ML Engineer', description: str = None, shared_memory: ~sherpa_ai.memory.shared_memory.SharedMemory = None, belief: ~sherpa_ai.memory.belief.Belief = None, policy: ~sherpa_ai.policies.base.BasePolicy = None, num_runs: int = 3, actions: ~typing.List[~sherpa_ai.actions.base.BaseAction] = [], validation_steps: int = 1, validations: ~typing.List[~sherpa_ai.output_parsers.base.BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, llm: ~langchain_core.language_models.base.BaseLanguageModel | None = None, prompt_template: ~sherpa_ai.prompts.prompt_template_loader.PromptTemplate = <sherpa_ai.prompts.prompt_template_loader.PromptTemplate object>, stop_checker: ~typing.Callable[[~sherpa_ai.memory.belief.Belief], bool] = <function BaseAgent.<lambda>>, **kwargs)[source]#

Bases: BaseAgent

A specialized agent for answering questions about machine learning topics.

This agent is designed to handle questions and research tasks related to machine learning, artificial intelligence, and data science. It can search for information from various sources and synthesize comprehensive responses.

name#

The name of the agent, defaults to “ML Engineer”.

Type:: str

description#

A description of the agent’s purpose and capabilities.

Type:: str

num_runs#

Number of action execution cycles, defaults to 3.

Type:: int

Example

>>> from sherpa_ai.agents.ml_engineer import MLEngineer
>>> agent = MLEngineer(name="AI Researcher")
>>> print(agent.name)
AI Researcher

name: str#

num_runs: int#

description: str#

create_actions()[source]#

Create and return the list of actions available to this agent.

This method defines the specific actions that the ML Engineer agent can perform, including deliberation, Google search, and arXiv search for finding information.

Returns:: List of action objects that the agent can use.
Return type:: List[BaseAction]

Example

>>> from sherpa_ai.agents.ml_engineer import MLEngineer
>>> agent = MLEngineer()
>>> actions = agent.create_actions()
>>> print(len(actions))
3
>>> print([action.__class__.__name__ for action in actions])
['Deliberation', 'GoogleSearch', 'ArxivSearch']

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

synthesize_output()[source]#

Generate the final answer based on the agent’s actions and belief state.

This method creates a SynthesizeOutput action and executes it with the current task, context, and internal history to produce a coherent response.

Returns:: The synthesized answer to the machine learning question.
Return type:: str

Example

>>> from sherpa_ai.agents.ml_engineer import MLEngineer
>>> agent = MLEngineer()
>>> agent.belief.current_task.content = "Explain neural networks"
>>> # In a real scenario, this would generate a response based on
>>> # the agent's actions and belief state
>>> # result = agent.synthesize_output()
>>> # print(result)

class sherpa_ai.agents.UserAgent(**data)[source]#

Bases: BaseAgent

A specialized agent that redirects tasks to human users.

This agent serves as a bridge between the AI system and human users, allowing tasks to be delegated to human experts when automated processing is not suitable or when human input is required.

name#

The name of the agent, defaults to “User”.

Type:: str

description#

A description of the agent’s purpose, defaults to “A user agent that redirects the task to an expert”.

Type:: str

user_id#

The identifier of the human user to redirect tasks to.

Type:: str

verbose_logger#

Logger for displaying messages to the user.

Type:: Any

Example

>>> from sherpa_ai.agents.user import UserAgent
>>> agent = UserAgent(user_id="expert1")
>>> agent.belief.current_task.content = "Analyze this data"
>>> result = agent.run()
>>> print(result)
@expert1 Please complete the following task:
Analyze this data

name: str#

description: str#

create_actions()[source]#

Create an empty list of actions as this agent delegates to humans.

Returns:: An empty list since this agent doesn’t perform actions.
Return type:: List[BaseAction]

Example

>>> from sherpa_ai.agents.user import UserAgent
>>> agent = UserAgent()
>>> actions = agent.create_actions()
>>> print(len(actions))
0

synthesize_output()[source]#

Return an empty string as this agent doesn’t synthesize output.

Returns:: An empty string.
Return type:: str

Example

>>> from sherpa_ai.agents.user import UserAgent
>>> agent = UserAgent()
>>> output = agent.synthesize_output()
>>> print(output)

run()[source]#

Redirect the task to a human user and return their response.

This method displays the task to the human user and waits for their input. The user’s response is then stored in the shared memory and returned.

Returns:: The human user’s response to the task.
Return type:: str

Example

>>> from sherpa_ai.agents.user import UserAgent
>>> agent = UserAgent(user_id="expert1")
>>> agent.belief.current_task.content = "Review this code"
>>> # In a real scenario, this would prompt the user and wait for input
>>> # result = agent.run()
>>> # print(result)

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sherpa_ai.agents.QAAgent(*args, custom_task_description_prompt=None, custom_action_plan_prompt=None, **kwargs)[source]#

Bases: BaseAgent

A specialized agent for answering questions and providing information.

This agent is designed to handle question-answering tasks by searching for information, synthesizing responses, and validating outputs. It can optionally include citations in its responses.
Attributes:
name (str): The name of the agent, defaults to “QA Agent”. description (str): A description of the agent’s purpose and capabilities. config (AgentConfig): Configuration settings for the agent. num_runs (int): Number of action execution cycles, defaults to 3. global_regen_max (int): Maximum number of output regeneration attempts, defaults to 5. citation_enabled (bool): Whether to include citations in responses, defaults to False.

Args:
custom_task_description_prompt (Optional[str]): Custom prompt for task description. If None, uses default from prompt template. custom_action_plan_prompt (Optional[str]): Custom prompt for action plan description. If None, uses default from prompt template.

Example:
>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> from sherpa_ai.config import AgentConfig
>>> agent = QAAgent(
...     name="Research Assistant",
...     config=AgentConfig(),
...     citation_enabled=True
... )
>>> print(agent.name)
Research Assistant
>>> # Using custom prompts
>>> agent = QAAgent(
...     name="Custom QA",
...     custom_task_description_prompt="You are an expert QA assistant.",
...     custom_action_plan_prompt="Plan to solve: {task}"
... )
>>> print(agent.description)
You are an expert QA assistant.

Your name is Custom QA.

name: str#

num_runs: int#

global_regen_max: int#

description: str#

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow'}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

config: AgentConfig#

citation_enabled: bool#

create_actions()[source]#

Create and return the list of actions available to this agent.

This method defines the specific actions that the QA agent can perform, including Google search for finding information.

Returns:: List of action objects that the agent can use.
Return type:: List[BaseAction]

Example

>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> agent = QAAgent()
>>> actions = agent.create_actions()
>>> print(len(actions))
1
>>> print(actions[0].__class__.__name__)
GoogleSearch

synthesize_output()[source]#

Generate the final answer based on the agent’s actions and belief state.

This method creates a SynthesizeOutput action and executes it with the current task, context, and internal history to produce a coherent response.

Returns:: The synthesized answer to the question.
Return type:: str

Example

>>> from sherpa_ai.agents.qa_agent import QAAgent
>>> agent = QAAgent()
>>> agent.belief.current_task.content = "What is machine learning?"
>>> # In a real scenario, this would generate a response based on
>>> # the agent's actions and belief state
>>> # result = agent.synthesize_output()
>>> # print(result)