跳到主要内容

总览:PySpur 是什么 · 全景 · 阅读地图

30 秒导读: PySpur 是一个"用 Python 搭 AI Agent 的可视化平台"——你在浏览器里拖拽节点、连线,画出一张有向无环图(DAG),这张图被存成一份 JSON,后端拿到后按依赖顺序、异步并发地逐个执行每个节点,每个节点用 litellm 等库真正去调大模型 / 工具 / 向量库。本章只做路由和全局认知:讲清它是什么、大盘怎么转、各部件在哪个文件夹、一条请求端到端走一遍,然后给出后续 5 章的阅读地图。

1. 这是什么(零基础也能懂)​

一句话定义: PySpur 是一个 Graph UI for building AI Agents in Python——把"搭 AI Agent"这件事从"写一堆散乱的 prompt 和胶水代码"变成"在画布上画一张流程图"。

这个定位不是我猜的,它直接写在包的元数据里:

  • 包名与描述:backend/pyproject.toml:6-8 —— name = "pyspur"、description = "PySpur is a Graph UI for building AI Agents in Python"、当前 version = "0.1.18"。
  • README 开头的一句话卖点:README.md:3 —— "AI engineers use PySpur to iterate over AI agents visually without reinventing the wheel."

解决什么问题 / 给谁用: 给做 AI Agent 的工程师用。README 把痛点总结成三样(README.md:30-34):调 prompt 调到手软("Prompt Hell")、看不见步骤之间怎么互相影响("Workflow Blindspots")、只能盯着终端里的原始 JSON 硬看("Terminal Testing Nightmare")。PySpur 的答案是给 Agent 一个"游乐场":可视化地搭、可视化地调、逐节点看中间结果。

它靠什么技术栈实现: 后端是一套典型的 Python 服务组合,从依赖清单一眼可知(backend/pyproject.toml:25-82):

依赖干什么
fastapi对外的 HTTP API 层(接收前端画布、跑工作流)
litellm统一封装 100+ 家大模型的调用接口——节点靠它真正"干活"
pydantic描述并校验工作流结构、节点配置、节点输入输出
alembic + SQLAlchemy数据库迁移与 ORM(存工作流、运行记录、task 记录)

用起来什么样: 装好后是一条命令起服务(README.md:75-91):

pip install pyspur
pyspur init my-project        # 生成一个带 .env 的项目目录
pyspur serve --sqlite         # 默认在 http://localhost:6080 起 App(sqlite 库)

然后打开浏览器,在画布上拖节点、连线、点运行。你也可以不碰 UI、纯用 Python 代码搭同一张图(README.md:44、68:"Add new nodes by creating a single Python file")——UI 和代码是同一份图的两种编辑方式。

一句话直觉/类比: 把它想成"AI Agent 版的 n8n / 节点式音频软件"——画布上的每个框是一个会调模型或工具的函数,连线就是数据流;你画完的图不是截图,而是一份能被后端真正执行的程序。

本节到此不碰代码细节。记住一句就够:画布上的图 = 一份可执行的 DAG。

2. 三种 Spur:workflow / chatbot / agent​

"Spur"(马刺,项目名的来源)是 PySpur 对"一张可执行的图"的统称。同一套画布 + 执行引擎,按用途分成三种类型,由一个枚举 SpurType 定义(backend/pyspur/schemas/workflow_schemas.py:12-23):

类型枚举值一句话
WorkflowSpurType.WORKFLOW标准工作流:一张普通的节点 + 连线 DAG
ChatbotSpurType.CHATBOT本质还是工作流,但 IO 是聊天式的,并带会话(session)管理
AgentSpurType.AGENT自主 Agent 节点:会调工具、循环,同样有聊天式 IO 和会话

这三者不是三套代码,而是同一份 WorkflowDefinitionSchema 上的一个字段(spur_type,workflow_schemas.py:104)。区别只在于多加几条校验/行为:例如 chatbot 类型会强制要求输入节点带 user_message、session_id 字段,输出节点带 assistant_message 字段(校验逻辑见 workflow_schemas.py:157-236 的 validate_chatbot_input_node / validate_chatbot_output_node);执行时 chatbot 还会自动注入/存储消息历史(workflow_executor.py:877-916)。

数据模型的细节看 01-workflow-model.md;chatbot/agent 执行时的差异看 02-executor.md 与 05-llm-and-agent.md。

3. 顶层全景(它大概怎么转)​

怎么读这张图: 从左到右是一条主线——人在前端画布上画图,图被序列化成 JSON 存进后端,运行时后端把 JSON 反序列化成对象、交给执行器按依赖顺序跑,每个节点用 litellm 等干活,过程写进 task 记录。

  ┌─────────────────────┐        画 DAG              ┌──────────────────────────┐
  │  前端 React 画布      │  拖拽节点 + 连线 ──────────▶ │  WorkflowDefinitionSchema │
  │  frontend/src        │                            │  { nodes[], links[] }     │
  │  (canvas / store)    │ ◀──── 逐节点结果回显 ────    │  (一份 JSON,可存库)       │
  └─────────────────────┘                            └────────────┬─────────────┘
                                                                  │ POST /wf/{id}/run/
                                                                  ▼
                                              ┌───────────────────────────────────┐
                                              │  后端 FastAPI  backend/pyspur/api   │
                                              │  workflow_run.py: run_workflow_*   │
                                              └────────────────┬──────────────────┘
                                                               │ new WorkflowExecutor(...)
                                                               ▼
                                     ┌────────────────────────────────────────────┐
                                     │  WorkflowExecutor  execution/               │
                                     │  按依赖拓扑调度 + asyncio 并发跑各节点        │
                                     └───────┬───────────────────────────┬────────┘
                                             │ 每个节点                    │ 每步写记录
                                             ▼                            ▼
                              ┌──────────────────────────┐   ┌──────────────────────┐
                              │  BaseNode 子类  nodes/    │   │  TaskRecorder         │
                              │  用 litellm/工具/向量库干活 │   │  execution/           │
                              └──────────────────────────┘   │  写 TaskModel 到数据库 │
                                                             └──────────────────────┘

主线一句话: 画布画出 DAG → 序列化成 nodes + links 的 JSON → FastAPI 接收 → WorkflowExecutor 拓扑 + 异步执行 → 每个 BaseNode 用 litellm 等干活 → 结果回显 + 存 task 记录。

4. 部件职责表​

每一行是一个部件、它干什么、它在哪个文件夹——按图找源码用这张表。

部件职责位置(相对克隆根)
前端画布React 画 DAG、编辑节点配置、回显每个节点的运行结果frontend/src(画布在 frontend/src/components/canvas,状态在 frontend/src/store)
API 层FastAPI 路由:创建/保存工作流、发起运行、恢复暂停、查运行记录backend/pyspur/api/*(运行入口 workflow_run.py,路由挂载 api/api_app.py:35-42)
数据模型用 Pydantic 描述并校验工作流结构(节点/连线/类型)backend/pyspur/schemas/workflow_schemas.py
执行器把 DAG 拓扑排序、异步并发执行、失败/暂停向下游传播backend/pyspur/execution/workflow_executor.py
节点体系BaseNode 抽象 + 各类节点(LLM/逻辑/循环/集成…)的实现backend/pyspur/nodes/(基类 nodes/base.py)
节点注册/工厂按名字创建节点实例;支持内置清单 + 装饰器注册表两条路backend/pyspur/nodes/factory.py、nodes/registry.py、nodes/node_types.py
task 记录每个节点执行时写一条 TaskModel(状态/输入/输出/耗时)backend/pyspur/execution/task_recorder.py

5. 一条端到端主线走一遍(高层,不进代码)​

跟着"点一次运行"发生了什么走一遍,先建立全局认知;每一步真正的代码在后续章节展开。入口是 run_workflow_blocking(backend/pyspur/api/workflow_run.py:261)。

  1. 取出图。 按 workflow_id 从库里取出这个工作流的当前版本,把它的 definition(那份 JSON)用 WorkflowDefinitionSchema.model_validate(...) 反序列化成对象(workflow_run.py:267-272)。

  2. 建一次运行。 先落一条 RunModel(状态 PENDING),再建一个 TaskRecorder 和 WorkflowExecutionContext(workflow_run.py:285-301)——前者负责把每个节点的执行写进库,后者携带 run_id、db、workflow 定义等上下文。

  3. 造执行器。 new WorkflowExecutor(workflow=..., task_recorder=..., context=...)(workflow_run.py:302-306)。构造时它就把图"消化"好:建节点字典、算出每个节点依赖谁(workflow_executor.py:68-69 调 _build_node_dict / _build_dependencies)。

  4. 喂输入、跑起来。 找到唯一的输入节点,把初始输入喂给它,await executor(...)(workflow_run.py:307-310)。__call__ 只是转调 run()(workflow_executor.py:919-932)。

  5. 拓扑 + 异步执行各节点。 run() 内部走 _execute_workflow;每个节点被包成一个 asyncio.Task(workflow_executor.py:160-172 的 _get_async_task_for_node_execution),在 _execute_node 里先 await 自己的依赖节点再执行自己(workflow_executor.py:461-463)——依赖天然形成拓扑序,互不依赖的分支并发跑。

  6. 每个节点干活。 节点实例的 __call__ 校验输入、调子类实现的 run()、再校验输出(nodes/base.py:175-246);LLM 类节点在 run() 里用 litellm 真正调模型(如 nodes/llm/single_llm_call.py)。

  7. 收尾。 全部完成则 RunModel 置 COMPLETED、把各节点输出按标题汇总回写(workflow_run.py:340-351);若中途有节点要求人工介入,会抛 PauseError、把运行置 PAUSED(workflow_run.py:352-380,详见 04-human-in-loop.md)。

一句话:取图 → 建 run/executor → 喂输入 → 拓扑异步跑节点 → 汇总输出或暂停。

6. 阅读地图(建议顺序,由浅入深)​

这套文档共 6 章。本章(index)是路由;下面 5 章各挖一个核心机制,建议按序号读,由浅入深、彼此不重叠。

章节讲什么什么时候读
01-workflow-model.md数据模型:一张 DAG 如何用 nodes + links 描述,WorkflowDefinitionSchema 有哪些校验(唯一输入节点、chatbot 字段要求等)想搞懂"图到底长什么样、存的是什么"先读它
02-executor.md执行引擎:依赖如何建、拓扑如何排、asyncio 如何并发、上游失败/暂停如何向下游传播读完 01 知道图长啥样后,读它看"图怎么被跑起来"
03-node-system.md节点体系:BaseNode 抽象、每个节点的输入/输出/配置为何是"动态 Pydantic 模型"、工厂 + 注册表如何按名字造节点想加自己的节点、或理解节点如何被创建时读
04-human-in-loop.md人在环中:PauseError 如何暂停工作流、run/task 的状态机、如何恢复并只重跑被阻塞的下游节点关心"等人审批""断点续跑"时读
05-llm-and-agent.mdLLM 与 Agent 节点:结构化输出(JSON Schema)、工具调用、Agent 的循环关心"节点里到底怎么调模型 / 让 Agent 自己循环"时读

最短路径: 只想懂大盘 → 读本章即可。想改执行行为 → 01 → 02。想扩节点 → 03(必要时回看 01)。想做审批/续跑 → 02 → 04。

7. 巧妙之处速览(细节在各章)​

先给读者一份"值得带走的精华"清单,每条只点到为止,深入见对应章节。

  • 图即程序,一份 JSON 打通前后端。 前端画布和后端执行器共用同一份 WorkflowDefinitionSchema(workflow_schemas.py:98-104)——UI 拖出来的东西、代码里搭出来的东西、库里存的东西,是同一个 schema,没有翻译层。(见 01)

  • 拓扑排序"免费"由 await 依赖得到。 执行器不显式做拓扑排序,而是让每个节点在 _execute_node 里 await 自己的依赖任务(workflow_executor.py:461-463),依赖关系天然决定先后,互不依赖的分支自动并发。(见 02)

  • 节点输入/输出是运行时动态生成的 Pydantic 模型。 节点收到的不是固定结构,而是"前驱节点名 → 前驱输出"的复合模型,__call__ 里按实际输入现造模型再校验(nodes/base.py:196-224)。(见 03)

  • 暂停用异常实现,续跑只重算被阻塞的下游。 人工介入靠抛 PauseError 打断执行,恢复时把已完成节点的输出作为 precomputed_outputs 传回、只跑 get_blocked_nodes 算出的下游子图(workflow_run.py:1219-1278)。(见 04)

  • 三种 Spur 复用同一引擎,只加校验与消息注入。 chatbot 的会话记忆是在同一个 run() 里"进来时注入历史、出去时存历史"(workflow_executor.py:877-916),而非另起一套执行路径。(见 02/05)

8. 代码地图(导航索引)​

一张表直接跳源码。行号 as-of sourceCommit;优先用符号名 grep 定位(抗行号漂移)。

主题文件路径符号名
包定位 / 依赖清单backend/pyproject.tomlname / description / dependencies
Spur 三类型枚举backend/pyspur/schemas/workflow_schemas.pySpurType
工作流数据模型(DAG)backend/pyspur/schemas/workflow_schemas.pyWorkflowDefinitionSchema / WorkflowNodeSchema / WorkflowLinkSchema
chatbot 输入/输出校验backend/pyspur/schemas/workflow_schemas.pyvalidate_chatbot_input_node / validate_chatbot_output_node
运行入口(阻塞)backend/pyspur/api/workflow_run.pyrun_workflow_blocking
运行入口(非阻塞 / 批量 / 恢复)backend/pyspur/api/workflow_run.pyrun_workflow_non_blocking / batch_run_workflow_non_blocking / process_pause_action
路由挂载backend/pyspur/api/api_app.pyinclude_router(workflow_run_router 挂在 /wf)
执行引擎backend/pyspur/execution/workflow_executor.pyWorkflowExecutor / run / _execute_node / _build_dependencies / get_blocked_nodes
单节点异步任务包装backend/pyspur/execution/workflow_executor.py_get_async_task_for_node_execution
task 记录backend/pyspur/execution/task_recorder.pyTaskRecorder / create_task / update_task
节点基类backend/pyspur/nodes/base.pyBaseNode / BaseNodeConfig / BaseNodeOutput / run
节点工厂 / 注册表backend/pyspur/nodes/factory.py · nodes/registry.pyNodeFactory.create_node / NodeRegistry.register
LLM 节点(litellm 干活)backend/pyspur/nodes/llm/single_llm_call.py · nodes/llm/agent.py—
前端画布frontend/src/components/canvas/EditorCanvas.tsxEditorCanvas