使用 ADK 构建长期运行的 AI 代理

# 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:
    """确保所有状态机键已初始化以防止错误。"""
    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. 如果当前步骤是 'START'：询问姓名、电子邮件和入职日期。然后调用 'send_welcome_packet'。
2. 如果当前步骤是 'WELCOME_SENT'：告知用户正在等待文档签名。不要调用其他工具。
3. 如果当前步骤是 'DOCUMENTS_SIGNED'：委托 IT 配置给 'it_agent'。
4. 如果当前步骤是 'IT_PROVISIONED'：询问硬件追踪 ID，然后调用 'check_hardware_delivery'。
5. 如果当前步骤是 'HARDWARE_DELIVERED'：调用 'send_day_one_schedule'。
6. 如果当前步骤是 'COMPLETED'：确认入职完成。

始终保持在工具和当前状态中。不要跳过步骤。"""

# 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})。文档待签名。",
    }

# 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,
)

# 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": "文档签名已处理，代理已恢复。"}

# 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,
            }))

# 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,
)

JSON

复制

第二个回合验证了代理拒绝调用任何工具，并停留在 `WELCOME_SENT` 状态。第二个测试案例预先设置状态为 `IT_PROVISIONED`，并确认代理在模拟的48小时硬件延迟后正确恢复，依次调用 `check_hardware_delivery` 和 `send_day_one_schedule`，而不会丢失新员工的原始上下文。

本地运行评估：

Shell

复制

这些黄金测试可以直接插入到 CI/CD 管道中，在生产之前捕获状态机的回归问题。

## **部署到代理运行时**

当评估通过后，就可以进行部署了。给你的编码代理这个提示：

`"将此部署到代理运行时，并启用 Cloud Trace，以便我们可以监控生产中的暂停和恢复延迟。"`
Shell

复制

编码代理搭建了连接你的 ADK 应用程序到代理运行时的 AgentEngineApp 包装器：

Python

复制

使用单个命令进行部署：

`agents-cli deploy`
Shell

复制

代理运行时处理会话持久性、自动扩展（包括空闲时间的缩放至零），以及开箱即用的 Cloud Trace 集成。与本地针对 SQLite 运行的检查点和恢复架构相同，在生产环境中针对托管云存储工作，无需更改代码。

![图像 4：Diagram-3](https://storage.googleapis.com/gweb-developer-goog-blog-assets/images/Diagram-3.original.png)

## **接下来做什么**

无状态代理只是代理可以实现的一部分。本教程中的模式——持久化状态机、持续的检查点和恢复、基于事件的空闲时间处理以及多代理委托——将代理从对话玩具转变为可靠的生产后台进程，能够管理跨越数天或数周的工作流。

开始步骤：

*   克隆 [new-hire-onboarding 仓库](https://github.com/GoogleCloudPlatform/generative-ai/tree/main/agents/adk/new-hire-onboarding)，并在本地运行实时演示
*   探索 [ADK 文档](https://adk.dev/)以了解会话管理、多代理模式和评估框架
*   安装 [Agents CLI](https://google.github.io/agents-cli/) 以搭建、测试和部署自己的长运行代理

入职代理只是一个例子。任何包含人工介入暂停、跨系统交接或多日时间线的工作流都是这种架构的理想候选者。发票争议、采购审批、销售线索序列、合规审计——模式相同。定义状态机，持久化检查点，度过空闲时间，然后从离开的地方继续。 

[](https://developers.googleblog.com/build-long-running-ai-agents-that-pause-resume-and-never-lose-context-with-adk/)
上一篇

下一篇

[](https://developers.googleblog.com/building-with-gemini-embedding-2/)