本文将带你从零开始构建一个具有完整记忆管理功能的智能个人助理。我们不仅会深入探讨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 ToolContext


def 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_preferences
        
        return 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 或搜索失败，继续
            pass
        
        return 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_value
        return 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] = value
        
        summary_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 ToolContext


def 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_preferences
        
        return 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] = value
        
        if 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:
            pass
        
        return 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] = tasks
        
        due_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_status
                task["updated_at"] = datetime.datetime.now().isoformat()
                tool_context.state[tasks_key] = tasks
                return 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().hour
        if 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

技术总结与最佳实践

通过这个完整的实现，我们总结出以下关键最佳实践：

🎯 设计原则

1. 分层记忆架构：即时记忆（session）→ 短期记忆（user state）→ 长期记忆（MemoryService）
2. 前缀驱动的作用域管理：通过命名约定控制数据生命周期
3. 优雅降级：在记忆服务不可用时提供合理的替代方案

🔧 实现技巧

1. 结构化状态存储：使用字典而非扁平字符串存储复杂信息
2. 索引维护：为频繁查询的数据维护索引结构
3. 时间戳记录：为所有状态变更添加时间信息

🚀 性能优化

1. 分层查找策略：直接查找 → 索引查找 → 模糊搜索
2. 合理的错误处理：不让单点故障影响整个系统
3. 监控和指标：为关键操作添加性能监控

结语：构建真正智能的代理

通过本文的深入探讨和实际实现，我们看到了如何利用Google ADK构建一个具有"记忆"的智能代理。这不仅仅是技术的堆砌，更是对用户体验的深度思考。

一个好的AI代理应该像一个优秀的人类助手一样：

• 记住重要的事情：用户的偏好、习惯、重要日期

• 理解上下文：知道当前在讨论什么，为什么讨论

• 学习和适应：从交互中获得洞察，不断改善服务质量

• 个性化服务：基于了解提供定制化的帮助

Google ADK通过其精心设计的会话管理、状态系统和记忆服务，为我们提供了实现这一目标的强大工具。而本文提供的完整实现，则展示了如何将这些概念转化为可在生产环境中运行的实际系统。

随着AI技术的不断发展，具有记忆和学习能力的代理将成为数字化转型的关键推动力。掌握这些核心技术，不仅能够构建更好的用户体验，更能为企业创造真正的价值。

---