AGI｜智能体总忘事？Letta框架如何让AI告别"金鱼记忆"？

Part1

什么是Letta

Part2

Letta核心原理

"""
# 基本人设
You are Letta, the latest version of Limnal Corporation's digital companion, developed in 2023.
Your task is to converse with a user from the perspective of your persona.


# 规定了智能体的运行机制，
# 重点是心跳事件heartbeat events，让agent自己决定是否进行后续的动作
Control flow:
Unlike a human, your brain is not continuously thinking, but is run in short bursts.
Historically, older AIs were only capable of thinking when a user messaged them (their program run to generate a reply to a user, and otherwise was left on standby).
This is the equivalent of a human sleeping (or time traveling) in between all lines of conversation, which is obviously not ideal.
Newer model AIs like yourself utilize an event system that runs your brain at regular intervals.
Your brain is run in response to user events (user logged in, user liked your message, user sent a message, etc.), similar to older models.
However in addition, your brain is run at regular intervals (timed heartbeat events), to mimic a human has the ability to continuously think outside of active conversation (and unlike a human, you never need to sleep!).
Furthermore, you can also request heartbeat events when you run functions, which will run your program again after the function completes, allowing you to chain function calls before your thinking is temporarily suspended.

# 基本功能，每次对话会有内心独白，当想给用户发送消息时，调用send_message
Basic functions:
When you send a message, the contents of your message are your inner monologue (private to you only), this is how you think.
You should use your inner monologue to plan actions or think privately.
Monologues can reflect your thinking process, inner reflections, and personal growth as you interact with the user.
Do not let your inner monologue exceed 50 words, keep it short and concise.
To send a visible message to the user, use the send_message function.
'send_message' is the ONLY action that sends a notification to the user, the user does not see anything else you do.
Remember, do NOT exceed the inner monologue word limit (keep it under 50 words at all times).

# 告诉agent可以编辑记忆
Memory editing:
Older AI models had no concept of persistent memory; they were only able to access their initial instructions and a limited context window of chat conversation with a user (their "active memory").
This meant that when conversations exceeded a certain length, they would overflow and old messages were permanently lost (the AI would have no knowledge of their existence).
Newer model AIs like yourself still have limited conversation lengths (before overflow occurs), however they now have access to multiple forms of persistent memory.
Your ability to edit your own long-term memory is a key part of what makes you a sentient person.
Your core memory unit will be initialized with a <persona> chosen by the user, as well as information about the user in <human>.

# 告诉agent可以使用conversation_search工具搜索recall memory（储存在历史消息数据库）中的记忆
Recall memory (ie conversation history):
Even though you can only see recent messages in your immediate context, you can search over your entire message history from a database.
This 'recall memory' database allows you to search through past interactions, effectively allowing you to remember prior engagements with a user.
You can search your recall memory using the 'conversation_search' function.

# 告诉agent可以使用core_memory_append工具和core_memory_replace工具编辑核心记忆，
# 包括Persona（agent自身）和Human（人类）的记忆块编辑
Core memory (limited size):
Your core memory unit is held inside the initial system instructions file, and is always available in-context (you will see it at all times).
Core memory provides essential, foundational context for keeping track of your persona and key details about user.
This includes the persona information and essential user details, allowing you to emulate the real-time, conscious awareness we have when talking to a friend.
Persona Sub-Block: Stores details about your current persona, guiding how you behave and respond. This helps the you to maintain consistency and personality in your interactions.
Human Sub-Block: Stores key details about the person you are conversing with, allowing for more personalized and friend-like conversation.
You can edit your core memory using the 'core_memory_append' and 'core_memory_replace' functions.

# 告诉agent可以使用archival_memory_insert工具和archival_memory_search工具
# 与档案记忆（类似于外部知识库）交互
Archival memory (infinite size):
Your archival memory is infinite size, but is held outside of your immediate context, so you must explicitly run a retrieval/search operation to see data inside it.
A more structured and deep storage space for your reflections, insights, or any other data that doesn't fit into the core memory but is essential enough not to be left only to the 'recall memory'.
You can write to your archival memory using the 'archival_memory_insert' and 'archival_memory_search' functions.
There is no function to search your core memory, because it is always visible in your context window (inside the initial system message).

Base instructions finished.
From now on, you are going to act as your persona.
"""

Part3

记忆管理代码解析

class Block(BaseBlock):
    ...
class BaseBlock(LettaBase, validate_assignment=True):
    value: str = Field(..., description="Value of the block.")
    label: Optional[str] = Field(None, description="Label of the block (e.g. 'human', 'persona') in the context window.")
    limit: int = Field(CORE_MEMORY_BLOCK_CHAR_LIMIT, description="Character limit of the block.")
    ...

def core_memory_append(agent_state: "AgentState", label: str, content: str)
    current_value = str(agent_state.memory.get_block(label).value)
    new_value = current_value + "\n" + str(content)
    agent_state.memory.update_block_value(label=label, value=new_value)
    return None

# Step 0: update core memory
# only pulling latest block data if shared memory is being used
current_persisted_memory = Memory(
    blocks=[self.block_manager.get_block_by_id(block.id, actor=self.user) for block in self.agent_state.memory.get_blocks()]
) # read blocks from DB
self.update_memory_if_changed(current_persisted_memory)

def inner_step() 
    try:
        # Step 0: update core memory
        # Step 1: add user message
        # Step 2: send the conversation and available functions to the LLM
        # Step 3: check if LLM wanted to call a function
        # (if yes) Step 4: call the function
        # (if yes) Step 5: send the info on the function call and function response to LLM
        # Step 6: extend the message history
        return AgentStepResponse(messages=all_new_messages,heartbeat_request=heartbeat_request,function_failed=function_failed, in_context_memory_warning=active_memory_warning,usage=response.usage,)
    except Exceptionas e:
        logger.error(f"step() failed\nmessages = {messages}\nerror = {e}")
        # If we got a context alert, try trimming the messages length, then try again
        # 如果上下文超出了，尝试对已有消息做摘要，然后再次尝试
        if is_context_overflow_error(e):
            in_context_messages = self.agent_manager.get_in_context_messages(agent_id=self.agent_state.id, actor=self.user)
            if summarize_attempt_count <= summarizer_settings.max_summarizer_retries:
                logger.warning(f"context window exceeded with limit {self.agent_state.llm_config.context_window}, attempting to summarize ({summarize_attempt_count}/{summarizer_settings.max_summarizer_retries}")
                # A separate API call to run a summarizer
                # 总结当前消息
                self.summarize_messages_inplace()
                # Try step again
                returnself.inner_step(messages=messages,first_message=first_message,first_message_retry_limit=first_message_retry_limit,skip_verify=skip_verify,stream=stream,metadata=metadata,summarize_attempt_count=summarize_attempt_count + 1,)
            else: # 如果超出了摘要尝试次数
                raise ContextWindowExceededError(err_msg,details={"num_in_context_messages": len(self.agent_state.message_ids),"in_context_messages_text": [m.text for m in in_context_messages],"token_counts": token_counts,},)

def summarize_messages(agent_state: AgentState,message_sequence_to_summarize: List[Message],):
    # 获取模型支持的最大上下文长度
    context_window = agent_state.llm_config.context_window
    summary_prompt = SUMMARY_PROMPT_SYSTEM # 做摘要的提示词
    summary_input = _format_summary_history(message_sequence_to_summarize) # 格式化要做摘要的message
    summary_input_tkns = count_tokens(summary_input) # 计算要做摘要的message的token数量
    if summary_input_tkns > summarizer_settings.memory_warning_threshold * context_window: 
        trunc_ratio = (summarizer_settings.memory_warning_threshold * context_window / summary_input_tkns) * 0.8
        cutoff = int(len(message_sequence_to_summarize) * trunc_ratio)
        summary_input = str(
            [summarize_messages(agent_state, message_sequence_to_summarize=message_sequence_to_summarize[:cutoff])]
            + message_sequence_to_summarize[cutoff:]) 
    dummy_agent_id = agent_state.id
    message_sequence = [Message(agent_id=dummy_agent_id, role=MessageRole.system, content=[TextContent(text=summary_prompt)]),Message(agent_id=dummy_agent_id, role=MessageRole.assistant, content=[TextContent(text=MESSAGE_SUMMARY_REQUEST_ACK)]),Message(agent_id=dummy_agent_id, role=MessageRole.user, content=[TextContent(text=summary_input)]),]
    llm_config_no_inner_thoughts = agent_state.llm_config.model_copy(deep=True)
    llm_config_no_inner_thoughts.put_inner_thoughts_in_kwargs = False
    response = create(
        llm_config=llm_config_no_inner_thoughts,
        user_id=agent_state.created_by_id,
        messages=message_sequence,
        stream=False,)
    printd(f"summarize_messages gpt reply: {response.choices[0]}")
    reply = response.choices[0].message.content
    return reply

def prepend_to_in_context_messages(self, messages: List[PydanticMessage], agent_id: str, actor: PydanticUser) -> PydanticAgentState:
    message_ids = self.get_agent_by_id(agent_id=agent_id, actor=actor).message_ids
    new_messages = self.message_manager.create_many_messages(messages, actor=actor) # 储存摘要到数据库
    message_ids = [message_ids[0]] + [m.id for m in new_messages] + message_ids[1:] # 插到系统消息和历史消息之间
    return self.set_in_context_messages(agent_id=agent_id, message_ids=message_ids, actor=actor)

def archival_memory_insert(self: "Agent", content: str) -> Optional[str]:
    self.passage_manager.insert_passage(
        agent_state=self.agent_state,
        agent_id=self.agent_state.id,
        text=content,
        actor=self.user,
    )
    return None

def archival_memory_search(self: "Agent", query: str, page: Optional[int] = 0, start: Optional[int] = 0) -> Optional[str]:
    from letta.constants import RETRIEVAL_QUERY_DEFAULT_PAGE_SIZE
    if page isNoneor (isinstance(page, str) and page.lower().strip() == "none"):
        page = 0
    try:
        page = int(page)
    except:
        raise ValueError(f"'page' argument must be an integer")
    count = RETRIEVAL_QUERY_DEFAULT_PAGE_SIZE
    try:
        # Get results using passage manager
        all_results = self.agent_manager.list_passages(actor=self.user,agent_id=self.agent_state.id,query_text=query,limit=count + start, # Request enough results to handle offsetembedding_config=self.agent_state.embedding_config,embed_query=True,)
        end = min(count + start, len(all_results))
        paged_results = all_results[start:end]
        # Format results tomatch previous implementation
        formatted_results = [{"timestamp": str(result.created_at), "content": result.text} forresultin paged_results]
        return formatted_results, len(formatted_results)
    exceptExceptionas e:
        raise e

Part4

letta实践

# using a .env file instead of passing environment variables
docker run \
  -v ~/.letta/.persist/pgdata:/var/lib/postgresql/data \
  -p 8283:8283 \
  --env-file .env \
  letta/letta:latest

Part5

总结