第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提供内置内存管理支持
- 有效的内存管理需要平衡相关性和容量
- 内存管理对于构建智能和上下文感知的智能体至关重要
结论
内存管理是智能体系统的基本组成部分,使它们能够维护上下文、从经验中学习并提供个性化体验。通过提供短期和长期内存的有效管理,智能体可以超越简单的反应性行为,实现真正的智能和适应性。
掌握内存管理对于构建能够处理复杂、上下文相关任务的智能体系统至关重要。它提供了维护状态、保留信息和从经验中学习所需的工具和技术,使智能体能够提供个性化和智能的交互。
参考文献
- Google ADK文档:https://google.github.io/adk-docs/
- LangChain内存管理:https://python.langchain.com/docs/modules/memory/
- 向量数据库和语义搜索:https://arxiv.org/abs/2308.11452
- 智能体内存研究:https://arxiv.org/abs/2308.11452