本地 RAG 管道：Bright Data 与 ChromaDB

在本文中，你将了解：

什么是检索增强生成（RAG）和 ChromaDB，以及它们各自带来了什么。
为什么将 Bright Data 的网络解锁器 API 与 ChromaDB 搭配，是让语言模型基于新鲜、真实世界数据的实用方式。
如何构建一个端到端管道，在你自己的机器上收集网页内容、本地嵌入这些内容，并基于它们回答问题。

在进入工具和代码之前，先明确这些概念，并看看它们如何在 RAG 工作流中结合起来，会很有帮助。

什么是检索增强生成（RAG）？

大型语言模型只知道它训练时接触过的内容。问它关于上周才上线的页面、内部 wiki 或小众产品目录的问题，它要么会猜测，要么会告诉你它不知道。检索增强生成弥合了这一差距。

这个想法很简单。你不是依赖模型的记忆，而是把自己的文档存储在一个可搜索索引中。当问题传入时，你首先检索与其最相关的文本块，然后通过将这些文本块粘贴为上下文来增强提示，最后让模型基于这些材料生成答案。模型仍然负责写作，但事实来自你控制的数据。

这很重要有两个原因。答案保持最新，因为你决定什么进入索引以及何时刷新它。答案也保持准确，因为模型是在真实源文本上推理，而不是从训练数据中填补空白。对于知识库、支持助手、研究工具，以及任何基于专有或快速变化信息构建的东西，RAG 已成为默认方法。

检索侧讲完后，让我们看看这些文档实际存放在哪里。

什么是 ChromaDB？

ChromaDB（通常简称为 Chroma）是一个为 AI 应用构建的开源向量数据库。普通数据库匹配精确值；向量数据库匹配含义。它通过存储嵌入来做到这一点，嵌入是文本的数值表示，相似的想法会在向量空间中彼此靠近。当你查询它时，Chroma 会返回其嵌入最接近你的问题的已存储文本块，即使它们没有任何完全相同的词重合。

Chroma 适合作为第一次 RAG 构建的原因，是它对你的要求很少。它通过单个 pip install 安装，嵌入在你的 Python 进程中运行，无需管理单独的服务器，并通过 PersistentClient 将所有内容持久化到磁盘，因此你的索引在重启后仍然存在。你可以接入任何嵌入模型，从 OpenAI 或 Google API，到完全不接触网络的本地模型。对于一个需要在笔记本电脑上运行并保持私密的项目来说，这种组合很难被超越。

RAG 与微调：区别是什么？

刚接触这一领域的人经常会把 RAG 和微调放在一起权衡，好像它们是相互竞争的选项。它们解决的是不同问题：

微调会改变模型本身。你用示例重新训练它，使其适应语气、格式或行为。当你希望模型以不同方式行动时，它是正确的工具，但它很慢、成本高，而且你嵌入进去的知识会在数据变化的那一刻过时。
RAG不改变模型，而是在提问时改变它看到的内容。你在每次请求时将相关上下文注入提示中。当你希望模型知道特定且最新的内容时，它是正确的工具，而更新它的成本就像向索引中添加一份文档一样低。

大多数团队最终采用的粗略规则是：如果你需要新知识，就选择 RAG；如果你需要新行为，就考虑微调。许多生产系统会同时使用两者。在本教程中，我们将完全专注于 RAG，因为目标是基于变化过于频繁、无法为其重新训练模型的网页内容来回答问题。

为什么将 Bright Data 集成到 RAG + ChromaDB 管道中？

RAG 系统的效果取决于你喂给它的文档质量。垃圾进，听起来很自信的垃圾出。因此，大多数管道中的真正瓶颈不是向量搜索或模型，而是一开始就获取干净、可靠、最新的源材料。

当你的数据已经位于一组 PDF 文件夹中时，这很直接。一旦你的知识需要来自开放网络，事情就会变得困难。公共页面隐藏在机器人检测之后，使用 JavaScript 渲染内容，按地区提供不同结果，并向任何看起来自动化的东西抛出 CAPTCHA。维护你自己的爬虫工具来应对所有这些本身就是一项工作，而且每当目标网站更改其标记时，这项工作都会中断。

这就是 Bright Data 的网络解锁器 API 发挥作用的地方。你向它发送目标 URL，它会返回页面内容，并在幕后处理反机器人措施、代理轮换、JavaScript 渲染和地理定位。你甚至可以要求它以干净的 Markdown 返回页面，这对 RAG 来说接近理想状态，因为它会去掉导航和样板内容，留下可读文本供你分块和嵌入。无需浏览器自动化，也无需照看代理池。

与 Chroma 的本地向量存储和本地运行模型结合后，你会得到一个在需要保持最新的地方保持最新、在其他所有地方保持私密的管道：只有数据收集步骤会访问网络，而嵌入、存储、检索和生成都留在你的机器上。

这种模式尤其适用于：

基于最新文章、论文或文档来回答问题的研究助手，而不是基于模型的训练快照。
跟踪一组竞争对手网站上的产品、定价或功能页面的竞争情报工具。
基于实时文档的内部支持机器人，因此答案反映当前文档，而不是几个月前的版本。
拉取新鲜列表或新闻，并让分析师用自然语言查询它们的市场监控系统。

通过让 Bright Data 的网页数据基础设施处理收集，让 Chroma 处理检索，你无需编写一行爬虫代码，就能获得一个具有生产形态的 RAG 引擎。

如何使用 Bright Data 和 ChromaDB 构建本地 RAG 管道

在这个引导部分中，你将构建一个包含三个阶段的管道：

收集网页内容：脚本调用 Bright Data 的网络解锁器 API，以获取一组页面，并将每个页面作为 Markdown 返回。
嵌入并存储：第二个脚本对该内容分块，使用 sentence-transformer 模型在本地生成嵌入，并将向量写入 ChromaDB。
检索并生成：查询脚本会嵌入你的问题，从 Chroma 中拉取最相关的文本块，并将它们传给本地模型，以生成带来源的、有依据的答案。

注意：这只是许多可能设计中的一种。你可以将本地模型替换为 API 调用以获得更高质量的答案，在检索和生成之间添加重排序步骤，或将收集阶段指向数百个 URL 而不是三个。结构保持不变。

按照以下步骤构建一个由 Bright Data 的网络解锁器 API 和 ChromaDB 提供支持的完全本地 RAG 管道。

先决条件

要继续操作，你需要：

一个拥有活跃网络解锁器区域的 Bright Data 账户。登录你的控制面板，前往账户设置，并复制你的 API 令牌（它将采用 UUID 格式）。同时记下你的区域名称；两者你都需要。
本地安装 Python 3.10+。
已安装 Ollama，并为生成步骤拉取了一个模型（本教程使用 llama3.1，但任何聊天模型都可以）。如果你更愿意使用托管模型，步骤 5 会展示在哪里替换为 API 调用。

步骤 1：项目设置

创建一个工作目录，设置虚拟环境，并安装依赖项：

mkdir local-rag-pipeline && cd local-rag-pipeline
python -m venv venv
source venv/bin/activate        # On Windows: venv\Scripts\activate
pip install requests chromadb sentence-transformers

第一次运行该管道时，sentence-transformers 会下载嵌入模型（几百兆字节）。之后它会从缓存加载并离线运行。

如果你还没有为生成步骤拉取模型，请拉取一个：

ollama pull llama3.1

步骤 2：创建你的项目结构

创建管道要写入的文件夹：

mkdir -p data/raw data/chroma

你的项目结构将如下所示：

local-rag-pipeline/
├── data/
│   ├── raw/            # Raw Markdown collected from the web
│   └── chroma/         # Persisted ChromaDB index
├── collect.py          # Stage 1: fetch pages via Bright Data
├── ingest.py           # Stage 2: chunk, embed, and store
└── rag.py              # Stage 3: retrieve and generate

步骤 3：使用 Bright Data 收集网页数据

创建文件 collect.py：

import json
import requests
from pathlib import Path

API_KEY = "your-brightdata-api-token-here"
ZONE = "web_unlocker1"
BASE_URL = "https://api.brightdata.com/request"
RAW_DATA_PATH = "data/raw/pages.json"

HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

TARGETS = [
    "https://en.wikipedia.org/wiki/Retrieval-augmented_generation",
    "https://en.wikipedia.org/wiki/Vector_database",
    "https://en.wikipedia.org/wiki/Large_language_model",
]

def fetch_pages():
    results = []
    for url in TARGETS:
        print(f"Fetching: {url}")
        response = requests.post(
            BASE_URL,
            headers=HEADERS,
            json={
                "zone": ZONE,
                "url": url,
                "format": "raw",
                "data_format": "markdown",
            },
            timeout=60,
        )
        response.raise_for_status()
        results.append({"url": url, "content": response.text})
        print(f"  -> {len(response.text)} chars")

    Path(RAW_DATA_PATH).parent.mkdir(parents=True, exist_ok=True)
    with open(RAW_DATA_PATH, "w") as f:
        json.dump(results, f, indent=2)

    print(f"Saved {len(results)} pages to {RAW_DATA_PATH}")

if __name__ == "__main__":
    fetch_pages()

将 your-brightdata-api-token-here 替换为你的实际 API 令牌，并更新 ZONE 以匹配你的网络解锁器区域名称。

以下是每个部分的作用：

API_KEY 和 ZONE：你的 Bright Data 凭据。API 令牌是来自你账户设置的 UUID 格式令牌，而不是区域密码。
TARGETS：要摄取的页面。这里的三篇 Wikipedia 文章提供了一个连贯的语料库，可用于提问。替换为你自己的 URL；这正是网络解锁器发挥价值的地方，因为通过 API，新闻站点、产品页面以及会阻止普通请求的 JavaScript 密集型应用都会干净地返回。
fetch_pages：遍历每个 URL，并向网络解锁器端点发送 POST 请求。data_format: "markdown" 选项会告诉 Bright Data 返回可读 Markdown，而不是原始 HTML，这省去了一个解析步骤。结果会写入单个 JSON 文件，供下一阶段读取。

注意： 如果某个网站在 Bright Data 的即时访问模式下受限，某些页面可能会返回 bad_endpoint 消息。这是预期行为；Bright Data 会在响应中显示该错误，而不是静默失败。如果你需要对受限目标拥有完整访问权限，请联系你的客户经理。

步骤 4：将内容嵌入 ChromaDB

创建文件 ingest.py：

import json
import chromadb
from chromadb.utils import embedding_functions

RAW_DATA_PATH = "data/raw/pages.json"
CHROMA_PATH = "data/chroma"
COLLECTION_NAME = "web_knowledge"

def chunk_text(text, size=800, overlap=100):
    words = text.split()
    chunks, i = [], 0
    while i < len(words):
        chunks.append(" ".join(words[i:i + size]))
        i += size - overlap
    return chunks

def main():
    with open(RAW_DATA_PATH) as f:
        pages = json.load(f)

    client = chromadb.PersistentClient(path=CHROMA_PATH)
    embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
        model_name="all-MiniLM-L6-v2"
    )
    collection = client.get_or_create_collection(
        name=COLLECTION_NAME,
        embedding_function=embed_fn,
    )

    documents, metadatas, ids = [], [], []
    for page in pages:
        for idx, chunk in enumerate(chunk_text(page["content"])):
            documents.append(chunk)
            metadatas.append({"source": page["url"], "chunk": idx})
            ids.append(f"{page['url']}#{idx}")

    collection.upsert(documents=documents, metadatas=metadatas, ids=ids)

    print(f"Indexed {len(documents)} chunks from {len(pages)} pages")
    print(f"Collection now holds {collection.count()} chunks")

if __name__ == "__main__":
    main()

这一阶段承担了将原始文本转换为可搜索内容的主要工作：

chunk_text：将每个页面拆分为大约 800 个词的重叠窗口。分块很重要，因为你想检索聚焦的段落，而不是整个页面，并且 100 个词的重叠可以避免跨越边界的句子被切成两半。
SentenceTransformerEmbeddingFunction：加载 all-MiniLM-L6-v2，这是一个小型、快速且本地运行的嵌入模型。每当你添加或查询文档时，Chroma 会自动调用它，因此你永远不需要手动处理向量。
get_or_create_collection：打开磁盘上的持久集合，首次运行时创建它，之后复用它。
collection.upsert：将文本块、它们的元数据和稳定 ID 写入 Chroma。使用 upsert 而不是 add 意味着你可以在收集新鲜内容后重新运行脚本，而不会遇到重复 ID 错误。

步骤 5：构建检索和生成步骤

创建文件 rag.py：

import sys
import requests
import chromadb
from chromadb.utils import embedding_functions

CHROMA_PATH = "data/chroma"
COLLECTION_NAME = "web_knowledge"
OLLAMA_URL = "http://localhost:11434/api/generate"
MODEL = "llama3.1"

PROMPT_TEMPLATE = """You are a research assistant. Answer the question using only the context below.
If the context does not contain the answer, say you don't have enough information.

Context:
{context}

Question: {question}

Answer:"""

def retrieve(question, n_results=4):
    client = chromadb.PersistentClient(path=CHROMA_PATH)
    embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
        model_name="all-MiniLM-L6-v2"
    )
    collection = client.get_collection(
        name=COLLECTION_NAME, embedding_function=embed_fn
    )
    results = collection.query(query_texts=[question], n_results=n_results)
    chunks = results["documents"][0]
    sources = [m["source"] for m in results["metadatas"][0]]
    return chunks, sources

def generate(question, chunks):
    prompt = PROMPT_TEMPLATE.format(context="\n\n".join(chunks), question=question)
    response = requests.post(
        OLLAMA_URL,
        json={"model": MODEL, "prompt": prompt, "stream": False},
        timeout=120,
    )
    response.raise_for_status()
    return response.json()["response"]

def main(question):
    chunks, sources = retrieve(question)
    answer = generate(question, chunks)

    print("\n=== Answer ===")
    print(answer.strip())
    print("\n=== Sources ===")
    for src in dict.fromkeys(sources):   # de-duplicate, keep order
        print(f"  - {src}")

if __name__ == "__main__":
    main(" ".join(sys.argv[1:]))

流程如下：

retrieve：使用摄取时使用的同一模型嵌入你的问题（这种一致性是相似性搜索能工作的原因），并向 Chroma 请求四个最近的文本块。它会返回每个文本块的文本和源 URL。
generate：构建一个提示，将模型限定在检索到的上下文中，并指示它在答案不存在时承认这一点，这是防止幻觉的最有效保护措施。然后它通过 Ollama 的 REST API 调用本地模型。
main：将两者连接起来，并打印答案以及去重后的来源列表，因此每个响应都可以追溯到它来自的页面。

如果你更喜欢托管模型，这是唯一需要更改的函数。将对 Ollama 的 requests.post 调用替换为对 Anthropic 或 OpenAI API 的调用，并传入相同提示；检索部分保持完全不变。

步骤 6：运行管道

按顺序运行三个阶段。首先，收集页面：

python collect.py

然后将它们分块并嵌入到 Chroma 中：

python ingest.py

现在提出一个问题：

python rag.py "What problem does retrieval-augmented generation solve?"

步骤 7：检查结果

查询会返回一个有依据的答案，后面跟着它所依赖的来源：

=== Answer ===
Retrieval-augmented generation addresses the fact that a language model only
knows what it was trained on. By retrieving relevant passages from an external
index at query time and adding them to the prompt, the model can answer using
current, specific information it was never trained on, which also reduces
hallucination because the response is anchored to real source text.

=== Sources ===
  - https://en.wikipedia.org/wiki/Retrieval-augmented_generation
  - https://en.wikipedia.org/wiki/Large_language_model

截图建议：一次 rag.py 查询的终端输出，显示生成的答案以及其下方的来源列表。

多尝试几个问题，以感受检索质量：

python rag.py "How does a vector database differ from a relational database?"
python rag.py "What are common limitations of large language models?"

如果答案感觉单薄，首先值得调整两个旋钮。提高 rag.py 中的 n_results 会向模型提供更多上下文，这有助于宽泛问题，但代价是提示更长。调整 ingest.py 中的 size 和 overlap 会改变文本的切分方式；更小的文本块会使精确查找更敏锐，更大的文本块会保留更多周围上下文。任何分块更改后都要重新运行 ingest.py，以便索引反映它。

整个循环没有一行抓取或代理代码。Bright Data 从网络交付了干净的 Markdown，Chroma 在本地处理嵌入和相似性搜索，本地模型生成了答案，所有内容都基于你选择收集的页面。