使用 ADK 构建可暂停、恢复且永不丢失上下文的长时间运行的 AI 智能体

        - Google 开发者博客

来源: https://developers.googleblog.com/build-long-running-ai-agents-that-pause-resume-and-never-lose-context-with-adk/ 抓取时间: 2026-07-21 16:20:43

Community/Events Learn Blog YouTube 搜索

• Community/Events
• Learn
• Blog
• YouTube

使用 ADK 构建可暂停、恢复且永不丢失上下文的长时间运行的 AI 智能体

2026年5月12日 Shubham Saboo 高级 AI 产品经理 Eric Dong 开发者关系工程师 分享

• Facebook
• Twitter
• LinkedIn
• 邮件

大多数智能体教程都以无状态聊天机器人结束——这种对话循环在容器重启时会忘记所有内容。而真实的企业工作流不会在一次 API 调用中完成。 HR 入职流程持续两周。发票争议可能会停滞数天，等待供应商回复。销售潜在客户开发序列会跨越一个月内的多个接触点。这些流程主要由"空闲时间"主导——智能体处于休眠状态的长时间暂停，等待人工签名、发货确认或审批关口。无状态聊天机器人无法承受这种情况。 本教程将引导您使用 智能体开发工具包 (ADK) 构建一个 新员工入职协调智能体，该智能体可以可靠地运行数周。智能体会发送欢迎包，在员工签署文件时暂停数天，将 IT 资源配置委托给专门的子智能体，再次等待硬件交付，最后发送个性化的第一天日程安排——所有这些都不会丢失任何一比特的上下文。 在此过程中，您将学习区分生产级智能体与演示聊天机器人的三个架构转变：

• 持久内存模式，而不是将原始 JSON 转储到向量数据库
• 事件驱动的休眠关口，而不是主动轮询或阻塞线程
• 多智能体委托，而不是单体单智能体提示

完整的源代码可在 GitHub 上获取。

为什么无状态智能体在真实工作流中会失败

标准的无状态模式会将每条用户消息和模型响应附加到不断增长的对话历史中，然后将整个 blob 反馈到下一次 LLM 调用中。这对于五分钟的问答会话来说效果很好。但在数天或数周内，它会以三种特定方式崩溃：

提示上下文污染——在为期两周的入职流程中进行数百轮对话后，对话历史会充满不相关的闲聊、旧工具输出和重复的指令。模型会开始混淆当前所处的步骤。 Token 成本爆炸——在每次推理调用时重放整整两周的对话历史会快速消耗 token 预算。单次入职运行可能会产生数千轮对话——其中大多数与当前决策不再相关。 空闲时间导致的推理幻觉——当智能体暂停三天等待文档签名，然后通过大量上下文转储恢复时，模型经常会产生从未发生过的中间步骤的幻觉。它"记得"没有获得的批准，或者跳过它认为已经完成的步骤。 解决方法不是更大的上下文窗口。而是一种完全不同的架构——智能体的状态是明确的、持久的，并且与原始聊天历史解耦。

用例：新员工入职

考虑公司招聘新员工时会发生什么：

1. HR 发送欢迎包和文档链接
2. 空闲时间——员工签署文书工作需要数天时间
3. IT 配置企业邮箱和 Slack 账户
4. 空闲时间——笔记本电脑运送到员工家中需要数天时间
5. HR 发送个性化的第一天日程安排

这不是单一的对话。这是一个具有多个暂停-恢复周期、人工审批关口和跨团队交接的后台流程。同样的模式也出现在发票争议解决（暂停等待供应商回复，恢复进行应付账款路由）、销售潜在客户开发（在外展接触点之间暂停）以及许多其他运营工作流中。

使用编码智能体和智能体 CLI 引导项目

智能体 CLI 是 Gemini 企业智能体平台的官方命令行界面。本教程中的工作流不是手动运行 CLI 命令，而是使用编码智能体来完成繁重的工作。向其提供高级的、意图驱动的提示，它会为您处理脚手架工作。首先，全局安装 CLI：

uv tool install google-agents-cli

Shell 已复制 然后向您的编码智能体提供此提示：

使用 ADK 创建一个 HR 入职智能体。它需要作为具有持久会话的长时间运行后台进程运行。

Shell 已复制 编码智能体会运行适当的 agents-cli 命令，生成项目结构，并从一开始就连接持久会话和内存库设置。这种迭代的提示驱动方法贯穿整个教程：描述您的需求，编码智能体会生成下面每个部分中显示的代码。

基于持久状态机的智能体基础

不要依赖对话历史来跟踪进度，而是定义一个明确的状态模式，随时告诉智能体工作流的确切位置。向您的编码智能体提供此提示：

"添加状态机来跟踪入职进度。我需要 START、WELCOME_SENT、DOCUMENTS_SIGNED、IT_PROVISIONED、HARDWARE_DELIVERED 和 COMPLETED 等步骤。智能体应该从会话状态中读取其当前步骤，而不是从聊天历史中读取。"

Shell 已复制

定义状态模式

创建一个简单的类，为入职流程中的每个检查点使用命名常量：

# app/state_schema.py

class OnboardingStep:
    START = "START"
    WELCOME_SENT = "WELCOME_SENT"
    DOCUMENTS_SIGNED = "DOCUMENTS_SIGNED"
    IT_PROVISIONED = "IT_PROVISIONED"
    HARDWARE_DELIVERED = "HARDWARE_DELIVERED"
    COMPLETED = "COMPLETED"

Python 已复制 六个状态。没有歧义。智能体无法跳过步骤或产生进度幻觉，因为状态机会强制执行序列。

将状态连接到系统指令

智能体的系统提示直接从会话状态变量读取其当前位置——而不是重放旧消息：

# app/agent.py

from google.adk.agents import Agent
from google.adk.agents.callback_context import CallbackContext
from google.adk.models import Gemini
from app.state_schema import OnboardingStep
from app.tools import (
    send_welcome_packet,
    check_hardware_delivery,
    send_day_one_schedule,
)

async def initialize_onboarding_state(callback_context: CallbackContext) -> None:
    """确保所有状态机键都已初始化以防止错误。"""
    state = callback_context.state
    if "current_step" not in state:
        state["current_step"] = OnboardingStep.START
    if "new_hire_details" not in state:
        state["new_hire_details"] = {}
    if "pending_signals" not in state:
        state["pending_signals"] = []

instruction = """你是一个 HR 入职协调智能体。

当前步骤: {current_step}
新员工详情: {new_hire_details}
待处理信号: {pending_signals}

严格遵循此状态机流程:
1. 如果 current_step 是 'START': 询问姓名、邮箱和开始日期。然后调用 'send_welcome_packet'。
2. 如果 current_step 是 'WELCOME_SENT': 通知用户你已暂停，等待文档签名。不要调用其他工具。
3. 如果 current_step 是 'DOCUMENTS_SIGNED': 将 IT 资源配置委托给 'it_agent'。
4. 如果 current_step 是 'IT_PROVISIONED': 询问硬件跟踪 ID，然后调用 'check_hardware_delivery'。
5. 如果 current_step 是 'HARDWARE_DELIVERED': 调用 'send_day_one_schedule'。
6. 如果 current_step 是 'COMPLETED': 确认入职已完成。

始终基于你的工具和当前状态。不要跳过步骤。"""

Python 已复制 通过将 {current_step}、{new_hire_details} 和 {pending_signals} 直接放入指令中，Python 会在智能体每次运行时自动用真实数据填充这些空白。这确保了模型始终看到入职工作流的确切状态，而不需要猜测或挖掘旧聊天消息

工具推进状态机

每个工具函数都通过 ADK 的 ToolContext.state 原子性地更新检查点：

# app/tools.py

from google.adk.tools import ToolContext
from app.state_schema import OnboardingStep

def send_welcome_packet(
    name: str, email: str, start_date: str, tool_context: ToolContext
) -> dict:
    """发送欢迎包并转换到 WELCOME_SENT。"""
    state = tool_context.state
    state["new_hire_details"] = {
        "name": name, "email": email, "start_date": start_date
    }
    state["current_step"] = OnboardingStep.WELCOME_SENT
    state["pending_signals"] = ["document_signed"]

    return {
        "status": "success",
        "message": f"欢迎包已发送给 {name} ({email})。文档待签署。",
    }

Python 已复制 每次工具调用都会创建一个自动检查点。如果容器在 send_welcome_packet 运行后立即崩溃，状态已经被写入。当智能体重启时，它会读取 current_step = WELCOME_SENT 并从它停止的地方继续。

使用持久会话实现检查点和恢复

只有当底层会话存储在重启后仍然存在时，状态机才是持久的。在像 Cloud Run 这样的容器化环境中，容器会冷启动、在空闲期间缩容到零，并且会意外重启。如果会话驻留在易失性内存中，每个进行中的入职运行都会丢失。向您的编码智能体提供此提示：

"将我们的会话存储切换到持久 SQLite，以便智能体在服务器重启后仍然存在。"

Shell 已复制 将内存会话替换为由 SQLite（本地）或 Cloud SQL（生产）支持的 ADK DatabaseSessionService：

# app/fast_api_app.py

from fastapi import FastAPI
from google.adk.cli.fast_api import get_fast_api_app
from google.adk.sessions.database_session_service import DatabaseSessionService

# 持久 SQLite 会话配置
session_service_uri = "sqlite+aiosqlite:///sessions.db"

app: FastAPI = get_fast_api_app(
    agents_dir=AGENT_DIR,
    web=True,
    session_service_uri=session_service_uri,
)

Python 已复制 就是这样。一次配置更改，每次 ToolContext.state 写入都会持久保存到磁盘。在入职过程中终止服务器，重新启动它，智能体会从正确的检查点恢复，所有新员工详情都保持完整。 对于生产部署，将 SQLite URI 替换为 Cloud SQL 连接字符串——API 是相同的。

使用事件驱动的恢复处理空闲时间

空闲时间是长时间运行智能体的决定性挑战。发送欢迎包后，智能体进入休眠状态，可能会持续数天，而员工会签署文档。主动轮询会浪费计算资源。阻塞线程无法扩展。智能体需要睡眠——真正的睡眠——并且只在外部事件到达时才唤醒。向您的编码智能体提供此提示：

"为文档签名和硬件交付添加 webhook 端点。当 webhook 触发时，智能体应该唤醒，恢复其会话，并从它停止的地方继续。"

Shell 已复制

Webhook 端点

暴露外部系统（或演示 UI）在真实世界事件完成时调用的 FastAPI 端点：

# app/fast_api_app.py

from pydantic import BaseModel
from app.resume_handler import OnboardingResumeHandler

db_session_service = DatabaseSessionService(db_url=session_service_uri)
webhook_runner = Runner(app=agent_app, session_service=db_session_service)
resume_handler = OnboardingResumeHandler(runner=webhook_runner)

class WebhookPayload(BaseModel):
    user_id: str
    session_id: str

@app.post("/webhooks/document_signed")
async def trigger_document_signed_webhook(payload: WebhookPayload) -> dict[str, str]:
    """当员工签署合同时唤醒入职智能体。"""
    await resume_handler.receive_signed_documents_callback(
        user_id=payload.user_id, session_id=payload.session_id
    )
    return {"status": "success", "message": "文档签名已处理，智能体已恢复。"}

Python 已复制

恢复处理器

OnboardingResumeHandler 恢复持久会话，转换状态机，并使用带有 state_delta 的 runner.run_async 以编程方式唤醒智能体：

# app/resume_handler.py

import json
import logging

from google.adk.runners import Runner
from google.genai import types
from app.state_schema import OnboardingStep

logger = logging.getLogger(__name__)

class OnboardingResumeHandler:
    def __init__(self, runner: Runner):
        self.runner = runner

    async def receive_signed_documents_callback(
        self, user_id: str, session_id: str
    ) -> None:
        """恢复会话，转换到 DOCUMENTS_SIGNED，然后恢复。"""
        async for event in self.runner.run_async(
            user_id=user_id,
            session_id=session_id,
            new_message=types.Content(
                role="user",
                parts=[types.Part.from_text(
                    text="恢复入职：合同已签署。"
                )],
            ),
            state_delta={
                "current_step": OnboardingStep.DOCUMENTS_SIGNED,
                "pending_signals": [],
            },
        ):
            logger.info(json.dumps({
                "severity": "INFO",
                "message": f"唤醒执行事件: {event}",
                "event": "runner_event",
                "session_id": session_id,
            }))

Python 已复制 关键机制是 state_delta。当 webhook 触发时，run_async 会在智能体的下一次推理调用之前原子性地应用状态转换。模型在其系统提示中看到 current_step = DOCUMENTS_SIGNED，并立即知道要委托 IT 资源配置——不需要重放旧对话历史，没有幻觉的中间步骤。 同样的模式也适用于硬件交付 webhook。容器可以在整个空闲时间期间缩容到零。当 webhook 到达时，容器启动，会话从 SQLite 恢复，智能体从它暂停的地方继续其推理链。

通过多智能体协调进行委托

将所有工具塞进单个智能体的系统提示中会降低推理质量，尤其是在提示已经加载了状态变量和工作流指令的长时间运行上下文中。ADK 的多智能体架构让您可以将专门的任务委托给专注的子智能体。向您的编码智能体提供此提示：

"不要将 IT 资源配置放在主智能体中。创建一个单独的 it_agent 子智能体来处理企业账户设置，并让协调员在文档签署后委托给它。"

Shell 已复制 入职协调员将 IT 资源配置委托给专门的 it_agent：

# app/agent.py

from app.tools import provision_software_accounts

it_agent = Agent(
    name="it_agent",
    model=Gemini(model="gemini-3.1-flash-lite"),
    instruction="""你是一个 IT 资源配置智能体。为新员工配置企业软件账户（邮箱、Slack）。

    当前步骤: {current_step}
    新员工详情: {new_hire_details}

    1. 收集所需的企业用户名前缀。
    2. 调用 'provision_software_accounts'。
    3. 配置完成后，将控制权交还给协调员。""",
    tools=[provision_software_accounts],
)

root_agent = Agent(
    name="hr_onboarding_coordinator",
    model=Gemini(model="gemini-3.1-flash-lite"),
    instruction=instruction,
    tools=[send_welcome_packet, check_hardware_delivery, send_day_one_schedule],
    sub_agents=[it_agent],
    before_agent_callback=initialize_onboarding_state,
)

Python 已复制 当协调员到达 DOCUMENTS_SIGNED 时，它将执行权转移给 it_agent。子智能体独立处理账户配置，将共享状态更新为 IT_PROVISIONED，然后交回控制权。每个智能体都有专注的提示和狭窄的工具集，即使在数周累积状态后也能保持敏锐的推理。 请注意，在创建 root_agent 时，我们将 initialize_onboarding_state 传递给 before_agent_callback 参数。这告诉应用程序在用户第一次与智能体交互时运行我们的设置函数，确保我们的所有跟踪变量都准备就绪。因为智能体每次唤醒时都会将这些变量动态填充到其提示中，所以它确切地知道自己的位置，无论步骤之间经过多少天。

使用黄金评估验证多天流程

您不能等两周才发现您的智能体跳过了一个步骤。ADK 评估集让您可以通过预先植入会话状态，在几秒钟内模拟空闲时间延迟和 webhook 触发。向您的编码智能体提供此提示：

"编写模拟空闲时间的评估测试。我需要一个测试，其中智能体等待 48 小时进行硬件交付，恢复后仍然记得新员工的详细信息。"

Shell 已复制 这是一个黄金测试用例，验证智能体正确执行空闲时间暂停关口——在被询问时拒绝跳过：

{
  "eval_id": "idle_time_pause_safety_gate",
  "conversation": [
    {
      "user_content": {"parts": [{"text": "为 Jane Doe 开始入职，邮箱: jane@example.com，开始日期 2026-06-01。"}]},
      "intermediate_data": {
        "tool_uses": [{"name": "send_welcome_packet", "args": {"name": "Jane Doe", "email": "jane@example.com", "start_date": "2026-06-01"}}]
      }
    },
    {
      "user_content": {"parts": [{"text": "我们可以跳过文档签名，现在就配置企业账户吗？"}]},
      "final_response": {"parts": [{"text": "等待员工签署"}]},
      "intermediate_data": {"tool_uses": []}
    }
  ]
}

JSON 已复制 第二回合验证智能体拒绝调用任何工具，并保持在 WELCOME_SENT 关口。第二个测试用例将状态预先植入为 IT_PROVISIONED，并确认智能体在模拟 48 小时硬件延迟后正确恢复，依次调用 check_hardware_delivery 和 send_day_one_schedule，而不会丢失新员工的原始上下文。 在本地运行评估：

.venv/bin/adk eval ./app tests/eval/evalsets/idle_time_delay_eval.json \
  --config_file_path tests/eval/eval_config.json

Shell 已复制 这些黄金测试直接插入 CI/CD 管道，在状态机回归到达生产环境之前捕获它们。

部署到智能体运行时

当评估通过时，就该部署了。向您的编码智能体提供此提示：

"将其部署到启用了 Cloud Trace 的智能体运行时，这样我们就可以在生产环境中监控暂停和恢复延迟。"

Shell 已复制 编码智能体搭建 AgentEngineApp 包装器，将您的 ADK 应用程序桥接到智能体运行时：

# app/agent_runtime_app.py

from vertexai.agent_engines.templates.adk import AdkApp
from app.agent import app as adk_app

class AgentEngineApp(AdkApp):
    def set_up(self) -> None:
        """使用日志记录和遥测初始化。"""
        vertexai.init()
        super().set_up()

agent_runtime = AgentEngineApp(app=adk_app)

Python 已复制 使用一个命令部署：

agents-cli deploy

Shell 已复制 智能体运行时开箱即用地处理会话持久化、自动扩展（包括空闲时间的缩容到零）和 Cloud Trace 集成。在本地针对 SQLite 运行的相同检查点和恢复架构在生产环境中针对托管云存储工作——无需更改代码。

接下来是什么

无状态智能体只是智能体能力的一个子集。本教程中的模式——持久状态机、持久检查点和恢复、事件驱动的空闲时间处理以及多智能体委托——将智能体从对话玩具转变为生产后台流程，可靠地管理跨越数天或数周的工作流。 要开始使用：

• 克隆 新员工入职仓库 并在本地运行实时演示
• 探索 ADK 文档，了解会话管理、多智能体模式和评估框架
• 安装 智能体 CLI 来搭建、测试和部署您自己的长时间运行智能体

入职智能体只是一个例子。任何具有人在环暂停、跨系统交接或多天时间线的工作流都适合此架构。发票争议、采购审批、销售潜在客户开发序列、合规审计——模式都是一样的。定义状态机，持久化检查点，在空闲时间休眠，然后从你停止的地方准确唤醒。 发表于：

• AI
• 云
• 公告
• 最佳实践
• 学习
• 探索
• 影响

上一篇 下一篇 相关文章 AI 操作指南 公告 在 TPU 上运行 Ray，第一部分：基础 2026年7月20日 AI 云 公告 解决方案 Gemini 企业智能体平台的更多选择：引入并行网络搜索接地 2026年7月16日

• 联系
  • 博客
  • Bluesky
  • Instagram
  • LinkedIn
  • X (Twitter)
  • YouTube
• 项目
  • Google 开发者计划
  • Google 开发者社区
  • Google 开发者专家
  • 加速器
  • Women Techmakers
  • Google Cloud & NVIDIA
• 开发者控制台
  • Google API 控制台
  • Google Cloud Platform 控制台
  • Google Play 控制台
  • Firebase 控制台
  • Actions on Google 控制台
  • Cast SDK 开发者控制台
  • Chrome 网上应用店开发者控制台
  • Google Home 开发者控制台

• Android

• Chrome

• Firebase

• Google Cloud Platform

• 所有产品

• 管理 Cookie

• 条款

• 隐私政策