AG Day 24 搜尋工具:讓 Agent 查網路
執行需求:CPU+API key。昨天的 AG Day 23 引用與出處:答案要能追溯(原文連結)讓我們能把答案綁回具體段落與來源,但整條流程仍有一個硬性限制:來源必須已經躺在知識庫裡。這在研究助理的實際使用情境中很快就會撞牆——使用者問的是這個月才發布的消息,而知識庫是上週抓的;或者問的是你根本沒想過要收錄的主題。今天要補上這一塊:把「上網搜尋」與「擷取網頁」包成代理可以呼叫的工具,讓系統第一次具備主動發現新來源的能力。文章需要 LLM 與搜尋服務的 API key 才能完整執行,--dry-run 模式則改用本地示範語料,離線可跑,輸出會明確標示為示範。
引言
先釐清一個常見誤解:很多人以為「讓代理查網路」就是接一個搜尋 API,把結果字串丟給模型。這樣做確實會動,但品質與成本都很難控制。原因有三個。第一,搜尋結果通常包含標題、摘要與網址,卻沒有真正可引用的正文;模型看到摘要後會「以為自己讀過了」,於是寫出沒有段落依據的結論——這正是我們昨天才剛修好的問題。第二,搜尋服務回傳的欄位與排序邏輯各家不同,如果不把它正規化成統一的工具契約,後面要換供應商就會到處改程式。第三,網路是充滿錯誤與延遲的環境:逾時、429、404、內容農場、重複頁面都會發生,如果工具遇到錯誤就拋例外,整條代理流程會直接中斷。
所以今天的工作不是「接一個 API」,而是設計兩個工具:search_web 負責廣度,fetch_page 負責深度。搜尋便宜、快、覆蓋面大,用來發現候選;擷取較貴、較慢、但拿到的是可引用的正文,用來把候選變成真正能入庫、能引用的來源。這兩個工具的分工,與 AG Day 22 檢索品質:查詢改寫與 rerank(原文連結)裡「召回與精排」的漏斗是同一個邏輯,只是場景從本地知識庫換成了開放網路。
今天的產出會直接接回昨天的引用鏈路:抓到的頁面會被寫進 documents 與 chunks,附上抓取時間與內容雜湊,之後產出的報告就能引用它們。也就是說,從今天起,研究助理的知識庫是可以「長大」的。
原理/觀念
工具契約比工具實作重要
在代理系統裡,工具不只是函式,而是「模型與程式之間的合約」。這份合約包含五個要素:工具名稱(模型看得到、要能望文生義)、描述(模型判斷何時該用的依據)、參數 schema(型別、必填、範圍)、回傳格式(模型要能解析的穩定結構)、錯誤語意(失敗時模型該如何反應)。前四項在 AG Day 5 談 function calling 時已經建立觀念,今天要特別強調第五項。
為什麼錯誤語意重要?因為代理是「會自己決定下一步」的系統。工具回傳 {"error": "timeout", "retryable": true},模型就知道可以換個查詢或稍後重試;工具回傳 {"error": "invalid_input", "retryable": false},模型就該修正參數而不是重試。相反地,如果工具直接拋出 Python 例外,LangGraph 的節點會中斷,除非我們在上層包了例外處理(AG Day 7 錯誤處理:逾時、重試與工具失敗(原文連結)談過),不然整個任務就死了。把錯誤變成「回傳值」而不是「例外」,是工具設計的第一原則。
搜尋與擷取為什麼要分開
搜尋 API 的價值在於它已經幫你做完三件事:找到可能相關的頁面、排出大致順序、抽出簡短摘要。這些摘要適合用來「判斷要不要深入」,但不適合直接當成引用來源,因為它不是原始正文,可能被壓縮、改寫、甚至過時。擷取工具則相反:它只處理一個網址,但拿到的是清洗後的正文,可以分塊、向量化、保存快照。
分成兩個工具的另一個好處是成本控制。搜尋一次可能回傳十個結果,若每個都立刻抓取全文,延遲與流量都會失控。讓模型先看搜尋摘要決定要抓哪幾頁,是把「判斷」交給模型、把「預算」留在程式端的做法。這個模式在後面的多代理架構(AG Day 29 起)會再次出現:先廣泛探索,再針對重點深入。
選哪個搜尋服務
本系列選擇 Tavily Search API 作為示範,理由是它針對 AI 應用設計,回傳結構包含標題、網址與已抽取的內容摘要,對代理呼叫相對友善,且官方提供 Python 客戶端套件。要強調的是:這不是唯一選擇,任何能回傳結構化結果的搜尋服務都可以套用同一套工具契約。實務上常見的替代方案包括自架的 SearXNG(需要自己維護實例)、搜尋引擎官方 API(如 Bing、Google 的搜尋 API),以及企業內部的搜尋閘道。實際的參數名稱、計費方式與可用欄位,一律以各服務的官方文件為準,本系列不寫死任何價格或額度數字。
還有一個技術路線值得知道:不依賴第三方 API,直接用 httpx 抓搜尋結果頁面再解析。這條路在示範時很吸引人,但正式使用時會遇到反爬機制、版面變動、以及服務條款的問題。我的建議是:教學與離線測試可以用本地語料模擬,正式流程請用有明確授權的 API。我們今天的 dry-run 模式正是走本地語料這條路。
工具回傳的大小控制
網頁正文動輒數千到數萬字,直接塞進模型上下文會產生三個問題:token 成本暴增、關鍵資訊被稀釋、以及超過上下文上限。因此擷取工具的回傳必須經過處理:只回傳清洗後正文、設定字數上限、必要時回傳「已入庫並可檢索」的摘要與識別碼,讓模型後續用檢索工具(昨天的 retrieve())取得需要的段落,而不是把全文塞進對話。
這個設計模式稱為「以識別碼代替內容」:工具回傳的不是全文,而是「我已經把這頁寫進知識庫了,你可以用 doc_id 去檢索」。這樣即使是長文件,也不會撐爆上下文,而且後續引用、查核都能沿用同一套機制。
去重、冪等與時效
同一個主題常常在不同查詢中撈到同一個網址;同一份文件也可能被抓取多次。如果每次抓取都新增一筆記錄,知識庫會迅速膨脹,檢索結果也會被重複段落佔滿。處理方式是用「來源識別 + 內容雜湊」做冪等入庫:以正規化後的網址(或內容雜湊)當文件識別,已存在且雜湊相同就跳過;雜湊不同則視為內容更新,保留新的並記錄時間。這正是昨天 content_hash 欄位的用途,今天終於派上用場。
完整實作
今天在 src/research_agent/tools.py 實作。第一步,先定義所有工具共用的一致性回傳格式:
# research-agent/src/research_agent/tools.py
from dataclasses import dataclass, asdict
from typing import Any
@dataclass
class ToolResult:
ok: bool
data: Any = None
error: str | None = None
retryable: bool = False
def to_dict(self) -> dict[str, Any]:
return {k: v for k, v in asdict(self).items() if v is not None}
段落說明:retryable 是給模型看的提示,最後再轉成 JSON 字串當作工具訊息。刻意讓 to_dict() 過濾掉 None 欄位,輸出的 JSON 會更乾淨、也更省 token。
第二步,實作搜尋工具。它有一段真實路徑(Tavily)與一段離線路徑(本地示範語料):
import json
import os
from pathlib import Path
FIXTURE_DIR = Path("data/fixtures")
def _mock_search(query: str, max_results: int) -> list[dict[str, Any]]:
"""離線示範語料:僅供 dry-run 流程驗證,不代表真實搜尋結果。"""
path = FIXTURE_DIR / "search_demo.json"
if not path.exists():
return []
records = json.loads(path.read_text(encoding="utf-8"))
keyword = query.strip()
matched = [r for r in records if keyword in r.get("keywords", [])]
return [
{"title": r["title"], "url": r["url"], "content": r["summary"]}
for r in matched[:max_results]
]
def search_web(query: str, max_results: int = 5, dry_run: bool | None = None) -> dict[str, Any]:
"""搜尋網路並回傳正規化後的結果清單。"""
offline = os.environ.get("RESEARCH_AGENT_DRY_RUN") == "1" if dry_run is None else dry_run
limit = max(1, min(max_results, 10))
if offline:
results = _mock_search(query, limit)
return ToolResult(ok=True, data={"query": query, "results": results}).to_dict()
try:
from tavily import TavilyClient # 需先安裝 tavily-python
client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
raw = client.search(query=query, max_results=limit, search_depth="basic")
results = [
{
"title": item.get("title", ""),
"url": item.get("url", ""),
"content": item.get("content", ""),
}
for item in raw.get("results", [])
]
return ToolResult(ok=True, data={"query": query, "results": results}).to_dict()
except KeyError:
return ToolResult(ok=False, error="缺少 TAVILY_API_KEY", retryable=False).to_dict()
except Exception as exc:
return ToolResult(ok=False, error=f"搜尋失敗:{type(exc).__name__}", retryable=True).to_dict()
段落說明:limit = max(1, min(max_results, 10)) 是防止模型要求一次抓一百筆而失控的護欄。search_depth="basic" 依官方文件有不同深度可選,較深的模式會回傳更完整的內容但也更貴,請依需求與官方文件選擇。最後把例外轉成 ToolResult 是關鍵——即使外部服務掛掉,代理也能拿到可解讀的錯誤訊息。
第三步,實作擷取工具。使用 AG Day 21 文件擷取與分塊:從網頁到知識庫(原文連結)建立的清洗流程,加上逾時、重試與大小限制:
import httpx
import trafilatura
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
HEADERS = {"User-Agent": "research-agent/0.1(教學示範)"}
MAX_CHARS = 20000
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=0.5, min=0.5, max=8),
retry=retry_if_exception_type(httpx.TransportError),
reraise=True,
)
def _download(url: str) -> str:
with httpx.Client(timeout=httpx.Timeout(15.0), follow_redirects=True, headers=HEADERS) as client:
response = client.get(url)
response.raise_for_status()
return response.text
def fetch_page(url: str, dry_run: bool | None = None) -> dict[str, Any]:
"""抓取單一網址並抽取出可引用的正文。"""
offline = os.environ.get("RESEARCH_AGENT_DRY_RUN") == "1" if dry_run is None else dry_run
if offline:
return ToolResult(
ok=True,
data={"url": url, "title": "(示範頁面)", "text": "這是離線示範正文,非真實網頁內容。"},
).to_dict()
try:
html = _download(url)
text = trafilatura.extract(html, include_comments=False, include_tables=True)
if not text:
return ToolResult(ok=False, error="無法從頁面抽出正文", retryable=False).to_dict()
truncated = text[:MAX_CHARS]
return ToolResult(
ok=True,
data={"url": url, "title": "", "text": truncated, "truncated": len(text) > MAX_CHARS},
).to_dict()
except httpx.HTTPStatusError as exc:
code = exc.response.status_code
return ToolResult(ok=False, error=f"HTTP {code}", retryable=code in (429, 500, 502, 503)).to_dict()
except Exception as exc:
return ToolResult(ok=False, error=f"擷取失敗:{type(exc).__name__}", retryable=True).to_dict()
段落說明:retry 只針對傳輸層錯誤重試,HTTP 狀態錯誤則交由 fetch_page 判斷是否為可重試的 429 或 5xx,這樣就不會對 404 白等三次。MAX_CHARS 是護欄,截斷時回報 truncated=True,讓上層知道內容不完整而不是誤以為全部到齊。
第四步,把抓到的內容真的寫進知識庫,並且做到冪等。這一步是今天最關鍵的整合點:
import hashlib
import sqlite3
import uuid
from datetime import datetime, timezone
from . import storage
CHUNK_SIZE = 800
def _normalize_url(url: str) -> str:
return url.split("#", 1)[0].rstrip("/")
def _sha256(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def _split_chunks(text: str, size: int = CHUNK_SIZE) -> list[str]:
paragraphs = [p.strip() for p in text.split("\n") if p.strip()]
chunks: list[str] = []
buffer = ""
for paragraph in paragraphs:
if len(buffer) + len(paragraph) + 1 > size and buffer:
chunks.append(buffer)
buffer = paragraph
else:
buffer = f"{buffer}\n{paragraph}" if buffer else paragraph
if buffer:
chunks.append(buffer)
return chunks
def save_page(conn: sqlite3.Connection, url: str, title: str, text: str) -> str:
"""冪等寫入:同一網址同內容不重複新增;內容變更則更新並記錄。"""
doc_id = _normalize_url(url)
digest = _sha256(text)
row = conn.execute(
"SELECT content_hash FROM documents WHERE id = ?", (doc_id,)
).fetchone()
if row is not None and row["content_hash"] == digest:
return doc_id
fetched_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
conn.execute(
"INSERT OR REPLACE INTO documents (id, url, title, fetched_at, content_hash)"
" VALUES (?, ?, ?, ?, ?)",
(doc_id, url, title, fetched_at, digest),
)
conn.execute("DELETE FROM chunks WHERE doc_id = ?", (doc_id,))
for order, chunk in enumerate(_split_chunks(text)):
conn.execute(
"INSERT INTO chunks (id, doc_id, ord, text) VALUES (?, ?, ?, ?)",
(f"{doc_id}#{order}", doc_id, order, chunk),
)
conn.commit()
return doc_id
段落說明:doc_id 直接使用正規化後的網址,讓「同一頁」天然變成同一個識別;內容雜湊相同則整段跳過,避免重複入庫。這裡先寫 SQLite,實務上還要接著把 chunks 送去向量化寫入 Chroma(AG Day 20 建立的流程);為避免範例過長,這一步在下一篇整合章節會補齊。另外提醒:以網址當主鍵是簡化做法,若遇到內容會隨時間變動的頁面,建議改用「網址加時間戳的複合識別」,這部分依你的需求自行調整。
第五步,把兩個工具包成 function calling 的 schema,讓模型能選擇呼叫:
SEARCH_TOOL_SCHEMA = {
"name": "search_web",
"description": "搜尋網路以發現可能的來源。回傳標題、網址與摘要,適合先廣泛探索。",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜尋查詢字串"},
"max_results": {"type": "integer", "minimum": 1, "maximum": 10, "default": 5},
},
"required": ["query"],
},
}
FETCH_TOOL_SCHEMA = {
"name": "fetch_page",
"description": "抓取指定網址的正文並寫入知識庫。回傳 doc_id,後續用檢索工具取得內容。",
"parameters": {
"type": "object",
"properties": {"url": {"type": "string", "description": "要抓取的網址"}},
"required": ["url"],
},
}
段落說明:描述文字要寫給模型看,所以「什麼時候用」比「它做什麼」更重要。搜尋的描述強調「先廣泛探索」,擷取的描述強調「寫入知識庫後用檢索取得」,這兩句話會直接影響模型的分工決策。參數的 minimum/maximum 是給模型的第一層護欄,程式端的 min/max 是第二層——兩層都要有,不要只靠提示。
第六步,寫一個把兩者串起來的迷你流程,示範「搜尋到候選、決定抓哪一頁、入庫」的閉環。這裡不依賴模型也能跑,方便你驗證工具本身:
def research_urls(query: str, top_n: int = 2, dry_run: bool | None = None) -> list[str]:
"""搜尋後挑前 top_n 個結果抓取並入庫,回傳已入庫的 doc_id。"""
found = search_web(query, max_results=top_n + 3, dry_run=dry_run)
if not found.get("ok"):
print("搜尋失敗:", found.get("error"))
return []
saved: list[str] = []
conn = storage.connect()
try:
for item in found["data"]["results"][:top_n]:
page = fetch_page(item["url"], dry_run=dry_run)
if not page.get("ok"):
print("略過:", item["url"], page.get("error"))
continue
doc_id = save_page(conn, page["data"]["url"], page["data"]["title"], page["data"]["text"])
saved.append(doc_id)
finally:
conn.close()
return saved
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="搜尋與擷取工具")
parser.add_argument("action", choices=["search", "fetch", "research"])
parser.add_argument("value", help="查詢字串或網址")
parser.add_argument("--dry-run", action="store_true", help="離線模式,輸出為示範")
args = parser.parse_args()
if args.action == "search":
print(json.dumps(search_web(args.value, dry_run=args.dry_run), ensure_ascii=False, indent=2))
elif args.action == "fetch":
print(json.dumps(fetch_page(args.value, dry_run=args.dry_run), ensure_ascii=False, indent=2))
else:
print(research_urls(args.value, dry_run=args.dry_run))
段落說明:research_urls() 是「工具層」的最小整合示範;真正由模型決定要抓哪幾頁的版本,會放在 AG Day 28 的 MCP 客戶端與 LangGraph 流程裡。執行方式與示範輸出如下:
uv run python -m research_agent.tools research "邊緣運算功耗" --dry-run
['https://example.invalid/edge-power']
段落說明:上面輸出中的網址是 dry-run 示範語料,不是真實網址,也不代表 Tavily 的實際回傳格式。要驗證真實路徑,請設定 TAVILY_API_KEY 後拿掉 --dry-run,並先確認服務額度與授權條款。搜尋結果屬於外部內容,正式使用前請確認你的使用情境符合該服務的條款。
常見錯誤與踩雷
第一個錯誤是把搜尋摘要當成事實來源。摘要來自搜尋服務的抽取與壓縮,可能有誤差或省略脈絡;如果你讓模型直接引用摘要,就會產生「看起來有出處、其實沒讀過原文」的報告。正確做法是搜尋只用來決定要抓哪些頁,引用一律回到擷取後的正文段落。
第二個錯誤是把整份 HTML 塞進模型。除了成本與上下文上限之外,HTML 標籤、導覽列、廣告文字都會干擾模型判斷。務必經過 trafilatura.extract() 這類清洗步驟,並設定字數上限;我習慣在擷取工具裡就截斷,而不是留到上層才處理,因為上層很容易忘記。
第三個錯誤是工具遇錯就抛例外。httpx 的 raise_for_status() 很好用,但要把拋出的例外在工具邊界攔下來、轉成結構化錯誤。否則一次 429 就讓整個研究任務失敗——而 429 在搜尋服務上其實非常常見。
第四個錯誤是沒有做網址正規化與去重。帶有 #fragment、追蹤參數、結尾斜線的網址會被視為不同頁面,造成同一份內容重複入庫。請先做基本正規化(去掉 fragment、統一結尾),再當成識別。
第五個錯誤是把 API key 寫進程式或提交進版本控制。金鑰一律從環境變數讀取(本系列統一用 TAVILY_API_KEY),並在缺少時回傳明確的設定提示而不是神秘錯誤。另外要記得:git add 前檢查有沒有不小心加入 .env。
第六個錯誤是忽略 robots、服務條款與著作權。自動抓取不是無限制的權利,請尊重來源網站的 robots.txt、服務條款與授權,避免高頻抓取造成對方負擔。本系列是教學示範,正式產品請諮詢法務並做好速率控制。
第七個錯誤是忘記記錄抓取時間。少了時間戳,之後就無法回答「這份資料是什麼時候的快照」,而時效性在研究場景中往往是關鍵。我們在 save_page() 裡已經寫入 fetched_at,請不要為了省一個欄位而拿掉它。
效能與實務提醒
先講速率與配額。搜尋服務通常以「每分鐘請求數」與「每月點數」雙重限制;擷取則是對目標網站造成負擔。實務上我建議三個設計:對同一查詢做短期快取(例如五分鐘),對同一網址做永久快取並用內容雜湊判斷是否需要重抓,以及把所有外部呼叫集中到一個閘道模組,方便統一加限速與觀測。這樣要調整節流策略時只需要改一個地方。
再講並行。抓取多頁本質上是可並行的 I/O 工作,用 asyncio 搭配 httpx.AsyncClient 可以大幅縮短總時間;但並行數要設上限(例如同時 5 個),否則很容易觸發對方的速率限制。這裡的取捨是「並行換延遲」對「限速換穩定」,建議從保守值開始,觀察錯誤率再往上調。
第三個提醒是成本可觀測性。搜尋與擷取都是外部計費服務,請從第一天就記錄「每次任務用了幾次搜尋、幾頁擷取、多少字元」,寫進 SQLite 的 events 表或未來的追蹤平台(AG Day 35 追蹤平台:Langfuse 觀測實戰(原文連結)會接手)。當一份報告的成本異常升高時,通常就是代理在重複搜尋或抓取失敗重試,這些紀錄是你唯一的線索。
最後提醒資料衛生。網路來源良莠不齊,請在入庫時保留來源網域,讓後續檢索與報告可以依網域做加權(例如優先官方與學術來源)。這類政策不需要寫死在程式裡,而是放在設定檔,方便不同研究主題調整。下一篇我們會把這一整套工具從「自家函式」升級為「標準協定」,讓它們能被任何支援該協定的客戶端使用。
小結
今天我們讓研究助理跨出了知識庫的邊界。重點回顧:
- 工具契約:名稱、描述、參數 schema、回傳格式、錯誤語意五件事缺一不可,尤其錯誤要以回傳值呈現而非例外。
- 搜尋與擷取分離:
search_web負責廣度、fetch_page負責深度,先搜尋篩選再擷取,控制成本與延遲。 - 冪等入庫:以正規化網址當文件識別、以內容雜湊判斷是否更新,避免重複與膨脹。
- 大小與安全護欄:
max_results上限、正文截斷、逾時與有限重試、金鑰只從環境變數讀取。 - 可追溯:抓取時間與來源一併保存,讓新抓到的內容能直接接上昨天的引用鏈路。
至此,研究助理已經具備「自己找資料」的能力。但這些工具目前只能被我們自己的程式呼叫;如果哪天你想讓別的代理、別的客戶端也能用同一組工具,就得再複製一份程式,這顯然不合理。明天的協定,就是為了解決這個問題。
結語
回頭看,我們從 Day 20 到 Day 24 建立了一條完整的研究鏈路:擷取、分塊、向量化、改寫與精排、引用與查核、主動搜尋。這條鏈路的每個工具目前都是專案內部函式,只有這份程式碼能呼叫它們。接下來我們要把視野放大到整個生態系:讓工具不再綁在單一應用裡,而是透過標準協定被任何支援的客戶端使用,這也是 2026 年 AI 代理工程最受矚目的基礎建設之一。
明天,我們會進入「AG Day 25 MCP 概念:Model Context Protocol 架構」,從頭拆解這個協定的角色分工、訊息格式與傳輸方式。我們不會馬上寫 FastMCP,而是先親手用最原始的方式走一遍握手與工具呼叫流程,把底層機制看清楚,之後用框架才不會覺得是黑盒子。
延伸資源
- Tavily 官方文件:
https://docs.tavily.com/。搜尋與擷取 API 的參數、回傳欄位與額度規範,一律以這份文件為準。 - trafilatura 官方文件:
https://trafilatura.readthedocs.io/。正文抽取的參數與輸出格式說明。 - httpx 官方文件:
https://www.python-httpx.org/。逾時設定、重新導向與非同步客戶端用法。 - tenacity 官方文件:
https://tenacity.readthedocs.io/。重試策略、退避與例外條件的組合方式。 - Model Context Protocol 官方網站:
https://modelcontextprotocol.io/。明天的架構篇會以 2025-06-18 版規格為主軸。
留言
張貼留言