ADK-Python高级特性：实时流式处理与音频支持
ADK-Python高级特性：实时流式处理与音频支持【免费下载链接】adk-python一款开源、代码优先的Python工具包，用于构建、评估和部署灵活可控的复杂 AI agents项目地址: https://gitcode...
·
ADK-Python高级特性：实时流式处理与音频支持
项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python ADK-Python框架提供了强大的实时流式处理和音频支持能力，通过精心设计的双向流式通信架构、实时音频处理与转录功能、SSE与WebSockets协议实现以及自定义音频双向流应用开发支持，为构建下一代语音交互和实时AI应用提供了完整的技术栈。本文详细介绍了ADK-Python在实时处理领域的核心架构、关键技术实现和最佳实践。
双向流式通信架构设计
ADK-Python的双向流式通信架构是其实时处理能力的核心，通过精心设计的异步消息队列和事件驱动机制，实现了高效、低延迟的实时数据交换。该架构支持音频流、文本流和活动信号的实时双向传输，为构建实时对话系统和流式AI应用提供了强大的基础。
核心组件架构
ADK的双向流式架构基于以下几个核心组件：
消息队列机制
LiveRequestQueue是双向流式通信的核心，它基于asyncio.Queue实现了一个高效的异步消息队列：
class LiveRequestQueue:
def __init__(self):
# 确保当前线程有事件循环
try:
asyncio.get_running_loop()
except RuntimeError:
# 没有运行中的循环，创建一个新的
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 创建队列（将使用我们确保存在的事件循环）
self._queue = asyncio.Queue()
def send_content(self, content: types.Content):
self._queue.put_nowait(LiveRequest(content=content))
def send_realtime(self, blob: types.Blob):
self._queue.put_nowait(LiveRequest(blob=blob))
async def get(self) -> LiveRequest:
return await self._queue.get()
实时消息格式
LiveRequest定义了双向流式通信中的标准消息格式，支持多种类型的实时数据：
| 字段 | 类型 | 描述 | 使用场景 |
|---|---|---|---|
| content | Optional[Content] | 文本内容消息 | 文本对话模式 |
| blob | Optional[Blob] | 二进制数据块 | 音频/视频流传输 |
| activity_start | Optional[ActivityStart] | 活动开始信号 | 用户开始输入 |
| activity_end | Optional[ActivityEnd] | 活动结束信号 | 用户结束输入 |
| close | bool | 关闭队列信号 | 终止流式会话 |
流式工具管理
ActiveStreamingTool负责管理流式工具的生命周期和资源：
class ActiveStreamingTool(BaseModel):
"""管理流式工具在调用期间的相关资源"""
model_config = ConfigDict(
arbitrary_types_allowed=True,
extra='forbid',
)
task: Optional[asyncio.Task] = None
"""该流式工具的活跃任务"""
stream: Optional[LiveRequestQueue] = None
"""该流式工具的活跃（输入）流"""
调用上下文集成
InvocationContext维护了所有活跃流式工具的状态，确保在多工具环境下的正确协调：
class InvocationContext:
active_streaming_tools: Optional[dict[str, ActiveStreamingTool]] = None
"""本次调用中运行的流式工具"""
run_config: RunConfig
"""运行配置，包含流式模式设置"""
流式处理流程
双向流式通信的处理流程遵循清晰的序列模式：
并发控制机制
ADK使用异步锁机制确保在多线程环境下的线程安全：
# 创建active_streaming_tools修改的异步锁
streaming_lock = asyncio.Lock()
# 线程安全的active_streaming_tools访问
async with streaming_lock:
if invocation_context.active_streaming_tools is None:
invocation_context.active_streaming_tools = {}
if tool.name in invocation_context.active_streaming_tools:
invocation_context.active_streaming_tools[tool.name].task = task
else:
invocation_context.active_streaming_tools[tool.name] = ActiveStreamingTool(
task=task, stream=stream
)
错误处理与恢复
架构设计了完善的错误处理机制：
- 连接异常处理：自动重连机制确保网络波动时的服务连续性
- 消息丢失防护：消息确认机制防止数据丢失
- 资源清理：完善的资源释放机制避免内存泄漏
- 超时控制：可配置的超时设置防止无限等待
性能优化策略
ADK的双向流式架构采用了多项性能优化技术：
| 优化技术 | 实现方式 | 效果 |
|---|---|---|
| 零拷贝传输 | 使用共享内存和引用计数 | 减少内存复制开销 |
| 批量处理 | 消息聚合和批量发送 | 提高网络利用率 |
| 异步IO | 基于asyncio的完全异步架构 | 高并发支持 |
| 连接复用 | 长连接和连接池管理 | 减少连接建立开销 |
| 压缩传输 | 可选的数据压缩 | 减少带宽占用 |
扩展性设计
架构支持水平扩展和垂直扩展：
- 水平扩展：通过负载均衡支持多实例部署
- 垂直扩展：支持硬件加速和GPU加速
- 插件架构：可插拔的工具和协议支持
- 协议兼容：支持多种流式协议和标准
这种精心设计的双向流式通信架构使得ADK-Python能够处理高并发、低延迟的实时AI应用场景，为开发者提供了强大而灵活的实时处理能力。
实时音频处理与转录功能
ADK-Python框架提供了强大的实时音频处理与转录功能，使开发者能够构建支持语音交互的智能体应用。该功能基于Google Cloud Speech-to-Text服务，提供了高效的音频转录能力，支持实时双向流式处理。
音频转录架构设计
ADK-Python的音频处理架构采用模块化设计，主要包括以下核心组件：
核心功能特性
1. 实时音频转录
ADK-Python支持实时音频流的转录处理，能够将音频数据实时转换为文本内容：
from google.adk.flows.llm_flows.audio_transcriber import AudioTranscriber
from google.adk.agents.invocation_context import InvocationContext
# 初始化音频转录器
transcriber = AudioTranscriber(init_client=True)
# 执行音频转录
def process_audio_stream(invocation_context: InvocationContext):
"""处理实时音频流并转录为文本"""
transcribed_contents = transcriber.transcribe_file(invocation_context)
for content in transcribed_contents:
# 处理转录后的文本内容
text = content.parts[0].text
role = content.role # 'user' 或 'model'
print(f"{role}: {text}")
return transcribed_contents
2. 说话人分离与合并
系统能够智能识别不同说话人，并将同一说话人的连续音频片段合并处理，减少转录延迟：
# 音频数据捆绑处理流程
def bundle_audio_segments(transcription_cache):
"""将同一说话人的音频数据合并处理"""
bundled_audio = []
current_speaker = None
current_audio_data = b''
for entry in transcription_cache:
speaker, audio_data = entry.role, entry.data
if speaker == current_speaker:
current_audio_data += audio_data.data
else:
if current_speaker is not None:
bundled_audio.append((current_speaker, current_audio_data))
current_speaker = speaker
current_audio_data = audio_data.data
if current_speaker is not None:
bundled_audio.append((current_speaker, current_audio_data))
return bundled_audio
3. 配置管理
ADK-Python提供了灵活的音频转录配置选项：
from google.genai import types
from google.adk.agents.run_config import RunConfig
# 配置输入音频转录
input_config = types.AudioTranscriptionConfig(
enabled=True,
language_code="zh-CN", # 支持中文转录
model="latest_short" # 使用最新的短音频模型
)
# 配置输出音频转录
output_config = types.AudioTranscriptionConfig(
enabled=True,
language_code="en-US", # 输出使用英文
model="telephony" # 使用电话音频优化模型
)
# 创建运行配置
run_config = RunConfig(
input_audio_transcription=input_config,
output_audio_transcription=output_config
)
实时流式处理集成
ADK-Python的音频处理功能与实时流式处理深度集成：
使用场景示例
1. 实时语音助手
from google.adk import Agent
from google.adk.tools import function_tool
from google.genai import types
@function_tool
async def process_voice_command(audio_data: bytes, tool_context) -> str:
"""处理语音命令并返回执行结果"""
# 这里可以集成自定义的语音处理逻辑
return "命令执行成功"
voice_agent = Agent(
model="gemini-2.0-flash-live-preview",
name="voice_assistant",
description="实时语音助手，支持语音命令处理",
instruction="""
你是一个语音助手，能够处理用户的语音命令。
当收到音频数据时，先进行转录，然后根据文本内容执行相应操作。
支持的命令包括：天气查询、日程安排、设备控制等。
""",
tools=[process_voice_command],
generate_content_config=types.GenerateContentConfig(
safety_settings=[
types.SafetySetting(
category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH,
)
]
)
)
2. 多语言会议转录
def setup_multilingual_transcription():
"""配置多语言会议转录系统"""
configs = {
"english": types.AudioTranscriptionConfig(
language_code="en-US",
model="video" # 视频会议优化模型
),
"chinese": types.AudioTranscriptionConfig(
language_code="zh-CN",
model="video"
),
"japanese": types.AudioTranscriptionConfig(
language_code="ja-JP",
model="video"
)
}
return configs
# 实时语言检测和切换
def detect_and_switch_language(audio_sample):
"""根据音频特征检测语言并切换配置"""
# 实现语言检测逻辑
detected_lang = "chinese" # 示例检测结果
return detected_lang
性能优化策略
ADK-Python在音频处理方面采用了多项性能优化措施：
| 优化策略 | 描述 | 效果 |
|---|---|---|
| 音频分段合并 | 合并同一说话人的连续音频片段 | 减少API调用次数，降低延迟 |
| 缓存管理 | 智能管理转录缓存，避免重复处理 | 提高处理效率 |
| 异步处理 | 支持异步音频处理流程 | 提高系统吞吐量 |
| 连接池 | 重用Speech-to-Text客户端连接 | 减少连接建立开销 |
错误处理与恢复
系统提供了完善的错误处理机制：
async def robust_audio_processing(invocation_context):
"""健壮的音频处理流程，包含错误处理"""
try:
transcriber = AudioTranscriber(init_client=True)
result = await transcriber.transcribe_file(invocation_context)
return result
except Exception as e:
# 记录错误日志
logging.error(f"音频转录失败: {str(e)}")
# 尝试恢复或降级处理
if "quota" in str(e).lower():
# 配额不足，启用降级模式
return await fallback_transcription(invocation_context)
else:
# 其他错误，重新抛出
raise
扩展性与自定义
开发者可以轻松扩展音频处理功能：
class CustomAudioTranscriber(AudioTranscriber):
"""自定义音频转录器，支持特殊音频格式"""
def __init__(self, custom_config=None):
super().__init__(init_client=False)
self.custom_config = custom_config or {}
# 自定义初始化逻辑
async def transcribe_custom_format(self, audio_data, format_type):
"""支持自定义音频格式的转录"""
# 实现特定格式的转录逻辑
pass
ADK-Python的实时音频处理与转录功能为构建下一代语音交互应用提供了强大的基础设施，结合Google Cloud Speech-to-Text服务的先进能力，开发者可以快速构建高性能、高准确度的语音智能体应用。
SSE与WebSockets流式协议实现
ADK-Python框架为实时流式处理提供了强大的SSE（Server-Sent Events）和WebSockets协议支持，这些协议是实现高效双向通信的关键技术。通过精心设计的架构和API，ADK让开发者能够轻松构建支持实时音频、视频和数据流的智能代理系统。
流式模式配置与枚举
ADK通过StreamingMode枚举定义了三种流式处理模式，为不同类型的实时应用场景提供灵活支持：
from enum import Enum
class StreamingMode(Enum):
NONE = None # 非流式模式
SSE = 'sse' # Server-Sent Events模式
BIDI = 'bidi' # 双向流式模式
在运行时配置中，开发者可以通过RunConfig类指定流式模式和相关参数：
from google.adk.agents.run_config import RunConfig, StreamingMode
# 配置SSE流式模式
run_config = RunConfig(
streaming_mode=StreamingMode.SSE,
max_llm_calls=100,
support_cfc=True # 支持组合函数调用
)
# 配置双向流式模式（支持音频）
run_config = RunConfig(
streaming_mode=StreamingMode.BIDI,
speech_config=speech_config,
response_modalities=["AUDIO"],
realtime_input_config=realtime_config
)
SSE协议实现架构
ADK的SSE实现基于FastAPI框架，提供了标准的Server-Sent Events端点。核心实现位于Web服务器模块中：
SSE端点的核心实现代码展示了事件流的处理机制：
@app.post("/run_sse")
async def run_agent_sse(req: RunAgentRequest) -> StreamingResponse:
"""SSE端点实现"""
async def event_generator():
# 创建运行配置
stream_mode = StreamingMode.SSE if req.streaming else StreamingMode.NONE
run_config = RunConfig(streaming_mode=stream_mode)
# 执行代理并流式返回事件
async for event in agent.run_async(
run_config=run_config,
session=session
):
# 序列化事件为SSE格式
sse_event = event.model_dump_json()
logger.debug("Generated event in agent run streaming: %s", sse_event)
yield f"data: {sse_event}\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"}
)
WebSockets双向通信实现
对于需要全双工通信的场景，ADK提供了WebSockets端点支持真正的双向实时交互：
@app.websocket("/run_live")
async def run_agent_live(websocket: WebSocket):
"""WebSockets实时通信端点"""
await websocket.accept()
try:
# 处理实时请求队列
live_queue = LiveRequestQueue()
while True:
# 接收客户端消息
data = await websocket.receive_text()
request = LiveRequest.parse_raw(data)
# 处理不同类型的实时请求
if request.content:
await process_content(request.content, websocket)
elif request.blob:
await process_blob(request.blob, websocket, live_queue)
elif request.activity_start:
await process_activity_start(websocket)
elif request.activity_end:
await process_activity_end(websocket)
except WebSocketDisconnect:
logger.info("WebSocket connection closed")
except Exception as e:
logger.exception("Error during live websocket communication: %s", e)
await websocket.close(code=1011, reason="Internal server error")
实时请求队列管理
ADK设计了专门的LiveRequestQueue类来管理双向流式通信中的请求队列：
class LiveRequestQueue:
"""用于以实时（双向流式）方式发送LiveRequest的队列"""
def __init__(self):
self._queue = asyncio.Queue()
self._active = True
async def send_content(self, content: types.Content):
"""发送内容消息"""
if self._active:
await self._queue.put(LiveRequest(content=content))
async def send_realtime(self, blob: types.Blob):
"""发送实时数据块"""
if self._active:
await self._queue.put(LiveRequest(blob=blob))
async def send_activity_start(self):
"""发送活动开始信号"""
if self._active:
await self._queue.put(LiveRequest(activity_start=True))
async def send_activity_end(self):
"""发送活动结束信号"""
if self._active:
await self._queue.put(LiveRequest(activity_end=True))
async def get(self) -> LiveRequest:
"""获取下一个请求"""
return await self._queue.get()
def close(self):
"""关闭队列"""
self._active = False
MCP工具集的SSE集成
ADK支持通过SSE协议集成Model Context Protocol（MCP）工具集，实现外部服务的流式访问：
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
from google.adk.tools.mcp_tool.mcp_session_manager import SseConnectionParams
# 配置MCP SSE连接
mcp_toolset = MCPToolset(
connection_params=SseConnectionParams(
url='http://localhost:3000/sse',
headers={'Accept': 'text/event-stream'},
),
tool_filter=[
'read_file',
'read_multiple_files',
'list_directory',
'directory_tree',
'search_files',
'get_file_info',
'list_allowed_directories',
],
)
流式工具回调机制
ADK提供了完善的流式工具回调机制，支持在实时处理过程中插入自定义逻辑：
def before_tool_callback(ctx: CallbackContext, tool_name: str, args: dict) -> Optional[dict]:
"""工具执行前回调"""
if tool_name == "monitor_stock_price" and ctx.state().get("execution_context") == "live_streaming":
# 在实时流式上下文中预处理参数
processed_args = preprocess_live_args(args)
return processed_args
def after_tool_callback(ctx: CallbackContext, tool_name: str, result: dict) -> Optional[dict]:
"""工具执行后回调"""
if tool_name == "monitor_stock_price":
# 异步处理流式结果
processed_result = {
"async_processed": True,
"original_result": result,
"processed_at": datetime.now().isoformat()
}
return processed_result
性能优化与错误处理
ADK在流式协议实现中包含了多项性能优化和健壮性措施：
class StreamingProtocolHandler:
"""流式协议处理器"""
def __init__(self, max_buffer_size: int = 1024, timeout: float = 30.0):
self.max_buffer_size = max_buffer_size
self.timeout = timeout
self._buffer = []
self._last_activity = time.time()
async def handle_stream(self, stream: AsyncGenerator):
"""处理事件流"""
try:
async for event in stream:
# 检查超时
if time.time() - self._last_activity > self.timeout:
raise TimeoutError("Streaming timeout")
# 处理缓冲区
self._buffer.append(event)
if len(self._buffer) > self.max_buffer_size:
await self._flush_buffer()
self._last_activity = time.time()
except asyncio.CancelledError:
logger.info("Streaming cancelled")
except Exception as e:
logger.error("Streaming error: %s", e)
await self._cleanup()
async def _flush_buffer(self):
"""刷新缓冲区"""
if self._buffer:
# 批量处理事件以提高性能
processed = await self._process_batch(self._buffer)
self._buffer.clear()
return processed
协议选择指南
根据不同的应用场景，ADK提供了清晰的协议选择指南：
| 特性 | SSE | WebSockets |
|---|---|---|
| 通信方向 | 单向（服务器到客户端） | 双向 |
| 协议开销 | 低（HTTP） | 中等（WebSocket） |
| 连接管理 | 简单 | 复杂 |
| 实时性 | 中等 | 高 |
| 适用场景 | 通知、日志流 | 聊天、实时协作 |
| 音频支持 | 有限 | 完整 |
通过这种分层架构设计，ADK-Python为开发者提供了灵活而强大的流式处理能力，无论是简单的服务器推送事件还是复杂的双向音频流处理，都能找到合适的协议实现方案。
自定义音频双向流应用开发
ADK-Python 提供了强大的实时音频流处理能力，让开发者能够构建支持双向音频通信的智能助手应用。通过集成 Gemini 2.0 Flash Live 模型，您可以创建具有自然对话体验的语音交互应用。
核心配置与初始化
音频双向流应用的核心在于正确的模型配置和流式处理设置。以下是一个完整的音频流代理配置示例：
from google.adk import Agent
from google.adk.tools.tool_context import ToolContext
from google.genai import types
import random
def audio_stream_handler(audio_data: bytes, tool_context: ToolContext) -> str:
"""处理传入的音频流数据"""
# 音频处理逻辑
return "音频处理完成"
def generate_audio_response(text: str) -> bytes:
"""将文本转换为音频输出"""
# 文本转语音逻辑
return b"\x00\xFF\x00\xFF" # 示例音频数据
audio_agent = Agent(
model='gemini-2.0-flash-live-preview-04-09',
name='audio_assistant',
description='支持双向音频流的智能语音助手',
instruction="""
你是一个专业的语音助手，能够实时处理音频输入并生成音频响应。
支持以下功能：
- 实时语音对话
- 音频内容分析
- 多轮语音交互
- 情感识别和响应
音频处理流程：
1. 接收用户音频输入
2. 分析音频内容
3. 生成适当的文本响应
4. 将文本转换为语音输出
""",
tools=[audio_stream_handler, generate_audio_response],
generate_content_config=types.GenerateContentConfig(
safety_settings=[
types.SafetySetting(
category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
threshold=types.HarmBlockThreshold.BLOCK_ONLY_HIGH,
),
],
# 音频流配置
audio_config=types.AudioConfig(
sample_rate=24000,
format=types.AudioFormat.LINEAR16
)
)
)
音频流处理架构
ADK-Python 的音频双向流处理采用模块化架构，确保高效的数据流转和处理：
实时音频配置参数
下表详细说明了音频流处理的关键配置参数：
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
sample_rate |
int | 24000 | 音频采样率（Hz） |
format |
AudioFormat | LINEAR16 | 音频编码格式 |
channels |
int | 1 | 音频通道数 |
bit_depth |
int | 16 | 音频位深度 |
vad_enabled |
bool | True | 语音活动检测 |
noise_suppression |
bool | True | 噪声抑制 |
echo_cancellation |
bool | True | 回声消除 |
高级音频处理功能
ADK-Python 支持多种高级音频处理特性，让您的应用具备专业级的音频处理能力：
from google.genai import types
# 高级音频配置示例
advanced_audio_config = types.AudioConfig(
sample_rate=48000,
format=types.AudioFormat.LINEAR16,
channels=2,
vad_config=types.VoiceActivityDetectionConfig(
enabled=True,
sensitivity=0.8,
min_silence_duration=500 # 毫秒
),
noise_reduction=types.NoiseReductionConfig(
enabled=True,
level=types.NoiseReductionLevel.MEDIUM
),
audio_transcription=types.AudioTranscriptionConfig(
language_code="zh-CN",
enable_automatic_punctuation=True
)
)
音频流状态管理
有效的状态管理是确保音频流连续性的关键。ADK-Python 提供了完善的状态管理机制：
class AudioStreamState:
"""音频流状态管理类"""
def __init__(self):
self.is_streaming = False
self.audio_buffer = bytearray()
self.last_processed_time = 0
self.conversation_context = []
self.audio_quality_metrics = {
'signal_noise_ratio': 0,
'clarity_score': 0,
'latency_ms': 0
}
def update_audio_quality(self, metrics: dict):
"""更新音频质量指标"""
self.audio_quality_metrics.update(metrics)
def add_to_context(self, message: str, role: str):
"""添加上下文信息"""
self.conversation_context.append({
'role': role,
'content': message,
'timestamp': time.time()
})
# 保持上下文长度合理
if len(self.conversation_context) > 10:
self.conversation_context.pop(0)
错误处理与重连机制
在音频流应用中，稳定的错误处理和自动重连机制至关重要：
import asyncio
from typing import Optional
class AudioStreamManager:
"""音频流管理器"""
def __init__(self, max_retries: int = 3):
self.max_retries = max_retries
self.current_retries = 0
self.reconnect_delay = 1.0
async def handle_stream_error(self, error: Exception) -> bool:
"""处理流错误并决定是否重试"""
if isinstance(error, ConnectionError):
return await self._handle_connection_error()
elif isinstance(error, AudioQualityError):
return await self._handle_quality_error()
return False
async def _handle_connection_error(self) -> bool:
"""处理连接错误"""
if self.current_retries < self.max_retries:
self.current_retries += 1
await asyncio.sleep(self.reconnect_delay)
self.reconnect_delay *= 2 # 指数退避
return True
return False
性能优化建议
为了获得最佳的音频流性能，请考虑以下优化策略：
- 音频数据缓冲：使用环形缓冲区减少内存碎片
- 实时优先级：设置适当的线程优先级
- 内存管理：及时释放不再使用的音频资源
- 网络优化：使用 UDP 协议减少延迟
- 压缩算法：选择合适的音频压缩格式
# 性能优化示例
optimized_config = {
'buffer_size': 4096, # 4KB 缓冲区
'thread_priority': 'high',
'compression': 'opus',
'jitter_buffer': 100, # 毫秒
'packet_loss_concealment': True
}
通过以上配置和最佳实践，您可以构建出高性能、高可靠性的自定义音频双向流应用，为用户提供流畅自然的语音交互体验。
总结
ADK-Python框架通过其先进的实时流式处理和音频支持能力，为开发者提供了构建高性能语音交互应用的强大工具集。从底层的双向流式通信架构到顶层的音频应用开发接口，ADK提供了完整的解决方案。其核心优势包括：高效的异步消息队列机制、智能的音频分段合并处理、灵活的流式协议选择（SSE/WebSockets）、完善的状态管理和错误恢复机制。这些特性使得开发者能够轻松构建出支持实时音频对话、多语言会议转录、语音助手等复杂应用场景的智能系统，为实时AI应用的发展奠定了坚实的技术基础。
小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用
更多推荐
30
0- 0
已为社区贡献4条内容
所有评论(0)