Docs / 文档Docs

DuanFlow 文档

端流把 Python 装饰器、inline 云端任务、Agent skill、日志和 artifact contract 收敛到同一个云端执行层。人类开发者可以把函数挂到 GPU/NPU，编程 Agent 会先判断任务是否可由 GPU、NPU、CuPy、cuDF 或云端隔离加速，再进入登录、费用确认和执行流程。

DuanFlow Documentation

DuanFlow combines Python decorators, inline cloud tasks, agent skills, logs, and artifact contracts into one cloud execution layer. Developers can attach functions to GPUs/NPUs; coding agents first decide whether a workload benefits from GPU, NPU, CuPy, cuDF, or cloud isolation before entering login, cost confirmation, and execution.

快速开始

快速开始有两条路径：已有脚本先用 inline Python 无侵入上云；需要长期复用时，再用 @app.function 声明稳定函数边界。Agent 使用 Skill 时，会先判断任务是否适合 GPU/NPU 加速或云端长运行。

Quickstart

There are two quickstart paths: use inline Python for existing scripts and repo commands first; introduce @app.function when the workload deserves a reusable function boundary. Agents using the skill first judge whether GPU/NPU acceleration or long-running cloud execution fits the task.

terminal + app.py

pip install duanflow
duanflow auth login
duanflow whoami

# Agent-friendly path: first judge acceleration fit, then ask for cost confirmation.
# Example: model evals can use cloud GPU; large dataframe jobs may use cuDF.

# Developer path: turn a stable function boundary into cloud GPU compute.

import subprocess
import duanflow as df

app = df.App("repo-eval")

@app.function(gpu="L40S", timeout=3600, artifacts=["reports/"])
def run_evals():
    subprocess.run("pytest evals/ --json-report", shell=True, check=True)
    return {"report": "reports/report.json"}

run_evals.remote(_duanflow_wait=True, _duanflow_stream_logs=True)

应用、镜像与 Secret

df.App 是部署单元。它描述应用名称、运行镜像、依赖、环境变量、Secret 和默认资源策略。一个 App 可以同时包含函数、Endpoint、队列 worker 和训练任务。

Apps, Images, and Secrets

df.App is the deployment unit. It describes the app name, runtime image, dependencies, environment variables, secrets, and default resource policy. One app can contain functions, endpoints, queue workers, and training jobs.

Image固定 CUDA、Python 和系统依赖，避免线上环境漂移。Pin CUDA, Python, and system dependencies to avoid runtime drift.

Secrets把 HF_TOKEN、数据库密码和对象存储凭据注入运行时。Inject HF_TOKEN, database passwords, and object storage credentials at runtime.

app.py

import duanflow as df

image = (
    df.Image.cuda("12.4", python="3.11")
    .pip_install("torch", "transformers", "accelerate")
    .apt_install("ffmpeg")
)

app = df.App(
    "customer-ai-api",
    image=image,
    secrets=[df.Secret.from_name("hf-prod")],
    env={"MODEL_ID": "Qwen/Qwen2.5-32B-Instruct"},
)

GPU 函数

GPU 函数适合被 Python 代码、批处理、队列或另一个 Endpoint 调用。模型加载应该放在模块级或 class 初始化阶段，这样 warm worker 可以复用权重。

GPU Functions

GPU functions are best for compute called by Python code, batches, queues, or another endpoint. Put model loading at module scope or inside a lifecycle class so warm workers can reuse weights.

inference.py

from transformers import AutoModelForCausalLM, AutoTokenizer
import duanflow as df

app = df.App("qwen-summarizer")

tokenizer = AutoTokenizer.from_pretrained(os.environ["MODEL_ID"])
model = AutoModelForCausalLM.from_pretrained(os.environ["MODEL_ID"]).cuda()

@app.function(gpu="H100", memory="80Gi", concurrency=12)
def summarize(document):
    prompt = "Summarize this document in 5 bullets:\n" + document
    tokens = tokenizer(prompt, return_tensors="pt").to("cuda")
    output = model.generate(**tokens, max_new_tokens=512)
    return tokenizer.decode(output[0], skip_special_tokens=True)

Inline 云端任务

Inline Python 是 Agent 无侵入运行现有命令的主路径。Agent 先判断任务是否适合 GPU/NPU 加速、云端隔离或长运行，例如模型 eval、训练、批处理、CuPy/cuDF 数据分析；命中后再登录、估算费用并等待确认。

Inline Cloud Tasks

Inline Python is the default non-invasive path for agents running existing commands. The agent first judges whether GPU/NPU acceleration, cloud isolation, or long runtime applies, then logs in, estimates cost, and waits for confirmation.

terminal

import subprocess
import duanflow as df

app = df.App("repo-command")

@app.function(gpu="L40S", timeout=3600, artifacts=["reports/"])
def run_command():
    subprocess.run(
        "pytest evals/ --json-report --json-report-file reports/eval.json",
        shell=True,
        check=True,
    )
    return {"report": "reports/eval.json"}

print(run_command.remote(_duanflow_wait=True, _duanflow_stream_logs=True))

任务、队列与批处理

长时间训练用 @app.job，事件驱动处理用 @app.worker，海量数据 fan-out 用 .map()。这些模式共用同一套日志、重试和资源策略。

Jobs, Queues, and Batches

Use @app.job for long training runs, @app.worker for event-driven processing, and .map() for large fan-out workloads. They share logs, retries, and resource policies.

jobs.py

audio_queue = df.Queue("audio-inbox", max_retries=3)

@app.worker(queue=audio_queue, gpu="A10G", scale_to=12)
def transcribe_audio(job):
    transcript = whisper.transcribe(job["input_url"])
    return df.storage.write_json(job["output_url"], transcript)

@app.job(gpu="A100-80G", gpu_count=4, timeout=10800)
def train_lora(dataset_uri):
    return trainer.fit(dataset_uri, rank=16)

日志与 Artifact

Agent 不应该只拿到一段终端输出。DuanFlow 把失败原因、资源状态、成本、报告和产物路径整理成可读也可解析的结果契约。

Logs and Artifacts

Agents need more than terminal output. DuanFlow turns failures, resource state, cost, reports, and artifact paths into a result contract that is readable and machine-parseable.

result.json

{
  "job_id": "job_8f31",
  "status": "failed",
  "resource": {"gpu": "L40S", "memory_used": "47.2Gi"},
  "error_type": "cuda_oom",
  "artifacts": ["reports/report.json", "logs/pytest.jsonl"],
  "suggested_next_actions": ["retry batch_size=8", "request A100-80G"]
}

Agent Skill 安装

Agent 接入主推 Skill，安装方式只有两种：用 npx duanflow-skill install 安装，或者直接把 https://duanflow.zeabur.app/skills/duanflow-cloud-exec/SKILL.md 交给 Agent。Agent 先自行判断任务是否能被 GPU/NPU、CuPy、cuDF、PyTorch/JAX 或 torch-npu 加速；命中后按 Skill 指令登录、估算费用、等待确认、运行 inline Python，并返回 artifacts。df.AgentTask 只是保留下来的高级编排 API，适合团队系统化工作流。

Agent Skill Install

Agent integration is primarily a skill with two install paths: use npx duanflow-skill install, or give the agent https://duanflow.zeabur.app/skills/duanflow-cloud-exec/SKILL.md. The agent first judges whether GPU/NPU, CuPy, cuDF, PyTorch/JAX, torch-npu, or cloud isolation fits the workload, then follows the skill to log in, estimate cost, wait for approval, run inline Python, and return artifacts. df.AgentTask remains as an advanced orchestration API for team workflows.

1. Install skill 用 npx 安装，或直接给 Agent 一个 Zeabur 上的 SKILL.md 链接。 Install with npx, or give the agent the hosted SKILL.md URL.

2. Agent judges acceleration 先判断 GPU/NPU、CuPy、cuDF、RAPIDS、torch-npu 或云端隔离是否能带来收益。 First decide whether GPU/NPU, CuPy, cuDF, RAPIDS, torch-npu, or cloud isolation helps.

3. Preflight and execute Agent 运行 whoami、估算时间和费用，确认后用 inline Python 调 SDK。 The agent runs whoami, estimates time and cost, then uses inline Python after approval.

4. Report back 最终回传 job id、加速原因、命令、资源、费用、日志状态和产物路径。 The final answer includes job id, acceleration reason, command, resource, cost, log status, and artifact paths.

Optional. AgentTask 保留的高级能力。内部 Agent 平台可以用 SDK 建模 workspace、权限、预算、输出和 PR 交付。 Internal agent platforms can model workspace, permissions, budget, outputs, and PR delivery through the SDK.

install

# Install from a skills repository once published
npx duanflow-skill install

# Or give this hosted SKILL.md URL to your agent
https://duanflow.zeabur.app/skills/duanflow-cloud-exec/SKILL.md

agent prompt after install

Use the duanflow-cloud-exec skill.
Run this repo's GPU evals in the cloud:

1. First decide whether this eval benefits from GPU/NPU or cloud isolation.
2. Run `duanflow whoami`, estimate runtime and cost, then ask me to confirm.
3. If confirmed, run `pytest evals/ --json-report` on L40S via inline Python.
4. Return acceleration reason, job id, logs, cost, artifact paths, and failure classification.
5. If it fails from CUDA OOM, estimate the retry before submitting again.

what the agent runs

duanflow whoami

import subprocess
import duanflow as df

app = df.App("agent-eval")

@app.function(gpu="L40S", timeout=3600, artifacts=["reports/"])
def run_eval():
    subprocess.run(
        "pytest evals/ --json-report --json-report-file reports/eval.json",
        shell=True,
        check=True,
    )
    return {"report": "reports/eval.json"}

print(run_eval.remote(_duanflow_wait=True, _duanflow_stream_logs=True))

agent final report

{
  "job_id": "job_8f31",
  "acceleration_reason": "model eval uses CUDA GPU and local machine has no suitable device",
  "command": "pytest evals/ --json-report",
  "gpu": "L40S",
  "estimated_cost": "about 6 CNY before confirmation",
  "status": "failed_then_retried_successfully",
  "first_failure": "cuda_oom",
  "artifacts": ["reports/eval.json", "logs/events.jsonl"]
}

advanced_agent_task.py

import duanflow as df

task = df.AgentTask(
    name="fix-gpu-inference-oom",
    agent="codex",
    workspace=df.GitRepo(
        url="github.com/acme/model-service",
        branch="main",
        write_branch="agent/fix-gpu-oom",
    ),
    objective="Reproduce CUDA OOM, fix the minimal cause, rerun GPU tests, and open a draft PR.",
    resources=df.Resources(cpu=16, memory="64Gi", gpu="A100-80G"),
    budget=df.Budget(max_cost="¥50", timeout="3h"),
    permissions=[
        df.Permission.shell_install(scope="project"),
        df.Permission.network(hosts=["pypi.org", "github.com"]),
        df.Permission.git(branch_prefix="agent/", pull_request=True),
    ],
    outputs=[
        df.Output.patch(),
        df.Output.pull_request(draft=True),
        df.Output.test_report(command="pytest tests/inference -q"),
        df.Output.benchmark(name="batch_latency"),
    ],
)

run = df.agent.submit(task)
print(run.stream_events())

部署、日志与版本

部署会生成不可变 release。你可以查看构建日志、运行日志、请求日志和调度事件；回滚时只需要指定历史 release。

Deploy, Logs, and Releases

Deployments create immutable releases. You can inspect build logs, runtime logs, request logs, and scheduler events; rollback points to a previous release.

terminal

duanflow deploy app.py --env prod --name customer-ai-api
duanflow logs customer-ai-api --follow
duanflow logs customer-ai-api --kind scheduler --since 30m
duanflow releases customer-ai-api
duanflow rollback customer-ai-api --to rel_20260505_1842

SDK 速查

这些是 mock 页面中使用的核心 SDK 概念，方便演示时快速解释能力边界。

SDK Reference Cheatsheet

These are the core SDK concepts used throughout the mock pages, useful when explaining the product surface during a demo.

df.AppDeployment unit, image, env, secrets

@app.functionRemote Python compute on CPU/GPU

@app.endpointHTTPS API backed by autoscaling workers

@app.jobLong-running training or migration task

df.QueueEvent-driven queue with retry policy

duanflow logsBuild, runtime, request, scheduler logs

df.AgentTaskAdvanced agent workflow orchestration