Ejecutor

El entorno de ejecución de adk-runner que orquesta la ejecución de agentes.

Descripción general

Runner gestiona el ciclo de vida completo de la ejecución de agentes:

• Gestión de sesiones (crear y recuperar sesiones)
• Inyección de memoria (buscar e inyectar memorias relevantes)
• Gestión de artefactos (acceso a artefactos con alcance limitado)
• Transmisión de eventos (procesar y reenviar eventos)
• Transferencias de agentes (gestionar traspasos entre varios agentes)

Rendering architecture…

Instalación

[dependencies]
adk-runner = "2.1.0"

RunnerConfig

Configura el ejecutor con los servicios necesarios:

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 (Recomendado)

Usa el constructor de typestate para construir un Runner. El constructor impone los campos obligatorios en tiempo de compilación y establece valores predeterminados para todos los campos opcionales, por lo que añadir nuevos campos en futuras versiones no interrumpirá tu código:

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()?;

El constructor requiere tres campos: app_name, agent y session_service. Todo lo demás es opcional y tiene valores predeterminados adecuados. El método build() solo está disponible una vez establecidos los tres campos obligatorios; omitir uno de ellos genera un error en tiempo de compilación, no en tiempo de ejecución.

Campos de configuración

Caché de prompts

El almacenamiento en caché es una cuestión del proveedor y no requiere configuración de Runner. Cada integración de proveedor lo gestiona cuando se ensambla la solicitud:

Los aciertos de caché son observables sin ninguna configuración adicional: la integración con Gemini registra cachedContentTokenCount en cada respuesta.

context_cache_config y cache_capable son experimentales y deben dejarse sin establecer. Controlan el cachedContents explícito API de Gemini desde el Runner. Ese API requiere que la caché reemplace system_instruction, tools y tool_config; enviar una caché junto con cualquiera de ellos se rechaza con INVALID_ARGUMENT. El Runner selecciona una caché antes de que el agente resuelva sus herramientas, por lo que no puede construir esa solicitud, y habilitar estos campos actualmente no produce aciertos de caché. La caché garantizada (en lugar de basada en el mejor esfuerzo) para Gemini pertenece a la integración del modelo, junto con la forma en que lo hacen los demás proveedores.

Ejecución de agentes

Ejecuta un agente con una entrada del usuario:

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

Método práctico para cadenas

Para sitios de llamada sencillos, run_str() acepta argumentos &str simples y gestiona internamente la conversión al newtype:

let mut stream = runner.run_str(
    "user-123",
    "session-456",
    Content::new("user").with_text("Hello!"),
).await?;

Si la cadena no supera la validación (está vacía, contiene bytes nulos o supera el límite de longitud), run_str() devuelve un error antes de iniciar el bucle del agente. El método existente run() con UserId/SessionId tipados permanece sin cambios.

Interrupción y aislamiento de ejecuciones

Una ejecución se registra en cuanto run() devuelve su flujo, y se elimina del registro cuando ese flujo se descarta, incluso si se descarta sin haber sido consultado nunca.

// Cancel one tenant's run without touching another that shares the session ID
let cancelled = runner.interrupt_identity("my-app", "user-1", "session-1");

Un ID de sesión solo es único dentro de una aplicación y un usuario, por lo que interrupt(session_id) es la forma amplia y interrupt_identity la precisa. Prefiere interrupt_identity cuando un solo Runner sirve a más de una aplicación o usuario.

Las ejecuciones se rastrean mediante un ID de ejecución único en lugar de mediante el ID de sesión, por lo que dos ejecuciones para la misma identidad se rastrean por separado y cada una solo se da de baja a sí misma.

La persistencia está vinculada a la identidad

Cada evento que Runner persiste —turnos del usuario, respuestas del modelo, eventos de transferencia, eventos de plugins y eventos de compactación— se escribe mediante SessionService::append_event_for_identity con la triple (app_name, user_id, session_id) completa. Por lo tanto, un SessionService cuya clave natural es compuesta puede vincular cada evento a su tenant, y puede rechazar o ignorar por completo la ruta de append_event del ID de sesión sin procesar.

Flujo de ejecución

┌─────────────────────────────────────────────────────────────┐
│                     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

El contexto proporcionado a los agentes durante la ejecución:

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

Opciones de ejecución:

pub struct RunConfig {
    /// Streaming mode for responses
    pub streaming_mode: StreamingMode,
    // ... other fields (tool_confirmation_decisions, cached_content, etc.)
}

ToolExecutionStrategy

Controla cómo se distribuyen varias llamadas a herramientas de una única respuesta de LLM:

Se establece por agente mediante 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()?;

En el modo Auto, el ciclo de despacho consulta tanto is_read_only() como is_concurrency_safe(). Las llamadas cuyas herramientas seleccionadas devuelven true para ambos métodos se ejecutan primero de forma concurrente; todas las llamadas restantes se ejecutan secuencialmente. Parallel omite estas comprobaciones de metadatos como una anulación explícita por parte del llamador. Los resultados siempre se vuelven a ensamblar en el orden original devuelto por LLM, independientemente de la estrategia. Las herramientas fallidas producen una respuesta de error JSON sin abortar el lote.

pub enum StreamingMode {
    /// No streaming, return complete response
    None,
    /// Server-Sent Events (default)
    SSE,
    /// Bidirectional streaming (realtime)
    Bidi,
}

Transferencias entre agentes

Runner gestiona automáticamente las transferencias entre varios agentes:

// 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 hará lo siguiente:

1. Detectar la solicitud de transferencia en el evento
2. Encontrar el agente de destino en sub_agents
3. Actualizar el estado de la sesión con el nuevo agente activo
4. Continuar la ejecución con el nuevo agente

Compactación del contexto

Para sesiones de larga duración, habilite la compactación automática del contexto para mantener acotada la ventana de contexto de 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),
    }),
    // ...
};

Cuando se activa la compactación, los eventos más antiguos se reemplazan por un evento de resumen. conversation_history() utiliza automáticamente el resumen en lugar de los eventos originales.

Consulte Compactación del contexto para obtener la documentación completa.

Integración con Launcher

Launcher utiliza Runner internamente:

// 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()?;

Uso personalizado de Runner

Para escenarios avanzados, use Runner directamente:

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

---

Anterior: ← Tipos principales | Siguiente: Launcher →