使用 LangChain 构建 AI 数据管道：以 Crawlbase 作为数据来源

一个 AI 数据管道的好坏只取决于你喂给它的文本，而开放网络是你能用来给模型打底的、最丰富的新鲜且特定领域知识来源。问题在于把那些文本弄成可用的形态：大多数页面是导航、广告和 JavaScript 渲染内容的一团乱麻，而普通的 HTTP 抓取永远看不到这些。本指南将向你展示如何用 LangChain 和 Crawlbase 构建一个 AI 数据管道，把 Crawling API 用作你的文档来源，让页面以干净的 markdown 形式抵达，然后用检索增强生成（RAG）对它们进行分割、嵌入和查询。

这个管道的形态很简单，并在 Python 中端到端运行：Crawlbase 获取并清理页面，LangChain 把它分割成块并嵌入到一个向量存储中，而在查询时你检索最相关的块并把它们作为上下文交给一个 LLM。Crawlbase 处理代理轮换、封锁和渲染，于是你的管道代码保持专注于数据，而不是与反机器人防御搏斗。下面的一切都可运行；换上你自己的 URL 和 token，你就拥有了一个在实时网络内容之上工作的 RAG 系统。

为什么用 Crawlbase 作为 LangChain 的文档来源

LangChain 自带用于文件、数据库和少数几个网络来源的文档加载器，但大规模加载真实网页正是大多数管道卡住的地方。对一个现代站点的裸请求要么返回一个没有内容的 JavaScript 空壳，要么返回一个封锁页，而即便你拿回了 HTML，它也满是污染你嵌入的样板内容。垃圾块意味着垃圾检索，这意味着一个会自信地引用 cookie 横幅的 LLM。

Crawling API 干净利落地解决了获取层。你把一个 URL 发给它，它在一个可信的住宅 IP 背后渲染页面，并且它能把内容以干净的 markdown 形式而非原始 HTML 交回给你。那份 markdown 恰恰是你想要的 LangChain 文档：可读的散文，保留了标题，并把导航、脚本和广告标记剥离掉。把预先清理好的 markdown 喂进你的分割器，是一个基于网络的 RAG 管道里最大的单一质量杠杆，而这与LLM 就绪的 markdown 网络抓取中探讨的是同一个想法。

这种关注点分离正是让管道保持可维护的原因。Crawlbase 拥有网络访问：轮换 IP、解决 CAPTCHA、渲染 JavaScript 以及返回结构化输出。LangChain 拥有编排：分块、嵌入、检索，以及给答案搭框架的提示词。模型拥有推理。你可以改变你如何分块或查询哪个模型，而无需触碰数据是如何被获取的，反过来也成立。

架构：从 URL 到有据可依的答案

这个管道有四个阶段，每个都只有一项工作。获取：Crawling API 获取每个 URL 并返回干净的 markdown。分割：LangChain 的文本分割器把每个文档拆成有重叠的块，小到足以精确地嵌入和检索。嵌入并存储：每个块被变成一个向量并写入一个向量存储（我们在本地使用 Chroma）。检索并生成：在查询时你嵌入问题，拉取最接近的块，并把它们作为打底上下文传给一个 LLM。

前三个阶段是一个离线摄取作业，你在你的来源变化时运行它。第四个在每次用户提问时运行。把摄取和查询分开正是让管道可扩展的原因：你爬取并嵌入一次，然后针对存储的向量廉价地回答许多问题。更广泛的模式，包括为什么清理在你嵌入之前就很要紧，在如何为 AI 和 ML 构建并清理网络抓取数据中有所讲解。

搭建项目

你需要 Python 3.10 或更新版本。创建一个虚拟环境并安装这些库：用于获取的官方 Crawlbase 客户端、用于编排的 LangChain 包、用于向量存储的 Chroma，以及用于嵌入和聊天模型的 OpenAI 集成。

python -m venv .venv
source .venv/bin/activate

pip install crawlbase langchain langchain-community langchain-openai langchain-chroma

你还需要两个凭据：来自你仪表板的一个 Crawlbase token，以及一个嵌入/LLM 提供商密钥（这里是一个 OpenAI 密钥）。crawlbase 包给你 CrawlingAPI 客户端；langchain-chroma 封装本地的 Chroma 存储；langchain-openai 同时提供嵌入和聊天模型。把你的密钥导出为环境变量，这样代码里就不会存有任何敏感信息。

export CRAWLBASE_TOKEN="your_crawlbase_token"
export OPENAI_API_KEY="your_openai_key"

第 1 步：用 Crawling API 获取干净的 markdown

从获取开始。官方客户端暴露一个 get 方法，它接收一个 URL 和一些选项；传入 format=markdown 会在响应 body 里返回干净的 markdown 形式的页面。把它包进一个小函数里，把每个获取到的页面变成一个 LangChain Document，在元数据里携带来源 URL，这样你之后就能引用它。

import os
from crawlbase import CrawlingAPI
from langchain_core.documents import Document

api = CrawlingAPI({"token": os.environ["CRAWLBASE_TOKEN"]})

def load_page(url):
    # format=markdown returns clean markdown, not raw HTML
    response = api.get(url, {"format": "markdown"})
    if response["status_code"] != 200:
        raise RuntimeError(f"Fetch failed for {url}: {response['status_code']}")
    body = response["body"]
    text = body.decode("utf-8") if isinstance(body, bytes) else body
    return Document(page_content=text, metadata={"source": url})

urls = [
    "https://example.com/docs/getting-started",
    "https://example.com/docs/pricing",
]
docs = [load_page(u) for u in urls]
print(f"Loaded {len(docs)} documents")

对于重 JavaScript 的页面，在选项字典里加上 "ajax_wait": "true" 和一个以毫秒为单位的 "page_wait"，并使用一个 JavaScript token。因为获取被隔离在 load_page 里，把那些选项换进去不会触碰任何下游阶段。如果一个站点以一个非 200 的状态响应，这个函数会带着那个码抛出，于是一个坏来源会响亮地浮现，而不是用一个错误页毒化你的存储。

第 2 步：把文档分割成块

整页太大，无法有用地嵌入：一个长文档对应单个向量会把不同的主题模糊地搅在一起，并损害检索精度。改为把每个文档分割成有重叠的块。LangChain 的 RecursiveCharacterTextSplitter 会先尝试在段落和句子边界处断开，于是块保持连贯，而且因为来自 Crawlbase 的 markdown 保留了标题和列表，那些分割会落在自然的接缝上。

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=150,
)

chunks = splitter.split_documents(docs)
print(f"Split into {len(chunks)} chunks")

对于散文，约 1000 个字符的 chunk_size 加上 150 个字符的重叠是一个明智的默认。重叠把一点上下文带过边界，这样一个被分割到两个块里的事实就不会丢失。两者都按你的内容来调：更密集、更技术性的页面常常用更小的块检索得更好，而长篇文章则能容忍更大的块。来自 load_page 的元数据会被自动复制到每个块上，因此每一个仍然知道它的来源 URL。

第 3 步：嵌入并存入一个向量数据库

现在把每个块变成一个向量并持久化它。一个嵌入模型把文本映射到高维空间里的一个点，语义相似的段落在那里彼此挨得很近，而这正是让按含义检索成为可能的东西。Chroma 在本地存储那些向量并处理相似度搜索；传入 persist_directory 会把索引写到磁盘，于是你只需付一次嵌入成本。

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vector_store = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
)
print(f"Stored {len(chunks)} vectors")

这一块是摄取的结尾。在你的来源变化时运行它一次，而不是在每次查询时。要在之后复用这个存储，用 Chroma(persist_directory="./chroma_db", embedding_function=embeddings) 重新打开它，而不是从文档重新构建。Chroma 对本地开发很方便；当你超出单台机器的承载时，同样的 LangChain 接口前置于像 Pinecone 或 pgvector 这样的托管存储，于是你其余的代码不变。

第 4 步：检索并生成答案

向量就位后，查询路径就很短了。把存储变成一个检索器，嵌入用户的问题，拉取最接近的块，并把它们传给一个聊天模型，附带一个提示词，告诉它只从提供的上下文里回答。LangChain 的表达式语言把这些接成一条链。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

retriever = vector_store.as_retriever(search_kwargs={"k": 4})
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_template(
    "Answer using only the context below.\n\n"
    "Context:\n{context}\n\nQuestion: {question}"
)

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = chain.invoke("What does the getting-started guide say about setup?")
print(answer)

设置 k=4 检索四个最相关的块；对宽泛的问题把它调高，对紧凑的问题把它调低。temperature 为 0 让模型锚定在检索到的文本上，而不是即兴发挥。因为提示词把答案限制在提供的上下文里，响应保持立足于你实际爬取到的内容，而且由于每个块都携带它的 source 元数据，你可以通过用 retriever.invoke(question) 直接检视检索到的文档来呈现引用。

运行整条管道

把这四个步骤按顺序放进一个脚本里，你就有了一条完整的管道：对你的 URL 调用 load_page，分割，嵌入进 Chroma，然后构建链并调用它。第一次运行会爬取并嵌入，这要花一会儿；后续重新打开持久化存储的运行会在远不到一秒内回答，因为昂贵的工作已经做完了。往列表里加更多 URL 并重新运行摄取，以拓宽系统所知道的范围。

从这里开始，同样的结构会自然地延伸。给摄取作业排程，让它按一个节奏刷新来源，把它指向站点地图以爬取整个版块，或者随着你的语料库增长把 Chroma 换成一个托管的向量存储。对于大批量爬取，你可以把获取移到异步的 Crawling API 上，或者通过 Web MCP 从一个智能体来驱动它，并在你需要在自己的获取器前面做 IP 轮换时，把一切路由通过 Smart AI Proxy。管道契约不变：干净的文本进去，有据可依的答案出来。关于这一点在提取那一侧的更多内容，参见AI 数据提取是如何工作的。

常见问题

为什么用 Crawlbase 而不是一个内置的 LangChain 网络加载器？

内置加载器假定一个页面在一个普通请求下会返回可用的 HTML，而现代站点很少这样做：它们在浏览器中渲染内容并封锁自动化流量。Crawling API 在一个轮换的住宅 IP 背后渲染页面并返回干净的 markdown，于是你的文档是真实内容，而不是空壳或封锁页。那种干净直接提升了块的质量和检索准确度。

对于一个 RAG 管道，我该请求 HTML 还是 markdown？

Markdown。传入 format=markdown，这样页面就以可读的散文形式回来，保留了标题和列表，并把导航、脚本和广告标记剥离掉。那些结构线索帮助分割器在自然边界处断开，而移除样板内容能把低价值文本挡在你的向量存储之外。只有当你需要用选择器解析特定元素而不是嵌入页面时，才请求 HTML。

我如何处理重 JavaScript 的页面？

使用一个 JavaScript token，并在你传给 api.get 的选项里加上 ajax_wait 和 page_wait。Crawling API 随后会在一个真实浏览器中渲染页面，等待异步内容，并返回完成的 markdown。因为获取被隔离在 load_page 函数里，启用渲染不会影响下游的分割、嵌入或检索。

我该用什么样的块大小和重叠？

对于一般散文，从每块大约 1000 个字符、150 个字符的重叠开始。更小的块在密集的技术内容上提升精度；更大的块适合上下文跨越多个段落的长篇文章。重叠把一点上下文带过边界，这样一个被分割到两个块之间的事实仍然可被检索到。把这些当作默认值，并针对你自己的检索结果来调。

我必须用 OpenAI 来做嵌入和 LLM 吗？

不必。这条管道在设计上与提供商无关。把 OpenAIEmbeddings 和 ChatOpenAI 换成任何 LangChain 支持的嵌入模型和聊天模型，包括本地的，而分割、存储和检索的代码保持不变。Crawlbase 完全处在获取那一侧，因此你对模型的选择从不影响数据是如何被获取的。

我如何让知识库保持新鲜？

针对那些会变化的 URL，按一个时间表重新运行摄取阶段（获取、分割、嵌入），并在两次之间重新打开持久化存储用于查询。对于大型或频繁更新的语料库，把爬取指向站点地图，并把获取移到异步的 Scraper API 上，这样你就能在不阻塞你的应用的情况下摄取许多页面。

IK

Ian Kalvin

技术支持工程师 · Crawlbase

Crawlbase 技术支持工程师，从生产环境抓取与代理配置中真正出问题的第一线写作。

NZ

Neil Zamora

高级架构师 · Crawlbase

Crawlbase 高级架构师，专注大规模爬取背后的系统：代理轮换、反机器人韧性，以及隐藏这些复杂性的 API。