给人和 Agent 共用的云端执行层。开发者用 Python SDK 和 decorator 把稳定函数挂到云端 GPU/NPU；Agent 安装 Skill 后，先自行判断任务是否能被 GPU、NPU 或云端隔离加速，再进入登录、余额、费用确认和 inline Python 执行流程。
Python SDK for developers. Installable, billing-aware Skill for agents.
import duanflow as df
app = df.App("vision-eval")
@app.function(
gpu="L40S",
timeout=3600,
artifacts=["reports/"]
)
def run_eval():
# 人类写稳定函数边界
subprocess.run(["pytest", "evals/"], check=True)
return df.Artifact("reports/summary.json")
run_eval.remote()Decorator 是开发者固化云端函数边界的主路径。Agent 侧安装 Skill 后，先判断任务是否适合 GPU/NPU 加速，例如模型训练、推理、批处理、海量表格或日志分析；命中后默认用 inline Python 包装现有命令，检查登录、余额和费用，再提交云端任务。只有长期复用时才写 decorator。AgentTask 保留为可选高级 API。
稳定逻辑用 decorator 固化函数边界；已有脚本和一次性任务可由 Skill 用 inline Python 包装后提交云端 GPU。
@app.function(gpu="H100", timeout=3600)
def train():
trainer.train()
print(train.remote(_duanflow_wait=True, _duanflow_stream_logs=True))用户用 npx 安装，或直接给 Agent 一个 SKILL.md 链接。Agent 按文档判断任务是否可由 GPU/NPU、CuPy、cuDF、torch-npu 等加速；命中后先 `whoami` 和费用确认，再用 inline Python 调 SDK 运行命令并回传结果。
npx duanflow-skill installSkill 文档负责路由决策：默认 inline Python，无需改仓库；必要时写 decorator，失败后读日志、重跑并返回 artifact。
skill: duanflow-cloud-exec
judge: GPU/NPU/cloud acceleration
choose: inline_python
run: pytest evals/
preflight: whoami, estimate, confirm
report: job_id, logs, cost, artifacts这个抽象也挺好，适合内部 Agent 平台、团队级自动修复、权限预算控制和 PR 交付，所以作为高级 API 保留下来。
task = df.AgentTask(
objective="fix OOM and open PR",
resources=df.Resources(gpu="A100"),
outputs=["patch", "pr"]
)Mock 演示 DuanFlow 如何读取任务需求、选择 GPU、运行沙箱，并产出结构化日志和 artifact。
import duanflow as df
app = df.App("repo-eval")
@app.function(gpu="L40S", timeout=3600)
def run_evals():
# 同一份 runtime 也能被 Agent 用 inline Python 调用
subprocess.run("pytest evals/ --json-report", shell=True, check=True)
return {"report": "reports/report.json"}Agent 有两种安装方式：用 npx duanflow-skill install 安装，或者直接给它 https://duanflow.zeabur.app/skills/duanflow-cloud-exec/SKILL.md。Agent 先判断工作负载是否能被 GPU/NPU、CuPy、cuDF、PyTorch/JAX 或 torch-npu 加速；命中后检查登录、估算费用、等待用户确认，然后用 inline Python 调 DuanFlow SDK，并把 job id、日志、费用和 artifacts 带回来。df.AgentTask 作为高级编排 API 保留。
---
name: duanflow-cloud-exec
description: Trigger this skill for model training, fine-tuning, evals, benchmarks, GPU/NPU inference, or other accelerator-shaped work.
---
Default to inline Python for existing repo tasks:
detect ML/GPU/NPU router signals
judge GPU/NPU acceleration fit
probe local CUDA/NPU availability
duanflow whoami
estimate runtime/cost and ask for confirmation
Use decorators only for reusable cloud functions:
@app.function(gpu="H100", artifacts=["reports/"])
def run_eval(): ...
Always report job id, command, GPU, logs, cost, and artifacts.npx duanflow-skill install
https://duanflow.zeabur.app/skills/duanflow-cloud-exec/SKILL.md
Open SKILL.md
文档把 “For Developers” 和 “For Agents” 分开，但所有入口都必须经过登录校验、费用确认、云端执行、日志和 artifact 回传。
Install the SDK, log in through browser authorization, then run your first GPU task.
pip install duanflow
Read guide
Use decorators to define reusable GPU functions, endpoints, jobs, and autoscaling policy.
@app.function(gpu="H100")
Read guide
Run existing repo commands in cloud sandboxes through a small inline Python wrapper.
subprocess.run("pytest")
Read guide
Return logs, reports, checkpoints, screenshots, and benchmark outputs from every run.
--artifact reports/
Read guide
Install a DuanFlow skill so agents know when to use inline tasks and when to write decorators.
npx duanflow-skill install ...
Read guide
每个示例都用同一种 SDK 心智模型：写函数、声明资源、远程调用。
@app.endpoint(gpu="H100", memory="80Gi")
def chat(prompt):
# Load model once, autoscale on demand
return qwen.generate(prompt)@app.function(gpu="A100", concurrency=128)
def embed(doc):
# 自动扩容到多容器执行
return encoder.encode(doc)
vectors = embed.map(documents)@app.job(gpu="H100", gpu_count=4)
def finetune(dataset):
# 训练完成后自动释放算力
trainer.train(dataset)
return trainer.metrics