news 2026/10/20 11:06:18

【Agent】【workflow】19.工作流烹饪书 - LlamaIndex工作流示例

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
【Agent】【workflow】19.工作流烹饪书 - LlamaIndex工作流示例

案例目标

本案例展示了LlamaIndex工作流系统的多种使用模式，通过一系列示例演示如何构建、组合和扩展工作流。主要目标包括：

  • 展示LlamaIndex工作流的基本构建块和设计模式
  • 演示不同类型的工作流（顺序、条件、循环等）
  • 提供可重用的工作流组件和模板
  • 展示工作流的可视化和调试功能
  • 说明如何将工作流与现有LlamaIndex组件集成

技术栈与核心依赖

主要技术栈
  • LlamaIndex工作流系统- 提供事件驱动的工作流框架
  • OpenAI GPT模型- 用于语言理解和生成
  • Pydantic- 用于数据验证和设置管理
  • Matplotlib- 用于工作流可视化
核心依赖
from llama_index.core.workflow import ( step, Workflow, StartEvent, StopEvent, Event, Context, draw_all_possible_flows ) import matplotlib.pyplot as plt from llama_index.llms.openai import OpenAI from typing import List, Optional, Tuple, Any from pydantic import BaseModel

环境配置

环境要求
  • Python 3.8+
  • OpenAI API密钥（通过环境变量OPENAI_API_KEY设置）
  • 已安装的LlamaIndex库（包含工作流模块）
  • Matplotlib用于可视化
LLM配置示例
from llama_index.llms.openai import OpenAI # 初始化OpenAI模型 llm = OpenAI(model="gpt-4o", temperature=0.1)

案例实现

1. 简单顺序工作流

最基本的顺序执行工作流，包含三个步骤：处理输入、转换数据和生成输出。

class ProcessInputEvent(Event): input: str class TransformEvent(Event): transformed: str class GenerateOutputEvent(Event): output: str class SimpleWorkflow(Workflow): @step async def process_input(self, ctx: Context, ev: StartEvent) -> ProcessInputEvent: input_data = ev.get("input") # 处理输入 processed = f"Processed: {input_data}" return ProcessInputEvent(input=processed) @step async def transform_data(self, ctx: Context, ev: ProcessInputEvent) -> GenerateOutputEvent: # 转换数据 transformed = f"Transformed: {ev.input}" return GenerateOutputEvent(output=transformed) @step async def generate_output(self, ctx: Context, ev: GenerateOutputEvent) -> StopEvent: # 生成最终输出 final_output = f"Final: {ev.output}" return StopEvent(result=final_output)

2. 条件分支工作流

基于条件选择不同执行路径的工作流，展示了如何实现决策逻辑。

class CheckConditionEvent(Event): condition: bool class PathAEvent(Event): data: str class PathBEvent(Event): data: str class ConditionalWorkflow(Workflow): @step async def check_condition(self, ctx: Context, ev: StartEvent) -> CheckConditionEvent: # 检查条件 condition = ev.get("condition", False) return CheckConditionEvent(condition=condition) @step async def path_a(self, ctx: Context, ev: CheckConditionEvent) -> PathAEvent: # 条件为真时执行路径A if ev.condition: return PathAEvent(data="Executed Path A") @step async def path_b(self, ctx: Context, ev: CheckConditionEvent) -> PathBEvent: # 条件为假时执行路径B if not ev.condition: return PathBEvent(data="Executed Path B") @step async def merge_paths(self, ctx: Context, ev: Optional[PathAEvent] = None, ev: Optional[PathBEvent] = None) -> StopEvent: # 合并路径结果 if ev is not None and isinstance(ev, PathAEvent): result = ev.data elif ev is not None and isinstance(ev, PathBEvent): result = ev.data else: result = "No path executed" return StopEvent(result=result)

3. 循环工作流

实现循环逻辑的工作流，通过上下文管理循环状态和终止条件。

class LoopStartEvent(Event): max_iterations: int class LoopIterationEvent(Event): iteration: int max_iterations: int class LoopCompleteEvent(Event): result: str class LoopWorkflow(Workflow): @step async def start_loop(self, ctx: Context, ev: StartEvent) -> LoopStartEvent: # 初始化循环 max_iterations = ev.get("max_iterations", 3) ctx.data["current_iteration"] = 0 return LoopStartEvent(max_iterations=max_iterations) @step async def loop_iteration(self, ctx: Context, ev: LoopStartEvent) -> LoopIterationEvent: # 执行循环迭代 current = ctx.data.get("current_iteration", 0) ctx.data["current_iteration"] = current + 1 return LoopIterationEvent(iteration=current + 1, max_iterations=ev.max_iterations) @step async def check_continue(self, ctx: Context, ev: LoopIterationEvent) -> Optional[LoopIterationEvent]: # 检查是否继续循环 if ev.iteration < ev.max_iterations: return LoopIterationEvent(iteration=ev.iteration + 1, max_iterations=ev.max_iterations) return None @step async def complete_loop(self, ctx: Context, ev: LoopIterationEvent) -> LoopCompleteEvent: # 完成循环 return LoopCompleteEvent(result=f"Completed {ev.iteration} iterations") @step async def finalize(self, ctx: Context, ev: LoopCompleteEvent) -> StopEvent: # 最终化结果 return StopEvent(result=ev.result)

4. 并行工作流

并行执行多个任务的工作流，展示了如何处理并行分支和结果合并。

class ParallelTaskAEvent(Event): result_a: str class ParallelTaskBEvent(Event): result_b: str class ParallelTaskCEvent(Event): result_c: str class ParallelMergeEvent(Event): results: List[str] class ParallelWorkflow(Workflow): @step async def start_parallel_tasks(self, ctx: Context, ev: StartEvent) -> Tuple[ParallelTaskAEvent, ParallelTaskBEvent, ParallelTaskCEvent]: # 启动并行任务 return ( ParallelTaskAEvent(result_a="Task A Result"), ParallelTaskBEvent(result_b="Task B Result"), ParallelTaskCEvent(result_c="Task C Result") ) @step async def merge_parallel_results(self, ctx: Context, ev: ParallelTaskAEvent, ev: ParallelTaskBEvent, ev: ParallelTaskCEvent) -> ParallelMergeEvent: # 合并并行结果 results = [ev.result_a, ev.result_b, ev.result_c] return ParallelMergeEvent(results=results) @step async def finalize_parallel(self, ctx: Context, ev: ParallelMergeEvent) -> StopEvent: # 最终化并行结果 return StopEvent(result=f"Merged Results: {', '.join(ev.results)}")

5. 流式处理工作流

处理流式数据的工作流，展示了如何处理连续输入和生成流式输出。

class StreamInputEvent(Event): chunk: str class StreamProcessEvent(Event): processed_chunk: str class StreamOutputEvent(Event): output_chunk: str class StreamingWorkflow(Workflow): @step async def process_stream(self, ctx: Context, ev: StreamInputEvent) -> StreamProcessEvent: # 处理流式输入 processed = f"Processed: {ev.chunk}" return StreamProcessEvent(processed_chunk=processed) @step async def generate_stream_output(self, ctx: Context, ev: StreamProcessEvent) -> StreamOutputEvent: # 生成流式输出 output = f"Output: {ev.processed_chunk}" return StreamOutputEvent(output_chunk=output) @step async def collect_stream(self, ctx: Context, ev: StreamOutputEvent) -> Optional[StopEvent]: # 收集流式输出 if not hasattr(ctx.data, "output_chunks"): ctx.data.output_chunks = [] ctx.data.output_chunks.append(ev.output_chunk) # 在实际应用中，这里会有更复杂的终止条件 if len(ctx.data.output_chunks) >= 3: final_output = " ".join(ctx.data.output_chunks) return StopEvent(result=final_output) return None

6. 错误处理工作流

包含错误处理逻辑的工作流，展示了如何处理异常和恢复。

class ErrorEvent(Event): error: str class RetryEvent(Event): attempt: int class ErrorHandlingWorkflow(Workflow): @step async def risky_operation(self, ctx: Context, ev: StartEvent) -> Union[StopEvent, ErrorEvent]: # 可能失败的操作 try: # 模拟可能失败的操作 import random if random.random() > 0.7: # 30%的失败概率 raise ValueError("Random operation failed") return StopEvent(result="Operation succeeded") except Exception as e: return ErrorEvent(error=str(e)) @step async def handle_error(self, ctx: Context, ev: ErrorEvent) -> RetryEvent: # 处理错误 if not hasattr(ctx.data, "retry_count"): ctx.data.retry_count = 0 ctx.data.retry_count += 1 if ctx.data.retry_count < 3: return RetryEvent(attempt=ctx.data.retry_count) # 超过最大重试次数 return StopEvent(result=f"Operation failed after {ctx.data.retry_count} retries: {ev.error}") @step async def retry_operation(self, ctx: Context, ev: RetryEvent) -> StartEvent: # 重试操作 return StartEvent(input="retry")

7. 工作流可视化

使用LlamaIndex提供的可视化工具来展示工作流结构。

# 可视化工作流 def visualize_workflow(workflow_class, filename): # 创建工作流实例 workflow = workflow_class() # 绘制所有可能的流程 draw_all_possible_flows(workflow, filename=filename) # 显示图像 plt.figure(figsize=(12, 8)) img = plt.imread(filename) plt.imshow(img) plt.axis('off') plt.show() # 可视化简单工作流 visualize_workflow(SimpleWorkflow, "simple_workflow.png")

案例效果

本案例提供了多种工作流模式的完整实现，每个工作流都展示了不同的设计模式和最佳实践：

顺序工作流效果

顺序工作流演示了基本的线性执行流程，每个步骤按顺序执行，前一个步骤的输出作为后一个步骤的输入。这种模式适用于简单的数据处理管道。

条件分支工作流效果

条件分支工作流展示了如何根据条件选择不同的执行路径，实现了基本的决策逻辑。这种模式适用于需要根据不同条件执行不同操作的场景。

循环工作流效果

循环工作流演示了如何实现迭代逻辑，通过上下文管理循环状态和终止条件。这种模式适用于需要重复执行某些操作直到满足特定条件的场景。

并行工作流效果

并行工作流展示了如何同时执行多个任务并合并结果，提高了处理效率。这种模式适用于可以并行处理的独立任务。

流式处理工作流效果

流式处理工作流演示了如何处理连续的数据流，适用于实时数据处理和生成场景。这种模式特别适合处理大量数据或需要实时响应的应用。

错误处理工作流效果

错误处理工作流展示了如何实现健壮的错误处理和恢复机制，包括重试逻辑和错误传播。这种模式提高了系统的可靠性和容错能力。

工作流可视化效果

工作流可视化功能提供了直观的流程图表示，帮助开发者理解工作流的结构和执行路径。这对于复杂工作流的调试和优化非常有用。

案例实现思路

1. 事件驱动设计

所有工作流都基于事件驱动的设计模式，通过定义不同类型的事件来表示工作流中的状态变化和数据流动。每个步骤接收特定类型的事件并产生新的事件，形成清晰的数据流和控制流。

2. 步骤装饰器模式

使用@step装饰器标记工作流中的步骤函数，使框架能够自动识别和管理这些步骤。每个步骤函数定义了输入事件类型和输出事件类型，建立了明确的类型契约。

3. 上下文管理

通过Context对象在工作流的不同步骤之间共享状态和数据。上下文提供了持久化存储机制，允许步骤之间传递复杂状态信息，而不需要通过事件链显式传递。

4. 类型安全

利用Python类型提示和Pydantic模型确保事件数据的类型安全。这种设计在编译时和运行时都能提供类型检查，减少了错误并提高了代码的可维护性。

5. 异步执行

工作流框架基于异步执行模型，允许非阻塞操作和高效资源利用。这对于I/O密集型任务和并行处理特别重要，提高了整体性能。

6. 可视化支持

工作流框架内置了可视化功能，可以自动生成流程图表示工作流结构。这种可视化帮助开发者理解复杂的工作流逻辑，并提供了调试和文档化的工具。

扩展建议

1. 集成外部系统

扩展工作流以与外部系统和服务集成，例如数据库、API和消息队列。这可以通过创建特定的事件类型和步骤来实现，使工作流能够与外部资源交互。

2. 动态工作流

实现动态工作流生成，根据运行时条件动态构建工作流结构。这将允许更灵活的自适应工作流，能够根据输入数据和执行结果调整执行路径。

3. 工作流模板库

创建可重用的工作流模板库，涵盖常见模式和最佳实践。这将加速开发过程，并确保一致性和质量。模板可以包括数据处理、决策树、并行处理等常见模式。

4. 高级错误处理

实现更复杂的错误处理策略，包括断路器模式、重试策略和错误恢复机制。这将提高工作流的健壮性和可靠性，特别是在处理不可靠的外部系统时。

5. 性能监控

添加性能监控和分析功能，跟踪工作流执行时间、资源使用和瓶颈。这将帮助优化工作流性能，并提供运行时洞察。

6. 工作流版本控制

实现工作流版本控制和迁移机制，允许工作流的演进和更新而不破坏现有实例。这对于长期维护和生产环境部署非常重要。

7. 分布式执行

扩展工作流框架以支持分布式执行，允许步骤在不同的节点或容器中运行。这将提高可扩展性和容错能力，适用于大规模处理需求。

总结

本案例全面展示了LlamaIndex工作流系统的功能和灵活性，通过一系列示例演示了如何构建各种类型的工作流。从简单的顺序流程到复杂的并行处理，从基本的数据转换到高级的错误处理，这些示例提供了构建复杂应用程序的基础。

工作流系统的事件驱动设计、类型安全和异步执行模型使其成为构建复杂AI应用的强大工具。通过可视化功能，开发者可以轻松理解和调试复杂的工作流逻辑，提高了开发效率和代码质量。

这些示例不仅展示了工作流系统的当前功能，也为未来的扩展和定制提供了基础。通过结合LlamaIndex的其他组件，如查询引擎、数据索引和LLM集成，开发者可以构建端到端的AI应用程序，满足各种复杂需求。

总的来说，LlamaIndex工作流系统提供了一个灵活、强大且易于使用的框架，用于构建复杂的AI工作流，是现代AI应用开发的重要工具。

版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/10/20 11:05:43

126、RDMA编程最佳实践：避免CPU忙等与上下文切换

一、从一次线上事故说起 去年双十一大促，我们团队负责的分布式存储集群突然出现严重的尾延迟抖动。监控面板上，P99延迟从正常的200μs飙升到5ms，整整25倍的恶化。更诡异的是，CPU利用率并不高，只有30%左右，但上下文切换次数暴涨到每秒几十万次。 我登录到出问题的机器，…

作者头像 李华
网站建设 2026/10/20 11:05:36

128、RDMA编程最佳实践：多QP与多CQ的负载均衡

一、一个让我熬夜到凌晨三点的bug 去年做分布式存储项目，集群规模从4节点扩展到16节点，原本跑得稳稳的RDMA代码突然开始间歇性丢包。现象很诡异：业务层报写延迟抖动，从正常的5微秒跳到200微秒，偶尔还出现连接断开。抓了三天日志，最后发现是QP（Queue Pair）和CQ（Comple…

作者头像 李华
网站建设 2026/10/20 11:04:56

IS-IS区域迁移与路由汇总实战：从NET地址到LSP收敛

我第一次正经做IS-IS区域迁移和路由汇总&#xff0c;是在一个把两个城域网合并成一张骨干网的改造项目里。当时接入侧一个区域用了老的地址规划&#xff0c;核心侧已经换了新的区域地址&#xff0c;两边的IS-IS进程虽然都通&#xff0c;但LSP互相看不到&#xff0c;路由表里全是…

作者头像 李华
网站建设 2026/10/20 11:03:01

声纹识别Web身份认证实战：从MFCC特征提取到前后端落地

简介&#xff1a;基于声纹识别的Web身份认证系统&#xff0c;面向需要为企业或业务系统提供安全、便捷登录方案的技术人员&#xff0c;也适合信息安全、人工智能方向的学生作为课设参考。方案以人声生物特征代替传统口令&#xff0c;用户无需记忆复杂密码&#xff0c;且不涉及人…

作者头像 李华
网站建设 2026/10/20 11:01:42

智能化小区网络规划：从拓扑选型到设备落地实践

简介&#xff1a;智能化小区网络规划毕业设计论文以doc文档形式打包&#xff0c;全文围绕智能化小区宽带网络的建设流程展开&#xff0c;适合网络工程、智能建筑、物联网技术等专业学生完成毕业设计&#xff0c;也适合从事住宅小区弱电规划与网络运维的技术人员查阅。文档先介绍…

作者头像 李华