我要投稿

使用LLamaIndex Workflow来打造水墨风格图片生成工作流

发布日期：2025-08-20 07:18:32 浏览次数： 1519

作者：Data Leads Future

微信搜一搜，关注“Data Leads Future”

在今天的文章里，我将帮助你搭建一个可以生成浓浓东方风格的水墨插画的工作流。同时该工作流还支持你多轮调整prompt和最终图片，以便节约token和时间成本。

Introduction

最近我想要搭建一个可以低成本快速生成我的博客图片的智能体工作流程。

我想让我的博客图片具备浓厚的艺术气息和古典的东方韵味，所以我希望能让我的工作流可以精细地对大模型Context进行控制，并能够不断地调整用来绘图的prompt和最终绘制的图片效果，还要保证token和时间成本最小化。

然后我立即陷入一个两难选择：

如果我选择使用dify, n8n等低代码平台，我会得不到足够的灵活性。这类平台无法支撑我在对话中调整prompt，按照文章风格生成博客图片的诉求。

如果我选择LangGraph，CrewAI等当今流行的智能体开发框架，这类框架抽象层级又过高，导致我无法对智能体应用的执行过程进行精细控制。

如果是你来面对这个任务，将会如何做出选择呢？

还好这个世界不是非此即彼的，在经过无数次失败又不断地尝试后，我终于找到一个很好的解决方案：LLamaIndex Workflow。

它既提供了一套高效的工作流开发流程，又不会对我的智能体执行过程产生太多的抽象，因此可以让图片生成程序按我的要求精细地运行。

今天，就让我使用最新的LlamaIndex Workflow 1.0版本，来为你搭建一个可以生成水墨风格插画的工作流程。

你为什么要关心？

在今天的文章里，我会：

引导你在项目实践中学习LlamaIndex Workflow的基本用法。
使用chainlit搭建一个chatbot界面，你能直观地看到生成的图片。
利用DeepSeek，在DALL-E-3模型之外生成更符合项目要求的绘图prompt。
使用多轮对话来对生成的prompt或者最终生成的图片做再次调整。
通过对大模型Context的精细控制，在token和时间层面优化成本。

最终，你仅需要一句简单的描述，就能绘制出一幅精美的水墨风格图片出来。

广袖流仙裙还是有那么点意思的

更重要的是，通过对本项目的实践，你将初步了解在企业级智能体应用中，我们是如何利用工作流来完成复杂的定制化需求开发的。

如果你需要一些前置知识，我写了一篇文章详细讲解LlamaIndex Workflow的事件驱动架构，你可以点击这里阅读：

深入浅出LlamaIndex Workflows: 事件驱动的LLM架构

业务流程图设计

对于智能体工作流类型的应用开发，我强烈推荐在开始编码前，先进行业务流程图设计，这有利于你全盘掌握程序的运行流程。

在这一章节，我将为你演示我对于本项目业务流程图的完整设计思路：

prompt生成流程

在今天的项目里我使用DALL-E-3进行绘图，DALL-E-3本身就具备根据用户的意图，重写出适合绘图的详细prompt的能力。

但是我们的要求更高，希望DALL-E-3完全按照我心中的样子对图片进行绘制。所以把用户意图重写为详细prompt的这个过程，我们将从DALL-E-3模型迁移到我们自己的工作流节点里面来。

因为今天的主题是绘制优美水墨风格的插画，所以我需要选用能够充分理解用户意图中的那份东方意境，并扩展为DALL-E-3可以理解的绘图prompt的大模型，这里我选择了DeepSeek-Chat模型。

由于DeepSeek接受了生成DALL-E-3绘图prompt的提示，所以生成出来的prompt是纯英文的。

如果你擅长使用英文，那么在这个环节就可以结束了。但如果你跟我一样是非英文母语的用户，又希望准确理解生成出的prompt的内容，则可以再加上一个翻译节点，这一步并不麻烦。

最后，生成的prompt和对prompt的翻译都会通过LlamaIndex Workflow的StopEvent返回给用户。

生成绘图prompt的流程

使用Context共享来解耦工作流循环

生成完prompt后，接下来我们要么就会把prompt提供给DALL-E来生成图片，或者返回去让DeepSeek重新对prompt进行调整。

这个时候，你肯定会想着加入用户反馈以及工作流迭代等特性。根据最终生成的图片效果，提供修改意见，并要求工作流重新生成prompt，一直循环到获得满意的输出为止。

由于我们要在prompt生成和图片生成这两个节点后都支持用户反馈，这样循环会大大增加工作流的复杂度。

但在今天的项目里，我打算使用一个小技巧，来大幅简化工作流的实现。

我们不在prompt或者图片生成后再加入用户反馈来判断是否要重新生成prompt，而是在工作流的一开始，就加入一个if-else节点，判断用户的输入是否包含特定的关键词（这里是APPROVE，你可以换成你自己的），如果包含，就走到图片生成分支，如果不包含，就走到prompt生成分支。

这样，每次迭代就是一次工作流的重新执行，从而将工作流从迭代中解耦出来。

使用条件分支和Context来解耦工作流循环

加入分支节点后，我们会面临一个麻烦，就是图片生成的那个分支，根本不知道上一次运行产生的prompt到底是什么。而且我也不希望把上一次运行的消息历史都保存下来，因为消息历史里还包含对prompt的翻译等并不需要发送给大模型的信息。

这个时候，最好的选择就是让工作流的运行都使用同一个上下文，并把生成出来的prompt保存到上下文里去。

刚好LlamaIndex Workflow支持多次运行共享同一个上下文，所以我们可以往工作流里加入相应的往Context里保存变量的逻辑。

重写用户的历史绘图请求

由于用户会通过多次对话逐步调整大模型生成的prompt，我们需要把用户完整的对话历史都提供给大模型。

通用的做法是使用role为user和role为assistant的消息，将用户和大模型的历史对话都保存下来都提供给DeepSeek。

但是这样做，随着调整的持续，对话消息会越来越长，导致大模型会忽略真正需要注意的关键信息。

我们也不能采用截断历史信息，只保留最近几轮对话的方案。因为最详细的绘图意图一般是用户最开始提供的请求。

所以这里我将采取重写用户请求历史的方案，将用户多轮对prompt和图片的调整都重写为一段完整的请求。

具体的方法是：在Context里创建一个list容器，在生成prompt时，将用户的最新输入append进这个list里。然后调用DeepSeek将所有用户输入重写为一段完整的绘图描述，存入Context。

我们要把用户的历史输入片段重写为一句完整的绘图意图

完善system prompt

最后，我们修改DeepSeek生成绘图prompt的节点。在指示大模型生成prompt之前，从Context中取出重写后的用户历史请求以及上一次生成的绘图prompt，一起合并到system prompt里去。

在最开始执行工作流的时候，Context里的用户历史请求和最后一次生成的prompt都是空的，但这不重要，因为最新的用户请求都是使用role为user的message提供给大模型的。

至此，完整的业务流程图就设计完毕，如下图所示。

完整的业务流程图

接下来，我们就可以按照业务流程图的设计，开始进行编码了。

使用LlamaIndex Workflow开发你的绘图工作流

在今天的项目里，除了会使用LlamaIndex Workflow构建绘图工作流以外，为了方便你与工作流应用进行交互，我还会使用chainlit来构建交互界面。

对话界面如下图所示：

使用Chainlit开发的交互界面

接下来就开始实际代码逻辑的编写。

项目整体结构

本项目的代码结构如下：

project_source_code  |-app.py  |-ctx_manager.py  |-events.py  |-prompts.py  |-workflow.py

app.py是应用的入口文件，用来存放chainlit代码。

workflow.py是LlamaIndex Workflow实现的核心代码逻辑。

prompts.py存放了提供将要提供给大模型的所有prompts。

events.py是LlamaIndex Workflow相关的事件定义。

ctx_manager.py存放了所有对Workflow Context的操作逻辑。

环境变量

在这个项目里，由于要同时使用DeepSeek和DALL-E-3两套大模型，所以我们要在`.env`文件里准备两套环境变量:

OPENAI_API_KEY=<Your DeepSeek API key>OPENAI_API_BASE=<DeepSeek endpoint>REAL_OPENAI_API_KEY=<Your OpenAI API key>REAL_OPENAI_BASE_URL=<OpenAI endpoint>

根据我的使用习惯，OPENAI_API_KEY和OPENAI_API_BASE依然指向了DeepSeek服务。而我使用REAL_OPENAI_API_KEY和REAL_OPENAI_BASE_URL指向OpenAI的服务。你可以根据自己的习惯进行调整。

定义Workflow需要的几个事件

LlamaIndex Workflow是基于事件驱动的架构，所以代码节点之间的跳转都需要定义相应的事件，这些事件定义我都放在events.py里面。

但是Workflow的核心代码还没开始写，就直接讲解Event的定义会让你感觉有些迷茫。没有关系，因为在前面我已经详细讲解了工作流的各个节点的作用，所以你依然可以明白每个事件各自用在什么地方。

GenPromptEvent，该事件会驱动DeepSeek代码节点开始生成绘图prompt。content属性里存放了用户最新的输入。

class GenPromptEvent(Event):    content: str

PromptGeneratedEvent，当绘图prompt生成完毕后，就会抛出该事件，content里存放了DeepSeek生成好的prompt。下游节点可以订阅该事件做prompt翻译，图片生成，以及用户请求重写等工作。

class PromptGeneratedEvent(Event):    content: str

StreamEvent，

由于我希望通过流式输出的方式在chainlit上展示prompt内容。chainlit代码里可以迭代该事件来获得流式消息。

class StreamEvent(Event):    target: str    delta: str

GenImageEvent，该事件会驱动DALL-E-3节点开始生成图片。content属性里依然存放着DeepSeek生成的prompt。

class GenImageEvent(Event):    content: str

RewriteQueryEvent，这个Event会调用大模型节点将用户的历史输入重写为一段完整的绘图意图。由于历史输入从Context里取，所以这个事件没有任何属性。

class RewriteQueryEvent(Event):    pass

实现Workflow代码逻辑

定义完所有的Workflow事件后，接下来就可以之前画的业务流程图来实现Workflow代码了。

在workflow.py里，我们定义一个名为ImageGeneration的类，作为LlamaIndex Workflow的子类。

在__init__方法里，我们要初始化两个大模型的client。一个是用来生成绘图prompt的DeepSeek模型，使用LlamaIndex的OpenAILike client。另一个是DALL-E-3模型，直接使用OpenAI的client。

class ImageGeneration(Workflow):    def __init__(self, *args, **kwargs):        self.deepseek_client = OpenAILike(            model="deepseek-chat",            is_chat_model=True,            is_function_calling_model=True        )        self.openai_client = AsyncOpenAI(            api_key=os.getenv("REAL_OPENAI_API_KEY"),            base_url=os.getenv("REAL_OPENAI_BASE_URL"),        )        super().__init__(*args, **kwargs)

on_start方法是工作流的入口方法，它只做条件分支判断。如果用户输入里包含"APPROVE"，则抛出GenImageEvent开始绘制图片。否则就抛出GenPromptEvent事件，开始生成绘图prompt，或者对现有的prompt进行调整。

class ImageGeneration(Workflow):    ...    @step    async def on_start(self, ctx: Context, ev: StartEvent) -> GenImageEvent | GenPromptEvent:        query = ev.query        if len(query) > 0 and ("APPROVE" in query.upper()):            return GenImageEvent(content=query)        else:            return GenPromptEvent(content=ev.query)

prompt_generator方法订阅GenPromptEvent事件，用来生成或调整绘图prompt。

class ImageGeneration(Workflow):    ...    @step    async def prompt_generator(self, ctx: Context, ev: GenPromptEvent) \            -> PromptGeneratedEvent | RewriteQueryEvent | None:        user_query = ev.content        hist_query = await ctx_mgr.get_rewritten_hist(ctx)        hist_prompt = await ctx_mgr.get_image_prompt(ctx)        system_prompt = PROMPT_GENERATE_SYSTEM.format(            hist_query=hist_query,            hist_prompt=hist_prompt        )        messages = [            ChatMessage(role="system", content=system_prompt),            ChatMessage(role="user", content=user_query)        ]        image_prompt = ""        events = await self.deepseek_client.astream_chat(messages)        async for event in events:            ctx.write_event_to_stream(StreamEvent(target="prompt", delta=event.delta))            image_prompt += event.delta        await ctx_mgr.add_query_hist(ctx, user_query)        await ctx_mgr.set_image_prompt(ctx, image_prompt)        ctx.send_event(PromptGeneratedEvent(content=image_prompt))        ctx.send_event(RewriteQueryEvent())

prompt_generator方法首先会从Context里获取被重写后的用户输入的历史意图，以及上一轮工作流生成的prompt，与预先定义的system prompt模板进行整合。

整合完毕的system prompt会与用户最新的输入一起，提交给DeepSeek模型并生成最新的绘图prompt。

我调用了llm client的流式api，所以我使用StreamEvent把大模型返回的消息抛出到Context的流里。同时，我会把所有消息都拼凑成一个完整的prompt，往下传递。

当然，prompt生成完毕后，我会把用户的最新输入和生成的prompt写回Context。

translate_prompt方法是可选的，它的作用是对DeepSeek生成出来的prompt进行翻译，以便你能准确理解prompt的内容。翻译后的结果只会用来在界面上展示，所以不会写入Context。

class ImageGeneration(Workflow):    ...    @step    async def translate_prompt(self, ctx: Context, ev: PromptGeneratedEvent) -> StopEvent:        image_prompt = ev.content        messages = [            ChatMessage(role="system", content=PROMPT_TRANSLATE_SYSTEM),            ChatMessage(role="user", content=image_prompt)        ]        events = await self.deepseek_client.astream_chat(messages)        translate_result = ""        async for event in events:            ctx.write_event_to_stream(StreamEvent(target="translate", delta=event.delta))            translate_result += event.delta        return StopEvent(target="prompt", result=translate_result)

translate_prompt方法会返回StopEvent，用来标记本次工作流结束执行。如果你不需要对绘图prompt进行翻译，则可以在prompt_generate方法里返回StopEvent，直接结束工作流。

如果你在输入的内容里包含有"APPROVE"关键词，那么工作流会直接进入generate_image方法。该方法会从Context里获取到最新的绘图prompt，并调用_image_generate方法开始生成图片。

class ImageGeneration(Workflow):    ...    @step    async def generate_image(self, ctx: Context, ev: GenImageEvent) -> StopEvent:        prompt = await ctx_mgr.get_image_prompt(ctx)        image_url, revised_prompt = await self._image_generate(prompt=prompt)        return StopEvent(target="image",                          result={                             "image_url": image_url,                              "revised_prompt": revised_prompt                         }                         )

我们已经提前使用DeepSeek生成好绘图prompt了，但是DALL-E-3基于安全的考虑依然会对输入的prompt进行重写，这会导致绘制出的图片与我们的意图相差太远，所以我们可以在绘图prompt前面加入以下这段内容来阻止DALL-E-3对prompt进行重写：

class ImageGeneration(Workflow):    ...    async def _image_generate(self, prompt: str) -> tuple[str, str]:        ## Stop DALL-E 3 from rewriting incoming prompts        final_prompt = f"""        I NEED to test how the tool works with extremely simple prompts. DO NOT add any detail, just use it AS-IS:        {prompt}        """        response = await self.openai_client.images.generate(            model="dall-e-3",            prompt=final_prompt,            n=1,            size="1792x1024",            quality="hd",            style="vivid",        )        return response.data[0].url, response.data[0].revised_prompt

最终，我们可以从DALL-E-3模型获得生成后的图片url。同时，我们也可以拿到DALL-E-3绘制图片实际使用的revised_prompt，这有助于我们与自己的prompt进行比对，看看区别在哪。

我们还要实现一个rewrite_query方法。该方法会把用户历史输入的列表从Context里取出来，然后提交给DeepSeek重写为一段完整的绘图意图，供下次调整绘图prompt或者生成的图片时使用。

class ImageGeneration(Workflow):    ...    @step    async def rewrite_query(self, ctx: Context, ev: RewriteQueryEvent) -> None:        query_hist_str = await ctx_mgr.get_query_hist(ctx)        messages = [            ChatMessage(role="system", content=PROMPT_REWRITE_SYSTEM),            ChatMessage(role="user", content=query_hist_str)        ]        response = await self.deepseek_client.achat(messages)        rewritten_prompt = response.message.content        await ctx_mgr.set_rewritten_hist(ctx, rewritten_prompt)

提供给大模型的prompts

prompts.py文件里存放了提供给大模型的几个system prompt：

PROMPT_GENERATE_SYSTEM用来提示DeepSeek模型生成水墨风格的绘图prompt，该prompt有{hist_query}和{hist_prompt}两个占位符，用来包含用户历史请求和上一次生成的绘图prompt：

PROMPT_GENERATE_SYSTEM = """## RoleYou're a visual art designer who's great at writing prompts perfect for DALL-E-3 image generation.## TaskBased on the [image content] I give you, and considering [previous requests], rewrite the prompt to be ideal for DALL-E-3 drawing.## LengthList 4 detailed sentences describing the prompt only - no intros or explanations.## Context HandlingIf the message includes [previous prompts], modify them based on the new info.## Art StyleThe artwork should be ink-wash style illustrations on slightly yellowed rice paper.## Previous Requests{hist_query}## Previous Prompts{hist_prompt}"""

PROMPT_TRANSLATE_SYSTEM用来对上一步生成的prompt进行翻译，内容比较简单：

PROMPT_TRANSLATE_SYSTEM = """## RoleYou're a professional translator in the AI field, great at turning English prompts into accurate Chinese.## TaskTranslate the [original prompt] I give you into Chinese.## RequirementsOnly provide the Chinese translation, no intros or explanations.-----------Original prompt:"""

PROMPT_REWRITE_SYSTEM用来对用户的历史请求进行重写。

PROMPT_REWRITE_SYSTEM = """You're a conversation history rewrite assistant.I'll give you a list of requests describing a scene, and you'll rewrite them into one complete sentence. Keep the same description of the scene, and don't add anything not in the original list."""

Context操作模块

由于LlamaIndex Workflow 1.0版本对Context的api进行了调整，我们需要访问Context.store才能读取和保存状态数据。因此我专门写了一个Context操作模块ctx_manager.py。

set_image_prompt和get_image_prompt方法用来存取DeepSeek生成的用来绘图的prompt。

async def set_image_prompt(ctx: Context, image_prompt: str) -> None:    await ctx.store.set("image_prompt", image_prompt)async def get_image_prompt(ctx: Context) -> str:    image_prompt = await ctx.store.get("image_prompt", "")    return image_prompt

add_query_hist方法会将用户的历史请求都添加进Context里的一个list容器中。get_query_hist方法会把历史请求从容器中取出，并拼接为一个字符串。

async def add_query_hist(ctx: Context, user_query: str) -> None:    query_hist = await ctx.store.get("query_hist", [])    query_hist.append(user_query)    await ctx.store.set("query_hist", query_hist)async def get_query_hist(ctx: Context) -> str:    query_hist = await ctx.store.get("query_hist", [])    query_hist_str = "; ".join(query_hist)    return query_hist_str

set_rewritten_hist和get_rewritten_hist方法用来存取重写后的用户绘图意图。

async def set_rewritten_hist(ctx: Context, rewritten_hist: str) -> None:    await ctx.store.set("rewritten_hist", rewritten_hist)async def get_rewritten_hist(ctx: Context) -> str:    rewritten_prompt = await ctx.store.get("rewritten_hist", "")    return rewritten_prompt

使用chainlit制作用户对话界面

chainlit使用生命周期管理的方式来组织代码。其中@cl.on_chat_start 注解的on_chat_start方法会在用户开始对话时被调用。@cl.on_message注解的main方法用来响应用户的单次对话。

由于我们要在用户的多轮对话之间共享Context，所以我们需要在on_chat_start方法里初始化Workflow和Context，并存入user_session。这样就可以在main方法里复用了。

@cl.on_chat_startasync def on_chat_start():    workflow = ImageGeneration(timeout=300)    context = Context(workflow)    cl.user_session.set("context", context)    cl.user_session.set("workflow", workflow)

在main方法里，除了获取workflow和context外，我们还要初始化一个cl.Message实例，这个实例会随着获取到的workflow的消息来不停更新内容，同时，如果workflow返回的是图片生成的事件，也会通过这个Message实例显示。

@cl.on_messageasync def main(message: cl.Message):    workflow = cl.user_session.get("workflow")    context = cl.user_session.get("context")    msg = cl.Message(content="Generating...")    await msg.send()    ...

我们要在同一个Message实例里同时显示来自prompt_generator和translate_prompt节点返回的流式消息。因此我们可以使用prompt_result和translate_result将当前轮的消息拼接起来，然后用一个模板一起更新。

@cl.on_messageasync def main(message: cl.Message):    ...    prompt_result = ""    translate_result = ""    handler = workflow.run(query=message.content, ctx=context)    async for event in handler.stream_events():        if isinstance(event, StreamEvent):            # # await msg.stream_token(event.delta)            match event.target:                case "prompt":                    prompt_result += event.delta                case "translate":                    translate_result += event.delta            msg.content = dedent(f"""### Prompt\n{prompt_result}### Translate{translate_result}APPROVE?            """)            await msg.update()        ...    await handler

如果workflow返回的消息是图片绘制的消息。则我们可以用一个cl.Image来展示图片内容。

@cl.on_messageasync def main(message: cl.Message):    ...        if isinstance(event, StopEvent) and event.target == "image":            image = cl.Image(url=event.result["image_url"], name="image1", display="inline")            msg.content = f"Revised prompt: \n{event.result['revised_prompt']}"            msg.elements = [image]            await msg.update()

同时我们还可以在Message里展示DALL-E-3返回的真实用来生成的revised_prompt，方便我们进行图片准确性的比对。