如何从实时网络数据构建新的向量数据库管道

TL;DR:

• 向量数据库并不能使网络数据保持更新。 新鲜度来自一个循环，该循环渲染源，清理它，生成特征指纹，更新更改的部分，并删除过时的部分。
• 源记录比嵌入本身更重要。 每个部分都需要一个稳定的ID、源URL、标题、位置、内容指纹和集合上下文。
• 区块ID应具有确定性。 从源、位置和标准化内容中推导，使未更改的部分在观察中保持其身份。
• Upsert必须与过时部分的删除配对。 否则，已删除的段落在源更改后仍可搜索。
• 检索评估需要新鲜度检查和相关性检查。 在判断搜索质量之前，测量当前源修订是否已被索引。
• Scrapeless Scraping Browser 提供渲染的实时网页输入。 Chroma 存储和搜索标准化向量；这两个产品并不互相替代。
• 免费开始。 新的 Scrapeless 帐户包括免费 Scraping Browser 运行时 - 在 app.scrapeless.com 注册。

引言：向量索引是一个快照

向量数据库回答关于当前存储记录的问题。它不知道源页面在上次提取运行后发生了更改、消失、移动或渲染了不同的区域变体。

这种区别是新鲜向量数据库管道的基础。网页层观察当前页面。确定性转换清理并拆分它。嵌入函数将块映射到向量中。向量数据库进行更新插入新的修订并移除不再属于页面的记录。评估检查检索相关性和源的新鲜度。

本教程使用 Scrapeless Scraping Browser 作为渲染页面的边界，并使用 Chroma 1.5.9 进行本地向量存储。无需凭证的验证捕获了一个通过HTTP公开的页面，然后执行干净 → 块 → 指纹 → 更新插入 → 查询路径针对 Chroma。Scrapeless 云渲染步骤仍然是一个明确的API密钥前提。

一览管道

一个新鲜的向量数据库网络数据管道将观察与索引分开。

每个阶段独立存储证据。一个糟糕的检索可以追溯到渲染、提取、块、嵌入、索引或查询层。

向量数据库的功能与局限

向量数据库存储向量和相关记录，然后根据索引和距离政策搜索附近的向量。Chroma 可以在集合中存储嵌入、文档和元数据，并一起查询这些记录。 Chroma 概述描述了集合和检索表面。

该数据库不发现URL，执行JavaScript，识别主要文章，选择法律集合范围，或判断源修订是否为当前。那些职责留在数据提取管道中。

以下分离保持了所有权的明确性：

• Scrapeless Scraping Browser 渲染批准的页面并保留会话/区域上下文。
• 清理代码删除了导航、重复的chrome、脚本和空文本。
• 块代码定义了检索单元。
• 嵌入函数定义了向量空间。
• Chroma 存储当前块并返回最近的记录。
• 源清单决定哪些之前的记录是过时的。

先决条件

• Python 3.12。
• Chroma 1.5.9、HTTPX 0.28.1 和 Beautiful Soup 4.14.3 用于执行本地验证。
• Node.js 和 scrapeless-scraping-browser CLI 用于渲染云输入。
• 一个 Scrapeless 账户和API密钥。
• 一个或多个具有文档化的提取目的和保留政策的公共源。

安装向量和清理依赖项：

python -m pip install chromadb==1.5.9 httpx==0.28.1 beautifulsoup4==4.14.3

阶段 1 — 渲染实时网页源

渲染阶段返回一个紧凑的移交：最终URL、标题和可见的主要文本。请勿将原始屏幕截图或未经检查的页面外壳传递到向量管道中。

注意：以下块需要您的 Scrapeless API 密钥。无凭证的验证无法创建云浏览器会话，因此未呈现任何云输出作为已完成的运行。

SESSION=$(scrapeless-scraping-browser new-session \
  --name vector-ingest --ttl 300 --proxy-country US --json \
  | jq -r '.taskId')

scrapeless-scraping-browser --session-id "$SESSION" open \
  "https://www.iana.org/help/example-domains"
scrapeless-scraping-browser --session-id "$SESSION" wait 3000
scrapeless-scraping-browser --session-id "$SESSION" eval '
JSON.stringify({
  url: location.href,
  title: document.title,
  text: document.body.innerText
})' > rendered-page.json

scrapeless-scraping-browser stop "$SESSION"

规则：

1. 仅输出翻译文本 — 不解释，不添加额外代码围栏。
2. 精确保留Markdown/HTML结构（标题、列表、链接、表格）。
3. 保持任何占位符标记如 bash python -m pip install chromadb==1.5.9 httpx==0.28.1 beautifulsoup4==4.14.3 或 1.5.9 完全不变；永远不翻译、重排序、合并或重新格式化它们。
4. 不要添加或删除 ``` 代码围栏，也不要将普通文本包装到代码块中。

在嵌入之前验证交接。拒绝意外的登录页面、访问消息、空主体或软404。仅成功的HTTP状态并不能证明意图内容被渲染。

Scraping Browser快速入门 解释了会话设置。Scraping Browser产品页面 和 定价 涉及托管浏览器层。

阶段 2 — 清理并在稳定边界上分块

清理应保留含义，同时让无关的变化消失。合并重复的空白，删除交接前已知的导航/页脚区域，并将最终的URL和标题与文本并列保留。

执行的示例使用80词的窗口和15词的重叠，因为公共测试页面较短。这些值是演示输入，而不是普遍建议。生产分块大小应根据目标语料库、嵌入模型、查询集和评估结果进行选择。

分块位置很重要。如果每个小编辑都移动每个后续分块，确定的ID将在页面中变化。优先选择语义边界，如标题和段落，当源提供它们时，然后在每个部分内应用一个有限大小的策略。

阶段 3 — 以明确的元数据嵌入

每个向量记录需要足够的元数据来解释其来源和修订。

下面的示例使用64坐标的确定性术语哈希向量来保持本地运行的自包含性。它证明了向量数据库的底层构造，而不是语义模型的质量。在生产中用评估过的嵌入模型替换它，并在向量空间变化时重新索引集合。

SHA-256 生成源和记录指纹。FIPS 180-4 指定了安全哈希算法。内容指纹检测变更；这不是授权或信任决策。

使用Scrapeless开始抓取

使用Scrapeless增强您的网页抓取和自动化工作流程！
今天注册并获得 $5的免费信用 — 无需信用卡。

现在在 Scrapeless仪表板 领取您的免费信用。

阶段 4 — 更新当前分块并删除过时的分块

以下程序读取 rendered-page.json，创建确定的分块ID，将其更新到一个持久的Chroma集合中，删除不再存在的先前ID，并运行查询。

import hashlib
import json
import math
import os
import re
from pathlib import Path

import chromadb

VECTOR_SIZE = 64
CHUNK_WORDS = 80
OVERLAP_WORDS = 15


def hash_embedding(text: str) -> list[float]:
    vector = [0.0] * VECTOR_SIZE
    for token in re.findall(r"[a-z0-9]+", text.lower()):
        digest = hashlib.sha256(token.encode()).digest()
        bucket = int.from_bytes(digest[:2], "big") % VECTOR_SIZE
        vector[bucket] += 1.0 if digest[2] % 2 == 0 else -1.0
    length = math.sqrt(sum(value * value for value in vector)) or 1.0
    return [value / length for value in vector]


def make_chunks(text: str) -> list[str]:
    words = text.split()
    step = CHUNK_WORDS - OVERLAP_WORDS
    return [" ".join(words[start:start + CHUNK_WORDS]) for start in range(0, len(words), step)]


input_path = Path(os.environ.get("RENDERED_PAGE_PATH", "rendered-page.json"))
database_path = os.environ.get("CHROMA_PATH", "chroma-data")
page = json.loads(input_path.read_text(encoding="utf-8"))
clean_text = " ".join(page["text"].split())
chunks = make_chunks(clean_text)
fingerprint = hashlib.sha256(clean_text.encode()).hexdigest()

ids = [
    hashlib.sha256(f'{page["url"]}:{position}:{chunk}'.encode()).hexdigest()[:24]
    for position, chunk in enumerate(chunks)
]
metadata = [
    {
        "source_url": page["url"],
        "title": page["title"],
        "position": position,
        "content_sha256": fingerprint,
    }
    for position in range(len(chunks))
]

client = chromadb.PersistentClient(path=database_path)
collection = client.get_or_create_collection(
    name="live_web_pages",
    metadata={"hnsw:space": "cosine"},
)

previous = collection.get(where={"source_url": page["url"]})
stale_ids = sorted(set(previous["ids"]) - set(ids))
if stale_ids:
    collection.delete(ids=stale_ids)

collection.upsert(
    ids=ids,
    documents=chunks,
    metadatas=metadata,
    embeddings=[hash_embedding(chunk) for chunk in chunks],
)

query = collection.query(
    query_embeddings=[hash_embedding("example domains documentation")],
    n_results=min(2, len(ids)),
    include=["documents", "metadatas", "distances"],
)

print(json.dumps({
    "chromadb_version": chromadb.__version__,
    "vector_size": VECTOR_SIZE,
    "chunk_count": len(chunks),
    "stored_count": collection.count(),
    "stale_deleted": len(stale_ids),
    "fingerprint_prefix": fingerprint[:12],
    "top_ids": query["ids"][0],
}, indent=2))

实时公共页面运行生成了以下输出：

{
  "chromadb_version": "1.5.9",
  "vector_size": 64,
  "chunk_count": 2,
  "stored_count": 2,
  "stale_deleted": 0,
  "fingerprint_prefix": "9ed322155bab",
  "top_ids": [
    "8289a31c06c87c9e1abac29d",
    "3ee123fc6659b0c9168231ff"
  ]
}

第二次执行返回相同的指纹、ID 和存储计数。这是期望的未更改源行为。

阶段 5 — 在重新嵌入之前检测变化

在分块之前计算规范化的源指纹。如果指纹和集合上下文与先前成功的观察匹配，则向量阶段可以停止而不更改记录。

当指纹变化时，首先建立新的分块集。更新当前ID，然后删除先前ID与当前ID之间的差异。保持旧源清单，直到新的写入和协调完成，以便失败的转变不会抹去最后已知的索引状态。

不要仅依赖时间戳。页面可以返回相同的时间戳，但内容不同，或者新的时间戳没有实际文本变化。基于内容的ID使比较确定。

阶段 6 — 添加混合搜索而不丧失来源

稠密向量捕获由嵌入模型定义的关系，而词汇匹配则有助于精确标识符、产品代码和稀有名称。混合检索层可以结合两者，但每个结果仍应返回源URL、页面标题、分块位置和内容指纹。
来源描述了一个实体是如何生成的以及哪个活动产生了它。 W3C PROV-O 词汇表为当管道需要可互操作的血统时提供了实体、活动和代理的正式模型。

适用于 RAG 管道的清洁网络文本更深入地探讨了向量存储之前的获取、提取和分块边界。

评估检索质量和时效性

使用两个独立的问题集评估管道。

时效性测试询问最新的批准源修订是否存在，移除的段落是否缺失，区域和页面状态是否符合政策，以及每个结果是否指向当前指纹。

检索测试询问相关片段是否出现于代表性问题中，确切的标识符是否仍可找到，无关的片段是否低于接受阈值，以及引用是否解析为展示给模型的证据。

BEIR 检索基准论文展示了为什么检索质量在数据集和任务之间会有所不同。使用从您实际语料库中提取的查询集，而不是将一个通用评分视为通用。

成本和操作清单

• 在收集之前注册源所有者、目的、区域、频率和保留时间。
• 在嵌入之前拒绝意外的页面状态。
• 使用每个记录存储源、块、嵌入和适配器版本。
• 按照指纹跳过未更改的标准化内容。
• 在成功的更新插入后删除陈旧的块 ID。
• 当嵌入模型或向量维度发生变化时重新索引。
• 除非所有者批准其他限制，否则每个目标主机不超过三个工作线程。
• 单独衡量渲染成功率、清洁文本产出、块变更、索引年龄、查询相关性和引用有效性。

结论：时效性是一个调和循环

一个新鲜的向量数据库管道是一个调和系统，而非一次性导入。渲染已批准源，验证页面状态，标准化内容，创建确定性块，在版本化模型下嵌入它们，更新当前记录，并移除陈旧的 ID。

Scrapeless Scraping Browser 拥有渲染的网页观察。Chroma 拥有向量存储和搜索。它们之间的清单证明了索引所代表的源修订。

准备构建一个新鲜的网络知识管道吗？

加入我们的社区以索取免费计划并与构建可追踪的 RAG 吞吐量的开发者联系：Discord · Telegram。

在 app.scrapeless.com 注册以获取免费的 Scraping Browser 运行时，并将渲染页面的交接连接到您评估过的向量存储。

常见问题

问：向量数据库是否会自动保持网络数据的新鲜？

不会。向量数据库存储它接收到的记录。您的管道必须观察源，比较修订，更新更改的块，并删除陈旧的块。

问：网络数据管道应该使用哪个向量数据库？

选择适合您部署、元数据过滤、索引、耐用性、访问控制和操作要求的数据库。本教程使用 Chroma 作为本地可执行示例，而不是作为通用排名。

问：实时网络向量管道是否需要代理？

动态或区域依赖的源通常需要稳定的浏览器出口。锁定批准的国家，以便连续观察指向相同的区域页面状态。

问：当源 DOM 变化时会发生什么？

在索引之前重新检查渲染和提取适配器。拒绝意外的页面状态，而不是嵌入访问消息、空壳或仅包含导航的页面。

问：收集器应该使用多少并发？

除非网站所有者批准其他限制，否则每个目标主机不超过三个工作线程。向量索引可以与源收集独立扩展。

问：抓取公共网络内容以供向量数据库使用是否合法？

公共可用性并不能解决所有法律问题。审核源条款、robots 指导、版权、隐私、保留以及预期的下游用途；具体项目请咨询法律顾问。

问：这个管道可以在没有 AI 代理的情况下运行吗？

可以。渲染器、确定性转换、嵌入功能、Chroma 客户端和评估套件可以作为定时软件运行，而无需 AI 代理。