做开发的人几乎都遇到过这样的场景想找一个靠谱的 AI 模型 API、一套开源方案或者一份高质量的数据集结果打开搜索引擎看到的是满屏广告、过时教程、包装过度的“AI 神器”真正想找的资源淹没在信息噪音里。更烦人的是同一个东西在 A 平台可能标价几千在 B 平台换了马甲就成了上万块的“企业级解决方案”。这不是个别现象而是 AI 赛道快速膨胀带来的典型问题信息透明度不够货源分散价格体系混乱。这篇文章要讲的就是如何从工程角度搭建一套“全网低价 AI 货源检索工具”。它不是简单的爬虫脚本而是一个覆盖数据采集、清洗、比价、聚合、检索的完整链路。读完这篇文章你能理解这类工具的技术原理搞清楚它和普通搜索引擎、传统爬虫工具的本质区别并且拿到一套可以直接运行的最小示例了解从零搭建时需要处理的关键难点和容易踩坑的地方。1. 这类工具到底解决了什么问题先想清楚需求本质。所谓“AI 货源”范围非常宽包括模型 API、训练数据集、开源代码、算力资源、AI 工具账号、部署方案等。对开发者和企业来说采购 AI 相关资源时最大的痛点不是资源少而是资源太分散同一个资源在不同平台的价格差异可能非常大。如果没有专门的检索工具传统做法是这样的先打开几个主流平台分别搜索再手动记录价格和参数最后用表格对比。这个流程有三个明显问题一是覆盖不全。AI 行业平台众多新平台和渠道不断出现人工不可能持续盯住所有渠道。二是数据滞后。AI 产品的定价变动频繁尤其是 API 按量计费的产品今天看到的价格可能明天就失效了。搜索结果里大量内容也是过时的教程和广告页面。三是对比困难。不同平台对同类产品的计费方式不同有的按次收费有的按 Token 收费有的按小时租用 GPU很难用一个统一口径去对比。这套检索工具要解决的就是这三个问题用程序化方式扩大覆盖范围用定时任务保证数据新鲜度用统一的标准化字段把不同平台的资源拉平对比。从技术属性看它属于垂直领域的数据采集与检索系统和通用比价网站有相似之处但差异也明显。最大的差异在于结构化难度电商商品有相对统一的 SKU 和价格字段而 AI 货源的计费单位、规格参数、功能边界差异极大做数据标准化比做普通比价系统更难。所以这套工具的核心竞争力不在于“能爬”而在于清洗和归一化能力——能否把不同平台、不同表述、不同计量方式的资源转为同一个可比较的框架。2. 核心概念与整体设计思路在动手写代码之前需要先理解几个关键概念。货源数据源指检索工具的采集对象包括公开的商品页面、API 文档定价页、开发者社区评测、开源仓库说明等。按数据类型可以分成三类结构化数据源指有明确字段的页面或接口比如云厂商的定价 API、第三方比价网站的数据表格半结构化数据源指 HTML 页面中嵌套着可识别数字的页面比如大多数官网的产品页非结构化数据源指博客、论坛、评测文章等需要自然语言处理才能提取信息的文本。价格归一化指把不同平台的计费方式换算成统一口径。这是全网低价检索最核心的一步。比如“每 1000 Token 0.002 美元”和“每 1M Token 2 美元”其实是同一个价格但如果不做归一化排序算法的结果就会失真。数据指纹指用于判断数据是否重复或者是否已更新的唯一标识。因为同一款 AI 模型可能出现在多个平台的销售页面上展示名略有差异但其实是同一个商品所以需要通过名称规范化、URL 指纹等方式去重。整体架构建议采用分层设计模块名称职责关键技术点采集层从多个数据源抓取原始页面请求调度、限速、代理池、失败重试解析层从 HTML/接口中提取结构化字段XPath、CSS 选择器、JSON 解析数据标准化层字段清洗、价格归一化、格式统一正则、单位换算函数去重与聚合层识别同一货品聚合多个来源抓取指纹、相似度计算存储层保存商品数据、价格历史、抓取日志SQLite、MySQL、时序数据库查询检索层提供按价格、关键词、类目检索能力SQL 查询、倒排索引、API 服务对于个人开发者或小团队第一版不需要做分布式爬虫单机 定时任务 SQLite 基本够用。文章后续代码示例都基于这个轻量级架构。3. 环境准备与工程目录设计先说明一点本文演示的是通用实现思路具体的依赖版本以你实际安装时项目支持的版本为准不要死记硬背版本号。核心技术栈如下Python 3.9 及以上版本建议使用 3.11性能更好Type Hint 支持更完善。网络请求使用的库requests、httpx。HTML 解析使用的库BeautifulSoup4、lxml。数据存储SQLite内置模块 sqlite3生产环境可换 MySQL 或 PostgreSQL。轻量 Web 查询接口Flask 或 FastAPI。定时任务方案Linux cron 或 APScheduler。推荐使用虚拟环境管理依赖mkdir ai-source-search cd ai-source-search python -m venv venv source venv/bin/activate # Windows 下为 venv\Scripts\activate pip install requests beautifulsoup4 lxml flask httpx建议的工程文件结构ai-source-search/ ├── collector/ │ ├── __init__.py │ ├── base.py # 抓取器抽象基类 │ ├── http_client.py # 请求客户端封装 │ └── sources/ # 各数据源抓取器 │ ├── platform_a.py │ └── platform_b.py ├── parser/ │ ├── __init__.py │ └── field_parser.py # 字段清洗和归一化函数 ├── storage/ │ ├── __init__.py │ └── database.py # SQLite 存储操作 ├── retriever/ │ ├── __init__.py │ └── query.py # 查询服务 ├── config.py # 数据源配置 ├── main.py # 入口脚本 └── requirements.txt这里的核心思想是每个数据源对应一个独立的抓取器。AI 货源的页面结构差异很大不可能写一个通用函数通吃所有网站所以抽象基类加上各平台的子类实现是更可维护的做法。4. 核心流程拆解从抓取到可检索数据一次完整的“低价货源检索”过程包含五个阶段。第一阶段是配置数据源。在 config.py 中定义每个平台的入口 URL、抓取频率、解析规则。需要注意不同平台的页面结构会变抓取规则要集中管理方便失效后快速定位。第二阶段是采集原始页面。使用 requests 或 httpx 发起请求。这里有两个关键点一是设置合理的 User-Agent 和请求头模拟真实浏览器行为二是控制请求频率避免对目标站点造成压力。更稳妥的做法是访问对方提供的对外接口或公开数据文件如果平台没有公开 API就遵循 robots.txt 约束以较低频率访问公开页面。第三阶段是解析字段。从 HTML 或 JSON 中提取名称、规格、价格、计费单位、链接、更新时间。这一阶段最容易出错因为源站页面格式一旦改版选择器就会失效。第四阶段是清洗与归一化。这是整个链路里最影响最终质量的一步。需要做四类处理去除 HTML 标签残留和转义符将“每 1000 tokens $0.002”和“1M tokens $2.00”统一成同一种基准单位把模型名中的大小写、全半角、空格差异统一处理时间字段统一为时间戳格式便于排序。第五阶段是存储与建立索引。将处理后的数据写入 SQLite 表建立名称索引和价格索引。查询时按关键词模糊匹配按归一化价格升序排列就能得到“全网低价”的排序结果。5. 完整示例代码实现下面这套示例能跑通最小闭环抓取两个模拟数据源的页面解析出 AI 模型货源信息统一价格单位存入 SQLite再通过 Web 接口按价格升序检索。示例一HTTP 客户端封装# collector/http_client.py import random import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class HttpClient: 统一请求客户端内置重试、限速和基础请求头。 USER_AGENTS [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15, Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36, ] def __init__(self, min_interval: float 1.0, max_interval: float 3.0): self.session requests.Session() self.min_interval min_interval self.max_interval max_interval retry Retry(total2, backoff_factor0.5, status_forcelist[429, 500, 502, 503, 504]) adapter HTTPAdapter(max_retriesretry) self.session.mount(http://, adapter) self.session.mount(https://, adapter) def get(self, url: str, timeout: int 15, headers: dict | None None): # 基础限速避免对目标站点造成过大访问压力 time.sleep(random.uniform(self.min_interval, self.max_interval)) request_headers { User-Agent: random.choice(self.USER_AGENTS), Accept: text/html,application/json;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, } if headers: request_headers.update(headers) response self.session.get(url, headersrequest_headers, timeouttimeout) response.raise_for_status() return response这个封装解决的是采集环节最基础的三个问题请求头伪装、超时控制、失败重试。在真实项目中如果采集目标数量较大可以考虑引入代理池但第一版不建议做因为代理池的维护成本很高而且对于公开信息采集来说没必要过度设计。示例二价格归一化与字段清洗# parser/field_parser.py import re from datetime import datetime from typing import Any class FieldParser: 字段清洗和价格归一化工具类。 staticmethod def clean_text(value: str) - str: 清理文本中的空白字符和 HTML 残留标签。 if not value: return text re.sub(r[^], , value) text re.sub(r\\u0026|amp;|nbsp;, , text) text re.sub(r\\s, , text) return text.strip() staticmethod def normalize_model_name(name: str) - str: 统一模型名称的小写、全半角和多余空格。 if not name: return name name.strip().lower() name name.replace(, ().replace(, )) name name.replace(, :).replace(, ,) name re.sub(r\\s, -, name) return name staticmethod def normalize_price_to_dollar(price_value: float, unit: str) - float: 将不同计费方式归一化为每 1M Token 的美元价格。 示例 - unit per_1k_tokens 表示原始价格是每 1000 Token 的价格 - unit per_1m_tokens 表示原始价格是每 1M Token 的价格 - 转换时统一乘以 1000得到每 1M Token 的价格 if unit per_1k_tokens: return round(price_value * 1000, 6) if unit per_1m_tokens: return round(price_value, 6) raise ValueError(f不支持的计费单位: {unit}) staticmethod def parse_price_string(price_string: str) - float | None: 从 $0.002、0.002美元 等文本中提取数字。 if not price_string: return None match re.search(r\\d\\.?\\d*, price_string.replace(,, )) if not match: return None return float(match.group()) staticmethod def parse_time(time_str: str) - str: 把常见时间字符串转成时间戳格式解析失败时返回当前时间。 formats [ %Y-%m-%d %H:%M:%S, %Y-%m-%d, %Y/%m/%d, ] for fmt in formats: try: return datetime.strptime(time_str, fmt).strftime(%Y-%m-%d %H:%M:%S) except (ValueError, TypeError): continue return datetime.now().strftime(%Y-%m-%d %H:%M:%S)价格归一化是很容易被低估的一步。表面上看只是单位换算但实际的换算场景比这个复杂得多有的 API 还区分输入价格和输出价格有的按 CPU 时长计费有的按实例规格计费。在第一版中建议只覆盖按 Token 或按固定单价计费的产品其他场景后续再扩展。示例三SQLite 存储层# storage/database.py import sqlite3 from datetime import datetime class Database: SQLite 操作封装负责建表、写入和查询。 def __init__(self, db_path: str ai_sources.db): self.conn sqlite3.connect(db_path) self.conn.row_factory sqlite3.Row self.create_tables() def create_tables(self): self.conn.execute( CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, normalized_name TEXT NOT NULL, category TEXT, platform TEXT, price_usd REAL, original_price TEXT, unit TEXT, currency TEXT, product_url TEXT, description TEXT, updated_at TEXT ) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_products_name ON products (normalized_name) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_products_price ON products (price_usd) ) self.conn.commit() def upsert_product(self, product: dict): 使用 INSERT OR REPLACE 写入商品数据。 真实场景中建议增加 fingerprint 字段做更精确的去重。 now datetime.now().strftime(%Y-%m-%d %H:%M:%S) self.conn.execute( INSERT OR REPLACE INTO products (name, normalized_name, category, platform, price_usd, original_price, unit, currency, product_url, description, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( product.get(name), product.get(normalized_name), product.get(category), product.get(platform), product.get(price_usd), product.get(original_price), product.get(unit), product.get(currency), product.get(product_url), product.get(description, ), now, )) self.conn.commit() def search_cheapest(self, keyword: str, category: str | None None, limit: int 20): 按关键词模糊匹配并按价格升序返回结果。 sql SELECT * FROM products WHERE name LIKE ? OR normalized_name LIKE ? OR category LIKE ? params [f%{keyword}%, f%{keyword}%, f%{keyword}%] if category: sql AND category ? params.append(category) sql ORDER BY price_usd ASC LIMIT ? params.append(limit) rows self.conn.execute(sql, params).fetchall() return [dict(row) for row in rows] def close(self): self.conn.close()写这一层时需要注意price_usd 字段是整个检索系统的核心排序字段必须保证写入前已经完成归一化否则排序出来的“低价”没有参考意义。示例四数据源抓取器以模拟平台 A 为例# collector/sources/platform_a.py from collector.http_client import HttpClient from parser.field_parser import FieldParser from bs4 import BeautifulSoup class PlatformACollector: 平台 A 的页面结构假设如下 div classmodel-card h3GPT-4-like Model/h3 span classprice$0.003 / 1K tokens/span /div def __init__(self): self.client HttpClient() def fetch(self, list_url: str) - list[dict]: response self.client.get(list_url) soup BeautifulSoup(response.text, lxml) products [] for card in soup.select(.model-card): name FieldParser.clean_text(card.select_one(h3).get_text()) price_text FieldParser.clean_text(card.select_one(.price).get_text()) price_value FieldParser.parse_price_string(price_text) if not name or price_value is None: continue products.append({ name: name, normalized_name: FieldParser.normalize_model_name(name), category: llm_api, platform: platform_a, price_usd: FieldParser.normalize_price_to_dollar(price_value, per_1k_tokens), original_price: price_text, unit: per_1k_tokens, currency: USD, product_url: list_url, description: 来自平台 A 的模型 API 货源, }) return products这个抓取器的写法体现了“一个平台一个类”的设计原则。每个平台的页面结构差异很大把解析逻辑封装在各自的类里修改互不影响。如果未来平台 A 改版只需要调整这一个文件中的 CSS 选择器。示例五查询服务与主流程# retriever/query.py from flask import Flask, jsonify, request from storage.database import Database app Flask(__name__) db Database() app.route(/search) def search(): keyword request.args.get(q, ) category request.args.get(category, None) if not keyword: return jsonify({error: 缺少关键词参数 q}), 400 results db.search_cheapest(keyword, category) return jsonify({ count: len(results), items: results, order: price_usd asc }) if __name__ __main__: app.run(host0.0.0.0, port8000, debugTrue)# main.py from collector.sources.platform_a import PlatformACollector from storage.database import Database PLATFORM_A_PAGE https://example-a.com/pricing # 替换为实际页面 def main(): db Database() collector PlatformACollector() products collector.fetch(PLATFORM_A_PAGE) for product in products: db.upsert_product(product) print(f写入 {len(products)} 条货源数据) db.close() if __name__ __main__: main()运行主流程python main.py预期输出类似写入 12 条货源数据启动查询服务python retriever/query.py浏览器访问http://127.0.0.1:8000/search?qmodelcategoryllm_api返回 JSON 数组中 items 会按 price_usd 从低到高排列。这就是“低价检索”的最小可用版本。6. 运行效果与验证方式判断这套系统是否跑通的依据有三个维度第一采集数量是否符合预期。如果配置了两个数据源每个数据源预期 10 条数据入库后总数应该在 20 条左右。如果明显偏少通常是解析选择器失效或者页面结构出现了变化。第二价格是否经过归一化。检查数据库中的 price_usd 字段不同平台的同类产品应该在同一量级。比如平台 A 显示“$0.003/1K tokens”平台 B 显示“$2/1M tokens”归一化后都应该是 3 美元每 1M Token。第三查询排序是否合理。用价格明显有差异的关键词测试比如搜索同一个模型名应该看到多个平台的价格升序结果。如果查询返回为空优先确认以下三个位置数据库表 products 是否有数据关键词是否大小写和名称不匹配normalized_name 已转小写但 LIKE 匹配是否统一转小写category 条件是否传了不存在的值。生产环境建议增加监控。最简单的方式是每天抓取后统计入库数量和时间写入日志如果某天入库数量为 0触发告警。这一步在数据源改版时能帮你快速发现异常。7. 常见问题与排查方法针对这套检索工具最常遇到的故障整理了一份排查清单问题现象可能原因排查方式解决方案抓取请求超时目标站点响应慢或访问频率过高查看请求日志确认访问间隔增大请求间隔检查网络必要时使用代理解析结果为空页面改版或 CSS 选择器失效用浏览器开发者工具检查页面结构更新选择器建议添加抓取告警价格排序异常价格归一化单位配置错误对比原始价格文本和 price_usd检查 unit 字段重新执行归一化重复数据过多缺少有效去重逻辑查询相同 normalized_name 的数量增加 fingerprint 字段做唯一性约束数据库写入失败SQLite 被多进程同时写查看报错信息使用连接锁或改为排队写入查询效果差关键词匹配策略太简单分析返回结果与实际需求偏差引入倒排索引或更精确的匹配方案在使用这类工具时有一个原则要强调访问任何平台数据时都要严格遵守网站的 robots.txt 规则和服务条款只采集公开信息控制访问频率不做影响目标站点正常运行的操作。这些内容只用于个人学习研究不要超出合理范围使用。8. 工程实践建议完成了最小示例之后如果想把它发展成一个更稳定的系统有几点工程建议值得关注。第一不要低估数据源配置的维护成本。全网低价检索系统的本质是 爬虫 解析规则 的组合而解析规则是有生命周期的每隔一段时间就会因为源站改版而失效。建议把每个数据源的解析规则独立成 JSON 或 YAML 配置而不是硬编码在 Python 文件里。这样非技术人员也能在源站改版后快速修改规则。{ platform_a: { list_url: https://example-a.com/pricing, item_selector: .model-card, name_selector: h3, price_selector: .price } }第二价格采集必须有历史记录。第一版可能只关心“当前谁最便宜”但真实采购场景中还需要知道价格走势判断现在是否是合适的入手时机。建议增加 price_history 表每次抓取时都对当前低价做快照。一个月后你就能利用这些历史数据分析价格波动规律。第三巧用缓存减少抓取压力。对于内容变化不频繁的页面可以缓存当日抓取结果只有当缓存过期时才重新请求。这既减少了对目标站点的访问压力也降低了自身系统被封的风险。第四搜索功能不要止步于 SQL LIKE 匹配。当数据量到几千条之后SQL LIKE 的模糊匹配效果会明显不足。这时候可以考虑使用轻量级方案SQLite FTS5 全文搜索或者接入 Elasticsearch。更好的方案是基于向量检索的思路把产品描述预置成向量结合关键词过滤实现语义搜索。需要明确的是当数据量很少时不需要做语义检索关键词过滤更精准、成本更低。第五对“低价”的定义要透明。系统返回的低价排序是基于归一化后的价格来的。但真实世界里的价格比价没那么简单有些低价是营销策略后续会有其他费用有些价格单位不透明把隐藏成本藏在别的维度。工程上要做的是把这些信息也展示在结果中而不是只给一个数字要让使用者知道这个低价是怎么算出来的。9. 总结与后续方向回到最开始的问题这款“全网低价 AI 货源检索工具”真正解决的是什么它解决的是 AI 资源采购中的信息分散和价格不透明问题。它不靠复杂的人工智能算法而是靠扎实的数据工程通过可配置的数据源采集机制把不同平台的货源数据统一格式通过价格归一化让不同计费口径变得可比再通过排序服务输出低价结果。从开发角度来看这个项目难度适中适合有 Python 基础、想深入理解爬虫全流程和数据治理的开发者练习。做完这套系统你不仅掌握了一个实用的检索工具还能接触到采集层、解析层、存储层、服务层这一整套数据管线的设计方法。下一步值得深入的方向有这么几个把抓取器从同步改为异步用 httpx 的 AsyncClient 配合 asyncio 提升采集效率引入 Redis 做 URL 去重和简单计数器对产品描述做向量化实现语义级检索增加价格监控告警当某类货源价格低于自定义阈值时主动通知。先从这个最小闭环开始把一个数据源跑通、把价格归一化逻辑验证清楚再逐步扩大覆盖范围才是做好这类工具的正确节奏。