AI 时代的数据工程师 ：从隧道代理抓取到向量数据库存储的全闭环实战

在 2026 年单纯抓取网页 HTML 已经没有意义了。真正的价值在于将这些数据转化为向量Embedding并存入向量数据库构建属于你自己的 RAG检索增强生成系统。今天我们将升级数据管道实现“高效抓取 - 自动清洗 - 向量存储”的一体化流程。升级版工具箱代理服务隧道代理解决封禁与高成本。数据清洗BeautifulSoupre去除噪音。向量数据库Milvus(使用pymilvus及其内置的轻量化模式)。Embedding 模型Sentence-Transformers将文本转为坐标。pipinstallrequests beautifulsoup4 pymilvus sentence-transformers核心代码实现全自动化 AI 数据工厂下面这段代码展示了如何利用隧道代理突破反爬限制并迅速将数据入库。importrequestsimportreimportrandomfrombs4importBeautifulSoupfrompymilvusimportMilvusClient,model# # 1. 配置信息 (代理与数据库)# # 亿牛云爬虫代理配置PROXY_DOMAINPROXY.16yun.cnPROXY_PORT6447PROXY_USER16YUN_USERPROXY_PASS16YUN_PASS# 初始化 Milvus Lite (本地轻量化向量数据库)milvus_clientMilvusClient(ai_data_factory.db)COLLECTION_NAMEweb_knowledge_base# 初始化嵌入模型 (用于将文本转为向量)embedding_fnmodel.DefaultEmbeddingFunction()defsetup_db():初始化向量库集合ifmilvus_client.has_collection(COLLECTION_NAME):milvus_client.drop_collection(COLLECTION_NAME)milvus_client.create_collection(collection_nameCOLLECTION_NAME,dimension768# 默认模型维度)print(f[√] 向量库{COLLECTION_NAME}初始化成功)# # 2. 带有代理保护的抓取函数# deffetch_and_clean(url): 使用隧道代理抓取并清洗数据 proxy_metafhttp://{PROXY_USER}:{PROXY_PASS}{PROXY_DOMAIN}:{PROXY_PORT}proxies{http:proxy_meta,https:proxy_meta}headers{User-Agent:Mozilla/5.0 (Windows NT 10.0; Win64; x64) AI-Data-Engineer/1.0,Accept:text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8}try:print(f[*] 正在通过代理请求:{url})# 使用隧道代理发起请求自动切换出口IPresponserequests.get(url,headersheaders,proxiesproxies,timeout15)ifresponse.status_code200:# --- 数据清洗逻辑 ---soupBeautifulSoup(response.text,html.parser)# 移除脚本、样式、导航栏等噪音forscript_or_styleinsoup([script,style,nav,footer,header]):script_or_style.decompose()# 获取纯文本并去除多余空行textsoup.get_text()clean_textre.sub(r\n,\n,text).strip()print(f[√] 成功抓取并清洗数据长度:{len(clean_text)}字符)returnclean_textexceptExceptionase:print(f[-] 抓取失败:{e})returnNone# # 3. 向量化与入库# defsave_to_vector_db(text,url): 将清洗后的文本向量化并存入 Milvus ifnottextorlen(text)50:return# 简单切片将长文本切成 500 字的小块方便 AI 检索chunks[text[i:i500]foriinrange(0,len(text),500)]# 转换为向量vectorsembedding_fn.encode_documents(chunks)# 构造入库数据data[{id:i,vector:vectors[i],text:chunks[i],source:url}foriinrange(len(chunks))]# 插入数据库milvus_client.insert(collection_nameCOLLECTION_NAME,datadata)print(f[√] 已将{len(chunks)}条切片存入向量数据库)# # 4. 主程序运行# if__name____main__:# 初始化环境setup_db()# 模拟抓取任务 (这里可以替换为你的目标 AI 新闻站或技术文档)target_urls[https://www.example.com/ai-news-1,http://httpbin.org/ip# 仅做代理测试]forurlintarget_urls:contentfetch_and_clean(url)ifcontent:save_to_vector_db(content,url)# 尝试检索模拟 AI 提问print(\n[*] 模拟 RAG 检索测试...)search_resmilvus_client.search(collection_nameCOLLECTION_NAME,data[embedding_fn.encode_queries([最新的 AI 动态是什么])][0],limit1,output_fields[text,source])forhitinsearch_res[0]:print(f找到最相关内容:{hit[entity][text][:100]}...)print(f来源:{hit[entity][source]})关键技术点解析1. 为什么代码中不写“IP 切换”逻辑因为我们使用了隧道代理。普通的代理 IP 池需要你手动去维护requests.get(proxiesrandom.choice(ips))而隧道代理在云端自动完成了负载均衡和 IP 轮换。你的代码只需要像连接单个代理一样简单剩下的“脏活累活”全由代理服务器在后台搞定。2. 低成本的秘密Milvus Lite在开发阶段或小型项目中不需要部署沉重的 Docker 容器。代码中使用的MilvusClient(ai_data_factory.db)会在本地生成一个 SQLite 样式的数据库文件这极大降低了硬件成本。3. 数据清洗的重要性AI 模型最讨厌 HTML 标签中的div、span或 JavaScript 代码。通过BeautifulSoup的decompose()方法剔除噪音不仅能节省向量数据库的存储空间还能显著提高 Embedding 的准确率。结语从 2026 年的视角来看一个优秀的采集方案必须具备“数据全周期管理”的能力。通过隧道代理保证稳定的输入再通过向量数据库完成知识的沉淀你不仅是在写代码更是在为 AI 时代构建基础设施。