尧图网站建设 
Agent Zero 响应流扩展response_stream深度解析流式日志、Include 别名替换与实时响应同步机制【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zeroAgent Zero 的extensions/python/response_stream目录是负责整段助手响应流更新的后端扩展点它接管 LLM 输出流式到达过程中的日志记录、§§include(...)文件别名占位符替换以及 UI 实时响应气泡的同步更新。读完本文你将掌握该扩展点的三个核心扩展模块_10_log_from_stream.py、_15_replace_include_alias.py、_20_live_response.py的职责边界、实现细节与调用契约并理解它如何与response_stream_chunk、response_stream_end以及before_main_llm_call协同共同构建出边生成边渲染、边掩码边记录的完整流式输出链路。扩展点在生命周期中的位置与职责边界response_stream是 Agent Zero 后端数十个生命周期扩展点之一全部位于 extensions/python 目录下。该目录的 DOX即 AGENTS.md明确约定了每个直接子目录对应一个命名扩展点且 Python 文件按确定性的文件名顺序加载数字前缀即排序依据。与响应流相关的扩展点共有四个构成一条完整的流处理流水线扩展点职责对应实现response_stream_chunk响应流逐块掩码与打印_10_mask_stream.pyresponse_stream整段响应流日志、别名替换、实时 UI 更新_10_log_from_stream.py、_15_replace_include_alias.py、_20_live_response.pyresponse_stream_end流结束时的掩码收尾与日志清理_10_mask_end.py、_15_log_from_stream_end.pyreasoning_stream姊妹扩展点推理流thinking的实时日志_10_log_from_stream.py从源码结构看response_stream与姊妹扩展点reasoning_stream处理 reasoning 流在设计中刻意保持了对称两者都通过log_item_generating这个params_temporary键来承载正在生成的日志条目并由 before_main_llm_call/_10_log_for_stream.py 提供共享的标题构造工具函数build_heading/build_default_heading。这意味着任何 agent 的每次主模型调用都会先由before_main_llm_call创建一个Calling LLM...日志占位再由流扩展点持续刷新它。核心契约Local Contractsresponse_stream/AGENTS.md 中定义了四条不可破坏的本地契约它们是理解该模块一切实现的前提流式输出必须与 UI 日志项保持同步扩展点产出的LogItem通过loop_data.params_temporary在整轮消息循环中传递后续阶段依赖同一 id 继续更新解析出的流快照是部分数据嵌套的工具字段如tool_args内部的子字段在对应 token 到达之前可能是None消费方必须容忍缺失字段保留 include 别名替换语义prompt / 工具参数中依赖的§§include(...)占位符行为不能被破坏实时响应中不得暴露未脱敏的密钥与response_stream_chunk/response_stream_end的掩码机制共同保证。模块一_10_log_from_stream.py—— 流式日志与实时步骤描述该模块定义LogFromStream扩展类是响应流扩展点的日志中枢。其核心逻辑分三步推导标题 → 创建/复用日志项 → 刷新内容与 KVPS。智能标题推导标题并非固定文案而是根据当前已解析出的流快照动态生成源码 extensions/python/response_stream/_10_log_from_stream.py若parsed中存在headline字段 →build_heading(agent, parsed[headline])否则若存在tool_name→ 标题为Using {tool_name}应对 LLM 跳过了 headline 的情形否则若存在thoughts→ 标题为Thinking... 按文本长度开方生成的进度条| * ceil(sqrt(len(text))/2)兜底 →Receiving...。标题统一经由 before_main_llm_call/_10_log_for_stream.py 的build_heading生成其实现会给所有 agentA0、A1、A2…加上agent_name:前缀保证多 agent 场景下日志归属一目了然。日志项复用与 KVPS 刷新与before_main_llm_call相同模块通过loop_data.params_temporary[log_item_generating]判断是否已创建日志项if log_item_generating not in loop_data.params_temporary: loop_data.params_temporary[log_item_generating] ( self.agent.context.log.log(typeagent, headingheading) )创建后立即读取该日志项的 KVPS并保留上一轮已有的reasoning字段避免推理内容被覆盖再注入 UI 步骤描述step通用工具调用 →Using {tool_name}...code_execution_tool且带tool_args时进一步细分Python →Writing Python code... (N)、Node.js →Writing Node.js code... (N)、terminal →Writing terminal command... (N)其中N为代码字符数源码 L53-L68。最后kvps.update(parsed)将整个已解析快照并入 KVPS再调用log_item.update(heading..., contenttext, kvpskvps)一次性刷新。注意对快照的部分性处理此处直接合并parsed而嵌套字段如tool_args.code在到达前可能为None模块用isinstance(tool_args, dict)、isinstance(code, str)等防御式判断规避了空值崩溃。模块二_15_replace_include_alias.py—— Include 别名占位符替换该模块定义ReplaceIncludeAlias扩展类职责是在流式快照中就地对tool_args中的文件包含占位符进行替换源码 extensions/python/response_stream/_15_replace_include_alias.py。占位符语法Agent Zero 使用§§include(路径)作为文件内容内联语法例如在工具参数中写入§§include(projects/myproject/README.md)扩展会将其替换为对应文件的完整文本内容。默认正则模式定义在 helpers/strings.pydef replace_file_includes(text: str, placeholder_pattern: str r§§include\(([^)])\)) - str:_15_replace_include_alias.py调用时显式传入该模式replace_file_includes(new_val, r§§include\(([^)])\))。递归替换的防御式实现replace_placeholders是一个递归函数深度遍历tool_args的数据结构str→ 执行replace_file_includesdict→ 对每个 value 递归list→ 对每个元素递归tuple→ 递归并还原为 tuple其他类型→ 原样返回。该扩展只在parsed同时包含tool_args与tool_name时生效if tool_args in parsed and tool_name in parsed替换后的结果写回parsed[tool_args]从而让下游工具执行阶段拿到的是已展开的完整内容。底层语义失败时保留占位符helpers/strings.py 的replace_file_includes实现值得关注它先用files.fix_dev_path(path)修正开发环境路径再读取文件内容若文件不可读异常则原样保留占位符文本而非抛错。这一失败即保留的语义正是契约中保留 include 别名替换语义的具体体现——prompt / 工具参数的构造不会因单个文件缺失而中断。模块三_20_live_response.py—— 实时响应气泡的同步该模块定义LiveResponse扩展类负责把流中识别出的最终response工具调用实时渲染为 UI 上的回复气泡源码 extensions/python/response_stream/_20_live_response.py。响应识别与消息提取模块先调用 helpers/extract_tools.py 的normalize_tool_request(parsed)归一化工具请求然后按优先级提取文本message tool_args.get(text) if not isinstance(message, str) or not message.strip(): message tool_args.get(message)只有当tool_name response且message为非空字符串时才继续否则直接return此时流快照仍在生成中尚未构成最终回复。这保证了只有真正的最终回复才会触发响应气泡的创建。与生成日志共享 ID 的巧妙设计创建响应日志项时模块从loop_data.params_temporary中取出已有的log_item_generating复用其 idgen_item loop_data.params_temporary.get(log_item_generating) shared_id gen_item.id if gen_item and gen_item.id else self.agent.context.log.log(typeresponse, headingficon://chat {self.agent.agent_name}: Responding, idshared_id)共享 id 的意义在于UI 的 branching分支视图能将生成中日志项与最终回复气泡合并为同一节点避免把同一回复渲染成两条独立记录。此后每次流更新都执行log_item.update(contentmessage)持续刷新气泡文案整个逻辑包在try/except中即使异常也静默通过不影响主链路。相邻扩展点分块掩码与流结束收尾要理解response_stream的完整语境必须同时看它的上下游邻居——二者共同保证了不暴露未脱敏密钥的契约。response_stream_chunk逐块脱敏extensions/python/response_stream_chunk/_10_mask_stream.py 的MaskResponseStreamChunk在每个块到达时执行通过get_secrets_manager(self.agent.context)获取密钥管理器首次调用时用create_streaming_filter()创建流式过滤器并以_resp_stream_filter为键缓存在 agent 数据上agent.get_data/agent.set_data用filter_instance.process_chunk(stream_data[chunk])处理当前块并回写stream_data[chunk]对stream_data[full]整体执行secrets_mgr.mask_values(...)保证累积文本与单块一致通过 helpers/print_style.py 的PrintStyle().stream(processed_chunk)实时打印脱敏后的内容。流式过滤器之所以必要是因为密钥可能跨块被截断一个 secret 的前半段在块 A、后半段在块 B逐块过滤器通过维护内部状态识别并掩码跨块片段。response_stream_end掩码收尾与步骤清理extensions/python/response_stream_end/_10_mask_end.py 的MaskResponseStreamEnd在流结束时若_resp_stream_filter仍存在调用finalize()冲刷缓冲中剩余的掩码内容打印 tail 后清理过滤器agent.set_data(filter_key, None)异常时静默通过保证流结束阶段不阻断主流程。extensions/python/response_stream_end/_15_log_from_stream_end.py 的LogFromStream则做日志收尾从params_temporary取出log_item_generating移除step键工具步骤描述在流结束后不再有意义并触发一次 KVPS 更新。这与response_stream/_10_log_from_stream.py的写入 step形成对称的开启/关闭配对。姊妹扩展点reasoning_streamextensions/python/reasoning_stream/_10_log_from_stream.py 与响应流日志几乎同构但针对思考过程标题固定为Reasoning... {进度条}KVPS 写入reasoning与step形如Reasoning... (字符数)。两处日志共享log_item_generating键从代码结构看推理流与响应流在同一轮循环内先后刷新同一个日志项最终在 UI 中呈现为完整的思考 行动 回复过程记录。扩展的执行机制调用契约与加载顺序response_stream目录下的三个扩展类均为Extension子类必须实现execute方法方法签名需与钩子点传入的参数匹配详见 extensions/python/AGENTS.md 的 Local Contracts。执行入口位于 helpers/extension.pycall_extensions_async(extension_point, agent, **kwargs)L228-L240遍历该扩展点在该 agent 上注册的扩展类逐个实例化并execute(**kwargs)若返回 awaitable 则 awaitcall_extensions_sync(...)L243-L255同步版本若扩展返回 awaitable 会直接抛ValueError防止异步对象泄漏。两个入口都通过_get_extension_classes获取该扩展点 该 agent的类列表因此扩展加载天然支持按 agent 差异化——不同 agent 配置可以启用不同的扩展集。文件名数字前缀_10_、_15_、_20_决定了同一扩展点内多个扩展的执行顺序这在流式掩码、日志记录等有严格先后依赖的场景中至关重要。变更验证清单按照 response_stream/AGENTS.md 的 Verification 约定任何修改该扩展点的行为后都应进行以下冒烟验证流式响应冒烟测试发起一次包含思考 工具调用 最终回复的对话确认 UI 日志依次呈现Reasoning...、Using xxx... / Writing Python code... (N)、最终响应气泡且整个过程中日志标题实时更新实时更新验证确认log_item_generating与响应气泡在流式生成中持续刷新且 branching 视图中二者合并为同一节点共享 id 生效include 别名替换验证在工具参数中写入§§include(文件路径)确认工具实际执行时拿到的是展开后的内容对不存在的路径确认占位符被原样保留而不抛错密钥掩码验证让模型输出包含配置密钥的文本确认终端打印与最终日志中密钥均被脱敏涉及 response_stream_chunk 与 response_stream_end 的配合。由于该扩展点处于 LLM 调用的热路径上修改时还应遵守 extensions/python/AGENTS.md 的通用建议保持模块导入轻量避免在热路径引入重依赖并保证与helpers.extension.call_extensions_async/call_extensions_sync的调用契约兼容。【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Agent Zero 响应流扩展（response_stream）深度解析：流式日志、Include 别名替换与实时响应同步机制
Agent Zero 响应流扩展response_stream深度解析流式日志、Include 别名替换与实时响应同步机制【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zeroAgent Zero 的extensions/python/response_stream目录是负责整段助手响应流更新的后端扩展点它接管 LLM 输出流式到达过程中的日志记录、§§include(...)文件别名占位符替换以及 UI 实时响应气泡的同步更新。读完本文你将掌握该扩展点的三个核心扩展模块_10_log_from_stream.py、_15_replace_include_alias.py、_20_live_response.py的职责边界、实现细节与调用契约并理解它如何与response_stream_chunk、response_stream_end以及before_main_llm_call协同共同构建出边生成边渲染、边掩码边记录的完整流式输出链路。扩展点在生命周期中的位置与职责边界response_stream是 Agent Zero 后端数十个生命周期扩展点之一全部位于 extensions/python 目录下。该目录的 DOX即 AGENTS.md明确约定了每个直接子目录对应一个命名扩展点且 Python 文件按确定性的文件名顺序加载数字前缀即排序依据。与响应流相关的扩展点共有四个构成一条完整的流处理流水线扩展点职责对应实现response_stream_chunk响应流逐块掩码与打印_10_mask_stream.pyresponse_stream整段响应流日志、别名替换、实时 UI 更新_10_log_from_stream.py、_15_replace_include_alias.py、_20_live_response.pyresponse_stream_end流结束时的掩码收尾与日志清理_10_mask_end.py、_15_log_from_stream_end.pyreasoning_stream姊妹扩展点推理流thinking的实时日志_10_log_from_stream.py从源码结构看response_stream与姊妹扩展点reasoning_stream处理 reasoning 流在设计中刻意保持了对称两者都通过log_item_generating这个params_temporary键来承载正在生成的日志条目并由 before_main_llm_call/_10_log_for_stream.py 提供共享的标题构造工具函数build_heading/build_default_heading。这意味着任何 agent 的每次主模型调用都会先由before_main_llm_call创建一个Calling LLM...日志占位再由流扩展点持续刷新它。核心契约Local Contractsresponse_stream/AGENTS.md 中定义了四条不可破坏的本地契约它们是理解该模块一切实现的前提流式输出必须与 UI 日志项保持同步扩展点产出的LogItem通过loop_data.params_temporary在整轮消息循环中传递后续阶段依赖同一 id 继续更新解析出的流快照是部分数据嵌套的工具字段如tool_args内部的子字段在对应 token 到达之前可能是None消费方必须容忍缺失字段保留 include 别名替换语义prompt / 工具参数中依赖的§§include(...)占位符行为不能被破坏实时响应中不得暴露未脱敏的密钥与response_stream_chunk/response_stream_end的掩码机制共同保证。模块一_10_log_from_stream.py—— 流式日志与实时步骤描述该模块定义LogFromStream扩展类是响应流扩展点的日志中枢。其核心逻辑分三步推导标题 → 创建/复用日志项 → 刷新内容与 KVPS。智能标题推导标题并非固定文案而是根据当前已解析出的流快照动态生成源码 extensions/python/response_stream/_10_log_from_stream.py若parsed中存在headline字段 →build_heading(agent, parsed[headline])否则若存在tool_name→ 标题为Using {tool_name}应对 LLM 跳过了 headline 的情形否则若存在thoughts→ 标题为Thinking... 按文本长度开方生成的进度条| * ceil(sqrt(len(text))/2)兜底 →Receiving...。标题统一经由 before_main_llm_call/_10_log_for_stream.py 的build_heading生成其实现会给所有 agentA0、A1、A2…加上agent_name:前缀保证多 agent 场景下日志归属一目了然。日志项复用与 KVPS 刷新与before_main_llm_call相同模块通过loop_data.params_temporary[log_item_generating]判断是否已创建日志项if log_item_generating not in loop_data.params_temporary: loop_data.params_temporary[log_item_generating] ( self.agent.context.log.log(typeagent, headingheading) )创建后立即读取该日志项的 KVPS并保留上一轮已有的reasoning字段避免推理内容被覆盖再注入 UI 步骤描述step通用工具调用 →Using {tool_name}...code_execution_tool且带tool_args时进一步细分Python →Writing Python code... (N)、Node.js →Writing Node.js code... (N)、terminal →Writing terminal command... (N)其中N为代码字符数源码 L53-L68。最后kvps.update(parsed)将整个已解析快照并入 KVPS再调用log_item.update(heading..., contenttext, kvpskvps)一次性刷新。注意对快照的部分性处理此处直接合并parsed而嵌套字段如tool_args.code在到达前可能为None模块用isinstance(tool_args, dict)、isinstance(code, str)等防御式判断规避了空值崩溃。模块二_15_replace_include_alias.py—— Include 别名占位符替换该模块定义ReplaceIncludeAlias扩展类职责是在流式快照中就地对tool_args中的文件包含占位符进行替换源码 extensions/python/response_stream/_15_replace_include_alias.py。占位符语法Agent Zero 使用§§include(路径)作为文件内容内联语法例如在工具参数中写入§§include(projects/myproject/README.md)扩展会将其替换为对应文件的完整文本内容。默认正则模式定义在 helpers/strings.pydef replace_file_includes(text: str, placeholder_pattern: str r§§include\(([^)])\)) - str:_15_replace_include_alias.py调用时显式传入该模式replace_file_includes(new_val, r§§include\(([^)])\))。递归替换的防御式实现replace_placeholders是一个递归函数深度遍历tool_args的数据结构str→ 执行replace_file_includesdict→ 对每个 value 递归list→ 对每个元素递归tuple→ 递归并还原为 tuple其他类型→ 原样返回。该扩展只在parsed同时包含tool_args与tool_name时生效if tool_args in parsed and tool_name in parsed替换后的结果写回parsed[tool_args]从而让下游工具执行阶段拿到的是已展开的完整内容。底层语义失败时保留占位符helpers/strings.py 的replace_file_includes实现值得关注它先用files.fix_dev_path(path)修正开发环境路径再读取文件内容若文件不可读异常则原样保留占位符文本而非抛错。这一失败即保留的语义正是契约中保留 include 别名替换语义的具体体现——prompt / 工具参数的构造不会因单个文件缺失而中断。模块三_20_live_response.py—— 实时响应气泡的同步该模块定义LiveResponse扩展类负责把流中识别出的最终response工具调用实时渲染为 UI 上的回复气泡源码 extensions/python/response_stream/_20_live_response.py。响应识别与消息提取模块先调用 helpers/extract_tools.py 的normalize_tool_request(parsed)归一化工具请求然后按优先级提取文本message tool_args.get(text) if not isinstance(message, str) or not message.strip(): message tool_args.get(message)只有当tool_name response且message为非空字符串时才继续否则直接return此时流快照仍在生成中尚未构成最终回复。这保证了只有真正的最终回复才会触发响应气泡的创建。与生成日志共享 ID 的巧妙设计创建响应日志项时模块从loop_data.params_temporary中取出已有的log_item_generating复用其 idgen_item loop_data.params_temporary.get(log_item_generating) shared_id gen_item.id if gen_item and gen_item.id else self.agent.context.log.log(typeresponse, headingficon://chat {self.agent.agent_name}: Responding, idshared_id)共享 id 的意义在于UI 的 branching分支视图能将生成中日志项与最终回复气泡合并为同一节点避免把同一回复渲染成两条独立记录。此后每次流更新都执行log_item.update(contentmessage)持续刷新气泡文案整个逻辑包在try/except中即使异常也静默通过不影响主链路。相邻扩展点分块掩码与流结束收尾要理解response_stream的完整语境必须同时看它的上下游邻居——二者共同保证了不暴露未脱敏密钥的契约。response_stream_chunk逐块脱敏extensions/python/response_stream_chunk/_10_mask_stream.py 的MaskResponseStreamChunk在每个块到达时执行通过get_secrets_manager(self.agent.context)获取密钥管理器首次调用时用create_streaming_filter()创建流式过滤器并以_resp_stream_filter为键缓存在 agent 数据上agent.get_data/agent.set_data用filter_instance.process_chunk(stream_data[chunk])处理当前块并回写stream_data[chunk]对stream_data[full]整体执行secrets_mgr.mask_values(...)保证累积文本与单块一致通过 helpers/print_style.py 的PrintStyle().stream(processed_chunk)实时打印脱敏后的内容。流式过滤器之所以必要是因为密钥可能跨块被截断一个 secret 的前半段在块 A、后半段在块 B逐块过滤器通过维护内部状态识别并掩码跨块片段。response_stream_end掩码收尾与步骤清理extensions/python/response_stream_end/_10_mask_end.py 的MaskResponseStreamEnd在流结束时若_resp_stream_filter仍存在调用finalize()冲刷缓冲中剩余的掩码内容打印 tail 后清理过滤器agent.set_data(filter_key, None)异常时静默通过保证流结束阶段不阻断主流程。extensions/python/response_stream_end/_15_log_from_stream_end.py 的LogFromStream则做日志收尾从params_temporary取出log_item_generating移除step键工具步骤描述在流结束后不再有意义并触发一次 KVPS 更新。这与response_stream/_10_log_from_stream.py的写入 step形成对称的开启/关闭配对。姊妹扩展点reasoning_streamextensions/python/reasoning_stream/_10_log_from_stream.py 与响应流日志几乎同构但针对思考过程标题固定为Reasoning... {进度条}KVPS 写入reasoning与step形如Reasoning... (字符数)。两处日志共享log_item_generating键从代码结构看推理流与响应流在同一轮循环内先后刷新同一个日志项最终在 UI 中呈现为完整的思考 行动 回复过程记录。扩展的执行机制调用契约与加载顺序response_stream目录下的三个扩展类均为Extension子类必须实现execute方法方法签名需与钩子点传入的参数匹配详见 extensions/python/AGENTS.md 的 Local Contracts。执行入口位于 helpers/extension.pycall_extensions_async(extension_point, agent, **kwargs)L228-L240遍历该扩展点在该 agent 上注册的扩展类逐个实例化并execute(**kwargs)若返回 awaitable 则 awaitcall_extensions_sync(...)L243-L255同步版本若扩展返回 awaitable 会直接抛ValueError防止异步对象泄漏。两个入口都通过_get_extension_classes获取该扩展点 该 agent的类列表因此扩展加载天然支持按 agent 差异化——不同 agent 配置可以启用不同的扩展集。文件名数字前缀_10_、_15_、_20_决定了同一扩展点内多个扩展的执行顺序这在流式掩码、日志记录等有严格先后依赖的场景中至关重要。变更验证清单按照 response_stream/AGENTS.md 的 Verification 约定任何修改该扩展点的行为后都应进行以下冒烟验证流式响应冒烟测试发起一次包含思考 工具调用 最终回复的对话确认 UI 日志依次呈现Reasoning...、Using xxx... / Writing Python code... (N)、最终响应气泡且整个过程中日志标题实时更新实时更新验证确认log_item_generating与响应气泡在流式生成中持续刷新且 branching 视图中二者合并为同一节点共享 id 生效include 别名替换验证在工具参数中写入§§include(文件路径)确认工具实际执行时拿到的是展开后的内容对不存在的路径确认占位符被原样保留而不抛错密钥掩码验证让模型输出包含配置密钥的文本确认终端打印与最终日志中密钥均被脱敏涉及 response_stream_chunk 与 response_stream_end 的配合。由于该扩展点处于 LLM 调用的热路径上修改时还应遵守 extensions/python/AGENTS.md 的通用建议保持模块导入轻量避免在热路径引入重依赖并保证与helpers.extension.call_extensions_async/call_extensions_sync的调用契约兼容。【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考