将网页转换为可用数据使用 Nstdata Crawl 将 URL 转换为 AI、RAG 和数据工作流的干净输出。 设置爬虫 |
Markdown
JSON
{
"title": "...", "url": "..." } 屏幕截图
|
检索增强生成系统的可信度取决于它可以检索到的证据。如果网站摄取存储了 cookie 横幅、重复导航、过时的副本以及没有源元数据的片段，更强的语言模型也无法修复缺失的来源。
本教程构建了一个小型、可检查的管道，并展示了生产系统需要更强组件的地方。它使用本地哈希向量基线，以便检索逻辑能够使用 Python 的标准库运行。在数据合同正常工作后，将该基线替换为生产嵌入模型和向量索引。
RAG 爬虫管道需要六个属性：覆盖范围、干净内容、稳定身份、有用分段、可检索向量和源基础的答案。速度很重要，但完整性和可追溯性更为重要。
每个获取页面的最低记录应包含：
| 字段 | 重要性 |
|---|---|
source_url | 允许答案引用证据 |
canonical_url | 防止 URL 参数重复 |
title 和标题 | 改善显示和语义边界 |
crawled_at | 支持新鲜度政策 |
content_hash | 检测未更改或重复内容 |
markdown | 提供具有结构的规范化文本 |
http_status | 将不可用页面与空提取分开 |
| locale/access context | 解释区域或语言差异 |
Nstdata 爬虫 可以将 URL 转换为 Markdown 和 JSON 等格式，同时处理渲染基础设施。它不会替代您的知识库政策：您还必须决定哪些路径被允许，刷新频率，以及什么 qualifies 为合格文档。
在爬取之前，请检查网站的规则和您的授权。机器人排除协议 定义了爬虫如何发现 robots.txt 指令，但机器人遵从性只是法律和合同审查的一部分。请参阅 Nstdata 的 网络爬虫法律指南 以获取更广泛的清单。
从站点地图、已知文档根目录或策划的 URL 种子开始。添加明确的允许和拒绝规则。对于文档网站，您可以允许 /docs/ 并排除登录、搜索、变更日志分页、查询参数和可下载的二进制文件。
Nstdata 的实时 API 当前在下面的路由接受经过身份验证的爬取请求。该路由已于 2026 年 9 月 3 日验证达到身份验证；没有账户密钥无法测试成功响应。确认请求字段在 当前爬取文档 中，以便在生产中使用。
curl --request POST \ --url 'https://api.nstdata.io/api/v1/crawl/scrape?async=true' \ --header 'Authorization: Bearer YOUR_NSTDATA_API_KEY' \ --header 'Content-Type: application/json' \ --data '{ "url": "https://example.com/docs/getting-started", "formats": ["markdown"] }'
在规范化之前存储原始响应。原始保留使解析器升级可重现，并在提取规则发生变化时为您提供证据。切勿记录 API 密钥、cookie 或经过身份验证页面的个人数据。
如果您在发现方法之间进行选择，请阅读 爬取与抓取。抓取一个已知的 URL 适合有针对性的刷新；当链接发现是工作的一部分时，爬虫是合适的。
规范化应去除重复导航、页脚文本、cookie 通知、不可见控件和空标题，而不扁平化有意义的文档结构。Markdown 有助于，因为标题、列表、代码和链接以紧凑的形式保留。CommonMark 提供了一种一致的 Markdown 解析的有用基线。
使用两个身份：
不要盲目丢弃每个近重复项。产品文档可能会重复共享警告，但包含不同的程序。确切的哈希对于精确重复是安全的；模糊相似性应产生可审查的候选项。
新鲜度属于同一合同中。存储 crawled_at、源修改提示（如果可用）、和提取器版本。在刷新时，仅重新嵌入已更改的块并删除被移除页面的向量。
一个块应足够大以回答一个可能的问题，同时又要足够小以便精确检索。开始时使用头部感知的片段，然后按段落或句子拆分长部分。为每个子块附加面包屑标题。
一个实际的起始策略是：
没有普遍最佳的块大小。根据实际问题进行评估。如果答案需要跨越长程序散布的事实，则检索相邻块或使用父子检索，而不是将每个块做得巨大。
以下脚本使用 Python 的标准库实现完整的本地机制。它使用可确定的哈希词袋向量——而不是语义生产嵌入。这个限制是故意的：你可以在本地运行数据流，检查 SQLite 记录，并在以后仅替换 embed()。
import hashlib import json import math import re import sqlite3 from datetime import datetime, timezone DIMENSIONS = 256 def normalize(text: str) -> str: text = re.sub(r"\r\n?", "\n", text) text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\n{3,}", "\n\n", text) return text.strip() def chunk_markdown(markdown: str, max_words: int = 90): chunks, heading, buffer = [], "", [] for line in normalize(markdown).splitlines(): if line.startswith("#"): if buffer: chunks.append((heading, "\n".join(buffer))) buffer = [] heading = line.lstrip("# ") else: buffer.append(line) if len(" ".join(buffer).split()) >= max_words: chunks.append((heading, "\n".join(buffer))) buffer = [] if buffer: chunks.append((heading, "\n".join(buffer))) return [(h, t.strip()) for h, t in chunks if t.strip()] def embed(text: str): vector = [0.0] * DIMENSIONS for token in re.findall(r"[a-z0-9]+", text.lower()): slot = int(hashlib.sha256(token.encode()).hexdigest()[:8], 16) % DIMENSIONS vector[slot] += 1.0 length = math.sqrt(sum(x * x for x in vector)) or 1.0 return [x / length for x in vector] def cosine(a, b): return sum(x * y for x, y in zip(a, b)) def ingest(db, url, title, markdown): cleaned = normalize(markdown) page_hash = hashlib.sha256(cleaned.encode()).hexdigest() crawled_at = datetime.now(timezone.utc).isoformat() db.execute("DELETE FROM chunks WHERE source_url = ?", (url,)) for index, (heading, text) in enumerate(chunk_markdown(cleaned)): embedding_text = f"{title}\n{heading}\n{text}" db.execute( "INSERT INTO chunks VALUES (?, ?, ?, ?, ?, ?, ?)", (url, title, heading, index, text, json.dumps(embed(embedding_text)), f"{page_hash}:{crawled_at}"), ) db.commit() def search(db, question, limit=3): query_vector = embed(question) rows = db.execute( "SELECT source_url, title, heading, body, vector FROM chunks" ).fetchall() ranked = [ (cosine(query_vector, json.loads(vector)), url, title, heading, body) for url, title, heading, body, vector in rows ] return sorted(ranked, reverse=True)[:limit] db = sqlite3.connect(":memory:") db.execute("""CREATE TABLE chunks ( source_url TEXT, title TEXT, heading TEXT, chunk_index INTEGER, body TEXT, vector TEXT, version TEXT )""") sample = """# Acme Docs ## Authentication Send an API key in the Authorization header. Never expose the key in client code. ## Retries Retry rate limits with exponential backoff and jitter. Do not retry invalid credentials. """ ingest(db, "https://example.com/docs", "Acme Docs", sample) for score, url, title, heading, body in search(db, "How should I handle rate limits?"): print(f"{score:.3f}\t{heading}\t{url}\t{body}")
该脚本在本地使用 Python 3 执行，并使用包含的说明文件。它为速率限制问题排名第一的“重试”部分。这确认了块存储和检索的电路；它不验证在真实语料库上的语义质量。 对于生产环境，将本地向量函数替换为嵌入 API，并用向量索引替换线性扫描。 pgvector 向 PostgreSQL 添加了精确和近似的向量搜索；管理型向量数据库是另一个选择。无论存储引擎如何，都应保留相同的元数据。
将最佳片段传递给答案模型，并附上来源 URL 和严格的指示：根据提供的上下文回答，引用声明，并说明证据不足时。不要让模型在缺少网站内容的情况下默默用一般知识替代。
一个强大的回答阶段应：
重新排序可以在初始向量检索后提高精确度。混合检索——语义向量加关键词评分——对于错误代码、产品名称和精确的 API 参数特别有用。
Nstdata 的 网络搜索 MCP 服务器指南 提供了将实时网络证据连接到 AI 代理的相关上下文。固定知识库和实时搜索解决不同的问题：前者是可控且快速的，而后者可以发现更新的页面。
从真实的支持工单、文档标题和已知故障案例中构建一个问题集。对于每个问题，标注预期的源页面以及语料库是否包含答案。
至少测量：
按照该顺序调试。如果正确的页面从未被爬取，调整嵌入将是徒劳。如果检索到了正确的片段但答案忽略了它，请更改提示或模型阶段。
最常见的错误是嵌入原始 HTML。它填充索引中的菜单、脚本文本和重复的模板内容。其他代价高昂的错误包括在去重之前分块、丢失标题路径、遗漏规范 URL、刷新整个语料库而不是改动的页面，以及允许已删除的内容保持可搜索性。
安全性同样重要。除非检索强制执行源授权模型，否则请勿将私人页面爬取到共享索引中。将抓取的文本视为不受信任的输入：它可能包含针对下游代理的提示注入。保持系统指令的独立，并限制答复时工具可以执行的操作。
在承诺之前进行更广泛的供应商比较，可以使用 Nstdata 的 最佳网络抓取 API 指南。然后在最难处理的页面上测试 Nstdata Crawl，而不仅仅是静态主页。
在发布之前，确认每个片段都有一个规范源、标题路径、爬取时间戳、版本和内容哈希。验证删除传播、访问过滤器、重试限制、可观察性和文档刷新政策。保持一个小的黄金问题集在 CI 中，以便解析器或模型的更改不能默默降低检索质量。
爬虫是第一个组件，但它为所有下游组件设定了上限。干净、版本化的 Markdown 加上明确的质量门槛，为嵌入和答案阶段提供了可实际使用的证据。
Q: 什么是 RAG 的爬虫？
RAG 的爬虫是发现网站页面、提取干净内容和元数据、分块并嵌入以及存储以供检索增强生成的过程。
Q: 我应该为 RAG 存储 HTML 还是 Markdown？
Markdown 通常更容易进行分块和审计，因为它移除了许多页面的装饰，同时保留了标题、列表、链接、表格和代码。当重新处理或合规要求时，请单独保留原始 HTML。
Q: RAG 知识库应该多久更新一次？
更新频率应跟随源的波动性和业务风险。文档网站可能需要事件驱动或每日更新；稳定的档案可能需要更少。使用内容哈希来重新处理已更改的页面，并及时删除已删除的内容。
Q: 该示例使用生产嵌入吗？
不。它使用确定性本地哈希向量来演示该管道，而没有依赖。请在生产使用前用语义嵌入模型和向量索引替换它。
110M+ 真实 IP，访问成功率 99.9%
高并发任务平均响应约 0.5 秒
仅 $0.1/GB 起
