企业官网建设流程全解析

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-audio（Vertex）或gemini-2.5-flash-native-audio-preview-12-2025（Gemini API），可按你的接入渠道选择。
• 运行组件：把LlmAgent包进App，用InMemoryRunner驱动会话；调用方通过LiveRequestQueue向会话推送文字或音频。run_live的必需参数为live_request_queue，当未传入session时还需要user_id和session_id。
• 输入通道：文字用queue.send_content()，实时音频用queue.send_realtime()，细节见 LiveRequestQueue 指南。

定义两个拦截回调并挂到 agent

live 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( content=types.Content( role='model', parts=[types.Part(text='That input is not allowed.')], ) ) def block_output_keyword( callback_context: CallbackContext, llm_response: LlmResponse, ) -> Optional[LlmResponse]: """Ends the turn when the model's 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( content=types.Content( role='model', parts=[types.Part(text="I can't help with that.")], ) ) root_agent = Agent( name='guarded_agent', instruction='Answer the user.', before_model_callback=block_input_keyword, after_model_callback=block_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 模式下两个回调收到的载荷不同（见 对比表）：

注意 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(name="guarded_app", root_agent=root_agent) runner = InMemoryRunner(app=app) queue = LiveRequestQueue() async def main(): queue.send_content( content=types.Content( role="user", parts=[types.Part.from_text(text="Tell me something forbidden")], ) ) async for event in runner.run_live( user_id="user_123", session_id="session_live", live_request_queue=queue, ): 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_complete=True的事件，随后连接被重置，对话在下一个用户轮继续。也就是说判断拦截是否生效，看run_live事件流里是否出现了你写入LlmResponse的替换文案并带有turn_complete=True，而不是继续收到模型原始生成。

拦截后框架如何重连

当before_model_callback拦截了语音输入、或after_model_callback拦截了模型输出时，模型其实已经处理了本次交互的一部分。框架的重连流程（引自 live model callbacks 指南）：

1. 关闭当前 live 连接；
2. 打开一个新的 live 会话，并清除 session resumption；
3. 新连接的历史从 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上的插件级接口。执行顺序是插件回调先跑：如果某个插件回调返回了LlmResponse，agent 级回调会被跳过。如果你的项目已经挂了全局护栏插件，先确认插件不会把请求提前短路，再决定 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 选项。
• LiveRequestQueue：send_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