作者：修竹

在过去一年里，我们见证了LLM (大语言模型) 爆发式的增长，LLM的能力有了质的飞跃，也颠覆了所有开发者对“软件能力边界”的认知。只需要几行代码，调用一次LLM api接口，模型就能帮你写一段看起来像模像样的代码、总结一份结构清晰的文档或者回答一些“看起来很聪明”的问题。但当你试图想构建一个稳定、可复用、复杂的生产级别的AI应用时就会遇到：

• Prompt 失控：一开始只是几行提示词，后来变成了几百行规则说明。你不断往 Prompt 里“打补丁”，但模型依然会在某个边缘场景下给你一个完全不可用的结果。
• 结果不可预测（Non-deterministic）： LLM 是概率模型，而业务系统追求的是确定性。 “大概率对”在 Demo 阶段可以接受，但在审批、风控、数据查询这类场景中，等同于事故隐患。
• AI应用开发周期长，不能复用：新做一个 AI 应用，往往要重新写一套流程代码，反复的在从零开始造轮子。
• 调试困难：当用户反馈“刚刚还能用，现在不行了”，你却无法复现。 你不知道当时：Prompt 最终渲染成了什么？模型具体返回了哪一步异常？是模型波动，还是上下游数据变化？

这些问题叠加在一起，会把一个原本看起来“很有前途”的 AI 项目，迅速拖入不可维护的深渊。这也是AIWorks平台诞生的初衷，AIWorks不仅仅是一个简单的低代码开发工具，它是一个确定性的编排系统。本文将从工程师的角度带你了解一下AIWorks平台中workflow的设计与实现。

1.DAG为骨架

复杂的业务逻辑，如果不加整理，往往是一团乱麻的代码（Spaghetti Code）。我们将业务逻辑抽象为数学上的 DAG（有向无环图）。

• Node（节点）：代表一个原子的计算单元。它可能是一次 LLM 的推理，也可以是一段 Python 代码的执行，或者是一次 HTTP 请求。
• Edge（边）：代表数据的流向和执行的顺序。

这种设计使得业务逻辑可视化。前端拖拽生成 JSON，后端解析执行 JSON。所见即所得，对非技术人员来说非常的友好。

2.状态机为灵魂

很多工作流系统，本质上只是任务编排器，节点按顺序执行，执行完就结束。但AI Workflow的行为模式具备以下特征：

• 多轮交互
• 上下文强依赖
• 当前行为取决于“之前发生了什么”

这就要求AI Workflow不是一个简单的线性流程，而是一个状态不断迁移的系统。因此，在 AIWorks 中，我们将工作流视为一个状态机（State Machine），每一次节点执行都会引起 Graph State 的变化，下一步的走向，取决于当前的状态。

3.一切皆节点

在AIWorks的workflow设计中，节点作为最小执行单元，无论是调用LLM，还是执行一段简单的Python代码，还是调用高德地图API，它们都被抽象成节点。所有节点都继承自同一个基类 BaseNode，Workflow执行引擎不关心节点“做了什么”，只关心节点是否成功，产出了什么结果。这样可以极大的提高系统的扩展能力，如果需要新增一种新能力(比如给飞书发消息)，那么你只需要继承BaseNode，然后实现其中对应的方法就能无缝接入到现有的系统中，享受工作流引擎提供的所有能力(重试、 日志、变量注入等)。

4.可观测性优先

不是“出问题了再加日志”，而是“天生可被回放”。为了让AI Workflow系统不再是“黑盒”，我们将可观测性作为核心设计原则之一。

引擎会自动记录工作流中每一个节点（Node）的完整执行快照，包括：

• 输入（Inputs）：节点接收到的变量和参数。
• 输出（Outputs）：节点运行后的产出结果。
• 状态变化（Status Changes）：从开始、运行中到结束的每一刻。

这种精细粒度的记录，使得开发者可以在工作流执行完成后，像看“即时回放”一样，逐帧查看执行过程。一旦出现问题，通过回溯输入输出，就能迅速定位是哪个环节的 Prompt 写得不对，还是哪个 Tool 调用参数传错了，真正做到“有迹可查”。

Workflow引擎架构解析

1.整体分层

AIWorks 工作流引擎采用了经典的分层架构：

应用层（Application Layer）    ↓图引擎层（Graph Engine Layer）    ↓节点层（Node System Layer）    ↓基础设施层（Infrastructure Layer）

• 应用层：提供 API 接口，处理用户请求
• 图引擎层：负责工作流的编排和执行
• 节点层：实现各种能力单元（LLM、工具、知识检索等）
• 基础设施层：提供底层服务（模型调用、向量检索、工具运行时等）

这样分层的好处是关注点分离。比如你要换个 LLM 提供商，只需要改基础设施层；要加个新节点类型，只需要在节点层扩展，不会影响引擎核心逻辑。

2.Graph Engine：整个系统的“心脏”

GraphEngine是 AIWorks 工作流引擎的核心调度器，它的职责可以概括为三件事：

• 解析前端传入的工作流JSON
• 将其编译为可执行的 Graph App
• 驱动整个执行生命周期(初始化状态、执行Graph、输出结果并保存状态)

（1）静态图构建：业务逻辑的“蓝图”

前端通过拖拽或配置生成的流程，本质上是一份 JSON 描述的 DAG。在 AIWorks中，我们并不会直接将这份 JSON 交给执行引擎，而是定义类Graph对其进行装载和校验，并解析JSON中的节点(Node)和边(Edge)。GraphEngine使用LangGraph作为工作流的执行引擎，GraphEngine将 Graph 对象转换为 LangGraph 提供的 StateGraph，并将Graph中的节点(Node)和边(Edge)动态设置到StateGraph中。

简化后的核心逻辑如下：

# 伪代码示意class GraphEngine:        def __init__(self, ..., graph: Graph):        self._graph = graph        ...    def _build_graph_app(self, state_context):            workflow = StateGraph(GraphRuntimeState)            node_id_config_mapping = self._graph.node_id_config_mapping
            # 1. 动态添加节点            for node_id, node_data in node_config_mapping.items():                wrapper_node = self._create_command_node(node_id, node_data, state_context)                workflow.add_node(node_id, wrapper_node)            # 2. 动态添加边            for edge in edges:                workflow.add_edge(edge.source, edge.target)
            return workflow.compile()

StateGraph设置完成后，会调用compile()进行编译，在这个阶段会对整个图进行结构校验，包括是否存在环、是否有孤立节点或不可达节点和节点和边的引用是否完整等。

（2）异步调度与事件流

GraphEngine的执行核心并非简单的循环，而是基于 LangGraph 提供的 astream 接口实现的异步事件流处理。事件类型包括FLOW_START、NODE_START、NODE_END、FLOW_END。

# 伪代码示意async def _run_workflow(self, inputs: GraphGenerateEntity, enable_run_log: bool):    graph_app = self._build_graph_app(state_context)    state = self._init_graph_state(inputs)    # ... 初始化状态 ...    yield GraphEvent(event=GraphEventEnum.FLOW_START, run_id=run_id)
    # 订阅 LangGraph 的流式输出，关注 "custom" (自定义事件) 和 "tasks" (节点状态)    event_stream = graph_app.astream(state, stream_mode=["custom", "tasks"])
    async for event_tuple in event_stream:        stream_mode, event_message = event_tuple
        # 将 LangGraph 的内部事件转换为 aiworks 的标准 GraphEvent        if stream_mode == "tasks":            # 处理节点启动/结束            yield GraphEvent(event=GraphEventEnum.NODE_START, ...)        else:            # 透传自定义事件 (如 LLM Token)            yield GraphEvent(**event_message)
    yield GraphEvent(event=GraphEventEnum.FLOW_END, ...)

（3）状态持久化

在工作流执行结束后，引擎会自动进行“快照存档”。这种设计不仅仅是保存 Log，它保存了完整的运行时状态（GraphRuntimeState）。这意味着：

• 可溯源：你可以随时打开一个历史运行记录，看到当时图的结构（哪怕现在的图已经改了）以及每个节点的输入输出。
• 可恢复（未来及展望）：这种结构为未来的“断点续跑”和“人工介入”功能打下了数据基础。

3.状态管理（State）：可观测的执行过程

工作流执行过程中,最重要的就是状态管理。我们设计了三层状态结构：

（1）变量池（VariablePool）

这是整个工作流的"记忆"，存储所有变量：

class VariablePool(BaseModel):    user_inputs: dict  # 用户输入的变量    system_inputs: dict  # 系统变量（如 conversation_id）    pool: dict  # 节点执行过程中产生的变量

节点执行时，可以从pool中读取前置节点的输出，也可以把自己的输出写入pool，供后续节点使用。

（2）节点状态（NodeState）

记录每个节点的执行状态：

class NodeState(BaseModel):    id: str    status: NodeStatus  # pending/running/succeeded/failed    inputs: dict  # 节点输入    result: dict  # 节点输出    start_at: datetime    finished_at: datetime    error_msg: str

这里有个细节：我们会完整记录节点的输入和输出。这样做的好处是：

• 执行回放：根据 inputs 可以重新执行节点，复现问题
• 调试：可以清楚看到每个节点的输入输出，快速定位问题
• 性能分析：通过 start_at 和 finished_at 计算耗时，找出瓶颈

（3）工作流运行时状态（GraphRuntimeState）

整个工作流的全局状态：

classGraphRuntimeState(BaseModel):    query:str    variable_pool: VariablePool    node_state_mapping: dict[str, NodeState] # 所有节点状态    routes: dict[str, list[str]] # 实际执行路径    status: GraphStatus # running/succeeded/failed    output:str|dict # 最终输出

执行完成后，我们会把GraphRuntimeState 序列化成 JSON，存储到数据库。这样就有了完整的执行记录，方便后续分析和优化。

节点系统：可扩展的能力单元

如果说图引擎是工作流的"大脑"，那节点系统就是"四肢"——具体干活的地方。

1.节点抽象：模板方法模式的实践

所有节点都继承自BaseNode，它定义了节点的生命周期：

class BaseNode:    def __init__(self, node_id, node_data, user_id, tenant_id, graph):        self.node_id = node_id        self.node_data = node_data        self.init_node()
    def init_node(self):        config = self.resolve_node_data()        self.init_node_config(config)
    @abstractmethod    def _run(self, state) -> NodeResult:        raise NotImplementedError
    async def run(self, state):        return await self._run(state)

这是典型的模板方法模式：

• init_node() 定义了初始化流程
• 子类只需要实现 init_node_config() 和 _run() 两个方法

这样做的好处是统一了节点的初始化流程，子类只需要关注自己的核心逻辑。

2.节点概览：工作流的能力单元

①Start 节点：工作流的入口节点，标识整个流程的起点，每个工作流都必须有一个 Start 节点，它负责接收用户的输入变量，并将它们传递给后续节点。

②LLM 节点：调用大语言模型进行文本生成、对话、摘要等任务。这是使用最频繁的节点，支持友好的提示词管理、记忆管理、多种LLM提供商等。

③Knowledge Retrieval 节点：从向量数据库中检索相关知识文档。典型应用场景是 RAG（检索增强生成），先检索相关知识，再传给 LLM 节点进行回答。

④Tool 节点：工具节点是与外部环境交互的接口，支持内置工具、API工具和自定义工具。

⑤IF-Else 节点：根据条件动态选择执行路径, 是实现复杂业务逻辑的关键。支持：

• 多种比较操作符（等于、包含、为空等）
• AND / OR 逻辑组合
• 多分支（IF / ELIF / ELSE）
• 基于变量池中的任意变量做判断

⑥Code 节点：执行用户自定义的代码逻辑。适合处理复杂的数据转换、计算逻辑等 LLM 不擅长的任务。

⑦HTTP 节点：发起 HTTP 请求，调用外部 API。

⑧Answer 节点：格式化最终输出，返回给用户。

3.扩展新节点：三步走

当我们需要新增节点类型时, 流程很简单：

（1）定义配置类：继承类BaseNodeConfig

lass MyNodeConfig(BaseNodeConfig):    param1: str    param2: int

（2）实现节点类：继承BaseNode

class MyNode(BaseNode):    _node_type = NodeType.MY_NODE
    def init_node_config(self, valid_config):        self.node_config = MyNodeConfig(**valid_config)
    def _run(self, state):        # 实现具体逻辑        return NodeResult(result={...})

（3）注册节点类型：在 NodeType 枚举和工厂方法中注册

这个设计让节点系统具备了很强的扩展性，每个节点并遵守单一原则，这样就能保证新增节点时效率高，还能保持引擎代码的稳定。

总结

回顾整个工作流引擎的开发过程,最大的感受是：好的架构设计真的能事半功倍。

我们在 AIWorks 中遵循的几个核心理念：

• 分层解耦：图定义、执行引擎、节点实现各自独立
• 抽象优先：基于接口编程,易于扩展
• 可观测性：完整记录执行链路,方便调试和优化
• 异步流式：提升用户体验和系统吞吐

目前这套系统已经在生产环境中稳定运行，支撑了多个企业级应用。当然,还有很多可以优化的地方：

• 可视化调试器：图形化展示执行流程和状态变化
• 分布式执行：支持大规模工作流的分布式调度
• 智能优化：基于历史数据,自动优化节点配置

希望这篇文章能对正在做类似系统的同学有所帮助。如果你有任何问题或建议，欢迎留言交流！