用于 ADK 的 LiveKit 运行器¶
ADK 提供了 LiveKitRunner 类，让你可以通过
LiveKit（一个用于 WebRTC 和 SIP 电话的开源平台）来运行实时智能体。此
集成充当传输适配器，负责处理音频和视频的捕获、播放、
打断、字幕和呼叫控制，因此你的 ADK 智能体无需任何改动即可从浏览器、
电话或游戏客户端访问。
使用场景¶
你可以在多种场景中使用 LiveKit，包括浏览器应用、移动应用、SIP 电话呼叫、游戏和沉浸式客户端。
浏览器和移动应用¶
智能体作为普通参与者加入房间，因此任何 LiveKit 客户端 SDK 都可以与之通信。 该连接器在 LiveKit 自身组件绑定的频道上发布字幕和说话状态， 因此这些组件无需额外配置即可与 ADK 智能体配合使用：
| LiveKit 资源 | ADK 智能体获得的能力 |
|---|---|
| Client SDKs | 浏览器、Swift、Android、Flutter、React Native、Unity、C++、Rust 和 ESP32 |
| UI components | 预构建的语音助手小组件，支持 React、SwiftUI、Compose 和 Flutter |
| Starter apps | 各平台可运行的应用，以及 Agents Playground，无需前端即可与智能体对话 |
电话呼叫¶
SIP 呼叫方是普通的 LiveKit 参与者，因此一旦配置了 呼入中继和 分发规则指向你的 worker， 电话呼叫就能到达智能体。
呼叫方的身份在对方开口说话之前就已存入 ADK 会话状态，因此函数工具可以像读取其他状态值一样读取它：
from google.adk.tools.tool_context import ToolContext
async def greet_by_account(tool_context: ToolContext) -> str:
"""在问候之前查找呼叫方信息。"""
number = tool_context.state.get("livekit_caller_phone_number") # '+15105550100'
if not number:
return "I could not see the number you are calling from."
return await crm.lookup(number) # 你自己的客户查询逻辑
连接器将键盘输入缓冲为单个轮次，因此六位数的账号会作为一次输入到达，而不是被打断六次。
游戏和沉浸式客户端¶
LiveKit 的 Unity SDK 为 Unity 应用添加了实时音频、 视频和数据通道，由 LiveKit Cloud 或你自建的服务器提供支持。将 ADK 智能体放入房间，玩家就可以与一个同时作用于 游戏世界的角色进行对话，例如语音驱动的 NPC 或游戏内助手：
from google.adk.integrations.livekit import current_call
from google.adk.tools.tool_context import ToolContext
async def open_the_door(door_id: str, tool_context: ToolContext) -> str:
"""在游戏世界中打开一扇门。"""
call = current_call(tool_context)
return await call.perform_rpc(method="open_door", payload=door_id)
客户端返回的任何内容都会成为模型叙述的工具结果，因此智能体会描述实际发生了什么。在 Unity 客户端上，注册一个 RPC 方法即可；ADK 负责管理 对话、工具调用和会话。
快速开始¶
- ADK >= 2.9.0，需要安装
livekit额外依赖。 - 实时模型的凭据。
- LiveKit 服务器，自建或使用 LiveKit Cloud。两者暴露相同的 API，因此
同一份 worker 代码可以同时适用于两者。本地开发时，运行
livekit-server --dev。 - 在环境中设置
LIVEKIT_URL、LIVEKIT_API_KEY和LIVEKIT_API_SECRET。
从你已有的智能体开始。添加 LiveKitToolset() 可为其提供呼叫控制功能，如
挂断或转接呼叫方。该工具集仅在有呼叫时激活，因此
adk web 仍然可以不变地运行智能体：
from google.adk.agents import Agent
from google.adk.integrations.livekit import LiveKitToolset
from google.adk.runners import InMemoryRunner
root_agent = Agent(
model="gemini-live-2.5-flash-native-audio",
name="support_agent",
instruction="You help customers troubleshoot their home internet.",
tools=[check_line_status, LiveKitToolset()], # check_line_status 是你自己的工具
)
runner = InMemoryRunner(agent=root_agent, app_name="support")
要连接智能体，请将一个已连接的房间传递给 runner。在生产环境中，你运行一个 worker，LiveKit 每次呼叫时分发一次，同一份代码同时服务于浏览器和 电话。
from google.adk.integrations.livekit import LiveKitRunner
from livekit.agents import AgentServer
from livekit.agents import cli
from livekit.agents import JobContext
server = AgentServer()
@server.rtc_session(agent_name="support")
async def entrypoint(ctx: JobContext) -> None:
"""将一次分发的呼叫桥接到 ADK 智能体。"""
await ctx.connect()
# LiveKit 没有 ADK 用户或会话 ID。示例从任务元数据中读取它们。
await LiveKitRunner(
runner=runner, room=ctx.room, user_id="live-user", session_id=ctx.room.name
).start()
if __name__ == "__main__":
cli.run_app(server)
部署 worker¶
worker 主动连接到 LiveKit 服务器，并通过同一连接接收分发的任务，因此它需要出站网络访问，不需要公网地址，也不需要负载均衡器。 除此之外，它是一个普通的 ADK 容器，部署方式与任何 ADK 智能体相同。将 入口点指向你的 worker 模块：
Agents CLI 根据你 pyproject.toml 中的 deployment_target 将该容器部署到 Agent Runtime、Cloud
Run 或 GKE。参见
使用 Agents CLI 部署，或手动部署到
Cloud Run 或 GKE。
Cloud Run 探测 $PORT，而 worker 在固定端口上提供健康检查端点，因此
请确保两者匹配。在创建服务器时从环境变量读取端口：
使用 --port=8081 部署也能达到相同效果，因为这是 worker 在
生产环境中使用的端口。worker 在呼叫之间也会闲置，因此请使用 --no-cpu-throttling 和
--min-instances=1 以保持它持续接受分发任务。
每个分发的任务都在独立的进程中运行，因此请使用持久化的
会话服务。InMemoryRunner 类在呼叫之间不持久化任何数据。