智能体设计模式

第8章:内存管理

为持久智能体智能实现有效的短期和长期内存系统

第8章:内存管理

有效的内存管理对于智能体保留信息至关重要。智能体需要不同类型的内存,就像人类一样,才能高效运行。本章深入探讨内存管理,特别解决智能体的即时(短期)和持久(长期)内存需求。

在智能体系统中,内存是指智能体保留和利用来自过去交互、观察和学习体验的信息的能力。这种能力允许智能体做出明智的决策、维护对话上下文并随时间改进。智能体内存通常分为两种主要类型:

  • 短期内存(上下文内存): 类似于工作内存,它保存当前正在处理或最近访问的信息。对于使用大语言模型(LLM)的智能体,短期内存主要存在于上下文窗口中。此窗口包含最近的消息、智能体回复、工具使用结果以及当前交互中的智能体反思,所有这些都为LLM的后续响应和行动提供信息。上下文窗口具有有限容量,限制智能体可以直接访问的最近信息量。有效的短期内存管理涉及将最相关的信息保留在此有限空间内,可能通过总结较旧的对话段或强调关键细节等技术。具有"长上下文"窗口的模型的出现只是扩展了这种短期内存的大小,允许在单个交互中保存更多信息。然而,这种上下文仍然是短暂的,一旦会话结束就会丢失,并且每次处理都可能成本高昂且效率低下。因此,智能体需要单独的内存类型来实现真正的持久性、从过去的交互中回忆信息并构建持久的知识库。
  • 长期内存(持久内存): 这作为智能体需要在各种交互、任务或延长时间内保留的信息存储库,类似于长期知识库。数据通常存储在智能体直接处理环境之外,通常在数据库、知识图或向量数据库中。在向量数据库中,信息被转换为数值向量并存储,使智能体能够基于语义相似性而不是精确关键词匹配检索数据,这个过程称为语义搜索。当智能体需要来自长期内存的信息时,它查询外部存储,检索相关数据,并将其集成到短期上下文中以供立即使用,从而将先验知识与当前交互相结合。

实际应用和用例

内存管理对于智能体跟踪信息并随时间智能执行至关重要。这对于智能体超越基本问答能力至关重要。应用包括:

  • 聊天机器人和对话AI: 维护对话流程依赖于短期内存。聊天机器人需要记住先前的用户输入以提供连贯的响应。长期内存使聊天机器人能够回忆用户偏好、过去的问题或先前的讨论,提供个性化和连续的交互。
  • 面向任务的智能体: 管理多步任务的智能体需要短期内存来跟踪先前的步骤、当前进度和整体目标。此信息可能驻留在任务的上下文中或临时存储中。长期内存对于访问不在直接上下文中的特定用户相关数据至关重要。
  • 个性化体验: 提供定制交互的智能体利用长期内存来存储和检索用户偏好、过去的行为和个人信息。这允许智能体调整其响应和建议。
  • 学习和改进: 智能体可以通过从过去的交互中学习来完善其性能。成功的策略、错误和新信息存储在长期内存中,促进未来的适应。强化学习智能体以这种方式存储学习的策略或知识。
  • 信息检索(RAG): 设计用于回答问题的智能体访问知识库,其长期内存,通常在检索增强生成(RAG)中实现。智能体检索相关文档或数据以告知其响应。
  • 自主系统: 机器人或自动驾驶汽车需要内存来存储地图、路线、对象位置和学习的行为。这涉及用于即时环境的短期内存和用于一般环境知识的长期内存。

内存使智能体能够维护历史、学习、个性化交互并管理复杂的、时间相关的问题。

实践代码:Google Agent Developer Kit(ADK)中的内存管理

Google Agent Developer Kit(ADK)提供了一种结构化方法来管理上下文和内存,包括用于实际应用的组件。对ADK的会话、状态和内存的扎实理解对于构建需要保留信息的智能体至关重要。

就像在人类交互中一样,智能体需要回忆先前交换的能力来进行连贯和自然的对话。ADK通过三个核心概念及其相关服务简化了上下文管理。

与智能体的每次交互都可以被视为唯一的对话线程。智能体可能需要访问来自早期交互的数据。ADK将此结构如下:

  • 会话: 记录该特定交互的消息和行动(事件)的单个聊天线程,还存储与该对话相关的临时数据(状态)。
  • 状态(session.state): 存储在会话中的数据,包含仅与当前活动聊天线程相关的信息。
  • 内存: 来自各种过去聊天或外部来源的信息的可搜索存储库,作为超出直接对话的数据检索资源。

ADK提供专门的服务来管理构建复杂、有状态和上下文感知智能体所需的关键组件。SessionService通过处理其启动、记录和终止来管理聊天线程(会话对象),而MemoryService监督长期知识(内存)的存储和检索。

SessionService和MemoryService都提供各种配置选项,允许用户根据应用需求选择存储方法。内存选项可用于测试目的,尽管数据在重启后不会持久。对于持久存储和可扩展性,ADK还支持基于数据库和云的服务。

会话:跟踪每个聊天

ADK中的会话对象设计用于跟踪和管理单个聊天线程。在与智能体开始对话时,SessionService生成一个会话对象,表示为google.adk.sessions.Session。此对象封装了与特定对话线程相关的所有数据,包括唯一标识符(id、app_name、user_id)、作为事件对象的事件时间记录、称为状态的会话特定临时数据存储区域,以及指示最后更新时间的时间戳。开发者通常通过SessionService间接与会话对象交互。SessionService负责管理对话会话的生命周期,包括启动新会话、恢复先前会话、记录会话活动(包括状态更新)、识别活动会话以及管理会话数据的删除。ADK提供几种具有不同存储机制的SessionService实现,用于会话历史和临时数据,如InMemorySessionService,它适用于测试但不提供跨应用程序重启的数据持久性。

# 示例:使用InMemorySessionService
# 这适用于本地开发和测试,其中数据
# 不需要在应用程序重启后持久化

from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.adk.agents import LlmAgent
from google.genai import types
import asyncio
import nest_asyncio

# 创建会话服务
session_service = InMemorySessionService()

# 定义智能体
agent = LlmAgent(
    name="memory_agent",
    model="gemini-2.0-flash",
    instruction="你是一个有记忆的智能体。你可以记住和回忆信息。",
    description="具有内存管理能力的智能体"
)

# 创建运行器
runner = Runner(
    agent=agent,
    app_name="memory_app",
    session_service=session_service
)

# 运行智能体
async def run_agent_with_memory():
    """运行具有内存的智能体"""
    # 创建会话
    session = await session_service.create_session(
        app_name="memory_app",
        user_id="user_123",
        session_id="session_456"
    )
    
    # 发送消息
    content = types.Content(
        role='user',
        parts=[types.Part(text="请记住我的名字是张三,我喜欢蓝色。")]
    )
    
    # 运行智能体
    events = runner.run(
        user_id="user_123",
        session_id="session_456",
        new_message=content
    )
    
    # 处理响应
    for event in events:
        if event.is_final_response():
            print(f"智能体响应: {event.content.parts[0].text}")
            break

# 示例使用
if __name__ == "__main__":
    nest_asyncio.apply()
    asyncio.run(run_agent_with_memory())

状态管理

状态是会话中存储的临时数据,用于在对话过程中维护上下文。状态可以包含用户偏好、对话历史、临时变量等。

# 状态管理示例
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.adk.agents import LlmAgent
from google.genai import types
import asyncio
import nest_asyncio

# 创建会话服务
session_service = InMemorySessionService()

# 定义智能体
agent = LlmAgent(
    name="state_agent",
    model="gemini-2.0-flash",
    instruction="你是一个有状态的智能体。你可以维护和更新状态。",
    description="具有状态管理能力的智能体"
)

# 创建运行器
runner = Runner(
    agent=agent,
    app_name="state_app",
    session_service=session_service
)

# 运行智能体
async def run_agent_with_state():
    """运行具有状态的智能体"""
    # 创建会话
    session = await session_service.create_session(
        app_name="state_app",
        user_id="user_123",
        session_id="session_456"
    )
    
    # 更新状态
    session.state["user_name"] = "张三"
    session.state["user_preference"] = "蓝色"
    session.state["conversation_count"] = 1
    
    # 发送消息
    content = types.Content(
        role='user',
        parts=[types.Part(text="请告诉我我的偏好。")]
    )
    
    # 运行智能体
    events = runner.run(
        user_id="user_123",
        session_id="session_456",
        new_message=content
    )
    
    # 处理响应
    for event in events:
        if event.is_final_response():
            print(f"智能体响应: {event.content.parts[0].text}")
            break

# 示例使用
if __name__ == "__main__":
    nest_asyncio.apply()
    asyncio.run(run_agent_with_state())

内存管理

内存是智能体的长期知识存储,可以跨会话和交互保留信息。内存通常使用向量数据库实现,支持语义搜索。

# 内存管理示例
from google.adk.sessions import InMemorySessionService
from google.adk.memory import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.agents import LlmAgent
from google.genai import types
import asyncio
import nest_asyncio

# 创建会话和内存服务
session_service = InMemorySessionService()
memory_service = InMemoryMemoryService()

# 定义智能体
agent = LlmAgent(
    name="memory_agent",
    model="gemini-2.0-flash",
    instruction="你是一个有长期内存的智能体。你可以存储和检索信息。",
    description="具有长期内存管理能力的智能体"
)

# 创建运行器
runner = Runner(
    agent=agent,
    app_name="memory_app",
    session_service=session_service,
    memory_service=memory_service
)

# 运行智能体
async def run_agent_with_memory():
    """运行具有长期内存的智能体"""
    # 创建会话
    session = await session_service.create_session(
        app_name="memory_app",
        user_id="user_123",
        session_id="session_456"
    )
    
    # 存储信息到内存
    await memory_service.store(
        app_name="memory_app",
        user_id="user_123",
        content="用户张三喜欢蓝色,住在北京。",
        metadata={"type": "user_profile", "timestamp": "2024-01-01"}
    )
    
    # 发送消息
    content = types.Content(
        role='user',
        parts=[types.Part(text="请告诉我关于我的信息。")]
    )
    
    # 运行智能体
    events = runner.run(
        user_id="user_123",
        session_id="session_456",
        new_message=content
    )
    
    # 处理响应
    for event in events:
        if event.is_final_response():
            print(f"智能体响应: {event.content.parts[0].text}")
            break

# 示例使用
if __name__ == "__main__":
    nest_asyncio.apply()
    asyncio.run(run_agent_with_memory())

一览

什么: 智能体需要不同类型的内存来有效运行。短期内存用于当前交互的上下文,而长期内存用于跨会话保留信息。没有适当的内存管理,智能体无法维护上下文或从过去的经验中学习。

为什么: 内存管理通过以下方式提供解决方案:

  • 维护对话上下文和连贯性
  • 实现个性化和定制体验
  • 支持学习和改进
  • 提供跨会话的信息持久性
  • 启用复杂的状态管理

经验法则: 当智能体需要维护上下文、个性化交互或从过去经验中学习时使用内存管理。它特别适用于:

  • 对话系统
  • 个性化推荐
  • 学习系统
  • 状态管理
  • 知识管理

关键要点

  • 内存管理包括短期(上下文)和长期(持久)内存
  • 短期内存受上下文窗口限制,需要有效管理
  • 长期内存使用向量数据库实现语义搜索
  • 现代框架如Google ADK提供内置内存管理支持
  • 有效的内存管理需要平衡相关性和容量
  • 内存管理对于构建智能和上下文感知的智能体至关重要

结论

内存管理是智能体系统的基本组成部分,使它们能够维护上下文、从经验中学习并提供个性化体验。通过提供短期和长期内存的有效管理,智能体可以超越简单的反应性行为,实现真正的智能和适应性。

掌握内存管理对于构建能够处理复杂、上下文相关任务的智能体系统至关重要。它提供了维护状态、保留信息和从经验中学习所需的工具和技术,使智能体能够提供个性化和智能的交互。

参考文献

  1. Google ADK文档:https://google.github.io/adk-docs/
  2. LangChain内存管理:https://python.langchain.com/docs/modules/memory/
  3. 向量数据库和语义搜索:https://arxiv.org/abs/2308.11452
  4. 智能体内存研究:https://arxiv.org/abs/2308.11452
第8章:内存管理