LLM应用开发框架llmflows：轻量级工作流编排实战指南-编程实验室

1. 项目概述：一个为LLM应用构建量身定制的轻量级框架

最近在折腾大语言模型应用开发的朋友，估计都经历过类似的“甜蜜的烦恼”：想法很美好，但真要把想法变成可运行、可维护的代码，中间隔着无数个坑。从Prompt的反复调试，到多个模型或工具的链式调用，再到状态管理和错误处理，每一步都可能让你从“调参侠”变成“debug狂魔”。正是在这种背景下，我注意到了GitHub上一个名为llmflows的开源项目。它的定位非常清晰：一个极简、灵活、面向开发者的Python框架，专门用于构建基于大语言模型的可靠应用流。

简单来说，llmflows不想成为另一个“全家桶”式的AI应用平台。它没有试图去解决所有问题，而是聚焦于一个核心痛点：如何优雅地编排和管理LLM调用之间的复杂依赖和数据流转。当你需要串联起多个提示词、多个模型（比如先用GPT-4生成大纲，再用Claude润色文本，最后调用DALL-E生成配图），或者需要将LLM调用与数据库查询、API请求等传统代码逻辑混合时，llmflows提供了一套直观的“乐高积木”式抽象。它用Flow和Step这两个核心概念，让你能以近乎绘制流程图的方式，声明式地定义应用的工作流，而框架则负责背后的异步执行、状态追踪和错误处理。

对我而言，llmflows最吸引人的地方在于它的“轻量级”哲学。它的代码库相当精炼，学习曲线平缓，你不需要先啃完几百页文档才能上手。它拥抱Python的原生特性，与asyncio异步生态良好集成，并且对流行的LLM API（如OpenAI、Anthropic）提供了开箱即用的支持。这意味着你可以快速地将一个实验性的Jupyter Notebook脚本，重构为一个结构清晰、易于扩展的生产级应用原型。接下来，我将深入拆解llmflows的设计思想、核心用法，并分享在真实项目中集成它时积累的一些实战经验与避坑指南。

2. 核心设计哲学：为什么是“Flow”与“Step”？

在深入代码之前，理解llmflows背后的设计哲学至关重要。这能帮助我们在正确的场景下使用它，而不是把它当成一个“银弹”。当今的LLM应用开发范式大致可以分为两类：一类是使用LangChain、LlamaIndex这类功能丰富的“重型”框架，它们提供了从文档加载、向量存储到智能体（Agent）的完整工具链；另一类则是直接从HTTP客户端调用API，然后在业务逻辑里用if-else和循环手动拼接一切。llmflows巧妙地找到了一个中间地带。

2.1 解决复杂编排的依赖地狱

想象一个内容生成场景：我们需要根据用户输入的主题，先让LLM生成一份内容大纲，然后基于大纲撰写详细文章，最后对文章进行语法和风格检查。如果用最原始的方式，代码可能长这样：

import openai def generate_content(topic): # Step 1: 生成大纲 outline_prompt = f"为‘{topic}’生成一份文章大纲。" outline = openai.ChatCompletion.create(model="gpt-4", messages=[{"role": "user", "content": outline_prompt}]) # Step 2: 撰写文章 article_prompt = f"根据以下大纲撰写一篇完整的文章：\n{outline.choices[0].message.content}" article = openai.ChatCompletion.create(model="gpt-4", messages=[{"role": "user", "content": article_prompt}]) # Step 3: 检查文章 check_prompt = f"检查以下文章的语法和风格，并提供修改建议：\n{article.choices[0].message.content}" feedback = openai.ChatCompletion.create(model="gpt-4", messages=[{"role": "user", "content": check_prompt}]) return { "outline": outline.choices[0].message.content, "article": article.choices[0].message.content, "feedback": feedback.choices[0].message.content }

这段代码的问题显而易见：逻辑耦合严重。每一步都硬编码在函数里，如果想调整顺序（比如先检查再撰写）、增加步骤（比如添加一个标题生成步骤）、或者替换模型（某一步用Claude），都需要直接修改函数体。这违反了“开闭原则”，也让单元测试变得困难。llmflows通过引入Step（步骤）的概念，将每个独立的LLM调用（或任何可调用对象）封装成一个具有明确输入和输出的单元。每个Step只关心自己的任务，并通过依赖声明来获取所需的数据。

2.2 声明式工作流与自动化的状态管理

llmflows的另一个核心概念是Flow（流）。一个Flow就是一个由多个Step构成的有向无环图（DAG）。你只需要定义好每个Step是什么，以及它们之间的依赖关系（即哪个Step的输出是另一个Step的输入）。之后，Flow对象会负责以正确的顺序执行这些步骤，并自动将上游步骤的输出传递给下游步骤作为输入。

这种声明式的编程模式带来了几个巨大优势：

可读性与可维护性：代码结构清晰地反映了业务逻辑的流程图，新人也能快速理解应用的数据流向。
可复用性：定义好的Step可以像乐高积木一样，在不同的Flow中重复组合使用。
内置的并发与异步：对于没有依赖关系的Step，llmflows可以自动并行执行，提升整体效率。
状态追踪与可观测性：框架会自动记录每个Step的输入、输出、开始和结束时间，甚至错误信息。这对于调试复杂流程和监控生产环境应用的状态至关重要。

注意：llmflows并不强制你使用特定的LLM提供商。虽然它提供了对OpenAI、Anthropic等主流API的便捷封装，但其Step的本质是一个接收输入字典、返回输出字典的异步函数。这意味着你可以轻松地将自定义函数、数据库查询、甚至其他微服务调用包装成一个Step，集成到流中。这种灵活性是它区别于某些“绑定过深”的框架的关键。

3. 快速上手：构建你的第一个LLM工作流

理论说得再多，不如动手跑一遍。让我们通过一个具体的例子，看看如何用llmflows实现上面提到的“大纲 -> 文章 -> 检查”三步走内容生成流程。假设我们已经配置好了OpenAI的API密钥（环境变量OPENAI_API_KEY）。

3.1 基础组件：LLM、提示词模板与步骤

首先，我们需要引入几个核心类：

from llmflows.flows import Flow, Step from llmflows.llms import OpenAI from llmflows.prompts import PromptTemplate

OpenAI：这是llmflows提供的LLM包装器。它封装了与OpenAI API的通信细节，让你可以专注于定义模型参数（如model_name,temperature等）。
PromptTemplate：提示词模板类。它允许你创建带有变量的提示词字符串，例如“为{topic}生成一份文章大纲”。在运行时，这些变量会被实际值替换。
Step：工作流的基本执行单元。一个Step需要绑定一个LLM（或任何callable）、一个提示词模板，并定义其输入变量来自哪里。
Flow：工作流容器，用于连接和运行多个Step。

3.2 分步构建工作流

接下来，我们一步步构建这个流。

第一步：定义LLM和提示词模板我们为所有步骤使用同一个GPT-4模型，但可以为不同步骤设置不同的temperature（创造性）。

# 1. 初始化LLM llm = OpenAI(model_name="gpt-4") # 2. 创建三个步骤的提示词模板 outline_template = PromptTemplate("为以下主题生成一份详细的内容大纲：{topic}") article_template = PromptTemplate("根据以下大纲，撰写一篇结构完整、内容详实的文章：\n{outline}") review_template = PromptTemplate("以专业编辑的身份，严格检查以下文章的语法、逻辑和风格，并提供具体的修改建议：\n{article}")

第二步：创建三个Step每个Step都需要一个唯一的名称、绑定的LLM对象、使用的提示词模板，以及它需要哪些输入变量。

# 3. 创建步骤 outline_step = Step( name="Outline_Generator", llm=llm, prompt_template=outline_template, output_key="outline" # 该步骤的输出将存储在变量“outline”中 ) article_step = Step( name="Article_Writer", llm=llm, prompt_template=article_template, output_key="article" ) review_step = Step( name="Article_Reviewer", llm=llm, prompt_template=review_template, output_key="feedback" )

第三步：创建Flow并连接Step这是最关键的一步，我们需要声明步骤之间的依赖关系。Flow的构造函数接受一个步骤列表，而依赖关系是通过在创建Step时指定input_variables（或在后续连接时）来定义的。更直观的方式是使用Flow的connect方法。

# 4. 创建Flow并连接步骤 flow = Flow(outline_step) # 以起始步骤初始化Flow flow.connect(outline_step, article_step) # outline_step的输出作为article_step的输入 flow.connect(article_step, review_step) # article_step的输出作为review_step的输入

flow.connect(source_step, dest_step)的含义是：dest_step的输入变量会自动从source_step的输出中寻找匹配的output_key。在我们的例子中，article_step的提示词模板需要变量{outline}，而outline_step的输出键正好是"outline"，因此连接后数据会自动传递。

第四步：运行Flow并获取结果最后，我们通过flow.start()方法来启动工作流，需要为起始步骤（outline_step）提供初始输入。

# 5. 运行Flow result = flow.start(topic="量子计算对现代密码学的挑战与机遇", verbose=True) # 6. 查看结果 print("生成的大纲：", result["outline"]) print("\n生成的文章：", result["article"]) print("\n审核反馈：", result["feedback"])

将verbose设置为True后，你会在终端看到框架打印的执行日志，清晰地展示了每个步骤的开始、结束和传递的数据，这对于调试非常有帮助。

3.3 第一个工作流的深度解析

通过这个简单的例子，我们已经触及了llmflows的核心使用模式。但其中有几个细节值得深入探讨：

输入变量的自动解析与传递：llmflows的依赖解析机制是其便利性的核心。当你连接两个步骤时，框架会检查目标步骤提示词模板中的所有变量（如{outline},{article}）。它会尝试从两个地方寻找这些变量的值：一是flow.start()提供的初始参数，二是所有已执行的上游步骤的output_key。这种自动匹配极大地减少了模板代码。
output_key的重要性：每个Step的output_key是其输出结果在流程全局命名空间中的“变量名”。它必须唯一，并且下游步骤通过这个键名来引用其输出。良好的命名习惯（如final_answer,summary_text）能让流程更清晰。
错误处理与重试：在实际应用中，LLM API调用可能因网络波动、速率限制等原因失败。基础的Step执行包含了简单的重试逻辑。但对于更复杂的错误处理（如根据错误类型选择不同降级方案），我们需要深入到Flow的异步执行器和自定义步骤逻辑中，这将在后续章节讨论。

实操心得：在初次设计Flow时，我建议先在纸上或白板上画出步骤图，明确每个步骤的输入和输出键。这能帮助你厘清数据流，避免出现循环依赖或缺失依赖。llmflows目前对循环依赖的检测可能有限，依赖良好的设计来避免。

4. 进阶用法：解锁框架的真正潜力

掌握了基础流程后，我们可以探索llmflows更强大的功能，以应对真实世界中的复杂需求。

4.1 多路径与条件执行

现实中的工作流很少是简单的直线。例如，一个客服问答流可能需要先判断用户意图（分类），然后根据不同的意图（如“退货”、“咨询”、“投诉”）走不同的处理子流程。llmflows通过支持更复杂的图结构来应对这种情况。

虽然llmflows本身不提供内置的“条件节点”或“开关”，但我们可以通过组合多个Flow或在一个Step内部实现逻辑判断来达到目的。一种模式是创建一个“路由步骤”（Router Step），该步骤的LLM调用专门用于分类，其输出是一个决定下一步走向的标签。

from llmflows.flows import Flow, Step # 假设已有几个处理不同意图的步骤 classify_step = Step(name="intent_classifier", llm=llm, prompt_template=intent_prompt, output_key="intent") handle_return_step = Step(name="handle_return", llm=llm, prompt_template=return_prompt, output_key="return_response") handle_inquiry_step = Step(name="handle_inquiry", llm=llm, prompt_template=inquiry_prompt, output_key="inquiry_response") # 创建两个独立的子流 return_flow = Flow(handle_return_step) inquiry_flow = Flow(handle_inquiry_step) # 在主逻辑中手动路由 async def main_flow(user_query): # 1. 分类 classification_result = await classify_step.run(user_query=user_query) intent = classification_result["intent"] # 2. 根据意图选择执行路径 if "return" in intent.lower(): result = await return_flow.start(classification_result) return result["return_response"] elif "inquiry" in intent.lower(): result = await inquiry_flow.start(classification_result) return result["inquiry_response"] else: return "抱歉，我暂时无法处理您的问题。"

这种方式将路由逻辑放在了Python代码层面，保持了灵活性。对于极其复杂的动态流，可能需要结合更高级的工作流引擎，但llmflows作为编排LLM调用的核心层，已经能处理大部分场景。

4.2 集成外部工具与函数调用

LLM应用不仅仅是调用大模型，经常需要与外部世界交互：查询数据库、调用API、执行计算等。llmflows的Step可以包装任何异步函数，使其无缝融入工作流。

假设我们有一个从数据库获取用户信息的函数：

import asyncpg from llmflows.flows import Step async def get_user_profile(user_id: int) -> dict: conn = await asyncpg.connect(DATABASE_URL) row = await conn.fetchrow("SELECT name, membership_level FROM users WHERE id = $1", user_id) await conn.close() return {"user_name": row["name"], "membership": row["membership_level"]} # 将异步函数包装成一个Step db_step = Step( name="Fetch_User_Profile", callable_fn=get_user_profile, # 关键参数：传入可调用对象 output_key="user_profile" ) # 这个Step可以像普通LLM Step一样被接入Flow # 下游的LLM Step可以通过 {user_profile} 来引用这个字典

关键点：当使用callable_fn时，该Step将不再需要llm和prompt_template参数。它的输入是传递给函数的参数，输出是函数的返回值。这极大地扩展了llmflows的边界，使其成为一个通用的任务编排框架。

4.3 异步执行与性能优化

默认情况下，flow.start()是同步的（尽管内部是异步的）。对于I/O密集型的LLM应用，充分利用异步可以大幅缩短端到端延迟。llmflows完全构建在asyncio之上，你可以直接使用其异步接口。

import asyncio from llmflows.flows import AsyncFlow, Step # 注意使用AsyncFlow async def main(): # 定义步骤（同上） outline_step = Step(...) article_step = Step(...) # 创建异步流 async_flow = AsyncFlow(outline_step) async_flow.connect(outline_step, article_step) # 异步运行 result = await async_flow.start(topic="异步编程的优势") print(result) # 运行异步主函数 asyncio.run(main())

性能提示：当流中存在多个互不依赖的步骤时，AsyncFlow会自动并发执行它们。例如，在一个总结流中，如果需要同时总结一篇文档的“核心观点”和“写作风格”，这两个总结步骤可以并行。确保在设计流程时，将可以并行的任务拆分成独立的Step，以最大化利用异步优势。

4.4 提示词管理的最佳实践

随着应用增长，提示词会散落在各个Step的定义中，难以维护。llmflows的PromptTemplate支持从文件加载，这有助于实现提示词的集中化管理。

# prompts/outline.txt 为以下主题生成一份详细的内容大纲： 主题：{topic} 要求： 1. 结构清晰，至少包含三级标题。 2. 每个主要部分下列出3-5个关键点。 3. 大纲语言为中文。 # 在代码中加载 from llmflows.prompts import PromptTemplate outline_template = PromptTemplate.from_file("prompts/outline.txt")

更进一步，可以建立一个提示词注册表或配置层，根据步骤名称动态加载模板，并与版本控制结合，实现提示词的版本化和A/B测试。

5. 实战配置与部署考量

将基于llmflows开发的应用从本地脚本推向可用的服务，需要考虑一系列工程化问题。

5.1 配置管理与环境隔离

绝对不要将API密钥等敏感信息硬编码在代码中。llmflows的LLM类（如OpenAI）通常会从环境变量读取配置。建议使用pydantic-settings或python-dotenv来管理配置。

# config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str anthropic_api_key: str | None = None database_url: str class Config: env_file = ".env" # .env 文件 OPENAI_API_KEY=sk-... ANTHROPIC_API_KEY=sk-ant-... DATABASE_URL=postgresql://user:pass@localhost/dbname # 在应用中使用 from llmflows.llms import OpenAI, Anthropic from config import settings llm_gpt = OpenAI(api_key=settings.openai_api_key, model_name="gpt-4") llm_claude = Anthropic(api_key=settings.anthropic_api_key, model_name="claude-3-sonnet") if settings.anthropic_api_key else None

5.2 日志记录、监控与可观测性

llmflows在verbose=True时会在控制台打印日志，但这对于生产环境远远不够。我们需要结构化的日志记录和链路追踪。

结构化日志：可以配置Python的logging模块，在自定义的Step执行逻辑中记录关键事件（开始、结束、错误、耗时、输入输出摘要）。建议记录到一个中心化的日志服务。
链路追踪：每个Flow和Step的执行都应该有一个唯一的trace_id。这个trace_id可以贯穿整个请求生命周期，方便在分布式系统中追踪问题。你可以在调用flow.start()时传入一个自定义的上下文字典，其中包含trace_id，并在每个步骤的日志中带上它。
监控指标：收集每个Step的耗时、令牌使用量、成功率等指标，这对于容量规划、成本控制和性能优化至关重要。可以考虑在Step的包装函数或装饰器中埋点。

5.3 错误处理与弹性设计

LLM服务的不稳定性是生产环境必须面对的挑战。llmflows的Step内置了简单的重试机制（通过其底层的HTTP客户端），但对于复杂的错误处理，我们需要更精细的控制。

策略一：步骤级重试与回退可以为重要的Step配置更积极的指数退避重试策略。如果主要模型（如GPT-4）持续失败，可以设计一个回退逻辑，自动切换到备用模型（如Claude或本地模型）。

from tenacity import retry, stop_after_attempt, wait_exponential from llmflows.llms import OpenAI, Anthropic class RobustStep(Step): def __init__(self, primary_llm, fallback_llm=None, **kwargs): super().__init__(llm=primary_llm, **kwargs) self.fallback_llm = fallback_llm @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) async def _run_with_retry(self, **inputs): try: return await super().run(**inputs) except Exception as e: # 捕获API错误、超时等 if self.fallback_llm and isinstance(e, (OpenAIError, TimeoutError)): self.llm = self.fallback_llm # 切换为备用LLM return await super().run(**inputs) # 用备用模型重试 else: raise

策略二：流程级熔断与降级对于整个Flow，可以设计一个顶层的熔断器。如果连续失败次数超过阈值，则暂时短路整个流程，直接返回一个预设的降级响应（如“系统繁忙，请稍后再试”），并报警通知开发人员。

5.4 部署模式：从脚本到服务

一个简单的llmflows应用可以作为一个独立的Python脚本运行。但对于长期运行或需要处理并发请求的服务，建议将其封装成Web API。FastAPI是一个极佳的选择，它能天然地支持asyncio。

from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from your_llmflows_module import create_content_flow # 导入你封装好的流程创建函数 import asyncio app = FastAPI() task_queue = asyncio.Queue() # 一个简单的内存队列，生产环境建议使用Redis或RabbitMQ class ContentRequest(BaseModel): topic: str request_id: str @app.post("/generate_content") async def generate_content(request: ContentRequest, background_tasks: BackgroundTasks): """异步处理内容生成请求""" # 将任务放入队列，立即返回请求ID await task_queue.put(request) return {"status": "accepted", "request_id": request.request_id, "message": "任务已加入队列"} # 后台工作进程 async def worker(): while True: request = await task_queue.get() try: flow = create_content_flow() result = await flow.start(topic=request.topic) # 将结果存储到数据库或缓存中，键为 request_id save_result(request.request_id, result) except Exception as e: save_result(request.request_id, {"error": str(e)}) finally: task_queue.task_done() @app.on_event("startup") async def startup_event(): # 启动后台工作协程 asyncio.create_task(worker())

这种异步任务队列的模式，将耗时的LLM流程与快速的HTTP响应解耦，避免了请求阻塞，提升了服务的吞吐量和用户体验。

6. 避坑指南与性能调优

在实际项目中使用llmflows一段时间后，我积累了一些宝贵的经验教训，这里分享出来，希望能帮你绕过一些常见的“坑”。

6.1 常见问题与排查技巧

问题1：步骤依赖解析失败，提示“Missing input variable: X”

原因：这是最常见的问题。下游步骤的提示词模板中引用了变量{X}，但框架在已执行的步骤输出和初始输入中找不到名为"X"的键。
排查：
1. 检查上游步骤的output_key是否确实设置为"X"（注意大小写）。
2. 检查flow.connect()的连接顺序是否正确。确保包含output_key="X"的步骤在下游步骤之前被连接和执行。
3. 如果变量来自flow.start()的初始参数，请确认参数名与模板变量名一致。
技巧：在开发阶段，将verbose设为True，仔细查看每个步骤执行前后的输入输出日志，能快速定位数据流断点。

问题2：流程执行速度慢，没有达到预期的并行效果

原因：llmflows只能对没有依赖关系的步骤进行并行。如果流程被设计成严格的串行链（A->B->C），那么它就无法并行。
优化：审视你的业务流程，看是否有步骤可以独立进行。例如，在生成报告的应用中，“获取数据A”、“获取数据B”、“获取数据C”这三个步骤如果没有依赖，应该被设计成三个独立的Step，然后同时连接到一个“生成报告”的步骤。这样，前三个数据获取步骤就可以并行执行。

问题3：API调用频繁超时或报错

原因：可能是网络问题、提供商速率限制（RPM/TPM）或服务不稳定。
解决：
1. 超时设置：在初始化LLM时，合理设置request_timeout参数（例如OpenAI(request_timeout=60)），避免因偶发网络延迟导致整个流程卡死。
2. 速率限制：严格遵守LLM提供商的速率限制。对于高并发场景，需要在应用层实现请求队列或使用令牌桶算法进行限流。llmflows本身不提供分布式限流，需要自行集成。
3. 指数退避重试：如前所述，使用tenacity等库为Step添加智能重试逻辑。

问题4：提示词模板中的变量被意外替换或格式混乱

原因：Python的字符串格式化与提示词模板的变量语法冲突。例如，如果提示词中包含{}或%s等字符。
解决：
1. 对于花括号{}，如果它不是变量，需要进行转义，或者使用PromptTemplate的替代语法（如果支持）。
2. 最稳妥的方式是将复杂的提示词保存在单独的.txt或.jinja2文件中，通过from_file加载。这既避免了转义问题，也便于管理和版本控制。

6.2 性能调优实战建议

批量处理：如果需要对大量独立项目（如1000条商品描述）执行相同的工作流，不要用for循环调用flow.start()1000次。应该改造你的Flow，使其初始输入是一个列表，并在Step内部使用LLM的批量处理能力（如果API支持），或者利用asyncio.gather并发运行多个Flow实例。注意遵守API的批量限制。
缓存中间结果：对于一些确定性高、计算成本高的步骤（例如，将一段固定文本翻译成另一种语言），可以考虑缓存其结果。可以为Step添加一个装饰器，根据输入参数的哈希值将输出缓存到Redis或本地磁盘，下次相同输入直接返回缓存结果。
精简上下文，管理令牌：LLM调用的成本和延迟与输入输出的令牌数直接相关。在设计流程时，要有意识地控制每个步骤传递的上下文大小。例如，如果上一步生成了一个很长的文档，下一步只需要其摘要，那么就应该插入一个“摘要步骤”，而不是将全文传递给下一步。
超时与取消：对于可能有长时间运行步骤的流程，实现超时和取消机制是必要的。可以利用asyncio.wait_for为整个flow.start()或单个step.run()设置超时，并在超时时优雅地取消任务，释放资源。

6.3 与现有技术栈的集成

llmflows不是一个孤岛，它需要融入你的现有技术生态。

与FastAPI/ Django集成：如前所述，将Flow包装成API端点。注意管理好每个请求的异步上下文和生命周期。
与Celery/ Dramatiq集成：对于需要离线处理的重任务，可以将flow.start()封装成一个Celery任务。确保你的任务函数和llmflows的代码都支持序列化，并且运行在兼容的环境中。
与向量数据库集成：虽然llmflows不直接提供RAG（检索增强生成）功能，但你可以轻松实现。创建一个Step，其callable_fn函数接收查询文本，调用向量数据库（如Pinecone, Weaviate）进行检索，并返回相关片段。然后将这个Step作为LLM生成步骤的上游，将检索结果作为上下文注入提示词。
与实验跟踪工具集成：为了迭代和优化提示词、比较不同模型，需要记录每次流程运行的详细信息（输入、输出、中间结果、成本、延迟）。可以将llmflows与MLOps平台（如Weights & Biases, MLflow）或自定义日志系统集成，在每个Step执行后发送跟踪事件。

llmflows作为一个专注且灵活的框架，为构建LLM应用提供了坚实的编排基础。它的价值在于让你从繁琐的流程控制代码中解放出来，专注于业务逻辑和提示词工程本身。随着应用的复杂化，你可能会需要在其上构建更多的工具和抽象层，但它的核心设计——清晰的依赖管理和声明式的工作流定义——始终是构建可靠、可维护AI应用的关键。