第十一篇：赋予 AI 以“记忆” —— Google ADK 智能代理中的记忆机制与能力演进-海口c网

本文将带你从零开始构建一个具有完整记忆管理功能的智能个人助理。我们不仅会深入探讨Google ADK的核心概念，更会通过实际代码展示如何在生产环境中实现持久化、智能化的代理记忆系统。

前言：为什么记忆对AI代理如此重要？

想象一下，每次你与Siri对话时，它都完全忘记了之前的交流。你需要重新介绍自己，重新说明你的偏好，重新建立上下文。这样的体验无疑是糟糕的。一个真正智能的代理，应该能够：

记住用户偏好：你喜欢什么样的音乐、餐厅、工作方式

维护对话上下文：理解当前讨论的主题和进展

学习历史经验：从过往交互中获得洞察，提供更好的服务

个性化服务：基于了解的信息提供定制化建议

本文将通过Google ADK框架，展示如何构建这样一个"有记忆"的智能代理。

核心概念深入解析

1. 会话管理：Session的生命周期

在ADK中，Session 是所有交互的基础容器。让我们先看看一个完整的会话是如何定义的：

# Session 对象包含以下核心属性：
{"id": "session_123",           # 唯一标识符"state": {                     # 状态字典"user:name": "张三",        # 用户级别状态"current_task": "booking",  # 会话级别状态"temp:cache": "..."        # 临时状态},"events": [...],              # 事件历史"metadata": {                 # 元数据"app_name": "assistant","user_id": "user_456","created_at": "2024-01-01T00:00:00Z"}
}

2. 状态前缀：巧妙的作用域设计

ADK采用前缀系统来管理不同作用域的状态：

前缀	作用域	生命周期	使用场景
(无前缀)	当前会话	会话结束后清除	临时上下文、对话状态
`user:`	用户级别	跨会话持久化	用户偏好、个人信息
`app:`	应用级别	全局共享	系统配置、共享设置
`temp:`	临时	不持久化	中间计算结果、缓存

这种设计的巧妙之处在于，开发者通过简单的命名约定就能控制数据的生命周期和可见性。

实战：构建智能个人助理

现在让我们通过实际代码来构建一个具有完整记忆管理功能的个人助理。

基础版本：核心记忆功能

首先，我们实现基础的偏好管理工具：

"""
个人助理 - 记忆管理示例这个模块实现了一个具有记忆功能的个人助理，能够：
1. 保存和检索用户偏好
2. 维护会话状态
3. 提供个性化的交互体验
"""import json
from typing import Optional, Dict, Any
from google.adk.agents import Agent
from google.adk.tools import ToolContextdef save_user_preference(tool_context: ToolContext, preference_name: str, preference_value: str) -> str:"""保存用户偏好到用户配置文件中，供将来参考。Args:tool_context: 工具上下文对象preference_name: 偏好设置的名称（如 'home_city', 'favorite_restaurant'）preference_value: 偏好设置的值Returns:确认消息"""try:# 使用 user: 前缀的状态会在使用持久化 SessionService 时跨会话保存state_key = f"user:{preference_name}"tool_context.state[state_key] = preference_value# 同时更新一个用户偏好列表，便于管理preferences_key = "user:saved_preferences"saved_preferences = tool_context.state.get(preferences_key, [])if preference_name not in saved_preferences:saved_preferences.append(preference_name)tool_context.state[preferences_key] = saved_preferencesreturn f"好的，我已经记住您的{preference_name}是{preference_value}。"except Exception as e:return f"抱歉，保存偏好时遇到了问题：{str(e)}"def recall_user_info(tool_context: ToolContext, info_key: str) -> str:"""从用户配置文件中回忆特定的用户信息。Args:tool_context: 工具上下文对象info_key: 要查询的信息键名Returns:用户信息或未找到信息的消息"""try:# 首先从用户状态中查找state_key = f"user:{info_key}"value = tool_context.state.get(state_key)if value:return f"我记得您的{info_key}是：{value}"# 如果没有找到具体信息，尝试搜索记忆（如果配置了 MemoryService）try:search_results = tool_context.search_memory(query=f"用户 {info_key}")if search_results and hasattr(search_results, 'results') and search_results.results:# 处理搜索结果（简化版本）return f"从过去的对话中，我找到了关于您{info_key}的信息：{search_results.results[0] if search_results.results else '未找到相关信息'}"except Exception:# 如果没有配置 MemoryService 或搜索失败，继续passreturn f"抱歉，我还没有记录您的{info_key}信息。您可以告诉我，我会记住的。"except Exception as e:return f"查询信息时遇到了问题：{str(e)}"def list_user_preferences(tool_context: ToolContext) -> str:"""列出已保存的所有用户偏好设置。Args:tool_context: 工具上下文对象Returns:用户偏好列表"""try:preferences_key = "user:saved_preferences"saved_preferences = tool_context.state.get(preferences_key, [])if not saved_preferences:return "我还没有保存任何关于您的偏好信息。"preferences_list = []for pref_name in saved_preferences:state_key = f"user:{pref_name}"value = tool_context.state.get(state_key, "未设置")preferences_list.append(f"- {pref_name}: {value}")return f"以下是我记住的您的偏好设置：\n" + "\n".join(preferences_list)except Exception as e:return f"获取偏好列表时遇到了问题：{str(e)}"def update_conversation_context(tool_context: ToolContext, context_key: str, context_value: str) -> str:"""更新当前对话的上下文信息（仅在当前会话中有效）。Args:tool_context: 工具上下文对象context_key: 上下文键名context_value: 上下文值Returns:确认消息"""try:# 使用无前缀的状态键，仅在当前会话中保存tool_context.state[context_key] = context_valuereturn f"好的，我已经记录了当前对话的{context_key}：{context_value}"except Exception as e:return f"更新对话上下文时遇到了问题：{str(e)}"def get_conversation_summary(tool_context: ToolContext) -> str:"""获取当前对话的摘要信息。Args:tool_context: 工具上下文对象Returns:对话摘要"""try:# 收集当前会话的非用户前缀状态session_data = {}user_data = {}for key, value in tool_context.state.items():if key.startswith("user:"):user_data[key[5:]] = value  # 移除 "user:" 前缀elif not key.startswith("app:") and not key.startswith("temp:"):session_data[key] = valuesummary_parts = []if user_data:summary_parts.append("用户信息：")for key, value in user_data.items():summary_parts.append(f"  - {key}: {value}")if session_data:summary_parts.append("当前对话上下文：")for key, value in session_data.items():summary_parts.append(f"  - {key}: {value}")if not summary_parts:return "当前没有记录的对话信息。"return "\n".join(summary_parts)except Exception as e:return f"获取对话摘要时遇到了问题：{str(e)}"# 创建个人助理代理
root_agent = Agent(name="personal_assistant",model="gemini-2.0-flash",description="一个能够记住用户偏好并提供个性化服务的智能个人助理",instruction="""你是一个非常有帮助的个人助理。你具有以下能力：1. **记忆用户偏好**：使用 'save_user_preference' 工具来记住重要的用户信息，如：- home_city（居住城市）- favorite_restaurant（喜欢的餐厅）- preferred_language（偏好语言）- work_schedule（工作时间）- dietary_restrictions（饮食限制）等等2. **回忆用户信息**：使用 'recall_user_info' 工具来查询用户的历史偏好和信息3. **管理对话上下文**：使用 'update_conversation_context' 工具记录当前对话的重要信息4. **提供个性化服务**：基于记住的用户信息来个性化你的回复和建议5. **状态总结**：使用 'get_conversation_summary' 和 'list_user_preferences' 工具来回顾信息**使用指南**：
- 当用户分享重要的个人信息时，主动使用保存工具
- 在提供建议时，先回忆相关的用户偏好
- 根据用户的居住地、喜好等信息来个性化回复
- 保持友好、专业的态度
- 如果不确定某个信息是否重要，可以询问用户是否需要记住记住：你的目标是成为一个真正了解用户并能提供个性化帮助的智能助理。""",tools=[save_user_preference,recall_user_info,list_user_preferences,update_conversation_context,get_conversation_summary]
)

高级版本：企业级记忆管理

对于更复杂的场景，我们需要支持分类管理、任务追踪等高级功能：

"""
高级个人助理 - 完整记忆管理示例这个版本包含了更完整的记忆管理功能，包括：
1. 多种类型的偏好管理
2. 智能上下文理解
3. 任务追踪和提醒
4. 更好的错误处理
"""import json
import datetime
from typing import Optional, Dict, Any, List
from google.adk.agents import Agent
from google.adk.tools import ToolContextdef save_user_preference(tool_context: ToolContext, preference_name: str, preference_value: str, category: str = "general") -> str:"""保存用户偏好，支持分类管理。Args:tool_context: 工具上下文对象preference_name: 偏好设置的名称preference_value: 偏好设置的值category: 偏好分类（如 'personal', 'work', 'food', 'travel'）Returns:确认消息"""try:# 保存具体偏好state_key = f"user:pref_{preference_name}"tool_context.state[state_key] = {"value": preference_value,"category": category,"updated_at": datetime.datetime.now().isoformat()}# 更新偏好列表preferences_key = "user:saved_preferences"saved_preferences = tool_context.state.get(preferences_key, {})if category not in saved_preferences:saved_preferences[category] = []if preference_name not in saved_preferences[category]:saved_preferences[category].append(preference_name)tool_context.state[preferences_key] = saved_preferencesreturn f"好的，我已经记住您在{category}类别下的{preference_name}是{preference_value}。"except Exception as e:return f"抱歉，保存偏好时遇到了问题：{str(e)}"def recall_user_info(tool_context: ToolContext, info_key: str) -> str:"""智能回忆用户信息，支持模糊搜索。"""try:# 直接查找direct_key = f"user:pref_{info_key}"direct_value = tool_context.state.get(direct_key)if direct_value:if isinstance(direct_value, dict):return f"我记得您的{info_key}是：{direct_value['value']} (分类：{direct_value.get('category', '未分类')})"else:return f"我记得您的{info_key}是：{direct_value}"# 模糊搜索all_prefs = {}for key, value in tool_context.state.items():if key.startswith("user:pref_"):pref_name = key[10:]  # 移除 "user:pref_" 前缀if info_key.lower() in pref_name.lower():all_prefs[pref_name] = valueif all_prefs:results = []for pref_name, pref_data in all_prefs.items():if isinstance(pref_data, dict):results.append(f"- {pref_name}: {pref_data['value']}")else:results.append(f"- {pref_name}: {pref_data}")return f"我找到了相关信息：\n" + "\n".join(results)# 尝试搜索长期记忆try:search_results = tool_context.search_memory(query=f"用户 {info_key}")if search_results and hasattr(search_results, 'results') and search_results.results:return f"从过去的对话中，我找到了关于{info_key}的信息：{search_results.results[0] if search_results.results else '未找到相关信息'}"except Exception:passreturn f"抱歉，我还没有记录关于{info_key}的信息。您可以告诉我，我会记住的。"except Exception as e:return f"查询信息时遇到了问题：{str(e)}"def list_preferences_by_category(tool_context: ToolContext, category: Optional[str] = None) -> str:"""按分类列出用户偏好。"""try:preferences_key = "user:saved_preferences"saved_preferences = tool_context.state.get(preferences_key, {})if not saved_preferences:return "我还没有保存任何关于您的偏好信息。"if category:if category not in saved_preferences:return f"没有找到{category}分类的偏好设置。"prefs_list = []for pref_name in saved_preferences[category]:state_key = f"user:pref_{pref_name}"pref_data = tool_context.state.get(state_key, "未设置")if isinstance(pref_data, dict):prefs_list.append(f"- {pref_name}: {pref_data['value']}")else:prefs_list.append(f"- {pref_name}: {pref_data}")return f"{category}分类的偏好设置：\n" + "\n".join(prefs_list)else:all_prefs = []for cat, prefs in saved_preferences.items():all_prefs.append(f"\n【{cat}】")for pref_name in prefs:state_key = f"user:pref_{pref_name}"pref_data = tool_context.state.get(state_key, "未设置")if isinstance(pref_data, dict):all_prefs.append(f"  - {pref_name}: {pref_data['value']}")else:all_prefs.append(f"  - {pref_name}: {pref_data}")return "您的所有偏好设置：" + "\n".join(all_prefs)except Exception as e:return f"获取偏好列表时遇到了问题：{str(e)}"def add_task_reminder(tool_context: ToolContext, task_description: str, due_date: Optional[str] = None, priority: str = "medium") -> str:"""添加任务提醒。"""try:tasks_key = "user:tasks"tasks = tool_context.state.get(tasks_key, [])task = {"id": len(tasks) + 1,"description": task_description,"due_date": due_date,"priority": priority,"status": "pending","created_at": datetime.datetime.now().isoformat()}tasks.append(task)tool_context.state[tasks_key] = tasksdue_info = f"，截止日期：{due_date}" if due_date else ""return f"好的，我已经为您添加了任务：{task_description} (优先级：{priority}{due_info})"except Exception as e:return f"添加任务时遇到了问题：{str(e)}"def list_tasks(tool_context: ToolContext, status: str = "all") -> str:"""列出任务列表。"""try:tasks_key = "user:tasks"tasks = tool_context.state.get(tasks_key, [])if not tasks:return "您目前没有任何任务。"if status != "all":tasks = [task for task in tasks if task.get("status") == status]if not tasks:return f"没有找到状态为'{status}'的任务。"task_list = []for task in tasks:status_icon = "✓" if task.get("status") == "completed" else "○"due_info = f" (截止：{task.get('due_date')})" if task.get('due_date') else ""priority_icon = {"high": "🔴", "medium": "🟡", "low": "🟢"}.get(task.get('priority', 'medium'), "🟡")task_list.append(f"{status_icon} {priority_icon} {task['description']}{due_info}")return f"您的任务列表：\n" + "\n".join(task_list)except Exception as e:return f"获取任务列表时遇到了问题：{str(e)}"def update_task_status(tool_context: ToolContext, task_id: int, new_status: str) -> str:"""更新任务状态。"""try:tasks_key = "user:tasks"tasks = tool_context.state.get(tasks_key, [])if not tasks:return "没有找到任何任务。"for task in tasks:if task.get("id") == task_id:old_status = task.get("status", "unknown")task["status"] = new_statustask["updated_at"] = datetime.datetime.now().isoformat()tool_context.state[tasks_key] = tasksreturn f"任务'{task['description']}'的状态已从'{old_status}'更新为'{new_status}'。"return f"没有找到ID为{task_id}的任务。"except Exception as e:return f"更新任务状态时遇到了问题：{str(e)}"def get_personalized_greeting(tool_context: ToolContext) -> str:"""根据用户信息生成个性化问候。"""try:# 获取用户姓名name_key = "user:pref_name"user_name = tool_context.state.get(name_key)if isinstance(user_name, dict):user_name = user_name.get('value')# 获取时间偏好timezone_key = "user:pref_timezone"timezone = tool_context.state.get(timezone_key)if isinstance(timezone, dict):timezone = timezone.get('value')# 构建问候语greeting_parts = []current_hour = datetime.datetime.now().hourif current_hour < 12:time_greeting = "早上好"elif current_hour < 18:time_greeting = "下午好"else:time_greeting = "晚上好"if user_name:greeting_parts.append(f"{time_greeting}，{user_name}！")else:greeting_parts.append(f"{time_greeting}！")# 检查待办任务tasks_key = "user:tasks"tasks = tool_context.state.get(tasks_key, [])pending_tasks = [task for task in tasks if task.get("status") == "pending"]if pending_tasks:greeting_parts.append(f"您还有{len(pending_tasks)}个待完成的任务。")return " ".join(greeting_parts) + " 我能为您做些什么？"except Exception as e:return f"生成个性化问候时遇到了问题：{str(e)}"# 创建高级个人助理代理
advanced_assistant = Agent(name="advanced_personal_assistant",model="gemini-2.0-flash",description="一个具有完整记忆管理、任务追踪和个性化服务的高级个人助理",instruction="""你是一个高级智能个人助理，具有强大的记忆管理和任务管理能力。**核心能力：**1. **分类偏好管理**：- 使用 'save_user_preference' 保存不同类别的用户偏好- 支持的分类：personal（个人）、work（工作）、food（饮食）、travel（旅行）、health（健康）等- 使用 'list_preferences_by_category' 查看特定分类的偏好2. **智能信息检索**：- 使用 'recall_user_info' 进行智能搜索，支持模糊匹配- 自动从用户状态和长期记忆中检索相关信息3. **任务管理**：- 使用 'add_task_reminder' 添加任务提醒- 使用 'list_tasks' 查看任务列表- 使用 'update_task_status' 更新任务状态4. **个性化交互**：- 使用 'get_personalized_greeting' 生成个性化问候- 基于用户历史和偏好定制回复**交互原则：**
- 主动询问并记录重要的用户信息
- 在提供建议时考虑用户的偏好和历史
- 定期提醒用户的待办任务
- 保持友好、专业和高效的服务态度
- 在不确定时主动澄清用户需求**示例使用场景：**
- "记住我喜欢中餐，分类为饮食偏好"
- "提醒我明天下午3点开会"
- "我的工作相关偏好有哪些？"
- "将任务1标记为完成"记住：你的目标是成为用户最可靠、最了解他们需求的智能助理。""",tools=[save_user_preference,recall_user_info,list_preferences_by_category,add_task_reminder,list_tasks,update_task_status,get_personalized_greeting]
)

实际使用效果展示

让我们看看这个系统在实际使用中的表现：

技术深入：状态管理最佳实践

状态设计原则

# ✅ 好的状态设计
{"user:profile_name": "张三",                    # 清晰的键名"user:preferences": {                           # 结构化数据"work": ["remote", "flexible_hours"],"food": ["chinese", "spicy"]},"current_context": "booking_restaurant",       # 描述性上下文"temp:api_cache": {...}                        # 明确的临时数据
}# ❌ 避免的状态设计
{"n": "张三",                                   # 不清晰的键名"data": "mixed_random_stuff",                  # 混乱的数据"user:temp_value": "should_be_temp_prefix"     # 错误的前缀使用
}

采用adk web 进行访问

# 访问 http://localhost:8000