Agent Zero 并行工具运行时源码剖析helpers/parallel_tools.py 架构、契约与实战【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zeroAgent Zero 的parallel工具允许在一次回复中并发启动多个独立工具调用包括子代理调用并以稳定的 job ID 进行等待、收集或取消。本文以 helpers/parallel_tools.py 的 DOX 文档为骨架结合 tools/parallel.py、helpers/defer.py、prompts/agent.system.tool.parallel.md 以及 tests/test_parallel_tool.py 等仓库源码完整讲解并行作业的输入归一化、后台任务调度、生命周期管理、日志与 Prompt 渲染等核心机制。读完本文你将理解 Agent Zero 并行执行的全部契约、边界与排查要点并能直接依据文中的代码路径深入源码验证。一、模块定位谁拥有并行执行的运行时在 Agent Zero 的代码结构中并行能力被拆成两层层文件职责工具入口tools/parallel.py定义ParallelToolhelpers.tool.Tool子类解析action、timeout、job_ids等参数并编排调用顺序运行时实现helpers/parallel_tools.py拥有全部共享运行时归一化工具调用、启动后台作业、等待/取消作业、渲染 prompt extras 与结果 JSONDOX 文档helpers/parallel_tools.py.dox.md明确规定了二者的所有权边界parallel_tools.py拥有运行时实现DOX 文件则承载关于职责、契约、副作用与验证的持久化笔记由于helpers/目录刻意保持扁平文件级 DOX 必须与实现保持同步。对外暴露的公共概念DOX 中列出的五个公共概念与源码一一对应NormalizedToolCall—— 归一化后的工具调用index、tool_name、tool_argsParallelJob—— 单个并行作业的数据载体start_parallel_jobs(...)—— 启动一批后台作业await_parallel_jobs(...)—— 按 job ID 等待/收集结果cancel_parallel_jobs(...)—— 取消作业build_parallel_jobs_extras(...)/format_parallel_results(...)—— 分别渲染 Prompt 附加信息与最终结果 JSON。二、输入归一化兼容多种工具调用信封2.1 三种取值入口extract_tool_calls()会依次从tool_calls、calls、items三个键中取出待包装的调用列表helpers/parallel_tools.py这为不同 provider 或模型输出格式提供了兼容别名tools/parallel.py的execute()首先就调用它。2.2 归一化规则normalize_parallel_tool_calls()helpers/parallel_tools.py是整个并行能力的入口校验器规则如下接受 JSON 字符串如果raw_calls是字符串先尝试json.loads解码用于恢复 provider/model 对数组的字符串化输出解码失败抛出ValueError必须是数组非字符串、非列表直接拒绝空数组拒绝数量上限最多DEFAULT_MAX_CALLS 8个helpers/parallel_tools.py超出报错逐项归一化每个条目调用 helpers/extract_tools.py 的normalize_tool_request()。该函数同时支持tool_name/tool_args、tool/args、以及 OpenAI 风格的type: functionname/parameters三种信封tool_name:action冒号后缀会自动注入action字段method字段也会映射为action忽略规划字段从完整 agent 回复对象复制来的thoughts、headline等字段被忽略只取工具名与参数。测试 tests/test_parallel_tool.py 验证了三种形状tool_name/tool_args、tool/args、JSON 字符串数组都能正确归一化。2.3 硬性禁用清单DISALLOWED_PARALLEL_TOOLS {document_query, response}helpers/parallel_tools.pydocument_query文档解析与问答过重必须在主上下文中顺序执行response必须保持顶层调用以结束消息循环包裹在parallel内会破坏循环终止语义。归一化阶段还会拒绝parallel嵌套parallelcannot be nested inside anotherparallelcall.从源头阻止递归。对应测试见 tests/test_parallel_tool.py。三、作业模型ParallelJob 与生命周期状态机3.1 数据结构ParallelJobhelpers/parallel_tools.py是一个 dataclass核心字段包括id稳定句柄格式为工具名(最多12个字母数字)-uuid8见_new_job_idparent_context_id父上下文 ID用于跨上下文取回作业kindtool直接工具作业或subordinate子代理作业state六态状态机——pending → running → success | error | cancelled | timeoutstarted_at/completed_at/elapsed()时间戳与耗时计算result/error结果与错误信息worker_context_id后台 worker 上下文 IDlog_item/deferred_task对应的可见日志项与底层后台任务句柄。终态集合TERMINAL_STATES {success, error, cancelled, timeout}helpers/parallel_tools.py被等待、刷新、渲染等多处复用。3.2 作业注册表父上下文通过私有数据键PARALLEL_JOBS_KEY _parallel_jobs存放进行中的作业字典_jobs_for_context()helpers/parallel_tools.py。已收集的终态作业会从注册表中移除这是 DOX「Key Concepts」中明确记录的契约收集即消费避免注册表无限膨胀。3.3 后台任务载体DeferredTask每个作业启动时创建一个DeferredTaskhelpers/defer.py它基于单例EventLoopThread后台守护线程运行协程start_task(func, *args)将_run_parallel_job投递到后台事件循环is_ready()/is_alive()用于轮询完成状态result()在调用方事件循环中等待 future 结果kill()取消 future 并清理子任务供取消与超时场景使用。start_parallel_jobs()helpers/parallel_tools.py为每个调用创建作业、写入注册表、记录子日志然后task.start_task(_run_parallel_job, context.id, job.id)异步启动启动失败则立即_finish_job(job, error, ...)标记为错误。四、两种作业执行路径子代理与直接工具_run_parallel_job()helpers/parallel_tools.py按kind分派到两条独立路径成功时_finish_job(job, success, result...)异常时标记error并打印错误。4.1 子代理路径subordinatecall_subordinate在并行中代表隔离的子聊天上下文helpers/parallel_tools.py校验tool_args.message非空支持profile或agent_profile与attachments创建AgentContextType.USER类型的子上下文并通过set_data写入_parallel_parent_context_id、_parallel_job_id、_parallel_worker_kind三个内部键通过set_output_data暴露parent_context_id、parent_context_kind: parallel、parent_context_label、parallel_job_id、parallel_tool_name等元数据供 WebUI 侧栏将并行子聊天渲染为父聊天下的缩进手风琴见 tests/test_parallel_tool.py 对chats-store.js/chats-list.html的断言注入专属系统提示_subordinate_worker_system_prompt以隔离并行 worker 身份运行向父级返回简洁文本摘要运行worker_context.agent0.monologue()完成后history.new_topic()收尾persist_chat.save_tmp_chat在前后各保存一次临时聊天。关键契约子代理作业是独立子聊天绝不进入调度器任务列表。测试 tests/test_parallel_tool.py 断言快照中无scheduler_task_uuidtests/test_parallel_tool.py 则验证state_snapshot中子上下文出现在contexts而非tasks中。4.2 直接工具路径tool非call_subordinate的普通工具走_run_direct_tool_job()helpers/parallel_tools.py创建AgentContextType.BACKGROUND类型的隔离上下文命名parallel:tool_name同样写入三个 worker 内部键复制父项目上下文_copy_project调用execute_tool_call()helpers/parallel_tools.py执行完整工具生命周期handle_intervention→before_execution→tool_execute_before扩展 →execute→tool_execute_after扩展 →after_execution每一步前后穿插干预检查finally中无论成败都清理 worker 上下文。直接工具 worker 禁止递归调用parallelexecute_tool_call()第一行就拒绝tool_name parallel同时 extensions/python/tool_execute_before/_20_block_parallel_recursion.py 中的BlockParallelRecursion扩展会在tool_execute_before阶段对直接工具 worker 再次拦截抛出RepairableException。判定依据是is_parallel_worker()上下文存在_parallel_worker_kind tool或仅有旧的_parallel_job_id即视为直接 worker而kind subordinate的子代理 worker 不受此限制——它们可以正常使用包括parallel在内的子聊天工具。该差异由 tests/test_parallel_tool.py 精确验证。五、等待、收集与取消job ID 驱动的控制流5.1 await_parallel_jobs轮询 超时语义await_parallel_jobs()helpers/parallel_tools.py是核心等待循环参数校验job_ids为空抛ValueError以deadline now timeout为界循环每次先refresh_parallel_jobs()刷新状态再取回各 job未知 job ID 立即报错Unknown parallel job id(s): ...计算活跃作业state 不在终态若waitFalse或没有活跃作业则跳出循环未到 deadline 则asyncio.sleep(POLL_INTERVAL_SECONDS 0.5)继续轮询超时只停止等待不取消作业活跃作业被记入wait_timed_out_job_ids快照打上wait_timed_out: True标记后台任务继续运行后续可用相同job_ids再次 awaitcollectTrue时终态作业在返回前被cleanup_parallel_job清理并从注册表移除。关键语义区分DOX「Key Concepts」原文collect 只返回已完成的作业结果而不等待await 等待指定 job ID。tools/parallel.py中action直接控制这些语义start|background默认waitFalseawait|wait强制waitTruecollect走waitFalse的收集分支cancel走取消分支。5.2 超时行为的测试佐证tests/test_parallel_tool.py 用 fake 时钟验证timeout1 秒、任务仍在running时结果快照带wait_timed_outTrue最终 JSON 的status waiting、wait_timeout true、instruction提示用action: await再次等待同时断言task.killed 0未取消且 job 仍留在注册表中。5.3 cancel_parallel_jobs主动终止cancel_parallel_jobs()helpers/parallel_tools.py先刷新状态然后对每个 job 执行_cancel_job()若deferred_task存活则kill()再_finish_job(job, cancelled, ...)标记终态随后清理 worker 上下文并从注册表移除。测试 tests/test_parallel_tool.py 断言task.killed 1、state 为cancelled、job 已从注册表删除。5.4 上下文清理的完整闭环cleanup_parallel_job()与_remove_context()helpers/parallel_tools.py保证不留垃圾杀掉存活任务后对tool类型调用AgentContext.reset()AgentContext.remove()并persist_chat.remove_chat()删除磁盘上的临时聊天目录。DOX 明确记录该契约直接工具后台上下文清理同时移除内存上下文与磁盘临时聊天文件夹。tests/test_parallel_tool.py 验证了persist_chat.remove_chat被正确调用。六、可见日志并行子任务的行级渲染6.1 子日志先行每个包装调用在启动前都会创建父级可见的子日志项_log_parallel_child_startedhelpers/parallel_tools.py让 WebUI 能独立查看并发的子任务subordinate类型typesubagent标题 Calling Subordinate Agenttool类型优先解析工具并复用其原生get_log_object()输出保留特殊渲染徽章——code_execution_tool用code_exe、wait用progress、MCP 工具用mcp、普通工具回退tool类型DOX 契约原文。解析失败则创建通用tool类型日志并注入_tool_name。三个典型分支分别由 tests/test_parallel_tool.py通用回退、tests/test_parallel_tool.pycode_exe、tests/test_parallel_tool.pyprogress覆盖。6.2 包装器自身不产生可见日志tools/parallel.py的ParallelTool.before_execution()将self.log Noneafter_execution()只把结果写入模型历史hist_add_tool_resultwrapper 本身不产生可见 process-step 日志行。测试 tests/test_parallel_tool.py 断言agent.context.log.items []而 tool_results 记录了包装结果。这保证了并行子日志与常规工具日志形态一致job ID 则通过 wrapper 结果与 prompt extras 暴露而非可见参数。6.3 日志对象复用直接工具 worker 执行时复用父级可见的子日志对象execute_tool_call()临时将工具的get_log_object替换为返回job.log_item并在finally中还原helpers/parallel_tools.py。这样before_execution()不会创建第二个通用 worker 日志也不会丢失原生徽章类型。tests/test_parallel_tool.py 用 FakeTool 验证执行后日志仍保持progress类型且内容更新为结果。七、Prompt Extras 与结果格式化给模型看的紧凑反馈7.1 build_parallel_jobs_extrasbuild_parallel_jobs_extras()helpers/parallel_tools.py在每次上下文渲染时把进行中的并行工作注入系统提示先刷新状态过滤掉cancelled/timeout的作业无作业返回空串保持 prompt 干净有作业时输出固定格式running:下列出运行中作业job ID、工具名、状态、已运行秒数ready to collect with \parallel and job_ids: 下列出可收集的终态作业含耗时结尾提示用paralleljob_ids等待/收集或action: cancel取消。DOX 强调prompt extras 必须保持有界只暴露 job ID、工具名、状态与紧凑的结果/错误摘要防止系统提示被撑爆。测试 tests/test_parallel_tool.py 断言运行中与就绪作业同时出现在 extras 中。7.2 结果状态判定format_parallel_results()helpers/parallel_tools.py根据所有作业状态推导聚合状态条件status存在活跃作业且等待超时waiting存在活跃作业未超时running全部successsuccess全部cancelledcancelled部分successpartial其余error超时时附带wait_timeout: true仍有活跃作业时附带instruction指导模型用awaitjob_ids续等或用cancel停止。输出为紧凑 JSONindent2便于模型阅读并继续决策。format_started_jobs()则在wait: false启动场景返回status: started的作业清单及使用指引。7.3 快照字段_job_snapshot()helpers/parallel_tools.py统一产出job_id、tool_name、state、duration_seconds保留 3 位小数有 worker 时附加context_id按需包含result/error。该结构同时被启动、等待、取消、格式化四个环节复用。八、工具入口编排ParallelTool 的 action 分派tools/parallel.py 的execute()是全部运行时能力的编排中枢顺序如下解析action小写去空格coerce_timeout解析超时默认DEFAULT_TIMEOUT_SECONDS 300秒必须为正整数coerce_bool支持1/true/yes/on等字符串布尔action cancel→cancel_parallel_jobs否则extract_tool_calls取调用列表并start_parallel_jobs启动新作业wait默认值由 action 决定start|background|collect之外的 action 默认trueawait|wait强制truewaitfalse且无新作业与 job_ids 时报错无 job_ids 时返回format_started_jobs后台启动其余走await_parallel_jobswaittrue等待或waitfalse收集结果交给format_parallel_results全程ValueError统一包装为Error: ...消息返回break_loopFalse保持循环继续。参数契约与 prompts/agent.system.tool.parallel.md 完全一致tool_calls、job_ids、wait默认 true、actionstart|await|collect|cancel、timeout。该 prompt 还规定了使用纪律仅用于独立工作、不拆按工具类型分批、禁止嵌套、document_query与response不得入内、wait: false必须随后用job_ids收集等。tests/test_responses_tools.py 验证该 prompt 可被 schema 解析器正确识别为自由参数对象。九、验证路径与排查要点DOX 的 Verification 章节给出两条验证主线对应仓库中的实际资产单元/契约测试tests/test_parallel_tool.py覆盖归一化、递归防护、prompt extras、结果格式化、超时语义、取消、日志类型、子聊天快照与 WebUI 侧栏展示tests/test_responses_tools.py 覆盖 prompt schema 契约端到端验证修改并行执行、子聊天元数据或子代理行为后需要在真实 Agent Zero 会话中调用parallel并观察 WebUI 的子聊天手风琴展示。排查要点速查tool_calls报错 must be an array检查是否传了单对象而非数组或 JSON 字符串损坏at most 8 items单批调用超过DEFAULT_MAX_CALLSUnknown parallel job idjob 已被收集终态并从注册表移除或 ID 拼写错误cannot be nested / cannot be used inside a parallel worker分别在归一化阶段与tool_execute_before扩展处触发前者针对parallel入内后者针对直接工具 worker 递归调用等待超时但任务仍在跑这是设计行为timeout只限制单次等待后续可用同一job_ids再 await子代理并行结果异常先检查persist_chat的临时聊天目录save_tmp_chat/remove_chat与state_snapshot中的父子上下文关联。十、结语helpers/parallel_tools.py是 Agent Zero 并发能力的核心枢纽它用一层严谨的归一化吸收不同 provider 的工具调用信封差异用ParallelJob状态机 DeferredTask后台事件循环管理作业生命周期用隔离上下文同时保障子代理与直接工具的安全执行再用有界的 prompt extras 与紧凑 JSON 结果与模型高效沟通。结合 tools/parallel.py 的 action 编排、helpers/extract_tools.py 的信封解析、extensions/python/tool_execute_before/_20_block_parallel_recursion.py 的递归防护以及 tests/test_parallel_tool.py 的完整测试矩阵读者可以沿任意一条代码路径深入完整掌握这套并行运行时的设计全貌。【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考