大多数智能体教程只停留在无状态聊天机器人层面。真实工作流要运行数周。

借助 Google Agent Development Kit，构建能够暂停数天、重启后不丢失上下文的长期运行 AI 智能体。

代码 + 教程 ➡️ https://t.co/ofoXioD5m4 https://t.co/Ne5E5SfQJs

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

来源：https://developers.googleblog.com/build-long-running-ai-agents-that-pause-resume-and-never-lose-context-with-adk/ 作者：Eric Dong (https://developers.googleblog.com/search/?author=Eric+Dong)，开发者关系工程师

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

HR 入职流程跨度两周；发票纠纷因等待供应商回复而搁置数天；销售线索跟进序列在一个月内跨越多个接触点。这些流程的特点是高“空闲时间”——智能体长时间处于休眠状态，等待人工签名、发货确认或审批关卡。无状态聊天机器人无法应对这种情况。

本教程将带你使用 Agent Development Kit (ADK) (https://adk.dev/) 构建一个新员工入职协调智能体，它能可靠地运行数周。该智能体会发送欢迎包、在员工签署文件时暂停数天、将 IT 配置委托给专门的子智能体、再次等待硬件交付，最后发送个性化第一天日程——全程不丢失任何上下文。

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

• 持久的记忆模式，而非将原始 JSON 直接存入向量数据库
• 事件驱动的休眠门控，而非主动轮询或阻塞线程
• 多智能体委派，而非单一智能体提示的巨型单体

完整源代码可在 GitHub (https://github.com/GoogleCloudPlatform/generative-ai/tree/main/agents/adk/new-hire-onboarding) 上获取。

Diagram-1

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

标准的无状态模式会将所有用户消息和模型响应追加到不断增长的对话历史中，然后将整个内容块传回下一次 LLM 调用。这在五分钟的问答会话中没问题。但经过数天或数周，它会以三种特定方式崩溃：

提示上下文污染——经过两周入职流程中数百轮的交互，对话历史充满了无关的闲聊、旧的工具输出和重复的指令。模型开始混淆当前处于哪一步。

Token 成本爆炸——每次推理调用都回放两周的完整对话历史，会迅速消耗 Token 预算。单次入职运行可能生成数千轮对话——其中大部分与当前决策无关。

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

解决方法不是更大的上下文窗口，而是一个根本不同的架构——其中智能体的状态是显式的、持久的，并且与原始聊天历史解耦。

用例：新员工入职

考虑一家公司引入新员工时会发生什么：

1. HR 发送欢迎包和文件链接
2. 空闲时间——员工签署文件需数天
3. IT 配置公司邮箱和 Slack 账号
4. 空闲时间——笔记本电脑运送到员工家中需数天
5. HR 发送个性化第一天日程

live-onboarding-overview

这不是单次对话。它是一个具有多个暂停-恢复周期、人工审批关卡和跨团队交接的后台进程。同样的模式也出现在发票纠纷处理（等待供应商回复，恢复进行 AP 路由）、销售线索跟进（在联系之间的暂停）以及许多其他运营工作流中。

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

Agents CLI (https://github.com/google/agents-cli) 是 Gemini Enterprise Agent Platform 的官方命令行界面。本教程不是手动运行 CLI 命令，而是使用编码智能体来完成繁重工作——为它提供高级意图驱动提示，它会为你处理脚手架。首先，全局安装 CLI：

uv tool install google-agents-cli

Shell

复制

然后给你的编码智能体以下提示：

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

Shell

复制

编码智能体会运行适当的 agents-cli 命令，生成项目结构，并从一开始就配置好持久化会话和记忆库设置。本教程后续将继续采用这种迭代、提示驱动的方式：描述你的需求，编码智能体就会生成每一节中展示的代码。

Diagram-2

将智能体建立在持久化状态机上

与其依赖对话历史来跟踪进度，不如定义一个显式的状态模式，让智能体随时知道自己处于工作流的哪一步。给你的编码智能体以下提示：

"添加一个状态机来跟踪入职进度。我需要诸如 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 (https://cloud.google.com/run?e=48754805) 等容器化环境中，容器冷启动、空闲时缩到零、并意外重启。如果会话存在于易失性内存中，每个进行中的入职运行都会丢失。给你的编码智能体以下提示：

"将会话存储切换到持久化 SQLite，这样智能体就能在服务器重启后继续运行。"

Shell

复制

将内存中的会话替换为 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

持久化 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 端点

公开 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]: “”“当员工签署合同后唤醒入职智能体。”“” 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=[sen