Cloudflare Workflows V2：确定性重放执行与 5 万并发的工作流引擎

// worker.ts — Cloudflare Workers + Workflows V2 入口
import {
  WorkflowEntrypoint,
  WorkflowEvent,
  WorkflowStep,
} from "cloudflare:workers";

interface Env {
  AI: any;           // Cloudflare Workers AI binding
  KV: KVNamespace;   // KV 存储绑定
  WORKFLOW: Workflow; // Workflows binding
}

// 定义工作流参数
type Params = {
  topic: string;
};

// 工作流类——每个 step 的返回值自动持久化，重放时跳过已完成步骤
class AgentWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // 步骤 1：搜索相关资料（模拟调用搜索 API）
    const searchResults = await step.do("search", {
      retries: { max: 3, delay: "5s" },
    }, async () => {
      // 这里替换成实际搜索 API 调用
      const resp = await fetch(
        `https://api.example.com/search?q=${encodeURIComponent(event.payload.topic)}`
      );
      if (!resp.ok) throw new Error(`Search failed: ${resp.status}`);
      const data = await resp.json() as { snippets: string[] };
      return data.snippets;
    });

    // 步骤 2：用 LLM 总结搜索结果
    const summary = await step.do("summarize", {
      retries: { max: 2, delay: "10s" },
    }, async () => {
      const prompt = `基于以下资料，用中文写一段关于"${event.payload.topic}"的简要总结：\n\n${searchResults.join("\n")}`;
      const { response } = await this.env.AI.run("@cf/meta/llama-3.1-8b-instruct", {
        messages: [{ role: "user", content: prompt }],
      });
      return response as string;
    });

    // 步骤 3：写入 KV 存储
    await step.do("store", async () => {
      await this.env.KV.put(
        `summary:${event.payload.topic}`,
        JSON.stringify({ topic: event.payload.topic, summary, createdAt: Date.now() })
      );
    });

    return { topic: event.payload.topic, summary };
  }
}

export default {
  // HTTP 入口：触发工作流
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    if (url.pathname === "/trigger" && request.method === "POST") {
      const { topic } = await request.json() as Params;
      if (!topic) return new Response("Missing topic", { status: 400 });

      // 创建工作流实例
      const instance = await env.WORKFLOW.create({ params: { topic } });
      return Response.json({
        id: instance.id,
        status: "queued",
        topic,
      });
    }

    // 查询工作流状态
    if (url.pathname === "/status" && request.method === "GET") {
      const id = url.searchParams.get("id");
      if (!id) return new Response("Missing id", { status: 400 });
      const instance = await env.WORKFLOW.get(id);
      const status = await instance.status();
      return Response.json(status);
    }

    return new Response("Not found", { status: 404 });
  },
};

name = "agent-workflow"
main = "worker.ts"
compatibility_date = "2024-09-23"

[vars]
# 无需额外环境变量

[[kv_namespaces]]
binding = "KV"
id = "你的 KV namespace ID"

[ai]
binding = "AI"

[[workflows]]
binding = "WORKFLOW"
name = "agent-workflow"
class = "AgentWorkflow"

# 安装依赖
npm install wrangler --save-dev

# 登录 Cloudflare
npx wrangler login

# 创建 KV namespace（如果还没有）
npx wrangler kv namespace create "WORKFLOW_KV"

# 部署
npx wrangler deploy

# 触发一个工作流实例
curl -X POST https://agent-workflow.your-subdomain.workers.dev/trigger \
  -H "Content-Type: application/json" \
  -d '{"topic": "Cloudflare Workflows V2 确定性执行"}'

# 返回：{"id":"abc123","status":"queued","topic":"Cloudflare Workflows V2 确定性执行"}

# 查询执行状态
curl "https://agent-workflow.your-subdomain.workers.dev/status?id=abc123"

# 返回包含每个 step 的状态、耗时、输出摘要