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的校验逻辑接受BaseAgent或BaseNode两种类型)。
Runner在内部会把传入的参数归一化为App(Runner._resolve_app),所以直接传一个裸 Agent 也能跑;但只有App能携带这些跨切面配置。官方推荐路径是Runner(app=...),这一点在本文"App 与裸 Agent 的区别"一节详述。
快速开始:把 Agent 包装进 App
定义一个带工具调用的 Agent,再用App把它包起来。下面这个例子构建了一个带get_weather工具的天气 Agent,并将其放入名为weather_app的App中,同时挂载了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 run、adk web、adk 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路径有三点差异:
- 跳过校验:包装过程会绕过
App的字段校验(源码中用的是App.model_construct,见 src/google/adk/runners.py),因此一个App会拒绝的应用名,在这里反而会被接受; - 跨切面配置无法设置:
context_cache_config、events_compaction_config、resumability_config都会被留空。Runner没有对应的参数,App是设置它们的唯一途径; plugins参数已弃用:Runner(plugins=[...])会触发DeprecationWarning;如果同时传入app和plugins,会直接抛出ValueError——请把插件放到App上。相关校验逻辑在Runner._resolve_app(src/google/adk/runners.py)。
另外,当app与app_name同时给出时,会话查找以app_name为准,而app.name保持不变。同时传两者通常不是你想要的。Runner构造时还要求app、agent、node三者恰好提供一个,否则抛出ValueError(src/google/adk/runners.py)。
App 字段详解
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
name | str | 必填 | 应用名称,会话按它索引。 |
root_agent | BaseAgent或BaseNode | 必填 | 执行入口。BaseNode是google.adk.workflow中的工作流节点基类,因此Workflow也可以作为根。 |
plugins | list[BasePlugin] | [] | 应用级插件,其回调对每一次 Agent、模型调用与工具调用都会触发。 |
context_cache_config | ContextCacheConfig \| None | None | 为应用内所有 LLM Agent 启用上下文缓存,缺省表示缓存关闭。 |
events_compaction_config | EventsCompactionConfig \| None | None | 对较早的会话事件做摘要,防止上下文无限增长。 |
resumability_config | ResumabilityConfig \| None | None | 允许调用在长时间运行的函数调用处暂停,并在之后恢复。 |
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只导出App和ResumabilityConfig(见 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在初始化时自动装配InMemoryArtifactService、InMemorySessionService与InMemoryMemoryService(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,取值范围1–100(ge=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_size。compaction_interval表示"新用户发起的调用"数量达到多少时触发压缩(必须> 0);overlap_size表示从前一次压缩范围末尾向前包含多少个调用,使相邻摘要之间有重叠以保持上下文(>= 0); - token 预算对:
token_threshold+event_retention_size。token_threshold为调用后压缩的 token 阈值(必须> 0);event_retention_size表示触发压缩时保留多少个原始事件不被压缩(>= 0)。
校验器(_validate_trigger_params)强制要求:token_threshold与event_retention_size必须同时设置;compaction_interval与overlap_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)**的,详见下一节限制说明。
已知限制与注意事项
- 实验性配置:
EventsCompactionConfig、ResumabilityConfig、ContextCacheConfig三者都在构造时发出实验性警告(源码中均标注了@experimental装饰器),并可能在无通知的情况下变更; EventsCompactionConfig未被 re-export:google.adk.apps只导出App和ResumabilityConfig,请从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),仅供参考