mcpField
F
flowing

flowing

github.com/oaustegardAgent 工作流15743 次安装1 天前更新
代码仓库

关于 flowing

flowing 是一种 Python 技能，可将多步骤流程表示为任务图，而非文字指令。它适用于需要分支、验证、重试、并行任务或可恢复执行的上下文内流水线。

主要功能

  1. 01用任务图定义依赖关系
  2. 02通过条件和验证控制任务
  3. 03按谓词或异常重试任务
  4. 04并行运行相互独立的任务
  5. 05使用缓存结果从故障处恢复
  6. 06在日志中持久化检查点

适用场景

  1. 01构建包含条件步骤的流水线
  2. 02重试 LLM 生成，直到通过检查
  3. 03任务失败后继续运行流水线
  4. 04执行副作用而不阻塞主任务图

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/oaustegard/claude-skills --skill flowing

安装后在 Claude Code 中输入/flowing确认已加载。

包含的文件

除 SKILL.md 外，安装时一起下载的参考资料和脚本。

CHANGELOG.md
README.md
references/
reference.md
scripts/
flowing.py
tests/
test_flowing.py

详细介绍

不会被动态工作流取代——请先阅读

Claude Code 的动态工作流负责编排子 Agent（使用独立上下文，支持扇出至 16 个并发任务／1000 个 Agent）。此技能属于另一种原语：在单一上下文中控制你自己的工具调用，具有持久化副作用，并支持从检查点恢复。工作流运行时明确不能直接访问文件系统或 shell——由 Agent 执行工作，脚本只负责协调。Flowing 则正好相反：由脚本执行工作。

在上下文内的流水线（3 个以上步骤、分支、重试、验证、分离的副作用）中使用 flowing。需要大量子 Agent 时使用工作流。两者可以组合使用，并非相互竞争。不要为了工作流而放弃 flowing——否则会失去持久化副作用，以及 hub-spoke 所依赖的跨会话检查点。

Flowing — 用代码而非文字控制流程

当一个过程需要 3 个以上步骤、分支、重试或契约时，应将其编码为 Python 任务的 DAG，而不是用文字描述指令。像“先执行 X，再执行 Y，然后如果 Z 就重试 3 次”这样的文字，可能会被忽略并直接生成后续内容。@task 图具有结构性：在输入绑定之前，步骤无法运行；针对错误输入触发的门控也无法跳过。

运行器负责控制流程——分支、重试、验证、传播故障、并行处理。你只需在各个叶节点提供判断。运行器：scripts/flowing.py。

快速入门

from flowing import task, Flow

@task
def fetch_data():
    return {"items": [1, 2, 3]}

@task(depends_on=[fetch_data])
def process(fetch_data):          # param name must match the dep's name
    return sum(fetch_data["items"])

@task(depends_on=[process])
def store(process):
    print(f"Result: {process}")

Flow(store).run()                 # topo-sorts, runs each layer, parallel within a layer

每个任务都会以依赖项名称作为 kwargs 参数接收其依赖项。同一层中的独立任务会并行运行。

控制流程原语

将分支和契约编码为图结构，而不是写在任务主体中的 if 语句。

when= — 条件门控

仅当谓词（针对收集到的依赖项值）为真值时才运行任务。假值 → SKIPPED，且跳过状态会传播给依赖任务。

@task(depends_on=[fetch], when=lambda fetch: fetch["needs_processing"])
def process(fetch):
    return transform(fetch["payload"])
validate= — 边契约

在主体运行前检查收集到的依赖项值。引发异常 → FAILED，不重试（错误输入不会自行修复）。通过 → 继续。

def must_have_items(fetch):
    if not fetch.get("items"):
        raise ValueError("fetch returned empty payload")

@task(depends_on=[fetch], validate=must_have_items)
def process(fetch):
    return sum(fetch["items"])
retry_until= — 由谓词驱动的循环

运行主体，然后调用 retry_until(value)。为 True → 完成。为 False → 重试，并消耗 retry= 次数预算。适用于能自我纠正的大模型步骤：生成、检查、重新生成。

@task(retry=4, retry_until=lambda r: r["valid"])
def generate_until_valid():
    candidate = llm_call(...)
    return {"valid": passes_schema(candidate), "candidate": candidate}

这与单独使用 retry= 不同，后者只会在引发异常时重试。

其他能力

  • 并行执行 — 同一层中的独立任务在线程池中运行（max_workers=）。
  • detached=True — 副作用任务（内存写入、通知）会在主 DAG 完成后运行，失败时不会阻塞主 DAG。
  • 进程内恢复 — flow.run() → 修复 → flow.resume() 会从失败点重新运行，并在内存中缓存已成功的任务（仅限同一进程）。flow.override(task, value) 会注入修正后的结果。
  • 持久化日志（journal_path=） — 可选的内容寻址重放机制，可在容器终止后继续使用。Flow(term, journal_path="/path/run.jsonl").run() 会将每个成功任务的结果追加到只追加的 JSONL 文件中，文件以 step_key 为键；该键是对任务字节码、其 when/validate/retry_until 函数体及其依赖项的键（采用链式计算，因此上游更改会向下游传播）进行 SHA-256 计算得到的。之后调用 run() 时，即使是在全新的容器中，也会从日志中重放未更改的前缀，只执行键不存在的任务；编辑任务函数体会使其键失效，并重新运行该任务及其依赖项，而外观性选项（retry=、timeout_s=、name）不会。跨会话检查点的中心枢纽辐射式工作依赖此机制。注意：结果会被 pickle 序列化，因此不可 pickle 的返回值只会重新运行；闭包捕获的值不属于键的一部分（只有任务函数体自身的代码包含在内）。
  • timeout_s=、带指数退避的 retry=、fail_fast=。

在使用快速入门和上述三个基本功能之外的任何内容之前，请阅读 references/reference.md——其中介绍了所有 @task 参数、Flow 方法、恢复/覆盖、分离任务的自动发现，以及 validate=/when= 的签名匹配注意事项。

适用场景

  • 流程包含重要分支 → when= 可将其设为结构化分支。
  • 步骤有输入契约 → validate= 可让契约得到强制执行。
  • 大模型步骤需要收敛 → retry_until= 可将检查纳入循环。
  • 有 3 个或更多可并行执行的独立操作。
  • 多步骤流水线中，后期失败不应浪费前期工作。
  • 副作用不应阻塞关键路径 → detached=True。

不适用场景

  • 单个顺序操作——直接调用函数即可。
  • 下一步需要对上一步的结果进行无法用谓词表达的推理——使用思考循环。
  • 异步或分布式工作流——此方案基于单容器和线程池。

编写规范

如果你发现自己在写类似 “先调用 X，验证 Y，然后如果 Z 成立就重试最多 3 次” 的说明文字，那就是一个流程图。在发布之前重构。文字指令无法强制执行；@task 图可以。

评论

支持 Markdown