RAG多数据源接入：本地文件、网页抓取
RAG（检索增强生成）系统要回答得准，前提是把知识正确灌入向量库。
很多新手卡在“数据从哪来”这一步。
本文围绕本地文件、网页抓取、数据库表同步三种常见数据源，给出可直接照做的接入步骤，包含准备条件、代码思路、排错要点和最终验证，适合零基础运维和开发者快速搭建多源知识库。
接入前先确认三样东西
动手前先检查环境，避免中途翻车。
第一，确认已经具备可用的向量数据库和嵌入模型接口，比如 Chroma、Milvus 或云上向量库，保证有写入权限。
第二，准备好 Python 环境，并安装必要的依赖包，例如 langchain、pymysql、beautifulsoup4、requests 等。
第三，明确数据源的文件格式、网页结构、数据库表结构，提前清理无效字段，能省不少调试时间。
本地文件：解析和入库
本地文件是最容易接的数据源。
先建一个文件夹，把需要导入的 txt、md、pdf 放在里面。
下面用 LangChain 的目录加载器读取：
from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
loader = DirectoryLoader("./docs", glob="**/*.txt")
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(docs)
# 将 chunks 向量化并写入向量库如果遇到 PDF 乱码，建议先转成文本再导入，或者换用 PyPDFLoader 单独处理。
路径中不要出现中文和空格，免得解析报错。
网页抓取：从URL到纯文本
网页抓取适合接入公开的新闻、文档站。
先用 requests 获取页面，再用 BeautifulSoup 提取正文，最后切块写入。
示例：
import requests
from bs4 import BeautifulSoup
res = requests.get("https://example.com/article", timeout=10)
soup = BeautifulSoup(res.text, "html.parser")
content = soup.find("article").get_text(separator="\n")
# 处理内容后切块入库注意两点：一是抓取前查看站点 robots.txt，不要给对方服务器造成压力；
二是页面内容可能包含导航和脚本，需要按实际标签结构过滤。
企业内网页面还要处理登录态，可用 session 携带 Cookie。
数据库表同步：增量更新别忽视
数据库同步是知识库保持新鲜的常用方式。
先读取表数据，再按主键或时间戳去重。
以 MySQL 为例：
import pymysql
conn = pymysql.connect(host="localhost", user="root", password="xxx", database="knowledge")
cur = conn.cursor()
cur.execute("SELECT id, title, content, updated_at FROM articles WHERE updated_at > %s", (last_time,))
rows = cur.fetchall()
# 清洗、切块、向量化后写入向量库这里最重要的坑是增量同步。
如果每次全量导入，数据量大时很慢，还会产生大量重复向量。
建议在源表加 updated_at 字段，并用定时任务记录上次同步时间，只搬运新增和变更的数据。
高频踩坑点与验证结果
实际接入时，比较常见的问题有这些：读取文件时编码不对，可在打开时指定 encoding="utf-8"；
分块过大导致检索精度差，一般 chunk_size 在 200-500 字；
重复导入造成重复向量，需要先清理再写入；
数据库权限不足导致无法连接，确认账号具有 SELECT 权限。
验证方法很简单：向量库写入后，打开 RAG 的问答界面输入一条与刚导入数据相关的问题，看返回内容是否来自新数据。
也可以直接查询向量库统计条数，确认数量与源数据一致。
同步逻辑建议先在测试环境跑一遍，再用 cron 定时执行。
如果你正在做 RAG 多数据源接入，建议先按本地文件打通全流程，再逐步增加网页和数据库同步，每接入一种都单独验证。
遇到异常时，优先检查依赖版本和权限配置，再对照本文避坑点定位问题。
