跳到正文
DEV Community· EME GUG·· 2 小時前AI 評分62

如何用 Python 為 LLM Agent 建立搜尋與抓取層：快取、內容提取與成本控制

Building a Search & Fetch Layer for LLM Agents: Caching, Clean Extraction and Cost Control in Python

AI 導讀

作者分享以 Python 為 LLM Agent 建立搜尋與抓取閘道的方法，透過快取、預算控制、robots.txt 檢查及正文提取管理外部請求。文中以 trafilatura 提取正文，將一個 VnExpress 頁面的 HTML 由約 180KB 縮至約 6KB，並把輸出限制在 8000 個字元。作者稱內部問答工作負載的快取命中率穩定在 35-45%，並建議只對實際外部抓取計入預算。

正文

Tuần này trên Hacker News có một bài khá hot: DeepSeek ra API search và fetch với giá $0.75 cho 1000 request thành công. Đọc comment thì thấy nhiều người tính nhẩm ngay: agent của mình mỗi câu hỏi gọi 5-10 lần search, mỗi lần lại fetch 3-4 trang, vậy một ngày tốn bao nhiêu? Cùng lúc đó lại có tranh cãi về điều khoản scraping của Meta. Hai chuyện này gộp lại cho thấy một điều: lớp search/fetch của agent đang dần thành hạ tầng có chi phí thật và có trách nhiệm pháp lý thật, không còn là một hàm requests.get() viết vội nữa.

Mình đã viết lại lớp này cho mấy dự án nội bộ vài lần rồi. Bài này chia sẻ cách mình build một lớp fetch “tử tế”: tôn trọng robots.txt, lấy nội dung sạch để tiết kiệm token, có cache và giới hạn ngân sách. Dùng API trả phí hay tự host thì các nguyên tắc này vẫn đúng.

1. Kiến trúc tổng quan: đừng để agent gọi thẳng ra internet

Lỗi phổ biến nhất mình hay gặp: đưa cho LLM một tool fetch_url(url) rồi để nó gọi thẳng HTTP. Kết quả là agent fetch lại cùng một trang 4 lần trong một phiên, nhét nguyên 200KB HTML (có cả menu, footer, script tracking) vào context, và chẳng ai biết tháng đó đã tiêu bao nhiêu tiền.

Cách mình làm là thêm một lớp trung gian:

flowchart LR
    A[LLM Agent] -->|tool call| B[Fetch Gateway]
    B --> C{Cache hit?}
    C -->|Yes| H[Return cached text]
    C -->|No| D{Budget OK?}
    D -->|No| X[Reject + reason]
    D -->|Yes| E{robots.txt allow?}
    E -->|No| X
    E -->|Yes| F[HTTP fetch]
    F --> G[Extract main content]
    G --> I[(SQLite cache)]
    I --> H

Gateway này có 4 nhiệm vụ: cache, budget, compliance (robots.txt, rate limit), và extraction. Agent chỉ nhận về text sạch, có giới hạn độ dài. Nếu bị từ chối thì agent nhận lý do rõ ràng để tự xoay xở, không bị crash.

2. Fetch có trách nhiệm: robots.txt, User-Agent và extraction

Python có sẵn urllib.robotparser trong stdlib, không cần cài thêm. Phần extraction thì mình dùng trafilatura 2.0. Mình đã thử readability-lxml và newspaper4k, nhưng trafilatura cho kết quả ổn định nhất với trang tin tức và blog tiếng Việt.

python3.12 -m venv .venv && source .venv/bin/activate
pip install httpx==0.27.2 trafilatura==2.0.0

import httpx
import trafilatura
from urllib.parse import urlparse
from urllib.robotparser import RobotFileParser
from functools import lru_cache

UA = 'MyAgentBot/1.0 (+contact: bot-admin@example.com)'
MAX_CHARS = 8000  # ~2-3k token, đủ cho hầu hết câu hỏi

@lru_cache(maxsize=512)
def get_robots(origin: str) -> RobotFileParser:
    rp = RobotFileParser()
    try:
        r = httpx.get(f'{origin}/robots.txt', headers={'User-Agent': UA}, timeout=5)
        rp.parse(r.text.splitlines() if r.status_code == 200 else [])
    except httpx.HTTPError:
        rp.parse([])  # không đọc được robots -> coi như cho phép, tùy policy của bạn
    return rp

def fetch_clean(url: str) -> dict:
    p = urlparse(url)
    if p.scheme not in ('http', 'https'):
        return {'ok': False, 'reason': 'unsupported scheme'}
    origin = f'{p.scheme}://{p.netloc}'
    if not get_robots(origin).can_fetch(UA, url):
        return {'ok': False, 'reason': 'disallowed by robots.txt'}

    r = httpx.get(url, headers={'User-Agent': UA}, timeout=15, follow_redirects=True)
    if r.status_code != 200 or 'text/html' not in r.headers.get('content-type', ''):
        return {'ok': False, 'reason': f'status {r.status_code}'}

    text = trafilatura.extract(r.text, include_comments=False, include_tables=True,
                               output_format='markdown', url=url)
    if not text:
        return {'ok': False, 'reason': 'no main content'}
    return {'ok': True, 'url': str(r.url), 'text': text[:MAX_CHARS],
            'truncated': len(text) > MAX_CHARS}

Vài điểm mình rút ra sau khi chạy thực tế:

  • User-Agent phải định danh được và có thông tin liên hệ. Giả làm Chrome có thể qua được vài cái WAF, nhưng đó đúng là kiểu hành vi đang bị các điều khoản kiểu của Meta nhắm tới. Nếu làm sản phẩm thương mại thì đừng làm vậy.
  • Output dạng markdown thay vì plain text. LLM hiểu heading và bảng tốt hơn hẳn, mà token vẫn gần như không đổi.
  • Extraction giảm token rất mạnh. Trên một trang VnExpress mình đo được HTML thô khoảng 180KB, sau trafilatura còn khoảng 6KB. Nếu trả tiền theo token thì đây là khoản tiết kiệm lớn nhất, lớn hơn cả tiền API fetch.
  • Thêm chặn SSRF: không cho fetch IP private (10.x, 192.168.x, 169.254.169.254). Agent mà bị prompt injection thì có thể bị lừa đi đọc metadata endpoint của cloud.

3. Cache và budget: chỗ tiết kiệm tiền thật sự

Với mức $0.75/1000 request thì từng request nghe có vẻ rẻ. Nhưng agent chạy vòng lặp thì 50k request/tháng đến rất nhanh, nhất là khi có bug khiến agent retry vô hạn. Mình dùng SQLite vì không cần thêm service nào, lại đủ nhanh cho vài trăm request mỗi phút.

import sqlite3, json, time, hashlib

class FetchGateway:
    def __init__(self, db='fetch_cache.db', ttl=6*3600, daily_budget=2000):
        self.db = sqlite3.connect(db, check_same_thread=False)
        self.db.execute('PRAGMA journal_mode=WAL')
        self.db.execute('CREATE TABLE IF NOT EXISTS cache (k TEXT PRIMARY KEY, v TEXT, ts REAL)')
        self.db.execute('CREATE TABLE IF NOT EXISTS usage (day TEXT PRIMARY KEY, n INTEGER)')
        self.ttl, self.budget = ttl, daily_budget

    def _key(self, url):
        return hashlib.sha256(url.strip().lower().encode()).hexdigest()

    def _spend(self) -> bool:
        day = time.strftime('%Y-%m-%d')
        row = self.db.execute('SELECT n FROM usage WHERE day=?', (day,)).fetchone()
        if row and row[0] >= self.budget:
            return False
        self.db.execute('INSERT INTO usage VALUES (?,1) ON CONFLICT(day) DO UPDATE SET n=n+1', (day,))
        self.db.commit()
        return True

    def fetch(self, url: str) -> dict:
        k = self._key(url)
        row = self.db.execute('SELECT v, ts FROM cache WHERE k=?', (k,)).fetchone()
        if row and time.time() - row[1] < self.ttl:
            return {**json.loads(row[0]), 'cached': True}
        if not self._spend():
            return {'ok': False, 'reason': 'daily fetch budget exceeded'}
        result = fetch_clean(url)
        if result['ok']:
            self.db.execute('REPLACE INTO cache VALUES (?,?,?)', (k, json.dumps(result), time.time()))
            self.db.commit()
        return result

Với workload hỏi đáp nội bộ của team mình, cache hit rate ổn định ở mức 35-45%, vì người dùng hay hỏi lặp lại cùng chủ đề trong ngày. Nói cách khác, gần một nửa chi phí fetch biến mất chỉ nhờ khoảng 30 dòng code.

Một lưu ý: chỉ đếm budget khi thật sự gọi ra ngoài, còn cache hit thì không tính. Ngoài ra, trả lỗi budget exceeded về cho agent dưới dạng dữ liệu, đừng raise exception. LLM đời mới đủ thông minh để trả lời “mình không tra thêm được, dựa trên những gì đã có thì…” thay vì treo.

4. Nối vào agent: tool schema và luồng gọi 

sequenceDiagram
    participant U as User
    participant A as Agent
    participant G as FetchGateway
    participant W as Website
    U->>A: Câu hỏi
    A->>G: fetch(url)
    G->>G: check cache / budget / robots
    G->>W: GET (nếu cần)
    W-->>G: HTML
    G-->>A: markdown sạch (<=8000 chars)
    A-->>U: Câu trả lời + nguồn

Trong tool description, mình ghi rõ cho model biết giới hạn: “Returns cleaned markdown, max 8000 chars. May return cached content up to 6h old. Do not fetch the same URL twice.” Nghe thì đơn giản, nhưng chỉ một dòng này đã làm số lần gọi trùng giảm thấy rõ. Thêm nữa, nên log mọi tool call vào một bảng riêng (url, latency, cached, tokens) để cuối tuần mở ra xem agent đang tiêu tiền vào đâu. Mình hay chạy nhanh:

sqlite3 fetch_cache.db "SELECT day, n FROM usage ORDER BY day DESC LIMIT 7;"

Kết luận

API search/fetch rẻ đi là tin tốt, nhưng chi phí thật của agent nằm ở số lần gọi và lượng token bạn nhét vào context. Những việc nên làm ngay:

  1. Đừng cho agent gọi HTTP trực tiếp. Luôn đi qua một gateway, kể cả khi gateway chỉ có 100 dòng code.
  2. Extract nội dung chính bằng trafilatura (hoặc tương đương) trước khi đưa cho LLM, rồi cắt cứng theo số ký tự.
  3. Cache với TTL hợp lý. Tin tức thì 1-6h, tài liệu kỹ thuật thì 24h hoặc lâu hơn. SQLite + WAL là đủ dùng cho đa số team nhỏ.
  4. Đặt budget theo ngày và trả lỗi dưới dạng dữ liệu cho agent, đừng để exception làm vỡ vòng lặp.
  5. Tôn trọng robots.txt, dùng User-Agent định danh, chặn SSRF. Làm vậy để đúng đạo đức, và cũng để sản phẩm của bạn không thành case study trong bài HN tiếp theo về scraping.

Nếu dùng API trả phí như của DeepSeek, bạn vẫn nên giữ nguyên gateway này và chỉ thay hàm fetch_clean bằng client của provider. Cache và budget vẫn là thứ bạn phải tự kiểm soát.

來源：DEV Community · dev.to

#教學/實踐#Agent#搜尋#部署/工程