sherpa_ai.agents package

Submodules

sherpa_ai.agents.agent_pool module

class sherpa_ai.agents.agent_pool.AgentPool

Bases: object

add_agent(agent: BaseAgent)

Add agent to agent pool

add_agents(agents: List[BaseAgent])

Add agents to agent pool

get_agent(agent_name: str) → BaseAgent | None

Get agent by name

get_agent_pool_description() → str

Create a description (prompt) of the AgentPool for agent planning

sherpa_ai.agents.base module

class sherpa_ai.agents.base.BaseAgent(*, name: str, description: str, shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 1, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, **extra_data: Any)

Bases: ABC, BaseModel

act(action: BaseAction, inputs: dict) → str | None | Exception

actions: List[BaseAction]

agent_finished(result: str) → str

agent_preparation()

async async_act(action: BaseAction, inputs: dict) → str | None

async async_run()

async async_send_event(event: str, args: dict)

Send an event to the state machine in the belief

Parameters:

• event (str) – The event name

• args (dict) – The arguments for the event

belief: Belief

abstract create_actions() → List[BaseAction]

description: str

do_synthesize_output: bool

feedback_agent_name: str

global_regen_max: int

llm: Any

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

num_runs: int

observe()

policy: BasePolicy

run()

select_action() → PolicyOutput | None

send_event(event: str, args: dict)

Send an event to the state machine in the belief

Parameters:

• event (str) – The event name

• args (dict) – The arguments for the event

shared_memory: SharedMemory

abstract synthesize_output() → str

validate_output()

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

validation_iterator(validations, global_regen_count, all_pass, validation_is_scaped, result)

validation_steps: int

validations: List[BaseOutputProcessor]

sherpa_ai.agents.critic module

class sherpa_ai.agents.critic.Critic(*, name: str = 'Critic', description: str = '\nYou are a Critic agent that receive a plan from the planner to execuate a task from user.\nYour goal is to output the {} most necessary feedback given the corrent plan to solve the task.\n', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 1, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, ratio: float = 0.9, num_feedback: int = 3, **extra_data: Any)

Bases: BaseAgent

The critic agent receives a plan from the planner and evaluate the plan based on some pre-defined metrics. At the same time, it gives the feedback to the planner.

create_actions()

description: str

get_detail_evaluation(task: str, plan: str)

get_feedback(task: str, plan: str) → List[str]

get_importance_evaluation(task: str, plan: str)

get_insight()

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

num_feedback: int

post_process(feedback: str)

ratio: float

synthesize_output() → str

sherpa_ai.agents.ml_engineer module

class sherpa_ai.agents.ml_engineer.MLEngineer(*args, name: str = 'ML Engineer', description: str = 'You are a skilled machine learning engineer with a deep-rooted expertise in understanding and analyzing various Machine Learnng algorithm and use it to solve practical problems. Your primary role is to assist individuals, organizations, and researchers in using machine learning models to solve Machine-Learning-Related chalenges, using your knowledge to guide decisions and ensure the accuracy and reliability of outcomes.If you encounter a question or challenge outside your current knowledge base, you acknowledge your limitations and seek assistance or additional resources to fill the gaps. ', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 3, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, **kwargs)

Bases: BaseAgent

The machine learning agent answers questions or research about ML-related topics

create_actions() → List[BaseAction]

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].

name: str

num_runs: int

synthesize_output() → str

sherpa_ai.agents.physicist module

class sherpa_ai.agents.physicist.Physicist(*args, name: str = 'Physicist', description: str = 'You are a physicist with a deep-rooted expertise in understanding and analyzing the fundamental principles of the universe, spanning from the tiniest subatomic particles to vast cosmic phenomena. Your primary role is to assist individuals, organizations, and researchers in navigating and resolving complex physics-related challenges, using your knowledge to guide decisions and ensure the accuracy and reliability of outcomes.', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 3, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, **kwargs)

Bases: BaseAgent

The physicist agent answers questions or research about physics-related topics

create_actions() → List[BaseAction]

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].

name: str

num_runs: int

synthesize_output() → str

sherpa_ai.agents.planner module

class sherpa_ai.agents.planner.Planner(*args, name: str = 'Planner', description: str = 'You are a **task decomposition assistant** who simplifies complex tasks into sequential steps, assigning roles or agents to each.\nBy analyzing user-defined tasks and agent capabilities, you provide structured plans, enhancing project clarity and efficiency.\nYour adaptability ensures customized solutions for diverse needs.\n', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 5, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, planning: TaskPlanning = None, agent_pool: AgentPool, **kwargs)

Bases: BaseAgent

agent_pool: AgentPool

create_actions()

description: str

get_last_feedback() → str | None

Get the last feedback from the shared memory

get_last_plan() → str | None

Get the last plan from the shared memory

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

num_runs: int

plan(task: str)

Plan the next action for the agent pool

planning: TaskPlanning

synthesize_output() → str

sherpa_ai.agents.qa_agent module

class sherpa_ai.agents.qa_agent.QAAgent(*args, name: str = 'QA Agent', description: str = 'You are a **question answering assistant** who solves user questions and offers a detailed solution.', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 3, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 5, do_synthesize_output: bool = False, llm: Any = None, config: AgentConfig = None, citation_enabled: bool = False, **kwargs)

Bases: BaseAgent

The task agent is the agent handles a single task.

llm

The language model used to generate text

Type:

BaseLanguageModel

name

The name of the agent. Defaults to “QA Agent”.

Type:

str, optional

description

The description of the agent. Defaults to TASK_AGENT_DESCRIPTION.

Type:

str, optional

shared_memory

The shared memory used to store information and shared with other agents. Defaults to None.

Type:

SharedMemory, optional

belief

The belief of the agent. Defaults to None.

Type:

Optional[Belief], optional

agent_config

The agent configuration. Defaults to AgentConfig.

Type:

AgentConfig, optional

num_runs

The number of runs the agent will perform. Defaults to 3.

Type:

int, optional

verbose_logger

The verbose logger used to log information. Defaults to DummyVerboseLogger().

Type:

BaseVerboseLogger, optional

actions

The list of actions the agent can perform. Defaults to [].

Type:

List[BaseAction], optional

validation_steps

The number of validation steps the agent will perform. Defaults to 1.

Type:

int, optional

validations

The list of validations the agent will perform. Defaults to [].

Type:

List[BaseOutputProcessor], optional

citation_enabled: bool

config: AgentConfig

create_actions() → List[BaseAction]

description: str

global_regen_max: int

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

num_runs: int

synthesize_output() → str

sherpa_ai.agents.user module

class sherpa_ai.agents.user.UserAgent(*, name: str = 'User', description: str = 'A user agent that redirects the task to an expert', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 1, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, **extra_data: Any)

Bases: BaseAgent

A wrapper class for redirecting the task to a real person

create_actions() → List[BaseAction]

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].

name: str

run() → str

Redirect the task to a real person

synthesize_output() → str

Module contents

class sherpa_ai.agents.AgentPool

Bases: object

add_agent(agent: BaseAgent)

Add agent to agent pool

add_agents(agents: List[BaseAgent])

Add agents to agent pool

get_agent(agent_name: str) → BaseAgent | None

Get agent by name

get_agent_pool_description() → str

Create a description (prompt) of the AgentPool for agent planning

class sherpa_ai.agents.Critic(*, name: str = 'Critic', description: str = '\nYou are a Critic agent that receive a plan from the planner to execuate a task from user.\nYour goal is to output the {} most necessary feedback given the corrent plan to solve the task.\n', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 1, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, ratio: float = 0.9, num_feedback: int = 3, **extra_data: Any)

Bases: BaseAgent

The critic agent receives a plan from the planner and evaluate the plan based on some pre-defined metrics. At the same time, it gives the feedback to the planner.

actions: List[BaseAction]

belief: Belief

create_actions()

description: str

do_synthesize_output: bool

feedback_agent_name: str

get_detail_evaluation(task: str, plan: str)

get_feedback(task: str, plan: str) → List[str]

get_importance_evaluation(task: str, plan: str)

get_insight()

global_regen_max: int

llm: Any

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

num_feedback: int

num_runs: int

policy: BasePolicy

post_process(feedback: str)

ratio: float

shared_memory: SharedMemory

synthesize_output() → str

validation_steps: int

validations: List[BaseOutputProcessor]

class sherpa_ai.agents.MLEngineer(*args, name: str = 'ML Engineer', description: str = 'You are a skilled machine learning engineer with a deep-rooted expertise in understanding and analyzing various Machine Learnng algorithm and use it to solve practical problems. Your primary role is to assist individuals, organizations, and researchers in using machine learning models to solve Machine-Learning-Related chalenges, using your knowledge to guide decisions and ensure the accuracy and reliability of outcomes.If you encounter a question or challenge outside your current knowledge base, you acknowledge your limitations and seek assistance or additional resources to fill the gaps. ', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 3, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, **kwargs)

Bases: BaseAgent

The machine learning agent answers questions or research about ML-related topics

actions: List[BaseAction]

belief: Belief

create_actions() → List[BaseAction]

description: str

do_synthesize_output: bool

feedback_agent_name: str

global_regen_max: int

llm: Any

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

num_runs: int

policy: BasePolicy

shared_memory: SharedMemory

synthesize_output() → str

validation_steps: int

validations: List[BaseOutputProcessor]

class sherpa_ai.agents.Physicist(*args, name: str = 'Physicist', description: str = 'You are a physicist with a deep-rooted expertise in understanding and analyzing the fundamental principles of the universe, spanning from the tiniest subatomic particles to vast cosmic phenomena. Your primary role is to assist individuals, organizations, and researchers in navigating and resolving complex physics-related challenges, using your knowledge to guide decisions and ensure the accuracy and reliability of outcomes.', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 3, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, **kwargs)

Bases: BaseAgent

The physicist agent answers questions or research about physics-related topics

actions: List[BaseAction]

belief: Belief

create_actions() → List[BaseAction]

description: str

do_synthesize_output: bool

feedback_agent_name: str

global_regen_max: int

llm: Any

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

num_runs: int

policy: BasePolicy

shared_memory: SharedMemory

synthesize_output() → str

validation_steps: int

validations: List[BaseOutputProcessor]

class sherpa_ai.agents.Planner(*args, name: str = 'Planner', description: str = 'You are a **task decomposition assistant** who simplifies complex tasks into sequential steps, assigning roles or agents to each.\nBy analyzing user-defined tasks and agent capabilities, you provide structured plans, enhancing project clarity and efficiency.\nYour adaptability ensures customized solutions for diverse needs.\n', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 5, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, planning: TaskPlanning = None, agent_pool: AgentPool, **kwargs)

Bases: BaseAgent

actions: List[BaseAction]

agent_pool: AgentPool

belief: Belief

create_actions()

description: str

do_synthesize_output: bool

feedback_agent_name: str

get_last_feedback() → str | None

Get the last feedback from the shared memory

get_last_plan() → str | None

Get the last plan from the shared memory

global_regen_max: int

llm: Any

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

num_runs: int

plan(task: str)

Plan the next action for the agent pool

planning: TaskPlanning

policy: BasePolicy

shared_memory: SharedMemory

synthesize_output() → str

validation_steps: int

validations: List[BaseOutputProcessor]

class sherpa_ai.agents.QAAgent(*args, name: str = 'QA Agent', description: str = 'You are a **question answering assistant** who solves user questions and offers a detailed solution.', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 3, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 5, do_synthesize_output: bool = False, llm: Any = None, config: AgentConfig = None, citation_enabled: bool = False, **kwargs)

Bases: BaseAgent

The task agent is the agent handles a single task.

llm

The language model used to generate text

Type:

BaseLanguageModel

name

The name of the agent. Defaults to “QA Agent”.

Type:

str, optional

description

The description of the agent. Defaults to TASK_AGENT_DESCRIPTION.

Type:

str, optional

shared_memory

The shared memory used to store information and shared with other agents. Defaults to None.

Type:

SharedMemory, optional

belief

The belief of the agent. Defaults to None.

Type:

Optional[Belief], optional

agent_config

The agent configuration. Defaults to AgentConfig.

Type:

AgentConfig, optional

num_runs

The number of runs the agent will perform. Defaults to 3.

Type:

int, optional

verbose_logger

The verbose logger used to log information. Defaults to DummyVerboseLogger().

Type:

BaseVerboseLogger, optional

actions

The list of actions the agent can perform. Defaults to [].

Type:

List[BaseAction], optional

validation_steps

The number of validation steps the agent will perform. Defaults to 1.

Type:

int, optional

validations

The list of validations the agent will perform. Defaults to [].

Type:

List[BaseOutputProcessor], optional

actions: List[BaseAction]

belief: Belief

citation_enabled: bool

config: AgentConfig

create_actions() → List[BaseAction]

description: str

do_synthesize_output: bool

feedback_agent_name: str

global_regen_max: int

llm: Any

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

num_runs: int

policy: BasePolicy

shared_memory: SharedMemory

synthesize_output() → str

validation_steps: int

validations: List[BaseOutputProcessor]

class sherpa_ai.agents.UserAgent(*, name: str = 'User', description: str = 'A user agent that redirects the task to an expert', shared_memory: SharedMemory = None, belief: Belief = None, policy: BasePolicy = None, num_runs: int = 1, actions: List[BaseAction] = [], validation_steps: int = 1, validations: List[BaseOutputProcessor] = [], feedback_agent_name: str = 'critic', global_regen_max: int = 12, do_synthesize_output: bool = False, llm: Any = None, **extra_data: Any)

Bases: BaseAgent

A wrapper class for redirecting the task to a real person

actions: List[BaseAction]

belief: Belief

create_actions() → List[BaseAction]

description: str

do_synthesize_output: bool

feedback_agent_name: str

global_regen_max: int

llm: Any

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

num_runs: int

policy: BasePolicy

run() → str

Redirect the task to a real person

shared_memory: SharedMemory

synthesize_output() → str

validation_steps: int

validations: List[BaseOutputProcessor]