Agent-Docs/ADKLearning/google-adk-tutorial/06-session-and-state.md

第06章：会话与状态管理

📌 本章目标

• 理解 Session、State、Memory 三层上下文架构
• 掌握 SessionService 的使用方法
• 学习 State 的读写和管理
• 了解 MemoryService 和跨会话记忆
• 掌握 Runner 的使用方法

---

6.1 上下文管理架构

6.1.1 三层上下文模型

┌──────────────────────────────────────────────────────┐
│                    上下文管理架构                       │
│                                                       │
│  ┌────────────────────────────────────────────┐      │
│  │  Memory（记忆层）                           │      │
│  │  - 跨会话持久化知识                          │      │
│  │  - 可搜索的知识库                            │      │
│  │  - 管理者：MemoryService                     │      │
│  │  - 实现：InMemory / Vector DB / Cloud       │      │
│  └────────────────────────────────────────────┘      │
│                                                       │
│  ┌────────────────────────────────────────────┐      │
│  │  Session（会话层）                           │      │
│  │  - 单次对话线程                              │      │
│  │  - 消息历史（Events）                         │      │
│  │  - 管理者：SessionService                    │      │
│  │  - 实现：InMemory / Firestore / DB           │      │
│  └────────────────────────────────────────────┘      │
│                                                       │
│  ┌────────────────────────────────────────────┐      │
│  │  State（状态层）                              │      │
│  │  - 会话内的临时数据                           │      │
│  │  - temp: 临时状态（当前调用有效）              │      │
│  │  - 持久状态（会话期间有效）                    │      │
│  └────────────────────────────────────────────┘      │
└──────────────────────────────────────────────────────┘

6.1.2 各层对比

层级	生命周期	用途	管理服务
Memory	跨会话持久化	长期知识、用户偏好	MemoryService
Session	单次对话	消息历史、对话上下文	SessionService
State	会话期间	临时数据、中间结果	session.state
temp: State	单次调用	工具间数据传递	session.state

---

6.2 Session（会话）

6.2.1 Session 是什么？

Session 代表用户与 Agent 系统之间的一次对话交互。它包含：

• 消息历史：按时间顺序记录的所有消息和事件
• 状态数据：对话过程中产生的临时数据
• 元信息：应用名称、用户 ID、会话 ID 等

6.2.2 SessionService（会话服务）

"""
SessionService — 会话管理服务
负责创建、读取、更新和删除会话
"""

from google.adk.sessions import InMemorySessionService  # 导入内存会话服务
import asyncio                                      # 导入异步库


async def session_demo():
    """演示 SessionService 的使用"""

    # ========================================
    # 创建会话服务
    # ========================================

    # InMemorySessionService：将数据存储在内存中
    # 适用于开发和测试，数据在程序重启后丢失
    session_service = InMemorySessionService()

    # ========================================
    # 创建会话
    # ========================================

    session = await session_service.create_session(
        app_name="my_app",                    # 应用名称（用于标识应用）
        user_id="user_001",                   # 用户唯一标识
        session_id="session_001",             # 会话唯一标识
        # state={"key": "value"},            # 可选：初始化状态
    )

    print(f"会话创建成功: {session.id}")       # 打印会话 ID
    print(f"应用名称: {session.app_name}")     # 打印应用名称
    print(f"用户 ID: {session.user_id}")       # 打印用户 ID

    # ========================================
    # 获取会话
    # ========================================

    existing_session = await session_service.get_session(
        app_name="my_app",                    # 应用名称
        user_id="user_001",                   # 用户 ID
        session_id="session_001",             # 会话 ID
    )

    print(f"获取会话: {existing_session.id}")  # 打印会话信息

    # ========================================
    # 删除会话
    # ========================================

    await session_service.delete_session(
        app_name="my_app",                    # 应用名称
        user_id="user_001",                   # 用户 ID
        session_id="session_001",             # 会话 ID
    )

    print("会话已删除")                        # 确认删除


# 运行演示
asyncio.run(session_demo())                   # 执行异步函数

6.2.3 SessionService 实现对比

实现	存储方式	持久化	适用场景
InMemorySessionService	内存	❌	开发、测试
FirestoreSessionService	Google Firestore	✅	生产环境
SqliteSessionService	SQLite 文件	✅	本地生产
自定义	自定义后端	✅	特殊需求

---

6.3 State（状态管理）

6.3.1 State 基础操作

"""
State 状态管理
在会话中读写和管理数据
"""

from google.adk.agents import Agent          # 导入 Agent 类
from google.adk.runners import Runner        # 导入运行器
from google.adk.sessions import InMemorySessionService  # 导入会话服务
from google.genai import types               # 导入类型定义
import asyncio                               # 导入异步库


# ========================================
# 定义使用 State 的工具
# ========================================

def add_to_cart(item: str, price: float, ctx) -> dict:
    """
    将商品添加到购物车

    Args:
        item (str): 商品名称
        price (float): 商品价格
        ctx: 工具上下文（自动注入），用于访问会话状态

    Returns:
        dict: 操作结果
    """
    # 获取当前购物车（如果不存在则初始化为空列表）
    cart = ctx.state.get("cart", [])         # 从 state 中读取购物车

    # 添加新商品到购物车
    cart.append({                            # 追加商品信息
        "item": item,                        # 商品名称
        "price": price,                      # 商品价格
    })

    # 更新 state 中的购物车
    ctx.state["cart"] = cart                 # 写回 state

    # 计算购物车总价
    total = sum(item["price"] for item in cart)  # 求和

    return {                                 # 返回结果
        "status": "success",                 # 状态
        "message": f"已将 {item} 添加到购物车",  # 消息
        "cart_items": len(cart),             # 购物车商品数量
        "total": total,                      # 总价
    }


def get_cart(ctx) -> dict:
    """
    获取购物车内容

    Args:
        ctx: 工具上下文

    Returns:
        dict: 购物车内容
    """
    cart = ctx.state.get("cart", [])         # 读取购物车

    if not cart:                             # 如果购物车为空
        return {                             # 返回空购物车信息
            "status": "success",
            "message": "购物车是空的",
            "items": [],
            "total": 0,
        }

    total = sum(item["price"] for item in cart)  # 计算总价

    return {                                 # 返回购物车内容
        "status": "success",
        "items": cart,                       # 商品列表
        "total": total,                      # 总价
    }


def clear_cart(ctx) -> dict:
    """
    清空购物车

    Args:
        ctx: 工具上下文

    Returns:
        dict: 操作结果
    """
    ctx.state["cart"] = []                   # 清空购物车

    return {                                 # 返回结果
        "status": "success",
        "message": "购物车已清空",
    }


# ========================================
# 创建 Agent
# ========================================

shopping_agent = Agent(
    name="shopping_assistant",               # Agent 名称
    model="gemini-2.0-flash",               # 模型
    instruction=(                           # 指令
        "你是一个购物助手。\n"
        "帮助用户管理购物车：添加商品、查看购物车、清空购物车。"
    ),
    tools=[                                 # 注册工具
        add_to_cart,                        # 添加商品
        get_cart,                           # 查看购物车
        clear_cart,                         # 清空购物车
    ],
)

6.3.2 temp: 临时状态

"""
temp: 临时状态
只在当前调用（invocation）中有效，调用结束后自动清除
适用于工具间传递中间数据
"""

def step1_process(query: str, ctx) -> dict:
    """
    处理步骤一：预处理数据

    Args:
        query (str): 用户查询
        ctx: 工具上下文

    Returns:
        dict: 预处理结果
    """
    # 对查询进行预处理
    processed = query.strip().lower()       # 去除空格并转小写

    # 保存到临时状态（temp: 前缀）
    # 这个数据只在当前调用中有效
    ctx.state["temp:processed_query"] = processed  # 临时存储

    return {                                 # 返回结果
        "status": "success",
        "processed_query": processed,
    }


def step2_enhance(ctx) -> dict:
    """
    处理步骤二：增强数据

    Args:
        ctx: 工具上下文

    Returns:
        dict: 增强结果
    """
    # 从临时状态读取步骤一的数据
    processed = ctx.state.get("temp:processed_query")  # 读取临时状态

    if not processed:                        # 如果没有数据
        return {                             # 返回错误
            "status": "error",
            "error_message": "请先执行预处理步骤。"
        }

    # 增强处理
    enhanced = f"[增强] {processed}"         # 模拟增强处理

    return {                                 # 返回结果
        "status": "success",
        "enhanced_query": enhanced,
    }

# temp: 状态 vs 普通状态：
# - temp:xxx：当前调用结束后自动清除
# - xxx：会话期间持久保存

---

6.4 Runner（运行器）

6.4.1 Runner 基础用法

"""
Runner — ADK 的核心执行引擎
负责协调 Agent、会话服务和工具调用
"""

from google.adk.agents import Agent          # 导入 Agent 类
from google.adk.runners import Runner        # 导入运行器
from google.adk.sessions import InMemorySessionService  # 导入会话服务
from google.genai import types               # 导入类型定义
import asyncio                               # 导入异步库


# ========================================
# 定义一个简单的 Agent
# ========================================

def get_greeting(name: str) -> dict:
    """生成问候语"""
    return {                                 # 返回问候语
        "status": "success",
        "greeting": f"你好，{name}！欢迎来到 ADK 教程。",
    }


agent = Agent(
    name="greeting_agent",                   # Agent 名称
    model="gemini-2.0-flash",               # 模型
    instruction="你是一个问候助手。",         # 指令
    tools=[get_greeting],                    # 注册工具
)


# ========================================
# 使用 Runner 运行 Agent
# ========================================

APP_NAME = "demo_app"                       # 应用名称
USER_ID = "user_001"                        # 用户 ID
SESSION_ID = "session_001"                  # 会话 ID


async def run_agent():
    """运行 Agent 的完整流程"""

    # 第一步：创建会话服务
    session_service = InMemorySessionService()  # 内存会话服务

    # 第二步：创建会话
    session = await session_service.create_session(
        app_name=APP_NAME,                   # 应用名称
        user_id=USER_ID,                     # 用户 ID
        session_id=SESSION_ID,               # 会话 ID
    )

    # 第三步：创建 Runner
    runner = Runner(
        agent=agent,                         # 要运行的 Agent
        app_name=APP_NAME,                   # 应用名称
        session_service=session_service,     # 会话服务
        # max_turns=10,                      # 可选：最大对话轮数
    )

    # 第四步：构造用户消息
    user_message = types.Content(
        role='user',                         # 角色：用户
        parts=[types.Part(text='你好，请向我打招呼')],  # 消息内容
    )

    # 第五步：运行 Agent 并处理事件流
    events = runner.run_async(
        user_id=USER_ID,                     # 用户 ID
        session_id=SESSION_ID,               # 会话 ID
        new_message=user_message,            # 用户消息
    )

    # 第六步：遍历事件流，提取响应
    async for event in events:               # 遍历所有事件
        print(f"事件类型: {type(event).__name__}")  # 打印事件类型

        # 检查是否为最终响应
        if event.is_final_response():        # 如果是最终响应
            response_text = event.content.parts[0].text  # 提取文本
            print(f"Agent 回复: {response_text}")  # 打印回复

        # 检查是否为工具调用事件
        if hasattr(event, 'function_call'):  # 如果有函数调用
            print(f"工具调用: {event.function_call}")  # 打印调用信息


# 运行
asyncio.run(run_agent())                     # 执行异步函数

6.4.2 多轮对话

"""
多轮对话示例
在同一个会话中进行多轮交互
"""

from google.adk.agents import Agent          # 导入 Agent 类
from google.adk.runners import Runner        # 导入运行器
from google.adk.sessions import InMemorySessionService  # 导入会话服务
from google.genai import types               # 导入类型定义
import asyncio                               # 导入异步库


agent = Agent(
    name="chat_agent",                       # Agent 名称
    model="gemini-2.0-flash",               # 模型
    instruction="你是一个友好的聊天助手，记住用户告诉你的信息。",  # 指令
)


async def multi_turn_chat():
    """多轮对话演示"""

    # 初始化服务
    session_service = InMemorySessionService()  # 会话服务
    await session_service.create_session(    # 创建会话
        app_name="chat_app",                 # 应用名称
        user_id="user_001",                  # 用户 ID
        session_id="chat_001",               # 会话 ID
    )

    # 创建 Runner
    runner = Runner(
        agent=agent,                         # Agent
        app_name="chat_app",                 # 应用名称
        session_service=session_service,     # 会话服务
    )

    # 模拟多轮对话
    questions = [                            # 对话列表
        "我叫小明，今年25岁。",               # 第一轮：自我介绍
        "我叫什么名字？",                     # 第二轮：测试记忆
        "我今年多大？",                       # 第三轮：测试记忆
    ]

    for question in questions:               # 遍历每轮对话
        print(f"\n[用户]: {question}")       # 打印用户消息

        # 构造消息
        content = types.Content(
            role='user',                     # 角色
            parts=[types.Part(text=question)],  # 内容
        )

        # 运行 Agent
        events = runner.run_async(
            user_id="user_001",              # 用户 ID
            session_id="chat_001",           # 会话 ID（同一个会话）
            new_message=content,             # 新消息
        )

        # 获取响应
        async for event in events:           # 遍历事件
            if event.is_final_response():    # 最终响应
                print(f"[Agent]: {event.content.parts[0].text}")  # 打印回复


asyncio.run(multi_turn_chat())               # 执行多轮对话

---

6.5 Memory（记忆服务）

6.5.1 Memory 概述

Memory 是跨会话的持久化知识存储，Agent 可以搜索和检索历史信息。

"""
MemoryService — 记忆管理服务
管理跨会话的持久化知识
"""

from google.adk.memory import InMemoryMemoryService  # 导入内存记忆服务


async def memory_demo():
    """演示 MemoryService 的使用"""

    # ========================================
    # 创建记忆服务
    # ========================================

    # InMemoryMemoryService：将记忆存储在内存中
    # 适用于开发测试，数据在重启后丢失
    memory_service = InMemoryMemoryService()

    # ========================================
    # 添加记忆
    # ========================================

    await memory_service.add_session_events_to_memory(
        app_name="my_app",                   # 应用名称
        user_id="user_001",                  # 用户 ID
        session_id="session_001",            # 会话 ID
    )

    # ========================================
    # 搜索记忆
    # ========================================

    results = await memory_service.search(
        query="用户偏好",                     # 搜索查询
        app_name="my_app",                   # 应用名称
        user_id="user_001",                  # 用户 ID
    )

    # 处理搜索结果
    for result in results:                   # 遍历结果
        print(f"记忆内容: {result}")         # 打印记忆


# 运行演示
import asyncio
asyncio.run(memory_demo())

6.5.2 Memory 与 Session 的区别

┌────────────────────────────────────────────────────┐
│  Session（会话）                                     │
│  ┌──────────────────────────────────────────────┐  │
│  │ 消息1 → 消息2 → 消息3 → ... → 消息N          │  │
│  │ state: {cart: [...], preferences: {...}}      │  │
│  └──────────────────────────────────────────────┘  │
│  生命周期：会话开始 → 会话结束                        │
└────────────────────────────────────────────────────┘

┌────────────────────────────────────────────────────┐
│  Memory（记忆）                                      │
│  ┌──────────────────────────────────────────────┐  │
│  │ 会话1摘要 → 会话2摘要 → 会话3摘要 → ...       │  │
│  │ 用户偏好 → 历史交互 → 长期知识                  │  │
│  └──────────────────────────────────────────────┘  │
│  生命周期：跨会话持久化                               │
└────────────────────────────────────────────────────┘

---

📌 本章小结

• ADK 采用三层上下文架构：Memory > Session > State
• Session：单次对话线程，由 SessionService 管理
• State：会话内数据，通过 ctx.state 读写
• temp: State：临时状态，当前调用有效
• Memory：跨会话持久化知识，由 MemoryService 管理
• Runner 是核心执行引擎，协调 Agent 和会话

下一章：第07章 - 回调机制与事件系统 → 学习如何使用回调函数控制 Agent 行为。