agno Workflow 循环执行（Loop Execution）实战指南：端条件求值、迭代上限与并行子循环

agno Workflow 循环执行Loop Execution实战指南端条件求值、迭代上限与并行子循环【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本指南以cookbook/04_workflows/03_loop_execution目录下的可运行示例为讲解主体深入拆解 agno Workflow 引擎中「循环执行」这一核心编排原语。你将掌握Loop组件的完整参数语义、如何用 end-condition 函数控制循环何时收敛退出、如何借助max_iterations防止死循环、如何在循环体内嵌套Parallel并行步骤以及如何利用forward_iteration_output让输出在迭代间累积传递文中同时结合libs/agno/agno/workflow/loop.py等源码佐证底层执行机制。一、目录定位与可运行示例总览本目录是 agno cookbook 中「Workflow」主题cookbook/04_workflows下的第 3 个子主题03_loop_execution其 README.md 明确了目录范围与作用提供可运行的循环工作流示例Runnable workflow examples。目录实际包含以下资源文件说明loop_basic.py演示 Loop 的基本用法端条件end-condition评估器 最大迭代次数保护loop_with_parallel.py演示在循环体内混合Parallel并行步骤与串行步骤loop_iterative_accumulation.py演示 Loop 迭代间输出接力迭代累积TEST_LOG.md示例脚本在验证环境中的执行记录运行前置条件来自原 README激活演示环境.venvs/demo/bin/python使用direnv allow加载 API Key需本地存在.envrc文件。因此下面所有脚本默认都应通过.venvs/demo/bin/python script.py的方式执行。二、为什么需要 Loop从「线性工作流」到「收敛式工作流」在 agno Workflow 中默认步骤序列steps[...]是线性执行的前一步输出作为后一步输入全部跑完即结束。但在真实业务中很多任务天然是迭代型的例如调研类任务一次搜索可能内容不足需要反复搜索直到资料足够详实质量校验类任务生成结果后反复自检、重写直到通过质量门槛数值/状态累积任务每轮对结果做增量处理直到满足某个阈值后退出纠错与细化任务如文档修复fix-validate 循环直到无错误。agno 在 libs/agno/agno/workflow/loop.py 中提供了Loop这一工作流组件它把一组步骤包裹起来反复执行并通过end_condition判定何时停止。它的 docstring 明确说明了三种端条件形式这也是Loop最核心的设计点可调用函数接收List[StepOutput]返回boolCELCommon Expression Language表达式字符串如current_iteration 2None循环按max_iterations设定的次数运行完毕。三个示例文件恰好覆盖了「纯函数端条件」「函数端条件 并行体」「迭代数据传递」三种最常见形态。三、基础形态loop_basic.py—— 端条件评估器与迭代上限loop_basic.py 演示了「研究 → 反复补齐 → 生成内容」的典型场景是理解 Loop 的最小完整骨架。3.1 准备 Agent 与 Step脚本先创建两个 Agent再各自包成Stepresearch_agent Agent( nameResearch Agent, roleResearch specialist, tools[HackerNewsTools(), WebSearchTools()], instructionsYou are a research specialist. Research the given topic thoroughly., markdownTrue, ) content_agent Agent( nameContent Agent, roleContent creator, instructionsYou are a content creator. Create engaging content based on research., markdownTrue, ) research_hackernews_step Step( nameResearch HackerNews, agentresearch_agent, descriptionResearch trending topics on HackerNews, ) research_web_step Step( nameResearch Web, agentresearch_agent, descriptionResearch additional information from web sources, ) content_step Step( nameCreate Content, agentcontent_agent, descriptionCreate content based on research findings, )可以看到Step的标准用法name用于标识与后续结果引用agent指定承载执行的 Agentdescription则用于解释步骤用途。三个示例中Step均由 Agent 驱动agent...这是 cookbook 中最常见的形态不过从源码角度Step 也支持自定义 executor 函数见下文第四节示例。3.2 端条件评估器写一个返回 bool 的函数循环的停止决策完全交给「评估器」。评估器签名固定为接收本轮迭代产出的List[StepOutput]返回布尔值True表示满足条件、可退出循环def research_evaluator(outputs: List[StepOutput]) - bool: if not outputs: return False for output in outputs: if output.content and len(output.content) 200: print( f[PASS] Research evaluation passed - found substantial content ({len(output.content)} chars) ) return True print([FAIL] Research evaluation failed - need more substantial research) return False这段逻辑的实战要点在于判空保护outputs为空时直接返回False语义是「没有任何产出继续循环」遍历所有 step output因为一个迭代内可能执行多个步骤HackerNews 调研 Web 调研只要任一产出内容超过 200 字符即判定「调研充分」打印调试信息[PASS]/[FAIL]日志便于在终端观察循环在每一轮做出了什么决策。StepOutput的来源在 libs/agno/agno/workflow/types.py它是一个 dataclass主要字段包括step_name、content、success、stop、error、steps等其中content即步骤的正文输出评估器主要依赖它做质量判断。3.3 组装 WorkflowLoop 后续串行步骤workflow Workflow( nameResearch and Content Workflow, descriptionResearch topics in a loop until conditions are met, then create content, steps[ Loop( nameResearch Loop, steps[research_hackernews_step, research_web_step], end_conditionresearch_evaluator, max_iterations3, ), content_step, ], )这展示了 Loop 的核心语义循环体是[research_hackernews_step, research_web_step]每一轮按顺序执行每轮结束后框架调用research_evaluator(iteration_results)一旦返回True立即跳出循环把「本循环所有轮次的全部 step outputs」作为整体输出传递给后续步骤此处为content_step若连续 3 轮评估都返回False则由max_iterations3强制终止——这是防止死循环的关键安全阀。源码中 loop.py 的 execute 主循环 就是while iteration self.max_iterations配合每轮结束后的self._evaluate_end_condition(...)共同决定是否break。3.4 四种运行方式同步 / 流式 / 异步 / 异步流式__main__块展示了 agno Workflow 的四种调用方式方便你根据应用形态选用input_text ( Research the latest trends in AI and machine learning, then create a summary ) # Sync workflow.print_response(inputinput_text) # Sync Streaming workflow.print_response(inputinput_text, streamTrue) # Async asyncio.run(workflow.aprint_response(inputinput_text)) # Async Streaming asyncio.run(workflow.aprint_response(inputinput_text, streamTrue))其中print_response同步执行并将最终回复打印到终端streamTrue开启流式输出适合需要边生成边展示的场景aprint_response是异步版本必须配合asyncio.run或在异步环境中await异步流式则组合二者。四、循环体内编排loop_with_parallel.py—— Loop × Parallelloop_with_parallel.py 展示的进阶形态是循环体的每一步本身可以是一个组合结构。此处循环体内第一步是一个Parallel三个子步骤并行第二步是串行的情绪分析随后再由评估器决定是否进入下一轮。workflow Workflow( nameAdvanced Research and Content Workflow, descriptionResearch topics with parallel execution in a loop until conditions are met, then create content, steps[ Loop( nameResearch Loop with Parallel Execution, steps[ Parallel( research_hackernews_step, research_web_step, trend_analysis_step, nameParallel Research Analysis, descriptionExecute research and analysis in parallel for efficiency, ), sentiment_analysis_step, ], end_conditionresearch_evaluator, max_iterations3, ), content_step, ], )4.1 两种 Loop 编排形态的对比维度loop_basic.pyloop_with_parallel.py循环体步骤数2 个串行 step第 1 步为 3 路并行的Parallel第 2 步串行分析Agent 数量2research、content3research、analysis、content端条件判断依据任一产出 200 字符本轮全部产出总长度 500 字符4.2 基于「累计内容量」的评估器并行版本的评估器不再只看单条产出而是把本轮所有 output 的字符数累加def research_evaluator(outputs: List[StepOutput]) - bool: if not outputs: return False total_content_length sum(len(output.content or ) for output in outputs) if total_content_length 500: print( f[PASS] Research evaluation passed - found substantial content ({total_content_length} chars total) ) return True print( f[FAIL] Research evaluation failed - need more substantial research (current: {total_content_length} chars) ) return False注意这里用了output.content or 相比 basic 版更健壮当某个StepOutput.content为None时不会触发len()报错。4.3 一个迭代内的真实执行时序结合 loop.py execute 的源码每轮迭代的处理流程为依次遍历循环体内的步骤此处为Parallel、sentiment_analysis_stepParallel内部三个子步骤并发执行后产出一个合并 output每一步的产出都会被收集进iteration_results并更新loop_step_outputs供步骤间接力本轮所有步骤执行完 →iteration 1→ 调用_evaluate_end_condition(iteration_results, iteration)判断是否退出未满足条件且未达到max_iterations则进入下一轮重新执行一遍Parallel 分析。换句话说只要端条件不满足每个新迭代都会重新并行搜索、重新做趋势分析这是「越搜越深」类 Agentic 任务的标准写法。五、迭代间的数据接力loop_iterative_accumulation.py与forward_iteration_output前两个示例每一轮都使用原始输入即默认行为每轮收到的是最初传入 step_input 的数据。而 loop_iterative_accumulation.py 演示了另一种模式把上一轮的输出作为下一轮的输入实现累积、提炼、收敛等迭代算法。def increment_executor(step_input: StepInput) - StepOutput: Increment the previous steps numeric content by 10. last_content step_input.get_last_step_content() if last_content and last_content.isdigit(): new_value int(last_content) 10 return StepOutput(contentstr(new_value)) return StepOutput(content0) workflow Workflow( nameIterative Accumulation Workflow, descriptionDemonstrates loop iterations carrying forward output from previous iterations., steps[ Step( nameInitial Value, descriptionPass through the initial input value., executorlambda step_input: StepOutput(contentstep_input.input), ), Loop( nameIncrement Loop, descriptionIncrement value by 10 each iteration until it reaches 50., steps[ Step( nameIncrement Step, descriptionAdd 10 to the current value., executorincrement_executor, ) ], end_conditionlambda step_outputs: int(step_outputs[-1].content) 50, max_iterations10, forward_iteration_outputTrue, ), ], )5.1 关键参数forward_iteration_output在 Loop 的字段定义 中forward_iteration_output: bool False的注释写得很清楚若为 True则每次迭代的输出会被转发为下一次迭代的输入默认 False 时每一轮迭代都收到最初的步骤输入。对应地在 execute 主循环末尾有if self.forward_iteration_output: step_input current_step_input这段逻辑——即每轮结束后把「按最新产出更新过的 step_input」回填给下一轮。5.2get_last_step_content()读取最近一次步骤内容由于开启了输出接力Step 的 executor 就能通过StepInput读取前一轮的成果。StepInput.get_last_step_content 会从previous_step_outputs中取出最近一步的 content必要时递归下沉到嵌套结构的最深层increment_executor据此把上轮数字加 10。脚本注释中的预期轨迹起始值 35阈值为 ≥50第 1 轮迭代35 → 45不满足 ≥50继续第 2 轮迭代45 → 55≥50end_condition命中退出。这个例子虽然用一个纯函数模拟数值累积但它演示的机制可扩展到任何「每轮基于上轮结果精修」的任务例如长文档分块改写、多轮反思、数值逼近等。StepInput上还提供了更细粒度的取值工具见 types.py StepInput包括get_step_output(name)按步骤名取、get_step_content(name)取内容、get_all_previous_content()拼接全部历史内容可供需要精确引用上轮某一步产出的场景使用。六、源码视角Loop 的停止决策与执行语义为了在实战中写出可靠的端条件有必要理解 loop.py 底层的决策规则6.1 端条件三分支_evaluate_end_condition核心方法位于 loop.py#L290-L316求值规则为end_condition is None直接返回False即「永不提前结束」循环必然跑满max_iterationsend_condition为字符串按 CEL 表达式处理。注意 CEL 依赖cel-python包若未安装会打印错误提示源码中日志为pip install cel-python并返回False因此使用 CEL 前需确认依赖可用end_condition为可调用对象直接self.end_condition(iteration_results)异步执行时若函数是协程函数会自动await见_aevaluate_end_condition。此外在求值异常时框架会捕获异常、记录 warning 并返回False继续循环交由max_iterations兜底不会让整个 Workflow 崩溃——这解释了为什么迭代上限在工程上是必须配置的。6.2 CEL 端条件速查如果不想写 Python 函数end_condition可以直接传 CEL 字符串。Loop 的 docstringloop.py#L46-L68列出了可用的上下文变量CEL 变量含义current_iteration当前迭代序号从 1 开始max_iterations本 Loop 配置的最大迭代数all_success本轮所有步骤是否全部成功last_step_content本轮最后一个步骤的 content 字符串step_outputs步骤名 → content 字符串 的映射官方示例表达式包括current_iteration 2、current_iteration max_iterations、all_success、last_step_content.contains(DONE)、all_success current_iteration 2等。cookbook 中另有专门的 CEL 示例如 cel_iteration_limit.py、cel_content_keyword.py、cel_compound_exit.py、cel_step_outputs_check.py可对照学习如何用纯字符串声明退出逻辑。6.3 其它内置退出路径除end_condition外源码中还存在两条隐含的退出路径编写复杂流程时可留意子步骤请求提前终止如果循环体内某个StepOutput.stopTrue如用户主动 stop 或条件分支要求停止execute会记录early_termination并跳出 while 循环loop.py#L517-L520每轮人工审查per-iteration reviewLoop 支持通过human_review在开跑前确认、在每轮结束后暂停等待人工继续/否决对应方法见 create_iteration_review_requirement。开启后即使端条件未满足也会先暂停、由人来决定是否续跑。6.4 Loop 的返回结构Loop 无论正常收敛还是跑满次数最终返回一个聚合的StepOutputloop.py#L569-L577content一段摘要文本如Loop name completed N iterations with M total stepssteps包含各轮次扁平化后的全部 step outputs后续步骤可递归检索StepInput.get_step_content对 Loop/Condition 这类嵌套结构会下沉取到最深层内容success由所有子步骤的success取并stop任一子步骤设置了stop则置真。七、运行与验证建议运行入口优先使用.venvs/demo/bin/python loop_basic.py等命令在演示环境内运行并确保已direnv allow加载 API Key原 README 的前置条件依赖 Agent 的真实推理前两个示例依赖HackerNewsTools、WebSearchTools与底层大模型端条件是否满足取决于模型输出长度属于概率性行为loop_iterative_accumulation.py则是纯函数确定性示例便于离线理解数据接力语义验证记录参考TEST_LOG.md 记录了这两个脚本在自动化验证环境下因 35 秒超时未跑完卡在初始化 OpenAI 客户端阶段的情况说明此类 Agent 密集型脚本的实际耗时受网络与模型响应影响较大评测时应放宽超时阈值相关延伸示例若需组合 Loop 与其它编排原语可参考 router_with_loop.pyRouter 与 Loop 组合、early_stop_loop.py提前停止与 nested_workflow_with_loop.py嵌套 Workflow 作为 Loop 内步骤。八、小结通过cookbook/04_workflows/03_loop_execution的三个示例可以完整掌握 agno Workflow 的循环编排能力Loop(steps[...], end_conditionfn, max_iterationsN)是最基本的循环骨架配合返回bool的评估器可实现「调研直到内容足够」「重试直到达标」等收敛式流程Loop的 steps 可以是任意编排原语用Parallel做循环体内的并行扇出loop_with_parallel.py用Condition、Router、甚至嵌套Workflow都能组合出复杂控制流forward_iteration_outputTrue开启「上一轮输出作为下一轮输入」的数据接力配合StepInput.get_last_step_content()/get_step_content(name)等 API 可实现累积、精修、逼近类算法终止安全三保险end_condition求值为真、max_iterations封顶、步骤stop提前终止三者共同避免失控循环需要人工把关时还可叠加human_review的逐轮审查。当你需要在 agno 中实现任何「直到满足条件才继续往下走」的智能体流程时优先考虑用Loop包裹重试与评估逻辑并把「条件不够充分」作为常态而非异常来处理——这正是本目录三个示例背后共同的工程哲学。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考