Google ADK 支持构建异步长时间运行 AI 代理

Fire and Forget：用 Google ADK 构建异步、长时间运行的 AI Agent 2026 年 11 月 4 日 · 阅读时长 23 分钟 目录 为什么我们应该停止与 Agent“聊天” 为什么同步 HTTP 对长时间运行的 AI Agent 行不通 问题究竟出在哪里？ Fire-and-Forget 模式：HTTP 202 Accepted 与 Webhook Google ADK 中的人类参与（HITL）：暂停与恢复工作流 人工验证与长时间运行的操作 一个更复杂的 Agent 为什么主循环可以用纯 Python 编写 中断与人类参与 深入剖析幕后究竟发生了什么 结论与下一步

在 Google ADK 中暂停与恢复 Agent、人类参与，以及长时间运行的操作

如今大多数 Agent 用例都有一个 UI，用户和 Agent 在其中进行对话。这对某些问答节奏很快的场景确实有效，但我认为大多数 Agent 根本不会遵循“对话式”的模式。

因为说实话：如今的真实 Agent 生命周期很长。它们可能持续几分钟、几小时，甚至几天。而显然，你不会在聊天窗口前等上几个小时。

与人类的沟通可以遵循两种模式：同步和异步。

正如人与人之间的沟通遵循这两条原则，我认为 Agent 也应被同等对待。不是把它们当人，而是把它们当作同步或异步系统。

简单来说：Agent 的大部分时间都在等待。

它们消耗 RAM、消耗 CPU、消耗带宽。

如今大多数真实的 Agent 用例都需要人工验证步骤，或者对信号做出反应（例如 webhook 触发）。人工验证步骤可能需要好几个小时，在此期间 Agent 会怎样？如果连接中断了又会怎样？

在接下来的章节中，我想带你通过一个具体示例，演示如何用 Google 的 Agent Development Kit 实现异步概念。我在教程或博客文章中看到的大多数实现往往非常简陋，所以我想通过一系列文章呈现一个贴近真实的实现。

第 1 部分将涵盖 ADK 特有的概念——Runners、DatabaseSessionService、Events、FunctionCalls、RequestInput——它们被用于运行 Agent、持久化记忆、恢复执行等。我还会解释 webhook 的工作原理，以及如何使用它们与远程服务通信。这部分主要是本地开发，服务器运行在 localhost 上，数据库使用 SQLite。

第 2 部分将涵盖我们所构建内容的投产迁移。具体来说，我们会涉及 Cloud Run 这类服务，用来托管应用，并能在不使用时缩容到零（对成本友好）；Cloud Tasks 用来在后台带重试地委派任务；以及 Cloud Scheduler 用来调度任务。

让我们设想有一个 Agent 执行一项非常简单的任务：抓取 RSS 源并把最新内容做摘要，生成一份每日简报。

这是个不错的测试对象：它做的是真实的工作（API 调用 + LLM 摘要），而且耗时足够长，干等着可能会让人难受。我目前大约有 20 个源，读完全部大约需要 1 到 2 分钟。好吧，这还不算夸张，但想象一下扩展到 1000 个源，你就明白了。

这个 Agent 就是一个 LLM 加一个工具。工具本身不过是对源做一次循环：

import feedparser
FEEDS = [
 "https://blog.google/technology/ai/rss/",
 "https://blog.google/innovation-and-ai/infrastructure-and-cloud/google-cloud/rss/",
 "https://cloudblog.withgoogle.com/products/ai-machine-learning/rss/",
 "https://developers.googleblog.com/feeds/posts/default/",
 "https://simonwillison.net/atom/everything/",
 "https://hnrss.org/newest?q=htmx+OR+ADK+OR+%22agent+development%22&points=20",
 "https://huggingface.co/blog/feed.xml",
 "https://www.anthropic.com/rss.xml",
 "https://read.deeplearning.ai/the-batch/rss/",
 "https://thegradient.pub/rss/",
 "http://export.arxiv.org/rss/cs.AI",
 "http://export.arxiv.org/rss/cs.CL",
def fetch_rss_feeds() -> str:
 """Fetch the latest articles from the configured RSS feeds. Call this first."""
 results = []
 for url in FEEDS:
 entries = feedparser.parse(url)['entries']
 for entry in entries:
 if entry:
 title = entry.title or "(no title)"
 link = entry.link or ""
 published = entry.published if hasattr(entry, "published") else "unknown"
 summary = (entry.summary or "")[:300].strip()
 results.append(f"[{published}] {title}\n{link}\n{summary}\n")
 return "\n---\n".join(results) if results else "No articles found."

这是一个非常初级的实现。在生产环境中，我多半会使用 asyncio 库中的 TaskGroups 和 Semaphores。眼下，我们就把这个当作我们的工具。

这个 Agent 是一个带有该工具的简单 LLM：

from google.adk.agents import Agent
root_agent = Agent(
 name="daily-feed",
 model="gemini-4-argon",
 instruction="""
 You are a sharp tech analyst writing a daily AI & data engineering briefing.
 Your first action MUST be to call the fetch_rss_feeds tool to get today's articles.
 Then summarize ONLY the articles returned by that tool.
 Do NOT search the web. Do NOT add information from outside the fetched articles.
 For each entry, provide:
 - The title and source
 - A 1-2 sentence summary of why it matters
 - The article link
 Rules:
 - Be concise and opinionated. Skip fluff.
 - Prioritize novelty and impact.
 - Always cite your sources with the link.
 tools=[fetch_rss_feeds],
)

到目前为止，没什么难的。

在进入 Agent 的执行循环之前，我想先回顾一些 ADK 使用的概念和对象，因为它们在本文的剩余部分中至关重要。

现在我们已经有了这些构建模块，接下来把它们串起来。分步骤进行：

app_name=...
agent=...
session_service=...
async def _run(message: str, session_id: str, **kwargs):
 # Create a session attached to the service
 await session_service.create_session(
 app_name=app_name, 
 user_id="admin", 
 session_id=session_id
 )
 # Create a runner
 runner = Runner(

（原文较长，此处译出前 6000 字，完整内容请点「打开原文」。）