实时智能体的图工作流¶

Supported in ADKPython v2.0.0

实时智能体与其他 ADK 智能体一样，可以组合成相同的图工作流。节点和边的定义、路由以及状态相关内容请参阅图工作流，更广泛的多智能体架构请参阅工作流。在实时连接下发生变化的是执行模型。

在 run_live() 下，整个智能体管道运行在同一个开放连接和同一个事件循环中，因此调用方听到的是一个连续的对话。当控制权从一个智能体转移到下一个时，用户继续说话，且不会感知到切换。

这也影响了你的代码编写方式。对于请求/响应模式的智能体，每次智能体转换都是一次你可控的新调用；而在实时模式下，无论工作流跨越多少个智能体，整个工作流只有一个循环和一个队列。

在图中运行智能体¶

图 Workflow 是在 ADK 2.0 中编排实时智能体的方式。你将智能体定义为节点并用边连接它们，运行器会在单个实时会话中遍历图：

from google.adk.agents.llm_agent import Agent
from google.adk.workflow import START, Workflow

LIVE_MODEL = 'gemini-live-2.5-flash-native-audio'

greeter = Agent(
    model=LIVE_MODEL,
    name='greeter',
    mode='task',  # 节点使用实时连接时必须设置
    instruction='Greet the caller and confirm you are speaking with John Doe. '
    'Ask one question per turn. Complete your task once the name is confirmed.',
)

verifier = Agent(
    model=LIVE_MODEL,
    name='verifier',
    mode='task',
    instruction='Verify the caller by date of birth, then complete your task.',
)

root_agent = Workflow(
    name='intake',
    edges=[
        (START, greeter),
        (greeter, verifier),
    ],
)

使用 adk web 提供服务并启动实时会话，或将其传递给 Runner.run_live()。运行器检测到 Workflow 根节点后会通过实时连接驱动它；你会消费跨所有节点的单个事件流。可运行的示例请参阅 live_workflow 示例，其中包含一个带类型化切换和实时评估集的三阶段语音接待流程。

每个需要发言的智能体都必须设置 mode='task' 或 mode='chat'。 作为工作流中的节点，没有设置 mode 的 LlmAgent 会回退到 single_turn 模式，该模式在实时连接之外运行并完全忽略音频队列，因此调用方听不到它的任何输出。请在每个需要发言的节点上显式设置模式。

每个节点在该节点持续期间会打开自己的 Live API 会话，工作流的 LiveRequestQueue 在各节点间按顺序共享。单个队列无法同时向两个实时节点供数据，因此请将实时节点保持在一条路径上，而不是分叉。

读取单个事件流¶

事件流在节点切换时是连续的。使用一个循环和一个队列来消费它，并通过 event.author 判断是哪个智能体在发言。

queue = LiveRequestQueue()

async for event in runner.run_live(
    user_id='user_123',
    session_id='session_456',
    live_request_queue=queue,
):
    if event.content and event.content.parts:
        for part in event.content.parts:
            if part.inline_data and part.inline_data.mime_type.startswith('audio/'):
                await play_audio(part.inline_data.data)
            elif part.text:
                await display_text(f'[{event.author}] {part.text}')

不要为每个智能体打开新的 run_live() 循环或新的 LiveRequestQueue。一个循环和一个队列服务于整个工作流；用户输入会流向当前活跃的节点。

对话中途切换¶

协调者智能体可以在会话中途通过 transfer_to_agent 将对话传递给专业智能体。切换发生在同一个 run_live() 循环内：ADK 关闭协调者的实时连接，为专业智能体打开一个新连接，用户继续说话。

User: "I need help with billing"
Event: author="coordinator", function_call: transfer_to_agent(agent_name="billing")
Event: author="billing", text="I can help with your billing question..."

切换会为目标智能体启动新的 Live API 会话，因此协调者的会话恢复句柄不会传递过去。要将切换限制在协调者自己的团队内，请在子智能体上设置 disallow_transfer_to_peers；不允许的同级切换会抛出 ValueError。

旧版工作流智能体¶

新代码请使用图 Workflow。SequentialAgent、LoopAgent 和 ParallelAgent 已被弃用，推荐使用 Workflow，将在未来版本中移除。LoopAgent 和 ParallelAgent 在 run_live() 下会抛出 NotImplementedError 并导致实时会话崩溃，因此请确保它们不在任何实时路径上。

SequentialAgent 仍然可以在实时模式下运行。当它运行时，ADK 会为每个直接 LlmAgent 子智能体添加一个 task_completed 工具，并附加一条指令告诉模型在任务完成时调用它。调用 task_completed 会结束该子智能体的实时连接并推进到序列中的下一个智能体。

# ADK 在实时运行时将此注入到每个 LlmAgent 子智能体中。
def task_completed():
    """Signals that the agent has completed the user's task."""
    return 'Task completion signaled.'

事件流看起来像任何实时工作流：每个智能体运行一组事件，然后是一个 task_completed 函数响应，接着下一个智能体开始：

Event: author="researcher", function_call: task_completed()
Event: author="writer", text="Based on the research..."

task_completed 和 transfer_to_agent 因不同原因结束智能体的回合：

函数	模式	效果
`task_completed`	固定序列	结束当前智能体；序列中的下一个智能体开始
`transfer_to_agent`	动态路由	关闭当前实时会话；为目标智能体打开新会话