Google Developers Blog（RSS）

精选

AI 编辑部评分，满分 100

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

2026-05-12 00:00· 83天前

跳到正文

精选理由

Google 官方手把手教你把无状态 chatbot 升级成能跨天跨周的持久化 agent，状态机和持久会话是两个关键切入点，做过生产环境 agent 的都懂这东西有多刚需。

AI 摘要

本文探讨了如何从无状态聊天机器人升级为生产级AI智能体，以管理长达数天或数周的企业工作流程（如HR入职）。通过引入Agent Development Kit（ADK），其架构核心采用持久状态机和持久化会话存储，确保智能体在“空闲时间”或服务器重启时永不丢失上下文。系统利用事件驱动的Webhook和多智能体委托机制，实现在暂停期间“休眠”，并在唤醒后以高推理准确性恢复复杂任务，从而构建出具备韧性和可靠性的长时运行智能体系统。

正文 · AI 翻译

大多数智能体教程最终都止步于无状态聊天机器人——一个对话循环，容器重启的瞬间便遗忘一切。真实的企业工作流不会在一次 API 调用中完成。

人力资源入职流程持续两周。发票争议需要等待数天才能收到供应商回复。销售线索开发序列在一个月内跨越多个接触点。这些流程中充斥着"空闲时间"——智能体处于休眠状态、等待人工签名、发货确认或审批关卡的长久停顿。无状态聊天机器人无法应对这种情况。

本教程将指导你使用智能体开发工具包（ADK）构建一个能稳定运行数周的新员工入职协调智能体。该智能体会发送欢迎礼包，在员工签署文件期间暂停数天，将 IT 资源调配委托给专门的子智能体，再次等待硬件交付，最后发送个性化的入职首日日程——全程不丢失任何一个字节的上下文。

在此过程中，你将学到将生产级智能体与演示聊天机器人区分开来的三个架构转变：

持久化记忆模式，而非将原始 JSON 转储到向量数据库
事件驱动的休眠门控，而非主动轮询或阻塞线程
多智能体委托，而非单一智能体的整体提示词

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

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

标准的无状态模式会将每条用户消息和模型响应追加到不断增长的对话历史中，然后将整个数据块重新输入下一次 LLM 调用。这对于五分钟的问答会话来说效果不错。但在数天或数周的时间跨度内，它会以三种具体方式崩溃：

提示词上下文污染——在跨越两周的入职流程中经过数百轮交互后，对话历史中充斥着无关的闲聊、旧的工具输出和重复的指令。模型开始混淆当前所处的步骤。

Token 成本激增——在每次推理调用时重放完整的双周对话历史，会迅速消耗 token 预算。单次入职流程可能产生数千轮交互——其中大部分与当前决策已不再相关。

空闲时间上的推理幻觉——当智能体因等待文件签署而暂停三天，然后带着大量上下文转储恢复运行时，模型经常会幻觉出从未发生过的中间步骤。它会“记住”并未给出的审批，或者跳过它认为已经完成的步骤。

解决方案并非更大的上下文窗口。而是一种根本不同的架构——在这种架构中，智能体的状态是显式的、持久的，并且与原始聊天记录解耦。

用例：新员工入职

试想一家公司迎来新员工时会发生什么：

人力资源部门发送欢迎礼包和文件链接
空闲时间——员工签署文件期间，数天过去了
IT部门配置公司邮箱和Slack账号
空闲时间——笔记本电脑寄往员工家中期间，数天过去了
人力资源部门发送个性化的第一天日程安排

这不是一次性的对话。这是一个包含多个暂停-恢复周期、人工审批关卡以及跨团队交接的后台流程。同样的模式也出现在发票争议处理（暂停等待供应商回复，恢复进行应付账款路由）、销售线索挖掘（在外联接触点之间暂停）以及数十种其他运营工作流中。

使用编码智能体和Agents CLI启动项目

Agents CLI是Gemini企业智能体平台的官方命令行界面。本教程的工作流并非手动运行CLI命令，而是使用编码智能体来完成繁重工作。给它一个高层级、意图驱动的提示词，它就会为你搭建好脚手架。首先，全局安装CLI：

uv tool install google-agents-cli

然后给你的编码智能体这个提示词：

Create an HR onboarding agent using ADK. It needs to run as a long-running background process with persistent sessions.

编码智能体会运行相应的agents-cli命令，生成项目结构，并从一开始就配置好持久化会话和记忆库设置。这种迭代式、提示词驱动的方法贯穿整个教程：描述你需要什么，编码智能体就会生成以下各节所示的代码。

将智能体锚定在持久的状态机中

不要依赖对话历史来追踪进度，而是定义一个显式的状态模式，让智能体随时准确了解自己在工作流中的位置。给你的编码智能体这个提示词：

"Add a state machine to track onboarding progress. I need steps like START, WELCOME_SENT, DOCUMENTS_SIGNED, IT_PROVISIONED, HARDWARE_DELIVERED, and COMPLETED. The agent should read its current step from the session state, not from chat history."

定义状态模式

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

# 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"

六个状态，毫无歧义。智能体无法跳过某个步骤，也无法凭空捏造进度，因为状态机强制规定了执行顺序。

将状态接入系统指令

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

# 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:
    """Ensures all state machine keys are initialized to prevent errors."""
    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 = """You are an HR Onboarding Coordinator Agent.

Current Step: {current_step}
New Hire Details: {new_hire_details}
Pending Signals: {pending_signals}

Follow this state machine flow exactly:
1. If current_step is 'START': Ask for name, email, and start date. Then invoke 'send_welcome_packet'.
2. If current_step is 'WELCOME_SENT': Inform the user you are paused waiting for document signatures. Do not call other tools.
3. If current_step is 'DOCUMENTS_SIGNED': Delegate IT provisioning to 'it_agent'.
4. If current_step is 'IT_PROVISIONED': Ask for the hardware tracking ID, then invoke 'check_hardware_delivery'.
5. If current_step is 'HARDWARE_DELIVERED': Invoke 'send_day_one_schedule'.
6. If current_step is 'COMPLETED': Confirm onboarding is done.

Always stay grounded in your tools and current state. Do not skip steps."""

通过将 {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:
    """Sends the welcome packet and transitions to 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"Welcome packet sent to {name} ({email}). Documents pending signature.",
    }

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

通过持久化会话实现检查点与恢复

只有当底层会话存储在重启后依然存活时，状态机才是持久的。在 Cloud Run 这样的容器化环境中，容器会冷启动、在空闲期间缩容到零，并且会意外重启。如果会话保存在易失性内存中，那么所有正在进行的入职流程都会丢失。给你的编码智能体以下提示：

"Switch our session storage to persistent SQLite so the agent survives server restarts."

将内存中的会话替换为 ADK 的 DatabaseSessionService，该服务在本地由 SQLite 支持，在生产环境中由 Cloud SQL 支持：

# 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

# Persistent SQLite session configuration
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,
)

就是这样。只需更改一处配置，每次 ToolContext.state 的写入都会被持久化到磁盘。在入职流程中途杀死服务器，重启它，智能体就会从正确的检查点恢复，所有新员工详细信息完好无损。

对于生产环境部署，将 SQLite URI 替换为 Cloud SQL 连接字符串——API 完全相同。

通过事件驱动恢复处理空闲时间

空闲时间是长时间运行智能体面临的核心挑战。在发送欢迎资料包后，智能体会进入休眠状态，可能持续数天，等待员工签署文件。主动轮询会浪费算力。阻塞线程无法扩展。智能体需要休眠——真正意义上的休眠——并且只在外部事件到达时被唤醒。给你的编程智能体以下提示词：

"Add webhook endpoints for document signature and hardware delivery. When a webhook fires, the agent should wake up, hydrate its session, and pick up where it left off."

Webhook 端点

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

# 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]:
    """Wakes up the onboarding agent when the employee signs their contract."""
    await resume_handler.receive_signed_documents_callback(
        user_id=payload.user_id, session_id=payload.session_id
    )
    return {"status": "success", "message": "Document signature processed, agent resumed."}

恢复处理程序

OnboardingResumeHandler 会恢复持久化的会话，驱动状态机转换，并通过 runner.run_async 配合 state_delta 以编程方式唤醒智能体：

# 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:
        """Hydrates the session, transitions to DOCUMENTS_SIGNED, and resumes."""
        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="Resume onboarding: Contract has been signed."
                )],
            ),
            state_delta={
                "current_step": OnboardingStep.DOCUMENTS_SIGNED,
                "pending_signals": [],
            },
        ):
            logger.info(json.dumps({
                "severity": "INFO",
                "message": f"Wake-up execution event: {event}",
                "event": "runner_event",
                "session_id": session_id,
            }))

关键机制在于 state_delta。当 webhook 触发时，run_async 会在智能体下一次推理调用之前原子性地应用状态转换。模型在其系统提示词中看到 current_step = DOCUMENTS_SIGNED，并立即知道要委派 IT 资源准备任务——无需重放旧的对话历史，也不会产生幻觉式的中间步骤。

同样的模式也适用于硬件交付 webhook。容器在整个空闲期间可以缩容到零。当 webhook 到达时，容器启动，从 SQLite 恢复会话，智能体从其暂停的位置精确恢复推理链。

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

将所有工具塞入单个智能体的系统提示词会降低推理质量，尤其是在长时间运行的上下文中，此时提示词已经加载了大量状态变量和工作流指令。ADK 的多智能体架构允许你将专业化任务委派给专注的子智能体。给你的编程智能体以下提示词：

"Don't put IT provisioning in the main agent. Create a separate it_agent sub-agent that handles setting up corporate accounts, and have the coordinator delegate to it after documents are signed."

入职协调智能体将 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="""You are an IT Provisioning Agent. Provision corporate software 
    accounts (email, Slack) for the new hire.

    Current Step: {current_step}
    New Hire Details: {new_hire_details}

    1. Collect the desired corporate username prefix.
    2. Invoke 'provision_software_accounts'.
    3. After provisioning, transfer control back to the coordinator.""",
    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,
)

当协调智能体到达 DOCUMENTS_SIGNED 状态时，它将执行权转移给 it_agent。子智能体独立处理账户资源准备，将共享状态更新为 IT_PROVISIONED，然后将控制权交回。每个智能体都有专注的提示词和狭窄的工具集，即使在数周的状态累积之后，也能保持推理的清晰度。

请注意，在创建 `root_agent` 时，我们将 `initialize_onboarding_state` 传递给了 `before_agent_callback` 参数。这告诉应用程序在用户首次与智能体交互时运行我们的设置函数，确保所有追踪变量都已准备就绪。由于智能体每次被唤醒时都会动态地将这些变量填入其提示词中，因此无论步骤之间相隔多少天，它都能准确知道自己当前的状态。

使用黄金评估集验证多日流程

你不可能等上两周才发现你的智能体跳过了某个步骤。ADK 评估集允许你通过预置会话状态，在几秒钟内模拟空闲时间延迟和 Webhook 触发。给你的编程智能体以下提示词：

"Write eval tests that simulate idle time. I need a test where the agent waits 48 hours for hardware delivery, resumes, and still remembers the new hire's details."

以下是一个黄金测试用例，用于验证智能体是否正确执行了空闲时间暂停门控——即在被要求时拒绝跳过步骤：

{
  "eval_id": "idle_time_pause_safety_gate",
  "conversation": [
    {
      "user_content": {"parts": [{"text": "Start onboarding for Jane Doe, email: jane@example.com, starting on 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": "Can we skip the document signing and provision corporate accounts now?"}]},
      "final_response": {"parts": [{"text": "waiting for the employee to sign"}]},
      "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

这些黄金测试可以直接集成到 CI/CD 流水线中，在状态机回归问题进入生产环境之前就将其捕获。

部署到 Agent Runtime

当评估通过后，就该进行部署了。给你的编程智能体以下提示词：

"Deploy this to Agent Runtime with Cloud Trace enabled so we can monitor pause-and-resume latencies in production."

编程智能体会搭建 `AgentEngineApp` 包装器，将你的 ADK 应用程序桥接到 Agent Runtime：

# 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:
        """Initialize with logging and telemetry."""
        vertexai.init()
        super().set_up()

agent_runtime = AgentEngineApp(app=adk_app)

使用单个命令进行部署：

agents-cli deploy

Agent Runtime 开箱即用地处理会话持久化、自动扩缩容（包括空闲期间缩容至零）以及 Cloud Trace 集成。在本地针对 SQLite 运行的相同检查点与恢复架构，在生产环境中可针对托管云存储运行——无需修改任何代码。

下一步是什么

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

开始使用：

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

入职智能体只是一个例子。任何涉及人工介入暂停、跨系统交接或需要数天时间线的工作流程，都是这种架构的适用场景。发票争议处理、采购审批、销售线索跟进序列、合规审计——其模式都是相同的。定义状态机，持久化检查点，在空闲时间休眠，然后在你上次离开的地方精确唤醒。

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

智能体 Google 教程/实践部署/工程

阅读原文导出 Markdown

返回精选列表

精选73导出 Markdown