爬虫转大模型实战：采集能力还在，为什么权限日志成了最大短板？

Java技术那些事儿

347人浏览 · 2026-08-25 16:21:17

Java技术那些事儿 · 2026-08-25 16:21:17 发布

聊《我用爬虫经验做了次 AI 项目，最先失效的是旧方法》之前，先说一句实在的：别急着背概念，先看它在真实项目里到底解决什么问题。

摘要

摘要：从爬虫转做大模型应用，很多人以为信息采集能力可以直接复用。但实际项目里，最先翻车的往往不是爬取环节，而是权限控制、日志追踪和可观测性。本文结合一个真实的知识库项目，拆解爬虫技能在大模型时代的延续与断裂，给出学习路线的取舍建议。

---

爬虫技能的价值：别急着丢掉老本行

文章插图 1

转大模型之前，我也走过弯路。看到招聘JD上写"熟悉RAG、向量数据库、LangChain"，第一反应是赶紧学这些新东西，爬虫经验好像不值钱了。

但实际项目做下来，发现爬虫阶段培养的能力并没有消失，只是换了个形式延续。

我做过一个企业内部知识库项目，需求是从十几个内部系统（OA、ERP、Wiki）采集文档，构建RAG语料库。整个流程分三步：

1. 数据采集：爬虫负责从各系统抓取文档，处理登录态、反爬策略、增量同步
2. 数据清洗：解析HTML、PDF、Word，提取正文，去除噪音
3. 向量化入库：切片、Embedding、存入向量数据库

前两步完全是爬虫的老本行。真正卡壳的地方在第三步之后——当模型开始返回结果时，权限控制和日志追踪成了新难题。

爬虫阶段你只需要考虑"能不能抓到"，但大模型应用要考虑"谁能看到什么"、"为什么返回这个答案"、"调用链出了什么问题"。

这是第一个思维转换：从采集思维转向服务思维。

---

数据清洗：从"能抓"到"能喂"

文章插图 2

爬虫的数据清洗目标是提取有效信息，RAG的数据清洗目标是让切片可被向量模型理解。两者有重叠，但侧重点不同。

我遇到的一个具体场景：从内部Wiki采集的技术文档，HTML结构杂乱，包含大量导航、侧边栏、广告位。爬虫阶段我只需要提取正文，但RAG阶段需要保证切片后的文本语义完整。

import re
from langchain.text_splitter import RecursiveCharacterTextSplitter

def clean_and_chunk(raw_text: str, chunk_size: int = 512, chunk_overlap: int = 64) -> list:
    # 清理HTML标签和多余空白
    cleaned = re.sub(r'<[^>]+>', '', raw_text)
    cleaned = re.sub(r'\s+', ' ', cleaned).strip()

    # 按段落切分，保持语义完整
    paragraphs = [p.strip() for p in cleaned.split('\n') if p.strip()]

    # 递归字符切分器，避免切断句子
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=['\n\n', '\n', '。', '.', ' ', '']
    )

    chunks = splitter.split_text(cleaned)
    return chunks

这段代码的输入是爬虫抓回的原始文本，输出是适合Embedding的切片。关键逻辑在于separators的优先级顺序——先尝试按段落切，不行再按句子，最后才按字符切。这是爬虫阶段不会考虑的：切片质量直接影响检索效果。

爬虫阶段的清洗目标是"提取正文"，RAG阶段的清洗目标是"保持语义"。这个转变很多人没意识到，导致向量检索效果很差。

---

CSDN资料领取方式

代码解释：关键代码的实现原理

下面对文中两段关键代码做详细拆解，帮助理解实现原理。

代码一：数据清洗与切片

输入：raw_text（含HTML标签的原始文本）、chunk_size（默认512字符）、chunk_overlap（默认64字符）

核心逻辑：

第一步用正则 re.sub(r'<[^>]+>', '', raw_text) 剥离所有HTML标签，再用 re.sub(r'\s+', ' ', cleaned) 把连续空白压缩成单个空格。这一步和爬虫阶段的清洗类似，但RAG阶段对"干净"的定义更严格——残留的换行符可能导致切片边界混乱。

第二步按 \n 分割段落，过滤空行。这里有个细节：爬虫阶段可能直接对整个文档切分，但RAG阶段先按段落分组，是为了让 RecursiveCharacterTextSplitter 在切分时优先保持段落完整。

第三步调用 LangChain 的递归切分器。separators 列表决定了切分优先级：先尝试 \n\n（段落间），再尝试 \n（行内），然后是中文句号 。、英文句号 .，最后才按空格和字符硬切。chunk_overlap=64 保证相邻切片有64字符重叠，避免关键信息被切断在边界上。

输出：list[str]，每个元素是一段语义相对完整的文本，长度不超过 chunk_size。

异常处理：函数本身没有显式 try-except，但实际使用时需要处理三种情况：空文本（返回空列表）、超长段落（会被递归切分到字符级）、特殊字符（正则已处理大部分，但极端情况下可能产生异常编码）。生产环境建议在外层加一层校验。

代码二：权限检查装饰器

输入：被装饰的查询函数、user_id、query 及额外参数

核心逻辑：

这是一个典型的装饰器模式。@wraps(func) 保留原函数的元信息（名称、文档字符串），避免调试时混淆。

权限检查分两步：先调用 get_user_permissions(user_id) 获取用户权限集合，再用 check_query_permission(query, user_permissions) 判断当前查询是否越权。这里的关键是权限检查必须发生在查询执行之前，否则可能泄露敏感信息。

日志记录贯穿全程：查询开始记 info 日志并记录时间戳，成功时记录耗时，失败时记录异常堆栈。exc_info=True 确保异常信息完整输出，这对排查问题至关重要。

输出：与原函数相同的返回值，或在权限不足时抛出 PermissionError。

异常处理：try-except 捕获所有异常，记录错误日志后重新抛出（raise），不吞掉异常。这样做的好处是调用方仍能感知错误，同时日志系统保留了完整上下文。

---

知识库构建：RAG语料生产的断点在哪

这是我踩坑最深的地方。

真实案例

公司内部技术知识库项目，爬取了5000+篇技术文档，构建了完整的RAG链路。Demo阶段效果很好，检索准确率85%，模型回答质量也不错。但上线后，问题接踵而至。

排查过程

第一天上线，用户反馈"为什么这个答案不对"。我开始排查：

1. 现象：用户问的是"如何配置Redis集群"，返回的答案是关于"Redis基础命令"
2. 验证动作：检查向量数据库，发现相关文档确实存在，相似度分数也正常
3. 进一步排查：检查日志，发现检索请求的权限token和Demo阶段不同
4. 排除结果：Demo用的是管理员权限，生产环境用的是普通用户权限。某些文档在普通用户权限下被过滤掉了，导致检索结果不完整

失败原因

| 错误类型 | 具体表现 | 排查方法 |
|---------|---------|---------|
| 业务错误 | 权限配置导致文档过滤 | 对比不同角色的检索结果 |
| 配置错误 | 向量模型与切片策略不匹配 | 检查chunk_size和模型上下文长度 |
| 环境错误 | 生产环境网络延迟导致超时 | 监控API响应时间 |

这个案例暴露了一个核心问题：Demo和生产的差距不在代码，而在工程化能力。

爬虫阶段你只需要处理"数据对不对"，但大模型应用还要处理"谁能访问"、"访问日志怎么记录"、"出了问题怎么追踪"。

我后来在项目中加了完整的权限透传和日志追踪：

from functools import wraps
import logging
import time

logger = logging.getLogger(__name__)

def with_permission_check(func):
    @wraps(func)
    def wrapper(user_id, query, **kwargs):
        # 权限检查
        user_permissions = get_user_permissions(user_id)
        if not check_query_permission(query, user_permissions):
            logger.warning(f"用户 {user_id} 权限不足: {query}")
            raise PermissionError("无权访问该知识库")

        # 记录调用日志
        logger.info(f"用户 {user_id} 查询: {query}")
        start_time = time.time()

        try:
            result = func(user_id, query, **kwargs)
            logger.info(f"查询成功，耗时: {time.time() - start_time}s")
            return result
        except Exception as e:
            logger.error(f"查询失败: {e}", exc_info=True)
            raise

    return wrapper