mirror of
https://github.com/hanzoai/flow.git
synced 2026-10-17 01:29:18 +00:00
feat(serverless): run a Flow graph as a Hanzo Function on Fission
compiler.py + runner.py + FUNCTIONS.md — take a Flow (graph) and execute it as a Hanzo Function against the live Fission control plane. Rebased onto upstream rather than merged: my local tree was 21 commits behind and also carried a fix repointing four alembic migrations from `langflow.utils` to `flow.utils`. That fix is obsolete — upstream deleted those four migration files outright AND completed the rename (src/backend/base/langflow no longer exists, 630 files now under base/flow). Keeping my version would have resurrected four deleted migrations. Only the genuinely new work is carried forward. Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
Hanzo Dev
+107
@@ -0,0 +1,107 @@
|
||||
# Flow → Hanzo Functions (Fission)
|
||||
|
||||
Run a Flow (langflow-fork graph) as a Hanzo Function on the live Fission control
|
||||
plane (`fission` ns, `do-sfo3-hanzo-k8s`). Two compile models; GPU envs host the
|
||||
Zen-model nodes. Scaffold: `serverless/` (compiler + in-function runner).
|
||||
|
||||
## How a Flow runs today
|
||||
|
||||
A flow export is a node/edge JSON. Production builds it fresh per request:
|
||||
|
||||
```python
|
||||
graph = Graph.from_payload(flow.data, flow_id, flow_name) # flow/graph
|
||||
outputs = await run_flow(inputs=..., graph=graph, user_id=...) # flow/helpers/flow.py:200
|
||||
```
|
||||
|
||||
Functions just needs to do the same inside a function pod.
|
||||
|
||||
## Model A — flow-as-function (default, demo path)
|
||||
|
||||
The **whole flow** becomes ONE Fission function on a Python env that has
|
||||
`hanzo-flow`/`lfx` installed.
|
||||
|
||||
```
|
||||
flow.json ──compile──> Fission Function (env=flow-python | flow-gpu)
|
||||
package = { runner.py, flow.json }
|
||||
invoke: POST /v1/functions/{flow} {input_value, tweaks}
|
||||
→ runner: Graph.from_payload(flow.json) → run_flow(...) → outputs
|
||||
```
|
||||
|
||||
- **Env**: a custom `flow-python` environment = upstream `python-env` + the
|
||||
`hanzo-flow` wheel baked in (build via CI → `ghcr.io/hanzoai/flow-env`). The
|
||||
function package is just `runner.py` + the embedded `flow.json`.
|
||||
- **GPU / Zen models**: if any node is a model node (LLM/embeddings/Zen), compile
|
||||
to `flow-gpu` (the `flow-python` image on a GPU env: `fission env create
|
||||
--name flow-gpu --image ghcr.io/hanzoai/flow-env-gpu` with GPU resources).
|
||||
Inference runs in-pod or, better, the model node calls the LLM gateway
|
||||
(`api.hanzo.ai/v1`) so the flow function stays CPU and the gateway owns GPU —
|
||||
pick per node (see "GPU placement" below).
|
||||
- **Billing**: per invocation via the router's `usagemeter` (Provider
|
||||
`functions`); GPU-seconds when on `flow-gpu`. Same contract as
|
||||
`functions/INTEGRATION.md`.
|
||||
- **Pros**: one cold start, simple, exact-parity with current execution.
|
||||
**Cons**: the whole flow scales as a unit.
|
||||
|
||||
This is the demo path: real, minimal, and reuses the existing `run_flow`.
|
||||
|
||||
## Model B — node-as-function (opt-in, for scale / isolation / GPU)
|
||||
|
||||
Each vertex compiles to its own Fission function; edges become calls. A
|
||||
coordinator function walks the DAG (topological order, fan-out parallel
|
||||
branches) invoking node functions through the router.
|
||||
|
||||
```
|
||||
flow.json ──compile──> { coordinator-fn, node-fn × N }
|
||||
coordinator: topo-sort vertices → for each, POST /v1/functions/{flow}-{node}
|
||||
passing upstream outputs as input; aggregate; return sink output
|
||||
```
|
||||
|
||||
- **GPU placement**: only model/Zen nodes compile to `flow-gpu`; CPU nodes to
|
||||
`flow-python`. This is the real win — isolate + scale the expensive node, keep
|
||||
the rest cheap and scale-to-zero.
|
||||
- **Async DAGs**: long branches use Fission message-queue triggers (KEDA/NATS)
|
||||
instead of sync calls.
|
||||
- **Pros**: per-node scaling, GPU isolation, reuse of shared nodes across flows.
|
||||
**Cons**: N cold starts, orchestration + payload-passing complexity.
|
||||
|
||||
Use B for heavy/parallel/GPU-mixed flows; A everywhere else. One rule:
|
||||
**a node is its own function only when it needs isolation (GPU) or independent
|
||||
scale** — otherwise it stays in the flow-as-function (don't fragment for its own
|
||||
sake).
|
||||
|
||||
## GPU placement decision (both models)
|
||||
|
||||
| Node | Where it runs | Why |
|
||||
|------|---------------|-----|
|
||||
| LLM / Zen model node | call `api.hanzo.ai/v1` (gateway owns GPU + Zen models) | default — no GPU in the flow pod, gateway already pools/scales models |
|
||||
| Custom model / local weights | `flow-gpu` env (GPU Fission env) | when the node ships its own weights and can't go through the gateway |
|
||||
| Tools / IO / logic nodes | `flow-python` (CPU, scale-to-zero) | cheap, bursty |
|
||||
|
||||
So most flows are CPU functions that call the gateway for inference; only
|
||||
bring-your-own-model nodes need a GPU env.
|
||||
|
||||
## Scaffold (`serverless/`)
|
||||
|
||||
- `compiler.py` — `compile_flow(flow_json, runtime, gpu)` → `FunctionSpec`
|
||||
(env, package files, entrypoint, fission CLI commands). Model A implemented;
|
||||
Model B (`compile_flow_to_dag`) sketched.
|
||||
- `runner.py` — the in-function entry: load `flow.json`, `Graph.from_payload`,
|
||||
`run_flow`, return JSON. This is the function code Model A deploys.
|
||||
- `__init__.py` — exports.
|
||||
|
||||
Deploy (Model A), once `flow-env` is published by CI:
|
||||
|
||||
```sh
|
||||
fission env create --name flow-python --image ghcr.io/hanzoai/flow-env -n fission
|
||||
python -m serverless.compiler my_flow.json --runtime flow-python # emits pkg + commands
|
||||
fission fn create --name my-flow --env flow-python --src ./build/my-flow --entrypoint runner.handler -n fission
|
||||
curl https://api.hanzo.ai/v1/functions/my-flow -d '{"input_value":"hi"}'
|
||||
```
|
||||
|
||||
## Remaining seams
|
||||
|
||||
- Publish `ghcr.io/hanzoai/flow-env` (+ `-gpu`) via CI: `python-env` + the
|
||||
`hanzo-flow` wheel. (No local image builds — CI.)
|
||||
- Coordinator runtime for Model B (topo-walk + MQ-trigger async branches).
|
||||
- Wire `compile_flow` into the Flow UI ("Deploy as Function" → calls the
|
||||
`/v1/functions` management API in `functions/INTEGRATION.md` §3).
|
||||
@@ -0,0 +1,9 @@
|
||||
"""Flow -> Hanzo Functions (Fission) compiler.
|
||||
|
||||
Compile a Flow graph export into a deployable Fission function. See FUNCTIONS.md
|
||||
for the two models (flow-as-function, node-as-function) and GPU placement.
|
||||
"""
|
||||
|
||||
from .compiler import FunctionSpec, compile_flow
|
||||
|
||||
__all__ = ["FunctionSpec", "compile_flow"]
|
||||
@@ -0,0 +1,158 @@
|
||||
"""Compile a Flow export (node/edge JSON) into a deployable Fission function.
|
||||
|
||||
Model A (implemented): the whole flow becomes ONE function on a Python env that
|
||||
has hanzo-flow installed. The package is `runner.py` + the embedded `flow.json`;
|
||||
the function calls `flow.helpers.flow.run_flow` per request.
|
||||
|
||||
Model B (sketched, `compile_flow_to_dag`): each model/heavy node becomes its own
|
||||
function and a coordinator walks the DAG — use only for GPU isolation / per-node
|
||||
scale (see FUNCTIONS.md).
|
||||
|
||||
This module emits a `FunctionSpec` + a build dir + the `fission` CLI commands. It
|
||||
does NOT call k8s — the `/v1/functions` management API (functions/INTEGRATION.md
|
||||
§3) or the CLI deploys it. No image builds here (envs are published by CI).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
# Node types that imply model inference. A flow with any of these is a candidate
|
||||
# for a GPU env (Model A) or for splitting that node out (Model B). Kept as a
|
||||
# simple prefix/keyword set — the flow component registry is the source of truth;
|
||||
# this is the heuristic the compiler uses without importing the whole registry.
|
||||
_MODEL_NODE_KEYWORDS = ("llm", "model", "embedding", "openai", "anthropic", "zen", "huggingface")
|
||||
|
||||
# Default runtime env names (registered on the Fission control plane).
|
||||
ENV_CPU = "flow-python" # ghcr.io/hanzoai/flow-env (python-env + hanzo-flow)
|
||||
ENV_GPU = "flow-gpu" # ghcr.io/hanzoai/flow-env-gpu (same, GPU resources)
|
||||
|
||||
|
||||
@dataclass
|
||||
class FunctionSpec:
|
||||
"""A compiled, deployable Fission function for one flow."""
|
||||
|
||||
name: str
|
||||
env: str
|
||||
build_dir: Path
|
||||
entrypoint: str = "runner.handler"
|
||||
network: str = "mainnet" # mainnet -> ns fission ; testnet/devnet -> ns fission-sandbox
|
||||
gpu: bool = False
|
||||
files: list[str] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def namespace(self) -> str:
|
||||
return "fission" if self.network == "mainnet" else "fission-sandbox"
|
||||
|
||||
def cli_commands(self) -> list[str]:
|
||||
"""The `fission` CLI commands to deploy this function."""
|
||||
ns = self.namespace
|
||||
return [
|
||||
f"fission fn create --name {self.name} --env {self.env} "
|
||||
f"--src {self.build_dir}/'*' --entrypoint {self.entrypoint} -n {ns}",
|
||||
f"fission httptrigger create --name {self.name} --method POST "
|
||||
f"--url /{self.name} --function {self.name} -n {ns}",
|
||||
]
|
||||
|
||||
|
||||
def _flow_has_model_node(flow_json: dict) -> bool:
|
||||
nodes = flow_json.get("data", flow_json).get("nodes", [])
|
||||
for n in nodes:
|
||||
nid = json.dumps(n.get("data", {}).get("type", n.get("id", ""))).lower()
|
||||
if any(k in nid for k in _MODEL_NODE_KEYWORDS):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def compile_flow(
|
||||
flow_json: dict,
|
||||
name: str,
|
||||
*,
|
||||
runtime: str | None = None,
|
||||
network: str = "mainnet",
|
||||
gpu: bool | None = None,
|
||||
out_dir: str | Path = "build",
|
||||
) -> FunctionSpec:
|
||||
"""Compile `flow_json` (a Flow export) into a Model-A Fission function.
|
||||
|
||||
`gpu` defaults to auto: True if the flow contains a model node and you want
|
||||
in-pod inference. The common case keeps `gpu=False` and lets model nodes call
|
||||
the LLM gateway (api.hanzo.ai/v1) — see FUNCTIONS.md "GPU placement".
|
||||
"""
|
||||
if gpu is None:
|
||||
gpu = False # default: model nodes call the gateway, flow stays CPU.
|
||||
env = runtime or (ENV_GPU if gpu else ENV_CPU)
|
||||
|
||||
build_dir = Path(out_dir) / name
|
||||
if build_dir.exists():
|
||||
shutil.rmtree(build_dir)
|
||||
build_dir.mkdir(parents=True)
|
||||
|
||||
# Package = the runner + the embedded flow graph.
|
||||
(build_dir / "flow.json").write_text(json.dumps(flow_json))
|
||||
shutil.copy(Path(__file__).with_name("runner.py"), build_dir / "runner.py")
|
||||
|
||||
return FunctionSpec(
|
||||
name=name,
|
||||
env=env,
|
||||
build_dir=build_dir,
|
||||
network=network,
|
||||
gpu=gpu,
|
||||
files=["runner.py", "flow.json"],
|
||||
)
|
||||
|
||||
|
||||
def compile_flow_to_dag(flow_json: dict, name: str, *, network: str = "mainnet") -> list[FunctionSpec]:
|
||||
"""Model B sketch: one function per model/heavy node + a coordinator.
|
||||
|
||||
Splits out nodes that warrant isolation (GPU / independent scale); everything
|
||||
else stays in a single CPU function. The coordinator topo-walks the graph and
|
||||
invokes node functions through the router. Full impl is a follow-up — this
|
||||
returns the intended unit boundaries so the UI/caller can preview the split.
|
||||
"""
|
||||
nodes = flow_json.get("data", flow_json).get("nodes", [])
|
||||
specs: list[FunctionSpec] = []
|
||||
for n in nodes:
|
||||
ntype = str(n.get("data", {}).get("type", n.get("id", ""))).lower()
|
||||
if any(k in ntype for k in _MODEL_NODE_KEYWORDS):
|
||||
# one isolated (GPU-capable) function per model node
|
||||
specs.append(
|
||||
FunctionSpec(
|
||||
name=f"{name}-{n.get('id', ntype)}",
|
||||
env=ENV_GPU,
|
||||
build_dir=Path("build") / f"{name}-{n.get('id', ntype)}",
|
||||
network=network,
|
||||
gpu=True,
|
||||
)
|
||||
)
|
||||
# plus the coordinator (CPU) — the remaining graph + the call DAG.
|
||||
specs.append(
|
||||
FunctionSpec(name=f"{name}-coordinator", env=ENV_CPU, build_dir=Path("build") / f"{name}-coordinator", network=network)
|
||||
)
|
||||
return specs
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover - dev convenience
|
||||
import argparse
|
||||
|
||||
ap = argparse.ArgumentParser(description="Compile a Flow export to a Fission function.")
|
||||
ap.add_argument("flow_json", help="path to the flow export JSON")
|
||||
ap.add_argument("--name", help="function name (default: flow file stem)")
|
||||
ap.add_argument("--runtime", default=None, help=f"env name (default: {ENV_CPU} / {ENV_GPU})")
|
||||
ap.add_argument("--network", default="mainnet", choices=["mainnet", "testnet", "devnet"])
|
||||
ap.add_argument("--gpu", action="store_true", help="use the GPU env (in-pod inference)")
|
||||
args = ap.parse_args()
|
||||
|
||||
data = json.loads(Path(args.flow_json).read_text())
|
||||
spec = compile_flow(
|
||||
data,
|
||||
name=args.name or Path(args.flow_json).stem,
|
||||
runtime=args.runtime,
|
||||
network=args.network,
|
||||
gpu=args.gpu,
|
||||
)
|
||||
print(f"compiled '{spec.name}' -> env={spec.env} ns={spec.namespace} dir={spec.build_dir}")
|
||||
print("\n".join(spec.cli_commands()))
|
||||
@@ -0,0 +1,58 @@
|
||||
"""In-function entry for a flow deployed as a Hanzo Function (Model A).
|
||||
|
||||
The Fission Python env imports this module and calls `handler(context)` per
|
||||
request. The package ships this file plus `flow.json` (the compiled flow export).
|
||||
On each request we build the graph fresh and run it — identical to how the Flow
|
||||
server executes a flow (flow.helpers.flow.run_flow / Graph.from_payload), so
|
||||
behaviour matches the editor.
|
||||
|
||||
Identity is the gateway-minted X-User-Id / X-Org-Id (see functions/INTEGRATION.md);
|
||||
billing is handled by the router's usagemeter — nothing to do here.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
_FLOW_PATH = Path(__file__).with_name("flow.json")
|
||||
|
||||
|
||||
async def _run(input_value: str, tweaks: dict | None, user_id: str):
|
||||
# Imported lazily so the module loads even outside the flow env (e.g. tests).
|
||||
from flow.graph.graph.base import Graph
|
||||
from flow.helpers.flow import run_flow
|
||||
|
||||
flow_json = json.loads(_FLOW_PATH.read_text())
|
||||
payload = flow_json.get("data", flow_json)
|
||||
graph = Graph.from_payload(payload, flow_id=flow_json.get("id"), flow_name=flow_json.get("name"))
|
||||
|
||||
outputs = await run_flow(
|
||||
inputs={"input_value": input_value, "type": "chat"},
|
||||
tweaks=tweaks,
|
||||
graph=graph,
|
||||
user_id=user_id,
|
||||
output_type="chat",
|
||||
)
|
||||
# RunOutputs -> JSON-able
|
||||
return [o.model_dump() if hasattr(o, "model_dump") else str(o) for o in outputs]
|
||||
|
||||
|
||||
def handler(context):
|
||||
"""Fission Python env entrypoint.
|
||||
|
||||
Body: {"input_value": str, "tweaks": dict?}. Identity from gateway headers.
|
||||
"""
|
||||
import asyncio
|
||||
|
||||
body = context.request.get_json(silent=True) or {}
|
||||
input_value = body.get("input_value", "")
|
||||
tweaks = body.get("tweaks")
|
||||
# Gateway-minted identity (trust boundary). Fall back to a service id offline.
|
||||
user_id = context.request.headers.get("X-User-Id", "anonymous")
|
||||
|
||||
try:
|
||||
result = asyncio.run(_run(input_value, tweaks, user_id))
|
||||
return {"status": 200, "body": json.dumps({"outputs": result})}
|
||||
except Exception as exc: # surface a clean error to the caller
|
||||
return {"status": 500, "body": json.dumps({"error": str(exc)})}
|
||||
Reference in New Issue
Block a user