【一】使用cloudflare实现最低成本的RAG 个人知识库

1. 前言

赛博菩萨Cloudflare又出手了，不仅提供免费额度的AI 模型调用，还教咱们怎么做自己的RAG( Retrieval-Augmented Generation (检索增强生成))

RAG 是什么？

RAG 是一种结合了信息检索和大语言模型生成的技术架构,用来让 AI 回答问题时能够:

先从外部知识库检索相关信息再基于检索到的内容生成回答

而这一切通过Cloudflare实现的话，只需要开一个Workers Paid计划（每月5美元）😹，其中还包含了一堆其他的功能就不在这详细列出了。

文章以下内容根据Cloudflare的这篇博客完成，如果英文能力好的推荐阅读原文。

Build a Retrieval Augmented Generation (RAG) AI

1.1 想先看代码？

所有代码你都能在GitHub上找到：demo-rag-ai-tutorial

1.2 Live Demo

还有一个在线可以直接试试效果的页面：https://demo-rag.runnable.run/ui

2. 构建检索增强生成（RAG）AI · Cloudflare Workers AI 文档

本指南将带你完成使用 Cloudflare AI 搭建并部署第一个应用的全过程。你将构建一个全功能的 AI 应用，使用 Workers AI、Vectorize、D1 以及 Cloudflare Workers 等工具。

想要托管式方案？

AI Search 提供在 Cloudflare 上构建 RAG 流水线的全托管方式，涵盖数据摄取、索引与查询，开箱即用。立即开始。

完成本教程后，你将拥有一个 AI 工具，可将信息存储起来并使用大语言模型进行查询。这种模式称为检索增强生成（Retrieval Augmented Generation，RAG），你可以通过组合 Cloudflare AI 工具包中的多个能力来实现。构建该应用不需要有使用 AI 工具的经验。

注册 Cloudflare 账号。
安装 Node.js。

Node.js 版本管理器

使用 Volta 或 nvm 等 Node 版本管理器，以避免权限问题并切换 Node.js 版本。稍后会用到的 Wrangler 需要 16.17.0 或更高版本的 Node。

你还需要访问 Vectorize。本教程也会演示如何可选集成 Anthropic Claude，如需使用需准备 Anthropic API Key。

2.1. 创建新的 Worker 项目

C3（create-cloudflare-cli）是一款命令行工具，旨在帮助你快速在 Cloudflare 上创建并部署 Workers。

打开终端运行 C3 来创建你的 Worker 项目：

npm
yarn
pnpm

npm create cloudflare@latest -- rag-ai-tutorial

yarn create cloudflare rag-ai-tutorial

pnpm create cloudflare@latest rag-ai-tutorial

在交互式设置中选择如下选项：

对于“开始于什么？”，选择 Hello World example
对于“使用哪个模板？”，选择 Worker only
对于“使用哪种语言？”，选择 JavaScript
对于“是否使用 git 进行版本控制？”，选择 Yes
对于“是否部署你的应用？”，选择 No（部署前我们会做一些修改）

在你的项目目录中，C3 会生成若干文件。

C3 创建了哪些文件？

wrangler.jsonc：你的 Wrangler 配置文件
worker.js（位于 /src）：使用 ES Module 语法编写的最小化“Hello World!” Worker
package.json：最小化的 Node 依赖配置文件
package-lock.json：参见 npm 关于 package-lock.json
node_modules：参见 npm 关于 node_modules

然后进入新创建的项目目录：

cd rag-ai-tutorial

2.2. 使用 Wrangler CLI 进行开发

Workers 命令行工具 Wrangler 支持你创建、本地开发与部署 Workers 项目。C3 会默认为项目安装 Wrangler。

创建第一个 Worker 后，在项目目录运行 wrangler dev 启动本地开发服务器，便于在开发过程中进行本地测试。

npx wrangler dev

现在访问 http://localhost:8787 即可看到你的 Worker 正在运行。任何对代码的更改都会触发重构建，刷新页面即可看到最新输出。

2.3. 添加 AI 绑定

要使用 Cloudflare 的 AI 产品，可以在 Wrangler 配置文件中添加 ai 块作为远程绑定。这会在你的代码中设置一个到 Cloudflare AI 模型的绑定，以便与平台上的可用模型交互。

提示

如果你此前未使用过 Wrangler，它会尝试打开浏览器让你用 Cloudflare 账号登录。

如在此步骤遇到问题或无法使用浏览器界面，请参阅 wrangler login 文档。

本示例使用 @cf/meta/llama-3-8b-instruct 模型，该模型用于生成文本。

wrangler.jsonc
wrangler.toml

{
    "$schema": "./node_modules/wrangler/config-schema.json",
    "ai": {
    "binding": "AI",
    "remote": true
}
}

[ai]
binding = "AI"
remote = true

接着找到 src/index.js 文件。在 fetch 处理器内，你可以通过 AI 绑定发起模型调用：

export default {
  async fetch(request, env, ctx) {
    const answer = await env.AI.run("@cf/meta/llama-3-8b-instruct", {
      messages: [{ role: "user", content: `What is the square root of 9?` }],
    });


    return new Response(JSON.stringify(answer));
  },
};

通过 AI 绑定调用 LLM，我们可以在代码中直接与 Cloudflare AI 的大语言模型交互。本例使用的是 @cf/meta/llama-3-8b-instruct 模型，用于生成文本。

你可以使用 wrangler 部署 Worker：

npx wrangler deploy

向你的 Worker 发起请求将得到 LLM 生成的文本响应，并以 JSON 对象返回。

curl https://example.username.workers.dev

{"response":"Answer: The square root of 9 is 3."}

2.4. 使用 Cloudflare D1 与 Vectorize 添加向量嵌入

嵌入（Embeddings）让你在 Cloudflare AI 项目中为语言模型增加检索能力。这通过 Vectorize（Cloudflare 的向量数据库）来实现。

开始使用 Vectorize 时，先用 wrangler 创建一个嵌入索引。该索引存储 768 维向量，并使用余弦相似度来判断向量间的相似性：

npx wrangler vectorize create vector-index --dimensions=768 --metric=cosine

随后，将新建的 Vectorize 索引的配置添加到 Wrangler 配置文件：

wrangler.jsonc
wrangler.toml

{
    "$schema": "./node_modules/wrangler/config-schema.json",
    "vectorize": [
        {
            "binding": "VECTOR_INDEX",
            "index_name": "vector-index"
        }
    ]
}

# ... existing wrangler configuration


[[vectorize]]
binding = "VECTOR_INDEX"
index_name = "vector-index"

向量索引用于存储一组维度（浮点数），用来表示你的数据。当需要查询时，你也会把查询语句转换成向量。Vectorize 旨在高效找出与查询最相似的已存向量。

要实现搜索功能，你需要设置一个 Cloudflare D1 数据库。你可以在 D1 中存储应用数据，并将其转换为向量格式；当用户搜索并与某向量匹配时，再取回并展示对应数据。

使用 wrangler 创建一个新的 D1 数据库：

npx wrangler d1 create database

随后，将上一条命令输出的配置信息粘贴到 Wrangler 配置文件：

wrangler.jsonc
wrangler.toml

{
    "$schema": "./node_modules/wrangler/config-schema.json",
    "d1_databases": [
{
    "binding": "DB",
    "database_name": "database",
    "database_id": "abc-def-geh"
}
    ]
}

# ... existing wrangler configuration


[[d1_databases]]
binding = "DB" # available in your Worker on env.DB
database_name = "database"
database_id = "abc-def-geh" # replace this with a real database_id (UUID)

在本应用中，我们会在 D1 创建一个 notes 表，用于存储笔记并在 Vectorize 中检索。运行以下 SQL 创建该表：

npx wrangler d1 execute database --remote --command "CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY, text TEXT NOT NULL)"

现在，使用 wrangler d1 execute 往数据库新增一条笔记：

npx wrangler d1 execute database --remote --command "INSERT INTO notes (text) VALUES ('The best pizza topping is pepperoni')"

2.5. 创建一个 Workflow

在开始创建笔记前，引入 Cloudflare Workflow。它允许我们定义一个持久的工作流，以可靠地执行 RAG 流程中的所有步骤。

首先，在你的 Wrangler 配置文件中添加新的 [[workflows]] 配置块：

wrangler.jsonc
wrangler.toml

{
    "$schema": "./node_modules/wrangler/config-schema.json",
    "workflows": [
{
    "name": "rag",
    "binding": "RAG_WORKFLOW",
    "class_name": "RAGWorkflow"
}
    ]
}

# ... existing wrangler configuration


[[workflows]]
name = "rag"
binding = "RAG_WORKFLOW"
class_name = "RAGWorkflow"

在 src/index.js 中添加一个名为 RAGWorkflow 的类并继承 WorkflowEntrypoint：

import { WorkflowEntrypoint } from "cloudflare:workers";


export class RAGWorkflow extends WorkflowEntrypoint {
  async run(event, step) {
    await step.do("example step", async () => {
      console.log("Hello World!");
    });
  }
}

该类会定义一个简单的工作流步骤，在控制台输出 “Hello World!”。你可以按需添加更多步骤。

工作流本身不会自动执行。要触发它，需要调用 RAG_WORKFLOW 绑定，并传入工作流所需的参数。示例调用如下：

env.RAG_WORKFLOW.create({ params: { text } });

2.6. 创建笔记并写入 Vectorize

为让 Workers 处理多条路由，我们将引入 Workers 的路由库 hono，以创建一个用于添加笔记的路由。通过 npm 安装 hono：

npm
yarn
pnpm

npm i hono

yarn add hono

pnpm add hono

随后在 src/index.js 中引入 hono，并用它来改造 fetch 处理器：

import { Hono } from "hono";
const app = new Hono();


app.get("/", async (c) => {
  const answer = await c.env.AI.run("@cf/meta/llama-3-8b-instruct", {
    messages: [{ role: "user", content: `What is the square root of 9?` }],
  });


  return c.json(answer);
});


export default app;

这会在根路径 / 建立一个路由，其功能与之前的应用版本等效。

接下来，更新工作流以将笔记写入数据库，并为其生成对应的嵌入向量。

本示例使用 @cf/baai/bge-base-en-v1.5 模型，用于生成嵌入。嵌入会存储在 Vectorize 中并可被检索。用户查询同样会转换为嵌入，以便在 Vectorize 中用于搜索。

import { WorkflowEntrypoint } from "cloudflare:workers";


export class RAGWorkflow extends WorkflowEntrypoint {
  async run(event, step) {
    const env = this.env;
    const { text } = event.payload;


    const record = await step.do(`create database record`, async () => {
      const query = "INSERT INTO notes (text) VALUES (?) RETURNING *";


      const { results } = await env.DB.prepare(query).bind(text).run();


      const record = results[0];
      if (!record) throw new Error("Failed to create note");
      return record;
    });


    const embedding = await step.do(`generate embedding`, async () => {
      const embeddings = await env.AI.run("@cf/baai/bge-base-en-v1.5", {
        text: text,
      });
      const values = embeddings.data[0];
      if (!values) throw new Error("Failed to generate vector embedding");
      return values;
    });


    await step.do(`insert vector`, async () => {
      return env.VECTOR_INDEX.upsert([
        {
          id: record.id.toString(),
          values: embedding,
        },
      ]);
    });
  }
}

该工作流执行如下步骤：

接收 text 参数
向 D1 的 notes 表插入新记录，并获取新行的 id
使用 LLM 的嵌入模型将 text 转为向量
将 id 与向量 upsert 到 Vectorize 的 vector-index 索引

这样就为笔记创建了向量表示，后续可用于检索。

为完善功能，我们添加一个路由让用户提交笔记到数据库。该路由会解析 JSON 请求体，获取 text 参数，并创建工作流实例传入该参数：

app.post("/notes", async (c) => {
  const { text } = await c.req.json();
  if (!text) return c.text("Missing text", 400);
  await c.env.RAG_WORKFLOW.create({ params: { text } });
  return c.text("Created note", 201);
});

2.7. 查询 Vectorize 以检索笔记

为完善代码，你可以在根路径（/）进行 Vectorize 查询。先将问题转换为向量，再用 vector-index 索引查找最相似的向量。

topK 参数用于限制返回的向量数量。例如，topK = 1 时仅返回基于查询的“最相似”向量；设为 5 则返回最相似的 5 个。

基于相似向量列表，你可以取回与这些向量存储的记录 ID 对应的笔记。此处我们仅取回一条笔记，你可按需自定义。

你可以把这些笔记的文本作为上下文插入到 LLM 的提示词中。这正是检索增强生成（RAG）的基础：提供 LLM 之外的数据作为额外上下文，以增强其生成质量。

我们将更新提示词以包含上下文，并要求 LLM 在回答时使用这些上下文：

import { Hono } from "hono";
const app = new Hono();


// Existing post route...
// app.post('/notes', async (c) => { ... })


app.get("/", async (c) => {
  const question = c.req.query("text") || "What is the square root of 9?";


  const embeddings = await c.env.AI.run("@cf/baai/bge-base-en-v1.5", {
    text: question,
  });
  const vectors = embeddings.data[0];


  const vectorQuery = await c.env.VECTOR_INDEX.query(vectors, { topK: 1 });
  let vecId;
  if (
    vectorQuery.matches &&
    vectorQuery.matches.length > 0 &&
    vectorQuery.matches[0]
  ) {
    vecId = vectorQuery.matches[0].id;
  } else {
    console.log("No matching vector found or vectorQuery.matches is empty");
  }


  let notes = [];
  if (vecId) {
    const query = `SELECT * FROM notes WHERE id = ?`;
    const { results } = await c.env.DB.prepare(query).bind(vecId).run();
    if (results) notes = results.map((vec) => vec.text);
  }


  const contextMessage = notes.length
    ? `Context:\n${notes.map((note) => `- ${note}`).join("\n")}`
    : "";


  const systemPrompt = `When answering the question or responding, use the context provided, if it is provided and relevant.`;


  const { response: answer } = await c.env.AI.run(
    "@cf/meta/llama-3-8b-instruct",
    {
      messages: [
        ...(notes.length ? [{ role: "system", content: contextMessage }] : []),
        { role: "system", content: systemPrompt },
        { role: "user", content: question },
      ],
    },
  );


  return c.text(answer);
});


app.onError((err, c) => {
  return c.text(err);
});


export default app;

2.8. 引入 Anthropic Claude 模型（可选）

如果需要处理更大的文档，可以选择使用 Anthropic 的 Claude 模型，其上下文窗口更大，非常适合 RAG 工作流。

首先安装 @anthropic-ai/sdk 包：

npm
yarn
pnpm

npm i @anthropic-ai/sdk

yarn add @anthropic-ai/sdk

pnpm add @anthropic-ai/sdk

在 src/index.js 中，你可以更新 GET / 路由以检查 ANTHROPIC_API_KEY 环境变量：若存在则用 Anthropic SDK 生成文本；否则回退到当前的 Workers AI 代码：

import Anthropic from '@anthropic-ai/sdk';


app.get('/', async (c) => {
  // ... Existing code
  const systemPrompt = `When answering the question or responding, use the context provided, if it is provided and relevant.`


  let modelUsed: string = ""
  let response = null


  if (c.env.ANTHROPIC_API_KEY) {
    const anthropic = new Anthropic({
      apiKey: c.env.ANTHROPIC_API_KEY
    })


    const model = "claude-3-5-sonnet-latest"
    modelUsed = model


    const message = await anthropic.messages.create({
      max_tokens: 1024,
      model,
      messages: [
        { role: 'user', content: question }
      ],
      system: [systemPrompt, notes ? contextMessage : ''].join(" ")
    })


    response = {
      response: message.content.map(content => content.text).join("\n")
    }
  } else {
    const model = "@cf/meta/llama-3.1-8b-instruct"
    modelUsed = model


    response = await c.env.AI.run(
      model,
      {
        messages: [
          ...(notes.length ? [{ role: 'system', content: contextMessage }] : []),
          { role: 'system', content: systemPrompt },
          { role: 'user', content: question }
        ]
      }
    )
  }


  if (response) {
    c.header('x-model-used', modelUsed)
    return c.text(response.response)
  } else {
    return c.text("We were unable to generate output", 500)
  }
})

最后，需要在 Workers 应用中设置 ANTHROPIC_API_KEY 环境变量。可通过以下命令添加：

$ npx wrangler secret put ANTHROPIC_API_KEY

2.9. 删除笔记与向量

如果不再需要某条笔记，你可以从数据库删除它。删除笔记的同时也需要删除 Vectorize 中对应的向量。可在 src/index.js 中实现 DELETE /notes/:id 路由：

app.delete("/notes/:id", async (c) => {
  const { id } = c.req.param();

  const query = `DELETE FROM notes WHERE id = ?`;
  await c.env.DB.prepare(query).bind(id).run();

  await c.env.VECTOR_INDEX.deleteByIds([id]);

  return c.status(204);
});

2.10. 文本切分（可选）

针对大段文本，建议将其切分为更小的块。这样有助于 LLM 更有效地聚合相关上下文，而不必检索过大的文本。

为实现该功能，我们在项目中添加 NPM 包 @langchain/textsplitters：

npm
yarn
pnpm

npm i @langchain/textsplitters

yarn add @langchain/textsplitters

pnpm add @langchain/textsplitters

该包中的 RecursiveCharacterTextSplitter 类可将文本切分为更小的块。你可以按需自定义配置，默认配置在多数场景下已足够：

import { RecursiveCharacterTextSplitter } from "@langchain/textsplitters";


const text = "Some long piece of text...";


const splitter = new RecursiveCharacterTextSplitter({
  // These can be customized to change the chunking size
  // chunkSize: 1000,
  // chunkOverlap: 200,
});


const output = await splitter.createDocuments([text]);
console.log(output); // [{ pageContent: 'Some long piece of text...' }]

为使用该切分器，我们会更新工作流把文本切分为多个块，然后遍历每个块并针对每块执行后续工作流步骤：

export class RAGWorkflow extends WorkflowEntrypoint {
  async run(event, step) {
    const env = this.env;
    const { text } = event.payload;
    let texts = await step.do("split text", async () => {
      const splitter = new RecursiveCharacterTextSplitter();
      const output = await splitter.createDocuments([text]);
      return output.map((doc) => doc.pageContent);
    });


    console.log(
      "RecursiveCharacterTextSplitter generated ${texts.length} chunks",
    );


    for (const index in texts) {
      const text = texts[index];
      const record = await step.do(
        `create database record: ${index}/${texts.length}`,
        async () => {
          const query = "INSERT INTO notes (text) VALUES (?) RETURNING *";


          const { results } = await env.DB.prepare(query).bind(text).run();


          const record = results[0];
          if (!record) throw new Error("Failed to create note");
          return record;
        },
      );


      const embedding = await step.do(
        `generate embedding: ${index}/${texts.length}`,
        async () => {
          const embeddings = await env.AI.run("@cf/baai/bge-base-en-v1.5", {
            text: text,
          });
          const values = embeddings.data[0];
          if (!values) throw new Error("Failed to generate vector embedding");
          return values;
        },
      );


      await step.do(`insert vector: ${index}/${texts.length}`, async () => {
        return env.VECTOR_INDEX.upsert([
          {
            id: record.id.toString(),
            values: embedding,
          },
        ]);
      });
    }
  }
}

现在，当向 /notes 端点提交大段文本时，它们将被切分为更小的块，并由工作流分别处理。

2.11. 部署你的项目

如果你未在步骤 1中部署 Worker，请使用 Wrangler 部署到 *.workers.dev 子域，或部署到你已配置的自定义域名。若尚未配置任何子域或域名，发布过程中 Wrangler 会提示你进行设置。

npx wrangler deploy

可在 <YOUR_WORKER>.<YOUR_SUBDOMAIN>.workers.dev 预览你的 Worker。

注意

首次推送到 *.workers.dev 子域时，DNS 传播期间可能会看到 523 错误。这些错误通常在一分钟左右会自动消失。

3. 实测结果

3.1 新增笔记

3.2 工作流执行

3.3 查询知识库

4. 结语

到这我们完成了一个简单个人RAG应用，也是知识库+AI结合应用的一种方式。

5.相关资源

所有代码你都能在GitHub上找到：demo-rag-ai-tutorial

Cloudflare 官方也提供了一个版本： https://github.com/kristianfreeman/cloudflare-retrieval-augmented-generation-example/

进一步了解：

查看 RAG 架构参考图
查阅 Cloudflare 的 AI 文档
浏览教程在 Workers 上构建项目
试用示例以复制粘贴代码进行实验
了解 Workers 的工作原理参见参考
学习 Workers 的功能与特性参见平台
设置 Wrangler 以编程方式创建、测试与部署你的 Worker 项目

1. 前言​

1.1 想先看代码？​

1.2 Live Demo​

2. 构建检索增强生成（RAG）AI · Cloudflare Workers AI 文档​

2.1. 创建新的 Worker 项目​

2.2. 使用 Wrangler CLI 进行开发​

2.3. 添加 AI 绑定​

2.4. 使用 Cloudflare D1 与 Vectorize 添加向量嵌入​

2.5. 创建一个 Workflow​

2.6. 创建笔记并写入 Vectorize​

2.7. 查询 Vectorize 以检索笔记​

2.8. 引入 Anthropic Claude 模型（可选）​

2.9. 删除笔记与向量​

2.10. 文本切分（可选）​

2.11. 部署你的项目​

3. 实测结果​

3.1 新增笔记​

3.2 工作流执行​

3.3 查询知识库​

4. 结语​

5.相关资源​