LlamaIndex 工作流 并发执行

除了循环和分支之外，工作流还可以并发地执行步骤。当你有多个可以相互独立运行的步骤，并且这些步骤中包含需要等待的耗时操作时，这种并发执行的方式就非常有用，因为它允许其他步骤并行运行。

触发多个事件

到目前为止，在我们的示例中，每个步骤只触发了一个事件。但在很多情况下，你可能希望并行执行多个步骤。要做到这一点，你需要触发多个事件。你可以通过 send_event 来实现这一点：

import asyncio
import randomfrom llama_index.core.workflow import Workflow, step, Context, StartEvent, Event, StopEvent
from llama_index.utils.workflow import draw_all_possible_flowsclass StepTowEvent(Event):payload: strquery: strclass ParallelFlow(Workflow):@stepasync def step_one(self, ctx: Context, ev: StartEvent)-> StepTowEvent:print(ev.payload)ctx.send_event(StepTowEvent(payload="跳转到第二步", query="查询 1"))ctx.send_event(StepTowEvent(payload="跳转到第二步", query="查询 2"))ctx.send_event(StepTowEvent(payload="跳转到第二步", query="查询 3"))return None@step(num_workers=4)async def step_tow(self, ctx: Context, ev: StepTowEvent) -> StopEvent:print(ev.query)await asyncio.sleep(random.randint(1, 5))return StopEvent(result="结束！")async def run_workflow():w = ParallelFlow(timeout=60, verbose=True)result = await w.run(payload="开始工作流...")print(result)if __name__ == '__main__':asyncio.run(run_workflow())# 工作流可视化工具
draw_all_possible_flows(ParallelFlow, filename="parallel_workflow.html")

运行结果：

Running step step_one
开始工作流...
Step step_one produced no event
Running step step_tow
查询 1
Running step step_tow
查询 2
Running step step_tow
查询 3
Step step_tow produced event StopEvent
结束！
parallel_workflow.html

流程图：

在这个示例中，我们的起始步骤触发了三个 StepTwoEvents。step_two 步骤使用了 num_workers=4 进行装饰，这告诉工作流最多可以同时运行 4 个该步骤的实例（这也是默认值）。

收集事件

如果你执行前面的示例，你会发现工作流会在第一个完成的查询后停止。有时候这种行为是有用的，但在其他情况下，你可能希望在继续执行下一步之前，等待所有耗时操作都完成。你可以通过 collect_events 来实现这一点：

import asyncio
import randomfrom llama_index.core.workflow import Workflow, Event, step, StartEvent, Context, StopEventfrom llama_index.utils.workflow import draw_all_possible_flowsclass StepTwoEvent(Event):payload: strquery: strclass StepThreeEvent(Event):payload: strclass ConcurrentFlow(Workflow):@stepasync def step_one(self, ctx: Context, ev: StartEvent)-> StepTwoEvent:print(ev.payload)ctx.send_event(StepTwoEvent(payload="", query="查询 1"))ctx.send_event(StepTwoEvent(payload="", query="查询 2"))ctx.send_event(StepTwoEvent(payload="", query="查询 3"))return None@step(num_workers=4)async def step_two(self, ctx: Context, ev: StepTwoEvent)-> StepThreeEvent:print("第二步...", ev.query)await asyncio.sleep(random.randint(1, 5))return StepThreeEvent(payload="来自第二步")@stepasync def step_three(self, ctx: Context, ev: StepThreeEvent)-> StopEvent:print("运行第三步...")# 等待直到我们接收到 3 个事件result = ctx.collect_events(ev, [StepThreeEvent] * 3)if result is None:return None# 将这三个结果一并进行处理print("将这三个结果一并进行处理", result)return StopEvent(result="Done")async def run_workflow():w = ConcurrentFlow(timeout=60, verbose=True)result = await w.run(payload="开始工作流..")print(result)if __name__ == '__main__':asyncio.run(run_workflow())# 工作流可视化工具
draw_all_possible_flows(ConcurrentFlow, filename="concurrent_workflow.html")

运行结果：

Running step step_one
开始工作流..
Step step_one produced no event
Running step step_two
第二步... 查询 1
Running step step_two
第二步... 查询 2
Running step step_two
第二步... 查询 3
Step step_two produced event StepThreeEvent
Step step_two produced event StepThreeEvent
Step step_two produced event StepThreeEvent
Running step step_three
运行第三步...
Step step_three produced no event
Running step step_three
运行第三步...
Step step_three produced no event
Running step step_three
运行第三步...
将这三个结果一并进行处理 [StepThreeEvent(payload='来自第二步'), StepThreeEvent(payload='来自第二步'), StepThreeEvent(payload='来自第二步')]
Step step_three produced event StopEvent
Done
concurrent_workflow.html

流程图：

collect_events 方法位于 Context 上，它接收触发当前步骤的事件以及一个需要等待的事件类型数组作为参数。在本例中，我们正在等待 3 个相同类型的 StepThreeEvent 事件。

每当接收到一个 StepThreeEvent 事件时，step_three 步骤就会被触发，但 collect_events 在接收到全部 3 个事件之前会一直返回 None。一旦收集到所有 3 个事件，该步骤将继续执行，并可以将这三个结果一并进行处理。

collect_events 返回的结果是一个按接收顺序排列的事件数组，其中包含所收集到的所有事件。

多种事件类型

当然，你并不要求必须等待相同类型的事件。你可以根据需要等待任意组合的事件，例如在以下示例中：

import asynciofrom llama_index.core.workflow import Event, Workflow, step, Context, StartEvent, StopEvent
from llama_index.utils.workflow import draw_all_possible_flowsclass StepAEvent(Event):query: str
class StepBEvent(Event):query: str
class StepCEvent(Event):query: strclass StepACompleteEvent(Event):query: str
class StepBCompleteEvent(Event):query: str
class StepCCompleteEvent(Event):query: strclass ConcurrentFlow(Workflow):@stepasync def step_one(self, ctx: Context, ev: StartEvent)-> StepAEvent | StepBEvent | StepCEvent:print(ev.payload)ctx.send_event(StepAEvent(query="A 查询..."))ctx.send_event(StepBEvent(query="B 查询..."))ctx.send_event(StepCEvent(query="C 查询..."))return None@stepasync def step_a(self, ctx: Context, ev: StepAEvent)-> StepACompleteEvent:print(ev.query)return StepACompleteEvent(query="A 查询...完成")@stepasync def step_b(self, ctx: Context, ev: StepBEvent)-> StepBCompleteEvent:print(ev.query)return StepBCompleteEvent(query="B 查询...完成")@stepasync def step_c(self, ctx: Context, ev: StepCEvent)-> StepCCompleteEvent:print(ev.query)return StepCCompleteEvent(query="C 查询...完成")@stepasync def step_tow(self,ctx: Context,ev: StepACompleteEvent | StepBCompleteEvent | StepCCompleteEvent,) -> StopEvent:print("接收到的事件 ", ev.query)# 等待直到我们接收到 3 个事件result = ctx.collect_events(ev, [StepCCompleteEvent, StepACompleteEvent, StepBCompleteEvent],)if result is None:return None# do something with all 3 results togetherreturn StopEvent(result="Done")async def run_workflow():w = ConcurrentFlow(timeout=60, verbose=True)result = await w.run(payload="开始工作流..")print(result)if __name__ == '__main__':asyncio.run(run_workflow())# 工作流可视化工具
draw_all_possible_flows(ConcurrentFlow, filename="concurrent_workflow.html")

运行结果：

Running step step_one
开始工作流..
Step step_one produced no event
Running step step_a
A 查询...
Step step_a produced event StepACompleteEvent
Running step step_b
B 查询...
Step step_b produced event StepBCompleteEvent
Running step step_c
C 查询...
Step step_c produced event StepCCompleteEvent
Running step step_tow
接收到的事件  A 查询...完成
Step step_tow produced no event
Running step step_tow
接收到的事件  B 查询...完成
Step step_tow produced no event
Running step step_tow
接收到的事件  C 查询...完成
Step step_tow produced event StopEvent
Done
concurrent_workflow.html

流程图：

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.pswp.cn/web/83671.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

精粹汇总：大厂编程规范（持续更新）

精粹汇总：大厂编程规范（持续更新）

欢迎来到啾啾的博客🐱。 记录学习点滴。分享工作思考和实用技巧，偶尔也分享一些杂谈💬。 有很多很多不足的地方，欢迎评论交流，感谢您的阅读和评论😄。 目录 1 引言2 并发控制 (Concurrency Control)3 事务控…
阅读更多...
curl 检查重定向的命令总结

curl 检查重定向的命令总结

查看是否发生了重定向： curl -I http://yourdomain.com跟踪整个重定向链： curl -IL http://yourdomain.com禁止跳转，检查是否返回 301/302： curl -I --max-redirs 0 http://yourdomain.com如果你只想看跳没跳 HTTPS&#xff0c…
阅读更多...
STM32 Bootloader：使用文件头加载并启动应用程序

STM32 Bootloader：使用文件头加载并启动应用程序

文章目录 STM32 Bootloader：使用文件头加载并启动应用程序的完整解析一、系统整体流程二、镜像头结构 image\_header\_t三、Bootloader 主函数流程1. 初始化 UART2. 调用启动函数3. 拷贝 APP 并跳转启动 四、跳转执行 APP 的实现五、总结与扩展思路 明白了&#xff…
阅读更多...
无外接物理显示器的Ubuntu系统的远程桌面连接(升级版)

无外接物理显示器的Ubuntu系统的远程桌面连接(升级版)

文章目录 操作步骤实践截图配置 Xorg 的虚拟显示界面(升级版) 操作步骤 “远程连接”,在设置里直接打开就可以.进行配置就行. 1.配置 GRUB 以支持无显示器启动 sudo nano /etc/default/grub (里面有一行改为: GRUB_CMDLINE_LINUX_DEFAULT"quiet splash videovesa:off vi…
阅读更多...
ACCU-100安科瑞协调控制器：精准调控光伏逆变器

ACCU-100安科瑞协调控制器：精准调控光伏逆变器

产品概述 ACCU-100微电网协调控制器是一款应用于微电网、分布式发电、储能等领域的智能协调控制器。它能接入光伏系统、风力发电、储能系统以及充电桩等设备，通过对微电网系统进行数据采集与分析，实时监控各类设备的运行状态和健康状况。在此基础上&…
阅读更多...
长春光博会 | 麒麟信安：构建工业数字化安全基座，赋能智能制造转型升级

长春光博会 | 麒麟信安：构建工业数字化安全基座，赋能智能制造转型升级

6月10日-13日，2025长春国际光电博览会Light国际会议（简称长春光博会）在长春东北亚国际博览中心盛大举行，吉林省委书记黄强出席并宣布开幕，省委副书记、省长胡玉亭致辞。本届大会聚焦光电信息领域的前沿技术和最新产品&…
阅读更多...
书写时垂直笔画比水平笔画表现更好的心理机制分析

书写时垂直笔画比水平笔画表现更好的心理机制分析

你有写字的时候总是垂直方向笔画好写，水平方向的笔画不好写的情况存在吗？ 书写时垂直笔画比水平笔画表现更好的心理机制分析 从人类认知和行为模式的角度来理解这种现象。以下是深度心理分析： 核心心理动因 重力知觉内化： 垂直…
阅读更多...
SpringAI使用总结

SpringAI使用总结

SpringAI使用总结 基本使用ChatModel和ChatClient简单对话流式输出预设角色prompt（提示词）function call（工具调用）参考 基本使用 ChatModel和ChatClient SpringAi支持非常多的模型，为了统一处理，SpringA…
阅读更多...
历史交易数据涨跌分级

历史交易数据涨跌分级

历史交易数据涨跌分级 # encoding:utf-8 import sys,traceback from loguru import loggersys.path.append("..") from QhSpiderTool import QhDorpFiled from QhCsvMode import *def QhZhangDieFenJi(QhDfData,QhFangFa"A"):"""历史交易数…
阅读更多...
Kafka入门4.0.0版本（基于Java、SpringBoot操作）

Kafka入门4.0.0版本（基于Java、SpringBoot操作）

Kafka入门4.0.0版本（基于Java、SpringBoot操作） 一、kafka概述 Kafka最初是由LinkedIn公司开发的，是一个高可靠、高吞吐量、低延迟的分布式发布订阅消息系统，它使用Scala语言编写，并于2010年被贡献给了Apache基金会&…
阅读更多...
react react-router-dom中获取自定义参数v6.4版本之后

react react-router-dom中获取自定义参数v6.4版本之后

路由配置, AutnToken 组件作为权限、登录管理 import { createBrowserRouter, Navigate } from react-router-dom; import Layout from /layout/index; import Login from /pages/login; import Page404 from /pages/404;import AutnToken from /components/authToken; import…
阅读更多...
AI中的Prompt

AI中的Prompt

1. System 作用：设定 AI 的“角色设定”和“行为准则”。 内容：通常是描述 LLM 的身份、语气、行为范围、约束规则。 类似：在大语言模型中是最优先被考虑的提示。 示例： 你是一个专业的商品评价分析助手，请根据用户…
阅读更多...
从人工到智能：IACheck如何重构检测报告审核工作流？

从人工到智能：IACheck如何重构检测报告审核工作流？

从人工到智能：IACheck如何重构检测报告审核工作流？ 在当今AI技术迅猛发展的时代，各行各业正经历从“人工驱动”到“智能驱动”的根本性变革。检测认证（TIC）行业作为关乎质量与安全的重要支柱，也不例外。在…
阅读更多...
React事件处理：如何给按钮绑定onClick点击事件？

React事件处理：如何给按钮绑定onClick点击事件？

系列回顾： 在前几篇文章中，我们已经学会了如何使用 State 管理组件的内部数据，以及如何通过 Props 实现父子组件之间的通信。我们的组件现在已经有了“数据”和“外观”。但是，它还像一个只能看的“模型”，无法与用户进…
阅读更多...
【机器学习|学习笔记】粒子群优化（Particle Swarm Optimization, PSO）详解，附代码。

【机器学习|学习笔记】粒子群优化（Particle Swarm Optimization, PSO）详解，附代码。

【机器学习|学习笔记】粒子群优化（Particle Swarm Optimization, PSO）详解，附代码。 【机器学习|学习笔记】粒子群优化（Particle Swarm Optimization, PSO）详解，附代码。 文章目录 【机器学习|学习笔记】粒…
阅读更多...
深度剖析：AI 社媒矩阵营销工具，如何高效获客？

深度剖析：AI 社媒矩阵营销工具，如何高效获客？

在社交媒体营销领域，竞争日益激烈，传统的社媒矩阵运营方式面临诸多挑战。而 AI 社媒矩阵营销工具的出现，正以前所未有的方式重构社媒矩阵的底层架构，为营销人员带来了全新的机遇与变革。接下来，我们将从技术破局、实战…
阅读更多...
Spring XML 常用命名空间配置

Spring XML 常用命名空间配置

Spring XML 常用命名空间配置 下面是一个综合性的Spring XML配置样例&#xff0c;展示了各种常用命名空间的使用方式&#xff1a; <?xml version"1.0" encoding"UTF-8"?> <beans xmlns"http://www.springframework.org/schema/beans&quo…
阅读更多...
UE5场景漫游——开始界面及关卡跳转

UE5场景漫游——开始界面及关卡跳转

UE中实现UMG游戏界面搭建及蓝图控制&#xff0c;点击游戏界面中的按钮实现关卡的跳转效果。 一、游戏界面显示。1.创建UMG&#xff0c;2.搭建UI。3.关卡蓝图控制显示 二、点击按钮之后实现关卡跳转
阅读更多...
CSS 外边距合并（Margin Collapsing）问题研究

CSS 外边距合并（Margin Collapsing）问题研究

在 CSS 中&#xff0c;margin-top 属性会导致外部 DIV 移动的现象主要是由于 外边距合并&#xff08;Margin Collapsing&#xff09; 造成的。这是 CSS 盒模型的一个特性&#xff0c;可能会与直觉相悖。 外边距合并的原理 当一个元素&#xff08;如内部 DIV&#xff09;的 ma…
阅读更多...
清理电脑C磁盘，方法N：使用【360软件】中的【清理C盘空间】

清理电脑C磁盘，方法N：使用【360软件】中的【清理C盘空间】

1、先下载并打开【360安全卫士】&#xff0c;点击如下位置&#xff1a; 之后&#xff0c;可以把这个东西&#xff0c;创建快捷方式到电脑桌面&#xff0c;方便以后使用&#xff1a;
阅读更多...
最新文章

Copyright @ 2022~2023