运行器
来自 adk-runner 的执行运行时,用于编排 agent 执行。
概述
Runner 管理 agent 执行的完整生命周期:
- 会话管理(创建/检索会话)
- 内存注入(搜索并注入相关记忆)
- 工件处理(作用域限定的工件访问)
- 事件流处理(处理并转发事件)
- agent 转移(处理多 agent 交接)
安装
[dependencies]
adk-runner = "2.1.0"
RunnerConfig
使用必需的服务配置运行器:
use adk_runner::{Runner, RunnerConfig};
use adk_session::InMemorySessionService;
use adk_artifact::InMemoryArtifactService;
use std::sync::Arc;
let config = RunnerConfig {
app_name: "my_app".to_string(),
agent: Arc::new(my_agent),
session_service: Arc::new(InMemorySessionService::new()),
artifact_service: Some(Arc::new(InMemoryArtifactService::new())),
memory_service: None,
plugin_manager: None,
run_config: None,
compaction_config: None,
context_cache_config: None,
cache_capable: None,
request_context: None,
cancellation_token: None,
};
let runner = Runner::new(config)?;
RunnerConfigBuilder(推荐)
使用 typestate 构建器构造 Runner。该构建器在编译时强制要求必需字段,并为所有可选字段提供默认值,因此未来版本添加新字段时不会破坏你的代码:
use adk_runner::Runner;
let runner = Runner::builder()
.app_name("my_app")
.agent(Arc::new(my_agent))
.session_service(Arc::new(InMemorySessionService::new()))
// Optional fields — only set what you need
.artifact_service(Arc::new(InMemoryArtifactService::new()))
.build()?;
该构建器要求三个字段:app_name、agent 和 session_service。其他所有字段都是可选的,并且具有合理的默认值。只有在设置完全部三个必需字段后,build() 方法才可用——缺少其中任何一个字段都会导致编译时错误,而不是运行时错误。
配置字段
| 字段 | 类型 | 必填 | 描述 |
|---|---|---|---|
app_name | String | 是 | 应用标识符 |
agent | Arc<dyn Agent> | 是 | 要执行的根代理 |
session_service | Arc<dyn SessionService> | 是 | 会话存储后端 |
artifact_service | Option<Arc<dyn ArtifactService>> | 否 | 工件存储 |
memory_service | Option<Arc<dyn Memory>> | 否 | 长期记忆 |
plugin_manager | Option<Arc<PluginManager>> | 否 | 插件生命周期钩子 |
compaction_config | Option<EventsCompactionConfig> | 否 | 上下文压缩设置 |
run_config | Option<RunConfig> | 否 | 执行选项 |
context_cache_config | Option<ContextCacheConfig> | 否 | 运行器级上下文缓存生命周期(实验性功能——见下文) |
cache_capable | Option<Arc<dyn CacheCapable>> | 否 | 支持缓存的模型引用 |
request_context | Option<RequestContext> | 否 | 身份验证中间件上下文 |
cancellation_token | Option<CancellationToken> | 否 | 协作式取消 |
提示词缓存
缓存属于提供商级别的关注点,无需配置 Runner。 每个 提供商集成都会在组装请求时处理缓存:
| 提供商 | 机制 | 默认值 |
|---|---|---|
| Anthropic / Bedrock | cache_control 断点 | 开启(AnthropicConfig::prompt_caching,使用 with_prompt_caching(false) 选择退出) |
| OpenAI | 服务器端提示缓存,PromptCacheRetention 用于保留 | 自动 |
| Gemini | 2.5/3.x 上的隐式缓存 — 共享前缀可享受折扣,无需更改代码 | 自动 |
无需额外配置即可观测缓存命中:Gemini 集成会在每个响应中记录 cachedContentTokenCount。
context_cache_config和cache_capable为实验性功能,应保持未设置状态。它们驱动 Runner 中 Gemini 的显式cachedContentsAPI。该 API 要求缓存替换system_instruction、tools和tool_config—— 将缓存与其中任何一项一同发送都会因INVALID_ARGUMENT而被拒绝。Runner 会在 agent 解析其工具之前选择缓存,因此无法组装此请求;目前启用这些字段也不会产生缓存命中。Gemini 的有保证缓存(而非尽力而为的缓存)应归属于模型集成,与其他提供商的处理方式一致。
运行 Agent
使用用户输入执行 agent:
use adk_core::{Content, SessionId, UserId};
use futures::StreamExt;
let user_content = Content::new("user").with_text("Hello!");
let mut stream = runner.run(
UserId::new("user-123")?,
SessionId::new("session-456")?,
user_content,
).await?;
while let Some(event) = stream.next().await {
match event {
Ok(e) => {
if let Some(content) = e.content() {
for part in &content.parts {
if let Some(text) = part.text() {
print!("{}", text);
}
}
}
}
Err(e) => eprintln!("Error: {}", e),
}
}
字符串便捷方法
对于简单的调用场景,run_str() 接受普通的 &str 参数,并在内部处理 newtype 转换:
let mut stream = runner.run_str(
"user-123",
"session-456",
Content::new("user").with_text("Hello!"),
).await?;
如果字符串未通过验证(为空、包含 null 字节或超过长度限制),run_str() 会在启动 agent 循环之前返回错误。现有的、使用类型化 UserId/SessionId 的 run() 方法保持不变。
中断与运行隔离
run() 返回其流后,运行即会被注册;当该流被丢弃时会注销——包括从未对其进行轮询就被丢弃的情况。
| 方法 | 作用域 |
|---|---|
interrupt(session_id) | 取消该会话 ID 的所有进行中运行,涵盖所有应用和用户 |
interrupt_identity(app_name, user_id, session_id) | 取消一个完全匹配身份的运行 |
active_runs() | 每个正在运行的任务的标识;重复的标识表示并发运行 |
active_session_ids() | 正在运行的任务的去重会话 ID |
// Cancel one tenant's run without touching another that shares the session ID
let cancelled = runner.interrupt_identity("my-app", "user-1", "session-1");
会话 ID 仅在应用和用户范围内唯一,因此 interrupt(session_id) 是
广义形式,而 interrupt_identity 是精确形式。当单个 Runner 服务于多个应用或用户时,
优先使用 interrupt_identity。
运行通过唯一的运行 ID 而非会话 ID 进行跟踪,因此同一身份的两个运行会被分别跟踪,并且每个运行只注销自身。
持久化与身份绑定
Runner 持久化的每个事件——用户轮次、模型响应、转移事件、
插件事件和压缩事件——都会通过
SessionService::append_event_for_identity 使用完整的
(app_name, user_id, session_id) 三元组写入。因此,自然键为
复合键的 SessionService 可以将每个事件绑定到其租户,并且可以完全拒绝或忽略
原始会话 ID append_event 路径。
执行流程
┌─────────────────────────────────────────────────────────────┐
│ Runner.run() │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 1. Session Retrieval │
│ │
│ SessionService.get(app_name, user_id, session_id) │
│ → Creates new session if not exists │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 2. Agent Selection │
│ │
│ Check session state for active agent │
│ → Use root agent or transferred agent │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 3. Context Creation │
│ │
│ InvocationContext with: │
│ - Session (mutable) │
│ - Artifacts (scoped to session) │
│ - Memory (if configured) │
│ - Run config │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 4. Agent Execution │
│ │
│ agent.run(ctx) → EventStream │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 5. Event Processing │
│ │
│ For each event: │
│ - Update session state │
│ - Handle transfers │
│ - Forward to caller │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 6. Session Save │
│ │
│ SessionService.append_event(session, events) │
└─────────────────────────────────────────────────────────────┘
InvocationContext
执行期间提供给代理的上下文:
pub trait InvocationContext: CallbackContext {
/// The agent being executed
fn agent(&self) -> Arc<dyn Agent>;
/// Memory service (if configured)
fn memory(&self) -> Option<Arc<dyn Memory>>;
/// Current session
fn session(&self) -> &dyn Session;
/// Execution configuration
fn run_config(&self) -> &RunConfig;
/// Signal end of invocation
fn end_invocation(&self);
/// Check if invocation has ended
fn ended(&self) -> bool;
}
RunConfig
执行选项:
pub struct RunConfig {
/// Streaming mode for responses
pub streaming_mode: StreamingMode,
// ... other fields (tool_confirmation_decisions, cached_content, etc.)
}
ToolExecutionStrategy
控制如何分派单个 LLM 响应中的多个工具调用:
| 策略 | 行为 |
|---|---|
Sequential(默认) | 按 LLM 返回的顺序逐个执行工具 |
Parallel | 并发执行所有工具;安全责任由调用方承担 |
Auto | 并发执行安全的只读子集,然后按顺序执行所有剩余调用 |
通过 LlmAgentBuilder 按代理设置:
use adk_core::ToolExecutionStrategy;
let agent = LlmAgentBuilder::new("fast_agent")
.model(model)
.tool_execution_strategy(ToolExecutionStrategy::Auto)
.tool(Arc::new(
search_tool
.with_read_only(true)
.with_concurrency_safe(true),
))
.tool(Arc::new(save_tool)) // runs after the concurrent safe subset
.build()?;
在 Auto 模式下,调度循环会同时查询 is_read_only() 和 is_concurrency_safe()。所选工具针对这两种方法均返回 true 的调用会优先并发运行;其余调用随后按顺序运行。Parallel 作为显式调用方覆盖项,会绕过这些元数据检查。无论采用何种策略,结果始终会按照 LLM 返回的原始顺序重新组装。工具失败时会生成 JSON 错误响应,但不会中止批处理。
pub enum StreamingMode {
/// No streaming, return complete response
None,
/// Server-Sent Events (default)
SSE,
/// Bidirectional streaming (realtime)
Bidi,
}
代理转移
Runner 会自动处理多代理转移:
// In an agent's tool or callback
if should_transfer {
// Set transfer in event actions
ctx.set_actions(EventActions {
transfer_to_agent: Some("specialist_agent".to_string()),
..Default::default()
});
}
Runner 将:
- 检测事件中的转移请求
- 在 sub_agents 中查找目标代理
- 使用新的活动代理更新会话状态
- 继续由新代理执行
上下文压缩
对于长时间运行的会话,请启用自动上下文压缩,以保持 LLM 上下文窗口有界:
use adk_runner::{Runner, RunnerConfig, EventsCompactionConfig};
use adk_agent::LlmEventSummarizer;
use std::sync::Arc;
let summarizer = LlmEventSummarizer::new(model.clone());
let config = RunnerConfig {
// ... other fields ...
compaction_config: Some(EventsCompactionConfig {
compaction_interval: 3, // Compact every 3 invocations
overlap_size: 1, // Keep 1 event overlap for continuity
summarizer: Arc::new(summarizer),
}),
// ...
};
触发压缩时,较早的事件会被摘要事件替代。conversation_history() 会自动使用该摘要,而不是原始事件。
完整文档请参阅上下文压缩。
与 Launcher 集成
Launcher 在内部使用 Runner:
// Launcher creates Runner with default services
Launcher::new(agent)
.app_name("my_app")
.run()
.await?;
// Equivalent to using the builder:
let runner = Runner::builder()
.app_name("my_app")
.agent(agent)
.session_service(Arc::new(InMemorySessionService::new()))
.build()?;
自定义 Runner 用法
对于高级场景,请直接使用 Runner:
use adk_runner::Runner;
// Production configuration using the builder
let runner = Runner::builder()
.app_name("production_app")
.agent(my_agent)
.session_service(Arc::new(SqliteSessionService::new(db_pool)))
.artifact_service(Arc::new(S3ArtifactService::new(s3_client)))
.memory_service(Arc::new(QdrantMemoryService::new(qdrant_client)))
.build()?;
// Use in HTTP handler with run_str() for convenience
async fn chat_handler(runner: &Runner, request: ChatRequest) -> Response {
let stream = runner.run_str(
&request.user_id,
&request.session_id,
request.content,
).await?;
// Stream events to client
Response::sse(stream)
}
上一页:← 核心类型 | 下一页:Launcher →