Separate voice stages when they have different jobs: one collects the question, and another answers from documentation. An ADK Workflow passes a typed result between task-mode agents. Runner.run_live() carries their audio and transcript events through one application event stream, while each stage uses its own model session.
The saved connections on gemini-3.1-flash-live-preview demonstrate that handoff with a typed question. They show the recorded stages and their completion, rather than a comparison proving that two agents outperform one.
When you need this. A docs call can need a clarification before an answer. Keep intake separate when its questions, completion rule or voice differ from the answering stage. Use one agent when those distinctions do not help; a graph adds another model session and another place the conversation can stall.
gemini-3.1-flash-live-preview.Download the complete lab ZIP, extract it and open a terminal in adk-live-workflow/. Create the environment below and set GOOGLE_API_KEY in that terminal before running a live command.
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements-lock.txt1. Define the typed intake and answer nodes
DocsQuestion carries the page and the caller’s question from intake to answer. The live-specific choice here is mode="task" on each speaking agent.
"""A live graph workflow: an intake agent finds the page and question, then an answer agent replies."""
import os
from pathlib import Path
from google.adk.agents import Agent
from google.adk.workflow import START, Workflow
from pydantic import BaseModel, Field
DOCS = Path(__file__).parent / "docs"
PAGES = sorted(p.name for p in DOCS.glob("*.md"))
# Every runnable Live lab uses the same model; the saved workflow receipt keeps its original model id.
WORKFLOW_MODEL = os.getenv("LIVE_MODEL", "gemini-3.8-live")
class DocsQuestion(BaseModel):
"""The typed handoff from intake to answer."""
page: str = Field(description=f"One of: {', '.join(PAGES)}")
question: str = Field(description="The caller's question, in their words")
def build_workflow(model=WORKFLOW_MODEL, answer_mode="task") -> Workflow:
intake = Agent(name="intake", model=model, mode="task", output_schema=DocsQuestion, instruction=(
f"You take calls for the docs assistant. Find out which page the caller means ({', '.join(PAGES)}) "
"and what they want to know. Ask one short question per turn. Finish your task as soon as you have both."))
pages = "\n\n".join((DOCS / name).read_text() for name in PAGES)
answer = Agent(name="answer", model=model, mode=answer_mode, instruction=(
"Answer the caller's question in one or two spoken sentences, only from these pages, "
"then finish your task.\n\n" + pages))
# Every node that speaks needs mode="task" or "chat"; single_turn runs outside the live connection.
return Workflow(name="docs_call", edges=[(START, intake), (intake, answer)])
def transcript_by_node(rows) -> dict:
"""What each node said. A node that ends with finish_task can stop before its finished transcript
arrives, so fall back to its partial fragments."""
finished, fragments = {}, {}
for row in rows:
if "output_transcription" in row:
target = fragments if row.get("partial") else finished
target.setdefault(row["author"], []).append(row["output_transcription"]["text"])
return {author: " ".join(t.strip() for t in finished[author]) if author in finished
else "".join(fragments[author]).strip() + " [partial]" for author in {**fragments, **finished}}The intake instruction asks one question per turn and finishes once it has both fields. ADK supplies the finish_task tool for a task node. Its structured arguments become the next node’s input, rather than asking the answer agent to extract the request from a long conversation. The page field is described with the known filenames; its Pydantic type is still a string, so the description alone does not enforce membership. No path supplied by the model is opened here: the answer node receives the text of all bundled documentation pages in its instruction.
The answer node also has task mode and a finish instruction. Without a completion rule it can keep chatting instead of advancing or ending the workflow.
A speaking node defaults to single_turn. The answer node left the live path and tried a normal generation request, which a Live-only model rejects. Set mode="task" or mode="chat" on every speaking node. Check: test_a_single_turn_node_skips_the_live_connection.
Legacy LoopAgent under run_live. Running a LoopAgent under run_live() raises NotImplementedError. Use a Workflow graph for multi-node live sequences. Check: test_loop_agent_is_not_supported_under_run_live.
2. Send caller turns through one queue
python record.pyThe recording sends the first caller message, waits for the intake turn, then sends the page and question. The harness keeps one LiveRequestQueue for the workflow and reads a single event stream. event.author identifies whether intake or answer produced an audio or transcript event.
[
"Hi, I have a question about the docs.",
"The sessions page. What happens if I reuse a closed queue?"
]The queue is shared sequentially. Each speaking node opens its own Live API session for its part of the call. This is not a parallel microphone fan-out: two live nodes cannot independently consume the same queue.
3. Inspect the handoff and the unfinished transcript
[
{
"page": "sessions.md",
"question": "What happens if I reuse a closed queue?"
},
{
"result": "Answered user question about reusing a closed queue."
}
]The test loads the answer connection’s initial history as DocsQuestion and compares it with these arguments. That checks the actual input to the next node, rather than inferring a successful handoff from the final audio.
The answer node completed its task before ADK emitted its merged finished transcription. transcript_by_node therefore marks its assembled fragments as [partial]. Keep that distinction in a UI or a saved transcript: partial text is useful for display, but treating it as the complete spoken turn loses information.
finish_task arrives before finished transcription. The answer’s transcript contains only partial fragments, although audio was emitted. Preserve its partial status instead of labelling the assembled text complete. Check: test_finish_task_can_end_a_node_before_its_finished_transcript.
4. Replay the workflow offline
Inspect the saved session output
[two live nodes] {"intake": "Hello! I can help with that. Which page in the docs are you referring to?", "answer": "If you reuse a closed `LiveRequestQueue`, the next `run_live()` call will end immediately. You should create a fresh queue for every new `run_live()` call. Can I help with anything else? [partial]"}
[answer node left as single_turn] {} error=ClientError: 400 INVALID_ARGUMENT. {'error': {'code': 400, 'message': 'models/gemini-3.1-flash-live-preview only supports real-time bidirectional streaming via WebSocket (bidiGenerateContent). Please use the Gemini Live API (bidiGenerateContent via WebS
[LoopAgent under run_live] {} error=NotImplementedError: This is not supported yet for LoopAgent.python -m unittest -v test_voice_intake.pytest_a_single_turn_node_skips_the_live_connection (test_voice_intake.VoiceIntake.test_a_single_turn_node_skips_the_live_connection) ... ok
test_finish_task_can_end_a_node_before_its_finished_transcript (test_voice_intake.VoiceIntake.test_finish_task_can_end_a_node_before_its_finished_transcript) ... ok
test_loop_agent_is_not_supported_under_run_live (test_voice_intake.VoiceIntake.test_loop_agent_is_not_supported_under_run_live) ... ok
test_one_stream_carries_both_nodes_in_order (test_voice_intake.VoiceIntake.test_one_stream_carries_both_nodes_in_order) ... ok
test_the_typed_handoff_becomes_the_next_nodes_history (test_voice_intake.VoiceIntake.test_the_typed_handoff_becomes_the_next_nodes_history) ... ok
----------------------------------------------------------------------
Ran 5 tests in 0.231s
OKThe replay replaces the Live API connection with its recorded responses while ADK runs the workflow, emits the handoff and builds the next request. The tests verify node order, typed DocsQuestion handoff, partial transcription handling and the unsupported mode configurations.
How it works
When Runner.run_live() runs a Workflow, ADK steps through the graph nodes using the shared LiveRequestQueue. An Agent node configured with mode="task" opens a Live API session and receives an injected finish_task tool whose parameter schema matches the agent’s output_schema (DocsQuestion). While the intake node is active, it consumes caller turns from the queue and streams audio and transcript events with author="intake".
Once the intake model calls finish_task, ADK validates the arguments against DocsQuestion, closes the intake node’s Live API session, and advances the graph edge to the answer node. The answer node opens a new Live API session seeded with the serialized DocsQuestion payload and streams its spoken response on the same run_live() event generator until its own finish_task call ends the workflow.
What’s next
- Guard and evaluate a voice agent: check what callbacks can block after speech has already begun and inspect what a live evaluation scores.
Check your understanding
Read the explanation
The nodes consume one application queue sequentially; the answer node opens its own model session seeded with the typed payload.
References
Official sources checked October 21, 2026, against the installed Google ADK package.