Create a proper/extensible agent memory API

[gavento]

Both for simple memory as list of events and its formatting (now), but also to support summarization or more complex recall in the future (e.g. as modular subclasses to be reused) or integration with other implementations of agent memory (GPTeam? Mindstorms? ...).

See also:

• Paper on Generative agents regarding a) memory b) communication facilitation (who talks when to whom, who listens etc.)
• How is the re-implementation of Generative agents in GPTeam regarding the same questions? Is it directly useful or at least inspiring?
• How do mindstorms implement the above?

[jas-ho]

the memory API in GPTeam seems promising and very close to what I remember from Generative agents. Generative agents paper and GPTeam snippets side-by-side:

Our architecture comprises three main components. The first is the memory stream, a long-term memory module that records, in natural language, a comprehensive list of the agent’s experiences.

class Agent(BaseModel):
id: UUID
full_name: str
private_bio: str
public_bio: str
directives: Optional[list[str]]
last_checked_events: datetime
last_summarized_activity: datetime
memories: list[SingleMemory]  # <== that's the memory stream
plans: list[SinglePlan]
authorized_tools: list[ToolName]
world_id: UUID
notes: list[str] = []
plan_executor: PlanExecutor = None
context: WorldContext
location: Location
discord_bot_token: str = None
react_response: LLMReactionResponse = None
recent_activity: str = ""

The retrieval model combines relevance, recency, and importance to surface the records that are needed to inform the agent’s moment-to-moment behavior.

async def get_relevant_memories(
query: str, memories: list[SingleMemory], k: int = 5
) -> list[SingleMemory]:
"""Returns a list of the top k most relevant NON MESSAGE memories, based on the query string"""

memories_with_relevance = [
    RelatedMemory(memory=memory, relevance=await memory.relevance(query))
    for memory in memories
]
... # then sort by relevance and take top k

class Memory ... async def relevance(self, query: str) -> float: return ( IMPORTANCE_WEIGHT * self.importance

• SIMILARITY_WEIGHT * (await self.similarity(query))
• RECENCY_WEIGHT * self.recency )

[jas-ho]

The second is reflection, which synthesizes memories into higher-level inferences over time, enabling the agent to draw conclusions about itself and others to better guide its behavior.

class Agent:
...
async def run_for_one_step(self):
....
# Work through the plans
await self._do_first_plan()

    # Reflect, if we should   # <-- here they include reflection
    if await self._should_reflect():
        await self._reflect()
   ...

async def _reflect(self):  # <-- reflection is implemented very similarly to how described in Generative Agents
    # Get the reflection questions
    ...
    # Parse the response into an object
    ...
    # For each question in the parsed questions...
    ...
        # Get the reflection insights
        ...
        # For each insight in the parsed insights...
        ...
            # Get the related memory ids
            ...
            # Add a new memory
            ...

[jas-ho]

The third is planning, which translates those conclusions and the current environment into high-level action plans and then recursively into detailed behaviors for action and reaction. These reflections and plans are fed back into the memory stream to influence the agent’s future behavior.

class Agent:
...
async def run_for_one_step(self):
...
# if there's no current plan, make some
if len(self.plans) == 0:
print(f"{self.full_name} has no plans, making some...")
await self._plan()
...

async def _plan(self, thought_process: str = "") -> list[SinglePlan]:
    ...
    # Get a summary of the recent activity
    ...
    # Get the plans
    ...
    # Delete existing plans
    ...
    # Make a bunch of new plan objects, put em into a list
    ...
    # update the local agent object
    ...

[gavento]

Thanks for the notes, Jason.

After some reflection, I am less sure we need to have a robust API for Memory - we need simple memory interface that will cover 1. list memory 2. summarizing memory, and 3. embedding memory with vector similarity recall, and maybe more like this (e.g. tagging/clustering/... or e.g. a combo of summarizing and embedding, with summaries having more weight by default than the components).

Also, part of the initial goal is to have a minimalistic API first, if feasible - GPTeam already assumes Plans, Tools, and more (Locations).

But for next step, having agents with explicit plans may be an interesting direction -- but that is not just the function of memory but also the agents.

• One possible impl could be to use (embedding or summarizing) memory as storage for both observations, thoughts, and plans, and recall them by their tags (these 3 categories) in the agent's internal routine.

[gavento]

Gonna close this as resolved by #7, with design doc label.