Agno Function Workflow 实战指南用单个可调用函数编排多智能体工作流【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本指南以 cookbook/04_workflows/01_basic_workflows/03_function_workflows 目录下的演示为核心讲解 Agnoagno工作流Workflow中一种独特的编排形态——函数工作流Function Workflow不再把步骤声明成显式的Step对象列表而是将一个可调用函数整体赋给Workflow(steps...)让编排逻辑完全收拢在普通 Python 函数里。读完本文你将掌握函数工作流的函数签名约束、WorkflowExecutionInput数据结构以及同步 / 流式 / 异步 / 异步流式四种运行模式的写法并理解其在Workflow源码层的实现机制。一、什么是 Function Workflow在 Agno 的 cookbook 工作流体系中01_basic_workflows 目录之下并行组织了三种基础的流程形态01_sequence_of_steps/以Step/Steps对象列表串联的顺序执行工作流02_step_with_function/在单个步骤内使用函数类型执行器03_function_workflows/整体用一个执行函数替代步骤列表即本节主题。该目录的 README.md 对范围的定义非常简洁——Runnable workflow examples under 03_function_workflows并指明唯一示例文件function_workflow.py的作用是Demonstrates function workflow。换句话说这个示例的实质是Workflow的steps参数除了接受Step列表、Steps容器之外还接受单个可调用对象callable当传入可调用对象时每次运行工作流都会调用这个函数由函数内部自行完成对 Agent、Team 乃至其他 Workflow 的编排。在 Workflow 实现 中可以看到steps字段的声明约第 623 行而在运行参数解析、序列化等多个逻辑分支里都专门针对callable(self.steps)做了区分处理如run_parameters属性、状态字典序列化等可以确认可调用 steps是 Workflow 一等公民的运行方式。二、运行环境与前置条件README.md 明确给出了两个前置条件这也是整个 cookbook 的运行约定激活 demo 虚拟环境仓库统一使用.venvs/demo/bin/python运行示例该环境由仓库根目录 scripts/demo_setup.sh 之类的脚本创建加载 API Key使用direnv allow放行本地.envrc文件为示例中的模型与联网工具注入密钥。运行示例# 在仓库根目录执行 .venvs/demo/bin/python cookbook/04_workflows/01_basic_workflows/03_function_workflows/function_workflow.py脚本在if __name__ __main__:块中会依次跑完四种模式同步、同步流式、异步、异步流式因此一次执行即可观察到函数工作流在不同运行模式下的完整行为。三、示例全景一个内容策划流水线function_workflow.py 构造了一个研究 → 策划两阶段内容生产流水线其组件布局如下组件类型职责hackernews_agentAgent使用HackerNewsTools()从 Hacker News 帖子中提炼洞察web_agentAgent使用WebSearchTools()检索最新资讯与趋势research_teamTeam成员为上述两个 Agent统一做技术主题调研content_plannerAgent依据调研结果生成 4 周内容排期streaming_hackernews_agentAgent流式模式专用的调研 Agent组装方式如下模型均使用OpenAIChat示例中取idgpt-5.6-luna/gpt-5.2读者按自己 API Key 可用模型替换即可research_team Team( nameResearch Team, members[hackernews_agent, web_agent], instructionsResearch tech topics from Hackernews and the web, )四个Workflow实例共享同一套数据库与步骤函数但各自绑定一种模式的执行函数sync_workflow Workflow( nameContent Creation Workflow, descriptionAutomated content creation from blog posts to social media, dbSqliteDb(session_tableworkflow_session, db_filetmp/workflow.db), stepscustom_execution_function, )这里值得注意两点steps直接接收函数引用而非步骤列表这正是函数工作流的命名由来dbSqliteDb(session_table..., db_file...)让工作流运行数据持久化到本地 SQLite为函数工作流提供会话与断点恢复能力。四、执行函数签名与 WorkflowExecutionInput函数工作流的执行函数遵循统一签名约定。以示例中的同步版本为例def custom_execution_function( workflow: Workflow, execution_input: WorkflowExecutionInput, ) - str: print(fExecuting workflow: {workflow.name}) run_response research_team.run(execution_input.input) research_content run_response.content planning_prompt f...Core Topic: {execution_input.input} Research Results: {research_content[:500]}... content_plan content_planner.run(planning_prompt) return content_plan.content4.1 两个固定参数函数前两个位置参数由 Workflow 自动注入workflow当前Workflow实例可读取name等配置execution_inputWorkflowExecutionInput类型封装本次运行的全部输入。从源码层的run_parameters属性可以看到框架在解析执行函数签名时会显式剔除workflow、execution_input以及self这三个参数名把它们当作框架保留参数其余参数则被识别为工作流的运行参数见 workflow.py 第 866-884 行附近的signature(self.steps)内省逻辑。这意味着你可以在执行函数里声明额外参数由调用方在运行工作流时按名传入实现参数化。4.2 WorkflowExecutionInput 的字段WorkflowExecutionInput定义于 agno/workflow/types.py数据类约从第 248 行开始核心字段包括字段类型说明inputstr/Dict/List/BaseModel工作流主输入示例中即AI trends in 2024additional_dataDict[str, Any]附加上下文数据images/videos/audio/files各媒体类型列表多模态输入载体它还提供了get_input_as_string()方法能将字符串、Pydantic 模型、字典或列表统一序列化为字符串方便你在函数内部把输入拼进 Prompt。对比同一文件中的StepInput面向显式步骤携带previous_step_outputs、workflow_session等跨步骤数据可以看出函数工作流的执行函数是黑盒式的——它只接收本次运行的输入步骤间的数据流转完全由函数内部自行组织这正是该形态灵活性的来源。五、四种运行模式sync / stream / async / async-stream示例的进阶价值在于用四个形态相近的执行函数覆盖了 Agno 支持的全部运行模式。逐一拆解5.1 同步sync返回值str内部依次调用research_team.run()与content_planner.run()def custom_execution_function(workflow, execution_input) - str: ...工作流侧用sync_workflow.print_response(inputAI trends in 2024)触发。5.2 同步流式sync streaming返回类型标注为Iterator关键差异在内部def custom_execution_function_stream(workflow, execution_input) - Iterator: research_content for response in streaming_hackernews_agent.run( execution_input.input, streamTrue, stream_eventsTrue, ): if hasattr(response, content) and response.content: research_content str(response.content) ... yield from content_planner.run(planning_prompt, streamTrue, stream_eventsTrue)要点有二上游 Agent 开启streamTrue, stream_eventsTrue逐条消费事件把content累加为完整调研文本后再构造下游 Prompt这也是它单独准备streaming_hackernews_agent的原因末尾用yield from把content_planner.run(..., streamTrue)的事件流原样透传给调用方。工作流侧通过sync_stream_workflow.print_response(input..., streamTrue)触发逐块输出。5.3 异步async函数以async def定义返回str内部使用await content_planner.arun(planning_prompt)async def custom_execution_function_async(workflow, execution_input) - str: ... content_plan await content_planner.arun(planning_prompt) return content_plan.content运行侧需要用事件循环驱动asyncio.run(async_workflow.aprint_response(inputAI trends in 2024))5.4 异步流式async streaming返回类型为AsyncIterator上游用async for消费arun(..., streamTrue, stream_eventsTrue)下游同样async for逐条yieldasync def custom_execution_function_async_stream(workflow, execution_input) - AsyncIterator: async for response in streaming_hackernews_agent.arun(...): if hasattr(response, content) and response.content: research_content str(response.content) ... async for response in content_planner.arun(planning_prompt, streamTrue, stream_eventsTrue): yield response运行侧asyncio.run(async_stream_workflow.aprint_response(inputAI trends in 2024, streamTrue))5.5 模式对照速查模式函数形态返回类型内部调用工作流侧触发Syncdefstr.run()print_response()Sync StreamingdefIterator.run(streamTrue)yield fromprint_response(streamTrue)Asyncasync defstrawait .arun()asyncio.run(aprint_response())Async Streamingasync defAsyncIteratorasync foryieldasyncio.run(aprint_response(streamTrue))规律非常清晰同步用run/print_response异步加a前缀变成arun/aprint_response要流式就传streamTrue并把返回类型切换为对应的Iterator/AsyncIterator。示例中四个 Workflow 均命名为 Content Creation Workflow 且复用同一 SQLite 会话表进一步演示了同一业务在多运行形态下的平行复用。六、执行流程剖析一次运行内发生了什么把custom_execution_function的代码与调用链对齐一次同步运行的完整流程是用户在终端调用sync_workflow.print_response(inputAI trends in 2024)Workflow 构造WorkflowExecutionInput(inputAI trends in 2024)并调用执行函数steps指向的可调用对象执行函数把research_teamHackerNews Agent Web Agent当作第一步run()返回的content作为中间产物代码对research_content做[:500]截断后嵌入精心构造的planning_prompt——这是控制上下文窗口、约束下游输入规模的实用技巧执行函数再驱动content_planner产出最终内容排期并return content_plan.contentWorkflow 把返回值包装为运行结果同时借助配置的SqliteDb将会话写入tmp/workflow.db。在流式变体中第 3、5 步换为事件流逐条累计/透传中间产物仍是纯字符串拼接整体编排语义保持一致。可见函数工作流把步骤这一概念彻底函数化没有显式的Step生命周期、没有步骤间的输入传递管线一切顺序、分支、截断、Prompt 拼接都由你写在函数体里获得最大的表达自由度。七、函数工作流的适用场景与边界结合源码结构可以做如下归纳属于从代码与示例得出的推断性结论适合函数工作流的场景编排逻辑高度定制、不便拆成标准Step的流程希望把调研团队 规划 Agent这类已有 Agent / Team 组合直接封装为可复用流水线的场景需要以统一函数接口同时对外提供 sync / async / streaming 多形态 API 的服务层。需要注意的边界函数工作流内部对 Agent / Team 的调度如并行、重试、循环需要自行在函数内实现它不享受显式步骤容器的调度便利后者对应Steps/Parallel/Loop/Condition等组件如果业务需要基于步骤粒度做会话恢复、指标聚合与人工审批HITL更适合回到显式步骤路线函数工作流的持久化粒度在整个函数这一层。八、进一步探索示例源码function_workflow.py完整对照阅读四种执行函数写法同目录说明03_function_workflows/README.md相邻形态01_sequence_of_steps显式步骤序列、02_step_with_function步骤内函数可用于对比理解三种流程定义风格的差异核心数据结构agno/workflow/types.pyWorkflowExecutionInput等定义底层调度实现agno/workflow/workflow.pysteps参数对可调用对象的解析、签名内省与运行分发。函数工作流是 Agno 工作流体系中以代码为编排核心的代表性写法。它牺牲了显式步骤带来的结构化能力换来的是完全自由的函数体编排与极低的抽象负担——当你手中的 Agent、Team 已经足够聪明而流程本身又不足以复杂到需要一张步骤图时它就是最贴合直觉的选择。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Agno Function Workflow 实战指南：用单个可调用函数编排多智能体工作流
Agno Function Workflow 实战指南用单个可调用函数编排多智能体工作流【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本指南以 cookbook/04_workflows/01_basic_workflows/03_function_workflows 目录下的演示为核心讲解 Agnoagno工作流Workflow中一种独特的编排形态——函数工作流Function Workflow不再把步骤声明成显式的Step对象列表而是将一个可调用函数整体赋给Workflow(steps...)让编排逻辑完全收拢在普通 Python 函数里。读完本文你将掌握函数工作流的函数签名约束、WorkflowExecutionInput数据结构以及同步 / 流式 / 异步 / 异步流式四种运行模式的写法并理解其在Workflow源码层的实现机制。一、什么是 Function Workflow在 Agno 的 cookbook 工作流体系中01_basic_workflows 目录之下并行组织了三种基础的流程形态01_sequence_of_steps/以Step/Steps对象列表串联的顺序执行工作流02_step_with_function/在单个步骤内使用函数类型执行器03_function_workflows/整体用一个执行函数替代步骤列表即本节主题。该目录的 README.md 对范围的定义非常简洁——Runnable workflow examples under 03_function_workflows并指明唯一示例文件function_workflow.py的作用是Demonstrates function workflow。换句话说这个示例的实质是Workflow的steps参数除了接受Step列表、Steps容器之外还接受单个可调用对象callable当传入可调用对象时每次运行工作流都会调用这个函数由函数内部自行完成对 Agent、Team 乃至其他 Workflow 的编排。在 Workflow 实现 中可以看到steps字段的声明约第 623 行而在运行参数解析、序列化等多个逻辑分支里都专门针对callable(self.steps)做了区分处理如run_parameters属性、状态字典序列化等可以确认可调用 steps是 Workflow 一等公民的运行方式。二、运行环境与前置条件README.md 明确给出了两个前置条件这也是整个 cookbook 的运行约定激活 demo 虚拟环境仓库统一使用.venvs/demo/bin/python运行示例该环境由仓库根目录 scripts/demo_setup.sh 之类的脚本创建加载 API Key使用direnv allow放行本地.envrc文件为示例中的模型与联网工具注入密钥。运行示例# 在仓库根目录执行 .venvs/demo/bin/python cookbook/04_workflows/01_basic_workflows/03_function_workflows/function_workflow.py脚本在if __name__ __main__:块中会依次跑完四种模式同步、同步流式、异步、异步流式因此一次执行即可观察到函数工作流在不同运行模式下的完整行为。三、示例全景一个内容策划流水线function_workflow.py 构造了一个研究 → 策划两阶段内容生产流水线其组件布局如下组件类型职责hackernews_agentAgent使用HackerNewsTools()从 Hacker News 帖子中提炼洞察web_agentAgent使用WebSearchTools()检索最新资讯与趋势research_teamTeam成员为上述两个 Agent统一做技术主题调研content_plannerAgent依据调研结果生成 4 周内容排期streaming_hackernews_agentAgent流式模式专用的调研 Agent组装方式如下模型均使用OpenAIChat示例中取idgpt-5.6-luna/gpt-5.2读者按自己 API Key 可用模型替换即可research_team Team( nameResearch Team, members[hackernews_agent, web_agent], instructionsResearch tech topics from Hackernews and the web, )四个Workflow实例共享同一套数据库与步骤函数但各自绑定一种模式的执行函数sync_workflow Workflow( nameContent Creation Workflow, descriptionAutomated content creation from blog posts to social media, dbSqliteDb(session_tableworkflow_session, db_filetmp/workflow.db), stepscustom_execution_function, )这里值得注意两点steps直接接收函数引用而非步骤列表这正是函数工作流的命名由来dbSqliteDb(session_table..., db_file...)让工作流运行数据持久化到本地 SQLite为函数工作流提供会话与断点恢复能力。四、执行函数签名与 WorkflowExecutionInput函数工作流的执行函数遵循统一签名约定。以示例中的同步版本为例def custom_execution_function( workflow: Workflow, execution_input: WorkflowExecutionInput, ) - str: print(fExecuting workflow: {workflow.name}) run_response research_team.run(execution_input.input) research_content run_response.content planning_prompt f...Core Topic: {execution_input.input} Research Results: {research_content[:500]}... content_plan content_planner.run(planning_prompt) return content_plan.content4.1 两个固定参数函数前两个位置参数由 Workflow 自动注入workflow当前Workflow实例可读取name等配置execution_inputWorkflowExecutionInput类型封装本次运行的全部输入。从源码层的run_parameters属性可以看到框架在解析执行函数签名时会显式剔除workflow、execution_input以及self这三个参数名把它们当作框架保留参数其余参数则被识别为工作流的运行参数见 workflow.py 第 866-884 行附近的signature(self.steps)内省逻辑。这意味着你可以在执行函数里声明额外参数由调用方在运行工作流时按名传入实现参数化。4.2 WorkflowExecutionInput 的字段WorkflowExecutionInput定义于 agno/workflow/types.py数据类约从第 248 行开始核心字段包括字段类型说明inputstr/Dict/List/BaseModel工作流主输入示例中即AI trends in 2024additional_dataDict[str, Any]附加上下文数据images/videos/audio/files各媒体类型列表多模态输入载体它还提供了get_input_as_string()方法能将字符串、Pydantic 模型、字典或列表统一序列化为字符串方便你在函数内部把输入拼进 Prompt。对比同一文件中的StepInput面向显式步骤携带previous_step_outputs、workflow_session等跨步骤数据可以看出函数工作流的执行函数是黑盒式的——它只接收本次运行的输入步骤间的数据流转完全由函数内部自行组织这正是该形态灵活性的来源。五、四种运行模式sync / stream / async / async-stream示例的进阶价值在于用四个形态相近的执行函数覆盖了 Agno 支持的全部运行模式。逐一拆解5.1 同步sync返回值str内部依次调用research_team.run()与content_planner.run()def custom_execution_function(workflow, execution_input) - str: ...工作流侧用sync_workflow.print_response(inputAI trends in 2024)触发。5.2 同步流式sync streaming返回类型标注为Iterator关键差异在内部def custom_execution_function_stream(workflow, execution_input) - Iterator: research_content for response in streaming_hackernews_agent.run( execution_input.input, streamTrue, stream_eventsTrue, ): if hasattr(response, content) and response.content: research_content str(response.content) ... yield from content_planner.run(planning_prompt, streamTrue, stream_eventsTrue)要点有二上游 Agent 开启streamTrue, stream_eventsTrue逐条消费事件把content累加为完整调研文本后再构造下游 Prompt这也是它单独准备streaming_hackernews_agent的原因末尾用yield from把content_planner.run(..., streamTrue)的事件流原样透传给调用方。工作流侧通过sync_stream_workflow.print_response(input..., streamTrue)触发逐块输出。5.3 异步async函数以async def定义返回str内部使用await content_planner.arun(planning_prompt)async def custom_execution_function_async(workflow, execution_input) - str: ... content_plan await content_planner.arun(planning_prompt) return content_plan.content运行侧需要用事件循环驱动asyncio.run(async_workflow.aprint_response(inputAI trends in 2024))5.4 异步流式async streaming返回类型为AsyncIterator上游用async for消费arun(..., streamTrue, stream_eventsTrue)下游同样async for逐条yieldasync def custom_execution_function_async_stream(workflow, execution_input) - AsyncIterator: async for response in streaming_hackernews_agent.arun(...): if hasattr(response, content) and response.content: research_content str(response.content) ... async for response in content_planner.arun(planning_prompt, streamTrue, stream_eventsTrue): yield response运行侧asyncio.run(async_stream_workflow.aprint_response(inputAI trends in 2024, streamTrue))5.5 模式对照速查模式函数形态返回类型内部调用工作流侧触发Syncdefstr.run()print_response()Sync StreamingdefIterator.run(streamTrue)yield fromprint_response(streamTrue)Asyncasync defstrawait .arun()asyncio.run(aprint_response())Async Streamingasync defAsyncIteratorasync foryieldasyncio.run(aprint_response(streamTrue))规律非常清晰同步用run/print_response异步加a前缀变成arun/aprint_response要流式就传streamTrue并把返回类型切换为对应的Iterator/AsyncIterator。示例中四个 Workflow 均命名为 Content Creation Workflow 且复用同一 SQLite 会话表进一步演示了同一业务在多运行形态下的平行复用。六、执行流程剖析一次运行内发生了什么把custom_execution_function的代码与调用链对齐一次同步运行的完整流程是用户在终端调用sync_workflow.print_response(inputAI trends in 2024)Workflow 构造WorkflowExecutionInput(inputAI trends in 2024)并调用执行函数steps指向的可调用对象执行函数把research_teamHackerNews Agent Web Agent当作第一步run()返回的content作为中间产物代码对research_content做[:500]截断后嵌入精心构造的planning_prompt——这是控制上下文窗口、约束下游输入规模的实用技巧执行函数再驱动content_planner产出最终内容排期并return content_plan.contentWorkflow 把返回值包装为运行结果同时借助配置的SqliteDb将会话写入tmp/workflow.db。在流式变体中第 3、5 步换为事件流逐条累计/透传中间产物仍是纯字符串拼接整体编排语义保持一致。可见函数工作流把步骤这一概念彻底函数化没有显式的Step生命周期、没有步骤间的输入传递管线一切顺序、分支、截断、Prompt 拼接都由你写在函数体里获得最大的表达自由度。七、函数工作流的适用场景与边界结合源码结构可以做如下归纳属于从代码与示例得出的推断性结论适合函数工作流的场景编排逻辑高度定制、不便拆成标准Step的流程希望把调研团队 规划 Agent这类已有 Agent / Team 组合直接封装为可复用流水线的场景需要以统一函数接口同时对外提供 sync / async / streaming 多形态 API 的服务层。需要注意的边界函数工作流内部对 Agent / Team 的调度如并行、重试、循环需要自行在函数内实现它不享受显式步骤容器的调度便利后者对应Steps/Parallel/Loop/Condition等组件如果业务需要基于步骤粒度做会话恢复、指标聚合与人工审批HITL更适合回到显式步骤路线函数工作流的持久化粒度在整个函数这一层。八、进一步探索示例源码function_workflow.py完整对照阅读四种执行函数写法同目录说明03_function_workflows/README.md相邻形态01_sequence_of_steps显式步骤序列、02_step_with_function步骤内函数可用于对比理解三种流程定义风格的差异核心数据结构agno/workflow/types.pyWorkflowExecutionInput等定义底层调度实现agno/workflow/workflow.pysteps参数对可调用对象的解析、签名内省与运行分发。函数工作流是 Agno 工作流体系中以代码为编排核心的代表性写法。它牺牲了显式步骤带来的结构化能力换来的是完全自由的函数体编排与极低的抽象负担——当你手中的 Agent、Team 已经足够聪明而流程本身又不足以复杂到需要一张步骤图时它就是最贴合直觉的选择。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考