【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.8OpenAI API密钥通过环境变量OPENAI_API_KEY设置已安装的LlamaIndex库包含工作流模块Matplotlib用于可视化LLM配置示例from llama_index.llms.openai import OpenAI # 初始化OpenAI模型 llm OpenAI(modelgpt-4o, temperature0.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 fProcessed: {input_data} return ProcessInputEvent(inputprocessed) step async def transform_data(self, ctx: Context, ev: ProcessInputEvent) - GenerateOutputEvent: # 转换数据 transformed fTransformed: {ev.input} return GenerateOutputEvent(outputtransformed) step async def generate_output(self, ctx: Context, ev: GenerateOutputEvent) - StopEvent: # 生成最终输出 final_output fFinal: {ev.output} return StopEvent(resultfinal_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(conditioncondition) step async def path_a(self, ctx: Context, ev: CheckConditionEvent) - PathAEvent: # 条件为真时执行路径A if ev.condition: return PathAEvent(dataExecuted Path A) step async def path_b(self, ctx: Context, ev: CheckConditionEvent) - PathBEvent: # 条件为假时执行路径B if not ev.condition: return PathBEvent(dataExecuted 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(resultresult)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_iterationsmax_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(iterationcurrent 1, max_iterationsev.max_iterations) step async def check_continue(self, ctx: Context, ev: LoopIterationEvent) - Optional[LoopIterationEvent]: # 检查是否继续循环 if ev.iteration ev.max_iterations: return LoopIterationEvent(iterationev.iteration 1, max_iterationsev.max_iterations) return None step async def complete_loop(self, ctx: Context, ev: LoopIterationEvent) - LoopCompleteEvent: # 完成循环 return LoopCompleteEvent(resultfCompleted {ev.iteration} iterations) step async def finalize(self, ctx: Context, ev: LoopCompleteEvent) - StopEvent: # 最终化结果 return StopEvent(resultev.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_aTask A Result), ParallelTaskBEvent(result_bTask B Result), ParallelTaskCEvent(result_cTask 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(resultsresults) step async def finalize_parallel(self, ctx: Context, ev: ParallelMergeEvent) - StopEvent: # 最终化并行结果 return StopEvent(resultfMerged 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 fProcessed: {ev.chunk} return StreamProcessEvent(processed_chunkprocessed) step async def generate_stream_output(self, ctx: Context, ev: StreamProcessEvent) - StreamOutputEvent: # 生成流式输出 output fOutput: {ev.processed_chunk} return StreamOutputEvent(output_chunkoutput) 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(resultfinal_output) return None6. 错误处理工作流包含错误处理逻辑的工作流展示了如何处理异常和恢复。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(resultOperation succeeded) except Exception as e: return ErrorEvent(errorstr(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(attemptctx.data.retry_count) # 超过最大重试次数 return StopEvent(resultfOperation failed after {ctx.data.retry_count} retries: {ev.error}) step async def retry_operation(self, ctx: Context, ev: RetryEvent) - StartEvent: # 重试操作 return StartEvent(inputretry)7. 工作流可视化使用LlamaIndex提供的可视化工具来展示工作流结构。# 可视化工作流 def visualize_workflow(workflow_class, filename): # 创建工作流实例 workflow workflow_class() # 绘制所有可能的流程 draw_all_possible_flows(workflow, filenamefilename) # 显示图像 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应用开发的重要工具。

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

相关推荐

genshin-wish-export 快速上手指南：三步导出并分析原神抽卡记录

RAG Chunking 实战｜主流切分策略选型与进阶优化手段

CVPR 2026 Findings｜FSSG：前视声呐数据集

更多资讯

弹性伸缩实战：扩容失效、成本失控与业务报错的排查指南

Excel批量删除空格全攻略：识别空格类型与高效清洗方案

STM32驱动VL53L1X激光测距：从I2C通信到稳定读数

无线传感器网络覆盖优化：MATLAB虚拟力算法与PSO/GA代码实战

双指针精讲：从移动零到数组原地压缩的通用套路

Agent Harness 标准化实战：从混乱到可控的工程化之路

今日更新

智能硬件视频会议集成：从芯片选型到场景落地的全栈实践

FFmpeg音视频处理与5路RTMP推流实战：从原理到118克盒子部署

RK3528+1GB内存如何变身直播节点？硬编解码与六协议实战复盘

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

基于GMM的语音识别系统原理与MATLAB实现详解

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？