SAGE-Flownet

SAGE-Flownet 是一个轻量、原生 Python 的动态数据流执行引擎。它以“嵌套动态图”为核心模型，把用户写下的普通 Python 函数编译为可执行的 FlowTask DAG，并由运行时按需调度执行，让复杂的工作流像调用函数一样自然。

它不是另一个固定模板的流水线框架。Flownet 将远程调用、闭包、回调以及子图嵌套视为一等公民，在不牺牲 Python 语义的前提下，支持灵活组合和横向扩展。

安装与使用

pip install isage-flownet

⚠️ PyPI 包名 vs. import 名

SAGE 生态约定：PyPI 包名使用 isage- 前缀（因为 sage 在 PyPI 已被占用）， 但 Python import 名使用 sage.* 命名空间。

操作	命令
安装	pip install isage-flownet
import	from flownet import flow, init

import isage_flownet 不存在，请始终使用 flownet。

from flownet import flow, init, create_actor

@flow
def my_pipeline(init_stream):
    return init_stream.map(lambda x: x * 2)

init()  # 启动运行时

✧ 项目如何工作（概述）

• Flow 定义与编译：通过 @flow 装饰器将函数包装为 Flow 定义（FlowDef），编译为包含 Transformation 的 FlowTask 图。
• 数据流建模：DataStream.map/flatmap 等算子创建上下游关系，形成可追踪的依赖图和执行边。
• 运行时执行：FlownetRuntime 负责处理请求栈（TransformationFrame），并由 operator handler 决定目标调用与转发策略。
• Actor 调用统一化：ActorRef / ActorMethodRef 将本地与远程调用统一抽象，运行时根据目标地址完成派发。
• FlowRun 输出通道：运行时通过 FlowRunRef 的输出通道暴露“写入输入—等待输出”的调用方式，负责 fork/join 逻辑与结果收集。

✧ 预期目标与效果

• 保持原生 Python 开发体验：无需额外 DSL，工作流像写函数一样简单。
• 支持动态与嵌套的复杂流程：运行时编译、依赖跟踪与延迟执行适配不确定结构。
• 可扩展的分布式执行：为本地/远程统一执行提供清晰路径，便于水平扩展。
• 更可靠的并发与结果聚合：由运行时完成 fork/join、收集与异常处理，降低业务代码复杂度。

Intellectual Property

All Rights Reserved applies to repository contents authored by Jingyuan Tian, including code, documentation, benchmarks, and related assets.

The following copied lab projects are excluded from this claim and retain their original ownership/license context: applications/SAGE, applications/sage-studio, applications/sageData.

1. License terms: LICENSE
2. Ownership details: COPYRIGHT.md