agno 工作流高级概念全景实战基于 06_advanced_concepts 测试日志的 40 个示例深度解析【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南围绕cookbook/04_workflows/06_advanced_concepts目录下 40 个可运行工作流示例及其测试验证日志展开系统讲解 agno 工作流Workflow的高级能力背景执行、提前停止、护栏Guardrails、历史上下文、长时间运行、运行控制、会话状态、结构化输入输出等。读者读完可以掌握每个高级主题对应的示例文件、运行方式、预期行为以及测试日志所反映的真实运行结果从而在自己的项目中准确选用并验证这些高级特性。一、文档与示例全景测试日志揭示了什么本专题的核心关联文档为 cookbook/04_workflows/06_advanced_concepts/TEST_LOG.md它是一份于 2026-02-08 生成的自动化测试日志逐条记录了该目录下 40 个示例脚本的运行验证结果。每个条目包含四个要素文件路径示例脚本的相对位置按功能主题分子目录存放状态PASS通过或 FAIL失败执行描述使用的解释器.venvs/demo/bin/python、运行模式normal正常执行或startup仅启动校验及超时时间结果执行输出摘要包括超时、报错信息或关键输出片段。按 cookbook/04_workflows/06_advanced_concepts/README.md 的说明示例共覆盖 12 个主题子目录background_execution、early_stopping、guardrails、history、long_running、previous_step_outputs、run_control、run_params、session_state、structured_io、tools、workflow_agent。运行前提是激活 demo 虚拟环境.venvs/demo/bin/python并通过direnv allow加载 API 密钥需本地.envrc文件部分示例如 background_execution、long_running、run_control 中的远程/服务类示例还要求本地 AgentOS 服务可用。从测试日志整体看提前停止、护栏、运行控制中的多数示例以及基础结构化 IO 示例均验证通过而涉及外部搜索WebSearch、DuckDuckGo、远程服务连接或 LLM 长时生成的示例则普遍出现超时或连接失败。这恰好反映了该类示例对网络、外部服务与模型响应速度的依赖而非示例代码本身的逻辑错误。二、测试方法与执行模式解读日志中所有条目均以.venvs/demo/bin/python执行这是项目 cookbook 约定的 demo 虚拟环境。执行模式分为两类理解它们才能正确解读 PASS/FAIL 语义模式说明典型超时判定含义normal完整运行示例主逻辑35sworkflow_agent 为 120s在超时内完成全部执行才算成功startup仅验证进程能否正常启动2s / 8s进程能启动、运行到预期阶段即算通过随后被强制终止例如long_running目录下的三个示例disruption_catchup.py、events_replay.py、websocket_reconnect.py均以startup模式、2s 超时验证日志输出“Startup validation only; process terminated after 2.01s. Starting test in 2 seconds...”表明它们本就是为了测试长时间运行的故障恢复而设计测试框架只确认其能正常启动。同理background_execution/websocket_server.py以 8s 超时启动校验输出INFO: Finished server process [29101]说明服务进程正常启动并退出。这一设计告诉我们判断一个工作流示例是否可用必须结合其运行模式。startup 模式通过代表进程可启动不代表完整业务流程跑通normal 模式超时则通常指向外部依赖模型 API、搜索服务、远程服务器而非代码缺陷。三、background_execution后台执行与 WebSocket相关文件background_execution/background_poll.py、background_execution/websocket_client.py、background_execution/websocket_server.py见 子目录 README。该主题演示工作流如何以后台方式执行并与外部进程通信background_poll.py演示后台轮询background poll测试日志中为 FAIL——在 35s 内超时最后输出停留在Agent Run End调试行说明 Agent 运行未在限定时间内结束websocket_client.pyPASSstartup 8s输出[ERROR] Failed to connect: Multiple exceptions: [Errno 61] Connect call failed这是典型的本机无 WebSocket 服务在监听时的连接拒绝错误macOS/Linux 的 Errno 61 即 ECONNREFUSED进程能正常启动并尝试连接即视为校验通过websocket_server.pyPASSstartup 8s作为服务端正常启动并结束输出INFO: Finished server process。三者组合构成客户端-服务端闭环服务端监听、客户端连接、轮询式后台任务。运行前需按文件头部注释启动本地 AgentOS 服务否则客户端示例会如日志所示出现连接失败。四、early_stopping提前停止的四种姿势相关文件early_stopping/early_stop_basic.py、early_stopping/early_stop_condition.py、early_stopping/early_stop_loop.py、early_stopping/early_stop_parallel.py见 子目录 README。提前停止是工作流的高级控制能力日志中 4 个示例有 3 个通过early_stop_condition.pyPASS在 Condition 分支内部停止整个工作流。结合源码early_stop_condition.py其核心机制是合规检查函数compliance_checker在发现内容含violation/illegal关键词时返回StepOutput(stopTrue)从而立即终止后续步骤而should_run_compliance_check作为Condition.evaluator决定是否进入合规检查分支。其实现细节是StepOutput携带stop标志位stopTrue时工作流提前结束这是条件触发 数据流控制的典型组合early_stop_loop.pyPASS在循环中提前停止日志显示 14.3s 完成early_stop_parallel.pyPASS在并行分支中提前停止early_stop_basic.pyFAIL35s 超时输出停留在路由/端点统计表附近属长时生成场景下的超时。该主题的关键 API 是agno.workflow.condition.Condition带evaluator参数与StepOutput.stop字段可用于构建合规门禁质量闸门等需要中途熔断的业务流。五、guardrails工作流内的提示注入防护相关文件guardrails/prompt_injection.py见 子目录 README。该示例在normal模式下 PASS日志关键输出为ERROR Validation failed: Potential jailbreaking or prompt injection detected.——即工作流中的护栏成功识别出提示注入/越狱攻击并拒绝了输入。这验证了 agno 工作流可以在入口处对用户输入进行安全校验将护栏Guardrails能力嵌入工作流执行链路用于拦截恶意输入后再进入后续步骤。六、history历史上下文与连续执行相关文件history/continuous_execution.py、history/history_in_function.py、history/intent_routing_with_history.py、history/step_history.py见 子目录 README。四个示例分别演示continuous_execution.py连续执行跨多轮调用维持上下文history_in_function.py在函数step executor内部访问历史intent_routing_with_history.py结合历史做意图路由step_history.py按步骤查看历史记录。测试日志中四个示例均标记 FAIL 且全部为 35s 超时输出分别停留在 Student :、Content Manager :、Im getting an error message、fit their needs 等会话片段说明这些示例属于多轮长会话场景35s 内无法完成完整演示属于外部模型响应耗时导致的超时而非导入或语法错误所有示例都能运行到产生会话内容。七、long_running长时间运行与容错恢复相关文件long_running/disruption_catchup.py、long_running/events_replay.py、long_running/websocket_reconnect.py见 子目录 README。该主题聚焦长时间运行任务的三类故障处理能力disruption_catchup.py中断后的追赶catch-up恢复演示工作流中断后如何补齐错过的执行events_replay.py事件重放将已记录的事件重新播放以恢复状态websocket_reconnect.pyWebSocket 断线重连。三个示例均以startup模式、2s 超时验证并通过输出均为Starting test in 2 seconds...表明示例被设计为启动后等待若干秒再进入正式测试流程测试框架仅确认其能正常启动。运行前提是本地 AgentOS 服务可用见文件头部注释中的服务地址。八、previous_step_outputs访问前一步骤的输出相关文件previous_step_outputs/access_previous_outputs.py见 子目录 README。该示例演示步骤间数据传递的关键机制在后续步骤的 executor 中通过StepInput.previous_step_content读取前一步骤的内容该字段在early_stop_condition.py源码中也有使用。日志中该示例为 FAIL35s 超时最后输出为 Hacker News 抓取调试行说明其执行到获取 Top 15 新闻阶段时因外部数据抓取耗时而超时。该主题的核心价值在于步骤之间的StepInput/StepOutput数据契约是 agno 工作流状态流转的基础。九、run_control运行控制的八种武器相关文件run_control/cancel_run.py、run_control/deep_copy.py、run_control/event_storage.py、run_control/executor_events.py、run_control/metrics.py、run_control/remote_workflow.py、run_control/workflow_cli.py、run_control/workflow_serialization.py见 子目录 README。这是示例最丰富、最能体现运行期控制能力的子目录8 个示例中 4 个通过cancel_run.pyPASS跨线程取消运行中的工作流。源码cancel_run.py展示了完整实现工作流在线程 A 中以流式方式运行workflow.run(..., streamTrue)线程 B 延迟若干秒后调用workflow.cancel_run(run_id)运行线程通过RunEvent.run_content接收内容、通过RunEvent.run_cancelled与WorkflowRunEvent.workflow_cancelled感知取消事件并检查 chunk 的RunStatus.completed/cancelled状态。日志输出 Workflow cancellation example completed 证实取消链路可用deep_copy.pyPASS对工作流做深拷贝输出 First Step: Draft Outline Copy 表明拷贝后的工作流实例可独立运行executor_events.pyPASS监听执行器事件日志显示Marked run 544da40f-... as RunStatus.completed即运行状态被正确标记完成remote_workflow.pyPASS远程工作流日志输出Error: Failed to connect to remote server at http://localhost:7777——示例逻辑正确执行但因本地无远程服务而报连接错误该场景需按 README 提示启动本地 AgentOS 服务workflow_serialization.pyPASS工作流序列化/保存日志显示Error saving workflow: Label serialization-demo already exists for...说明重复保存同名标签触发了预期的去重保护逻辑event_storage.py、metrics.py、workflow_cli.pyFAIL三者均在 35s 超时分别停在 OpenAI 客户端创建、指标表输出等阶段属 LLM 调用与长任务执行耗时的场景其中workflow_cli.py输出中包含 Add observability and safety: logs/metrics, error handling, retries 等建议性内容说明已运行至执行阶段。run_control主题覆盖的 API 包括Workflow.cancel_run()、RunStatus、RunEvent、WorkflowRunEvent、工作流深拷贝与序列化是构建可运维、可干预生产级工作流的核心工具箱。十、session_state跨步骤共享的可变会话状态相关文件session_state/rename_session.py、session_state/state_in_condition.py、session_state/state_in_function.py、session_state/state_in_router.py、session_state/state_with_agent.py、session_state/state_with_team.py另有job_application_tracker.py见 子目录 README。该主题演示工作流会话状态session_state的共享与变更state_with_agent.pyPASS让 Agent 通过工具读写共享会话状态。源码state_with_agent.py展示了购物清单场景工作流以dbSqliteDb(db_filetmp/workflow.db)持久化构造时传入初始状态session_state{shopping_list: []}四个工具函数add_item/remove_item/remove_all_items/list_items通过RunContext.session_state读写同一份可变字典多个 AgentShopping Assistant、List Manager在不同步骤中操作共享列表最后通过workflow.get_session_state()取回状态。日志输出Final workflow session state: {shopping_list: []}证明清空操作生效state_in_condition.pyPASS3.2s 完成在 Condition 求值中读取状态state_in_function.py、state_in_router.py、state_with_team.pyFAIL均为 35s 超时其中state_with_team.py输出 Step Write Tests not found 提示步骤引用问题rename_session.pyFAIL以退出码 1 失败报AttributeError: NoneType object has no attribute session_data属于会话数据未初始化时的空引用异常是少数代码路径本身报错的示例。该主题的核心机制是RunContext.session_state作为跨步骤、跨 Agent 的可变共享内存配合SqliteDb可实现带持久化的多轮会话状态管理。十一、structured_io结构化输入与输出相关文件structured_io/image_input.py、structured_io/input_schema.py、structured_io/pydantic_input.py、structured_io/structured_io_agent.py、structured_io/structured_io_function.py、structured_io/structured_io_team.py见 子目录 README。该主题演示工作流输入输出的结构化约束6 个示例 3 个通过structured_io_agent.pyPASSAgent 步骤的结构化输出日志除成功执行外还提示PydanticDeprecatedSince20: min_items is deprecated... use min_length这是 Pydantic v2 的兼容性警告不影响运行structured_io_function.pyPASS函数步骤的结构化输出同样仅有min_items弃用警告image_input.pyPASS22.9s 完成图像多模态输入input_schemabegin▁of▁sentenceblob、pydantic_input.py、structured_io_team.pyFAIL均为 35s 超时其中两个示例与 DuckDuckGo 搜索ddgs.exceptions.DDGSException: No results found相关属外部搜索服务无结果导致的等待。日志中反复出现的min_items弃用警告提示读者在当前仓库的 Pydantic v2 环境下定义结构化模型时应使用min_length替代已弃用的min_items。十二、tools、workflow_agent 与 run_params工具、智能体化与运行参数12.1 tools工作流即工具tools/workflow_tools.py见 子目录 README演示将整个工作流封装为工具供其他组件调用即工作流组合复用的入口。日志中该示例为 FAIL35s 超时输出停留在量子计算相关内容生成阶段属外部模型/搜索耗时。12.2 workflow_agent工作流 智能体的融合workflow_agent/basic_workflow_agent.py与workflow_agent/workflow_agent_with_condition.py见 子目录 README演示以更长的 120s 超时运行basic_workflow_agent.pyFAIL120s 超时超时前已输出TOOL METRICS调试块说明进入了工具调用统计阶段属于深度推理多工具调用场景的耗时超限workflow_agent_with_condition.pyPASS13.5s 完成带条件分支的工作流智能体正常运行13.5s 内完成全部执行。12.3 run_params工作流级运行参数run_params/目录metadata_resolution.py、workflow_all_params.py、workflow_dependencies.py专门演示工作流级别的运行参数元数据解析metadata resolution、全部参数all params与依赖关系dependencies——涉及如何在运行时传入元数据、控制上下文标志context flags以及声明步骤间的依赖。该目录未出现在测试日志条目中仅子目录自带 TEST_LOG.md可作为进阶阅读材料。十三、测试结果汇总与实战建议将主日志TEST_LOG.md中的 40 个示例按状态归类主题示例数PASSFAIL主要失败原因background_execution321后台 Agent 运行超时early_stopping431长时生成超时guardrails110—history404多轮会话 35s 超时long_running330—startup 校验previous_step_outputs101外部数据抓取超时run_control844LLM 调用 / CLI 长任务超时session_state624超时 / 会话数据空引用structured_io633搜索服务无结果、超时tools101搜索/生成超时workflow_agent211深度推理 120s 超时从中可以提炼出四条实战结论PASS/FAIL 不等于代码正确/错误。绝大多数 FAIL 是 35s 超时或外部依赖模型 API、DuckDuckGo、远程服务不可用导致的示例本身均能启动并运行到业务阶段区分运行模式startup 模式2s/8s仅验证启动能力适合服务端、长任务、断线重连类示例normal 模式才验证完整业务链路善用停止与取消控制StepOutput(stopTrue)实现数据流层面的提前终止Workflow.cancel_run(run_id)实现运行期跨线程取消两者是控制工作流生命周期的主要手段共享状态与结构化 IO 是组合复杂工作流的基础RunContext.session_state支撑跨步骤/跨 Agent 状态共享Pydantic 模型注意用min_length而非弃用的min_items保证输入输出契约。十四、进一步探索主题总览与运行前提cookbook/04_workflows/06_advanced_concepts/README.md原始测试记录cookbook/04_workflows/06_advanced_concepts/TEST_LOG.md运行参数主题cookbook/04_workflows/06_advanced_concepts/run_params/工作流基础教程cookbook/04_workflows/复制示例到自己环境中验证时请先按 README 要求激活.venvs/demo/bin/python、通过direnv allow加载 API 密钥并为 background_execution、long_running、remote_workflow 等示例准备好本地 AgentOS 服务否则会复现日志中的连接失败与超时现象。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 agno 工作流高级概念全景实战：基于 06_advanced_concepts 测试日志的 40 个示例深度解析
agno 工作流高级概念全景实战基于 06_advanced_concepts 测试日志的 40 个示例深度解析【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南围绕cookbook/04_workflows/06_advanced_concepts目录下 40 个可运行工作流示例及其测试验证日志展开系统讲解 agno 工作流Workflow的高级能力背景执行、提前停止、护栏Guardrails、历史上下文、长时间运行、运行控制、会话状态、结构化输入输出等。读者读完可以掌握每个高级主题对应的示例文件、运行方式、预期行为以及测试日志所反映的真实运行结果从而在自己的项目中准确选用并验证这些高级特性。一、文档与示例全景测试日志揭示了什么本专题的核心关联文档为 cookbook/04_workflows/06_advanced_concepts/TEST_LOG.md它是一份于 2026-02-08 生成的自动化测试日志逐条记录了该目录下 40 个示例脚本的运行验证结果。每个条目包含四个要素文件路径示例脚本的相对位置按功能主题分子目录存放状态PASS通过或 FAIL失败执行描述使用的解释器.venvs/demo/bin/python、运行模式normal正常执行或startup仅启动校验及超时时间结果执行输出摘要包括超时、报错信息或关键输出片段。按 cookbook/04_workflows/06_advanced_concepts/README.md 的说明示例共覆盖 12 个主题子目录background_execution、early_stopping、guardrails、history、long_running、previous_step_outputs、run_control、run_params、session_state、structured_io、tools、workflow_agent。运行前提是激活 demo 虚拟环境.venvs/demo/bin/python并通过direnv allow加载 API 密钥需本地.envrc文件部分示例如 background_execution、long_running、run_control 中的远程/服务类示例还要求本地 AgentOS 服务可用。从测试日志整体看提前停止、护栏、运行控制中的多数示例以及基础结构化 IO 示例均验证通过而涉及外部搜索WebSearch、DuckDuckGo、远程服务连接或 LLM 长时生成的示例则普遍出现超时或连接失败。这恰好反映了该类示例对网络、外部服务与模型响应速度的依赖而非示例代码本身的逻辑错误。二、测试方法与执行模式解读日志中所有条目均以.venvs/demo/bin/python执行这是项目 cookbook 约定的 demo 虚拟环境。执行模式分为两类理解它们才能正确解读 PASS/FAIL 语义模式说明典型超时判定含义normal完整运行示例主逻辑35sworkflow_agent 为 120s在超时内完成全部执行才算成功startup仅验证进程能否正常启动2s / 8s进程能启动、运行到预期阶段即算通过随后被强制终止例如long_running目录下的三个示例disruption_catchup.py、events_replay.py、websocket_reconnect.py均以startup模式、2s 超时验证日志输出“Startup validation only; process terminated after 2.01s. Starting test in 2 seconds...”表明它们本就是为了测试长时间运行的故障恢复而设计测试框架只确认其能正常启动。同理background_execution/websocket_server.py以 8s 超时启动校验输出INFO: Finished server process [29101]说明服务进程正常启动并退出。这一设计告诉我们判断一个工作流示例是否可用必须结合其运行模式。startup 模式通过代表进程可启动不代表完整业务流程跑通normal 模式超时则通常指向外部依赖模型 API、搜索服务、远程服务器而非代码缺陷。三、background_execution后台执行与 WebSocket相关文件background_execution/background_poll.py、background_execution/websocket_client.py、background_execution/websocket_server.py见 子目录 README。该主题演示工作流如何以后台方式执行并与外部进程通信background_poll.py演示后台轮询background poll测试日志中为 FAIL——在 35s 内超时最后输出停留在Agent Run End调试行说明 Agent 运行未在限定时间内结束websocket_client.pyPASSstartup 8s输出[ERROR] Failed to connect: Multiple exceptions: [Errno 61] Connect call failed这是典型的本机无 WebSocket 服务在监听时的连接拒绝错误macOS/Linux 的 Errno 61 即 ECONNREFUSED进程能正常启动并尝试连接即视为校验通过websocket_server.pyPASSstartup 8s作为服务端正常启动并结束输出INFO: Finished server process。三者组合构成客户端-服务端闭环服务端监听、客户端连接、轮询式后台任务。运行前需按文件头部注释启动本地 AgentOS 服务否则客户端示例会如日志所示出现连接失败。四、early_stopping提前停止的四种姿势相关文件early_stopping/early_stop_basic.py、early_stopping/early_stop_condition.py、early_stopping/early_stop_loop.py、early_stopping/early_stop_parallel.py见 子目录 README。提前停止是工作流的高级控制能力日志中 4 个示例有 3 个通过early_stop_condition.pyPASS在 Condition 分支内部停止整个工作流。结合源码early_stop_condition.py其核心机制是合规检查函数compliance_checker在发现内容含violation/illegal关键词时返回StepOutput(stopTrue)从而立即终止后续步骤而should_run_compliance_check作为Condition.evaluator决定是否进入合规检查分支。其实现细节是StepOutput携带stop标志位stopTrue时工作流提前结束这是条件触发 数据流控制的典型组合early_stop_loop.pyPASS在循环中提前停止日志显示 14.3s 完成early_stop_parallel.pyPASS在并行分支中提前停止early_stop_basic.pyFAIL35s 超时输出停留在路由/端点统计表附近属长时生成场景下的超时。该主题的关键 API 是agno.workflow.condition.Condition带evaluator参数与StepOutput.stop字段可用于构建合规门禁质量闸门等需要中途熔断的业务流。五、guardrails工作流内的提示注入防护相关文件guardrails/prompt_injection.py见 子目录 README。该示例在normal模式下 PASS日志关键输出为ERROR Validation failed: Potential jailbreaking or prompt injection detected.——即工作流中的护栏成功识别出提示注入/越狱攻击并拒绝了输入。这验证了 agno 工作流可以在入口处对用户输入进行安全校验将护栏Guardrails能力嵌入工作流执行链路用于拦截恶意输入后再进入后续步骤。六、history历史上下文与连续执行相关文件history/continuous_execution.py、history/history_in_function.py、history/intent_routing_with_history.py、history/step_history.py见 子目录 README。四个示例分别演示continuous_execution.py连续执行跨多轮调用维持上下文history_in_function.py在函数step executor内部访问历史intent_routing_with_history.py结合历史做意图路由step_history.py按步骤查看历史记录。测试日志中四个示例均标记 FAIL 且全部为 35s 超时输出分别停留在 Student :、Content Manager :、Im getting an error message、fit their needs 等会话片段说明这些示例属于多轮长会话场景35s 内无法完成完整演示属于外部模型响应耗时导致的超时而非导入或语法错误所有示例都能运行到产生会话内容。七、long_running长时间运行与容错恢复相关文件long_running/disruption_catchup.py、long_running/events_replay.py、long_running/websocket_reconnect.py见 子目录 README。该主题聚焦长时间运行任务的三类故障处理能力disruption_catchup.py中断后的追赶catch-up恢复演示工作流中断后如何补齐错过的执行events_replay.py事件重放将已记录的事件重新播放以恢复状态websocket_reconnect.pyWebSocket 断线重连。三个示例均以startup模式、2s 超时验证并通过输出均为Starting test in 2 seconds...表明示例被设计为启动后等待若干秒再进入正式测试流程测试框架仅确认其能正常启动。运行前提是本地 AgentOS 服务可用见文件头部注释中的服务地址。八、previous_step_outputs访问前一步骤的输出相关文件previous_step_outputs/access_previous_outputs.py见 子目录 README。该示例演示步骤间数据传递的关键机制在后续步骤的 executor 中通过StepInput.previous_step_content读取前一步骤的内容该字段在early_stop_condition.py源码中也有使用。日志中该示例为 FAIL35s 超时最后输出为 Hacker News 抓取调试行说明其执行到获取 Top 15 新闻阶段时因外部数据抓取耗时而超时。该主题的核心价值在于步骤之间的StepInput/StepOutput数据契约是 agno 工作流状态流转的基础。九、run_control运行控制的八种武器相关文件run_control/cancel_run.py、run_control/deep_copy.py、run_control/event_storage.py、run_control/executor_events.py、run_control/metrics.py、run_control/remote_workflow.py、run_control/workflow_cli.py、run_control/workflow_serialization.py见 子目录 README。这是示例最丰富、最能体现运行期控制能力的子目录8 个示例中 4 个通过cancel_run.pyPASS跨线程取消运行中的工作流。源码cancel_run.py展示了完整实现工作流在线程 A 中以流式方式运行workflow.run(..., streamTrue)线程 B 延迟若干秒后调用workflow.cancel_run(run_id)运行线程通过RunEvent.run_content接收内容、通过RunEvent.run_cancelled与WorkflowRunEvent.workflow_cancelled感知取消事件并检查 chunk 的RunStatus.completed/cancelled状态。日志输出 Workflow cancellation example completed 证实取消链路可用deep_copy.pyPASS对工作流做深拷贝输出 First Step: Draft Outline Copy 表明拷贝后的工作流实例可独立运行executor_events.pyPASS监听执行器事件日志显示Marked run 544da40f-... as RunStatus.completed即运行状态被正确标记完成remote_workflow.pyPASS远程工作流日志输出Error: Failed to connect to remote server at http://localhost:7777——示例逻辑正确执行但因本地无远程服务而报连接错误该场景需按 README 提示启动本地 AgentOS 服务workflow_serialization.pyPASS工作流序列化/保存日志显示Error saving workflow: Label serialization-demo already exists for...说明重复保存同名标签触发了预期的去重保护逻辑event_storage.py、metrics.py、workflow_cli.pyFAIL三者均在 35s 超时分别停在 OpenAI 客户端创建、指标表输出等阶段属 LLM 调用与长任务执行耗时的场景其中workflow_cli.py输出中包含 Add observability and safety: logs/metrics, error handling, retries 等建议性内容说明已运行至执行阶段。run_control主题覆盖的 API 包括Workflow.cancel_run()、RunStatus、RunEvent、WorkflowRunEvent、工作流深拷贝与序列化是构建可运维、可干预生产级工作流的核心工具箱。十、session_state跨步骤共享的可变会话状态相关文件session_state/rename_session.py、session_state/state_in_condition.py、session_state/state_in_function.py、session_state/state_in_router.py、session_state/state_with_agent.py、session_state/state_with_team.py另有job_application_tracker.py见 子目录 README。该主题演示工作流会话状态session_state的共享与变更state_with_agent.pyPASS让 Agent 通过工具读写共享会话状态。源码state_with_agent.py展示了购物清单场景工作流以dbSqliteDb(db_filetmp/workflow.db)持久化构造时传入初始状态session_state{shopping_list: []}四个工具函数add_item/remove_item/remove_all_items/list_items通过RunContext.session_state读写同一份可变字典多个 AgentShopping Assistant、List Manager在不同步骤中操作共享列表最后通过workflow.get_session_state()取回状态。日志输出Final workflow session state: {shopping_list: []}证明清空操作生效state_in_condition.pyPASS3.2s 完成在 Condition 求值中读取状态state_in_function.py、state_in_router.py、state_with_team.pyFAIL均为 35s 超时其中state_with_team.py输出 Step Write Tests not found 提示步骤引用问题rename_session.pyFAIL以退出码 1 失败报AttributeError: NoneType object has no attribute session_data属于会话数据未初始化时的空引用异常是少数代码路径本身报错的示例。该主题的核心机制是RunContext.session_state作为跨步骤、跨 Agent 的可变共享内存配合SqliteDb可实现带持久化的多轮会话状态管理。十一、structured_io结构化输入与输出相关文件structured_io/image_input.py、structured_io/input_schema.py、structured_io/pydantic_input.py、structured_io/structured_io_agent.py、structured_io/structured_io_function.py、structured_io/structured_io_team.py见 子目录 README。该主题演示工作流输入输出的结构化约束6 个示例 3 个通过structured_io_agent.pyPASSAgent 步骤的结构化输出日志除成功执行外还提示PydanticDeprecatedSince20: min_items is deprecated... use min_length这是 Pydantic v2 的兼容性警告不影响运行structured_io_function.pyPASS函数步骤的结构化输出同样仅有min_items弃用警告image_input.pyPASS22.9s 完成图像多模态输入input_schemabegin▁of▁sentenceblob、pydantic_input.py、structured_io_team.pyFAIL均为 35s 超时其中两个示例与 DuckDuckGo 搜索ddgs.exceptions.DDGSException: No results found相关属外部搜索服务无结果导致的等待。日志中反复出现的min_items弃用警告提示读者在当前仓库的 Pydantic v2 环境下定义结构化模型时应使用min_length替代已弃用的min_items。十二、tools、workflow_agent 与 run_params工具、智能体化与运行参数12.1 tools工作流即工具tools/workflow_tools.py见 子目录 README演示将整个工作流封装为工具供其他组件调用即工作流组合复用的入口。日志中该示例为 FAIL35s 超时输出停留在量子计算相关内容生成阶段属外部模型/搜索耗时。12.2 workflow_agent工作流 智能体的融合workflow_agent/basic_workflow_agent.py与workflow_agent/workflow_agent_with_condition.py见 子目录 README演示以更长的 120s 超时运行basic_workflow_agent.pyFAIL120s 超时超时前已输出TOOL METRICS调试块说明进入了工具调用统计阶段属于深度推理多工具调用场景的耗时超限workflow_agent_with_condition.pyPASS13.5s 完成带条件分支的工作流智能体正常运行13.5s 内完成全部执行。12.3 run_params工作流级运行参数run_params/目录metadata_resolution.py、workflow_all_params.py、workflow_dependencies.py专门演示工作流级别的运行参数元数据解析metadata resolution、全部参数all params与依赖关系dependencies——涉及如何在运行时传入元数据、控制上下文标志context flags以及声明步骤间的依赖。该目录未出现在测试日志条目中仅子目录自带 TEST_LOG.md可作为进阶阅读材料。十三、测试结果汇总与实战建议将主日志TEST_LOG.md中的 40 个示例按状态归类主题示例数PASSFAIL主要失败原因background_execution321后台 Agent 运行超时early_stopping431长时生成超时guardrails110—history404多轮会话 35s 超时long_running330—startup 校验previous_step_outputs101外部数据抓取超时run_control844LLM 调用 / CLI 长任务超时session_state624超时 / 会话数据空引用structured_io633搜索服务无结果、超时tools101搜索/生成超时workflow_agent211深度推理 120s 超时从中可以提炼出四条实战结论PASS/FAIL 不等于代码正确/错误。绝大多数 FAIL 是 35s 超时或外部依赖模型 API、DuckDuckGo、远程服务不可用导致的示例本身均能启动并运行到业务阶段区分运行模式startup 模式2s/8s仅验证启动能力适合服务端、长任务、断线重连类示例normal 模式才验证完整业务链路善用停止与取消控制StepOutput(stopTrue)实现数据流层面的提前终止Workflow.cancel_run(run_id)实现运行期跨线程取消两者是控制工作流生命周期的主要手段共享状态与结构化 IO 是组合复杂工作流的基础RunContext.session_state支撑跨步骤/跨 Agent 状态共享Pydantic 模型注意用min_length而非弃用的min_items保证输入输出契约。十四、进一步探索主题总览与运行前提cookbook/04_workflows/06_advanced_concepts/README.md原始测试记录cookbook/04_workflows/06_advanced_concepts/TEST_LOG.md运行参数主题cookbook/04_workflows/06_advanced_concepts/run_params/工作流基础教程cookbook/04_workflows/复制示例到自己环境中验证时请先按 README 要求激活.venvs/demo/bin/python、通过direnv allow加载 API 密钥并为 background_execution、long_running、remote_workflow 等示例准备好本地 AgentOS 服务否则会复现日志中的连接失败与超时现象。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考