1. 智能體知識庫的重要性
我最近在做“歷史大事記”智能體時,踩了個實打實的坑:初期全靠大模型原生知識庫支撐,回答總是“缺斤短兩”:要麼漏了關鍵歷史事件,要麼對人物生卒、傳統習俗的描述模糊不清,甚至連一些廣為人知的紀念日都沒法精準對應。
為了補齊這個短板,我找遍了各種數據源,最終發現維基百科這一“寶藏庫”:它把全年365/366天的內容拆解得明明白白,大到影響世界的歷史拐點,小到行業先驅的誕辰、地方特有的民俗,甚至冷門的科普事件都有收錄,數據維度全、權威性也夠,完全能滿足智能體知識庫的擴充需求。
2. 代理IP在數據採集中的重要性
數據源找到了,但新的數據採集問題又來了:維基百科在國內沒法直接訪問,而且單天數據就包含幾十甚至上百條目數據,365/366天累計下來就是海量數據,手動採集根本不現實,自動化爬蟲又被網絡管理卡了脖子。這時候我才意識到,想要高效抓取這些數據,解決網絡壁壘的代理IP工具是剛需。
合理運用代理IP技術是實現高效穩定網絡爬蟲不可或缺的一環,代理IP幫助爬蟲獲得了更廣泛、更深入的數據資源。這些數據經過清洗、整合後就可以形成高質量的知識庫,為智能體的問題回答提供優質準確的數據支撐。
3. 選擇IPIDEA代理IP的原因
為了高效、穩定的完成數據抓取,我對市面上的一些代理服務商進行了詳細的分析和研究,包括IPIDEA、星空代理、IPIPGO等。具體對比如下:
|
優勢維度 |
IPIDEA核心表現 |
星空代理表現 |
IPIPGO表現 |
|
一、IP資源規模+質量 |
1億+優質IP,含1億+真實住宅IP、千萬級移動IP,自建純淨合規池,嚴格篩選 |
僅提及“海量IP”,無具體數值,以國內運營商授權節點為主,未強調“真實用户IP” |
9000萬+IP,住宅IP以動態為主,移動IP覆蓋有限 |
|
二、覆蓋範圍+精度 |
全球220+國家/地區,支持“國家-州/省-城市-ASN”四級定位,熱門地區IP密度高 |
僅覆蓋國內200+城市,無全球佈局,定位精度僅到城市級 |
全球200+國家/地區,定位精度到城市級,熱門地區(如美國數據中心IP僅3.5萬+)密度低 |
|
三、穩定性 |
99.9%正常運行時間+99.9%採集成功率,支持1-120分鐘自定義時效 |
IP可用率>98%,IP有效時長僅1-15分鐘,無長效選項 |
數據中心IP穩定性99.9%,整體採集成功率未明確,部分套餐帶寬有隱性限制 |
|
四、代理類型 |
8類細分類型(含移動、IPv6、無限量、長效ISP、獨享數據中心等特色類型) |
僅4類基礎套餐(計量/不限量A/B),無移動、IPv6、長效等類型 |
無IPv6、長效ISP代理,移動IP類型單一 |
|
五、高匿名性 |
住宅/移動IP為真實用户網絡,搭配IP輪轉+自定義指紋 |
IP為運營商節點,非真實用户IP,匿名性僅滿足基礎需求 |
住宅IP為真實節點,但規模小於IPIDEA,防封策略側重基礎輪轉 |
|
六、協議+集成能力 |
支持HTTP/HTTPS/Socks5,提供7種主流語言代碼示例,無縫對接AI/ML工作流與數據基礎設施 |
支持HTTP/HTTPS/Socks5,無多語言代碼示例,僅提供基礎代理服務 |
支持HTTP/HTTPS/Socks5,無多語言代碼示例,代理與輔助工具為合作模式 |
|
七、合規性 |
符合GDPR、CCPA等國際數據保護法規,SSL/TLS加密+DDoS防護+24/7安全監控 |
僅提及“運營商正規授權”,無國際合規認證,安全防護體系未明確 |
未提及國際合規認證,僅強調IP正規授權,安全防護體系較基礎 |
|
八、附加價值 |
配套抓取API\瀏覽器\抓取器等工具,服務阿里巴巴、華為等頭部企業,支持多買多送 |
僅提供代理服務,無配套工具,缺乏知名企業背書 |
合作工具類平台,缺乏頭部企業背書,優惠以折扣為主 |
可以看到,IPIDEA作為企業級代理IP服務提供商,其優勢在IP資源規模、覆蓋精度、穩定性、場景適配性、合規性等核心維度均顯著領先於同類平台。以1億+真實IP資源、220+國家四級定位構建全球網絡覆蓋,以99.9%穩定性+8類細分代理類型適配全場景需求,以國際合規認證+一體化工具降低企業使用成本與風險,最終形成“代理IP+數據採集+安全保障”的閉環服務。
顯然,這就是契合我需求的那個“它”。所以,最後我選擇了IPIDEA來作為我構建智能體知識庫的代理IP。
4. IPIDEA的使用流程和性能測試
接下來,我將介紹如何使用IPIDEA的服務來實現代理IP的提取,並對其提供的代理服務的實際表現進行一系列測試。
4.1 使用流程
4.1.1 賬號註冊登錄
介紹使用代理IP前的準備工作,包括註冊賬號、獲取API密鑰、選擇合適套餐、安裝必要Python庫等。
首先訪問IPIDEA官網,點擊右上角註冊按鈕,完成註冊登錄。
4.1.2 配置IP白名單
IP白名單僅允許預先添加信任的IP調用代理資源,可有效防止賬號盜用導致的代理資源浪費,鎖定合規使用的網絡範圍。
如上圖所示,賬號註冊成功之後,點擊右上角用户名,進入概覽頁。然後依次找到動態代理->動態住宅代理->代理使用->IP白名單管理->添加白名單,然後輸入自己當前的IP地址和備註信息,點擊確認按鈕即可。
4.1.3 配置提取API地址
在概覽頁,依次找到動態代理->動態住宅代理->API提取。如下圖配置之後,點擊生成鏈接,然後右側複製按鈕複製備用。
比如,我得到的是:[http://api.proxy.ipidea.io/getBalanceProxyIp?num=1&return_type=json&lb=1&sb=0&flow=1®inotallow=&protocol=http](http://api.proxy.ipidea.io/getBalanceProxyIp?num=1&return_type=json&lb=1&sb=0&flow=1®inotallow=&protocol=http)。
4.2 性能測試
代理IP的提取API地址已經有了,接下來我們對其進行一列測試驗證,確保它能夠符合的數據採集需求。
4.2.1 提取API有效性驗證
接下來,我們寫一段python代碼,來驗證API的有效性,確保我們能夠順利提取到代理IP,同時驗證我們的白名單IP配置是否生效:
import requests
import json
def fetch_proxy_IP():
api_url = "http://api.proxy.IPIDEA.io/getBalanceProxyIP?num=1&return_type=json&lb=1&sb=0&flow=1®ions=&protocol=http"
try:
# 發送 GET 請求(超時設置為 10 秒,避免無限等待)
response = requests.get(api_url, timeout=10)
# 檢查 HTTP 響應狀態碼(200 表示請求成功)
if response.status_code != 200:
print(f"請求失敗,HTTP 狀態碼:{response.status_code}")
return
# 解析 JSON 響應(將字符串轉為 Python 字典)
result = response.json()
print("完整響應數據:")
print(json.dumps(result, indent=2, ensure_ascii=False)) # 格式化打印
print("-" * 50)
# 根據響應狀態判斷是否獲取代理IP成功
if result.get("success") is True and result.get("code") == 0:
# 提取代理IP和端口
proxy_data = result.get("data", [])
if proxy_data:
proxy_IP = proxy_data[0].get("IP")
proxy_port = proxy_data[0].get("port")
request_IP = result.get("request_IP")
print("請求成功!")
print(f"你的請求 IP:{request_IP}")
print(f"獲取到的代理 IP:{proxy_IP}")
print(f"代理端口:{proxy_port}")
print(f"可用代理格式(HTTP):http://{proxy_IP}:{proxy_port}")
else:
print("響應中未包含代理IP數據")
else:
# 打印失敗原因
error_msg = result.get("msg", "未知錯誤")
error_code = result.get("code", "未知狀態碼")
print(f"獲取代理IP失敗!")
print(f"錯誤碼:{error_code}")
print(f"錯誤信息:{error_msg}")
# 針對 113 錯誤碼給出明確提示(你之前遇到的白名單問題)
if error_code == 113:
print("提示:請登錄 ipidea 後台,將你的請求 IP(result 中的 request_IP)添加到白名單後重試")
except requests.exceptions.Timeout:
print("錯誤:請求超時(超過 10 秒未響應)")
except requests.exceptions.ConnectionError:
print("錯誤:網絡連接失敗(可能是 API 地址不可達或網絡問題)")
except json.JSONDecodeError:
print("錯誤:響應數據不是有效的 JSON 格式")
except Exception as e:
print(f"未知錯誤:{str(e)}")
if __name__ == "__main__":
# 執行函數
fetch_proxy_IP()
如下圖所示,是測試程序的運行截圖。順利的請求到了1個代理IP,API驗證有效。
4.2.2 代理IP提取成功率
代理IP提取成功率是指從代理IP池中成功獲取有效IP地址的比例,是衡量代理服務質量和可靠性的關鍵指標。高成功率有助於用户快速獲得高質量的代理IP資源,確保數據採集順利進行。
循環提取100次,每次提取1個代理IP,根據成功提取到的數量來計算提取成功率。如下是核心代碼:
def test_1_extraction_success_rate():
"""測試1: 代理IP提取成功率"""
print("=" * 60)
print("測試1: 代理IP提取成功率 (測試100次)")
print("=" * 60)
successful_extractions = 0
failed_extractions = 0
errors = {}
for i in range(100):
proxy_data, _, error = extract_proxy()
if proxy_data and error is None:
successful_extractions += 1
print(f"第{i+1:3d}次提取: 成功 - {proxy_data['data'][0]['IP']}:{proxy_data['data'][0]['port']}")
else:
failed_extractions += 1
errors[error] = errors.get(error, 0) + 1
print(f"第{i+1:3d}次提取: 失敗 - {error}")
time.sleep(1)
success_rate = (successful_extractions / 100) * 100
print(f"\n提取成功率: {success_rate:.2f}% ({successful_extractions}/100)")
print("錯誤統計:")
for error, count in errors.items():
print(f" {error}: {count}次")
return success_rate, errors
運行截圖如下所示:
測試結果:提取成功率100%,這與官方所聲明的99.9%正常運行時間標準相符。
4.2.3 代理IP提取時間
代理IP提取時間是從代理服務器池獲取可用IP所需的時間。這對於需要多次更換IP以應對網絡管理、提高抓取效率或保護隱私的應用非常重要,能縮短數據採集時間並提升用户體驗。
循環提取100次,每次提取1個代理IP,然後根據成功提取到的數量來計算提取時間。如下是核心代碼:
def test_2_extraction_time():
"""測試2: 代理IP提取時間"""
print("\n" + "=" * 60)
print("測試2: 代理IP提取時間 (測試100次)")
print("=" * 60)
extraction_times = []
for i in range(100):
_, extraction_time, error = extract_proxy()
extraction_times.append(extraction_time)
if error is None:
print(f"第{i+1:3d}次提取時間: {extraction_time:.3f}秒")
else:
print(f"第{i+1:3d}次提取時間: {extraction_time:.3f}秒 (提取失敗: {error})")
avg_time = statistics.mean(extraction_times)
min_time = min(extraction_times)
max_time = max(extraction_times)
median_time = statistics.median(extraction_times)
print(f"\n提取時間統計:")
print(f" 平均時間: {avg_time:.3f}秒")
print(f" 最短時間: {min_time:.3f}秒")
print(f" 最長時間: {max_time:.3f}秒")
print(f" 中位時間: {median_time:.3f}秒")
return
運行截圖如下所示:
測試結果:平均時間不足1秒,這個耗時在代理IP提取領域屬於中上水平,還是非常優秀的。
|
最短時間 |
中位時間 |
最長時間 |
平均時間 |
|
0.709秒 |
0.915秒 |
1.226秒 |
0.931秒 |
4.2.4 代理IP連通成功率
代理IP連通成功率,是指代理服務器的網絡連通性。該指標對需使用代理IP進行網絡訪問的用户至關重要,直接影響工作效率與服務質量。連通成功率低會導致數據採集多次失敗、重試次數增加,消耗更多時間和資源。
循環提取100次,每次提取1個代理IP,然後向其發起tcp連接,根據連接成功的數量來計算連通成功率。如下是核心代碼:
def test_proxy_connectivity(IP: str, port: int) -> Tuple[bool, float, Optional[str]]:
"""測試代理連通性"""
start_time = time.time()
try:
# 測試TCP連接
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(10)
sock.connect((IP, port))
connect_time = time.time() - start_time
sock.close()
return True, connect_time, None
except Exception as e:
connect_time = time.time() - start_time
error_msg = f"連接失敗: {str(e)}"
return False, connect_time, error_msg
def test_3_connection_success_rate():
"""測試3: 代理IP連通成功率"""
print("\n" + "=" * 60)
print("測試3: 代理IP連通成功率 (測試100次)")
print("=" * 60)
successful_connections = 0
failed_connections = 0
connection_errors = {}
for i in range(100):
proxy_data, _, extract_error = extract_proxy()
if proxy_data and extract_error is None:
IP= proxy_data["data"][0]["IP"]
port = proxy_data["data"][0]["port"]
success, _, conn_error = test_proxy_connectivity(IP, port)
if success:
successful_connections += 1
print(f"第{i+1:3d}次連接: {IP}:{port} - 成功")
else:
failed_connections += 1
connection_errors[conn_error] = connection_errors.get(conn_error, 0) + 1
print(f"第{i+1:3d}次連接: {IP}:{port} - 失敗 ({conn_error})")
else:
failed_connections += 1
error_msg = f"提取失敗: {extract_error}"
connection_errors[error_msg] = connection_errors.get(error_msg, 0) + 1
print(f"第{i+1:3d}次連接: 提取失敗 - {extract_error}")
success_rate = (successful_connections / 100) * 100
print(f"\n連通成功率: {success_rate:.2f}% ({successful_connections}/100)")
print("連接錯誤統計:")
for error, count in connection_errors.items():
print(f" {error}: {count}次")
return success_rate, connection_errors
運行截圖如下:
測試結果:提取成功率100%,這與官方所聲明的99.9%正常運行時間標準相符。
4.2.5 代理IP響應時間
代理IP響應時間是指從客户端發送請求到通過代理服務器接收到響應數據之間的時間間隔。這個時間間隔可以用來衡量代理服務器的性能和效率。與代理IP連接時間類似,一般來説,響應時間越短,説明代理服務器處理請求的速度越快,用户體驗也就越好。
循環提取100次,每次提取1個代理IP,然後使用它來請求baidu,計算請求成功的時間。如下是核心代碼:
def test_proxy_response_time(IP: str, port: int) -> Tuple[Optional[float], float]:
"""測試代理響應時間"""
start_time = time.time()
proxy = f"http://{IP}:{port}"
proxies = {"http": proxy, "https": proxy}
test_url = "http://www.baidu.com"
try:
response = requests.get(test_url, proxies=proxies, timeout=15)
if response.status_code == 200:
response_time = time.time() - start_time
return response_time, response_time
except Exception as e:
pass
total_time = time.time() - start_time
return None, total_time
def test_5_response_time():
"""測試5: 代理IP響應時間"""
print("\n" + "=" * 60)
print("測試5: 代理IP響應時間 (測試100次)")
print("=" * 60)
response_times = []
for i in range(100):
proxy_data, _, extract_error = extract_proxy()
if proxy_data and extract_error is None:
IP= proxy_data["data"][0]["IP"]
port = proxy_data["data"][0]["port"]
resp_time, _ = test_proxy_response_time(IP, port)
if resp_time is not None:
response_times.append(resp_time)
print(f"第{i+1:3d}次響應時間: {resp_time:.3f}秒 - {IP}:{port}")
else:
print(f"第{i+1:3d}次響應時間: 測試失敗 - {IP}:{port}")
else:
print(f"第{i+1:3d}次: 提取失敗,無法測試響應時間")
if response_times:
avg_time = statistics.mean(response_times)
min_time = min(response_times)
max_time = max(response_times)
median_time = statistics.median(response_times)
print(f"\n響應時間統計:")
print(f" 平均時間: {avg_time:.3f}秒")
print(f" 最短時間: {min_time:.3f}秒")
print(f" 最長時間: {max_time:.3f}秒")
print(f" 中位時間: {median_time:.3f}秒")
else:
print("\n沒有成功的響應時間測試,無法計算響應時間統計")
return response_times
運行截圖:
測試結果:平均時間2.467秒,除了因為我這裏網絡波動引起的個別異常值外,整體耗時還是比較優秀的。
|
最短時間 |
中位時間 |
最長時間 |
平均時間 |
|
0.415秒 |
2.145秒 |
12.263秒 |
2.467秒 |
5. 使用動態住宅代理IP爬取維基百科數據
為了簡化爬蟲代碼,提高效率,我們直接訪問維基百科的純文本頁面。在任意維基百科頁面URL後加?actinotallow=raw,即可跳轉純文本版本。比如:[<font style="color:rgb(0, 0, 0);">https://zh.wikipedia.org/wiki/3%E6%9C%8830%E6%97%A5?actinotallow=raw</font>](https://zh.wikipedia.org/wiki/3%E6%9C%8830%E6%97%A5?actinotallow=raw):
接下來就簡單了,我們只需要對日期進行拼接,然後請求一個代理IP來抓取,然後將抓取到的數據,清洗整理之後,保存到文本文件中即可。完整代碼如下所示:
import requests
import re
from datetime import datetime
import time
from typing import List, Dict, Optional, Tuple
# 配置參數
PROXY_API_URL = "http://api.proxy.IPIDEA.io/getBalanceProxyIP?num=1&return_type=json&lb=1&sb=0&flow=1®ions=us&protocol=http"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
}
BASE_WIKI_URL = "https://zh.wikIPedia.org/wiki/{month}月{day}日?action=raw"
PROXY_RETRY_TIMES = 3
WIKI_RETRY_TIMES = 2
DELAY = 1
OUTPUT_FILE = "維基百科所有日期歷史信息_代理版_帶月日.txt"
def get_proxy_IP() -> Optional[Dict[str, str]]:
"""獲取代理IP,返回格式:{'http': 'http://IP:port', 'https': 'http://IP:port'}"""
for retry in range(PROXY_RETRY_TIMES):
try:
response = requests.get(PROXY_API_URL, timeout=10)
response.raise_for_status()
proxy_result = response.json()
if proxy_result.get("success") is True and proxy_result.get("code") == 0:
proxy_data = proxy_result.get("data", [])
if proxy_data:
IP= proxy_data[0].get("IP")
port = proxy_data[0].get("port")
ifIPand port:
proxy = {
"http": f"http://{IP}:{port}"
}
print(f"✅ 成功獲取代理IP:{IP}:{port}")
return proxy
print("❌ 代理接口返回數據為空")
else:
error_msg = proxy_result.get("msg", "未知錯誤")
error_code = proxy_result.get("code", "未知狀態碼")
print(f"❌ 獲取代理IP失敗(錯誤碼:{error_code},信息:{error_msg})")
if error_code == 113:
print("⚠️ 提示:請登錄ipidea後台,將當前請求IP添加到白名單")
except requests.exceptions.Timeout:
print(f"⚠️ 獲取代理IP超時,第{retry+1}/{PROXY_RETRY_TIMES}次重試...")
except requests.exceptions.ConnectionError:
print(f"⚠️ 獲取代理IP網絡連接失敗,第{retry+1}/{PROXY_RETRY_TIMES}次重試...")
except Exception as e:
print(f"⚠️ 獲取代理IP異常:{str(e)},第{retry+1}/{PROXY_RETRY_TIMES}次重試...")
time.sleep(DELAY)
print("❌ 多次獲取代理IP失敗")
return None
def get_max_days(month: int, year: int = 2024) -> int:
"""獲取指定月份的最大天數(2024是閏年,適配2月29天)"""
if month == 2:
return 29 if (year % 4 == 0 and year % 100 != 0) or (year % 400 == 0) else 28
elif month in [4, 6, 9, 11]:
return 30
else:
return 31
def generate_all_dates() -> List[Dict[str, int]]:
"""生成全年所有日期(月/日)"""
all_dates = []
for month in range(1, 13):
max_days = get_max_days(month)
for day in range(1, max_days + 1):
all_dates.append({"month": month, "day": day})
return all_dates
def fetch_wikitext_with_proxy(month: int, day: int) -> Optional[str]:
"""使用代理IP請求指定日期的維基百科原始wikitext內容"""
url = BASE_WIKI_URL.format(month=month, day=day)
print(f"\n📥 正在請求 {month}月{day}日 維基內容(使用代理)")
for retry in range(WIKI_RETRY_TIMES):
proxy = get_proxy_IP()
if not proxy:
print(f"⚠️ 第{retry+1}/{WIKI_RETRY_TIMES}次重試:獲取代理失敗,跳過該日期")
return None
try:
response = requests.get(
url,
headers=HEADERS,
proxies=proxy,
timeout=20,
verify=False
)
response.raise_for_status()
print(f"✅ {month}月{day}日 維基內容請求成功")
return response.text
except requests.exceptions.ProxyError:
print(f"⚠️ 第{retry+1}/{WIKI_RETRY_TIMES}次重試:代理失效或不可用")
except requests.exceptions.Timeout:
print(f"⚠️ 第{retry+1}/{WIKI_RETRY_TIMES}次重試:代理請求超時")
except requests.exceptions.ConnectionError:
print(f"⚠️ 第{retry+1}/{WIKI_RETRY_TIMES}次重試:代理網絡連接失敗")
except requests.exceptions.HTTPError as e:
print(f"❌ 第{retry+1}/{WIKI_RETRY_TIMES}次重試:請求失敗({e})")
except Exception as e:
print(f"⚠️ 第{retry+1}/{WIKI_RETRY_TIMES}次重試:未知異常 {str(e)}")
time.sleep(DELAY * 2)
print(f"❌ {month}月{day}日 多次代理請求失敗,跳過該日期")
return None
def add_month_day_to_year(item: str, month: int, day: int) -> str:
"""在年份後添加月日(支持公元/公元前、1-4位數年份)"""
# 正則匹配規則:支持 公元前xxxx年、xxxx年、xx年 格式,匹配後在年份後插入 月日
# 分組説明:\1=前綴(公元前)、\2=年份數字、\3=年字、\4=冒號及後面內容
pattern = r"(^(前|公元前)?(\d{1,4})(年)(:|:))(.*)"
match = re.match(pattern, item, re.UNICODE)
if match:
# 提取各部分並拼接:前綴 + 年份 + 年 + 月日 + 冒號 + 內容
prefix = match.group(1) # 包含"公元前"(如有)、年份、"年"、冒號
content = match.group(6) # 冒號後的內容
# 插入月日(格式:x月x日)
new_item = f"{match.group(2) if match.group(2) else ''}{match.group(3)}年{month}月{day}日{match.group(5)}{content}"
return new_item
else:
# 無明確年份格式的記錄,保持原樣
return item
def parse_wikitext(wikitext: str, month: int, day: int) -> Dict[str, List[str]]:
"""解析wikitext,提取大事記、出生、逝世信息,在年份後補充月日"""
section_pattern = r"==\s*(%s)\s*==\n([\s\S]*?)(?=\n==\s*[\u4e00-\u9fa5]+\s*==|\Z)"
target_sections = ["大事記", "出生", "逝世"]
result = {section: [] for section in target_sections}
for section in target_sections:
match = re.search(section_pattern % section, wikitext, re.IGNORECASE)
if not match:
result[section] = ["無相關記錄"]
continue
content = match.group(2).strIP()
if not content:
result[section] = ["無相關記錄"]
continue
# 基礎清洗
content = re.sub(r"\{\{[^\}]+\}\}", "", content)
content = re.sub(r"\[\[([^\]|]+)\|([^\]]+)\]\]", r"\2", content)
content = re.sub(r"\[\[([^\]]+)\]\]", r"\1", content)
content = re.sub(r"<ref[^>]*>[\s\S]*?</ref>", "", content)
content = re.sub(r"\n+", "\n", content)
items = re.findall(r"^\* (.+)$", content, re.MULTILINE)
# 核心步驟:為每條記錄的年份後添加月日
processed_items = []
for item in items:
processed_item = add_month_day_to_year(item, month, day)
processed_items.append(processed_item)
result[section] = processed_items if processed_items else ["無相關記錄"]
return result
def save_result(date_info: Dict[str, int], parsed_data: Dict[str, List[str]], file):
"""將解析結果寫入文件"""
month, day = date_info["month"], date_info["day"]
# 注意:這裏要和target_sections對應
for section in ["大事記", "出生", "逝世"]:
for idx, item in enumerate(parsed_data[section], 1):
file.write(f"{item}\n")
def main():
"""主函數:生成日期→獲取代理→代理請求維基→解析→保存"""
requests.packages.urllib3.disable_warnings()
all_dates = generate_all_dates()
total_dates = len(all_dates)
print(f"📅 共生成 {total_dates} 個日期,開始遍歷(每個日期將使用獨立代理IP)...")
print(f"⚠️ 注意:若持續獲取代理失敗,請檢查IP白名單配置\n")
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
for idx, date in enumerate(all_dates, 1):
month, day = date["month"], date["day"]
print(f"\n{'='*60}")
print(f"[{idx}/{total_dates}] 正在處理:{month}月{day}日")
print(f"{'='*60}")
wikitext = fetch_wikitext_with_proxy(month, day)
if not wikitext:
continue
parsed_data = parse_wikitext(wikitext, month, day)
save_result(date, parsed_data, f)
print(f"✅ {month}月{day}日 信息已保存")
print(f"\n🎉 所有日期處理完成!結果已保存至:{OUTPUT_FILE}")
print(f"⚠️ 部分日期可能因代理失效或網絡問題未獲取到內容,可查看控制枱日誌")
if __name__ == "__main__":
main()
爬取每個日期的事件前,先提取一個代理IP。然後使用代理IP去抓取,抓取到內容後進行內容清洗,保存到文件:維基百科所有日期歷史信息_代理版_帶月日.txt中,依次循環,直到抓取完365/366個日期的所有事件。
如下所示是代碼運行截圖,成功的將每個日期的大事件記錄抓取、清洗並保存了下來。
6. 智能體效果驗證
智能體的最終回答質量,從來都離不開優質數據源、高效採集能力與完整數據支撐的三重保障——而IPIDEA代理IP的核心功能與強悍性能,正是打通這一鏈路的關鍵紐帶。它絕非單純的“工具”,更像是全程賦能的“得力助手”:不僅在數據採集環節掃清障礙,更將自身優勢直接轉化為智能體回答質量的跨越式提升,讓知識庫的價值真正落地見效。
6.1 知識庫展示
如下圖所示,是我抓取清洗後的內容。這就是一份可以用來構建知識庫的優質數據集了。
經過分段解析後,會得到下圖所示的一個知識條目,適配智能體的知識庫檢索調用邏輯,讓大模型能直接精準抓取所需信息。
6.2 智能體表現對比
接下來,我們通過2個簡單的問題,來對比使用知識庫前後的智能體表現差異。
|
問題 |
無知識庫 |
有知識庫 |
|
1945年1月9日有哪些重大事件發生? |
|
|
|
1月1日發生過什麼重要事件? |
|
|
根據對第一個問題的回答進行對比分析後發現,引入知識庫顯著提升了AI系統的知識容量。從第二個問題的回答對比中可以看出,在知識庫的支持下,AI的回答更加貼合“歷史大事記”智能體的特定調性,其重點集中於歷史事件而非近期發生的事件上,從而更接近預期設定的標準。
7. 結論
回頭看“歷史大事記”智能體的搭建過程,最核心的感悟是:智能體的回答質量,本質是知識庫的“數據質量”決定的。初期依賴大模型原生知識庫,之所以出現“缺斤短兩”“描述模糊”的問題,核心就是缺乏結構化、全覆蓋、高權威的專屬數據支撐--而維基百科作為全球公認的優質數據源,恰好彌補了這一短板,但“國內無法直接訪問”“海量數據需自動化採集”的現實難題,又讓代理IP成為了不可或缺的“橋樑”。
在整個數據採集環節中,IPIDEA代理IP的選擇堪稱“關鍵一步”。它的億級純淨代理IP池解決了“代理失效、反爬機制”的顧慮,220+國家/地區的四級精準定位解決了網絡地域問題,99.9%的穩定性和多類型代理適配,確保了365/366個日期、數十萬條歷史數據的高效抓取。對比同類平台,其“真實IP佔比高、場景適配全、合規有保障”的優勢,不僅讓我避開了“劣質IP導致業務中斷”“定位不準無法訪問目標資源”等問題,更通過多語言代碼示例,降低了開發成本,最終實現了“採集-清洗-入庫”的全流程順暢落地。
最終,通過“維基百科數據源+IPIDEA代理IP+Python爬蟲”的組合,我成功構建了一份覆蓋全年日期、包含大事記、出生、逝世三大核心維度的高質量知識庫--不僅數據完整度遠超大模型原生知識庫,還通過“年份+月日”的格式優化,讓智能體能夠精準定位歷史事件,徹底解決了“回答模糊、遺漏關鍵信息”的問題。