当前位置: 首页 > news >正文

基于Playwright与向量化技术构建AI知识库:从网页采集到RAG应用实战

1. 项目概述:从零构建一个“会思考”的AI知识库

最近在折腾一个挺有意思的项目:想给我的团队搞一个垂直领域的AI知识库。这玩意儿不是简单的文档堆砌,而是希望它能“理解”我们行业内的专业内容,比如最新的技术博客、竞品动态、行业报告,然后能像专家一样回答我们的问题。想法很美好,但第一步就卡住了——数据从哪来?总不能靠人工一篇篇去复制粘贴吧?那效率太低了,而且很多信息源是动态更新的网页。

于是,我盯上了网页内容自动化采集。市面上工具很多,从简单的requests+BeautifulSoup,到无头浏览器PuppeteerSelenium。经过一番折腾,我最终选择了Playwright。理由很简单:它够新、够快、够稳。微软出品,原生支持异步,对现代单页应用(SPA)的渲染支持近乎完美,而且跨浏览器(Chromium, Firefox, WebKit)的特性让它在不同场景下的兼容性更有保障。最关键的是,它的API设计非常人性化,写起采集脚本来有种行云流水的感觉。

这个项目,就是围绕Playwright,打造一套从网页内容抓取、清洗、结构化到最终灌入向量数据库,为大模型提供“养料”的完整流水线。它解决的不仅仅是“把网页存下来”的问题,更是如何高效、准确、自动化地为AI知识库准备高质量数据源的问题。无论你是想构建个人学习笔记库、企业内部的FAQ系统,还是垂直领域的智能问答助手,这套实战经验都能给你提供一条清晰的路径。

2. 核心思路与架构设计:为什么是Playwright+向量化?

构建AI知识库,尤其是面向大模型(RAG,检索增强生成)的知识库,核心矛盾在于:如何将海量、非结构化的网页信息,转化为大模型能够高效“消化”和“回忆”的结构化知识片段。整个过程可以拆解为四个核心环节:采集 -> 解析 -> 处理 -> 存储与检索。我的技术选型正是围绕这四个环节展开的。

2.1 采集层:Playwright的压倒性优势

为什么不用更轻量的requests?因为现代网页太“狡猾”了。大量内容通过JavaScript动态加载,简单的HTTP请求只能拿到一个空壳HTML。Selenium虽然老牌,但启动慢、资源占用高,在需要高并发采集时显得笨重。

Playwright的优势在这里凸显:

  1. 无头浏览器驱动:能完整执行页面JavaScript,获取渲染后的最终DOM,对付React、Vue等框架构建的SPA页面毫无压力。
  2. 自动等待机制:内置智能等待,可以等待元素出现、网络请求完成或页面加载完毕,大大减少了编写复杂等待逻辑的代码量,提升了脚本的稳定性。
  3. 强大的选择器:支持CSS、XPath、Text等多种定位方式,甚至可以通过get_by_roleget_by_label进行可访问性定位,编写采集规则更精准。
  4. 网络拦截与模拟:可以拦截和修改网络请求,这对于处理反爬机制(如验证码图片请求)或直接抓取接口数据(XHR/Fetch)提供了可能。
  5. 并发与上下文隔离:通过创建多个Browser Context,可以在一次浏览器实例中模拟多个完全独立的会话,高效且节省资源。

注意:虽然Playwright功能强大,但它毕竟启动了完整的浏览器环境,资源消耗(内存、CPU)远高于requests。因此,在目标网页是纯静态或服务端渲染(SSR)的情况下,优先考虑requests+parsel/BeautifulSoup的组合,效率会高出一个数量级。我的原则是:能静态解析的绝不启动浏览器

2.2 处理与存储层:从HTML到向量

采集到的原始HTML是“脏数据”,包含导航栏、广告、页脚等噪音。直接扔给大模型,效果会大打折扣。因此,解析与清洗至关重要。我使用BeautifulSouplxml进行HTML解析,结合自定义规则(如根据CSS类名、标签结构)提取正文内容。

更关键的一步是文本分割(Text Splitting)。一篇长文章直接嵌入成一个大向量,检索精度会很低。我们需要将其切分成有语义关联的片段(Chunks)。这里我采用了递归字符分割语义分割相结合的策略:

  • 递归字符分割:按固定长度(如500字符)分割,并设置一段重叠区(如50字符),防止句子被生生切断。
  • 语义分割:利用langchainRecursiveCharacterTextSplittersentence-transformers的语义模型,尝试在句子的自然边界处进行分割,保证chunk的语义完整性。

处理后的文本片段,通过嵌入模型(Embedding Model)转化为高维向量。我推荐使用开源模型如bge-large-zh-v1.5(中文)或all-MiniLM-L6-v2(英文),它们效果不错且可以在本地部署。这些向量最终存入向量数据库,如ChromaMilvusQdrant。向量数据库的核心能力是近似最近邻搜索(ANN),当用户提问时,将问题也转化为向量,并快速从库中找出最相关的几个文本片段,作为上下文提供给大模型生成答案。

整个架构的流程图如下:

[目标URL列表] -> [Playwright 采集器] -> [原始HTML/截图] -> [解析与清洗模块] -> [纯净文本] -> [文本分割器] -> [文本片段(Chunks)] -> [嵌入模型] -> [向量] -> [向量数据库] ↑ [用户提问] -> [嵌入模型] -> [查询向量] -> [ANN检索] -> [相关片段] -> [大模型] -> [答案]

3. Playwright采集实战:从安装到编写健壮爬虫

理论说再多,不如一行代码。我们直接进入实战环节。

3.1 环境搭建与初始化

首先,安装Playwright。建议使用Python版本。

pip install playwright # 安装Playwright自带的浏览器(Chromium, Firefox, WebKit) playwright install chromium

我通常只安装Chromium,因为它在性能和兼容性上最平衡。如果需要测试页面在不同浏览器下的表现,可以再安装其他的。

一个最基本的采集脚本骨架如下:

import asyncio from playwright.async_api import async_playwright async def main(): async with async_playwright() as p: # 启动浏览器,headless=False可以看到浏览器界面,调试时非常有用 browser = await p.chromium.launch(headless=False, slow_mo=50) # slow_mo 让动作慢下来,方便观察 # 创建一个浏览器上下文,可以模拟独立的会话(cookies, localStorage隔离) context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...' ) page = await context.new_page() try: # 导航到目标页面 await page.goto('https://example.com/article/123', wait_until='networkidle') # 等待到网络空闲 # 在这里进行内容提取操作... # ... except Exception as e: print(f"抓取页面失败: {e}") finally: await browser.close() asyncio.run(main())

3.2 核心内容提取策略与代码示例

提取内容的核心在于定位元素。Playwright提供了多种方式。

场景一:提取特定标题和正文假设我们要抓取一篇博客文章。

# 等待文章标题区域加载 await page.wait_for_selector('article h1', state='visible') # 提取标题 title = await page.text_content('article h1') # 提取正文 - 假设正文在 <article> 标签下的所有 <p> 标签里 # 使用 page.query_selector_all 获取元素列表 paragraph_elements = await page.query_selector_all('article p') # 将每个段落的文本内容合并 body_text = '\n'.join([await el.text_content() for el in paragraph_elements if await el.text_content()]) print(f"标题: {title}") print(f"正文: {body_text[:200]}...") # 打印前200字符

场景二:处理分页或“加载更多”很多网站采用分页或滚动加载。

# 方法1:点击“下一页”按钮,直到没有下一页 has_next_page = True all_articles = [] while has_next_page: # 提取当前页所有文章链接 article_links = await page.query_selector_all('.article-list a.title') for link in article_links: href = await link.get_attribute('href') all_articles.append(href) # 尝试找到并点击“下一页”按钮 next_button = await page.query_selector('a.next-page') if next_button: await next_button.click() # 等待新内容加载,通常可以等待某个新出现的元素或网络请求 await page.wait_for_load_state('networkidle') # 或者等待列表区域更新 # await page.wait_for_selector('.article-list a.title:last-child') else: has_next_page = False # 方法2:模拟滚动触发加载(针对无限滚动页面) import time previous_height = await page.evaluate('document.body.scrollHeight') while True: # 滚动到页面底部 await page.evaluate('window.scrollTo(0, document.body.scrollHeight)') # 等待新内容加载 await page.wait_for_timeout(2000) # 等待2秒,网络请求可能更可靠 # 也可以等待某个加载动画消失 # await page.wait_for_selector('.loading-spinner', state='hidden') new_height = await page.evaluate('document.body.scrollHeight') if new_height == previous_height: break # 高度不再变化,说明已加载完毕 previous_height = new_height # 滚动结束后,再提取所有内容

场景三:应对反爬与复杂交互有些网站需要登录或有复杂验证。

# 1. 处理登录(以账号密码为例) await page.goto('https://example.com/login') await page.fill('input[name="username"]', 'your_username') await page.fill('input[name="password"]', 'your_password') # 有时需要等待一下再点击,或者勾选复选框 await page.click('button[type="submit"]') # 等待登录成功后的跳转或某个登录后特有的元素出现 await page.wait_for_selector('#user-avatar', state='visible') print("登录成功") # 2. 保存登录状态(Cookies),避免每次采集都登录 # 登录成功后保存上下文状态 await context.storage_state(path='auth_state.json') # 下次启动时,可以直接加载这个状态来恢复登录会话 context2 = await browser.new_context(storage_state='auth_state.json') page2 = await context2.new_page() await page2.goto('https://example.com/dashboard') # 此时应该已是登录状态 # 3. 处理弹窗或验证码(简单情况) # 监听对话框(alert, confirm, prompt) page.on('dialog', lambda dialog: dialog.accept()) # 对于图形验证码,通常需要引入第三方OCR服务或手动处理,这超出了自动化范畴。 # 更优的策略是寻找无需验证码的接口,或使用更专业的反爬解决方案。

3.3 高级技巧:网络拦截与性能优化

为了提高采集效率和应对特殊场景,Playwright的网络拦截功能非常有用。

# 拦截并记录或修改请求/响应 async def handle_route(route): # 可以在这里修改请求头,例如添加Referer或User-Agent headers = route.request.headers headers['my-custom-header'] = 'my-value' # 继续请求 await route.continue_(headers=headers) # 或者直接返回一个模拟的响应 # await route.fulfill(status=200, body='Mocked response') await page.route('**/*', handle_route) # 拦截所有请求 # 更精确的拦截,例如只拦截图片请求以节省带宽 await page.route('**/*.{png,jpg,jpeg}', lambda route: route.abort()) # 性能优化:禁用不必要的资源加载 browser = await p.chromium.launch( headless=True, args=['--disable-images', '--disable-javascript'] # 谨慎使用,可能破坏页面功能 ) # 更好的方式是通过路由选择性拦截 async def block_media(route): if route.request.resource_type in ['image', 'media', 'font', 'stylesheet']: await route.abort() else: await route.continue_() await page.route('**/*', block_media)

4. 从原始HTML到知识库Chunk:数据处理流水线

采集到HTML只是第一步,接下来是更繁琐但至关重要的数据处理。

4.1 解析与正文提取

我使用BeautifulSoup进行解析,因为它容错性好,API友好。

from bs4 import BeautifulSoup import re def extract_clean_content(html, url): """ 从HTML中提取纯净的正文内容。 :param html: 原始HTML字符串 :param url: 页面URL,可用于特定站点规则 :return: 字典,包含标题、正文、发布时间等元数据 """ soup = BeautifulSoup(html, 'lxml') # 1. 提取标题 - 多种后备方案 title = '' # 方案1: 查找 <title> 标签 if soup.title and soup.title.string: title = soup.title.string.strip() # 方案2: 查找 Open Graph 或 Twitter 的 title 属性 og_title = soup.find('meta', property='og:title') if og_title and og_title.get('content'): title = og_title['content'].strip() # 方案3: 查找最大的标题标签 (h1) if not title: h1 = soup.find('h1') if h1: title = h1.get_text().strip() # 2. 提取正文 - 核心挑战 # 简单策略:移除所有 script, style, nav, footer, header 等标签 for tag in soup(['script', 'style', 'nav', 'footer', 'header', 'aside', 'iframe']): tag.decompose() # 复杂策略:使用 readability-lxml 或 trafilatura 等专用库 # 这里演示一个启发式方法:寻找包含最多文本的连续块 body_text = '' # 假设正文在 <article> 或 <main> 标签内 article = soup.find('article') or soup.find('main') or soup.body if article: # 获取所有文本,并合并多余空白 body_text = ' '.join(article.get_text().split()) # 3. 提取元数据(如发布时间) publish_time = None # 查找常见的发布时间标签属性 time_tag = soup.find('time') if time_tag and time_tag.get('datetime'): publish_time = time_tag['datetime'] else: # 尝试从 meta 标签中找 meta_pub = soup.find('meta', property='article:published_time') if meta_pub: publish_time = meta_pub['content'] # 4. 基础清洗 # 移除多余的空白字符、换行符 body_text = re.sub(r'\s+', ' ', body_text).strip() return { 'url': url, 'title': title, 'content': body_text, 'publish_time': publish_time, 'source': 'web_crawler' }

4.2 文本分割的艺术与策略

直接将整篇文章存入向量数据库,检索时很可能因为内容太泛而找不到重点。文本分割的目标是创建语义上连贯、大小适中的片段。

from langchain.text_splitter import RecursiveCharacterTextSplitter # 或者使用 tiktoken 进行精确的token计数分割(适用于GPT模型) def split_text_into_chunks(text, chunk_size=500, chunk_overlap=50): """ 使用递归字符分割器将长文本切分成块。 :param text: 待分割的文本 :param chunk_size: 每个块的最大字符数 :param chunk_overlap: 块之间的重叠字符数,防止语义断裂 :return: 文本块列表 """ # 初始化分割器 text_splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, length_function=len, # 使用字符长度,对于中文更合适。英文可用 tiktoken 计算token。 separators=["\n\n", "\n", "。", "!", "?", ";", ",", " ", ""] # 中文分隔符优先级 ) chunks = text_splitter.split_text(text) return chunks # 实际应用 cleaned_data = extract_clean_content(html, url) if cleaned_data['content']: text_chunks = split_text_into_chunks(cleaned_data['content']) for i, chunk in enumerate(text_chunks): print(f"Chunk {i+1} (长度: {len(chunk)}): {chunk[:100]}...")

分割参数的心得

  • chunk_size:不宜过大或过小。太小(如100)会丢失上下文,太大(如2000)会降低检索精度。对于通用问答,500-1000字符是个不错的起点。这大致对应GPT等模型上下文窗口的一小部分,能容纳一个完整的观点或事实。
  • chunk_overlap:至关重要。设置为chunk_size的10%-20%。这能确保一个句子或关键概念如果恰好在边界处,不会完全丢失,相邻的chunk会包含它,保证了上下文的连续性。
  • 分隔符(separators:定义了分割的优先级。上面的列表表示:先按双换行分,再按单换行,再按句号...以此类推。对于中文,将句号、问号等标点加入分隔符列表非常重要。

4.3 元数据关联:让Chunk“有据可查”

每个文本块(Chunk)不能孤立存在,必须携带来源信息,这样当大模型引用时,我们可以追溯到原文。

def create_chunks_with_metadata(full_data, text_chunks): """ 为每个文本块附加元数据。 :param full_data: 从 extract_clean_content 返回的字典 :param text_chunks: 分割后的文本块列表 :return: 包含元数据的块字典列表 """ chunks_with_meta = [] for idx, chunk_text in enumerate(text_chunks): chunk_data = { 'id': f"{full_data['url']}#chunk_{idx}", # 唯一ID 'text': chunk_text, 'metadata': { 'source': full_data['url'], 'title': full_data['title'], 'chunk_index': idx, 'publish_time': full_data.get('publish_time'), 'total_chunks': len(text_chunks) } } chunks_with_meta.append(chunk_data) return chunks_with_meta

这样,每个chunk都包含了原文的URL、标题、发布时间以及它在原文中的位置。在后续的RAG流程中,这些元数据可以随答案一起返回给用户,增加可信度。

5. 向量化与入库:构建可检索的知识核心

处理好的文本块需要转化为向量,并存入专门的数据库。

5.1 嵌入模型选择与本地部署

对于中文场景,我强烈推荐BAAI/bge-large-zh-v1.5模型。它在中文语义相似度任务上表现优异,且完全开源。

# 使用 sentence-transformers 库 from sentence_transformers import SentenceTransformer import torch # 指定模型名称,会自动下载(首次) model_name = 'BAAI/bge-large-zh-v1.5' # 如果你有GPU device = 'cuda' if torch.cuda.is_available() else 'cpu' print(f"正在加载嵌入模型: {model_name}, 设备: {device}") embedding_model = SentenceTransformer(model_name, device=device) # 将文本列表转换为向量 texts = ["这是第一个文本块", "这是第二个文本块"] embeddings = embedding_model.encode(texts, normalize_embeddings=True) # 归一化,便于余弦相似度计算 print(f"向量维度: {embeddings.shape}") # 例如 (2, 1024)

实操心得normalize_embeddings=True非常重要。它将向量归一化为单位长度,这样后续计算余弦相似度就简化为点积运算,速度更快,且余弦相似度是衡量语义相似度的更佳指标。

5.2 向量数据库选型与操作(以Chroma为例)

Chroma是一个轻量级、易用的向量数据库,特别适合原型开发和中小规模项目。

import chromadb from chromadb.config import Settings # 1. 初始化客户端和集合(Collection) # 持久化模式 client = chromadb.PersistentClient(path="./chroma_db") # 数据将保存在本地目录 # 或者使用内存模式(重启后数据丢失) # client = chromadb.Client() # 创建一个集合,类似于数据库中的表 collection = client.get_or_create_collection( name="my_knowledge_base", metadata={"hnsw:space": "cosine"} # 使用余弦相似度作为距离度量 ) # 2. 准备要添加的数据 chunks_data = create_chunks_with_metadata(full_data, text_chunks) # 接上一节的数据 ids = [] documents = [] metadatas = [] embeddings_list = [] for chunk in chunks_data: ids.append(chunk['id']) documents.append(chunk['text']) metadatas.append(chunk['metadata']) # 注意:这里我们提前计算好嵌入向量。Chroma也支持传入模型自动计算,但自己控制更灵活。 embedding = embedding_model.encode([chunk['text']], normalize_embeddings=True)[0].tolist() embeddings_list.append(embedding) # 3. 批量添加到集合 if ids: collection.add( ids=ids, documents=documents, # 原始文本 metadatas=metadatas, embeddings=embeddings_list # 预计算的向量 ) print(f"成功添加 {len(ids)} 个文本块到知识库。") # 4. 查询:找到与问题最相关的片段 query = "如何安装Playwright?" query_embedding = embedding_model.encode([query], normalize_embeddings=True)[0].tolist() results = collection.query( query_embeddings=[query_embedding], n_results=3, # 返回最相关的3个结果 # include=['documents', 'metadatas', 'distances'] # 指定返回的内容 ) print("检索结果:") for i, (doc, meta, dist) in enumerate(zip(results['documents'][0], results['metadatas'][0], results['distances'][0])): print(f"\n--- 结果 {i+1} (距离: {dist:.4f}) ---") print(f"来源: {meta['title']} ({meta['source']})") print(f"内容: {doc[:200]}...")

距离(distances):这里返回的是余弦距离(1 - 余弦相似度)。值越小,表示语义越相似。通常,距离小于0.2可以认为是高度相关。

5.3 流程自动化与调度

将以上所有步骤串联起来,形成一个自动化流水线脚本。同时,需要考虑定时采集和增量更新。

import asyncio import hashlib from urllib.parse import urlparse import json import os class KnowledgeBaseCrawler: def __init__(self, start_urls, chroma_path="./chroma_db", model_name='BAAI/bge-large-zh-v1.5'): self.start_urls = start_urls self.visited_urls = set() self.chroma_client = chromadb.PersistentClient(path=chroma_path) self.collection = self.chroma_client.get_or_create_collection(name="web_knowledge") self.embedding_model = SentenceTransformer(model_name) # 用于去重或记录状态的简单文件 self.state_file = 'crawler_state.json' async def crawl_and_process(self, url): """针对单个URL的完整抓取处理流程""" if url in self.visited_urls: return print(f"正在处理: {url}") # 1. 使用Playwright抓取 html_content = await self.fetch_with_playwright(url) if not html_content: return # 2. 解析清洗 cleaned_data = extract_clean_content(html_content, url) if not cleaned_data.get('content'): print(f" 警告: 未从 {url} 提取到有效内容") return # 3. 文本分割 text_chunks = split_text_into_chunks(cleaned_data['content']) # 4. 生成向量并入库 chunks_with_meta = create_chunks_with_metadata(cleaned_data, text_chunks) ids_to_add = [] embeddings_to_add = [] metadatas_to_add = [] documents_to_add = [] for chunk in chunks_with_meta: # 检查是否已存在(通过ID或内容哈希) chunk_id = chunk['id'] # 简单查重:检查ID是否已存在(更严谨的做法是检查内容哈希) existing = self.collection.get(ids=[chunk_id]) if existing['ids']: # 已存在 print(f" 跳过已存在的块: {chunk_id}") continue embedding = self.embedding_model.encode([chunk['text']], normalize_embeddings=True)[0].tolist() ids_to_add.append(chunk_id) embeddings_to_add.append(embedding) metadatas_to_add.append(chunk['metadata']) documents_to_add.append(chunk['text']) # 批量添加 if ids_to_add: self.collection.add( ids=ids_to_add, embeddings=embeddings_to_add, metadatas=metadatas_to_add, documents=documents_to_add ) print(f" 已添加 {len(ids_to_add)} 个新块") self.visited_urls.add(url) async def fetch_with_playwright(self, url): """Playwright抓取封装""" # ... (实现细节,包含错误处理、重试等) pass def save_state(self): """保存爬取状态""" with open(self.state_file, 'w') as f: json.dump({'visited': list(self.visited_urls)}, f) def load_state(self): """加载爬取状态""" if os.path.exists(self.state_file): with open(self.state_file, 'r') as f: state = json.load(f) self.visited_urls = set(state.get('visited', [])) async def run(self): """主运行循环""" self.load_state() async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context() # 将browser和context保存为实例变量,供fetch_with_playwright使用 self.browser = browser self.context = context for url in self.start_urls: await self.crawl_and_process(url) # 可以在这里添加延时,避免请求过快 await asyncio.sleep(1) await browser.close() self.save_state() # 使用示例 async def main(): start_urls = [ 'https://playwright.dev/python/docs/intro', 'https://example.com/tech-article-1', # ... 更多初始URL ] crawler = KnowledgeBaseCrawler(start_urls) await crawler.run() # 定时任务可以使用 APScheduler 或 Celery 进行调度

6. 避坑指南与性能优化实战

在实际操作中,你会遇到各种各样的问题。以下是我踩过的一些坑和总结的经验。

6.1 常见问题与解决方案

问题可能原因解决方案
Playwright 超时页面加载慢、网络差、等待条件未满足1. 增加timeout参数(如page.goto(url, timeout=60000))。
2. 使用wait_for_selector等待特定元素而非networkidle
3. 设置更宽松的等待条件,如wait_until='domcontentloaded'
元素找不到页面结构变化、元素动态生成、iframe1. 使用更稳定的选择器(如>被网站屏蔽请求频率过高、User-Agent 被识别、IP 被封1. 在请求间添加随机延时await asyncio.sleep(random.uniform(1, 3))
2. 轮换 User-Agent 和浏览器上下文。
3. 使用代理 IP(需谨慎,确保合规)。
4. 模拟人类行为(如随机滚动、鼠标移动)。
提取内容杂乱正文提取算法不准,包含过多噪音1. 使用更专业的库如readability-lxmltrafilatura
2. 针对特定网站编写定制化的提取规则(XPath/CSS选择器)。
3. 训练一个简单的机器学习模型来识别正文区域(成本较高)。
向量检索不准chunk 分割不合理、嵌入模型不匹配、相似度阈值设置不当1. 调整chunk_sizechunk_overlap
2. 尝试不同的嵌入模型(针对中文/英文)。
3. 在检索后对结果进行重排序(Re-ranking)。
4. 设置相似度阈值,过滤掉低质量结果(如distance < 0.3)。
数据库性能下降向量数量过多、索引未优化1. 对于 Chroma,确保使用持久化模式并定期清理。
2. 对于大规模数据(>10万条),考虑使用 Milvus 或 Qdrant,它们为大规模 ANN 搜索优化。
3. 建立合适的索引(如 HNSW)。

6.2 性能优化技巧

  1. 异步并发采集:Playwright 原生支持异步,利用asyncio.gather可以并发抓取多个页面,极大提升效率。

    async def fetch_url(url, context): page = await context.new_page() try: await page.goto(url, wait_until='networkidle') content = await page.content() return url, content finally: await page.close() async def main(): async with async_playwright() as p: browser = await p.chromium.launch() context = await browser.new_context() urls = ['url1', 'url2', 'url3'] tasks = [fetch_url(url, context) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) # 处理 results await browser.close()
  2. 复用浏览器上下文:创建和销毁浏览器实例开销很大。在整个采集任务中,尽量复用同一个browser和多个context

  3. 选择性加载资源:如前所述,通过路由拦截禁用图片、字体、CSS等非必要资源,可以显著加快页面加载速度并减少带宽消耗。

  4. 嵌入模型批处理sentence-transformersencode方法支持批量输入。一次性传入一个文本列表进行向量化,比循环调用单次encode快得多。

  5. 增量更新与去重:每次采集前,检查 URL 或内容哈希是否已存在于知识库中,避免重复工作和存储浪费。可以在元数据中增加last_updated字段,实现基于时间的增量更新。

6.3 关于合规性与道德的思考

虽然技术很强大,但我们必须合法合规地使用。

  • 遵守robots.txt:在采集前,检查目标网站的robots.txt文件,尊重网站所有者设置的爬虫规则。
  • 控制请求速率:在请求间添加延迟,避免对目标服务器造成过大压力,这既是道德要求,也能降低被封禁的风险。
  • 识别公开数据与个人数据:只采集公开可访问的信息,绝不触碰需要登录才能访问的个人隐私数据或受版权严格保护的内容。
  • 注明数据来源:在最终的知识库应用呈现答案时,尽可能附上原文链接,尊重内容创作者。

构建AI知识库的自动化采集系统,是一个将软件工程、数据工程和机器学习结合起来的综合项目。从Playwright的精准抓取,到文本处理的细致清洗,再到向量化的语义升华,每一步都影响着最终知识库的“智商”。这套流程并非一成不变,你需要根据目标网站的特点、数据的性质以及最终应用的需求进行灵活调整和优化。最关键的还是动手去试,在真实的数据流中发现问题、解决问题,你的知识库才会越来越“聪明”。

http://www.jsqmd.com/news/1067880/

相关文章:

  • 企业级接口自动化测试框架构建:从动态参数到数据驱动的实战指南
  • Nacos安全加固实战:从CVE-2021-29441漏洞看鉴权配置与生产环境部署
  • 基于Frida的Android应用动态脱壳原理与实战指南
  • 密码学基础:对称加密、非对称加密、哈希
  • MeterSphere接口自动化场景构建:从变量传递到数据驱动的全流程实战
  • 旅游场景下即开即用的Vue3租房H5模板,含完整房源浏览与联系功能
  • Matlab一键绘制非线性系统庞加莱截面图的实操工具包
  • XSS攻防实战:从靶场到企业级防御体系构建
  • PBEWithMD5AndDES跨语言加解密:Java与Python兼容实现详解
  • 基于Playwright与FastAPI构建高可用GitHub趋势爬虫API服务
  • Web认证安全实战:从OWASP指南到代码落地的纵深防御体系
  • Apifox AI 如何智能生成API测试用例:从文档到自动化的实践指南
  • JMeter WebSocket压测全攻略:从环境配置到高并发调优
  • 实战指南:从零部署与调优OWASP ModSecurity CRS Web应用防火墙
  • pytest固件失效排查:从xUnit到fixture的正确使用指南
  • JDBC连接字符串反序列化漏洞深度剖析:从原理到实战化EXP开发
  • MATLAB语音加噪降噪全流程:含SNR自动计算、时频对比图与多种滤波实现
  • WSAIOS v3.0 架构设计与核心实现
  • Java密码安全存储实战:从BCrypt到Argon2的演进与实现
  • Pytest执行参数全解析:从基础筛选到CI/CD集成实战
  • DeepSeek-V4并行与THD模式:大模型推理的硬件级执行契约
  • Appium Python Client扩展开发:自定义命令与连接管理实战
  • 交通路口视频监控后台系统(Vue2+原生JS,含部署指南与毕设适配说明)
  • 从basic_pentesting_2靶机实战入门渗透测试:信息收集到权限提升全流程解析
  • FastAPI OAuth2 JWT认证系统实战:从密码哈希到令牌刷新的完整实现
  • JMeter压力测试实战避坑指南:从环境配置到结果分析的常见误区与解决方案
  • JMeter实战指南:从接口测试到性能压测的全流程解析
  • 行星齿轮箱振动仿真MATLAB工具:含时变刚度与齿隙建模
  • Python实现Ascon轻量级加密算法:从原理到AEAD工具开发
  • CNN-LSTM加注意力机制的RUL预测完整复现包:含双方案代码、数据与结果