FlowX

FlowX

用自然语言生成、运行和演进 AI Agent 工作流

English · 简体中文

⚙️ 安装配置 · 🚀 快速开始 · ✨ 核心特性 · 🧰 MCP 工具 · 🐍 Python SDK · 🤝 A2A · 🏗️ 架构 · 📖 文档 · English

FlowX 通过 MCP、Python SDK 或 A2A 让 AI Agent 构建、运行和演进工作流。描述流程后，FlowX 会生成工作流图、经过审计和测试的 Python 节点及 FastAPI 后端。

描述 → 生成 → 检查 → 运行 → 演进 → 复用

原始上游： AIpRoBuilder/FlowX · 署名说明 · 源码来源与验证。 FlowX 仍采用 Apache-2.0 许可证；欢迎符合许可证要求的 Fork 与复用。

🌟 为什么选择 FlowX？

将意图转换为可编辑、可复用的流程，而不是一次性回答。后续修改工作流时，无需重建未变化的节点。

示意演示，并非真实录屏。界面、卡片和耗时均为示意；FlowX 不捆绑该界面。参见分镜说明。

🚀 核心特性

• 🧩 生成工作流图、节点文档、Python 代码和后端。
• 🛡️ 审计和测试每个节点，并进行有界修复；失败的节点不会发布。
• 🔁 演进整个工作流或指定节点，复用未变化的节点。
• 🚀 本地运行工作流；MCP 还支持步骤输入和会话重置。
• 🔌 **选择接口：**本地 stdio MCP、Python SDK 或非流式 A2A。
• 📦 **复用产物：**工作流保存在 workspace/workflows/<workflow_id> 下。
• 🧠 **自选模型和 UI：**可选的 DeepSeek Harness 插件计划提供只读工作流查看器；创建、演进和运行工作流请使用独立的 FlowX 技能。

⚙️ 安装与配置

前置要求

• Python 3.10 或更高版本。
• Git，用于获取 FlowX 及其工作流运行时依赖。
• 用于本地 stdio 路径的支持 MCP 的 Agent 客户端。
• 兼容 OpenAI API 的 LLM 提供商（默认使用 DeepSeek）。

安装

克隆 FlowX 并安装核心、MCP 服务器、SDK 和 A2A 适配器：

pip

git clone https://github.com/AIpRoBuilder/FlowX.git
cd FlowX

# 创建并激活虚拟环境
python3.10 -m venv .venv
source .venv/bin/activate

# 安装 FlowX 及其运行时依赖
python -m pip install --upgrade pip
python -m pip install -e .

# Poetry
poetry env use python3.10
poetry install

# uv
uv sync

# conda
conda env create -f environment.yml
conda activate flowx

python3.10 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install -e .
python -m pip install -e ../ag_ui_worflow --no-deps

# Poetry
poetry env use python3.10
poetry install
poetry run pip install -e ../ag_ui_worflow --no-deps

# uv
uv sync
uv pip install --python .venv/bin/python -e ../ag_ui_worflow --no-deps

在源码检出目录中运行时，FlowX 会自动发现每个 packages/* 组件。仅当外部 ag_ui_workflow 源码尚未安装时，才需要设置 FLOWX_EXTRA_PATHS。

配置环境

cp .env.example .env

编辑 .env，设置提供商、模型、API 密钥和工作区：

FLOWX_LLM_PROVIDER=deepseek
FLOWX_LLM_MODEL=deepseek-flash
FLOWX_LLM_API_KEY=
FLOWX_DEFAULT_WORKSPACE=./.flowx_workspaces

安装后的 flowx-mcp 可以通过 FLOWX_CONFIG_ROOT 解析仓库检出目录之外的 .env 文件。FLOWX_EXTRA_PATHS 可以指向本地 ag_ui_workflow 源码，用于尚未安装到当前环境的情况。

SDK 和 A2A 服务器读取进程环境变量，不会自动加载此 .env 文件。请导出配置，或在创建客户端前由应用加载该文件。例如，在 macOS/Linux 的仓库根目录运行：

set -a; source .env; set +a

启动 MCP 服务器

# 默认本地 stdio 入口
flowx-mcp

# 详细日志
flowx-mcp --verbose

# 源码检出入口
python3.10 run_server.py

环境未激活时，请加上 poetry run 或 uv run 前缀。MCP 宿主通常会自行启动服务器；上述命令可用于检查入口。

在 MCP 客户端中注册 FlowX

将可执行文件替换为虚拟环境中 flowx-mcp 的绝对路径，并设置凭据和工作区。MCP 宿主可能不会继承 shell 的 PATH。

{
  "mcpServers": {
    "flowx": {
      "command": "/absolute/path/to/your/venv/bin/flowx-mcp",
      "env": {
        "FLOWX_LLM_PROVIDER": "deepseek",
        "FLOWX_LLM_MODEL": "deepseek-flash",
        "FLOWX_LLM_API_KEY": "<your key>",
        "FLOWX_DEFAULT_WORKSPACE": "/absolute/path/to/flowx_workspaces"
      }
    }
  }
}

请按 MCP 客户端的安全指南保存凭据，避免将其提交到仓库。

安装 FlowX Agent Skill

TARGET_PROJECT=/absolute/path/to/your/project
mkdir -p "$TARGET_PROJECT/.github/skills/flowx-workflows"
cp -R skills/flowx-workflows/. "$TARGET_PROJECT/.github/skills/flowx-workflows/"

mkdir -p ~/.copilot/skills/flowx-workflows
cp -R skills/flowx-workflows/. ~/.copilot/skills/flowx-workflows/

hermes skills install AIpRoBuilder/FlowX/skills/flowx-workflows
hermes skills list

python3 -c 'import sys; assert sys.version_info >= (3, 10), "Python 3.10+ required"' && \
  test ! -e "$HOME/.flowx/venv" && \
  test ! -L "$HOME/.flowx/venv" && \
  python3 -m venv "$HOME/.flowx/venv" && \
  "$HOME/.flowx/venv/bin/python" -m pip install 'git+https://github.com/AIpRoBuilder/FlowX.git' && \
  hermes skills install AIpRoBuilder/FlowX/skills/flowx-workflows

mkdir -p ~/.hermes/skills/flowx-workflows
cp -R skills/flowx-workflows/. ~/.hermes/skills/flowx-workflows/

⚡ 快速开始

完成安装与配置后，向支持 MCP 的 Agent 发送请求：

1. 描述工作流

Use flowx to build a workflow.

Build a workflow that accepts a support ticket, classifies its category,
assigns a priority, and drafts a reply. Name the reply node draft_reply.
Return the category, priority, and draft reply as structured output.

2. 构建并检查

1. 调用 create_workflow，然后调用 start_new_workflow(workflow_id, prompt)。
2. 轮询 get_build_status(workflow_id)，检查 status 和 result.ok；顶层 ok 只表示状态查询成功。
3. 若状态为 input_required，展示 awaiting.prompt 和 awaiting.test_log，然后调用 submit_background_build_input，传入准确的 workflow_id、revision、request_id、作为 step_id 的 awaiting.stepId 和用户 input，再继续轮询。
4. 使用 get_workflow 和 list_nodes 检查结果。步骤 ID 以生成的工作流为准。

MCP 构建在后台运行；Python SDK 仍提供同步方法。旧版同步 MCP 构建工具已移除，升级前请使用旧服务器完成或取消已暂停的旧构建。生成过程会将需求及可能的日志发送给 LLM 提供商，并可能执行生成的测试；运行服务前请检查生成的代码。

3. 运行、演进与复用

Run the workflow as a service.

Update the draft_reply node: for urgent tickets, include an escalation note.
Then rerun the service.

run_service 启动或重启生成的服务。run_step 向运行中的服务输入某个步骤的参数，并在工作流暂停等待更多输入时返回状态；reset_session 取消暂停的运行。start_evolve_workflow 可以修改整个工作流（target="workflow"）、某个节点（list_nodes 返回的 step id 或 node id），或由引擎自行判断；按上述方式轮询后台构建。

🧰 MCP 工具列表

这些 MCP 工具让宿主 Agent 在同一段对话中操作工作流层。UI 渲染刻意留给宿主负责。

🐍 Python SDK

无需 MCP 传输，直接在 Python Agent 或你自己的智能 UI 后端中嵌入工作流构建能力。完整示例见 examples/sdk_quickstart.py。

from flowx_sdk import FlowXClient

with FlowXClient(workspace="./workflows") as flowx:
    created = flowx.build_workflow_json(
        "Build a workflow that categorizes and prioritizes support tickets.",
    ).raise_for_failure()                                  # workflow.json + requirements
    workflow_id = created.workflow_id
    flowx.build_nodes(workflow_id, await_repair_input=False).raise_for_failure()  # all nodes, or {step_id: amendment}
    flowx.build_main_entrypoint(workflow_id).raise_for_failure()
    print(flowx.run_service(workflow_id))                  # run as a service
    input("Press Enter to stop the service... ")

每次构建调用都会返回包含各阶段结果、错误、日志和产物路径的 AgentResult；raise_for_failure() 会在阶段失败时抛出异常。使用服务时请保持客户端打开；退出 with 代码块会停止该客户端启动的服务。

build_nodes、build_workflow 和 evolve 默认在节点测试失败时暂停。若 status="input_required"，检查 awaiting，然后用 flowx.answer(request_id, input, awaiting["stepId"]) 继续。暂停时 ok 也可能为 True，不能仅凭 raise_for_failure() 判断构建完成。传入 await_repair_input=False 可自动根据测试日志修复。指导和输入选项参见 examples/sdk_guided_repair.py。

在终端回答节点构建暂停

安装 SDK、导出 LLM 配置并构建工作流图后，在仓库根目录运行技能的构建脚本， 将工作区和工作流 ID 替换为实际值：

python skills/flowx-workflows/scripts/build.py --workspace "$WS" nodes --workflow-id "$WF"

如果脚本输出 input_required JSON 事件，查看 awaiting.prompt 和 awaiting.test_log。保持同一个进程运行，在其标准输入中输入一行 JSON 并回车：

{"action":"answer","input":"调用 API 前先处理空输入。"}

每次暂停只发送一个回答；后续节点再次暂停时重复此操作。输入 test: <指导> 可修改生成的测试，skip 可跳过本次修改，空字符串根据测试日志修复。 发送 {"action":"abort"} 可取消构建；关闭标准输入也会取消。可选的 step_id 必须与 awaiting.stepId 一致。--auto-repair 会关闭这些测试修复暂停。请检查最终的 result.status、result.ok、阶段、错误和日志，不要仅凭暂停或回答判断成功。

节点审计违规会根据审计报告自动修复，不会单独暂停等待用户输入。详情参见 技能说明和动态工作流文档。

要修改现有工作流，可使用 evolve(workflow_id, command, target=...) 选择演进目标。list_nodes 会返回每个节点的 step id 与 node id；不传 target 时由引擎判断命令指向哪一部分。可运行示例见 examples/sdk_evolve.py。

如需对扩展后的 DynamicWorkflowOrchestrator 引擎进行不依赖 LLM 的直接冒烟测试，请参见 examples/dynamic_workflow_orchestrator.py。它覆盖逐节点生成/修复链、finalizer ID 以及增量业务图更新。

from flowx_sdk import FlowXClient, WORKFLOW_TARGET

workflow_id = "<existing workflow ID>"
with FlowXClient(workspace="./workflows") as flowx:
  flowx.evolve(workflow_id, "Add a language-detection step.", target=WORKFLOW_TARGET, await_repair_input=False).raise_for_failure()
  nodes = flowx.list_nodes(workflow_id)
  flowx.evolve(workflow_id, "Validate empty input.", target=nodes[0]["step_id"], await_repair_input=False).raise_for_failure()

🤝 A2A 服务器

通过非流式 Agent2Agent 端点，让另一个 Agent 将工作流生成委托给 FlowX：

flowx-a2a --host 127.0.0.1 --port 8001

它在 /.well-known/agent-card.json 提供 A2A Agent Card，在 /a2a（以及 /）提供 JSON-RPC。支持的方法有 message/send、tasks/get 和 tasks/cancel。发送 A2A 文本消息即可构建工作流；添加 metadata.workflowId 可修改现有工作流。所有阶段成功时任务状态为 completed，否则为 failed；两种情况下都会携带结果，包括生成的产物路径。

🏗️ 架构

生成式工作流层包含一个共享核心和三种面向 Agent 的接口：flowx_core 负责规划、生成、审计和运行工作流；flowx_mcp、flowx_sdk 与 flowx_a2a 将这些能力提供给不同的 Agent 宿主。生成后的运行时位于该层之下，智能 UI 位于该层之上。

flowchart TB
  UI["智能 UI / Agent 宿主"] --> MCP["本地 MCP：构建、修改、运行"]
  UI --> SDK["Python SDK：构建、修改、运行"]
  PEER["对等 Agent"] --> A2A["A2A：委托生成"]
  MCP --> CORE["FlowX · 生成式工作流层"]
  SDK --> CORE
  A2A --> CORE
  CORE --> FILES["工作流图 + 节点代码 + 后端"]
  FILES --> RUNTIME["FastAPI + ag_ui_workflow 运行时"]
  MCP -->|"作为服务运行 / 重启"| RUNTIME
  RUNTIME -.->|"AG-UI SSE 到自定义前端"| UI
  MCP -->|"后续变更"| CORE

仓库结构

产物流转

工作流编译是生成式工作流层的核心机制。规划与代码生成将意图转化为产物；审计在生成过程中和运行时提供反馈。下面的产物流程图与 assets/architecture.mmd 保持一致。

flowchart LR
    A[原始需求] --> B[requirement_analysis.md]
    B --> C[workflow.json]
    C --> D[node_docs/*.md]
    D --> E[nodes/*.py]
    E --> F[main.py]

    C --> G[图审计]
    E --> H[节点审计]
    F --> I[入口审计]
    F --> J[运行日志审计]

    G -->|反馈| C
    H -->|反馈| E
    I -->|反馈| F
    J -->|反馈| E

🧪 运行时兼容性

FlowX 面向请求作用域的 ag_ui_workflow API，已针对上游修订 0319046（2026-10-23）验证，包括 StepMeta 步骤元数据契约。贡献者可参阅 docs/RUNTIME.md 了解契约细节。

🙏 致谢

FlowX 感谢 CGraph 为工作流执行提供运行时 DAG 引擎框架。

🤝 参与贡献

欢迎提交 Issue 和 Pull Request。如果 FlowX 对你有帮助，欢迎给仓库点 Star、分享你生成的工作流，或提交一个你希望转化为工作流的意图。

当你重新分发或修改 FlowX 时，请保留 LICENSE 与 NOTICE 署名（参见 docs/PROVENANCE.md）。

💬 联系方式

有兴趣为 AI Agent 构建生成式工作流层，或将其连接到智能 UI？欢迎通过 peterxcx@gmail.com 联系。

微信二维码：

📜 许可证与来源

FlowX 采用 Apache License 2.0 许可证。在遵守其条款的前提下允许重新分发与修改；重新分发 FlowX 源码时，请保留 NOTICE 署名与源码来源记录。

原始上游： AIpRoBuilder/FlowX。

⭐ Star 历史

中文版为英文 README 的翻译；如有差异，以英文版为准。

↑ 返回顶部 ↑