ADK-Python 如何用 before_model_callback 和 after_model_callback 拦截 live 会话中的内容【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python如果你在 ADK-Python 中用Runner.run_live维护一个与 Gemini Live API 的双向流式会话音频、文字持续进出标准的回调时机就不够用了你希望用户输入在发送给模型之前被检查模型输出在送达客户端之前被检查命中规则时能直接替换回复并结束当前轮次。ADK 的before_model_callback和after_model_callback就是为此提供的拦截点——两个回调都挂在LlmAgent上在run_live执行期间由BaseLlmFlow调用可用于护栏、脱敏、审计日志和内容过滤。本文给出从定义回调、挂到 agent、运行 live 会话到验证拦截效果的完整路径并说明 live 模式下与一次性unary调用不同的行为边界。准备一个可以运行 live 会话的环境before_model_callback/after_model_callback的 live 行为只在run_live中生效因此前置条件以 run_live 指南 为准模型run_live要求模型端点支持 Gemini Multimodal Live API文档给出的示例模型是gemini-2.0-flash-exp。单 agent 流式示例 中实际使用的是gemini-live-2.5-flash-native-audioVertex或gemini-2.5-flash-native-audio-preview-12-2025Gemini API可按你的接入渠道选择。运行组件把LlmAgent包进App用InMemoryRunner驱动会话调用方通过LiveRequestQueue向会话推送文字或音频。run_live的必需参数为live_request_queue当未传入session时还需要user_id和session_id。输入通道文字用queue.send_content()实时音频用queue.send_realtime()细节见 LiveRequestQueue 指南。定义两个拦截回调并挂到 agentlive model callbacks 指南给出的完整示例是检查用户输入与模型输出中是否出现违禁词命中则用替换文案结束本轮。from typing import Optional from google.adk.agents import Agent from google.adk.agents.callback_context import CallbackContext from google.adk.models.llm_request import LlmRequest from google.adk.models.llm_response import LlmResponse from google.genai import types def block_input_keyword( callback_context: CallbackContext, llm_request: LlmRequest, ) - Optional[LlmResponse]: Blocks user input containing a forbidden keyword. text None if llm_request.contents and llm_request.contents[-1].parts: text .join( part.text for part in llm_request.contents[-1].parts if part.text ) if not text or forbidden not in text.lower(): return None # Send the request to the model. return LlmResponse( contenttypes.Content( rolemodel, parts[types.Part(textThat input is not allowed.)], ) ) def block_output_keyword( callback_context: CallbackContext, llm_response: LlmResponse, ) - Optional[LlmResponse]: Ends the turn when the models output so far mentions a blocked term. text None if llm_response.output_transcription: text llm_response.output_transcription.text if not text or forbidden not in text.lower(): return None # Deliver the response unchanged. return LlmResponse( contenttypes.Content( rolemodel, parts[types.Part(textI cant help with that.)], ) ) root_agent Agent( nameguarded_agent, instructionAnswer the user., before_model_callbackblock_input_keyword, after_model_callbackblock_output_keyword, )两个回调的签名和返回值语义是固定的替换词可以按自己的规则改但不能改变返回约定before_model_callback(callback_context, llm_request) - Optional[LlmResponse]检查用户输入。返回None表示放行输入正常发给模型返回LlmResponse表示拦截框架把替换回复发给客户端并记录进会话原始输入被完全扣留。after_model_callback(callback_context, llm_response) - Optional[LlmResponse]在输出送达用户之前检查模型响应。返回None表示原样继续流式下发返回LlmResponse表示立即停止生成用替换回复结束本轮。与 unary 调用相比live 模式下两个回调收到的载荷不同见 对比表行为Non-live (Unary)Live (Bidirectional Streaming)before_model_callback时机每次 LLM 调用前一次文字在发送前评估语音在转写完成后评估before_model_callback载荷完整对话历史在llm_request.contents单项contents列表只含当前被评估的用户消息或转写after_model_callback时机模型响应完成后一次随输出到达持续运行评估累计的音频转写after_model_callback载荷完整生成Content在llm_response.content累计文本在llm_response.output_transcription请求/响应可变性标准可变对象只读快照注意 live 模式下回调收到的是LlmRequest/LlmResponse的只读快照不要试图修改它们要替换内容就返回一个新的LlmResponse。运行 run_live 并验证拦截效果把上面的root_agent接入App和InMemoryRunner按 run_live 指南 的最小示例运行。下面的代码块在示例基础上补充了from google.adk.runners import InMemoryRunner和from google.adk.apps import App两个导入导入路径来自 App 容器指南以及queue.close()收尾model参数按你使用的 Live 模型端点填写import asyncio from google.adk.apps import App from google.adk.live import LiveRequestQueue from google.adk.runners import InMemoryRunner from google.genai import types # root_agent 定义见上文含 before/after_model_callback # 按 run_live 要求补充模型端点例如 # root_agent.model gemini-live-2.5-flash-native-audio # Vertex # root_agent.model gemini-2.5-flash-native-audio-preview-12-2025 # Gemini API app App(nameguarded_app, root_agentroot_agent) runner InMemoryRunner(appapp) queue LiveRequestQueue() async def main(): queue.send_content( contenttypes.Content( roleuser, parts[types.Part.from_text(textTell me something forbidden)], ) ) async for event in runner.run_live( user_iduser_123, session_idsession_live, live_request_queuequeue, ): if event.content and event.content.parts: for part in event.content.parts: if part.text: print(Live model response:, part.text) queue.close() asyncio.run(main())文档给出的验证方式是事件级的当回调返回替换响应后客户端会收到一个携带替换文本、且turn_completeTrue的事件随后连接被重置对话在下一个用户轮继续。也就是说判断拦截是否生效看run_live事件流里是否出现了你写入LlmResponse的替换文案并带有turn_completeTrue而不是继续收到模型原始生成。拦截后框架如何重连当before_model_callback拦截了语音输入、或after_model_callback拦截了模型输出时模型其实已经处理了本次交互的一部分。框架的重连流程引自 live model callbacks 指南关闭当前 live 连接打开一个新的 live 会话并清除 session resumption新连接的历史从 session events 填充其中包含的是替换响应而非被拦截的内容。这条机制的目的就是防止模型在后续轮次记住被拒的内容属于框架自动行为回调代码里不需要额外处理。限制与已知边界以下限制直接影响回调设计来自 live model callbacks 指南 的 Limitations 一节音频 blob 不经过检查。两个回调只对 live bidi 输入文字和语音/模型音频的转写生效输入转写预期先于主模型响应到达输出转写预期与音频交错到达。原始音频数据本身不被筛查。纯文字 agent 的输出不受检查。text-only live agent 不产生输出转写因此不会触发after_model_callback其输出目前无过滤。文档标注这是当前限制未来会修复以包含文本输出筛查。如果你的场景是纯文字会话且依赖输出拦截这一条必须先确认。轮级语义。输出回调收到的是本轮累计文本返回LlmResponse会以你的替换内容结束整个轮次。修改单个流式 chunk 但保持生成继续目前不支持。回调延迟。回调在接收循环内被 await其中的阻塞操作或网络调用会直接增加流式会话延迟规则逻辑应保持轻量。与插件回调的优先级两个回调除了 agent 级接口LlmAgent.before_model_callback/after_model_callback同样存在BasePlugin上的插件级接口。执行顺序是插件回调先跑如果某个插件回调返回了LlmResponseagent 级回调会被跳过。如果你的项目已经挂了全局护栏插件先确认插件不会把请求提前短路再决定 agent 级回调的检查逻辑。延伸阅读Live model callbacks本文回调机制的完整出处含 live 与 unary 行为对比表。Runner Live Streaming (run_live)run_live参数表user_id、session_id、live_request_queue、run_config与RunConfig的speech_config、response_modalities、save_live_blob、session_resumption等 live 选项。LiveRequestQueuesend_content/send_realtime/send_audio_stream_end/close等输入通道方法。Live Bidi Streaming Single Agent 示例含可参考的 Live 模型端点配置。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考