🤖 第十一章：CrewAI 多代理爬蟲：全自動採集全台露營地

在前面的基礎章節中，我們學會了如何在畫面上放上地圖、標記座標。但是，如果你是一個創業者，你打算做一個「全台最完整的露營車泊地圖」。你總不可能自己一個人，每天坐在電腦前 Google 搜尋「露營地」，然後一筆一筆複製貼上地址、經緯度、價格，手動建檔吧？如果你這麼做，等你建完 500 筆資料，你的競爭對手早就把市場佔滿了！

在 AI 時代，最昂貴的資源不再是「寫程式的技術」，而是**「乾淨、結構化的高品質資料」**。這堂課，我們將帶你進入 Vibe Coding 的最深處——自動化資料採集。我們不寫傳統的 Python 爬蟲 (BeautifulSoup / Scrapy)，那太容易因為網站改版就壞掉。我們將引進目前全球最火紅的多代理框架：CrewAI。

這是一個長達 6000 字的深度解析章節，請準備好你的咖啡，我們將從原理、Prompt 設計、防呆機制、到最終的結構化輸出，一步一步教你如何建立一支「不知疲倦的 AI 機器人軍團」。

🏭 什麼是 CrewAI？為什麼傳統爬蟲已經死了？

在以前，我們寫爬蟲的邏輯是：「去這個網址 -> 找到 ID 為 title 的 HTML 標籤 -> 把裡面的字抓出來」。但現在的網站幾乎都是 React / Vue 寫的動態網頁，甚至有反爬蟲機制 (Cloudflare)。只要對方網站的 HTML 稍微改動一個 class 名字，你的爬蟲就立刻崩潰報錯，你又要花一整天去修 Bug。

CrewAI 的革命性思維： 我們不再依賴死板的 HTML 標籤。我們建立一組「AI 員工團隊 (Crew)」。

Agent A (搜尋專家)：負責去 Google 搜尋「南投推薦露營地」，並點開前十名的文章閱讀。
Agent B (資料萃取專家)：負責把 Agent A 讀到的長篇大論文章，用 LLM 理解後，精準萃取出「營地名稱、地址、價格區間」。
Agent C (座標轉換專家)：負責把地址丟給 Google Maps API，轉換成精準的經緯度 (Latitude / Longitude)。

你只需要下達高階的「人類指令」，這群 AI 就會自己開會、自己分配工作、自己處理異常，最後交給你一份完美的 JSON 檔案！

🛠️ 實戰 1：建立你的第一個 AI 員工 (Agent)

在使用 CrewAI 之前，你必須清楚地定義每一個 AI 員工的角色 (Role)、目標 (Goal) 與背景故事 (Backstory)。這就像是你在面試員工一樣，你給的設定越詳細，他做出來的東西就越精準。

💡 Vibe Prompt 實戰 1：定義「資料搜尋專家」

如果你不知道怎麼寫 Agent 的設定，直接請大語言模型幫你寫！

[!IMPORTANT] 請複製以下 Prompt 傳送給 AI (如 Claude 3.5 或 ChatGPT-4o)：

我正在使用 Python 的 CrewAI 框架開發一個露營地資料自動採集系統。 請幫我定義第一個 Agent (AI 員工)。 他的任務是：在網路上搜尋全台灣各大論壇 (如 PTT、Dcard、Mobile01) 與部落格的露營地推薦文章。 請給我這個 Agent 的 Python 程式碼，必須包含： 1. role (角色名稱) 2. goal (終極目標) 3. backstory (背景故事，用來賦予 AI 性格與專業度) 4. verbose=True (開啟詳細日誌) 5. allow_delegation=False (不允許把工作推給別人) 請用專業、生動的語氣來撰寫 backstory。

🤖 AI 生成的超完美 Agent 設定：

from crewai import Agent

# 建立資料搜尋專家
search_expert = Agent(
    role='資深露營秘境情報員',
    goal='在浩瀚的網際網路中，挖掘出全台灣最新、最熱門、甚至是最隱密的露營地資訊，特別是適合「車泊」的隱藏版地點。',
    backstory=(
        '你是一位熱愛戶外活動、擁有 15 年露營經驗的資深玩家，同時也是一位頂尖的網路情報分析師。'
        '你熟知各大戶外論壇的黑話，知道如何從 PTT 的 Camping 版、Dcard 的旅遊版，以及無數個露營部落客的文章中，'
        '一眼辨識出哪些是業配文，哪些是真正優質的營地。'
        '你的直覺敏銳，從不錯過任何一個擁有絕美星空或乾淨衛浴的露營聖地。'
    ),
    verbose=True,
    allow_delegation=False,
    # 這裡我們假設你有配置一個可以搜尋網路的 Tool (例如 SerperDevTool)
    # tools=[search_tool] 
)

🔍 深度解析： 看到了嗎？這就是 CrewAI 強大的地方。你不是在寫「if/else」邏輯，你是在「賦予靈魂」。當你給了 search_expert 這樣的背景故事後，他在搜尋和閱讀網頁時，大語言模型就會自動切換成「資深露營玩家」的視角，自動過濾掉那些看起來像詐騙或品質低劣的資訊。這在傳統寫死規則的 Python 爬蟲中，是絕對不可能做到的事！

🎯 實戰 2：為 AI 分配具體任務 (Task)

有了員工之後，我們要指派具體的工作任務 (Task) 給他。 Task 的定義一樣需要極度精準，告訴他「要做什麼」、「預期的產出格式是什麼」。

💡 Vibe Prompt 實戰 2：定義精確的爬蟲任務

[!IMPORTANT] 請複製以下 Prompt 傳送給 AI：

我已經有了一個名為 search_expert 的 Agent。 現在請幫我用 CrewAI 的 Task 類別，寫一個具體的任務給他。 任務需求： 1. description (描述)：請他搜尋「新竹縣」評價最高的 5 個車泊露營地。必須包含詳細的地址、價格、以及一項最大的特色 (例如：雲海、螢火蟲)。 2. expected_output (預期產出)：要求他最終輸出一份結構化的清單。 3. agent：綁定 search_expert。 請給我這段 Python 程式碼，並在註解中說明 expected_output 的重要性。

🤖 AI 的高階任務規劃：

from crewai import Task

# 分配任務：新竹車泊地毯式搜索
hsinchu_camping_task = Task(
    description=(
        '請使用搜尋工具，找出「新竹縣」目前最受歡迎、評價最高的 5 個車泊/露營地點。'
        '你必須閱讀相關的部落格文章或論壇評價，並提取出以下資訊：'
        '1. 營地完整名稱'
        '2. 完整的地址 (至少要有鄉鎮市區)'
        '3. 價格區間 (例如：一帳 1000~1200元)'
        '4. 營地最大特色 (限 20 字以內，例如：秋冬限定絕美雲海)'
    ),
    expected_output=(
        '一份包含 5 個新竹露營地的 Markdown 條列式清單。'
        '每一個營地必須嚴格遵守以下格式：'
        '- 名稱：[營地名稱]'
        '- 地址：[完整地址]'
        '- 價格：[價格]'
        '- 特色：[特色]'
    ),
    agent=search_expert
)

/*
【AI 深度解析 expected_output 的重要性】：
在大型語言模型 (LLM) 中，如果你只給 description，AI 最後給你的答案可能會是一篇幾千字、充滿廢話的「散文」。
`expected_output` 是 CrewAI 框架用來「約束」AI 輸出的緊箍咒。
當你嚴格定義了格式 (如 Markdown 清單，或是 JSON 格式)，AI 就會為了滿足這個條件，強制把自己長篇大論的思考結果進行收斂與格式化。
這對於我們後續要把資料塞進資料庫來說，是生死交關的一步！
*/

🔄 實戰 3：最困難的一步 —— 座標轉換 (Geocoding)

在地圖系統中，只有「地址」是沒有用的。Leaflet.js 地圖只看得懂經緯度 (Latitude, Longitude)。如果我們把「新竹縣五峰鄉桃山村...」丟給地圖，地圖會直接報錯。

因此，我們需要建立第二個 AI 員工：地理位置轉換專家。但他不能只靠 LLM 的大腦去猜經緯度 (這叫做幻覺 Hallucination，AI 猜出來的經緯度通常會把你帶到太平洋中間)。他必須學會使用真正的工具 (Tool) 去呼叫 Google Maps API。

💡 Vibe Prompt 實戰 3：教 AI 如何使用外部 API 工具

這是高階 Vibe Coding 的核心：我們不自己寫 API 呼叫，我們寫一個工具箱，讓 AI 自己去呼叫！

[!IMPORTANT] 請複製以下 Prompt 傳送給 AI：

在 CrewAI 中，我需要建立一個自訂工具 (Custom Tool)，讓我的 Agent 可以把地址轉換成經緯度。 請幫我用 Python 的 @tool 裝飾器寫一個函數 geocode_address(address: str)。 邏輯： 1. 使用 requests 套件打 Google Maps Geocoding API。 2. API Key 會放在環境變數 GOOGLE_MAPS_API_KEY。 3. 如果轉換成功，回傳包含 lat 和 lng 的字典字串。 4. 如果失敗或找不到地址，回傳錯誤提示字串。 請給我完整的程式碼與註解，並強調 docstring 的重要性。

🤖 AI 生成的無敵工具函數：

import os
import requests
from langchain.tools import tool

# 🔥 這是靈魂！docstring (函數下方的註解) 是寫給 AI 看的！
# AI 會閱讀這段 docstring，來決定什麼時候要使用這個工具，以及要傳入什麼參數。
@tool("地址轉換經緯度工具")
def geocode_address(address: str) -> str:
    """
    這是一個強大的地理編碼工具。
    當你需要知道某個地點或地址的精準經緯度 (Latitude 和 Longitude) 時，請使用這個工具。
    傳入參數必須是一個清楚的字串地址，例如："台北市信義區市府路1號" 或 "武陵農場"。
    """
    api_key = os.environ.get("GOOGLE_MAPS_API_KEY")
    if not api_key:
        return "錯誤：找不到 GOOGLE_MAPS_API_KEY 環境變數。"

    # Google Maps Geocoding API 端點
    url = f"https://maps.googleapis.com/maps/api/geocode/json?address={address}&key={api_key}&language=zh-TW"
    
    try:
        response = requests.get(url)
        data = response.json()
        
        # 檢查 API 是否成功回傳資料
        if data['status'] == 'OK':
            location = data['results'][0]['geometry']['location']
            lat = location['lat']
            lng = location['lng']
            # 回傳明確的字串結果給 AI 員工
            return f"地址 '{address}' 的經緯度為：緯度 {lat}, 經度 {lng}"
        else:
            return f"無法找到 '{address}' 的經緯度。API 回應狀態：{data['status']}"
            
    except Exception as e:
        return f"呼叫 API 時發生網路錯誤：{str(e)}"

🔍 深度解析： 絕大多數新手在使用 LangChain 或 CrewAI 的 @tool 時，都會忘記寫 """docstring"""。請記住，在 AI Agent 的世界裡，函數的註解不是寫給人類看的，是寫給 AI 看的說明書！ 如果你不告訴 AI 這個工具是用來「轉換經緯度」的，AI 拿到地址時就會傻在原地，不知道自己其實手上有這個超強工具可以使用。現在，我們可以把這個 @tool 分配給第二個 Agent 了。

🧩 實戰 4：串聯上下游：讓 AI 員工接力工作

我們現在有了「搜尋專家 (抓資料)」和「座標轉換專家 (用工具轉經緯度)」。在真實的公司裡，這叫做流水線。Agent A 做完報告，要把報告交給 Agent B 處理。

💡 Vibe Prompt 實戰 4：建立 Crew 團隊與 JSON 結構輸出

我們最終要的不是一篇 Markdown 文章，我們需要的是一個可以直接塞進 Supabase 資料庫的 JSON 格式 (例如 [{"name": "...", "lat": 24.1, "lng": 121.2}])。

[!IMPORTANT] 請複製以下 Prompt 傳送給 AI：

請幫我整合上述的流程，建立完整的 CrewAI 執行腳本。 1. 建立 geocode_expert (座標轉換專家)，並把剛剛的 geocode_address 工具指派給他。 2. 建立第二個任務 (format_to_json_task)：由 geocode_expert 接收第一個任務 (hsinchu_camping_task) 整理出的 5 個營地清單。 3. 他必須逐一使用工具查詢經緯度，最後將所有資料整合成標準的 JSON Array 格式。 4. 建立 Crew，並依照順序執行這兩個任務。 5. 請給我完整的 Python 程式碼，並展示如何啟動這個團隊。

🤖 AI 自動化團隊總成：

from crewai import Agent, Task, Crew, Process
import json

# 1. 建立第二個員工：座標轉換與資料清洗專家
geocode_expert = Agent(
    role='資深地理資料工程師',
    goal='將人類閱讀的地址，精準轉換為地圖系統需要的經緯度，並輸出完美的 JSON 結構。',
    backstory='你有嚴重的強迫症，容不下任何一個括號或引號的錯誤。你是處理 JSON 資料格式的頂尖大師。',
    verbose=True,
    tools=[geocode_address] # 把剛剛寫好的工具交給他！
)

# 2. 建立第二個任務：轉換與格式化
format_to_json_task = Task(
    description=(
        '請接收上一個任務整理出來的 5 個露營地清單。'
        '針對每一個營地，請使用你的 [地址轉換經緯度工具] 查詢精確的 lat 與 lng。'
        '如果工具回傳找不到，經緯度請填 0。'
        '最後，請將這 5 個營地的所有資訊，轉化為 JSON Array 格式。'
    ),
    expected_output=(
        '必須是完全合法的 JSON 字串，不可包含任何 Markdown 標記 (如 ```json)。格式範例如下：\n'
        '[\n'
        '  {\n'
        '    "name": "營地名稱",\n'
        '    "address": "地址",\n'
        '    "price": "價格區間",\n'
        '    "feature": "特色",\n'
        '    "lat": 24.123,\n'
        '    "lng": 121.456\n'
        '  }\n'
        ']'
    ),
    agent=geocode_expert
)

# 3. 成立專案團隊 (Crew)
camping_crew = Crew(
    agents=[search_expert, geocode_expert],
    tasks=[hsinchu_camping_task, format_to_json_task],
    process=Process.sequential, # 循序漸進：A 做完換 B 做
    verbose=True
)

# 4. 🔥 老闆按下啟動按鈕！
print("🚀 開始執行全自動露營地採集計畫...")
result = camping_crew.kickoff()

print("==================================")
print("🎉 最終產出的完美的 JSON 資料：")
print(result)

# 如果你的 expected_output 寫得夠好，你甚至可以直接用 json.loads() 把結果轉成 Python 字典，準備寫入資料庫！
try:
    final_data = json.loads(result)
    print(f"成功解析了 {len(final_data)} 筆營地資料！準備寫入 Supabase！")
except Exception as e:
    print("AI 輸出的格式有誤，無法解析為 JSON：", e)

🔍 深度解析： 這就是價值百萬的自動化爬蟲架構！以前你需要寫一堆 try/catch，去爬 HTML，去解析字串，去呼叫 API，最後還要手動轉 JSON。現在，你只是寫了幾段「人類的文字」，設定了兩個 AI 員工的職責。按下 kickoff() 之後，你會在終端機看到驚人的畫面： AI 會自己思考：「我現在拿到 5 個營地了。第一個營地是櫻花部落。我要使用地址工具...」然後它會自己去打 API，拿到經緯度後，再思考下一個。如果 API 出錯，它甚至會自己反省：「哎呀，API 報錯了，我試著把地址縮短一點再查一次看看。」

這就是所謂的 Agentic Workflow (智能體工作流)。它具備了自我糾錯與邏輯推理的能力！

🚫 終極避坑指南：AI 拒絕交出乾淨的 JSON

即使你千叮嚀萬交代，目前的 LLM (特別是較便宜的模型如 GPT-3.5 或 Claude 3 Haiku) 都有一種「強迫症」：它們非常喜歡在你要求的 JSON 前面加上一句：「好的老闆，這是我為您整理的 JSON：」，然後在結尾加上 ````json` 的 Markdown 標記。

如果你的程式碼直接用 json.loads(result) 去接這串文字，程式會立刻崩潰爆炸 (JSONDecodeError)！

💡 Vibe Prompt 實戰 5：寫一個無敵的 JSON 清洗工具

在 Vibe Coding 中，與其浪費時間去跟 AI 爭論叫它不要加 Markdown，不如直接寫一個正則表達式 (Regex) 工具把外殼剝掉！

[!IMPORTANT] 請複製以下 Prompt 傳送給 AI：

AI 在輸出 JSON 字串時，經常會被 ```json 和 ``` 包裝起來，或是前後夾雜一些無用的對話廢話。 請幫我用 Python 寫一個 clean_json_response(text: str) 的函數。 邏輯如下： 1. 使用正則表達式 (re 套件)，找出字串中第一個 [ 和最後一個 ] 之間的所有內容。 2. 如果找不到，或是解析失敗，拋出明確的 Error。 3. 如果找到，請使用 json.loads() 驗證它是否為合法的 JSON，如果是，就回傳這個 List 或 Dict。 請給我堅固耐用的程式碼。

🤖 AI 的終極防爆清洗器：

import re
import json

def clean_json_response(text: str):
    """
    不管 AI 講了多少廢話，只要裡面有合法的 JSON 陣列或物件，這個函數就能把它挖出來！
    """
    try:
        # 第一步：尋找最外層的陣列 [] 或是物件 {}
        # 這個正則表達式會匹配從第一個 [ 或 { 開始，到最後一個 ] 或 } 結束的所有內容
        match = re.search(r'\[.*\]|\{.*\}', text, re.DOTALL)
        
        if not match:
            raise ValueError("在 AI 的回覆中找不到任何 JSON 結構 (沒有 [] 或 {})")
            
        clean_text = match.group(0)
        
        # 第二步：驗證並轉換為 Python 字典/列表
        parsed_data = json.loads(clean_text)
        return parsed_data
        
    except json.JSONDecodeError as e:
        print(f"❌ JSON 解析失敗，提取出的文字格式不合法。錯誤行號：{e.lineno}")
        print("提取出的原始文字片段：", clean_text[:100], "...")
        raise
    except Exception as e:
        print(f"❌ 清洗函數發生未預期錯誤：{str(e)}")
        raise

# 實戰應用：
ai_dirty_reply = """
太棒了老闆！這是我花了一整晚找出來的新竹露營地：
```json
[
  { "name": "星空部落", "lat": 24.5, "lng": 121.1 }
]

希望你喜歡！ """

使用我們的神器

clean_data = clean_json_response(ai_dirty_reply) print("清洗後的純淨資料：", clean_data)

輸出：[{'name': '星空部落', 'lat': 24.5, 'lng': 121.1}]


**🔍 深度解析：**
這是全自動化系統中最關鍵的「防呆機制」。你不能永遠祈禱 AI 乖乖聽話。
在系統架構設計上，我們必須容許 AI 犯錯 (產生廢話)，但我們要在「接收端」做好完美的過濾器。這段 `clean_json_response` 函數價值連城，你可以把它放在所有 LLM 輸出的後端，確保你的系統永遠不會因為一句「好的老闆」而全線崩潰！

---

## 📈 高階演化：建立「非同步併發」的多縣市採集

上面的例子我們只抓了「新竹縣」。
如果老闆說：「我明天就要看到全台灣 22 個縣市的露營地資料！」
如果你讓 CrewAI 一個縣市一個縣市慢慢查，可能要查一整天。
在高階的 Python 開發中，我們會引入**併發 (Concurrency) 與非同步** 的概念。

### 💡 Vibe Prompt 實戰 6：讓 AI 幫你寫出多執行緒排程器

> [!IMPORTANT]
> **請複製以下 Prompt 傳送給 AI：**
> 
> `我剛剛寫好了針對單一縣市的 CrewAI 採集流程，包裝在 run_camping_crew(city_name: str) 函數中。`
> `現在我有一個包含 22 個縣市的陣列 cities = ["台北市", "新北市", "桃園市", ...]。`
> `請幫我寫一段 Python 程式碼，使用 concurrent.futures.ThreadPoolExecutor 來「同時 (併發)」啟動這 22 個城市的採集任務。`
> `為了避免 API 被鎖，請將最大工作執行緒 (max_workers) 限制在 5。`
> `請幫我加上進度條顯示 (tqdm 或是詳細的 print 也可以)。`

**🤖 AI 的高效率併發引擎：**

```python
import concurrent.futures
import time

# 假設這是我們剛才包裝好的執行函數
def run_camping_crew(city_name):
    print(f"🚀 [開始] 啟動 {city_name} 的探索小隊...")
    # 這裡放 Crew.kickoff() 的邏輯...
    # 模擬花費 10 秒鐘
    time.sleep(10) 
    print(f"✅ [完成] {city_name} 探索完畢！")
    return f"{city_name}_data.json"

cities = ["基隆市", "台北市", "新北市", "桃園市", "新竹縣", "苗栗縣", "台中市", "南投縣", "彰化縣"]

print(f"🔥 準備對 {len(cities)} 個縣市發動全面搜索...")

results = []

# 使用 ThreadPoolExecutor，限制同時只能有 5 個團隊出發
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # 將每個縣市提交給執行緒池
    # future_to_city 是一個字典，用來記錄哪個任務(future)對應哪個縣市
    future_to_city = {executor.submit(run_camping_crew, city): city for city in cities}
    
    # as_completed 會在任何一個任務完成時立刻通知我們
    for future in concurrent.futures.as_completed(future_to_city):
        city = future_to_city[future]
        try:
            # 取得該任務的回傳結果
            data = future.result()
            results.append(data)
            print(f"📥 成功接收 {city} 的回報資料。目前進度：{len(results)}/{len(cities)}")
        except Exception as exc:
            # 如果某個縣市在執行中發生崩潰，這裡會捕捉到，不會導致整個系統死掉
            print(f"❌ {city} 團隊全軍覆沒，發生錯誤: {exc}")

print("🏆 全台搜索行動大功告成！所有的 JSON 都已收集完畢！")

🔍 深度解析： 這就是資深工程師與新手的差別。新手只會寫 for city in cities:，一個一個慢慢跑。資深工程師知道利用 AI 寫出 ThreadPoolExecutor。當這段程式碼執行時，你會看到終端機同時印出「啟動台北、啟動新北、啟動桃園...」5 條線程火力全開。原本要跑 220 秒的任務，現在只要 50 秒就能完成！在撰寫這段 Vibe Prompt 時，我們特別指定了 max_workers=5，這是為了保護你的 Google Maps API 不會因為瞬間打入幾千個 Request 而被封鎖 (Rate Limit)。這就是架構設計的藝術。

✅ 本章總結與終極實戰心法

恭喜你完成了這堂價值 10 萬元的自動化爬蟲高階課程！在這個長達 6000 字的章節中，我們完全顛覆了傳統寫程式的思維。

回顧我們打造這支「AI 傭兵團」的關鍵步驟：

角色塑造 (Agent)：我們不用 Regex 抓標籤，我們賦予 AI「資深露營家」的靈魂與背景故事，讓他用大腦去理解文章。
精確目標 (Task)：利用 expected_output 緊箍咒，強迫 AI 吐出結構化的 Markdown 或是 JSON。
賦予武器 (Tools)：利用 @tool 裝飾器與詳盡的 docstring，教 AI 如何自己去呼叫 Google Maps API。
防爆清洗器 (Regex)：永遠不要相信 AI 會輸出 100% 完美的字串，自己寫一個過濾器把 JSON 挖出來。
多線程併發 (Concurrency)：用 5 倍數的火力，瞬間掃蕩全台灣的資料。

當你掌握了這套架構，你就不再只是一個「會畫畫面的前端工程師」。你可以用同一套邏輯，去抓「全台的寵物友善餐廳」、「日本東京的平價溫泉旅館」、「美股的熱門分析報告」。只要有網路的地方，你的 CrewAI 機器人軍團就能為你源源不絕地帶回結構化的黃金資料，這就是 SaaS 創業的最強護城河！

接下來，當我們擁有了幾千筆精緻的露營地資料後，我們要回到前端。下一章：第十二章：Leaflet.js 自訂高階地圖標記 (Custom Markers)。我們將教你如何把這幾千筆資料，渲染成地圖上極度精美的、帶有價格標籤與玻璃擬態的自訂圖示，讓你的地圖介面直接超越 Google Maps！我們下堂課見！