零成本构建边缘 AI 应用：Cloudflare Workers + D1 + R2 + Vectorize + AI Gateway 生产级完全指南（2026）

前言：为什么 2026 年是边缘 AI 的拐点

2026 年，AI 推理不再属于"必须在数据中心跑 GPU 集群"的范畴。

Cloudflare Workers AI 自 2023 年推出以来，已经从"概念验证"进化为"生产级平台"。它让你能在全球 300+ 边缘节点上，以 50ms 级别延迟运行 Llama 3、Mistral、Stable Diffusion 等模型。更重要的是，围绕 Workers AI 打造一套完整应用所需的存储、数据库、向量检索、AI 网关能力，Cloudflare 全部原生提供，而且免费额度足够个人开发者折腾一年。

本文从实战出发，用一个完整的"AI 知识库问答"应用串联 Workers AI、D1、R2、Vectorize、AI Gateway 五大组件，深入讲解每个服务的底层原理、代码实现、性能调优和避坑指南。读完这篇，你就能从零搭建一套真正能在生产环境跑起来的边缘 AI 系统。

一、架构总览：一张图看清边缘 AI 应用的全貌

在动手之前，先搞清楚每个组件在架构中的位置和职责。

                                    ┌─────────────────────────────────────┐
                                    │           全球 300+ 边缘节点           │
                                    │                                      │
用户请求 ──▶  Workers ──┬──▶ Workers AI (Llama/Qwen) ──▶ LLM 推理       │
                        │                                           │
                        ├──▶ Vectorize (向量数据库) ──▶ RAG 检索       │
                        │                                           │
                        ├──▶ D1 (SQLite 边缘数据库) ──▶ 结构化数据      │
                        │                                           │
                        └──▶ R2 (对象存储) ──────────▶ 文件/图片存储     │
                                    │                                      │
                                    └─────────────────────────────────────┘
                                                ▲
                                                │
                                    ┌─────────────────────────┐
                                    │      AI Gateway         │
                                    │  (多模型路由/限流/监控)  │
                                    └─────────────────────────┘

关键组件职责：

• Workers：轻量级无状态函数，处理请求路由、业务逻辑、V8 沙箱隔离
• Workers AI：GPU 加速的 AI 推理，运行大语言模型、嵌入模型
• Vectorize：向量数据库，做语义检索和 RAG
• D1：SQLite 兼容的边缘关系数据库，存用户数据、会话历史
• R2：S3 兼容的对象存储，存图片、PDF、文档等二进制文件
• AI Gateway：统一接入层，做多模型路由、请求限流、费用监控

二、环境准备：Wrangler CLI 与项目初始化

2.1 安装与认证

# 全局安装 Wrangler CLI
npm install -g wrangler

# 验证安装
wrangler --version
# wrangler 4.x.x

# 登录 Cloudflare 账户
wrangler login
# 会自动打开浏览器，完成 OAuth 授权

2.2 初始化 Workers 项目

# 创建新项目（TypeScript 模板）
wrangler generate my-edge-ai-app
cd my-edge-ai-app

# 目录结构
my-edge-ai-app/
├── src/
│   └── index.ts          # Worker 入口
├── wrangler.toml         # 配置文件
├── package.json
└── tsconfig.json

2.3 wrangler.toml 核心配置

name = "my-edge-ai-app"
main = "src/index.ts"
compatibility_date = "2026-01-01"

# 启用 Workers AI（会自动绑定 AI 命名空间）
ai = { binding = "AI" }

# 绑定 D1 数据库
[[d1_databases]]
binding = "DB"
database_name = "ai-knowledge-base"
database_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

# 绑定 R2 存储桶
[[r2_buckets]]
binding = "ASSETS"
bucket_name = "ai-knowledge-assets"

三、Workers AI：50ms 延迟的边缘大模型推理

3.1 Workers AI 是什么，为什么值得关注

Workers AI 运行在 Cloudflare 拥有的 GPU（NVIDIA A100/H100）集群上，通过 V8 隔离的 WASM 运行时执行模型推理。关键特性：

• 全球分布：模型权重预加载到 300+ 节点，首次推理无需跨区域
• 毫秒级冷启动：比 Lambda/Vercel Functions 快 100 倍（~0ms vs ~500ms）
• SSE 流式输出：支持 Server-Sent Events，实时流式响应
• 零运维：无需管理 GPU 集群，模型自动缩放

3.2 支持的模型矩阵（2026 年 6 月）

3.3 基本对话：最简单的 Workers AI 调用

// src/index.ts
export interface Env {
  AI: Ai;
}

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);

    // 仅处理 /ai 路由
    if (url.pathname !== "/ai") {
      return new Response("Not Found", { status: 404 });
    }

    if (request.method !== "POST") {
      return new Response("Method Not Allowed", { status: 405 });
    }

    const { prompt, model = "@cf/meta/llama-3-8b-instruct" } = await request.json();

    if (!prompt) {
      return new Response(JSON.stringify({ error: "prompt is required" }), {
        status: 400,
        headers: { "Content-Type": "application/json" },
      });
    }

    // 调用 Workers AI
    const answer = await env.AI.run(model, {
      messages: [
        { role: "system", content: "你是一个专业的 AI 助手，用中文回答。" },
        { role: "user", content: prompt },
      ],
      stream: false,
    });

    return new Response(JSON.stringify({ answer, model }), {
      headers: { "Content-Type": "application/json" },
    });
  },
} satisfies ExportedHandler<Env>;

3.4 流式输出：让 AI 打字给你看

非流式响应的体验很糟糕，真实产品必须用 SSE：

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    if (request.method !== "POST") {
      return new Response("Method Not Allowed", { status: 405 });
    }

    const { prompt, model = "@cf/meta/llama-3-8b-instruct" } = await request.json();

    // 流式推理
    const stream = await env.AI.run(model, {
      messages: [
        { role: "system", content: "你是一个专业的 AI 助手，用中文回答。" },
        { role: "user", content: prompt },
      ],
      stream: true,
    });

    // 直接转发流式响应
    return new Response(stream, {
      headers: {
        "Content-Type": "text/event-stream; charset=utf-8",
        "Transfer-Encoding": "chunked",
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
        // CORS 头（如果前端跨域调用）
        "Access-Control-Allow-Origin": "*",
      },
    });
  },
} satisfies ExportedHandler<Env>;

前端接收 SSE 的代码：

async function streamChat(prompt: string) {
  const response = await fetch("/ai", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ prompt }),
  });

  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let partial = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    partial += decoder.decode(value, { stream: true });
    // SSE 格式：data: {"response":"xxx"}\n\n
    const lines = partial.split("\n");
    partial = lines.pop() ?? "";

    for (const line of lines) {
      if (line.startsWith("data: ")) {
        const data = JSON.parse(line.slice(6));
        if (data.response) {
          process.stdout.write(data.response); // 流式打印
        }
      }
    }
  }
}

3.5 Workers AI 的性能调优

延迟分析：从边缘节点到模型推理的完整链路：

用户请求 → DNS解析(5ms) → CF边缘Worker(0ms) → 模型推理(20-80ms) → 响应

实测数据（从中国用户访问新加坡节点）：

优化策略：

1. 模型选择：英文场景用 Llama，中文场景用 Qwen，不要跨语言使用
2. 上下文截断：通过 max_tokens 限制输出长度，避免无效计算
3. 推理参数调优：

const answer = await env.AI.run("@cf/qwen/qwen2.5-7b-instruct", {
  messages: [...],
  max_tokens: 512,       // 限制输出长度，节省 GPU 时间
  temperature: 0.7,       // 控制随机性（0=确定性，1=高随机）
  top_p: 0.9,            // nucleus sampling
});

四、D1：SQLite 兼容的边缘关系数据库

4.1 D1 的核心原理

D1 是基于 SQLite 的边缘数据库，通过以下技术实现全球分布：

• 存储层：Cloudflare Durable Object 存储数据，零复制读取
• 查询层：V8 隔离的 SQLite 运行时，WASM 编译执行
• 复制策略：每个 D1 写操作自动同步到全球 300+ 节点（最终一致）
• 索引优化：支持 B-Tree 索引（与 SQLite 完全兼容）

免费额度：5GB 存储，1000 万行读取/天，10 万行写入/月

4.2 创建 D1 数据库

# 创建数据库
wrangler d1 create ai-knowledge-base
# 输出：database_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

# 将 ID 写入 wrangler.toml
# [[d1_databases]]
# binding = "DB"
# database_name = "ai-knowledge-base"
# database_id = "刚才的ID"

4.3 定义 Schema

-- schema.sql

-- 用户表
CREATE TABLE IF NOT EXISTS users (
  id TEXT PRIMARY KEY,
  email TEXT UNIQUE NOT NULL,
  name TEXT,
  created_at INTEGER DEFAULT (unixepoch()),
  updated_at INTEGER
);

-- 对话会话表
CREATE TABLE IF NOT EXISTS sessions (
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  title TEXT,
  model TEXT DEFAULT '@cf/meta/llama-3-8b-instruct',
  token_count INTEGER DEFAULT 0,
  created_at INTEGER DEFAULT (unixepoch()),
  FOREIGN KEY (user_id) REFERENCES users(id)
);

-- 消息历史表（支持 RAG 上下文）
CREATE TABLE IF NOT EXISTS messages (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  session_id TEXT NOT NULL,
  role TEXT NOT NULL,          -- 'user' | 'assistant' | 'system'
  content TEXT NOT NULL,
  tokens INTEGER DEFAULT 0,
  created_at INTEGER DEFAULT (unixepoch()),
  FOREIGN KEY (session_id) REFERENCES sessions(id)
);

-- RAG 文档表（配合 Vectorize 使用）
CREATE TABLE IF NOT EXISTS documents (
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  title TEXT NOT NULL,
  content TEXT NOT NULL,
  chunk_count INTEGER DEFAULT 0,
  created_at INTEGER DEFAULT (unixepoch()),
  FOREIGN KEY (user_id) REFERENCES users(id)
);

-- 索引优化
CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);
CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id);
CREATE INDEX IF NOT EXISTS idx_messages_created ON messages(created_at);
CREATE INDEX IF NOT EXISTS idx_documents_user ON documents(user_id);

4.4 初始化数据库

wrangler d1 execute ai-knowledge-base --file=./schema.sql --local
wrangler d1 execute ai-knowledge-base --file=./schema.sql --remote

4.5 Workers 中使用 D1

export interface Env {
  DB: D1Database;
  AI: Ai;
}

// 获取会话历史（用于上下文）
async function getSessionHistory(sessionId: string, limit = 10): Promise<Array<{role: string; content: string}>> {
  const { results } = await env.DB
    .prepare(`
      SELECT role, content FROM messages
      WHERE session_id = ?
      ORDER BY created_at DESC
      LIMIT ?
    `)
    .bind(sessionId, limit)
    .all();

  return results.reverse().map((r: any) => ({
    role: r.role,
    content: r.content,
  }));
}

// 保存消息到 D1
async function saveMessage(sessionId: string, role: string, content: string, tokens = 0) {
  await env.DB
    .prepare(`INSERT INTO messages (session_id, role, content, tokens) VALUES (?, ?, ?, ?)`)
    .bind(sessionId, role, content, tokens)
    .run();
}

// RAG：获取相关文档片段
async function getRelevantDocuments(userId: string, queryEmbedding: number[], topK = 3) {
  // 从 Vectorize 获取相似文档 ID
  const vectorResults = await env.AI.run("@cf/baai/bge-base-zh-v1.5", {
    texts: [queryEmbedding], // 这里简化，实际需要用 Vectorize API
  });

  // 从 D1 获取文档内容
  const docIds = vectorResults.map((r: any) => r.id).slice(0, topK);
  if (docIds.length === 0) return [];

  const placeholders = docIds.map(() => "?").join(",");
  const { results } = await env.DB
    .prepare(`SELECT * FROM documents WHERE id IN (${placeholders}) AND user_id = ?`)
    .bind(...docIds, userId)
    .all();

  return results;
}

4.6 D1 的坑与最佳实践

坑 1：WASM SQLite 的 LIMIT 限制

D1 的 SQLite 是 WASM 编译版本，部分 SQL 语法有差异：

-- ✅ 支持
SELECT * FROM messages ORDER BY created_at DESC LIMIT 100;

-- ❌ 不支持（部分版本）
SELECT * FROM messages OFFSET 50 LIMIT 50;  -- 用 cursor 分页替代

正确做法：用游标分页（keyset pagination）替代 OFFSET：

// 替代 OFFSET 10，用 created_at 做游标
const { results } = await env.DB
  .prepare(`
    SELECT * FROM messages
    WHERE session_id = ? AND created_at < ?
    ORDER BY created_at DESC
    LIMIT 20
  `)
  .bind(sessionId, cursorTimestamp)
  .all();

坑 2：写操作不是实时的

D1 写操作是异步复制到全球节点的，通常 60 秒内完成。读取自己刚写入的数据有时会失败。

解决方案：对于需要立即读取刚写入数据的场景，使用 Durable Object 替代：

// 对于需要强一致性的写后读场景
export class UserSession implements DurableObject {
  async fetch(request: Request): Promise<Response> {
    // 强一致读写
  }
}

五、Vectorize：边缘向量数据库与 RAG 实战

5.1 Vectorize 架构解析

Vectorize 是 Cloudflare 原生的向量检索服务，底层基于 Facebook FAISS 的分布式实现：

• 索引类型：HNSW（Hierarchical Navigable Small World）—— 近似最近邻搜索
• 维度：支持 768 维（bge-base-en）和 1024 维（bge-base-zh）嵌入向量
• 免费额度：100 万维向量，30 万次查询/月
• 延迟：全球平均 P50 < 20ms

5.2 创建 Vectorize 索引

# 创建索引（维度要与嵌入模型匹配）
wrangler vectorize create ai-knowledge-embeddings \
  --dimensions=768 \
  --metric=cosine \
  --description="AI知识库嵌入向量"

# 输出：index_id = "xxxxxxxx"

5.3 文档向量化与入库

完整 RAG 流程：文档 → 分块 → 向量化 → 存储 → 检索 → 生成

// 1. 文档分块（chunking）- 合理的块大小是 RAG 效果的关键
function chunkText(text: string, chunkSize = 500, overlap = 50): string[] {
  const chunks: string[] = [];
  for (let i = 0; i < text.length; i += chunkSize - overlap) {
    chunks.push(text.slice(i, i + chunkSize));
  }
  return chunks;
}

// 2. 嵌入向量化
async function embedChunks(chunks: string[], env: Env): Promise<number[][]> {
  const results = await Promise.all(
    chunks.map(async (chunk) => {
      const res = await env.AI.run("@cf/baai/bge-base-zh-v1.5", { text: chunk });
      return res.data[0].embedding as number[];
    })
  );
  return results;
}

// 3. 存入 Vectorize
async function indexDocuments(
  docId: string,
  chunks: string[],
  embeddings: number[][],
  metadata: Record<string, string>,
  env: Env
) {
  const vectors = chunks.map((chunk, i) => ({
    id: `${docId}-${i}`,
    values: embeddings[i],
    metadata: {
      docId,
      chunkIndex: i,
      content: chunk.slice(0, 200), // Vectorize metadata 有大小限制
      ...metadata,
    },
  }));

  // 批量插入（每次最多 1000 条）
  await env.VECTORIZE.index(vectors);
}

// 4. 语义检索
async function semanticSearch(query: string, topK = 5, env: Env): Promise<any[]> {
  // 查询文本向量化
  const queryEmbedding = await env.AI.run("@cf/baai/bge-base-zh-v1.5", { text: query });

  // 向量检索
  const results = await env.VECTORIZE.query(queryEmbedding.data[0].embedding, {
    topK,
    returnMetadata: true,
  });

  return results.matches.map((m: any) => ({
    id: m.id,
    score: m.score,
    content: m.metadata?.content,
    docId: m.metadata?.docId,
  }));
}

5.4 完整的 RAG 生成流程

async function ragChat(userQuery: string, userId: string, env: Env) {
  // Step 1: 语义检索相关文档
  const relevantDocs = await semanticSearch(userQuery, 5, env);

  if (relevantDocs.length === 0) {
    return "抱歉，知识库中没有找到相关信息。";
  }

  // Step 2: 构建 RAG 上下文
  const context = relevantDocs
    .map((d, i) => `[参考文档 ${i + 1}]\n${d.content}`)
    .join("\n\n");

  // Step 3: 带上下文的增强生成
  const augmentedPrompt = `
根据以下参考文档回答用户问题。如果文档中没有相关信息，请如实说明。

【参考文档】
${context}

【用户问题】
${userQuery}

【回答要求】
1. 结合参考文档内容作答
2. 在回答中标注参考来源
3. 如果文档不足以回答，请说明"以下答案基于我的通用知识..."
`;

  // Step 4: 调用 LLM 生成答案
  const response = await env.AI.run("@cf/qwen/qwen2.5-7b-instruct", {
    messages: [
      {
        role: "system",
        content: "你是一个专业的知识库问答助手，根据提供的参考文档回答用户问题。",
      },
      { role: "user", content: augmentedPrompt },
    ],
    stream: false,
    max_tokens: 1024,
    temperature: 0.3, // 低温度，RAG 场景需要准确性
  });

  return {
    answer: response.response,
    references: relevantDocs.map((d) => ({
      docId: d.docId,
      score: d.score,
      snippet: d.content.slice(0, 100),
    })),
  };
}

5.5 Vectorize 性能调优

HNSW 参数调优：

# 创建索引时可以指定 HNSW 参数
wrangler vectorize create ai-knowledge-embeddings \
  --dimensions=768 \
  --metric=cosine \
  --hnsw_ef_construction=128 \  # 构建时精度（高 = 精准但慢）
  --hnsw_m=16                    # 内存/速度权衡

EF 搜索参数（在查询时动态调整）：

// 精确但慢（适合高质量答案场景）
const exactResults = await env.VECTORIZE.query(vector, { topK: 5 });

// 快速但略有精度损失（适合大量候选筛选）
const fastResults = await env.VECTORIZE.query(vector, {
  topK: 20,
  topK: { retrieve: 5 }, // 内部取 20 个再筛选回 5 个
});

六、R2：S3 兼容的边缘对象存储

6.1 R2 vs S3 vs 传统 CDN

R2 是 Cloudflare 的对象存储服务，核心差异：

R2 的最大优势：免费出口流量。对于图片、PDF 密集型应用，这能节省 90%+ 的存储成本。

6.2 创建 R2 存储桶

# 创建存储桶
wrangler r2 bucket create ai-knowledge-assets

# 公开访问配置（wrangler.toml）
[[r2_buckets]]
binding = "ASSETS"
bucket_name = "ai-knowledge-assets"

6.3 文件上传与公开访问

export interface Env {
  ASSETS: R2Bucket;
}

// 生成预签名上传 URL（让前端直传 R2）
export async function generateUploadUrl(
  filename: string,
  contentType: string,
  env: Env
): Promise<{ uploadUrl: string; fileKey: string }> {
  const fileKey = `${Date.now()}-${crypto.randomUUID()}-${filename}`;

  // 使用 Workers 的预签名 URL 功能
  const uploadUrl = `https://api.cloudflare.com/client/v4/accounts/${env.CF_ACCOUNT_ID}/r2/upload`;

  return {
    uploadUrl,
    fileKey,
  };
}

// 在 Workers 中处理上传（支持大文件分片）
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);

    // 上传文件
    if (url.pathname === "/upload" && request.method === "POST") {
      const formData = await request.formData();
      const file = formData.get("file") as File;
      const userId = (formData.get("userId") as string) ?? "anonymous";

      const fileKey = `${userId}/${Date.now()}-${file.name}`;

      await env.ASSETS.put(fileKey, file.stream(), {
        httpMetadata: {
          contentType: file.type,
          contentDisposition: `inline; filename="${file.name}"`,
        },
        customMetadata: {
          uploadedBy: userId,
          originalName: file.name,
        },
      });

      // 构建公开访问 URL
      const publicUrl = `https://pub-${env.CF_ACCOUNT_ID}.r2.dev/${fileKey}`;

      return Response.json({ fileKey, publicUrl });
    }

    // 列出用户文件
    if (url.pathname === "/files" && request.method === "GET") {
      const userId = url.searchParams.get("userId") ?? "anonymous";
      const prefix = `${userId}/`;

      const listed = await env.ASSETS.list({
        prefix,
        limit: 100,
      });

      const files = listed.objects.map((obj) => ({
        key: obj.key,
        size: obj.size,
        uploaded: obj.uploaded,
        etag: obj.httpEtag,
        url: `https://pub-${env.CF_ACCOUNT_ID}.r2.dev/${obj.key}`,
      }));

      return Response.json({ files });
    }

    return new Response("Not Found", { status: 404 });
  },
} satisfies ExportedHandler<Env>;

6.4 前端直传 R2（避免流量经过 Workers）

真实应用中，前端直接上传到 R2 能节省 Workers 的 CPU 配额：

// Workers 端：生成临时凭证
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    if (request.method !== "POST") {
      return new Response("Method Not Allowed", { status: 405 });
    }

    const { filename, contentType } = await request.json();
    const fileKey = `uploads/${Date.now()}-${filename}`;

    // 生成 R2 的 PUT URL（有效期 1 小时）
    const putUrl = await env.ASSETS.createSignedUrl(fileKey, {
      expirationTime: Date.now() + 3600 * 1000,
      httpMetadata: { contentType },
    });

    return Response.json({ putUrl, fileKey });
  },
} satisfies ExportedHandler<Env>;

// 前端：直接 PUT 到 R2
async function uploadToR2(file: File) {
  // 获取签名 URL
  const { putUrl, fileKey } = await fetch("/get-upload-url", {
    method: "POST",
    body: JSON.stringify({ filename: file.name, contentType: file.type }),
  }).then((r) => r.json());

  // 直接上传到 R2
  await fetch(putUrl, {
    method: "PUT",
    body: file,
    headers: { "Content-Type": file.type },
  });

  return fileKey;
}

七、AI Gateway：多模型路由与成本控制

7.1 为什么需要 AI Gateway

当你开始组合使用多个 AI 提供商（OpenAI、Anthropic、Cloudflare Workers AI、本地模型）时，AI Gateway 变得不可或缺：

• 统一接入：一个端点，对接所有 AI 提供商
• 成本控制：请求聚合、缓存、限流
• 可靠性：自动重试、熔断、故障转移
• 可观测性：详细的用量统计和延迟分析
• 模型路由：根据请求内容自动选择最合适的模型

7.2 Workers AI 作为 AI Gateway

下面用 Workers 构建一个轻量级 AI Gateway：

export interface Env {
  AI: Ai;
  // 可以扩展支持更多 AI 提供商
  OPENAI_API_KEY: string;
}

interface AIModelConfig {
  provider: "cloudflare" | "openai";
  model: string;
  cache?: boolean;
  maxTokens?: number;
}

const MODEL_ROUTING: Record<string, AIModelConfig> = {
  "fast": { provider: "cloudflare", model: "@cf/qwen/qwen2.5-7b-instruct", cache: true, maxTokens: 512 },
  "smart": { provider: "cloudflare", model: "@cf/meta/llama-3-8b-instruct", maxTokens: 2048 },
  "vision": { provider: "openai", model: "gpt-4o", maxTokens: 4096 },
};

// 请求缓存（基于 KV）
async function getCachedResponse(promptHash: string, env: Env): Promise<string | null> {
  return env.AI_CACHE.get(promptHash, "text");
}

async function cacheResponse(promptHash: string, response: string, env: Env) {
  // 缓存 1 小时（3600 秒）
  await env.AI_CACHE.put(promptHash, response, { expirationTtl: 3600 });
}

// AI Gateway 核心逻辑
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const { prompt, mode = "fast", useCache = true } = await request.json();

    // 计算缓存 key（基于 prompt + mode 的哈希）
    const promptHash = await crypto.subtle.digest(
      "SHA-256",
      new TextEncoder().encode(`${prompt}:${mode}`)
    ).then((b) => btoa(String.fromCharCode(...new Uint8Array(b))));

    // 尝试命中缓存
    if (useCache) {
      const cached = await getCachedResponse(promptHash, env);
      if (cached) {
        return Response.json({ answer: cached, cached: true, mode });
      }
    }

    // 模型路由
    const config = MODEL_ROUTING[mode] ?? MODEL_ROUTING["fast"];

    let answer: string;
    if (config.provider === "cloudflare") {
      const result = await env.AI.run(config.model, {
        messages: [{ role: "user", content: prompt }],
        max_tokens: config.maxTokens,
      });
      answer = result.response;
    } else if (config.provider === "openai") {
      // 调用 OpenAI API
      const openaiRes = await fetch("https://api.openai.com/v1/chat/completions", {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          Authorization: `Bearer ${env.OPENAI_API_KEY}`,
        },
        body: JSON.stringify({
          model: config.model,
          messages: [{ role: "user", content: prompt }],
          max_tokens: config.maxTokens,
        }),
      });
      const data = await openaiRes.json();
      answer = data.choices[0].message.content;
    }

    // 缓存结果
    if (useCache) {
      await cacheResponse(promptHash, answer, env);
    }

    return Response.json({ answer, cached: false, mode, model: config.model });
  },
} satisfies ExportedHandler<Env>;

7.3 限流与成本监控

// 基于 KV 的滑动窗口限流
async function rateLimit(userId: string, limit: number, windowSecs: number, env: Env): Promise<boolean> {
  const key = `rl:${userId}:${Math.floor(Date.now() / (windowSecs * 1000))}`;
  const current = await env.RATE_LIMIT_KV.get(key, "text");

  if (current && parseInt(current) >= limit) {
    return false; // 触发限流
  }

  await env.RATE_LIMIT_KV.put(
    key,
    String((parseInt(current ?? "0") + 1)),
    { expirationTtl: windowSecs }
  );
  return true;
}

// 在请求处理中加入限流
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const userId = request.headers.get("X-User-ID") ?? "anonymous";

    // 免费用户：每分钟 20 次
    if (!(await rateLimit(userId, 20, 60, env))) {
      return Response.json(
        { error: "Rate limit exceeded. Free tier: 20 requests/minute." },
        { status: 429, headers: { "Retry-After": "60" } }
      );
    }

    // ... 后续处理
    return Response.json({ success: true });
  },
} satisfies ExportedHandler<Env>;

八、生产部署：从开发到全球上线的完整流水线

8.1 开发阶段

# 本地开发（热重载）
wrangler dev --port 8787

# 预览部署（staging）
wrangler deploy --env preview

# 生产部署
wrangler deploy

8.2 环境隔离策略

# wrangler.toml
[env.preview]
name = "my-edge-ai-app-preview"
routes = ["preview.my-domain.com/*"]

[env.production]
name = "my-edge-ai-app"
routes = ["my-domain.com/*"]

8.3 成本估算（实际数字）

以一个中等规模 AI 应用为例（月活 10 万用户，每人每天 20 次对话）：

总计：约 $75/月——相当于一台低配云服务器的零头。

九、架构深度：从 V8 沙箱到 GPU 推理的底层原理

理解底层原理，才能真正优化应用。

9.1 Workers 的 V8 沙箱隔离

Workers 运行在 Cloudflare 的 V8 引擎上，每个 Worker 实例在一个独立的 V8 隔离环境中执行：

┌──────────────────────────────────────────────────────┐
│                  Cloudflare Edge Node               │
│                                                     │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐               │
│  │Worker A │ │Worker B │ │Worker C │  ...          │
│  │(V8 ISO) │ │(V8 ISO) │ │(V8 ISO) │               │
│  └─────────┘ └─────────┘ └─────────┘               │
│                                                     │
│  ┌──────────────────────────────────────────────┐  │
│  │           Cloudflare Global Network          │  │
│  │  (Anycast 路由 / Durable Objects / KV / D1) │  │
│  └──────────────────────────────────────────────┘  │
│                                                     │
│  ┌──────────────────────────────────────────────┐  │
│  │           GPU Cluster (Workers AI)           │  │
│  │     (NVIDIA A100 / H100, Triton Inference)   │  │
│  └──────────────────────────────────────────────┘  │
└──────────────────────────────────────────────────────┘

关键特性：

• 零共享内存：每个 Worker 隔离，防止数据泄露
• 毫秒级冷启动：V8 隔离比 Lambda 容器快 50 倍
• CPU 限制：每个 Worker 最多 10ms CPU 时间/请求（适合 I/O 密集型）
• 内存限制：128MB（对 SQLite 查询足够）

9.2 Workers AI 的推理链路

用户请求 → CF Edge Worker → 
  → AI Gateway (路由选择) → 
  → 模型路由 (选最优节点) → 
  → GPU 推理 (VLLM/Triton) → 
  → 流式响应 → 
  → Worker (组装 SSE) → 
  → 用户

Workers AI 使用 VLLM 作为推理引擎，实现了：

• PagedAttention：类操作系统分页管理 KV Cache，显存利用率提升 40%
• 连续批处理（Continuous Batching）：动态批处理多个请求，GPU 利用率提升 3 倍
• 投机解码（Speculative Decoding）：小模型预测 + 大模型验证，延迟降低 50%

十、避坑指南：我在生产环境中踩过的那些坑

坑 1：AI 响应内容过长导致超时

问题：Workers 有 50ms CPU 时间限制，AI 生成超长文本时会超时。

解决：始终设置 max_tokens，并在超时时用 cursor 续接：

// 设置合理的 token 上限
const response = await env.AI.run(model, {
  messages,
  max_tokens: 1024, // 必须设置，防止超时
});

// 如果需要更长内容，分段生成
if (response.response.length === 1024) {
  const continuation = await env.AI.run(model, {
    messages: [...messages, { role: "assistant", content: response.response }],
  });
  // 拼接续写内容
}

坑 2：D1 写操作延迟导致数据不一致

问题：在 /api/chat 中写入消息后立即读取，有时读不到。

解决：改用乐观更新模式，先显示本地数据，后台异步持久化：

async function chat(request: Request, env: Env) {
  const { sessionId, prompt } = await request.json();

  // 1. 先显示用户消息（乐观更新）
  const userMsg = { id: crypto.randomUUID(), role: "user", content: prompt };

  // 2. 并行处理：AI 推理 + 持久化用户消息
  const [aiResponse, _] = await Promise.all([
    env.AI.run("@cf/qwen/qwen2.5-7b-instruct", { messages: [...] }),
    env.DB.prepare("INSERT INTO messages ...").run(), // 不 await 结果
  ]);

  // 3. 用户立即看到 AI 响应
  return Response.json({ ... });
}

坑 3：Vectorize 元数据大小超限

问题：Vectorize 每条记录的 metadata 限制 2KB。

解决：大内容存 R2，小摘要存 Vectorize：

// ✅ 正确：Vectorize 只存摘要
{
  id: "doc-123-chunk-0",
  values: embedding,
  metadata: {
    docId: "doc-123",
    chunkIndex: 0,
    contentPreview: content.slice(0, 150), // 前 150 字符作为摘要
    r2Key: "documents/doc-123.txt",          // 完整内容存 R2
  }
}

坑 4：Wrangler 版本不一致导致部署失败

问题：本地 Wrangler 版本与 Cloudflare 控制台不兼容。

解决：

# 锁定版本（推荐）
npm install wrangler@4.x.x --save-dev

# 或者用 npx 始终使用最新版
npx wrangler deploy

总结：边缘 AI 的技术哲学

回顾整个架构，Cloudflare Workers + AI 生态代表了一种新的技术哲学：把复杂性留在平台，把简单留给开发者。

传统架构下，搭建一套 AI 应用需要：

• 云服务器 + GPU 实例 + 负载均衡：$200+/月起步
• 数据库 + 缓存 + 对象存储：$50+/月
• CDN + WAF + 流量管理：$30+/月
• AI 网关 + 监控系统：$20+/月

而 Cloudflare 的这套方案：

• 零服务器运维：没有 EC2、没有 Kubernetes、没有运维人员
• 零出口费用：R2 对象存储完全免费出海流量
• 零冷启动：V8 沙箱的毫秒级响应
• 零认知负荷：一个 Wrangler.toml 管理所有配置

2026 年的今天，如果你还在花 $300/月 维护一台 GPU 云服务器，是时候认真考虑边缘 AI 架构了。

当然，这套方案也有局限性：

• CPU 时间限制（50ms/请求）：不适合超长推理任务
• 模型受限：无法运行超过 70B 参数的模型
• 生态系统：不如 AWS/GCP 成熟，部分场景需要自建

对于个人开发者、Side Project、创业公司 MVP 阶段，这套方案的性价比是无可匹敌的。等业务规模超过这套方案的天花板时，再考虑迁移到更重的架构也不迟。

技术选型的本质，从来不是选"最好的"，而是选"当前阶段最合适的"。

相关资源：

• Cloudflare Workers AI 文档：https://developers.cloudflare.com/workers-ai/
• D1 文档：https://developers.cloudflare.com/d1/
• Vectorize 文档：https://developers.cloudflare.com/vectorize/
• R2 文档：https://developers.cloudflare.com/r2/
• Wrangler CLI：https://developers.cloudflare.com/workers/wrangler/

标签：Cloudflare|Workers AI|D1|R2|Vectorize|AI Gateway|边缘计算|Serverless|无服务器|向量数据库|RAG|嵌入向量|对象存储