Extraction Pipeline

Extract typed values from conversation evidence and use them as inputs to state, phase, and tool decisions. Choose when extraction runs and how an error or incomplete result affects the workflow.

A value matching the schema can still be factually wrong. Test the evidence needed for a transition as well as the JSON shape.

What Is Extraction?

Extraction turns unstructured conversation into structured data. As the user and model talk, extractors analyze the transcript and produce typed JSON: customer name, order items, emotional state, account numbers. These values flow into session State where they drive phase transitions, trigger watchers, and inform instruction composition.

Why Out-of-Band?

Extraction runs on the control lane, not on the conversation path. An LLM extraction call takes 1-5 seconds. Voice conversations cannot pause for that. Extractors run concurrently after each turn completes while the conversation continues uninterrupted.

Turn completes
  -> Transcript buffer finalizes the turn
  -> Extractors run concurrently (control lane)
  -> Results written to State under derived: prefix
  -> Watchers evaluate -> Phase transitions fire
  -> Conversation continues (no blocking)

Deterministic extraction (no model)

Not every field needs an LLM. Slots like quantities, money, yes/no, a name matched against a roster, or a date/time are recognizable on the CPU — no model, no network, no accelerator. The Extract kit declares a record of typed fields, each filled by a Recognizer, and compiles to a TurnExtractor that promotes recognized fields straight into State.

use gemini_adk_rs::Extract;   // the #[derive(Extract)] macro

#[derive(Extract)]
#[extract(name = "order", window = 3)]
struct Order {
    #[recognize(integer_near = ["want", "get"])]
    quantity: Option<i64>,
    #[recognize(one_of = ["pizza", "salad", "soda"])]
    item: Option<String>,
    #[recognize(fuzzy = ["Johnson", "Jackson"])]   // ASR-robust name match
    name: Option<String>,
    #[recognize(datetime)]                          // → { "time": "18:00", "day": "tomorrow" }
    #[extract(state = "when")]
    pickup: Option<serde_json::Value>,
    #[recognize(yes_no)]
    confirmed: Option<bool>,
}

Live::builder()
    .extract_record(Order::extract())   // deterministic, runs on the control lane
    .govern(order_flow)                 // Flow reads done(captured(["quantity", "item"]))
    .connect_from_env().await?;

Recognizer forms: integer/integer_near, money, regex, one_of, fuzzy (Jaro-Winkler, ASR-robust), yes_no, and datetime (a small on-device clock/calendar normalizer: 12h/24h time → HH:MM, relative days, weekdays, parts of day, ISO dates). You can also build a record fluently with Extract::record(name).field(..) instead of the derive.

Async field sources

A field can be filled by an async resolver instead of a recognizer — arguments bound from State, with an optional TTL cache:

let booking = Extract::record("booking")
    .field("slot", Recognizer::one_of(["morning", "afternoon"]))
    // args bound from State → fetcher → field; cached 30s by (field, args).
    .field_resolve("availability", ["slot"], Some(Duration::from_secs(30)), |args| async move {
        let slot = args.get("slot").and_then(|v| v.as_str()).unwrap_or("");
        Ok(serde_json::json!({ "open": slot == "afternoon" }))
    })
    // run a downstream agent when the record lands fields (result → booking:result):
    .on_complete(router_agent, AgentMode::Dispatch)
    .build();

(Closures can’t live in attributes, so resolver fields use the builder; #[derive(Extract)] covers the recognizer fields.)

Standalone `Resolver`

The async sibling of Recognizer is Resolver: a named value source whose inputs come from State and whose result lands under {name}:result (or {name}:error). It generalizes a sub-agent call to any async source — a tool call, an HTTP fetch, or an MCP request (see Orchestration):

use gemini_adk_rs::Resolver;

// From a sub-agent (its String output becomes the result):
Resolver::agent("availability", availability_agent).resolve(&state).await?;

// From any async system, with inputs bound from State:
Resolver::fetch("availability", |s: State| async move {
    let slot = s.get::<String>("slot").unwrap_or_default();
    Ok(serde_json::json!({ "open": slot == "afternoon" }))
}).resolve(&state).await?;   // or .dispatch(state) to run detached

Both deterministic Recognizer fields and async Resolver results live in State under the same conventions, so a Flow step completes on either — done(captured(["quantity"])) or done(resolved("availability")) — and a flow step’s on_enter can launch a resolver automatically.

TurnExtractor

The base trait for all extractors. Implement it for synchronous extraction (regex, keyword matching, heuristics):

use async_trait::async_trait;
use gemini_adk_rs::live::extractor::TurnExtractor;
use gemini_adk_rs::live::transcript::TranscriptTurn;
use gemini_adk_rs::llm::LlmError;

struct OrderNumberExtractor;

#[async_trait]
impl TurnExtractor for OrderNumberExtractor {
    fn name(&self) -> &str { "order_info" }  // State key for results
    fn window_size(&self) -> usize { 5 }     // Look at last 5 turns

    fn should_extract(&self, window: &[TranscriptTurn]) -> bool {
        // Skip trivial turns -- checked before async extraction
        window.last()
            .map(|t| t.user.split_whitespace().count() >= 3)
            .unwrap_or(false)
    }

    async fn extract(&self, window: &[TranscriptTurn]) -> Result<serde_json::Value, LlmError> {
        let text: String = window.iter()
            .map(|t| format!("{} {}", t.user, t.model))
            .collect::<Vec<_>>().join(" ");

        let re = regex::Regex::new(r"order\s+#?(\d+)").unwrap();
        let mut result = serde_json::Map::new();
        if let Some(caps) = re.captures(&text) {
            result.insert("order_number".into(), serde_json::json!(caps[1].to_string()));
        }
        Ok(serde_json::Value::Object(result))
    }
}

should_extract is checked before launching async work. Return false to skip the LLM round-trip entirely on trivial turns.

LlmExtractor

For extraction requiring understanding (sentiment, intent, entity recognition), LlmExtractor sends the transcript to an OOB LLM:

use gemini_adk_rs::live::extractor::LlmExtractor;

let extractor = LlmExtractor::new(
    "SentimentAnalysis",
    llm,  // Arc<dyn BaseLlm>
    "Analyze conversation sentiment and extract the customer's emotional state.",
    3,    // window size
)
.with_schema(serde_json::json!({
    "type": "object",
    "properties": {
        "sentiment": { "type": "string", "enum": ["positive", "neutral", "negative"] },
        "score": { "type": "number" }
    }
}))
.with_min_words(5);  // Skip "uh huh", "ok", "yes" turns

Each request carries the transcript window, the extractor’s instruction and its schema. The runtime also adds three things:

Every field may be null. Each top-level schema property also accepts null, and the instruction asks for null when the transcript does not state a field. Without this, the model fills fields nobody mentioned with placeholders such as "unknown", "" or 0.
Known values. When the extractor runs with access to state, the values already promoted for its fields are listed as Already known: {...}, and the caller’s latest turn is quoted after the transcript. The model returns a known field only when that turn gives a different value. A value equal to the known one apart from case, spacing or punctuation ("7 pm" and "7 PM") is not promoted again, so it is not taken for a correction.
No thinking. Requests send a thinking budget of 0. The turn pipeline waits for each extraction, so thinking costs seconds on every turn. On gemini-flash-latest a short extraction took 2.3 s at the median with a budget of 0, against 7.7 s at the model’s default, with the same results. .with_thinking_budget(None) restores the model’s default. When a model rejects the budget (some only work in thinking mode), the extractor sends the request again without it and stops sending it.

A transient failure (HTTP 5xx, 429 or a transport error) is retried once after 300 ms. A failed extraction loses what the caller said in that turn.

Choosing the extraction model

Name the model. In a spec, models.extraction sets it for every extract entry, and an entry’s own model overrides it; models.extraction_thinking_budget and an entry’s thinking_budget set the budget the same way:

"models": { "extraction": "gemini-3.5-flash-lite", "extraction_thinking_budget": 64 },
"extract": [
  { "name": "caller_identity", "instruction": "...", "schema": { ... } },
  { "name": "notes", "model": "gemini-3.5-flash", "thinking_budget": 0, "instruction": "...", "schema": { ... } }
]

A gemini-* name runs on GeminiLlm. SpecResources::models, an LlmRegistry, resolves any other name, such as a model served elsewhere, and takes precedence. An entry that names no model runs on SpecResources::extraction_llm; adk spec run creates it from GEMINI_TEXT_MODEL, else gemini-flash-latest. adk spec check warns (unpinned_extraction_model) when extraction names no model, and its fix pins gemini-3.5-flash-lite with a thinking budget of 64.

tests/extraction_latency.rs replays the pharmacy fixture’s three extractors over the turn “I need a refill of my Lisinopril”, all three at once as the turn pipeline runs them, six turns at a time (18 concurrent requests), through LlmExtractor, and checks the name, date of birth, medication and signals:

Model	Turn p50 / p90 / p99 / max	Correct
`gemini-3.5-flash-lite`, budget 64	0.67 / 0.85 / 1.16 / 1.18 s	60/60
`gemini-3.1-flash-lite`, budget 0	1.0 / 1.3 / 1.5 / 1.7 s	60/60
`gemini-3.5-flash`	1.1 / 1.9 / 3.3 / 5.7 s	60/60
`gemini-flash-latest` (`gemini-3.8-flash`)	6.7 / 19.5 / 49.1 / 62.2 s	60/60
`gemini-2.5-flash-lite`	0.4 / 0.7 / 0.9 / 1.0 s	0/60 (drops the date of birth)

gemini-flash-latest is a rolling alias: it served gemini-3.8-flash when this was measured, and its latency changes when the alias moves. With three turns at a time it took 3.3 s at the median and 9 s at most; under load its tail reaches the 49 and 58 s turns seen in live runs. Every call succeeded on the first try: the time is the model’s.

Run the probe against your own extractors before switching:

EXTRACTION_MODELS=gemini-3.5-flash-lite EXTRACTION_THINKING_BUDGET=64 \
  cargo test -p gemini-adk-fluent-rs --test extraction_latency -- --ignored --nocapture

`gemini-3.5-flash-lite`

Thinking. It rejects thinkingBudget: 0 (“Request contains an invalid argument”) and accepts budgets from 1. Below about 1024 it does not think at all (0 thought tokens at 1, 64, 128, 512, 768 and 1000); from 1024 it thinks about 550 tokens and takes about 2 s. thinkingLevel minimal and low do not think either; medium thinks about 520 tokens (2 s) and high about 1,400 (4.6 s). Send 64: it keeps thinking off at the smallest budget the model accepts. Without a configured budget the extractor sends 0, and on the rejection resends at 64, then with none, once per extractor.
Instructions are read literally without thinking. The pharmacy fixture once said “Leave it out until they have picked one.” The caller said “I need a refill of my lisinopril”, and the agent replied “I found two prescriptions: Lisinopril and Atorvastatin. Which one would you like to refill today?” Without thinking the model took the agent’s question as proof that nothing had been picked: lisinopril in 6 of 20 runs (0 of 20 at budgets 64 to 1000). Without the agent’s question, or with the agent acknowledging the pick, it was right 20 of 20; with medium thinking, 20 of 20, and its thought summary weighed exactly that question. Saying what counts fixed it at 64, 30 of 30: “A medication the caller named counts even if the assistant then asks which one; use null only if the caller has not named one.” gemini-3.1-flash-lite was right 20 of 20 with either wording, so it is the more forgiving choice when instructions are not tested.
Known values are re-stated without thinking. Told to return a known field “only if the user’s latest turn changes it”, it returned the known slot on “Yes, that’s all correct. Please book it.” in 17 of 20 runs, 7 of them in other words (“tomorrow at 7:00 PM” for “tomorrow at 7 pm”), which read as a correction and sent a live booking back to the availability stage. gemini-3.1-flash-lite and gemini-flash-latest did it in none of 20. Quoting the latest turn after the transcript, as the extractor now does, brought it to 0 of 30, and every model still caught a real change (“make that eight o’clock instead”) in all runs.
Picks read as agreement. One live run took “Seven is fine.” as agreeing to book. Offline it did not recur for gemini-3.5-flash-lite (0 of 40), but gemini-3.1-flash-lite marked “Seven o’clock is perfect” as a yes in 3 of 20, and gemini-flash-lite-latest did in earlier testing. Signals that gate a commit belong to a decision model or need explicit criteria.

Schema Definition

The fluent API’s extract_turns auto-generates the schema from a Rust struct:

use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

#[derive(Deserialize, Serialize, JsonSchema)]
struct DebtorState {
    /// "calm", "cooperative", "frustrated", "angry"
    emotional_state: Option<String>,
    /// 0.0 (refusing) to 1.0 (eager)
    willingness_to_pay: Option<f32>,
    /// "full_pay", "partial_pay", "dispute", "refuse", "delay"
    negotiation_intent: Option<String>,
    /// Whether debtor explicitly requested cease-and-desist
    cease_desist_requested: Option<bool>,
}

Live::builder()
    .extract_turns::<DebtorState>(
        llm,
        "Extract: emotional state, willingness to pay, negotiation intent, cease-and-desist.",
    )
    .connect(config).await?;

The type name (DebtorState) becomes the extractor name and State key. extract_turns auto-enables transcription, generates the JSON schema, and defaults to a 3-turn window. Use extract_turns_windowed for custom sizes.

Extraction Triggers

By default, extractors run on every TurnComplete event. For many use cases this is wasteful – trivial utterances (“yeah”, “ok”) rarely contain extractable data, and each extraction is an OOB LLM call. Extraction triggers control when extractors fire:

use gemini_adk_rs::live::extractor::ExtractionTrigger;

Live::builder()
    // Extract every 2 turns instead of every turn (reduces LLM costs by ~50%)
    .extract_turns_triggered::<DebtorState>(
        llm,
        "Extract debtor emotional state and negotiation intent",
        5,  // transcript window size
        ExtractionTrigger::Interval(2),
    )
    .connect(config).await?;

Trigger	When it fires	Use case
`EveryTurn`	After every TurnComplete	Default — high-frequency extraction
`Interval(n)`	Every N turns	Reduce LLM costs for slow-changing data
`AfterToolCall`	After tool dispatch completes	Extract from tool results
`OnPhaseChange`	When phase transitions fire	Re-extract on context shift

An EveryTurn LlmExtractor skips a turn the caller said nothing in, such as the model speaking a tool’s result. Earlier turns were already extracted, and the turn pipeline would otherwise wait on a call that cannot find anything new. When no turn in the window has caller words (input transcription is off), it extracts as usual.

The TurnExtractor trait also has a trigger() method with a default implementation returning EveryTurn, so custom extractors get the old behavior for free:

impl TurnExtractor for MyExtractor {
    fn trigger(&self) -> ExtractionTrigger {
        ExtractionTrigger::AfterToolCall
    }
    // ...
}

Transcript Window

Extractors receive a slice of TranscriptTurn values:

pub struct TranscriptTurn {
    pub turn_number: u32,
    pub user: String,                    // Accumulated user speech
    pub model: String,                   // Accumulated model speech
    pub tool_calls: Vec<ToolCallSummary>,
    pub timestamp: Instant,
}

The TranscriptBuffer is a ring buffer (default 50 turns) that evicts the oldest turns to prevent unbounded memory growth:

let mut buf = TranscriptBuffer::new();
let recent = buf.window(3);          // last 3 completed turns
let formatted = buf.format_window(3); // human-readable text
let snapshot = buf.snapshot_window(5); // cheap read-only clone for callbacks

Auto-Flatten

When an extractor returns a JSON object, the framework automatically flattens it to individual state keys under the derived: prefix. Given this result:

{ "emotional_state": "frustrated", "willingness_to_pay": 0.3 }

The framework writes:

derived:emotional_state = "frustrated"
derived:willingness_to_pay = 0.3

The prefix is transparent – state.get("emotional_state") auto-checks derived:emotional_state if the unprefixed key is not found:

.transition("close", S::is_true("cease_desist_requested"))
// Internally checks derived:cease_desist_requested

Concurrent Extraction

Multiple extractors run in parallel via futures::future::join_all:

Live::builder()
    .extractor(Arc::new(regex_extractor))      // instant
    .extract_turns::<DebtorState>(llm, "...")   // 1-3 seconds
    .on_extracted(|name, value| async move {
        println!("Extractor '{name}' produced: {value}");
    })
    .on_extraction_error(|name, error| async move {
        eprintln!("Extractor '{name}' failed: {error}");
    })
    .connect(config).await?;

Extraction and tool calls

Turn-end extraction runs on the control lane, and so do tool calls. A tool call that arrives while an extraction is running waits for it, so the extraction model’s latency adds directly to the tool’s response time. In a live run on gemini-flash-latest, two pharmacy tool calls waited 49 and 58 s behind the previous turn’s extraction. Pinned to gemini-3.5-flash-lite at a thinking budget of 64, the slowest tool wait across the same 44 runs was 1.4 s (see Choosing the extraction model).

A commit guard often reads a key an extractor writes, such as a caller’s confirmation. The caller says “yes” and the model calls the commit tool in the same turn, before that turn has been extracted. When the governed flow refuses a tool only because such a guard has not held yet, the tool lane runs the EveryTurn extractors that may write the guard’s keys on the turn in progress, then decides again. An extractor may write a key when a promotion rule targets it, or, without rules, when its schema has a field of that name. Extractors with another trigger keep their own moment. Each extractor reads its usual number of turns, ending with the turn in progress, so an older “yes” outside that window is not read again.

A confirmation, an intent or a pick among offered options can instead go to a decision model, which answers typed questions with probabilities instead of text.

Extraction to State to Watchers

The full data flow after each turn:

Extractors run concurrently on the control lane.
Results auto-flatten – each JSON field becomes a derived: key.
Computed state evaluates – derived variables that depend on extracted keys re-compute.
Watchers fire – any watcher observing a changed key triggers.
Phase transitions evaluate – guards check, machine transitions.

Live::builder()
    .extract_turns::<DebtorState>(llm, "Extract emotional state")
    .computed("call_risk_level", &["derived:sentiment_score"], |state| {
        let score: f64 = state.get("derived:sentiment_score").unwrap_or(0.5);
        if score < 0.3 { Some(json!("high")) } else { Some(json!("low")) }
    })
    .watch("derived:call_risk_level")
        .changed_to(json!("high"))
        .then(|_old, _new, state| async move {
            state.set("alert:risk_escalation", true);
        })
    .phase("negotiate")
        .instruction("Negotiate payment")
        .transition("close", S::is_true("cease_desist_requested"))
        .done()
    .connect(config).await?;

Real Example

The debt collection demo combines regex and LLM extractors:

// Regex: captures dollar amounts, phone numbers, disclosure acknowledgment
let regex_extractor = Arc::new(RegexExtractor::new("debt_fields", 10, |text, existing| {
    let mut extracted = HashMap::new();
    if !existing.contains_key("dollar_amount") {
        if let Some(m) = DOLLAR_RE.find(text) {
            extracted.insert("dollar_amount".into(), json!(m.as_str()));
        }
    }
    if !existing.contains_key("disclosure_given") {
        if DISCLOSURE_ACK_RE.is_match(text) {
            extracted.insert("disclosure_given".into(), json!(true));
        }
    }
    extracted
}));

let handle = Live::builder()
    .extractor(regex_extractor)
    .extract_turns::<DebtorState>(llm, "Extract debtor emotional state and intent")
    .computed("sentiment_score", &["emotional_state"], |state| {
        let emotion: String = state.get("emotional_state")?;
        Some(json!(match emotion.as_str() {
            "cooperative" => 0.9, "calm" => 0.7,
            "frustrated" => 0.4, "angry" => 0.2, _ => 0.5,
        }))
    })
    .computed("call_risk_level", &["derived:sentiment_score", "cease_desist_requested"], |state| {
        let sentiment: f64 = state.get("derived:sentiment_score").unwrap_or(0.5);
        let cease: bool = state.get("cease_desist_requested").unwrap_or(false);
        Some(json!(if cease { "critical" } else if sentiment < 0.3 { "high" } else { "low" }))
    })
    .phase("disclosure")
        .instruction("Deliver the Mini-Miranda disclosure")
        .transition("verify_identity", S::is_true("disclosure_given"))
        .transition("close", S::is_true("cease_desist_requested"))
        .done()
    .initial_phase("disclosure")
    .connect(config).await?;

Extracted fields flow through the full pipeline: extraction produces raw values, computed state derives higher-level signals, guards evaluate on every turn, and the phase machine transitions when conditions are met.