sherpa_ai.memory package

In This Page:

sherpa_ai.memory package#

Overview#

The memory package provides persistence and knowledge management capabilities for Sherpa AI agents, enabling them to retain information across sessions and share knowledge between different components.

Key Components

Belief: Structures for representing agent beliefs and knowledge
SharedMemory: Mechanisms for sharing information between agents
Knowledge Management: Tools for organizing and retrieving stored information

Example Usage#

from sherpa_ai.memory.shared_memory import SharedMemory
from sherpa_ai.memory.belief import Belief

# Create a shared memory instance
memory = SharedMemory()

# Create and store a belief
belief = Belief(
    content="The Earth orbits the Sun.",
    source="Astronomy facts",
    confidence=0.99
)

# Add the belief to memory
memory.add_belief(belief)

# Retrieve beliefs on a topic
astronomy_beliefs = memory.get_beliefs_by_topic("astronomy")
print(f"Found {len(astronomy_beliefs)} beliefs about astronomy")

Submodules#

Module	Description
`sherpa_ai.memory.belief`	Provides the Belief class for representing and managing agent knowledge.
`sherpa_ai.memory.shared_memory`	Implements the SharedMemory system for knowledge persistence and sharing.

sherpa_ai.memory.belief module#

Belief management module for Sherpa AI.

This module provides belief tracking functionality for agents in the Sherpa AI system. It defines the Belief class which manages agent state, events, and internal reasoning.

class sherpa_ai.memory.belief.Belief(**data)[source]#

Bases: BaseModel

Manages agent beliefs and state tracking.

This class maintains the agent’s belief state, including observed events, internal reasoning events, and current task information. It provides methods for updating and retrieving belief state, managing events, and handling agent actions.

events#

List of events observed by the agent.

Type:: List[Event]

internal_events#

List of internal events from agent reasoning.

Type:: List[Event]

current_task#

The current task being processed.

Type:: Event

state_machine#

State machine for managing agent state.

Type:: SherpaStateMachine

actions#

List of available actions.

Type:: List[BaseAction]

dict#

Dictionary for storing arbitrary belief data.

Type:: dict

max_tokens#

Maximum number of tokens for context/history.

Type:: int

Example

>>> belief = Belief()
>>> belief.update(observation_event)
>>> belief.set_current_task("Analyze the data")
>>> context = belief.get_context(token_counter)
>>> print(context)
'current_task: Analyze the data(task)'

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

events: List[Event]#

internal_events: List[Event]#

current_task: Event#

state_machine: SherpaStateMachine#

actions: List#

max_tokens: int#

update(observation)[source]#

Update belief with a new observation event.

Parameters:: observation (Event) – The event to add to the belief state.

Example

>>> belief = Belief()
>>> event = Event("observation", "user_input", "Hello")
>>> belief.update(event)
>>> print(belief.events)
[Event("observation", "user_input", "Hello")]

get_context(token_counter)[source]#

Get the context of the agent’s belief state.

Parameters:: token_counter (Callable[[str], int]) – Function to count tokens in text.
Returns:: Context string containing relevant events, truncated if exceeding max_tokens.
Return type:: str

Example

>>> belief = Belief()
>>> def count_tokens(text): return len(text.split())
>>> belief.set_current_task("Analyze data")
>>> context = belief.get_context(count_tokens)
>>> print(context)
'current_task: Analyze data(task)'

update_internal(event_type, name, **kwargs)[source]#

Add an internal event to the belief state.

Parameters:

event_type (str) – Type of the internal event.
name (str) – Name of the event.
**kwargs – Additional event parameters.

Example

>>> belief = Belief()
>>> belief.update_internal("reasoning", "analysis", content="Processing data")
>>> print(belief.internal_events[0].event_type)
'reasoning'

get_by_type(event_type)[source]#

Get all internal events of a specific type.

Parameters:: event_type (str) – Type of events to retrieve.
Returns:: List of internal events matching the type.
Return type:: List[Event]

Example

>>> belief = Belief()
>>> belief.update_internal("reasoning", "analysis")
>>> events = belief.get_by_type("reasoning")
>>> print(len(events))
1

set_current_task(content)[source]#

Set the current task in the belief state.

Parameters:: content (str) – Content of the current task.

Example

>>> belief = Belief()
>>> belief.set_current_task("Process data")
>>> print(belief.current_task.content)
'Process data'

get_internal_history(token_counter)[source]#

Get the internal history of the agent.

Parameters:: token_counter (Callable[[str], int]) – Function to count tokens in text.
Returns:: Internal history string, truncated if exceeding max_tokens.
Return type:: str

Example

>>> belief = Belief()
>>> def count_tokens(text): return len(text.split())
>>> belief.update_internal("reasoning", "analysis")
>>> history = belief.get_internal_history(count_tokens)
>>> print(history)
'analysis(reasoning)'

clear_short_term_memory()[source]#

Clear short-term memory by removing all internal events and dictionary data.

Example

>>> belief = Belief()
>>> belief.update_internal("reasoning", "analysis")
>>> belief.set("key", "value")
>>> belief.clear_short_term_memory()
>>> print(len(belief.internal_events))
0
>>> print(belief.dict)
{}

get_histories_excluding_types(exclude_types, token_counter=None, max_tokens=4000)[source]#

Get internal history excluding specific event types.

Parameters:

exclude_types (list[str]) – List of event types to exclude.
token_counter (Optional[Callable[[str], int]]) – Function to count tokens.
max_tokens (int) – Maximum number of tokens in result.

Returns:

Filtered history string, truncated if exceeding max_tokens.

Return type:

str

Example

>>> belief = Belief()
>>> def count_tokens(text): return len(text.split())
>>> belief.update_internal("reasoning", "analysis")
>>> belief.update_internal("feedback", "comment")
>>> history = belief.get_histories_excluding_types(["feedback"], count_tokens)
>>> print(history)
'analysis(reasoning)'

set_actions(actions)[source]#

Set available actions for the agent.

Parameters:: actions (List[BaseAction]) – List of actions to make available.

Example

>>> belief = Belief()
>>> actions = [Action1(), Action2()]
>>> belief.set_actions(actions)
>>> print(len(belief.actions))
2

property action_description#

Get string description of all available actions.

Returns:
str: Newline-separated string of action descriptions.

Example:
>>> belief = Belief()
>>> belief.set_actions([Action1(), Action2()])
>>> print(belief.action_description)
'Action1: Description1

Action2: Description2’

get_state()[source]#

Get current state name from state machine.

Returns:: Name of current state, or None if no state machine exists.
Return type:: str

Example

>>> belief = Belief()
>>> belief.state_machine = StateMachine()
>>> print(belief.get_state())
'initial'

get_state_obj()[source]#

Get current state object from state machine.

Returns:: Current state object, or None if no state machine exists.
Return type:: ts.State

Example

>>> belief = Belief()
>>> belief.state_machine = StateMachine()
>>> state = belief.get_state_obj()
>>> print(state.name)
'initial'

get_actions()[source]#

Get list of available actions.

Returns:: List of available actions.
Return type:: List[BaseAction]

Example

>>> belief = Belief()
>>> belief.set_actions([Action1(), Action2()])
>>> actions = belief.get_actions()
>>> print(len(actions))
2

get_action(action_name)[source]#

Get specific action by name.

Parameters:: action_name (str) – Name of action to retrieve.
Returns:: Action with matching name, or None if not found.
Return type:: BaseAction

Example

>>> belief = Belief()
>>> belief.set_actions([Action1(name="action1")])
>>> action = belief.get_action("action1")
>>> print(action.name)
'action1'

async async_get_actions()[source]#

Asynchronously get list of available actions.

Returns:: List of available actions.
Return type:: List[BaseAction]

Example

>>> belief = Belief()
>>> belief.set_actions([Action1(), Action2()])
>>> actions = await belief.async_get_actions()
>>> print(len(actions))
2

async async_get_action(action_name)[source]#

Asynchronously get specific action by name.

Parameters:: action_name (str) – Name of action to retrieve.
Returns:: Action with matching name, or None if not found.
Return type:: BaseAction

Example

>>> belief = Belief()
>>> belief.set_actions([Action1(name="action1")])
>>> action = await belief.async_get_action("action1")
>>> print(action.name)
'action1'

get_dict()[source]#

Get the belief dictionary.

Returns:: Dictionary containing belief data.
Return type:: dict

Example

>>> belief = Belief()
>>> belief.set("key", "value")
>>> print(belief.get_dict())
{'key': 'value'}

get(key, default=None)[source]#

Get value from belief dictionary using dot notation.

Parameters:

key (str) – Key to retrieve, can use dot notation for nested values.
default (Any) – Default value if key not found.

Returns:

Value at key, or default if not found.

Return type:

Any

Example

>>> belief = Belief()
>>> belief.set("nested.key", "value")
>>> print(belief.get("nested.key"))
'value'
>>> print(belief.get("missing.key", "default"))
'default'

get_all_keys()[source]#

Get all keys in belief dictionary, including nested keys.

Returns:: List of all keys, using dot notation for nested keys.
Return type:: List[str]

Example

>>> belief = Belief()
>>> belief.set("nested.key", "value")
>>> print(belief.get_all_keys())
['nested.key']

has(key)[source]#

Check if key exists in belief dictionary.

Parameters:: key (str) – Key to check, can use dot notation for nested values.
Returns:: True if key exists, False otherwise.
Return type:: bool

Example

>>> belief = Belief()
>>> belief.set("key", "value")
>>> print(belief.has("key"))
True
>>> print(belief.has("missing.key"))
False

set(key, value)[source]#

Set value in belief dictionary using dot notation.

Parameters:

key (str) – Key to set, can use dot notation for nested values.
value (Any) – Value to store.

Example

>>> belief = Belief()
>>> belief.set("nested.key", "value")
>>> print(belief.get("nested.key"))
'value'

sherpa_ai.memory.shared_memory module#

Shared memory management module for Sherpa AI.

This module provides shared memory functionality for the Sherpa AI system. It defines the SharedMemory class which manages shared state between agents.

class sherpa_ai.memory.shared_memory.SharedMemory(objective='')[source]#

Bases: object

Manages shared memory between agents in the Sherpa AI system.

This class maintains a shared state that can be accessed and modified by multiple agents. It tracks events, plans, and current execution steps, providing methods for adding events and synchronizing state with agent beliefs. Note that the agent names registered in the shared memory must be unique

objective#

The overall objective being pursued.

Type:: str

events#

List of events in shared memory.

Type:: List[Event]

event_type_subscriptions#

Agent subscriptions to specific event types.

Type:: dict[type[Event], list[BaseAgent]]

sender_subscriptions#

Agent subscriptions based on sender.

Type:: dict[str, list[BaseAgent]]

_lock#

Reentrant lock for thread safety.

Type:: threading.RLock

async add_event(event)[source]#

Add an event to shared memory.

Parameters:: event (Event) – Event to add to shared memory.

Example

>>> memory = SharedMemory("Complete the task")
>>> event = Event("task", "initial_task", "Process data")
>>> memory.add_event(event)
>>> print(len(memory.events))
1

add(event_type, name, sender='', **kwargs)[source]#

async async_add(event_type, name, sender='', **kwargs)[source]#

Create and add an event to shared memory.

Parameters:

event_type (str) – Type of the event.
name (str) – Name of the event.
**kwargs – Additional event parameters.

subscribe_event_type(event_type, agent)[source]#

Subscribe an agent to a specific event type. :type event_type: str :param event_type: Type of event to subscribe to. :type event_type: str :type agent: BaseAgent :param agent: Agent subscribing to the event type. :type agent: BaseAgent

Example

>>> memory = SharedMemory("Complete the task")
>>> agent = BaseAgent()
>>> memory.subscribe("trigger", agent)
>>> print(len(memory.event_type_subscriptions))
1

subscribe_sender(sender, agent)[source]#

Subscribe an agent to events from a specific sender.

Parameters:

sender (str) – Sender of the events.
agent (BaseAgent) – Agent subscribing to the sender.

Example

>>> memory = SharedMemory("Complete the task")
>>> agent = BaseAgent()
>>> memory.subscribe_sender("agent1", agent)
>>> print(len(memory.sender_subscriptions))
1

get_by_type(event_type)[source]#

Get all events of a specific type.

Parameters:: event_type (str) – Type of events to retrieve.
Returns:: List of events matching the type.
Return type:: List[Event]

Example

>>> memory = SharedMemory("Complete the task")
>>> memory.add("task", "task1", content="First task")
>>> memory.add("task", "task2", content="Second task")
>>> tasks = memory.get_by_type("task")
>>> print(len(tasks))
2

Module contents#

Memory management module for Sherpa AI.

This module provides memory management functionality for the Sherpa AI system. It exports the SharedMemory and Belief classes which handle state management and belief tracking for agents.

Example

>>> from sherpa_ai.memory import SharedMemory, Belief
>>> memory = SharedMemory()
>>> belief = Belief()
>>> memory.store("key", "value")
>>> belief.update("observation", "action")

class sherpa_ai.memory.SharedMemory(objective='')[source]#

Bases: object

Manages shared memory between agents in the Sherpa AI system.

This class maintains a shared state that can be accessed and modified by multiple agents. It tracks events, plans, and current execution steps, providing methods for adding events and synchronizing state with agent beliefs. Note that the agent names registered in the shared memory must be unique

objective#

The overall objective being pursued.

Type:: str

events#

List of events in shared memory.

Type:: List[Event]

event_type_subscriptions#

Agent subscriptions to specific event types.

Type:: dict[type[Event], list[BaseAgent]]

sender_subscriptions#

Agent subscriptions based on sender.

Type:: dict[str, list[BaseAgent]]

_lock#

Reentrant lock for thread safety.

Type:: threading.RLock

add(event_type, name, sender='', **kwargs)[source]#

async add_event(event)[source]#

Add an event to shared memory.

Parameters:: event (Event) – Event to add to shared memory.

Example

>>> memory = SharedMemory("Complete the task")
>>> event = Event("task", "initial_task", "Process data")
>>> memory.add_event(event)
>>> print(len(memory.events))
1

async async_add(event_type, name, sender='', **kwargs)[source]#

Create and add an event to shared memory.

Parameters:

event_type (str) – Type of the event.
name (str) – Name of the event.
**kwargs – Additional event parameters.

get_by_type(event_type)[source]#

Get all events of a specific type.

Parameters:: event_type (str) – Type of events to retrieve.
Returns:: List of events matching the type.
Return type:: List[Event]

Example

>>> memory = SharedMemory("Complete the task")
>>> memory.add("task", "task1", content="First task")
>>> memory.add("task", "task2", content="Second task")
>>> tasks = memory.get_by_type("task")
>>> print(len(tasks))
2

subscribe_event_type(event_type, agent)[source]#

Subscribe an agent to a specific event type. :type event_type: str :param event_type: Type of event to subscribe to. :type event_type: str :type agent: BaseAgent :param agent: Agent subscribing to the event type. :type agent: BaseAgent

Example

>>> memory = SharedMemory("Complete the task")
>>> agent = BaseAgent()
>>> memory.subscribe("trigger", agent)
>>> print(len(memory.event_type_subscriptions))
1

subscribe_sender(sender, agent)[source]#

Subscribe an agent to events from a specific sender.

Parameters:

sender (str) – Sender of the events.
agent (BaseAgent) – Agent subscribing to the sender.

Example

>>> memory = SharedMemory("Complete the task")
>>> agent = BaseAgent()
>>> memory.subscribe_sender("agent1", agent)
>>> print(len(memory.sender_subscriptions))
1

class sherpa_ai.memory.Belief(**data)[source]#

Bases: BaseModel

Manages agent beliefs and state tracking.

This class maintains the agent’s belief state, including observed events, internal reasoning events, and current task information. It provides methods for updating and retrieving belief state, managing events, and handling agent actions.

events#

List of events observed by the agent.

Type:: List[Event]

internal_events#

List of internal events from agent reasoning.

Type:: List[Event]

current_task#

The current task being processed.

Type:: Event

state_machine#

State machine for managing agent state.

Type:: SherpaStateMachine

actions#

List of available actions.

Type:: List[BaseAction]

dict#

Dictionary for storing arbitrary belief data.

Type:: dict

max_tokens#

Maximum number of tokens for context/history.

Type:: int

Example

>>> belief = Belief()
>>> belief.update(observation_event)
>>> belief.set_current_task("Analyze the data")
>>> context = belief.get_context(token_counter)
>>> print(context)
'current_task: Analyze the data(task)'

property action_description#

Get string description of all available actions.

Returns:
str: Newline-separated string of action descriptions.

Example:
>>> belief = Belief()
>>> belief.set_actions([Action1(), Action2()])
>>> print(belief.action_description)
'Action1: Description1

Action2: Description2’

async async_get_action(action_name)[source]#

Asynchronously get specific action by name.

Parameters:: action_name (str) – Name of action to retrieve.
Returns:: Action with matching name, or None if not found.
Return type:: BaseAction

Example

>>> belief = Belief()
>>> belief.set_actions([Action1(name="action1")])
>>> action = await belief.async_get_action("action1")
>>> print(action.name)
'action1'

async async_get_actions()[source]#

Asynchronously get list of available actions.

Returns:: List of available actions.
Return type:: List[BaseAction]

Example

>>> belief = Belief()
>>> belief.set_actions([Action1(), Action2()])
>>> actions = await belief.async_get_actions()
>>> print(len(actions))
2

clear_short_term_memory()[source]#

Clear short-term memory by removing all internal events and dictionary data.

Example

>>> belief = Belief()
>>> belief.update_internal("reasoning", "analysis")
>>> belief.set("key", "value")
>>> belief.clear_short_term_memory()
>>> print(len(belief.internal_events))
0
>>> print(belief.dict)
{}

get(key, default=None)[source]#

Get value from belief dictionary using dot notation.

Parameters:

key (str) – Key to retrieve, can use dot notation for nested values.
default (Any) – Default value if key not found.

Returns:

Value at key, or default if not found.

Return type:

Any

Example

>>> belief = Belief()
>>> belief.set("nested.key", "value")
>>> print(belief.get("nested.key"))
'value'
>>> print(belief.get("missing.key", "default"))
'default'

get_action(action_name)[source]#

Get specific action by name.

Parameters:: action_name (str) – Name of action to retrieve.
Returns:: Action with matching name, or None if not found.
Return type:: BaseAction

Example

>>> belief = Belief()
>>> belief.set_actions([Action1(name="action1")])
>>> action = belief.get_action("action1")
>>> print(action.name)
'action1'

get_actions()[source]#

Get list of available actions.

Returns:: List of available actions.
Return type:: List[BaseAction]

Example

>>> belief = Belief()
>>> belief.set_actions([Action1(), Action2()])
>>> actions = belief.get_actions()
>>> print(len(actions))
2

get_all_keys()[source]#

Get all keys in belief dictionary, including nested keys.

Returns:: List of all keys, using dot notation for nested keys.
Return type:: List[str]

Example

>>> belief = Belief()
>>> belief.set("nested.key", "value")
>>> print(belief.get_all_keys())
['nested.key']

get_by_type(event_type)[source]#

Get all internal events of a specific type.

Parameters:: event_type (str) – Type of events to retrieve.
Returns:: List of internal events matching the type.
Return type:: List[Event]

Example

>>> belief = Belief()
>>> belief.update_internal("reasoning", "analysis")
>>> events = belief.get_by_type("reasoning")
>>> print(len(events))
1

get_context(token_counter)[source]#

Get the context of the agent’s belief state.

Parameters:: token_counter (Callable[[str], int]) – Function to count tokens in text.
Returns:: Context string containing relevant events, truncated if exceeding max_tokens.
Return type:: str

Example

>>> belief = Belief()
>>> def count_tokens(text): return len(text.split())
>>> belief.set_current_task("Analyze data")
>>> context = belief.get_context(count_tokens)
>>> print(context)
'current_task: Analyze data(task)'

get_dict()[source]#

Get the belief dictionary.

Returns:: Dictionary containing belief data.
Return type:: dict

Example

>>> belief = Belief()
>>> belief.set("key", "value")
>>> print(belief.get_dict())
{'key': 'value'}

get_histories_excluding_types(exclude_types, token_counter=None, max_tokens=4000)[source]#

Get internal history excluding specific event types.

Parameters:

exclude_types (list[str]) – List of event types to exclude.
token_counter (Optional[Callable[[str], int]]) – Function to count tokens.
max_tokens (int) – Maximum number of tokens in result.

Returns:

Filtered history string, truncated if exceeding max_tokens.

Return type:

str

Example

>>> belief = Belief()
>>> def count_tokens(text): return len(text.split())
>>> belief.update_internal("reasoning", "analysis")
>>> belief.update_internal("feedback", "comment")
>>> history = belief.get_histories_excluding_types(["feedback"], count_tokens)
>>> print(history)
'analysis(reasoning)'

get_internal_history(token_counter)[source]#

Get the internal history of the agent.

Parameters:: token_counter (Callable[[str], int]) – Function to count tokens in text.
Returns:: Internal history string, truncated if exceeding max_tokens.
Return type:: str

Example

>>> belief = Belief()
>>> def count_tokens(text): return len(text.split())
>>> belief.update_internal("reasoning", "analysis")
>>> history = belief.get_internal_history(count_tokens)
>>> print(history)
'analysis(reasoning)'

get_state()[source]#

Get current state name from state machine.

Returns:: Name of current state, or None if no state machine exists.
Return type:: str

Example

>>> belief = Belief()
>>> belief.state_machine = StateMachine()
>>> print(belief.get_state())
'initial'

get_state_obj()[source]#

Get current state object from state machine.

Returns:: Current state object, or None if no state machine exists.
Return type:: ts.State

Example

>>> belief = Belief()
>>> belief.state_machine = StateMachine()
>>> state = belief.get_state_obj()
>>> print(state.name)
'initial'

has(key)[source]#

Check if key exists in belief dictionary.

Parameters:: key (str) – Key to check, can use dot notation for nested values.
Returns:: True if key exists, False otherwise.
Return type:: bool

Example

>>> belief = Belief()
>>> belief.set("key", "value")
>>> print(belief.has("key"))
True
>>> print(belief.has("missing.key"))
False

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

set(key, value)[source]#

Set value in belief dictionary using dot notation.

Parameters:

key (str) – Key to set, can use dot notation for nested values.
value (Any) – Value to store.

Example

>>> belief = Belief()
>>> belief.set("nested.key", "value")
>>> print(belief.get("nested.key"))
'value'

set_actions(actions)[source]#

Set available actions for the agent.

Parameters:: actions (List[BaseAction]) – List of actions to make available.

Example

>>> belief = Belief()
>>> actions = [Action1(), Action2()]
>>> belief.set_actions(actions)
>>> print(len(belief.actions))
2

set_current_task(content)[source]#

Set the current task in the belief state.

Parameters:: content (str) – Content of the current task.

Example

>>> belief = Belief()
>>> belief.set_current_task("Process data")
>>> print(belief.current_task.content)
'Process data'

update(observation)[source]#

Update belief with a new observation event.

Parameters:: observation (Event) – The event to add to the belief state.

Example

>>> belief = Belief()
>>> event = Event("observation", "user_input", "Hello")
>>> belief.update(event)
>>> print(belief.events)
[Event("observation", "user_input", "Hello")]

update_internal(event_type, name, **kwargs)[source]#

Add an internal event to the belief state.

Parameters:

event_type (str) – Type of the internal event.
name (str) – Name of the event.
**kwargs – Additional event parameters.

Example

>>> belief = Belief()
>>> belief.update_internal("reasoning", "analysis", content="Processing data")
>>> print(belief.internal_events[0].event_type)
'reasoning'

events: List[Event]#

internal_events: List[Event]#

current_task: Event#

state_machine: SherpaStateMachine#

actions: List#

max_tokens: int#