🤖 MemGPT 智能体内存管理架构

📑 目录

📋 项目简介

• GitHub: letta-ai/letta（原 cpacker/MemGPT）
• 语言: Python
• Star: 15k+
• 核心论文: "MemGPT: Towards LLMs as Operating Systems" (NeurIPS 2023)
• 理念: 将 LLM 视为操作系统，智能体视为进程，内存管理由框架自动完成

💡 核心设计理念

LLM 通过函数调用主动管理自己的内存，而非被动接受固定上下文。

🏗️ 整体架构

📐 架构演进

🔄 MemGPT vs Letta 对比

📦 BaseAgent 抽象类

class BaseAgent(ABC):
    """
    Abstract class for all agents.
    Only one interface is required: step.
    """

    @abstractmethod
    def step(
        self,
        input_messages: List[MessageCreate],
    ) -> LettaUsageStatistics:
        """
        Top-level event message handler
        for the agent.
        """
        raise NotImplementedError

• 极简抽象：只要求实现 step() 方法
• step() 是 Agent 的唯一入口点
• 输入：消息列表 → 输出：使用统计

📦 Agent 核心类

class Agent(BaseAgent):
    def __init__(
        self,
        interface: Optional[Union[AgentInterface,
                                  StreamingRefreshCLIInterface]],
        agent_state: AgentState,
        user: User,
        first_message_verify_mono: bool = True,
        mcp_clients: Optional[Dict[str,
            AsyncBaseMCPClient]] = None,
        save_last_response: bool = False,
    ):

• interface: 输出接口（CLI / Streaming / API）
• agent_state: 完整的智能体状态（含内存）
• user: 当前用户（权限控制）
• mcp_clients: MCP 协议客户端

⚙️ Agent 初始化流程

# 核心初始化组件
self.agent_state = agent_state
self.user = user

# 工具规则求解器
self.tool_rules_solver = ToolRulesSolver(
    tool_rules=agent_state.tool_rules
)

# 模型配置
self.model = self.agent_state.llm_config.model
self.supports_structured_output = \
    check_supports_structured_output(model=self.model)

# 状态管理器（Service Layer）
self.block_manager = BlockManager()
self.message_manager = MessageManager()
self.agent_manager = AgentManager()
self.passage_manager = PassageManager()

# 遥测
self.telemetry_manager = TelemetryManager() \
    if settings.llm_api_logging \
    else NoopTelemetryManager()

🧠 Memory 内存系统

🧩 Block 内存块

# Block 的核心属性
class Block:
    id: str           # 唯一标识
    label: str        # "persona", "human" 等
    value: str        # 内存内容（纯文本）
    read_only: bool   # 是否只读
    limit: int        # token 上限

# 使用示例
persona_block = Block(
    label="persona",
    value="I am a helpful assistant.",
    read_only=False,
    limit=2000
)

• Block 是最小的内存单元
• 每个 Block 有 label（标签）和 value（内容）
• read_only 块不允许 LLM 修改

🏷️ 内存块类型

📋 AgentState 状态

class AgentState:
    id: str                       # Agent ID
    name: str                     # Agent 名称
    agent_type: str               # Agent 类型
    memory: Memory                # 内存对象
    llm_config: LLMConfig         # LLM 配置
    embedding_config: EmbeddingConfig
    tool_rules: List[ToolRule]    # 工具规则
    message_ids: List[str]        # 上下文消息 ID
    response_format: ResponseFormat
    timezone: str                 # 时区

    # metadata
    created_at: datetime
    updated_at: datetime

AgentState 是 Agent 的完整快照，包含运行所需的所有信息。

🔧 Memory.compile()

# Memory.compile() 将所有 Block 编译为
# system prompt 的一部分

# 输入（多个 Block）
memory = Memory(
    blocks=[
        Block(label="persona", value="You are..."),
        Block(label="human", value="User name is..."),
        Block(label="scratchpad", value="Current plan..."),
    ]
)

# 输出（编译后的 prompt 片段）
compiled = memory.compile()
# → "\n<persona>\nYou are...\n</persona>
#    \n<human>\nUser name is...\n</human>
#    \n<scratchpad>\nCurrent plan...\n</scratchpad>"

compile() 用 XML 标签包裹每个 Block，便于 LLM 区分不同记忆类型。

📝 PromptGenerator

class PromptGenerator:
    """生成完整的 system prompt"""

    def generate_system_prompt(
        self,
        agent_state: AgentState,
    ) -> str:
        # 1. 加载基础系统提示
        system_prompt = self._load_base_prompt()

        # 2. 编译内存块到 prompt
        memory_str = agent_state.memory.compile()
        system_prompt += memory_str

        # 3. 添加工具描述
        tools_str = self._format_tools(agent_state.tools)
        system_prompt += tools_str

        # 4. 添加内存管理指令
        system_prompt += self._memory_instructions()

        return system_prompt

🔄 step() 主循环

🔄 step() 详细执行

@trace_method
def step(
    self,
    input_messages: List[MessageCreate],
) -> LettaUsageStatistics:
    # 重建 system prompt（含最新内存）
    self.agent_state = \
        self.agent_manager.rebuild_system_prompt(
            agent_id=self.agent_state.id,
            actor=self.user,
        )

    # 获取上下文中的消息
    messages = self.agent_manager.get_in_context_messages(
        agent_id=self.agent_state.id,
        actor=self.user,
    )

    # 追加新消息
    messages.extend(new_messages)

🤖 _get_ai_reply()

@trace_method
def _get_ai_reply(self, ...):
    # 1. 检查上下文窗口
    token_count = num_tokens_from_messages(messages)
    if token_count > self.agent_state.llm_config.context_window:
        raise ContextWindowExceededError(...)

    # 2. 准备工具列表
    functions_list = self._prepare_tools()
    functions_list = self._runtime_override_tool_json_schema(
        functions_list
    )

    # 3. 调用 LLM API
    response = create(
        model=self.model,
        messages=messages,
        tools=functions_list,
    )

    return response

🔧 工具执行引擎

Agent 通过函数调用主动操作自己的记忆，实现自我改进。

🛡️ ToolExecutionSandbox

class ToolExecutionSandbox:
    """工具执行沙盒环境"""

    def execute_tool(
        self,
        tool: Tool,
        function_args: dict,
        agent_state: AgentState,
    ) -> ToolExecutionResult:

        # 1. 参数校验和强制转换
        validated_args = coerce_dict_args_by_annotations(
            function_args,
            tool.source_code,
        )

        # 2. 在沙盒中执行
        result = self._run_in_sandbox(
            tool.source_code,
            validated_args,
            env=self._build_env(agent_state),
        )

        return ToolExecutionResult(
            output=result,
            status="success",
        )

💾 内存更新机制

内存更新后，system prompt 自动重建，LLM 在下一轮对话中就能使用新记忆。

💾 update_memory_if_changed

def update_memory_if_changed(
    self, new_memory: Memory
) -> bool:
    system_message = self.message_manager \
        .get_message_by_id(
            message_id=self.agent_state.message_ids[0],
            actor=self.user,
        )

    if new_memory.compile() not in \
            system_message.content[0].text:
        # 内存有变化，更新 DB
        for label in new_memory.list_block_labels():
            updated = new_memory.get_block(label).value
            if updated != self.agent_state.memory \
                    .get_block(label).value:
                self.block_manager.update_block(
                    block_id=...,
                    block_update=BlockUpdate(value=updated),
                )
        # 重建 system prompt
        self.agent_state = \
            self.agent_manager.rebuild_system_prompt(
                agent_id=self.agent_state.id,
            )
        return True
    return False

📏 上下文管理

• 上下文窗口: LLM 的 token 上限（如 128k）
• 消息累积: 对话越长，占用的 token 越多
• 溢出检测: 每次调用 LLM 前检查 token 数
• 自动压缩: 溢出时自动摘要旧消息

# 检查上下文溢出
token_count = num_tokens_from_messages(messages)
context_limit = self.agent_state.llm_config \
    .context_window

if token_count > context_limit:
    # 触发摘要压缩
    summarized = summarize_messages(messages)

📄 summarize_messages

def summarize_messages(
    messages: List[Message],
    model: str,
    token_limit: int,
) -> List[Message]:
    """将旧消息压缩为摘要"""

    # 1. 计算需要保留的消息数量
    cutoff = calculate_summarizer_cutoff(
        messages, token_limit
    )

    # 2. 分离需要摘要和需要保留的消息
    to_summarize = messages[:cutoff]
    to_keep = messages[cutoff:]

    # 3. 调用 LLM 生成摘要
    summary = create(
        model=model,
        messages=[
            {"role": "system",
             "content": "Summarize this conversation..."},
            *format_messages(to_summarize),
        ],
    )

    # 4. 用摘要替换旧消息
    return [summary_message] + to_keep

💓 心跳机制

• Agent 在空闲时会定期触发"心跳"
• 心跳允许 Agent 自主思考和管理记忆
• 类似于操作系统中的空闲进程调度

# 心跳触发流程
def heartbeat(self):
    # 发送内部心跳消息
    heartbeat_msg = get_heartbeat()

    # Agent 像处理用户消息一样处理心跳
    self.step(input_messages=[heartbeat_msg])

    # Agent 可以在心跳中：
    # - 整理归档记忆
    # - 反思对话历史
    # - 规划下一步行动

📋 ToolRulesSolver

class ToolRulesSolver:
    """工具调用规则求解器"""

    def __init__(self, tool_rules):
        self.tool_rules = tool_rules
        self.call_history = []

    def register_tool_call(self, tool_name: str):
        """记录一次工具调用"""
        self.call_history.append(tool_name)

    def should_allow_tool(
        self, tool_name: str
    ) -> bool:
        """根据规则判断是否允许调用"""
        for rule in self.tool_rules:
            if rule.matches(tool_name, self.call_history):
                return rule.allow
        return True  # 默认允许

ToolRulesSolver 实现了类似 Linux iptables 的链式规则匹配。

🗄️ BlockManager

class BlockManager:
    """内存块的 CRUD 管理"""

    def get_block_by_id(self, block_id, actor):
        # 从数据库加载 Block
        return self.orm.get(Block, block_id)

    def update_block(
        self, block_id, block_update, actor
    ):
        # 更新 Block 内容并持久化
        block = self.get_block_by_id(block_id, actor)
        if block.read_only:
            raise ValueError("Block is read-only!")
        block.value = block_update.value
        self.orm.update(block)
        return block

    def create_block(self, block_create, actor):
        # 创建新 Block
        block = Block(**block_create.dict())
        self.orm.add(block)
        return block

📨 MessageManager

class MessageManager:
    """消息持久化管理"""

    def get_message_by_id(self, message_id, actor):
        return self.orm.get(Message, message_id)

    def create_message(self, message, actor):
        msg = Message(**message.dict())
        self.orm.add(msg)
        return msg

    def get_in_context_messages(
        self, agent_id, actor
    ) -> List[Message]:
        """获取当前在上下文中的消息"""
        agent = self.agent_manager.get_agent(
            agent_id
        )
        return [self.get_message_by_id(mid, actor)
                for mid in agent.message_ids]

🔌 AgentInterface

class AgentInterface(ABC):
    """Agent 输出接口抽象"""

    @abstractmethod
    def internal_monologue(self, msg):
        """显示 Agent 内心独白"""
        pass

    @abstractmethod
    def assistant_message(self, msg):
        """显示 Agent 回复消息"""
        pass

    @abstractmethod
    def function_message(self, msg, **kwargs):
        """显示函数调用结果"""
        pass

# 实现: CLIInterface, StreamingInterface,
#       APIInterface, DiscordInterface 等

接口模式让同一个 Agent 可以在不同环境中运行（CLI、API、Discord Bot）。

🌐 LLMClient

class LLMClient:
    """统一的 LLM 调用接口"""

    def __init__(self):
        # 支持多个 Provider
        self.providers = {
            ProviderType.OPENAI: OpenAIProvider(),
            ProviderType.ANTHROPIC: AnthropicProvider(),
            ProviderType.LOCAL: LocalProvider(),
        }

    def call(
        self,
        model: str,
        messages: List[Message],
        tools: Optional[List] = None,
    ) -> ChatCompletionResponse:
        """根据模型名路由到对应 Provider"""
        provider = self._resolve_provider(model)
        return provider.chat(model, messages, tools)

LLMClient 实现了模型无关的抽象，同一套代码可以运行在不同的 LLM 上。

📊 数据流总览

⏱️ 消息处理时序图

⏱️ 内存读写时序

⏱️ 工具调用时序

⏱️ 上下文压缩时序

🏗️ 设计模式：分层架构

🏗️ 设计模式：策略模式

# 不同 LLM Provider
class LLMProvider(ABC):
    @abstractmethod
    def chat(self, model, messages):
        pass

class OpenAIProvider(LLMProvider):
    def chat(self, model, messages):
        return openai.ChatCompletion.create(
            model=model, messages=messages
        )

class AnthropicProvider(LLMProvider):
    def chat(self, model, messages):
        return anthropic.messages.create(
            model=model, messages=messages
        )

# 不同输出接口
class AgentInterface(ABC):
    def internal_monologue(self, msg): ...
    def assistant_message(self, msg): ...

class CLIInterface(AgentInterface):
    def assistant_message(self, msg):
        print(msg)

class DiscordInterface(AgentInterface):
    def assistant_message(self, msg):
        channel.send(msg)

🏗️ 设计模式：事件驱动

# Agent 通过 Interface 发出事件
class Agent:
    def step(self, input_messages):
        # ... 处理逻辑 ...

        # 发出内心独白事件
        self.interface.internal_monologue(thought)

        # 发出工具调用事件
        self.interface.function_message(
            f"Ran {func}({args})"
        )

        # 发出回复事件
        self.interface.assistant_message(reply)

# 不同 Interface 实现决定事件去向：
# CLIInterface → print 到终端
# StreamingInterface → SSE 推送
# DiscordInterface → 发送到频道

Agent 不关心输出方式，Interface 决定事件如何处理。

🏗️ 设计模式：沙盒模式

• ToolExecutionSandbox 隔离工具执行环境
• 防止恶意工具调用影响 Agent 主进程
• 支持超时控制和资源限制

class ToolExecutionSandbox:
    def execute_tool(self, tool, args, agent_state):
        # 1. 参数校验
        validated = coerce_dict_args_by_annotations(
            args, tool.source_code
        )

        # 2. 构建受限环境
        env = self._build_env(agent_state)

        # 3. 超时执行
        result = timeout(
            self._run(tool.source_code, validated, env),
            timeout_seconds=30,
        )

        return ToolExecutionResult(output=result)

📐 核心类 UML 图

📐 内存系统 UML 图

📐 服务层 UML 图

⚡ 性能优化：上下文管理

• Token 计数缓存: 避免重复计算消息的 token 数
• 增量摘要: 只摘要新增的旧消息，不重复摘要已压缩部分
• 分级压缩: 优先压缩最旧的消息，保留最近对话

# 增量摘要策略
def calculate_summarizer_cutoff(
    messages, token_limit
):
    """计算哪些消息需要摘要"""
    total = 0
    for i, msg in enumerate(reversed(messages)):
        total += count_tokens(msg)
        if total > token_limit * 0.8:
            return len(messages) - i - 1
    return len(messages)

⚡ 性能优化：并行处理

• 异步工具执行: 多个独立工具调用可并行
• 异步 MCP 客户端: MCP 协议使用 async/await
• 流水线处理: 消息持久化和 LLM 调用可重叠

# 异步 MCP 客户端
class AsyncBaseMCPClient(ABC):
    @abstractmethod
    async def call_tool(
        self, tool_name, args
    ) -> str:
        pass

    @abstractmethod
    async def list_tools(self) -> List:
        pass

⚡ 性能优化：记忆检索

• 向量检索: archival_memory_search 使用 embedding 检索
• Passage 索引: 记忆被切分为 Passage 存储在向量 DB
• 语义相似度: 使用 embedding 计算查询与记忆的相似度

# 记忆检索流程
class PassageManager:
    def search(
        self,
        query: str,
        agent_id: str,
        limit: int = 10,
    ) -> List[Passage]:
        """向量检索相关记忆"""
        # 1. 将查询转为 embedding
        query_embedding = self.embed(query)

        # 2. 在向量 DB 中搜索
        results = self.vector_db.search(
            query_embedding,
            filter={"agent_id": agent_id},
            limit=limit,
        )
        return results

📊 性能 Benchmark

✅ 最佳实践

❌ 反模式

• ❌ Block 过大: 单个 Block 超过 2000 token 会导致 prompt 膨胀
• ❌ 频繁读写: 每次 step 都修改记忆会增加 DB 压力
• ❌ 忽略心跳: 不配置心跳会让 Agent 缺少自主整理能力
• ❌ 过多工具: 工具描述占用大量 token，应按需加载
• ❌ 无 read_only: 不保护系统 Block 会让 LLM 覆盖核心指令
• ❌ 小模型+复杂工具: 小模型难以正确调用复杂工具

🌍 生态集成

📚 扩展阅读

• 论文: "MemGPT: Towards LLMs as Operating Systems" (NeurIPS 2023)
• GitHub: github.com/letta-ai/letta
• 文档: docs.letta.com
• Discord: discord.gg/letta

🎯 总结