ADK-Python Live 双向流式会话模型回调指南用 before/after_model_callback 实现输入拦截与输出管控【免费下载链接】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-pythonLive双向流式模式是 ADK-Python 中一类特殊的 Agent 运行方式客户端与模型之间保持一条持续打开的连接客户端不断流式上传音频、文本或视频模型则实时回传音频与文本。当需要在这样的实时会话中执行敏感词拦截、内容脱敏、审计日志、内容过滤等安全与合规任务时before_model_callback与after_model_callback就是官方提供的标准拦截点。本文基于官方指南 live_model_callbacks.md并结合仓库源码深入讲解这两个回调在 Live 会话中的执行时机、返回语义、重连机制、配置方式与已知限制帮助你在run_live场景下快速落地一套可运行的实时内容防护方案。什么是 Live 模型回调Live 模式的模型回调与非 Liveunary单次请求-响应模式使用同一套接口before_model_callback和after_model_callback。它们的核心语义在两种模式下一致before_model_callback在模型处理用户输入之前被调用用于检查或阻止用户内容after_model_callback在模型生成输出之后、交付给客户端之前被调用用于检查或替换模型输出。两者的函数签名都接收一个CallbackContext作为第一个参数并返回Optional[LlmResponse]返回None放行请求/响应按原样继续返回一个LlmResponse用返回的内容替换原始内容并结束当前轮次turn。在 Live 模式下这两个回调由BaseLlmFlow统一调度。从源码结构看Live 模式专门实现了独立的执行流模块 _live_llm_flow.py其中包含send_to_model发送用户数据到模型与receive_from_model接收模型输出并加工事件两个核心协程三个回调触发点代码注释中称为 Live callback site 1/2/3就分布在这两条数据通路上。BaseLlmFlow.run_live只是对run_live_flow的薄封装见 base_llm_flow.py。快速开始一个可运行的拦截示例官方指南给出了一个完整的敏感词拦截示例同时注册输入回调与输出回调当用户输入或模型输出中出现关键词forbidden时分别用固定的提示文本进行替换。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, )将上述 Agent 交给run_live运行后行为如下当模型输出命中关键词时客户端收到的是替换后的文本并且该文本所在事件带有turn_completeTrue随后连接被重置下一轮用户对话继续正常进行。这里有两个值得注意的接口细节输入回调读取llm_request.contents在 Live 模式下contents是当前正在评估的那一条用户消息或语音转写文本的单元素快照而不是完整历史示例中通过llm_request.contents[-1]取最后一条消息并拼接其所有文本 parts。输出回调读取llm_response.output_transcriptionLive 模式下模型输出以音频转写文本transcription的形式流式累积output_transcription.text即当前轮次已生成的累计文本。input_transcription与output_transcription是LlmResponse上的标准字段见 llm_response.py。工作原理三个回调触发点从 _live_llm_flow.py 的实现看Live 会话中before_model_callback与after_model_callback实际分布在三条不同的数据路径上官方文档称之为三个 callback site站点 1用户输入文本发送前在send_to_model协程中当LiveRequestQueue取出一个包含用户文本内容的请求时框架会先调用flow._screen_live_user_content内部实现为screen_live_user_content函数见 _live_llm_flow.py对内容执行before_model_callback回调会拿到一个llm_request.model_copy(update{contents: [content]})的副本其中contents只包含当前这条待评估的用户消息若回调返回了LlmResponse框架会将其终态化为事件并显式设置blocked_event.turn_complete True后推入事件队列值得注意的是文本输入被拦截时不会触发重连——因为模型尚未收到任何内容直接丢弃即可源码注释明确说明 a block here does not reconnect because the model has not yet received the content。站点 2模型输出转写文本累积时在receive_from_model协程中框架维护一个turn_output_transcription字符串逐块累积模型输出的转写文本。每收到一段新的输出转写都会构造一个副本callback_llm_response把累积文本放入output_transcription字段finishedFalse然后调用flow._handle_after_model_callback执行after_model_callback回调返回None输出继续流式下发不做任何改动回调返回LlmResponse立即终止生成将替换响应终态化为事件并标记turn_completeTrue随后 yield 一个_ReconnectSentinel(mode_ReconnectMode.RESTART)触发重启式重连源码见 _live_llm_flow.py。站点 3语音输入转写完成后当用户的语音输入被模型转写完成input_transcription.finished为 True 且有文本时框架把转写文本包装成一个roleuser的Content再次调用_screen_live_user_content执行before_model_callback。与站点 1纯文本输入不同语音输入被拦截后会触发RESTART重连因为模型在听语音的过程中可能已经开始生成响应必须断开连接才能丢弃这些在途内容。从源码结构可以推断三个站点的设计动机是Live 模式数据持续流动必须分别覆盖文本输入、模型输出流、语音输入转写三条路径才能保证任何方向的敏感内容都有机会被拦截。返回语义与重连机制before_model_callback 的返回行为返回效果None用户输入照常发送给模型或正常继续生成流程LlmResponse用户输入被拦截框架将替换响应发送给客户端并标记turn_completeTrue同时记入会话历史其中针对两种输入形式有细微差别纯文本输入原始消息完全不会到达模型且不触发重连语音输入转写文本拦截后会自动重置活动连接让模型丢弃在聆听期间已经开始生成的在途响应。after_model_callback 的返回行为返回效果None输出继续原样流式下发LlmResponse生成立即停止替换响应以turn_completeTrue交付给客户端并重置活动连接确保模型不保留被拦截的生成内容拒绝后的重连流程当before_model_callback拦截了语音输入或after_model_callback拦截了模型输出时模型已经处理了交换的一部分内容。为了防止模型在后续轮次中记住被拒绝的内容框架执行三步重连关闭当前活动的 Live 连接打开一个全新的 Live 会话并清除会话恢复session resumption句柄新连接的历史从会话事件session events重建——而会话事件中保存的是替换后的响应而非被拦截的原始内容。这一点在源码中有直接对应_ReconnectMode枚举定义了RESUME与RESTART两种模式见 _live_llm_flow.py拦截场景统一走RESTARTrun_live_flow收到RESTART哨兵事件后会拷贝一份新的 invocation context、清空live_session_resumption_handle、用run_config_for_new_live_session生成新的运行配置再递归调用run_live开启全新会话见 _live_llm_flow.py。配置选项Agent 级回调与插件级回调Live 模型回调复用了 ADK 标准的模型回调接口同时支持两种注册位置Agent 级回调在LlmAgent或便捷类Agent构造时通过before_model_callback/after_model_callback参数传入如上面的示例所示插件级回调在BasePlugin子类中实现before_model_callback/after_model_callback方法见 base_plugin.py。执行顺序约定插件回调先运行。如果某个插件回调返回了响应Agent 级回调会被跳过。这一顺序在统一回调处理函数handle_before_model_callback/handle_after_model_callback中有明确实现见 _model_response_finalizer.py先调用plugin_manager.run_before_model_callback若得到非空响应立即返回否则再调用_run_callbacks依次执行 Agent 的canonical_before_model_callbacks。此外模型回调也可通过配置驱动的方式注册LlmAgentConfig中提供了before_model_callbacks与after_model_callbacks两个CodeConfig列表字段支持以name: my_library.callbacks.before_model_callback的形式从 YAML/JSON 配置中引用回调函数见 llm_agent_config.py。已知限制官方文档明确列出了以下限制使用前需要评估是否适配你的场景音频 blob 本身不被筛查。模型回调运行在 Live 双向输入的文本以及语音/模型音频的转写文本上。输入转写预期在模型主响应之前到达输出转写预期与音频交错到达。纯文本输出模态不被筛查。纯文本的 Live Agent 不产生输出转写因此不会触发after_model_callback其输出处于未筛查状态。官方文档指出这是当前版本的限制未来将修复以纳入文本输出筛查。轮次级turn-level语义。输出回调接收到的是当前轮次的累计文本返回LlmResponse会以替换内容结束整轮。目前不支持在生成仍然活跃的情况下修改单个流式 chunk。请求与响应对象不可变。Live 模式下回调收到的LlmRequest与LlmResponse可能是只读副本不要尝试就地修改如需替换整轮内容应返回新的LlmResponse。若只是给事件附加注解应使用on_event_callback。回调延迟影响流式体验。回调是在接收循环内被 await 的阻塞操作或网络调用会为流式会话引入额外延迟回调内应避免重活。Live 与非 Live 模式对比两种模式共享同一套回调接口但语义存在显著差异官方文档用下表做了系统对比行为非 LiveUnaryLive双向流式before_model_callback时机每次 LLM 调用、生成开始前执行一次文本输入在发送前评估语音输入仅在转写完成后评估before_model_callback载荷llm_request.contents中包含完整对话历史contents为单元素列表只含正在评估的那条用户消息或转写文本after_model_callback时机每次完整模型响应完成后执行一次输出到达时持续执行评估累积的音频转写文本after_model_callback载荷llm_response.content中包含完整生成的Content累积文本在llm_response.output_transcription中拦截拒绝行为直接替换响应或中止执行替换整轮并重置连接以清除模型记忆中已被拒绝的内容请求与响应可变性标准可变对象只读快照这一差异的根源在于两种运行模式的数据形态非 Live 是一次请求、一次完整响应的批式生成Live 则是持续的音频/文本双向流。从 _live_llm_flow.py 的screen_live_user_content实现可以看到Live 模式在调用before_model_callback前用model_copy(update{contents: [content]})显式构造了单条消息快照正是为了适配流式场景下逐条评估的需求。总结Live 模型回调为 ADK-Python 的双向流式 Agent 提供了一套统一、标准的内容管控入口before_model_callback覆盖文本输入与语音转写两条用户输入路径after_model_callback覆盖模型输出转写流返回LlmResponse即可完成替换内容 结束轮次并在涉及模型已处理内容时自动触发RESTART重连以清除模型记忆。配合插件级回调优先执行、配置化注册等能力可以快速搭建实时会话场景下的 guardrail、脱敏与审计能力。使用时请务必注意其局限性——音频 blob 与纯文本输出模态当前不在筛查范围内且回调应保持轻量以避免拖慢流式会话。更深入的理解可继续阅读以下仓库文件官方指南docs/guides/flows/llm_flows/base_llm_flow/live_model_callbacks.mdLive 执行流实现三个回调站点、重连哨兵与重启逻辑src/google/adk/flows/llm_flows/_live_llm_flow.py回调统一处理与响应终态化src/google/adk/flows/llm_flows/_model_response_finalizer.pyBaseLlmFlow.run_live入口src/google/adk/flows/llm_flows/base_llm_flow.py插件级回调接口src/google/adk/plugins/base_plugin.py回调的配置化注册字段src/google/adk/agents/llm_agent_config.py转写字段定义src/google/adk/models/llm_response.py【免费下载链接】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），仅供参考