news 2026/9/13 19:26:47

ADK Python 应用容器 App 完全指南:从根 Agent 绑定到跨切面配置

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
ADK Python 应用容器 App 完全指南:从根 Agent 绑定到跨切面配置

ADK Python 应用容器 App 完全指南:从根 Agent 绑定到跨切面配置

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

App是 ADK Python(google.adk)应用的最顶层容器:它把根 Agent(或工作流根节点)与仅属于整个应用的配置——应用名称、插件、上下文缓存(context caching)、事件压缩(event compaction)与可恢复性(resumability)——绑定在一起。读完本文,你将掌握如何用App组织一个完整 ADK 应用、如何通过 CLI 或编程方式运行它、如何配置三项跨切面能力,以及它与旧版裸 Agent 传参方式(Runner(agent=..., app_name=...))之间的差异与迁移要点。

App 是什么:为什么需要这个顶层容器

一个 Agent 描述的只是对话中的一个参与者:一个模型、一段指令、一组工具,也许还有若干子 Agent。但在真实部署中,有很多东西不属于任何一个 Agent,而是属于整个应用:

  • 应用名称:会话(session)按应用名索引,一个应用只有一个名字;
  • 插件(plugins):需要观察整棵 Agent 树上的每一个 Agent、每一次模型调用和每一次工具调用;
  • 跨切面配置:上下文缓存、事件压缩、可恢复性作用于整棵 Agent 树,而不是树中的某一个节点。

App就是承载这些内容的地方。从源码看,它本质上是一个Pydantic 模型(继承自pydantic.BaseModel),持有root_agent加上上述应用级配置(见 src/google/adk/apps/app.py)。这样配置就跟随 Agent 定义一起流动,而不是散落在每一个构造Runner的调用点。值得注意的是,App没有单独的"根节点"字段:工作流的根BaseNode同样放在root_agent字段中(App的校验逻辑接受BaseAgentBaseNode两种类型)。

Runner在内部会把传入的参数归一化为AppRunner._resolve_app),所以直接传一个裸 Agent 也能跑;但只有App能携带这些跨切面配置。官方推荐路径是Runner(app=...),这一点在本文"App 与裸 Agent 的区别"一节详述。

快速开始:把 Agent 包装进 App

定义一个带工具调用的 Agent,再用App把它包起来。下面这个例子构建了一个带get_weather工具的天气 Agent,并将其放入名为weather_appApp中,同时挂载了LoggingPlugin

from google.adk.agents import LlmAgent from google.adk.apps import App from google.adk.plugins import LoggingPlugin def get_weather(city: str) -> str: """Returns a one-line weather report for the given city.""" return f"It is sunny in {city}." root_agent = LlmAgent( name="weather_agent", model="gemini-2.5-flash", instruction="Answer weather questions using the get_weather tool.", tools=[get_weather], ) app = App( name="weather_app", root_agent=root_agent, plugins=[LoggingPlugin()], )

这里的App是插件能够生效的关键:Runner(plugins=...)已弃用,而且也只有App能设置缓存、压缩与可恢复性三项配置。仓库中有一个完整可运行的官方示例,自定义了CountInvocationPlugin(统计 Agent 与 LLM 调用次数)并组合了上下文缓存与事件压缩配置,见 contributing/samples/core/app/agent.py。

运行你的 App:CLI 与编程两种方式

方式一:交给 CLI 命令

adk runadk webadk api_server都会为你构建Runner。它们会先在 Agent 模块中查找模块级变量app,只有在找不到App时才回退到root_agent。因此,把上面的App导出为模块级app变量,这三条命令就能自动拾取插件和全部跨切面配置。

当以这种方式加载 Agent 时,如果 app 名称与 Agent 所在目录不一致,Runner会打印一条警告日志并提示它期望的目录名。从源码看,这一逻辑位于Runner._enforce_app_name_alignment(src/google/adk/runners.py):Runner会通过_adk_origin_app_name元数据或模块路径推断 Agent 的来源目录,与app_name比对。重命名目录或 App 使二者一致即可消除警告。该警告同样会出现在会话查找失败的报错提示中(_format_session_not_found_message),提示可能是名称不一致导致Runner找不到会话。

方式二:编程驱动

如果你要自己驱动 App,则需要创建会话服务、把App交给Runner,然后运行一轮用户对话并逐条打印事件:

import asyncio from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.genai import types async def main() -> None: session_service = InMemorySessionService() session = await session_service.create_session( app_name=app.name, user_id="user" ) runner = Runner(app=app, session_service=session_service) async for event in runner.run_async( user_id="user", session_id=session.id, new_message=types.Content( role="user", parts=[types.Part(text="What is the weather in Zurich?")], ), ): if event.content and event.content.parts: print(event.author, event.content.parts[0].text) if __name__ == "__main__": asyncio.run(main())

注意:会话是在app.name下创建的。会话服务按应用名索引所有会话,所以App上的名称与查找会话时使用的名称必须一致。Runner在初始化时会把self.app_name = app_name or app.name(src/google/adk/runners.py),后续的get_session/create_session全部使用该名称。

App 与裸 Agent 的区别:两条路径并不等价

Runner同时接受App或裸 Agent,并且会在做任何其他事情之前先把裸 Agent 变成App。但这两条路径并不等价:

# Current: the App carries the application-wide configuration. runner = Runner(app=app, session_service=session_service) # Legacy: the agent is wrapped in an App for you. runner = Runner( app_name="weather_app", agent=root_agent, session_service=session_service, )

旧形式是 ADK 1.x 接受的写法,目前仍然支持,但它与App路径有三点差异:

  1. 跳过校验:包装过程会绕过App的字段校验(源码中用的是App.model_construct,见 src/google/adk/runners.py),因此一个App会拒绝的应用名,在这里反而会被接受;
  2. 跨切面配置无法设置context_cache_configevents_compaction_configresumability_config都会被留空。Runner没有对应的参数,App是设置它们的唯一途径;
  3. plugins参数已弃用Runner(plugins=[...])会触发DeprecationWarning;如果同时传入appplugins,会直接抛出ValueError——请把插件放到App上。相关校验逻辑在Runner._resolve_app(src/google/adk/runners.py)。

另外,当appapp_name同时给出时,会话查找以app_name为准,而app.name保持不变。同时传两者通常不是你想要的。Runner构造时还要求appagentnode三者恰好提供一个,否则抛出ValueError(src/google/adk/runners.py)。

App 字段详解

字段类型默认值说明
namestr必填应用名称,会话按它索引。
root_agentBaseAgentBaseNode必填执行入口。BaseNodegoogle.adk.workflow中的工作流节点基类,因此Workflow也可以作为根。
pluginslist[BasePlugin][]应用级插件,其回调对每一次 Agent、模型调用与工具调用都会触发。
context_cache_configContextCacheConfig \| NoneNone为应用内所有 LLM Agent 启用上下文缓存,缺省表示缓存关闭。
events_compaction_configEventsCompactionConfig \| NoneNone对较早的会话事件做摘要,防止上下文无限增长。
resumability_configResumabilityConfig \| NoneNone允许调用在长时间运行的函数调用处暂停,并在之后恢复。

App的 Pydantic 配置为extra="forbid"(src/google/adk/apps/app.py),会禁止未知关键字——拼错的字段名会抛出ValidationError,而不是被静默忽略。

应用命名规则

应用名必须以字母开头,之后可包含字母、数字、下划线和连字符;"user"被拒绝,因为它是保留给最终用户输入的名称。源码中的正则与校验见validate_app_name(src/google/adk/apps/app.py):

_VALID_APP_NAME_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9_-]*$")

如果你想在构造 App 之前先校验名称,可以直接从google.adk.apps.app导入validate_app_name使用。App的模型校验器也会在构造时调用它,并在root_agent缺失或类型错误时抛出相应的ValueError/TypeError

三个配置类型的导入位置

三个配置类型从不同位置导入,注意不要弄混:

from google.adk.agents.context_cache_config import ContextCacheConfig from google.adk.apps import ResumabilityConfig from google.adk.apps.app import EventsCompactionConfig

其中google.adk.apps只导出AppResumabilityConfig(见 src/google/adk/apps/init.py),EventsCompactionConfig需要从google.adk.apps.app导入。

服务挂在 Runner 上,而不是 App 上

App只保存声明式配置。会话、制品(artifact)、记忆(memory)与凭据(credential)服务都是Runner的构造参数——它们是部署层面的接线(wiring),而不是应用定义的一部分。因此,同一个App可以在测试中使用内存服务、在生产中使用持久化服务,而无需任何改动。

session_service是唯一必需的服务。本地开发时,InMemoryRunner会提供内存版的会话、制品和记忆服务,并接受同一个App

from google.adk.runners import InMemoryRunner runner = InMemoryRunner(app=app)

从源码看,InMemoryRunner在初始化时自动装配InMemoryArtifactServiceInMemorySessionServiceInMemoryMemoryService(src/google/adk/runners.py),非常适合测试与快速原型。Runner也支持auto_create_session=True,在会话不存在时自动创建。

配置跨切面功能

每个配置项只有设置到App上才会生效,未设置时均保持惰性(inert):

app = App( name="weather_app", root_agent=root_agent, context_cache_config=ContextCacheConfig( cache_intervals=10, ttl_seconds=1800, min_tokens=2048 ), events_compaction_config=EventsCompactionConfig( compaction_interval=5, overlap_size=1 ), resumability_config=ResumabilityConfig(is_resumable=True), )

上下文缓存:ContextCacheConfig

ContextCacheConfig定义在 src/google/adk/agents/context_cache_config.py,启用后作用于应用内所有LLM Agent:

  • cache_intervals:复用同一份缓存的最大调用次数,默认10,取值范围1100ge=1, le=100);
  • ttl_seconds:缓存存活时间(秒),默认1800(30 分钟),必须大于 0;
  • min_tokens:启用缓存所需的最小前置请求 token 数,默认0。注意它针对的是上一次请求的实际 prompt token 数。Gemini 模型自身的最小值始终生效:Gemini 2.5 为 2048 token,Gemini 3 为 4096 token;会话的首次请求不会创建缓存,缓存最早从第二轮开始。调高该值可以避免为小请求付出缓存存储开销;
  • create_http_options:可选的types.HttpOptions,用于给CachedContent.create()设置超时(如types.HttpOptions(timeout=10000)表示 10 秒)。缓存创建超时后请求会继续但不带缓存。

事件压缩:EventsCompactionConfig

EventsCompactionConfig定义在 src/google/adk/apps/_configs.py。它至少需要一个触发器,且两种触发器各是一对参数,必须成对设置:

  • 滑动窗口对compaction_interval+overlap_sizecompaction_interval表示"新用户发起的调用"数量达到多少时触发压缩(必须> 0);overlap_size表示从前一次压缩范围末尾向前包含多少个调用,使相邻摘要之间有重叠以保持上下文(>= 0);
  • token 预算对token_threshold+event_retention_sizetoken_threshold为调用后压缩的 token 阈值(必须> 0);event_retention_size表示触发压缩时保留多少个原始事件不被压缩(>= 0)。

校验器(_validate_trigger_params)强制要求:token_thresholdevent_retention_size必须同时设置;compaction_intervaloverlap_size必须同时设置;两组至少配置一组,否则抛出ValueError

从 src/google/adk/apps/compaction.py 的实现看,滑动窗口压缩的流程是:在每次调用完成后,统计自上次压缩以来新增的调用数,达到阈值后,从"新调用块起点往前数overlap_size个调用"处开始、到当前块最后一个调用为止,生成一个CompactedEvent摘要事件并追加到会话。例如compaction_interval=2, overlap_size=1时:调用 1、2 完成后生成覆盖 [1,2] 的摘要;调用 3、4 完成后生成覆盖 [2,4] 的摘要(与上一次重叠调用 2)。同时,token 阈值压缩会读取最近一次事件中的prompt_token_count(必要时以约 4 字符/token 估算),并在压缩前通过"最长自包含前缀"算法保证不会把成对的工具调用/响应拆散。

Summarizer 的默认行为:如果summarizer未设置,ADK 会用根 Agent 的模型自动构建LlmEventSummarizer——源码中_ensure_compaction_summarizer会检查config.summarizer,为空时要求根 Agent 是LlmAgent,并以LlmEventSummarizer(llm=agent.canonical_model)初始化(src/google/adk/apps/compaction.py)。因此,根 Agent 需要可用的 LLM 模型,否则压缩配置会因缺少摘要器而报错。

可恢复性:ResumabilityConfig

ResumabilityConfig定义在 src/google/adk/apps/_configs.py,目前只有一个字段is_resumable: bool = False。启用后,该能力作用于应用内所有 Agent:允许调用在遇到长时间运行的函数调用时暂停,并在之后从最后的事件恢复。注意 ADK 的恢复是**尽力而为(best-effort)**的,详见下一节限制说明。

已知限制与注意事项

  • 实验性配置EventsCompactionConfigResumabilityConfigContextCacheConfig三者都在构造时发出实验性警告(源码中均标注了@experimental装饰器),并可能在无通知的情况下变更;
  • EventsCompactionConfig未被 re-exportgoogle.adk.apps只导出AppResumabilityConfig,请从google.adk.apps.app导入EventsCompactionConfig
  • 恢复是尽力而为的:可能被恢复的工具必须具有幂等性,因为恢复只保证"至少一次"执行(at-least-once);并且暂停期间任何内存态都会丢失(这些约束在ResumabilityConfig的 docstring 中有明确说明,见 src/google/adk/apps/_configs.py)。

相关示例

官方提供了完整的应用配置示例(含自定义插件、上下文缓存与事件压缩组合),可直接参考: contributing/samples/core/app(示例代码见其中的 agent.py)。

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/9/13 19:26:38

低功耗Bandgap基准源设计实战:纳安级实现与温漂控制

1. 什么是低功耗Bandgap结构?它到底解决什么问题?Bandgap(带隙)基准源,是模拟电路里最基础也最“娇气”的模块之一——它不放大信号,不驱动负载,甚至不参与主信号通路,但整个芯片的精…

作者头像 李华
网站建设 2026/9/13 19:25:56

如何用 MLflow Agent Server 把 AI 代理托管为生产 REST API

如何用 MLflow Agent Server 把 AI 代理托管为生产 REST API 【免费下载链接】mlflow The open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI appli…

作者头像 李华
网站建设 2026/9/13 19:22:30

IS-95 CDMA基带链路全栈仿真:Simulink+S-Function可调试实现

简介:本资源是一个基于MATLAB Simulink构建的CDMA(码分多址)通信系统仿真工程包,面向通信工程专业本科生、研究生及无线通信入门学习者,用于深入理解CDMA核心机制——如扩频调制、多用户干扰建模、Rake接收、多径衰落信…

作者头像 李华
网站建设 2026/9/13 19:22:20

EC2302电容触摸芯片PCB设计要点与灵敏度调试实战指南

EC2302是一颗非常典型的单通道电容触摸感应芯片,常用于小家电、智能面板、灯具、玩具这类对成本敏感又需要可靠触摸响应的产品里。它的调试难点往往不在芯片本身,而在PCB设计——同样的固件和寄存器配置,板子画得不好,触摸灵敏度飘…

作者头像 李华
网站建设 2026/9/13 19:21:13

Boost.Asio+Qt4+Python2在电力SCADA嵌入式系统中的工程实践

简介:这是一套面向工业自动化与电力监控领域的SCADA系统通信管理机完整源码,适用于嵌入式Linux通信管理机及大型服务器部署场景,主要解决多协议数据采集、跨平台通信调度与Web端可视化集成等核心问题。资源共768个文件,涵盖395个h…

作者头像 李华