如果你最近在做 LLM 相关的产品大概率会频繁遇到一个词多智能体工作流。单 Agent 问答已经不算新鲜真正让团队头疼的是当业务被拆成“取数—分析—写报告—审核”四个步骤并且每天凌晨都要自动跑一遍时开发复杂度会迅速从“一个脚本”膨胀成“一套系统”。更麻烦的是这套系统不能只停留在“能跑”。它还需要被设计、被调试、被运营节点之间传什么数据失败时怎么重试某一步 Agent 的输出格式变了如何发现。这些问题在纯代码环境下往往只能靠日志和脑补来处理。于是“可视化工作区Visual Workspace”这种交互形态开始被关注。最近 Hacker News 的 Show HN 板块就有项目打出了这样的定位用可视化工作区来设计和运营 daily multi-agent workflows。这篇文章不打算只评价某个具体产品而是借这个方向把多智能体工作流的“设计—编排—运行—运营”完整链路拆解一遍并给出一套最小可运行的代码实现。读完你会明白一个判断可视化工作区的核心价值不在“画图好看”而在于让一套分布式、不确定、周期运行的 Agent 系统变得可理解、可控制、可运营。1. 为什么多智能体工作流需要可视化工作区先看一个最常见的场景。假设你要构建一个每日运营报告系统流程并不复杂先采集销售数据再做趋势分析然后生成 Markdown 报告最后交给审核节点确认。用代码实现这个流程几百行就能搞定。但一旦把其中每一步替换成“某个大模型 Agent”问题就出现了。1.1 多 Agent 系统的复杂度来自哪里单 Agent 应用里你只需要关注一次请求的输入输出。多 Agent 工作流则完全不同多个 Agent 之间产生了依赖关系、数据传递和状态流转。任何一个 Agent 的输出格式变化、一次偶发超时、一次 API 配额耗尽都可能导致下游连锁失败。这种复杂度通常体现在三个层面设计态Agent 之间怎么连线、谁先谁后、哪些步骤可以并行需要在一开始就梳理清楚。调试态某个节点的输出不符合预期时你要能完整看到这一条链路上的中间结果。运营态工作流是每天跑的不是跑一次就完。需要调度、监控、告警、重跑、动态调整。纯代码方案并不是不能做这三件事但它的表达成本很高。你需要在代码里维护拓扑结构、在日志里追踪状态、在告警规则里描述每个节点的失败语义。而“可视化工作区”要做的正是把这三件事重新收拢到一个图形界面里。1.2 没有可视化时团队怎么协作可以设想一个没有可视化工作区的团队协作过程算法工程师写好每个 Agent 的调用逻辑后端工程师把它们串成一个 Pipeline测试人员靠日志推断中间结果业务方想调整报告模板只能提工单等版本迭代。这种模式的问题在于Agent 工作流本质上是“业务逻辑”和“AI 能力”的混合体。业务方最关心的“报告格式”“审核条件”被埋在了代码深处非技术人员无法参与设计而技术团队最关心的“状态流转”“失败原因”又被日志切割得支离破碎。沟通成本会随着 Agent 数量上升而急剧增加。1.3 可视化工作区改变了什么可视化工作区的核心价值是让工作流从“只能被程序员阅读的代码”变成“能被业务方、测试方、运维方共同看见和操作的实体”。它相当于给多 Agent 系统配备了一个“IDE 控制台”的组合体画布上拖拽节点是在设计点击节点查看输入输出是在调试看运行时间线和告警列表是在运营。对这个方向的项目来说难点不在于画布渲染而在于背后必须有一套真正能执行、能记录、能追溯的工作流引擎。所以写这篇文章时我不会只停留在产品介绍而是要带你落地一个最小工作流引擎。这个引擎是所有可视化工作区产品的“心脏”。2. 核心概念Multi-Agent Workflow 与 Visual Workspace在进入代码之前有必要把几个概念放在一起讲清楚。很多读者分不清“多智能体系统”“多智能体工作流”“可视化工作区”和“工作流执行引擎”之间的区别这会影响后续实践中的选型判断。2.1 多智能体工作流一支 AI 流水线团队多智能体系统Multi-Agent System指的是多个 Agent 在同一任务中各自承担角色、相互协作。这里的 Agent 可以理解为一个“具有大模型能力、能调用工具、能做出决策”的独立单元。多智能体工作流Multi-Agent Workflow则更强调“流程化”把 Agent 之间的协作过程固化成可重复执行的步骤包含明确的依赖关系和状态传递。你可以把它想象成一支流水线团队数据采集员只负责取数分析师只看数据做洞察撰稿人负责把洞察写成报告审核员负责把关质量。每个人职责清晰但只有串成一条流水线才能真正完成“每日运营报告”这件事。与传统自动化流水线的差别在于传统流水线每个节点的行为是确定的输入相同输出就相同而 Agent 节点输出不稳定、延迟不稳定、失败模式也不稳定。这导致工作流设计必须额外考虑容错、校验和人工介入。2.2 可视化工作区Agent 系统的中央控制室可视化工作区Visual Workspace是以图形界面为载体的工作流设计与运营环境。用户通过拖拽节点、连线、配置参数来定义工作流而不是直接编写代码。“可视化”在这里不是目的而是手段。它真正解决的是认知负担问题当工作流有十几个节点、几十条依赖关系时图形化表达远比阅读代码更直观。更进一步运营人员可以在同一个界面上看到当前运行实例处于哪个节点、哪一步耗时长、哪一步失败率最高。2.3 图、节点、边与执行引擎几乎所有可视化工作区底层都遵循同一个抽象模型图Graph整个工作流的结构由节点和边组成。节点Node一个 Agent 或一个工具调用。边Edge节点之间的依赖关系和数据传递方向。执行引擎Engine读取图的定义按依赖关系调度节点维护运行状态和上下文。这个模型并不神秘。一篇 YAML 或 JSON 配置本质上就是一张图的序列化结果。可视化画布只是这张图的“渲染层”真正驱动工作流运转的是背后的执行引擎。2.4 可视化工作区与流程图、传统工作流引擎的区别类型能否执行编辑方式是否面向 Agent 场景典型代表普通流程图否纯图形否只表达设计意图Visio、draw.io传统工作流引擎是代码/DSL 为主偏向确定型任务Airflow、TemporalAgent 可视化工作区是图形 配置是支持 LLM 节点、人工审批本文方向及部分新产品从表中能看出Agent 可视化工作区处于“图形化 可执行 面向 LLM 复杂场景”的交叉点。这也是它区别于普通画图工具的本质。3. 可视化工作区适合谁、不适合谁任何技术方案都有边界。可视化工作区听起来很理想但并不意味着所有多 Agent 项目都应该上可视化。3.1 适合的团队和场景第一类是探索期团队。当你们还不确定工作流最终形态时可视化方式能大幅缩短调整周期。改一条连线、加一个分支比改代码、重新部署要快得多。第二类是运营型固定流程。比如每天的日报生成、每小时的数据巡检、每周的竞品监控。这类流程“稳定重复、需要被长期运营”可视化工作区能帮助团队在运行时快速定位问题。第三类是跨角色协作场景。如果业务方、运营方需要频繁参与 Agent 流程的设计和审核可视化工作区几乎是必需品。比如“流程中需要人工确认再继续执行”这类 human-in-the-loop 节点用画布表达远比用代码表达清楚。3.2 不适合的场景如果你的任务只是两个 Agent 之间的简单串联写脚本反而更高效。可视化工作区的引入本身就带有一定的平台成本和维护成本不要为了“高级”而过度设计。如果系统对延迟极度敏感比如用户请求链路中的实时调用那么每次经过可视化平台的调度反而会引入额外开销。这种场景更适合直接代码调用而不是图执行引擎。另外如果你的团队完全在离线环境工作、没有图形界面可用那可视化工作区也无从谈起。这种情况下基于配置文件的工作流定义和纯命令行工具会是更合适的选择。3.3 选择判断我的建议是先用手写代码跑通最小闭环确认工作流真的有“被反复调试和运营”的需求之后再考虑上可视化工作区或成熟平台。可视化解决的从来不是“能不能跑”而是“跑起来之后能不能看清楚、能不能控制住”。4. 环境准备与最小可运行架构由于本文展示的是自研最小实现环境要求很低。你不需要安装任何重量级框架只需要一个可运行 Python 3 的机器。4.1 运行环境操作系统Windows / macOS / Linux 均可。Python3.10 或更高版本本示例只用标准库无第三方依赖。开发工具任意文本编辑器或 IDE。可选如果你希望引擎直接读取 YAML 配置文件可以安装pyyaml但本文为了减少环境依赖引擎代码中直接使用字典结构与 YAML 结构保持一致。pip install pyyaml # 可选4.2 项目目录结构建议按下面的结构组织代码。把“Agent 实现”和“工作流配置”分开是可视化工作区的一个重要设计原则图结构与节点逻辑解耦才能让画布编辑和代码迭代互不干扰。multi-agent-workflow-demo/ ├── agents/ │ ├── __init__.py │ └── agents.py ├── engine/ │ └── engine.py └── config/ └── workflow_config.yaml4.3 模块职责agents/agents.py定义每个 Agent 的 handler对应一个节点行为。engine/engine.py包含一个最小工作流执行引擎读取配置、调度节点、维护上下文。config/workflow_config.yaml工作流图的配置样例是人读的工作流定义也是未来接入可视化画布的桥梁。4.4 设计原则这个最小架构遵循三个原则配置即工作流节点和依赖用数据表达不写死在 Python 业务代码里。逻辑可替换Agent 内部是调用大模型、调用工具、还是用规则函数由实现方决定引擎不关心。状态显式化每个运行实例都有统一的上下文记录节点输入输出和状态方便后续做 trace 追踪。这三个原则也是你在评估任何可视化工作区产品时应该检查的核心能力。5. 完整示例一个“每日运营报告”多 Agent 工作流下面我们用代码实现一个完整的“每日运营报告”工作流。它包含四个节点数据采集、趋势分析、报告撰写、内容审核完全对应标题中 daily multi-agent workflows 的日常运营场景。5.1 工作流图配置先看配置文件。这份 YAML 描述了工作流的完整拓扑结构。在可视化工作区里这张图通常由拖拽连线生成最终导出的也是类似的结构。# 文件路径config/workflow_config.yaml name: daily_operation_report description: 每日运营数据采集、分析、报告生成与审核工作流 schedule: 0 9 * * * # 每天9点执行留给调度器参考 nodes: - id: collect_data agent: collect_data_agent description: 从多个数据源采集前一天运营数据 - id: analyze_trend agent: analyze_trend_agent description: 基于采集数据产出核心洞察 depends_on: [collect_data] - id: write_report agent: write_report_agent description: 根据洞察撰写运营日报 depends_on: [analyze_trend] - id: review_report agent: review_report_agent description: 检查报告结构、口径与风险信息 depends_on: [write_report]这份配置里depends_on定义了节点依赖。只有depends_on中列出的上游节点成功完成后当前节点才会执行。这种结构可以直接映射为一张有向无环图DAG。5.2 每个 Agent 的实现agents.py中实现了四个 Agent。为了让你在没有模型 API Key 的情况下也能完整跑通流程本文用确定性函数模拟 Agent 行为。真实项目中你只需把函数内部的逻辑替换为 LLM 调用接口不变。# 文件路径agents/agents.py 各 Agent 的实现。 真实项目中这里通常是对 LLM API 的调用封装 本演示使用确定性函数方便在没有模型 Key 的情况下跑通编排逻辑。 import random def collect_data_agent(payload: dict) - dict: 数据采集 Agent。 真实场景调用数据接口、查询数仓或让另一个 Agent 去抓取数据。 date payload.get(date, 2025-01-01) return { date: date, gmv: round(128000 random.uniform(-5000, 5000), 2), order_count: 824, channel: { search: round(62.5, 1), recommend: round(24.3, 1), other: round(13.2, 1), }, } def analyze_trend_agent(payload: dict) - dict: 趋势分析 Agent。 真实场景把 collect_data 的输出交给大模型并要求返回结构化分析结果。 data payload[collect_data] gmv data[gmv] trend 增长 if gmv 120000 else 平稳 return { trend: trend, highlight: f昨日 GMV {gmv}渠道搜索占比最高, risk: 需关注退款率异常波动 if random.random() 0.5 else 无显著风险, } def write_report_agent(payload: dict) - dict: 报告撰写 Agent。 真实场景把分析结论交给大模型按团队模板生成 Markdown 报告。 analysis payload[analyze_trend] data payload[collect_data] report f# 每日运营报告{data[date]} ## 核心结论 {analysis[trend]}{analysis[highlight]}。 ## 风险提示 {analysis[risk]} ## 原始数据摘要 - GMV{data[gmv]} - 订单数{data[order_count]} return {report: report} def review_report_agent(payload: dict) - dict: 审核 Agent。 真实场景用规则或另一个 LLM 对报告做质检判断是否能发布。 report payload[write_report][report] if 每日运营报告 in report and 核心结论 in report: return {passed: True, comment: 报告结构完整通过审核} return {passed: False, comment: 报告结构不完整需要重写} AGENT_MAP { collect_data_agent: collect_data_agent, analyze_trend_agent: analyze_trend_agent, write_report_agent: write_report_agent, review_report_agent: review_report_agent, }关键点在于每个 Agent 函数的输入payload都包含了上游节点的输出返回值会被注册到全局上下文中。这种约定保证了节点之间数据传递的一致性也方便可视化工作区把每个节点的输入输出直观展示出来。5.3 最小工作流执行引擎engine.py是整套示例的核心。它做的事情和可视化工作区后台的调度执行器是同一件事读配置按依赖顺序执行节点把上游输出组装成下游输入最后汇总运行状态。# 文件路径engine/engine.py 一个极简的多 Agent 工作流执行引擎。 功能 - 按配置顺序执行各 Agent 节点 - 自动把上游节点的输出作为下游节点的输入 - 每个节点记录成功/失败状态 - 支持节点级异常捕获。 这是可视化工作区后台执行器的“最小可运行”版本。 import os import sys from datetime import datetime # 保证可以从项目根目录导入 agents 包 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from agents.agents import AGENT_MAP # 演示用配置结构等同于 config/workflow_config.yaml # 可视化工作区通常会把这张图保存为 yaml/json再由引擎加载 WORKFLOW_CONFIG { name: daily_operation_report, description: 每日运营数据采集、分析、报告生成与审核, nodes: [ {id: collect_data, agent: collect_data_agent, depends_on: []}, {id: analyze_trend, agent: analyze_trend_agent, depends_on: [collect_data]}, {id: write_report, agent: write_report_agent, depends_on: [analyze_trend]}, {id: review_report, agent: review_report_agent, depends_on: [write_report]}, ], } def build_payload(node: dict, context: dict) - dict: 把当前节点依赖的上游输出合并进 payload。 payload {} for dep_id in node.get(depends_on, []): dep_result context[results].get(dep_id) if dep_result and dep_result[status] success: payload[dep_id] dep_result[data] return payload def run_workflow(initial_inputs: dict) - dict: 执行整个工作流返回包含上下文、节点结果的字典。 context { status: running, started_at: datetime.now().isoformat(), inputs: initial_inputs, results: {}, } for node in WORKFLOW_CONFIG[nodes]: node_id node[id] agent_name node[agent] handler AGENT_MAP.get(agent_name) if handler is None: context[results][node_id] { status: failed, error: funknown agent: {agent_name}, } context[status] failed break # 全局输入先注入再注入上游节点输出 payload dict(initial_inputs) payload.update(build_payload(node, context)) print(f[{datetime.now().isoformat()}] 开始节点: {node_id} - {agent_name}) try: data handler(payload) context[results][node_id] {status: success, data: data} print(f[{datetime.now().isoformat()}] 节点成功: {node_id}) except Exception as exc: context[results][node_id] {status: failed, error: str(exc)} context[status] failed print(f[{datetime.now().isoformat()}] 节点失败: {node_id}, error{exc}) break if context[status] running: context[status] succeeded context[finished_at] datetime.now().isoformat() return context def main(): ctx run_workflow({date: 2025-06-17}) print(\n 工作流执行汇总 ) print(workflow:, WORKFLOW_CONFIG[name]) print(status:, ctx[status]) print(started_at:, ctx[started_at]) print(finished_at:, ctx[finished_at]) print(\n各节点输出摘要:) for node_id, result in ctx[results].items(): if result[status] success: print(f - {node_id}: success, 输出字段{list(result[data].keys())}) else: print(f - {node_id}: failed, error{result.get(error)}) if __name__ __main__: main()这段代码最重要的设计是context。它保存了整个工作流运行实例的输入、每个节点的结果和最终状态。可视化工作区的运行时间线、节点详情、失败告警本质上都是对这个