LLMFlows异步流教程：并行运行多个AI模型调用，提升应用效率300%-开发者社区

LLMFlows异步流教程：并行运行多个AI模型调用，提升应用效率300%

【免费下载链接】llmflowsLLMFlows - Simple, Explicit and Transparent LLM Apps项目地址: https://gitcode.com/gh_mirrors/ll/llmflows

LLMFlows是一个专注于简化AI应用开发的Python框架，通过其异步流功能，开发者可以轻松实现多个AI模型调用的并行执行，显著提升应用效率。本文将详细介绍如何利用LLMFlows的异步流特性，构建高效的AI应用。

为什么选择异步流处理AI模型调用？

在传统的AI应用开发中，模型调用通常是串行执行的，这意味着每个调用必须等待前一个调用完成才能开始。这种方式在处理多个相互独立的模型调用时效率极低，会造成大量的等待时间。

LLMFlows的异步流功能允许开发者将多个AI模型调用设计为流步骤（FlowStep），并自动并行执行那些所有输入都已准备就绪的步骤。这种并行处理能力可以将应用效率提升300%以上，特别是在需要处理复杂流程或多个独立任务的场景中。

图：LLMFlows复杂异步流结构示意图，展示了多个流步骤如何并行执行

LLMFlows异步流核心组件

LLMFlows提供了一系列用于构建异步流的核心组件，主要包括：

AsyncBaseFlow：所有异步流的基类，负责管理流步骤和执行流程
AsyncBaseFlowStep：异步流步骤的基类，支持并行执行
AsyncFlow：异步流的具体实现类
AsyncFlowStep：异步流步骤的具体实现类

这些组件定义在以下路径中：

llmflows/flows/async_base_flow.py
llmflows/flows/async_base_flowstep.py
llmflows/flows/async_flow.py
llmflows/flows/async_flowstep.py

快速入门：构建你的第一个异步流

下面我们通过一个实际示例来演示如何使用LLMFlows构建异步流。这个示例将创建一个电影相关内容生成器，包括生成电影标题、歌曲标题、主要角色和歌词等多个并行任务。

1. 安装LLMFlows

首先，你需要克隆仓库并安装LLMFlows：

git clone https://gitcode.com/gh_mirrors/ll/llmflows cd llmflows pip install -r requirements.txt

2. 创建异步流步骤

以下是创建多个异步流步骤的基本示例：

from llmflows.flows import AsyncFlow, AsyncFlowStep from llmflows.llms import OpenAI from llmflows.prompts import PromptTemplate # 创建LLM实例 open_ai_llm = OpenAI(api_key="your-api-key") # 创建提示模板 title_template = PromptTemplate("What is a good title of a movie about {topic}?") song_template = PromptTemplate("What is a good song title for a movie called {movie_title}?") characters_template = PromptTemplate("What are two main characters for a movie called {movie_title}?") lyrics_template = PromptTemplate("Write lyrics for a song based on {song_title} and characters {main_characters}") # 创建异步流步骤 flowstep1 = AsyncFlowStep( name="Movie Title", llm=open_ai_llm, prompt_template=title_template, output_key="movie_title" ) flowstep2 = AsyncFlowStep( name="Song Title", llm=open_ai_llm, prompt_template=song_template, output_key="song_title" ) flowstep3 = AsyncFlowStep( name="Main Characters", llm=open_ai_llm, prompt_template=characters_template, output_key="main_characters" ) flowstep4 = AsyncFlowStep( name="Song Lyrics", llm=open_ai_llm, prompt_template=lyrics_template, output_key="song_lyrics" )

3. 连接流步骤并运行

接下来，我们需要连接这些流步骤，并创建异步流：

# 连接流步骤 flowstep1.connect(flowstep2, flowstep3) # 电影标题生成后，并行生成歌曲标题和角色 flowstep2.connect(flowstep4) # 歌曲标题生成后，用于生成歌词 flowstep3.connect(flowstep4) # 角色生成后，用于生成歌词 # 创建异步流 soundtrack_flow = AsyncFlow(flowstep1) # 运行流 async def run_flow(): result = await soundtrack_flow.start(topic="friendship", verbose=True) print(result) import asyncio asyncio.run(run_flow())

完整的示例代码可以在examples/10_complex_async_flows.py中找到。

异步流的工作原理

LLMFlows的异步流通过以下机制实现并行执行：

依赖解析：流会自动分析各个步骤之间的依赖关系
就绪检查：持续检查哪些步骤的所有输入都已准备就绪
并行执行：对所有就绪的步骤进行并行执行
结果传递：将步骤的输出自动传递给依赖它的后续步骤

这种机制确保了系统资源得到最充分的利用，同时简化了开发者的工作，不需要手动管理线程或进程。

图：LLMFlows异步流执行流程示意图，展示了步骤如何根据依赖关系并行执行

异步流的最佳实践

1. 合理设计步骤依赖

为了最大化并行性，应该将相互独立的任务设计为可以并行执行的步骤。例如，在生成报告时，可以同时并行生成不同章节，最后再合并。

2. 设置适当的超时时间

在创建异步流步骤时，可以设置适当的超时时间，避免某个步骤阻塞整个流程：

flowstep = AsyncFlowStep( name="Example Step", llm=open_ai_llm, prompt_template=prompt_template, output_key="output", timeout=30 # 设置30秒超时 )

3. 使用回调函数监控进度

LLMFlows支持回调函数，可以用来监控流的执行进度和处理异常：

from llmflows.callbacks import AsyncBaseCallback class MyCallback(AsyncBaseCallback): async def on_flow_start(self, flow_name): print(f"Flow {flow_name} started") async def on_flow_end(self, flow_name, result): print(f"Flow {flow_name} ended with result: {result}") # 在流中使用回调 flow = AsyncFlow(first_step, callbacks=[MyCallback()])

相关的回调类定义在llmflows/callbacks/目录下。