[Golang] 將 PTT 資料爬蟲改用 Firecrawl API 的研究與實作

將 PTT 資料爬蟲改用 Firecrawl API 的研究與實作

前提

photomgr 專案(https://github.com/kkdai/photomgr)一直以來都是爬取 PTT Beauty 板資料的得力工具,幫不少人抓取正妹板的文章和圖片。原本我們是用 Google Cloud Platform(GCP)直接連到 PTT 網站抓資料,簡單又順手。不過最近 PTT 把資料架到 Cloudflare 保護後,GCP 的連線常常被擋,可能是因為 Cloudflare 的防爬蟲機制(像是驗證碼或 IP 限制)太強大,導致原本的爬蟲程式完全 GG。為了讓 photomgr 繼續運作,我們研究了幾個替代方案,最後決定改用 Firecrawl API 來爬 PTT 的資料。這篇部落格會講解我們為什麼要做這個改變、怎麼改,以及改完後的程式碼和成果。

相關變動

PTT 最近開始用 Cloudflare 保護網站,加入了像是驗證碼、IP 限制等防爬蟲機制。這些措施讓我們原本用 GCP 直接送 HTTP 請求去抓資料的方式行不通,因為 Cloudflare 會把 GCP 的請求擋掉或限速,導致爬蟲常常失敗。經過一番研究,我們發現 Firecrawl 這個服務很適合解決這個問題。它不僅能繞過 Cloudflare 的防爬,還能把網頁內容轉成乾淨的 markdown 格式,方便我們解析。於是,我們決定把 photomgr 的爬蟲邏輯從原本的直接連線改成用 Firecrawl API 來抓 PTT Beauty 板的文章列表和內文。

什麼是 Firecrawl?

Firecrawl 是一個專為網頁爬蟲設計的 API 服務(https://www.firecrawl.dev/),可以幫你抓取網頁內容並轉成結構化的格式,比如 markdown 或 JSON。它最大的優勢是能處理像 Cloudflare 這種防爬蟲保護,還能模擬瀏覽器行為,抓到動態載入的內容。Firecrawl 的 /v1/scrape 端點讓我們可以送一個網址過去,它就會回傳網頁的主要內容,省去自己處理複雜 HTML 的麻煩。對我們來說,這是個超佛心的工具,因為它讓爬蟲變得更穩定,還能省下不少寫解析邏輯的時間。

要如何修改?

為了讓 photomgr 用上 Firecrawl API,我們需要把原本 ptt.go 的爬蟲邏輯改成呼叫 Firecrawl 的 API。以下是改動的重點:

  1. 保留公開 API 不變ptt.go 裡的公開函數(像是 GetPostsGetPostDetails)已經被其他使用者依賴,所以我們不能改動這些函數的輸入輸出格式,只能改內部的實作邏輯。
  2. 使用 Firecrawl API:改用 Firecrawl 的 /v1/scrape 端點來抓 PTT Beauty 板的列表頁(https://www.ptt.cc/bbs/Beauty/index.html)和單篇文章頁(像是 https://www.ptt.cc/bbs/Beauty/M.1748080032.A.015.html)。
  3. 解析 markdown 資料:Firecrawl 回傳的資料是 markdown 格式,我們需要把它解析成結構化的 JSON,像是文章標題、網址、作者、日期、推文數(或「爆」標記)等。
  4. 環境變數管理 API Key:Firecrawl 的 API 需要一個 key,我們會從環境變數 FIRECRAWL_KEY 讀取,確保安全不硬寫在程式碼裡。
  5. 單元測試:新增單元測試來驗證爬蟲和解析邏輯,目標是至少 80% 的程式碼覆蓋率。

PTT Beauty 板有年齡限制,瀏覽時需要設定 over18=1 的 Cookie 來通過檢查。在 Firecrawl API 的請求中,我們需要在 headers 裡加入這個 Cookie,這樣才能順利抓到頁面內容。具體寫法如下:

{
  "url": "https://www.ptt.cc/bbs/Beauty/index.html",
  "headers": {
    "Cookie": "over18=1",
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
  },
  "formats": ["markdown"],
  "onlyMainContent": true,
  "waitFor": 1000
}

這段 JSON 告訴 Firecrawl 在送請求時要帶上 over18=1 的 Cookie,模擬一個通過年齡驗證的使用者。User-Agent 則是模擬瀏覽器,確保 PTT 伺服器不會因為奇怪的請求頭而擋掉我們。

相關程式碼

以下是改用 Firecrawl API 的一些核心程式碼範例,展示如何抓取和解析 PTT Beauty 板的資料:

抓取列表頁

go

package ptt

import (
    "encoding/json"
    "net/http"
    "os"
)

type FirecrawlResponse struct {
    Success bool `json:"success"`
    Data    struct {
        Markdown string `json:"markdown"`
    } `json:"data"`
}

func GetPosts() ([]Post, error) {
    apiKey := os.Getenv("FIRECRAWL_KEY")
    if apiKey == "" {
        return nil, errors.New("FIRECRAWL_KEY is not set")
    }

    url := "https://www.ptt.cc/bbs/Beauty/index.html"
    reqBody := map[string]interface{}{
        "url": url,
        "headers": map[string]string{
            "Cookie":     "over18=1",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
        },
        "formats":       []string{"markdown"},
        "onlyMainContent": true,
        "waitFor":       1000,
    }

    body, _ := json.Marshal(reqBody)
    req, _ := http.NewRequest("POST", "https://api.firecrawl.dev/v1/scrape", bytes.NewBuffer(body))
    req.Header.Set("Authorization", "Bearer "+apiKey)
    req.Header.Set("Content-Type", "application/json")

    client := &http.Client{}
    resp, err := client.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    var fcResp FirecrawlResponse
    json.NewDecoder(resp.Body).Decode(&fcResp)
    if !fcResp.Success {
        return nil, errors.New("Firecrawl API request failed")
    }

    // 解析 markdown 成 Post 結構
    posts, err := parseIndexMarkdown(fcResp.Data.Markdown)
    if err != nil {
        return nil, err
    }
    return posts, nil
}

func parseIndexMarkdown(markdown string) ([]Post, error) {
    // 使用正則表達式或 markdown 解析庫解析
    // 範例:提取標題、網址、作者、日期、推文數
    var posts []Post
    // 假設 Post 結構如下
    type Post struct {
        Title     string
        URL       string
        Author    string
        Date      string
        PushCount int
    }
    // 實作解析邏輯(簡化範例)
    lines := strings.Split(markdown, "\n")
    for _, line := range lines {
        if strings.Contains(line, "[正妹]") || strings.Contains(line, "[公告]") {
            // 解析標題、網址等
            post := Post{ /* 填入解析結果 */ }
            posts = append(posts, post)
        }
    }
    return posts, nil
}

抓取單篇文章

go

func GetPostDetails(url string) (PostDetail, error) {
    apiKey := os.Getenv("FIRECRAWL_KEY")
    if apiKey == "" {
        return PostDetail{}, errors.New("FIRECRAWL_KEY is not set")
    }

    reqBody := map[string]interface{}{
        "url": url,
        "headers": map[string]string{
            "Cookie":     "over18=1",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
        },
        "formats":       []string{"markdown"},
        "onlyMainContent": true,
        "waitFor":       1000,
    }

    body, _ := json.Marshal(reqBody)
    req, _ := http.NewRequest("POST", "https://api.firecrawl.dev/v1/scrape", bytes.NewBuffer(body))
    req.Header.Set("Authorization", "Bearer "+apiKey)
    req.Header.Set("Content-Type", "application/json")

    client := &http.Client{}
    resp, err := client.Do(req)
    if err != nil {
        return PostDetail{}, err
    }
    defer resp.Body.Close()

    var fcResp FirecrawlResponse
    json.NewDecoder(resp.Body).Decode(&fcResp)
    if !fcResp.Success {
        return PostDetail{}, errors.New("Firecrawl API request failed")
    }

    // 解析 markdown 成 PostDetail 結構
    detail, err := parsePostMarkdown(fcResp.Data.Markdown)
    if err != nil {
        return PostDetail{}, err
    }
    return detail, nil
}

func parsePostMarkdown(markdown string) (PostDetail, error) {
    // 假設 PostDetail 結構如下
    type PostDetail struct {
        Author    string
        Board     string
        Title     string
        Date      string
        ImageURLs []string
        Content   string
    }
    // 實作解析邏輯(簡化範例)
    var detail PostDetail
    lines := strings.Split(markdown, "\n")
    for _, line := range lines {
        if strings.HasPrefix(line, "作者") {
            detail.Author = strings.TrimPrefix(line, "作者")
        } else if strings.Contains(line, "https://i.imgur.com") {
            detail.ImageURLs = append(detail.ImageURLs, line)
        } // 其他欄位解析
    }
    return detail, nil
}

單元測試範例

go

package ptt

import (
    "testing"
    "github.com/stretchr/testify/assert"
)

func TestGetPosts(t *testing.T) {
    // 模擬 Firecrawl API 回應
    mockMarkdown := `
[正妹] OOO
jerryyuan
5/24
[搜尋同標題文章](https://www.ptt.cc/bbs/Beauty/search?q=thread%3A%5B%E6%AD%A3%E5%A6%B9%5D)
    `
    posts, err := parseIndexMarkdown(mockMarkdown)
    assert.NoError(t, err)
    assert.Len(t, posts, 1)
    assert.Equal(t, "[正妹] OOO", posts[0].Title)
}

總結

PTT 改用 Cloudflare 保護後,讓我們原本的 GCP 爬蟲方案直接陣亡,逼得我們得找新方法來抓資料。Firecrawl API 成了我們的救星,不僅能繞過 Cloudflare 的防爬,還能把 PTT 的頁面轉成乾淨的 markdown,省下不少解析的麻煩。我們改進了 photomgr 的 ptt.go,用 Firecrawl API 抓取 Beauty 板的列表和文章,保留了原本的公開 API 介面,確保現有使用者不受影響。透過環境變數管理 API Key、加上單元測試,程式碼安全又穩定。這次遷移讓我們學到如何應對網站保護機制的挑戰,也證明 Firecrawl 是爬蟲任務的超強幫手。未來我們會持續監控 PTT 的變化,確保爬蟲能一直順利跑下去!

[好書分享] 連結(Nexus) - 從石器時代到AI紀元

連結: 從石器時代到AI紀元
Nexus : A Brief History of Information Networks from the Stone Age to AI
作者: 哈拉瑞  
原文作者: Yuval Noah Harari  
譯者: 林俊宏  出版社:天下文化 
出版日期: 2024/09/10 

買書推薦網址:

前言:

這是 2025 年第 2 本讀完的書。當初會看這本書只是因為是近期最暢銷的書籍,沒有想過裡面真的有蠻多部分是很值得思索的部分。

蠻建議有興趣的人可以來看看,裡面透過不少歷史故事與相關的說明分享,讓你更了解人類歷史,資訊流的連結關係。

內容摘要:

現在與未來,AI將會做什麼?
哈拉瑞的新巨作《連結》,以激勵人心的方式,
講述了我們如何走到這一刻,
以及現在必須做出的、攸關生存與發展的急迫決擇。
 
《連結》透過人類大歷史的長鏡頭,
檢視資訊網路與資訊流,如何將我們帶到AI紀元。
哈拉瑞引導我們走過石器時代、經歷《聖經》正典化、
印刷術的發明、大眾媒體的興起、以及民粹主義的重燃……
例舉了羅馬帝國、秦帝國、天主教會、蘇聯等體制
如何利用資訊技術來實現目標,無論是好是壞。
資訊,既不是真理真相的原料,也不只是權力與武器;
《連結》探索了許多極端之間、充滿希望的中間立場,
在鐵幕落下、矽幕升起之際,重新發現我們共有的人性。

封面
【開場白】前方的路滿布荊棘
第一部 人類形成的網路
第1章 資訊是什麼?
第2章 故事──無限的連結
第3章 文件──紙老虎也會咬人
第4章 錯誤──絕對正確是一種幻想
第5章 決擇──民主與極權制度簡史
第二部 非生物的網路
第6章 新成員──電腦與印刷機不同之處
第7章 不停歇──網路監控無所不在
第8章 易出錯──存在於電腦間的現實
第三部 電腦政治學
第9章 民主制度──我們還能對話嗎?
第10章 極權主義──權力歸於演算法?
第11章 矽幕──全球帝國、或是全球分裂?
【結 語】打造自我修正機制

心得:

這本書講解著資訊與知識本身,造成人類相互連結與對立的相關故事。作者在以下的採訪中有著相當清楚的解釋:

在許多文案中,可能會讓人以為他是在探討著 AI 與人類的關係,但是其實做作者探討著更深層的問題 — 人與人之間的 「連結」問題。

這本書首先從資訊交換開始說起來,然後透過故事的方式將資訊將以保存以及流傳。這邊一開始有提到大家都有聽說過「歷史是勝者寫的」,也就是一種資訊的連結的傳遞方式。 所有你以為的政治故事其實都是因為勝利的那一方寫下來的,而故事造成傳遞效果會更加的強大。 也有提到聖經上的許多故事也是值得反覆的思考,主要也因為當初的時間與政治背景的不同。

故事能創造出「第三層次的現實」:存在於主體間的現實,主觀現實 與 主體間的現實。

故事造就時事變革

所以希特勒之所以能贏得 1933 年的大選,因為德國經濟危機狀況下,許多人相信了他提出的故事。

回過頭人與人「連結」的方式,就談到了關於「書籍」。書籍是最原始將人與人連結傳遞下來的介質,從「聖經」到各種重大書籍為主。但是隨著人類歷史與各地區政治的變革,聖經的內容也有了不同教義與流派的解釋。甚至也有了不同流傳版本的差異。

書籍與宗教的變革

但是宗教是否有自我修正的機制?宗教的本體是人,人類會有許多的偏見與錯誤,自然也會有相關的修正機制。這也代表著人類彼此間的相互連結,也是存在著許多修復的方式。

連結的修復機制

科學也有自我修復的機制,書中提到謝赫特曼的「準晶體」理論,震驚了整個科學界,並且也受到許多學者的反抗與輿論的壓力。但是著隨著科學技術的進步,整個理論慢慢被澄清。理論本身的連結也因此被修復了起來。

政治上 - 民主 與 極權 產生的連結

回歸到政治上,民主自然也是一種人與人之間的一種連結。人民透過民主制度選舉出的政府單位。如果發生了問題,也代表著人民有機會可以退回自己決定。 這會讓人覺得「民主」是脆弱與妥協的產物? 作者也提到,民主的本體就在於許多的選擇與討論與妥協下的變革,相比「極權」帶來的“容易二分法”,民主其實是相當的脆弱。所謂的民主,就是人民有著投票與反對的權利。而極權是沒有的:

Google Chrome 2025-05-04 13.22.48

相反複雜的民主,其實極權反而是簡單

史達林是大家都認為的蘇俄極權統治者,他建立了一個極端的極權組織與統治制度。到了現在許多的國家依舊是依照著相關的方式:

  • 秘密警察: 針對資訊的統一管理,並且鼓勵檢舉。讓人與人之間的連結切斷,彼此不相信。(史達林大清洗)
  • 剝奪人民的選舉與投票權: 鞏固著自己的政治權利。

但是卻因為自己過度的極權統治,讓自己在病倒的那一天沒有人敢靠近他給他足夠的醫治。

社群媒體造就的資訊不對稱與切斷連結

這邊有提出在社群媒體興盛的這個年代,人們越來越喜歡在社群媒體上「取得同溫層」的鼓勵。但是卻又更喜歡去「異溫層」層裡面爭吵與彼此對立。這其實也是因為 AI 計算出來這就是一個推薦系統上的問題。

Typora 2025-05-04 13.36.19

使用去中心化的資訊,是否可以達成民主?

這裡也有探討著最新的技術與民主的交互作用,是否可以用去中心化來保存資訊,達到真正的社群民主?但是作者也提出疑問,如果類似史達林的極權統治者,統治著 51% 的人民,是否就可以消除掉其他的聲音?

Google Chrome 2025-05-04 13.49.22

我們應該如何小心

如同前面提出的:「民主是相當複雜且脆弱的」,相比之下民粹極權都相對的簡單與易懂。 這也是民主本身需要解決的問題,但是在 SNS 上面我們會看到許多“簡單的暴力”的言論方式,來鼓勵許多對立的情緒。 你之所以會看到,其實並不一定代表那是大部分人的情緒,反而可能是推薦系統所造成的結果。

推薦系統會推薦著「你喜歡的」讓你有同感,但是更傾向推薦「你厭惡的」來讓你回覆文章。對於政治也是如此,如果你看到簡單的贊成與反對,應該要花一點時間去閱讀更多的書籍與資訊。來讓自己充分了解每一方的說法,而不是限於各種的反對與破壞「連結」的說法之中。

不論是你喜歡的資訊,或是你不喜歡的資訊。我們都希望每一個人要有主動研究資訊的能力。 就像是讀書一樣,要能夠取得資訊並且去思考真正的問題。 比較這些政治人物之前與現在的說法,是否只是因為某些政治取向的操作。

最後再次呼應著大家「民主是一個相對的概念」,並不存在的非黑即白的想法。「民主是複雜且脆弱的」,但是保有人民選舉與決定自我投票的權力才是真正的民主。

Google Chrome 2025-05-04 14.01.52

[Gemini][LINEBot] 透過 Google ADK 打造一個 Agent LINE Bot

image-20250410202925234

前言

雖然前幾天我才剛將 OpenAI Agents SDK 整合成一個簡單的 LINE Bot 範例程式,就在 20250410 的凌晨, Google 就宣佈了 ADK (Google Agent SDK) 的發佈。

本篇文章將介紹如何透過 Google Agent SDK (ADK) 來打造一個最簡單的 LINE Bot 功能,作為之後 MCP 與其他功能的起始專案。 (想不到才沒隔多久,就可以換成 Google ADK XD)

範例程式碼: https://github.com/kkdai/linebot-adk

05/25更新: SDK 有更改成 asynchronous 形式。

session = await session_service.create_session(app_name=app_name, 
                                    user_id=user_id, 
                                    session_id=session_id)

快速簡介 Google ADK

Repo: https://github.com/google/adk-python

intro_components.png

Google 推出的Agent Development Kit (ADK),這是一個開源框架,旨在簡化智能多代理系統的開發。以下是內容的重點:

  • ADK是一個開源框架,專為開發多代理系統而設計,提供從構建到部署的全方位支持。
  • 它支持模組化和可擴展的應用程序開發,允許多個專業代理的協作。

功能特點

  • 內建串流:支持雙向音頻和視頻串流,提供自然的人機互動。
  • 靈活的編排:支持工作流代理和LLM驅動的動態路由。
  • 集成開發者體驗:提供強大的CLI和可視化Web UI,便於開發、測試和調試。
  • 內建評估:系統性地評估代理性能。
  • 簡易部署:支持容器化部署。

支援視覺化測試 WebUI

adk-web-dev-ui-chat.png

(Refer: https://google.github.io/adk-docs/get-started/quickstart/#run-your-agent))

可以在本地端透過 WebUI 來做一些快速的測試,快速部署到 Google Cloud 。 相關的功能也會在後續的文章中陸續提到。

整合 LINE Bot SDK 需要注意的事項:

image-20250410205257808

接下來跟大家講一下,要加上 LINE Bot SDK 有哪一些需要注意的地方。

範例程式碼: https://github.com/kkdai/linebot-adk

Agent 起始的流程

目前是放在 Services 啟動的時候,就將 Agent 初始化。

# Initialize ADK client
root_agent = Agent(
    name="weather_time_agent",
    model="gemini-2.0-flash-exp",
    description=(
        "Agent to answer questions about the time and weather in a city."
    ),
    instruction=(
        "I can answer your questions about the time and weather in a city."
    ),
    tools=[get_weather, get_current_time],
)
print(f"Agent '{root_agent.name}' created.")

建立 Agent 之後,接下來要準備好 Runner 來執行 Agent 溝通的工作。

# Key Concept: Runner orchestrates the agent execution loop.
runner = Runner(
    agent=root_agent,  # The agent we want to run
    app_name=APP_NAME,   # Associates runs with our app
    session_service=session_service  # Uses our session manager
)

這樣之後就可以透過 async 來呼叫這個 runner 來取的 agent 的結果。(後續會提到)

針對不同使用者,使用記憶體來記憶對話

image-20250410210008468

ADK 中,有蠻多相關的 Memory Services 可以使用 :

  • InMemoryMemoryService
    • 使用 Serives 的記憶體來儲存,可以作為基本的儲存方式。但是如果使用 CloudRun ,當服務重啟就會消失掉。
  • VertexAiRagMemoryService
    • 使用 VertexAI 的 RAG 服務,這邊可能會有額外的儲存空間的費用會產生。

接下來分享一下,如何使用 InMemoryMemoryService 來儲存不同用戶的對話記憶。

async def get_or_create_session(user_id):
    if user_id not in active_sessions:
        # Create a new session for this user
        session_id = f"session_{user_id}"
        session = await session_service.create_session(
            app_name=APP_NAME,
            user_id=user_id,
            session_id=session_id
        )
        active_sessions[user_id] = session_id
        print(
            f"New session created: App='{APP_NAME}', User='{user_id}', Session='{session.id}'")
    else:
        # Use existing session
        session_id = active_sessions[user_id]
        print(
            f"Using existing session: App='{APP_NAME}', User='{user_id}', Session='{session_id}'")

    return session_id

首先以上 get_or_create_session() 可以透過 user_id 來建立或是取得使用者的 Session ID。這樣可以讓 ADK 透過正確的 Session ID 來繼續相關的對話。

async def call_agent_async(query: str, user_id: str) -> str:
    """Sends a query to the agent and prints the final response."""
    print(f"\n>>> User Query: {query}")

    # Get or create a session for this user
    session_id = await get_or_create_session(user_id)

    # Prepare the user's message in ADK format
    content = types.Content(role='user', parts=[types.Part(text=query)])

    final_response_text = "Agent did not produce a final response."  # Default

    try:
        # Key Concept: run_async executes the agent logic and yields Events.
        # We iterate through events to find the final answer.
        async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
            # Key Concept: is_final_response() marks the concluding message for the turn.
            if event.is_final_response():
                if event.content and event.content.parts:
                    # Assuming text response in the first part
                    final_response_text = event.content.parts[0].text
                elif event.actions and event.actions.escalate:  # Handle potential errors/escalations
                    final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
                # Add more checks here if needed (e.g., specific error codes)
                break  # Stop processing events once the final response is found
    except ValueError as e:
        # Handle errors, especially session not found
        print(f"Error processing request: {str(e)}")
        # Recreate session if it was lost
        if "Session not found" in str(e):
            active_sessions.pop(user_id, None)  # Remove the invalid session
            session_id = await get_or_create_session(user_id)  # Create a new one
            # Try again with the new session
            try:
                async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
                    # Same event handling code as above
                    if event.is_final_response():
                        if event.content and event.content.parts:
                            final_response_text = event.content.parts[0].text
                        elif event.actions and event.actions.escalate:
                            final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"
                        break
            except Exception as e2:
                final_response_text = f"Sorry, I encountered an error: {str(e2)}"
        else:
            final_response_text = f"Sorry, I encountered an error: {str(e)}"

    print(f"<<< Agent Response: {final_response_text}")
    return final_response_text

透過以上的程式碼,每一次使用者的資訊 (Query, User_ID). 傳入後,透過不同用戶的 user_id 來建立(或取得)不同溝通的紀錄(Session) 。

再來透過不同的 Session 來跑 ADK 的功能查詢。 (主要是透過 async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content): )

這樣就可以達成不同使用者,有不同的記憶內容。也不用另外來呼叫相關記憶體相關的 Function 。

快速總結與未來發展

學習到現在,或許大家會跟我一樣有疑問是: 「究竟 OpenAI Agents SDK 跟 Google ADK 有什麼差異?」

Google Chrome 2025-04-10 21.28.23

(表格整理 by Grok3)

就像這個表格整理的一樣,我覺得 ADK 使用起來沒有比 OpenAI Agent SDK 更簡單。但是由於內建許多有用的 WebUI 還有已經打包好許多的工具。 讓未來的開發上不會有後顧之憂,接下來也會將 MCP Server,語音或是多模態相關的應用整理近 ADK 來跟大家分享,敬請期待。

[Gemini][LINEBot] 打造一個 OpenAI Agents LINE Bot 並且使用 Google Gemini Model

A sleek, minimal interface displaying a task list for an AI agent, including ‘triage_agent,’ ‘guardrail,’ and ‘update_salesforce_record,’ over a fluid blue abstract background.

前言

OpenAI 在 03/11 發佈了新的 OpenAI-Agent SDK 的套件 (OpenAI-Agents-Python),裡面不僅僅支援多 Agent 可以相互作用外,還宣佈了可以支援 MCP Server

本篇文章將介紹如何透過 OpenAI-Agents SDK 來打造一個最簡單的 LINE Bot 功能,作為之後 MCP 與其他功能的起始專案。

範例程式碼: https://github.com/kkdai/linebot-openai-agent

快速簡介 OpenAI-Agents-SDK

OpenAI-Agent SDK 的套件 (OpenAI-Agents-Python) OpenAI推出了一系列新工具和API,包括Responses API和Agents SDK,這些工具旨在簡化開發者和企業構建智能代理的過程。Responses API結合了Chat Completions API的簡單性和Assistants API的工具使用能力,支持網頁搜索、文件搜索和電腦使用等內建功能。Agents SDK提供了改進的可觀察性和安全檢查,簡化多代理工作流程的編排,並支持智能代理之間的控制轉移,從而提升各行各業的生產力。

並且這個套件同時也提供支援 MCP Server 的功能,詳細部分下一次再介紹。

image-20250401123219075

透過 Custom Provider 來使用 Google Gemini

先來讓 OpenAI-Agents SDK 可以使用其他公司的模型,這邊使用的是 Custom Provider。官方的敘述如下:

image-20250401123524000

這邊我們使用 custom_example_provider.py 範例程式碼來參考,實際完成整合可以看 範例程式碼: https://github.com/kkdai/linebot-openai-agent

BASE_URL = os.getenv("EXAMPLE_BASE_URL") or ""
API_KEY = os.getenv("EXAMPLE_API_KEY") or ""
MODEL_NAME = os.getenv("EXAMPLE_MODEL_NAME") or ""

# Initialize OpenAI client
client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)
set_tracing_disabled(disabled=True)

這邊主要需要三個環境參數,以下開始詳細說明:

  • BASE_URL: 也就是 Custom Provider 的 API 網址,如果要使用 Google Gemini 請記得改成 https://generativelanguage.googleapis.com/v1beta/
  • API_KEY: 這邊就寫成自己的 Google Gemini API Key
  • MODEL_NAME: 這邊記得要改成 Gemini 的 Model ,要省費用可以使用 gemini-1.5-flash

然後這樣的透過 AsyncOpenAI() 就可以呼叫 Google Gemini 的服務了。

加入簡單的 Tools

這個範例參考原本的 Tools 的寫法,並且加入兩個 Tools 。

@function_tool
def get_weather(city: str):
    """Get weather information for a city"""
    print(f"[debug] getting weather for {city}")
    return f"The weather in {city} is sunny."


@function_tool
def translate_to_chinese(text: str):
    """Translate text to Traditional Chinese"""
    print(f"[debug] translating: {text}")
    return f"Translating to Chinese: {text}"

這邊可以看到有兩個工具可以使用,一個是 translate_to_chinese 另外一個是 get_weather ,當然因為這是一個範例,裡面就直接回覆天氣很好即可。

整合 LINE Bot SDK 跟 OpenAI-Agents-SDK

這邊程式碼分成兩個部分解釋,首先是針對 LINE Bot 的 WebHook 處理的部分。這邊沒有太多其他的部分:

for event in events:
        if not isinstance(event, MessageEvent):
            continue

        if event.message.type == "text":
            # Process text message
            msg = event.message.text
            user_id = event.source.user_id
            print(f"Received message: {msg} from user: {user_id}")

            # Use the user's prompt directly with the agent
            response = await generate_text_with_agent(msg)
            reply_msg = TextSendMessage(text=response)
            await line_bot_api.reply_message(
                event.reply_token,
                reply_msg
            )
        elif event.message.type == "image":
            return 'OK'
        else:
            continue

主要會去呼叫 generate_text_with_agent 並且等待他的結果。

async def generate_text_with_agent(prompt):
    """
    Generate a text completion using OpenAI Agent.
    """
    # Create agent with appropriate instructions
    agent = Agent(
        name="Assistant",
        instructions="You are a helpful assistant that responds in Traditional Chinese (zh-TW). Provide informative and helpful responses.",
        model=OpenAIChatCompletionsModel(
            model=MODEL_NAME, openai_client=client),
        tools=[get_weather, translate_to_chinese],
    )

    try:
        result = await Runner.run(agent, prompt)
        return result.final_output
    except Exception as e:
        print(f"Error with OpenAI Agent: {e}")
        return f"抱歉,處理您的請求時出現錯誤: {str(e)}"

這邊使用 OpenAIChatCompletionModel 的時候,需要透過 model=MODEL_NAME, openai_client=client) 來使用 Custom Model Provider 。這樣才能正確使用到 Google Gemini 的 Model 。

要使用 Tools ,就必須要使用到 tools=[get_weather, translate_to_chinese] 將所有支援的 Tools 加入進來。才能正確引用到。

成果與如何使用

image-20250401122056151

部署完畢後,這是一個簡單的截圖。使用也很簡單,直接就詢問他問題。如果跟翻譯有關,就會直接使用到 translate_to_chinese ,如果要抓取城市的天氣,他則會先跟你確認清楚城市的名稱後,一率都回覆你天氣晴朗。

快速總結與未來發展

本篇文章提供了範例程式碼,並且提供了如何使用 OpenAI-Agent SDK 的套件 (OpenAI-Agents-Python) 來串接 LINE Bot SDK 。 之後的幾篇文章,我們將開始串接一些有用的 MCP Server 並且讓我們的 LINE Bot 有更完整的功能。

[Google Cloud] GCP 上面避免 YouTube 阻擋同區網路流量(透過 Proxy 解決)

LINE 2025-03-28 15.31.08

前言

之前有一篇文章「[Google Cloud] 如何在 GCP Cloud Run 上面透過 LangChain 取得 YouTube 的相關資訊」,雖然有講過使用 Secret Manager 與 GCP. 相關的 LangChain YouTube 套件來試著抓取資料。但是近期 YouTube 又開始修該他的讀取規範,造成原來的方式不能成功,這裡記錄一下主要錯誤訊息,還有該如何解決。

主要的問題

有一天 YouTube 的字幕開始抓不到,查詢 Log 出現以下內容。

During handling of the above exception, another exception occurred:
youtube_transcript_api._errors.RequestBlocked:
Could not retrieve a transcript for the video https://www.youtube.com/watch?v=ViA4-YWx8Y4! 
This is most likely caused by:
YouTube is blocking requests from your IP. This usually is due to one of the following reasons:
- You have done too many requests and your IP has been blocked by YouTube
- You are doing requests from an IP belonging to a cloud provider (like AWS, Google Cloud Platform, Azure, etc.). Unfortunately, most IPs from cloud providers are blocked by YouTube.

There are two things you can do to work around this:
1. Use proxies to hide your IP address, as explained in the "Working around IP bans" section of the README (https://github.com/jdepoix/youtube-transcript-api?tab=readme-ov-file#working-around-ip-bans-requestblocked-or-ipblocked-exception).
2. (NOT RECOMMENDED) If you authenticate your requests using cookies, you will be able to continue doing requests for a while. However, YouTube will eventually permanently ban the account that you have used to authenticate with! So only do this if you don't mind your account being banned!
If you are sure that the described cause is not responsible for this error and that a transcript should be retrievable, please create an issue at https://github.com/jdepoix/youtube-transcript-api/issues. Please add which version of youtub

這邊快速提到主要出現的問題:

  • 由於透過 LangChain 架構中, GoogleApiYoutubeLoader 也是呼叫 https://github.com/jdepoix/youtube-transcript-api/ 來使用。
  • 由於部署在 GCP 上面,造成 YouTube 開始阻擋所有同一個雲端的需求。
  • 這邊套件是建議使用 Webshare 這類型的 Proxy 服務,來去抓取相關資料。

WebShare

image-20250328170602359

Webshare 是一個第三方的付費 Proxy 服務,可以讓你的 Web Request 可以透過這個服務達到以下的相關事項:

  • 想要在 GCP 上面存取(爬蟲) Google 服務(Map, YouTube, Google Search)
  • 想要爬一些比較有阻擋 CSP (Cloud Services Provider) IP 的 CDN 服務(比如說 CloudFlare)

他也有相關的免費 Proxy 額度可以使用:

  • 五個 Proxy
  • 一個月 1GB usage

對於 Youtube 字幕來說,這樣流量不會是問題。

使用上也蠻簡單的,這裡附上透過 Webshare 抓取 YouTube Transcript 的程式碼:

from youtube_transcript_api import YouTubeTranscriptApi
from youtube_transcript_api.proxies import WebshareProxyConfig
import os


def get_transcripts(video_id, languages):
    # Get proxy credentials from environment variables
    proxy_username = os.environ.get("PROXY_USERNAME")
    proxy_password = os.environ.get("PROXY_PASSWORD")

    ytt_api = YouTubeTranscriptApi(
        proxy_config=WebshareProxyConfig(
            proxy_username=proxy_username,
            proxy_password=proxy_password,
        )
    )
    transcript_list = ytt_api.fetch(video_id, languages=languages)
    transcript_texts = [snippet["text"] for snippet in transcript_list.to_raw_data()]
    return " ".join(transcript_texts)


# Example usage (only runs when script is executed directly)
if __name__ == "__main__":
    video_id = "YOUR_VIDEO_ID"
    languages = ["en", "de"]
    transcript_text = get_transcripts(video_id, languages)
    print(transcript_text)

比起直接連線來說,雖然會比較慢一點。但是真的可以直接抓取資料。

可能要避免的部分(額外費用)

雖然 Webshare 有免費的額度又好用,但是如果你抓取次數太頻繁可能會被 Webshare 阻擋掉叫你去付錢。這邊要小心。

youtube_transcript_api._errors.RequestBlocked:

Could not retrieve a transcript for the video https://www.youtube.com/watch?v=ViA4-YWx8Y4! This is most likely caused by:

YouTube is blocking your requests, despite you using Webshare proxies. Please make sure that you have purchased "Residential" proxies and NOT "Proxy Server" or "Static Residential", as those won't work as reliably! The free tier also uses "Proxy Server" and will NOT work!

The only reliable option is using "Residential" proxies (not "Static Residential"), as this allows you to rotate through a pool of over 30M IPs, which means you will always find an IP that hasn't been blocked by YouTube yet!

You can support the development of this open source project by making your Webshare purchases through this affiliate link: https://www.webshare.io/?referral_code=1yl49cgzfedr
Thank you for your support! <3

[Gemini][MCP] 在 Cline 上面使用 Gemini 來呼叫 MCP 的功能

What is Model Context Protocol (MCP)? How it simplifies AI ...

前情提要

最近 MCP 是非常熱門的討論話題,但是大家提到 MCP 不免會想到 Anthropic 的 Claude 或是其他語言的模型。這一篇文章要告訴大家關於 MCP 的一些基礎原理,並且如何使用 Google Gemini 來呼叫 MCP 。希望能給大家一些整理。

什麼是 MCP(Model Context Protocol)

根據 Anthropic 的文件上面有提到: MCP 是一個開放協議,用於標準化應用程式如何為大型語言模型(LLM)提供上下文。您可以將 MCP 想像成 AI 應用程式的 USB-C 接口。就像 USB-C 為您的設備提供了一個標準化的方式來連接各種外圍設備和配件一樣,MCP 為 AI 模型提供了一個標準化的方式來連接不同的數據源和工具。

這邊也分享 YT https://www.youtube.com/watch?v=McNRkd5CxFY&t=17s 上面的架構圖,讓大家更容易理解

Google Chrome 2025-03-23 16.41.43

(圖片來源 技术爬爬虾 TechShrimp :MCP是啥?技术原理是什么?一个视频搞懂MCP的一切。Windows系统配置MCP,Cursor,Cline 使用MCP

這邊可以看出來, 透過 MCP 這邊提到的 AI 客戶端(大家常使用的 ChatGPT, Claude, Gemini 的等等)都可以透過 MCP 架構直接來對這些服務做「操作」。

MCP 服務中的架構圖

image-20250323164924676

(架構圖: MCP Core architecture)

這個架構圖,有清楚的敘述出關於 MCP 的 Client Server 的架構,這邊再強調一下。

  • MCP Host: 使用這些 MCP 服務的應用,可能是 Cline, Windsurf 或是 Claude Desktop)
  • MCP Server: 許多地方應該都要介紹過,這裡不太贅述。之後也會有範例程式碼。就是一個萬用的溝通協定,讓每一個 MCP Host 可以更容易去使用一些外部功能。並且變成一個共通的協定。
  • MCP Client: 在每一個 Host 中,確定使用某個 MCP Server 後。在 Host 中,會有其相關的 client 這裡將會是用 Prompt 存在著。接下來會詳細敘述。

MCP 運作細節

完整細節可以參考網路上這位的說明,不過我將內容改成 Google Gemini 相關的應用。除了換掉影片中使用的 DeepSeek 之外,也可以讓整個使用更符合資訊安全相關的應用。

參考影片:

技术爬爬虾 TechShrimp MCP是怎么对接大模型的?抓取AI提示词,拆解MCP的底层原理

透過 CloudFlare 建立 AI API Gateway

(如果要查看相關細節,需要使用 OpenRouter 或是 OpenAI Compatible )

如果要查看 MCP 溝通細節,就得使用 OpenAI 或是 OpenRouter 。這裡查看影片可以看到完整教學。這裡將直接貼出相關細節。

Code 2025-03-23 17.35.03

以上是透過 Cloudflare 抓取封包後,來解析 MCP 的溝通機制:

  • 你會發現 MCP Server 所有安裝的 Server 跟功能都被當作 Promopt 輸入給該模型。
  • 每一個 MCP Server 能做什麼,會寫在 Prompt 之中(圖片可以看到 Github Repository MCP Server)

以上這些資料很重要,稍後會在下一個章節 MCP 跟 Function Call 來討論。

MCP 跟 Function Call 的差別

Expanding AI Horizons: The Rise of Function Calling in LLMs

原本 LLM 有一些模型會透過 Function Calling 來決定要如何呼叫一些工具,這裡其實需要 Model 額外的支持,雖然說大家常見的 LLM 提供商 OpenAI, Google, Anthropic 都已經支援 Function Calling ,但是每一次都要為了這些工具額外去寫相關的 Function Calling App 還是很讓人頭痛。

這也是為什麼 MCP 會紅的原因,他讓原本的 Function Calling App 變成一個共用機制。每一個相關的應用廠商,都可以寫出自己的 MCP Servers ,也可以讓使用的服務變得更多人可以使用。

並且,很重要的:

  • MCP 讓不能使用 Function Calling 的 Model 也能使用 MCP Servers 做出類似 Function Calling 的應用
  • MCP 讓不能使用 Function Calling 的 Model 也能使用 MCP Servers 做出類似 Function Calling 的應用
  • MCP 讓不能使用 Function Calling 的 Model 也能使用 MCP Servers 做出類似 Function Calling 的應用

因為 技术爬爬虾 TechShrimp MCP是怎么对接大模型的?抓取AI提示词,拆解MCP的底层原理 影片中使用到的相關模型是 DeepSeekChat 其實還不支援 Function Calling ,但是他也是可以在透過 MCP Server 來互動。

使用 Cline 作為 MCP host

image-20250323175901298

(Refer: Cline Plugin)

Cline 是一個 VS Code 的 Plugin ,目前已經有相當多的類似 AI IDE Plugin 。但是由於 Cline 本身內建支援 MCP Server 的溝通機制,並且有一個內建的推薦清單,讓開發者可以很快速地連接 MCP Servers ,這裡很推薦使用 Cline 作為你第一個了解 MCP 的工具。

相關 Cline 設定流程,讓你使用 MCP Servers

  • 安裝 Cline Plugin
  • 選擇你需要的 MCP Servers image-20250323180151743
  • 設定你的模型,這裡使用 Google Gemini ,並且使用比較省錢的 Gemini-1.5-flashCode 2025-03-23 18.02.07

這裡比較建議大家可以先安裝最安全跟簡單的 “Time MCP Servers“,也可以從 Cline 的介面上來理解。

Code 2025-03-23 18.03.33

這裡可以看到 “Time MCP Servers” 支援兩個功能:

  • Get_current_time
  • Convert_time

這時候如果詢問,請問幾點?

image-20250323180956464

  • 他可能會先請你告訴他你所在區域
  • 然後決定要跑 MCP Server (會請求你同意)
  • 同意後,才會告訴你現在的結果。

如果你要換成 告訴我現在幾點?並且轉換成法國的時間? Code 2025-03-23 18.11.23

這時候就會呼叫轉換的 API ,這時候也會要求同意。完成後就可以跑出相關的工作。

未來發展:

本篇文章快速做了一個簡介,並且使用 Cline 搭配了 Gemini 1.5-flash model 來呼叫 MCP Server 做一個查詢時間的展示。接下來的文章將會告訴大家,該如何寫一個 MCP Server ,並且分享給大家可能有用的應用。

參考資料: