agno Workflow 步骤序列化实战Condition、Loop、Parallel、Router 与 HITL 配置的保存与加载【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno导读本指南基于 agno 仓库中 cookbook/93_components/workflows 目录下的 8 个完整示例系统讲解如何将包含高级步骤类型条件Condition、循环Loop、并行Parallel、路由Router、自定义执行函数以及 Human-in-the-Loop 配置的 Workflow 保存到数据库并通过Registry在加载时恢复其中的函数与 Agent 依赖。读完本文你将掌握Workflow.save()与get_workflow_by_id()的完整调用链、Registry在反序列化中的核心作用以及如何在加载后继续运行带人工确认Confirmation、结构化用户输入User Input与路由选择Route Selection的工作流。一、背景为什么需要工作流序列化在 agno 中Workflow是由一组步骤Step及其编排组件Condition、Loop、Parallel、Router构成的组件化执行单元。实际生产中工作流往往需要保存一次、多次复用——例如将一个研究型工作流存档后由不同会话、不同用户重新加载运行。本目录cookbook/93_components/workflows/README.md聚焦的正是这一场景带高级步骤类型的工作流保存与加载。核心挑战在于普通步骤可以直接序列化但Condition的求值函数evaluator、Loop的结束条件end_condition、Router的选择函数selector以及Step的自定义执行器executor都是Python 函数对象无法直接写入数据库——它们必须通过Registry按名字序列化、按名字恢复。二、通用骨架Save → Load → Run 三步曲所有示例遵循同一套操作骨架以 save_custom_steps.py 为例from agno.db.postgres import PostgresDb from agno.registry import Registry from agno.workflow.step import Step from agno.workflow.workflow import Workflow, get_workflow_by_id # 1) 数据库连接同步数据库 db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url) # 2) 保存工作流 version workflow.save(dbdb) # 返回配置版本号 print(fSaved workflow as version {version}) # 3) 通过 id 加载工作流 loaded_workflow get_workflow_by_id(dbdb, idcustom-executor-workflow, registryregistry) # 4) 可选运行加载后的工作流 loaded_workflow.run(Write about AI trends)值得注意的约束来自底层实现workflow.py 中的save()明确指出save()目前不支持异步数据库使用前会校验db_必须是BaseDb的实例否则抛出ValueError(Async databases not yet supported for save(). Use a sync database.)。因此示例全部选用 PostgreSQL 作为持久化后端。关键参数说明参数说明db保存组件与配置的数据库实例可用PostgresDb等同步数据库stage组件的阶段标识默认publishedlabel组件的标签可选notes组件的备注说明可选返回值保存的配置版本号int供后续版本追溯save()还会递归保存步骤内引用的 Agent 等子实体以便为链接固定版本号从源码中_save_step_children的逻辑可以推断。加载侧的get_workflow_by_id()则接受db、id与registry三个核心参数其中id由Workflow.name自动生成下划线转连字符例如nameCustom Executor Workflow对应idcustom-executor-workflow。三、Registry函数与 Agent 的反序列化注册表Python 函数无法被 JSON 化存入数据库因此所有包含自定义函数的示例都必须显式注册registry Registry( nameCustom Steps Registry, functions[transform_content], # 按名字序列化、按名字恢复 )从源码结构看Registry定义在 libs/agno/agno/registry/registry.py它同时支持注册functions、agents与dbs。在 HITL 相关示例中Agent 也通过Registry解析registry Registry( nameHITL Confirmation Registry, agents[research_agent, processor_agent, writer_agent], dbs[db], )注册原则凡是工作流中用到的自定义函数evaluator、end_condition、selector、executor以及需要从数据库恢复的 Agent都必须放进Registry。加载时传入同一个registryag no 即可将数据库中存储的函数名字重新映射回内存中的真实函数对象。四、五种高级步骤类型的保存与恢复4.1 Condition条件求值器的 Registry 恢复save_conditional_steps.py 演示了一个判断主题是否为科技类的条件工作流。求值函数签名固定为(step_input: StepInput) - booldef is_tech_topic(step_input: StepInput) - bool: Returns True to execute the conditional steps, False to skip. topic step_input.input or step_input.previous_step_content or tech_keywords [ai, machine learning, programming, software, tech, startup, coding] is_tech any(keyword in topic.lower() for keyword in tech_keywords) print(fCondition: Topic is {tech if is_tech else not tech}) return is_tech registry Registry(nameCondition Workflow Registry, functions[is_tech_topic]) workflow Workflow( nameConditional Research Workflow, steps[ Condition( nameTechTopicCondition, descriptionCheck if topic is tech-related for HackerNews research, evaluatoris_tech_topic, steps[research_hackernews_step], ), research_web_step, write_step, ], dbdb, )要点StepInput.input是用户原始输入step_input.previous_step_content是上一步输出二者常配合做判断依据返回True执行steps分支返回False跳过示例未配置else_steps如需分支可加else_steps[...]加载时用get_workflow_by_id(dbdb, idconditional-research-workflow, registryregistry)恢复求值函数。Condition类定义于 libs/agno/agno/workflow/condition.py除了evaluator外还支持else_stepsfalse 分支与human_review见下文 HITL 章节。4.2 Custom Executor自定义执行函数的恢复save_custom_steps.py 演示了不依赖 Agent、纯函数执行的步骤。执行器签名固定为(step_input: StepInput) - StepOutputdef transform_content(step_input: StepInput) - StepOutput: previous_content step_input.previous_step_content or transformed f[TRANSFORMED] {previous_content} [END] return StepOutput( step_nameTransformContent, contenttransformed, successTrue, )自定义执行器通过Step(..., executortransform_content)挂载到步骤上同样依赖Registry(functions[transform_content])完成保存/加载。注意StepOutput支持step_name、content、success等字段用于向后续步骤传递结构化结果。4.3 Loop循环与结束条件的 Registry 恢复save_loop_steps.py 演示了循环研究直到内容足够的工作流。结束条件函数签名固定为(outputs: List[StepOutput]) - booldef check_research_complete(outputs: List[StepOutput]) - bool: Returns True to break the loop, False to continue. if not outputs: return False for output in outputs: if output.content and len(output.content) 500: print(fLoop: Research complete - found {len(output.content)} chars) return True print(Loop: Research incomplete - continuing) return False workflow Workflow( nameLoop Research Workflow, steps[ Loop( nameResearchLoop, steps[research_step], end_conditioncheck_research_complete, max_iterations3, # 循环迭代上限防止死循环 ), summarize_step, ], dbdb, )要点end_condition接收历史所有StepOutput列表返回True中断循环max_iterations是硬性上限示例设为3循环结束时统一进入后续summarize_step。Loop类定义于 libs/agno/agno/workflow/loop.py还支持human_review迭代前确认见下文。4.4 Parallel并行步骤无需 Registrysave_parallel_steps.py 演示了并行执行多个研究步骤后汇聚。Parallel是唯一不需要Registry的编排组件因为它只包含普通 Step 子步骤workflow Workflow( nameParallel Research Pipeline, steps[ Parallel( research_hn_step, research_web_step, nameParallelResearch, descriptionRun HackerNews and Web research in parallel, ), write_step, review_step, ], dbdb, )加载方式也最简洁loaded_workflow get_workflow_by_id(dbdb, idparallel-research-pipeline)无需registry参数。整体管线为并行研究 → 撰写 → 审校三段式充分体现Parallel在信息采集场景的聚合价值。Parallel类定义于 libs/agno/agno/workflow/parallel.py。4.5 Router选择器函数的 Registry 恢复save_router_steps.py 演示了根据主题动态路由到不同研究步骤。选择器函数签名固定为(step_input: StepInput) - List[Step]返回实际要执行的步骤列表def select_research_step(step_input: StepInput) - List[Step]: topic step_input.input or step_input.previous_step_content or topic_lower topic.lower() tech_keywords [ai, machine learning, programming, software, tech, startup, coding] selected_steps [] if any(keyword in topic_lower for keyword in tech_keywords): print(Router: Selected HackerNews step for tech topic) selected_steps.append(hackernews_step) if not selected_steps or news in topic_lower or general in topic_lower: print(Router: Selected Web step) selected_steps.append(web_step) return selected_steps workflow Workflow( nameRouter Research Workflow, steps[ Router( nameResearchRouter, selectorselect_research_step, choices[hackernews_step, web_step], ), summary_step, ], dbdb, )要点choices声明所有可选步骤selector从中挑选并返回选择器可返回多个步骤实现即路由又并行的组合语义选择器同样依赖Registry(functions[select_research_step])恢复。Router类定义于 libs/agno/agno/workflow/router.py。五、HITL 配置的序列化Confirmation、User Input 与组合组件HITLHuman-in-the-Loop配置由HumanReview数据类承载定义于 libs/agno/agno/workflow/types.py。它通过to_dict()/from_dict()完成序列化往返因此可以随工作流一起保存和恢复。5.1 核心字段与组件兼容性字段说明适用组件requires_confirmation执行前要求人工确认Step、Loop、Router、Conditionconfirmation_message确认提示语同上requires_user_input执行前收集结构化用户输入Step、Routeruser_input_message输入提示语Step、Routeruser_input_schema输入字段 SchemaUserInputField列表Stepallow_multiple_selections允许用户选择多条路由Routeron_reject拒绝时的行为skip/cancel/else/retryStep、Loop、Condition 等on_error出错时的行为fail/skip/pause通用max_retries最大重试次数默认3通用timeout/on_timeout暂停超时及行为cancel/skip/approve通用OnReject枚举定义在同文件 types.py取值包括skip跳过本步继续、cancel取消整个工作流、else_branchCondition 专属转走 else 分支、retry重试。构造时组件会校验字段兼容性不支持的组合直接抛出ValueError例如requires_output_review与requires_iteration_review不能同时设置。5.2 步骤级确认Confirmationsave_hitl_confirmation_steps.py 演示在ProcessData步骤前加入人工确认Step( nameProcessData, descriptionProcess and validate research (requires confirmation), agentprocessor_agent, human_reviewHumanReview( requires_confirmationTrue, confirmation_messageResearch complete. Ready to process data. Proceed?, on_rejectOnReject.skip, ), )保存/加载后脚本会遍历步骤校验requires_confirmation、confirmation_message、on_reject三个字段是否完整往返round-trip。运行时通过run_output.steps_requiring_confirmation获取待确认项requirement.confirm()/requirement.reject()响应再调用loaded_workflow.continue_run(run_output)继续执行while run_output.is_paused: for requirement in run_output.steps_requiring_confirmation: print(f[HITL] Step {requirement.step_name} requires confirmation) user_input input(\nContinue? (yes/no): ).strip().lower() if user_input in (yes, y): requirement.confirm() else: requirement.reject() run_output loaded_workflow.continue_run(run_output)5.3 结构化用户输入User Inputsave_hitl_user_input_steps.py 演示在内容生成前收集用户偏好语气、长度、是否包含示例。user_input_schema由UserInputField组成字段名、类型、描述都会随工作流往返human_reviewHumanReview( requires_user_inputTrue, user_input_messagePlease provide your content preferences:, user_input_schema[ UserInputField(nametone, field_typestr, descriptionTone: formal, casual, or technical, requiredTrue), UserInputField(namelength, field_typestr, descriptionLength: short, medium, or long, requiredTrue), UserInputField(nameinclude_examples, field_typebool, descriptionInclude practical examples?, requiredFalse), ], ),运行时按field_type做类型转换bool接受 true/yes/1/yint、float转数值其余按字符串处理收集后通过requirement.set_user_input(**user_values)回填再continue_run()。此例还混合了自定义执行器format_output证明函数注册与 HITL 配置可以并存于同一工作流。5.4 HITL 应用于组合组件Condition、Loop、Routersave_hitl_condition_loop_router.py 是 HITL 的组合压测把human_review同时挂到三种编排组件上ConditionevaluatorTrue恒真由人决定走steps还是else_stepson_rejectOnReject.else_branch表示拒绝时走快速摘要分支Loop循环体前要求确认Start iterative refinement? This runs up to 3 iterations.拒绝则skipRouter不用 selector而是用requires_user_inputTrueallow_multiple_selectionsTrue让用户从三个候选步骤Research / Analysis / Summary中自行勾选。运行时按需求类型分派处理steps_requiring_confirmation走confirm()/reject()steps_requiring_route走req.select(chosen)单选或req.select_multiple(chosen)多选。脚本用统一的save_and_verify()/run_workflow()函数遍历三个工作流完整展示保存→加载→校验 HITL 字段→运行的闭环。六、运行环境与前置条件依据目录 README.md 中的说明运行这些示例需要环境变量使用direnv allow加载需要项目根目录存在.envrcPostgreSQL启动本地库脚本为./cookbook/scripts/run_pgvector.sh示例默认连接串为postgresqlpsycopg://ai:ailocalhost:5532/ai执行方式使用虚拟环境解释器运行例如.venvs/demo/bin/python cookbook/93_components/workflows/save_parallel_steps.py。每个脚本的__main__段默认只做保存 加载 校验print_response/run调用被注释掉取消注释即可实际运行加载后的工作流注意 HITL 工作流会进入交互式暂停循环。需要说明的是示例中 Agent 使用了OpenAIChat(idgpt-5.6-luna)这一模型标识实际运行时请按你的可用模型替换。七、实践要点总结同步数据库Workflow.save()目前仅支持同步数据库如 PostgreSQL异步数据库会直接抛错id自动生成Workflow.name会自动转成连字符 id如HITL Loop Workflow→hitl-loop-workflow加载时使用该 id函数必须注册evaluator、end_condition、selector、executor 一律放进Registry(functions[...])否则加载后无法还原为可调用对象Agent 也需注册工作流步骤中引用的 Agent 通过Registry(agents[...], dbs[db])解析加载时才能还原HITL 全量往返HumanReview的全部字段含user_input_schema都会经to_dict/from_dict序列化往返加载后可在运行期逐一校验Parallel 特例只含普通 Step 的Parallel无需 Registry加载最简暂停恢复模式run()返回的run_output通过is_paused、steps_requiring_confirmation、steps_requiring_user_input、steps_requiring_route暴露暂停点处理完交互后必须调用continue_run(run_output)推进流程。八、延伸阅读工作流保存/加载核心实现libs/agno/agno/workflow/workflow.pysave()见 L1377get_workflow_by_id()见 L11892HITL 配置与枚举定义libs/agno/agno/workflow/types.py编排组件源码Conditioncondition.py、Looploop.py、Parallelparallel.py、Routerrouter.pyRegistry 注册表实现libs/agno/agno/registry/registry.py完整示例代码与测试记录cookbook/93_components/workflows含各示例的 TEST_LOG.md【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 agno Workflow 步骤序列化实战：Condition、Loop、Parallel、Router 与 HITL 配置的保存与加载
agno Workflow 步骤序列化实战Condition、Loop、Parallel、Router 与 HITL 配置的保存与加载【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno导读本指南基于 agno 仓库中 cookbook/93_components/workflows 目录下的 8 个完整示例系统讲解如何将包含高级步骤类型条件Condition、循环Loop、并行Parallel、路由Router、自定义执行函数以及 Human-in-the-Loop 配置的 Workflow 保存到数据库并通过Registry在加载时恢复其中的函数与 Agent 依赖。读完本文你将掌握Workflow.save()与get_workflow_by_id()的完整调用链、Registry在反序列化中的核心作用以及如何在加载后继续运行带人工确认Confirmation、结构化用户输入User Input与路由选择Route Selection的工作流。一、背景为什么需要工作流序列化在 agno 中Workflow是由一组步骤Step及其编排组件Condition、Loop、Parallel、Router构成的组件化执行单元。实际生产中工作流往往需要保存一次、多次复用——例如将一个研究型工作流存档后由不同会话、不同用户重新加载运行。本目录cookbook/93_components/workflows/README.md聚焦的正是这一场景带高级步骤类型的工作流保存与加载。核心挑战在于普通步骤可以直接序列化但Condition的求值函数evaluator、Loop的结束条件end_condition、Router的选择函数selector以及Step的自定义执行器executor都是Python 函数对象无法直接写入数据库——它们必须通过Registry按名字序列化、按名字恢复。二、通用骨架Save → Load → Run 三步曲所有示例遵循同一套操作骨架以 save_custom_steps.py 为例from agno.db.postgres import PostgresDb from agno.registry import Registry from agno.workflow.step import Step from agno.workflow.workflow import Workflow, get_workflow_by_id # 1) 数据库连接同步数据库 db_url postgresqlpsycopg://ai:ailocalhost:5532/ai db PostgresDb(db_urldb_url) # 2) 保存工作流 version workflow.save(dbdb) # 返回配置版本号 print(fSaved workflow as version {version}) # 3) 通过 id 加载工作流 loaded_workflow get_workflow_by_id(dbdb, idcustom-executor-workflow, registryregistry) # 4) 可选运行加载后的工作流 loaded_workflow.run(Write about AI trends)值得注意的约束来自底层实现workflow.py 中的save()明确指出save()目前不支持异步数据库使用前会校验db_必须是BaseDb的实例否则抛出ValueError(Async databases not yet supported for save(). Use a sync database.)。因此示例全部选用 PostgreSQL 作为持久化后端。关键参数说明参数说明db保存组件与配置的数据库实例可用PostgresDb等同步数据库stage组件的阶段标识默认publishedlabel组件的标签可选notes组件的备注说明可选返回值保存的配置版本号int供后续版本追溯save()还会递归保存步骤内引用的 Agent 等子实体以便为链接固定版本号从源码中_save_step_children的逻辑可以推断。加载侧的get_workflow_by_id()则接受db、id与registry三个核心参数其中id由Workflow.name自动生成下划线转连字符例如nameCustom Executor Workflow对应idcustom-executor-workflow。三、Registry函数与 Agent 的反序列化注册表Python 函数无法被 JSON 化存入数据库因此所有包含自定义函数的示例都必须显式注册registry Registry( nameCustom Steps Registry, functions[transform_content], # 按名字序列化、按名字恢复 )从源码结构看Registry定义在 libs/agno/agno/registry/registry.py它同时支持注册functions、agents与dbs。在 HITL 相关示例中Agent 也通过Registry解析registry Registry( nameHITL Confirmation Registry, agents[research_agent, processor_agent, writer_agent], dbs[db], )注册原则凡是工作流中用到的自定义函数evaluator、end_condition、selector、executor以及需要从数据库恢复的 Agent都必须放进Registry。加载时传入同一个registryag no 即可将数据库中存储的函数名字重新映射回内存中的真实函数对象。四、五种高级步骤类型的保存与恢复4.1 Condition条件求值器的 Registry 恢复save_conditional_steps.py 演示了一个判断主题是否为科技类的条件工作流。求值函数签名固定为(step_input: StepInput) - booldef is_tech_topic(step_input: StepInput) - bool: Returns True to execute the conditional steps, False to skip. topic step_input.input or step_input.previous_step_content or tech_keywords [ai, machine learning, programming, software, tech, startup, coding] is_tech any(keyword in topic.lower() for keyword in tech_keywords) print(fCondition: Topic is {tech if is_tech else not tech}) return is_tech registry Registry(nameCondition Workflow Registry, functions[is_tech_topic]) workflow Workflow( nameConditional Research Workflow, steps[ Condition( nameTechTopicCondition, descriptionCheck if topic is tech-related for HackerNews research, evaluatoris_tech_topic, steps[research_hackernews_step], ), research_web_step, write_step, ], dbdb, )要点StepInput.input是用户原始输入step_input.previous_step_content是上一步输出二者常配合做判断依据返回True执行steps分支返回False跳过示例未配置else_steps如需分支可加else_steps[...]加载时用get_workflow_by_id(dbdb, idconditional-research-workflow, registryregistry)恢复求值函数。Condition类定义于 libs/agno/agno/workflow/condition.py除了evaluator外还支持else_stepsfalse 分支与human_review见下文 HITL 章节。4.2 Custom Executor自定义执行函数的恢复save_custom_steps.py 演示了不依赖 Agent、纯函数执行的步骤。执行器签名固定为(step_input: StepInput) - StepOutputdef transform_content(step_input: StepInput) - StepOutput: previous_content step_input.previous_step_content or transformed f[TRANSFORMED] {previous_content} [END] return StepOutput( step_nameTransformContent, contenttransformed, successTrue, )自定义执行器通过Step(..., executortransform_content)挂载到步骤上同样依赖Registry(functions[transform_content])完成保存/加载。注意StepOutput支持step_name、content、success等字段用于向后续步骤传递结构化结果。4.3 Loop循环与结束条件的 Registry 恢复save_loop_steps.py 演示了循环研究直到内容足够的工作流。结束条件函数签名固定为(outputs: List[StepOutput]) - booldef check_research_complete(outputs: List[StepOutput]) - bool: Returns True to break the loop, False to continue. if not outputs: return False for output in outputs: if output.content and len(output.content) 500: print(fLoop: Research complete - found {len(output.content)} chars) return True print(Loop: Research incomplete - continuing) return False workflow Workflow( nameLoop Research Workflow, steps[ Loop( nameResearchLoop, steps[research_step], end_conditioncheck_research_complete, max_iterations3, # 循环迭代上限防止死循环 ), summarize_step, ], dbdb, )要点end_condition接收历史所有StepOutput列表返回True中断循环max_iterations是硬性上限示例设为3循环结束时统一进入后续summarize_step。Loop类定义于 libs/agno/agno/workflow/loop.py还支持human_review迭代前确认见下文。4.4 Parallel并行步骤无需 Registrysave_parallel_steps.py 演示了并行执行多个研究步骤后汇聚。Parallel是唯一不需要Registry的编排组件因为它只包含普通 Step 子步骤workflow Workflow( nameParallel Research Pipeline, steps[ Parallel( research_hn_step, research_web_step, nameParallelResearch, descriptionRun HackerNews and Web research in parallel, ), write_step, review_step, ], dbdb, )加载方式也最简洁loaded_workflow get_workflow_by_id(dbdb, idparallel-research-pipeline)无需registry参数。整体管线为并行研究 → 撰写 → 审校三段式充分体现Parallel在信息采集场景的聚合价值。Parallel类定义于 libs/agno/agno/workflow/parallel.py。4.5 Router选择器函数的 Registry 恢复save_router_steps.py 演示了根据主题动态路由到不同研究步骤。选择器函数签名固定为(step_input: StepInput) - List[Step]返回实际要执行的步骤列表def select_research_step(step_input: StepInput) - List[Step]: topic step_input.input or step_input.previous_step_content or topic_lower topic.lower() tech_keywords [ai, machine learning, programming, software, tech, startup, coding] selected_steps [] if any(keyword in topic_lower for keyword in tech_keywords): print(Router: Selected HackerNews step for tech topic) selected_steps.append(hackernews_step) if not selected_steps or news in topic_lower or general in topic_lower: print(Router: Selected Web step) selected_steps.append(web_step) return selected_steps workflow Workflow( nameRouter Research Workflow, steps[ Router( nameResearchRouter, selectorselect_research_step, choices[hackernews_step, web_step], ), summary_step, ], dbdb, )要点choices声明所有可选步骤selector从中挑选并返回选择器可返回多个步骤实现即路由又并行的组合语义选择器同样依赖Registry(functions[select_research_step])恢复。Router类定义于 libs/agno/agno/workflow/router.py。五、HITL 配置的序列化Confirmation、User Input 与组合组件HITLHuman-in-the-Loop配置由HumanReview数据类承载定义于 libs/agno/agno/workflow/types.py。它通过to_dict()/from_dict()完成序列化往返因此可以随工作流一起保存和恢复。5.1 核心字段与组件兼容性字段说明适用组件requires_confirmation执行前要求人工确认Step、Loop、Router、Conditionconfirmation_message确认提示语同上requires_user_input执行前收集结构化用户输入Step、Routeruser_input_message输入提示语Step、Routeruser_input_schema输入字段 SchemaUserInputField列表Stepallow_multiple_selections允许用户选择多条路由Routeron_reject拒绝时的行为skip/cancel/else/retryStep、Loop、Condition 等on_error出错时的行为fail/skip/pause通用max_retries最大重试次数默认3通用timeout/on_timeout暂停超时及行为cancel/skip/approve通用OnReject枚举定义在同文件 types.py取值包括skip跳过本步继续、cancel取消整个工作流、else_branchCondition 专属转走 else 分支、retry重试。构造时组件会校验字段兼容性不支持的组合直接抛出ValueError例如requires_output_review与requires_iteration_review不能同时设置。5.2 步骤级确认Confirmationsave_hitl_confirmation_steps.py 演示在ProcessData步骤前加入人工确认Step( nameProcessData, descriptionProcess and validate research (requires confirmation), agentprocessor_agent, human_reviewHumanReview( requires_confirmationTrue, confirmation_messageResearch complete. Ready to process data. Proceed?, on_rejectOnReject.skip, ), )保存/加载后脚本会遍历步骤校验requires_confirmation、confirmation_message、on_reject三个字段是否完整往返round-trip。运行时通过run_output.steps_requiring_confirmation获取待确认项requirement.confirm()/requirement.reject()响应再调用loaded_workflow.continue_run(run_output)继续执行while run_output.is_paused: for requirement in run_output.steps_requiring_confirmation: print(f[HITL] Step {requirement.step_name} requires confirmation) user_input input(\nContinue? (yes/no): ).strip().lower() if user_input in (yes, y): requirement.confirm() else: requirement.reject() run_output loaded_workflow.continue_run(run_output)5.3 结构化用户输入User Inputsave_hitl_user_input_steps.py 演示在内容生成前收集用户偏好语气、长度、是否包含示例。user_input_schema由UserInputField组成字段名、类型、描述都会随工作流往返human_reviewHumanReview( requires_user_inputTrue, user_input_messagePlease provide your content preferences:, user_input_schema[ UserInputField(nametone, field_typestr, descriptionTone: formal, casual, or technical, requiredTrue), UserInputField(namelength, field_typestr, descriptionLength: short, medium, or long, requiredTrue), UserInputField(nameinclude_examples, field_typebool, descriptionInclude practical examples?, requiredFalse), ], ),运行时按field_type做类型转换bool接受 true/yes/1/yint、float转数值其余按字符串处理收集后通过requirement.set_user_input(**user_values)回填再continue_run()。此例还混合了自定义执行器format_output证明函数注册与 HITL 配置可以并存于同一工作流。5.4 HITL 应用于组合组件Condition、Loop、Routersave_hitl_condition_loop_router.py 是 HITL 的组合压测把human_review同时挂到三种编排组件上ConditionevaluatorTrue恒真由人决定走steps还是else_stepson_rejectOnReject.else_branch表示拒绝时走快速摘要分支Loop循环体前要求确认Start iterative refinement? This runs up to 3 iterations.拒绝则skipRouter不用 selector而是用requires_user_inputTrueallow_multiple_selectionsTrue让用户从三个候选步骤Research / Analysis / Summary中自行勾选。运行时按需求类型分派处理steps_requiring_confirmation走confirm()/reject()steps_requiring_route走req.select(chosen)单选或req.select_multiple(chosen)多选。脚本用统一的save_and_verify()/run_workflow()函数遍历三个工作流完整展示保存→加载→校验 HITL 字段→运行的闭环。六、运行环境与前置条件依据目录 README.md 中的说明运行这些示例需要环境变量使用direnv allow加载需要项目根目录存在.envrcPostgreSQL启动本地库脚本为./cookbook/scripts/run_pgvector.sh示例默认连接串为postgresqlpsycopg://ai:ailocalhost:5532/ai执行方式使用虚拟环境解释器运行例如.venvs/demo/bin/python cookbook/93_components/workflows/save_parallel_steps.py。每个脚本的__main__段默认只做保存 加载 校验print_response/run调用被注释掉取消注释即可实际运行加载后的工作流注意 HITL 工作流会进入交互式暂停循环。需要说明的是示例中 Agent 使用了OpenAIChat(idgpt-5.6-luna)这一模型标识实际运行时请按你的可用模型替换。七、实践要点总结同步数据库Workflow.save()目前仅支持同步数据库如 PostgreSQL异步数据库会直接抛错id自动生成Workflow.name会自动转成连字符 id如HITL Loop Workflow→hitl-loop-workflow加载时使用该 id函数必须注册evaluator、end_condition、selector、executor 一律放进Registry(functions[...])否则加载后无法还原为可调用对象Agent 也需注册工作流步骤中引用的 Agent 通过Registry(agents[...], dbs[db])解析加载时才能还原HITL 全量往返HumanReview的全部字段含user_input_schema都会经to_dict/from_dict序列化往返加载后可在运行期逐一校验Parallel 特例只含普通 Step 的Parallel无需 Registry加载最简暂停恢复模式run()返回的run_output通过is_paused、steps_requiring_confirmation、steps_requiring_user_input、steps_requiring_route暴露暂停点处理完交互后必须调用continue_run(run_output)推进流程。八、延伸阅读工作流保存/加载核心实现libs/agno/agno/workflow/workflow.pysave()见 L1377get_workflow_by_id()见 L11892HITL 配置与枚举定义libs/agno/agno/workflow/types.py编排组件源码Conditioncondition.py、Looploop.py、Parallelparallel.py、Routerrouter.pyRegistry 注册表实现libs/agno/agno/registry/registry.py完整示例代码与测试记录cookbook/93_components/workflows含各示例的 TEST_LOG.md【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考