AI agent farm: 10 agent Claude song song scrape news Việt Nam

Marketing team cần news brief mỗi sáng: 5 nguồn tin lớn (vnexpress, tuoitre, dantri, vietnamnet, cafef), 5 nguồn ngành (ictnews, vneconomy, baomoi, zing, viettan). Mỗi nguồn 50-100 bài/ngày. Đọc tay = 2 giờ/ngày của 1 analyst. Tự build crawler đơn giản + AI tóm tắt = 5 phút mỗi sáng nhìn dashboard, có tin nóng thì click sâu. Đây là use case kinh điển của "agent farm" - nhiều AI agent chạy song song xử lý task lặp.

Bài này hướng dẫn build agent farm 10 Claude agent trên 1 VPS để scrape + AI process news VN. Cấu trúc bằng Python asyncio + Redis queue + Postgres storage. Không over-engineer như multi-agent framework (CrewAI, AutoGen) - mục tiêu là production simple cho workload thực tế, dưới 500 dòng code.

Vì sao 10 agent thay vì 1 monolithic crawler?

• Mỗi nguồn có HTML structure khác nhau, tách agent cho dễ maintain.
• 1 nguồn down (vnexpress chậm) không block 9 nguồn khác.
• Parallel I/O: 10 agent fetch HTTP song song, tận dụng bandwidth.
• Mỗi agent có rate limit riêng, không bị ban IP do tổng request quá tải.
• Dễ scale: nhân lên 20 agent khi thêm nguồn, không refactor logic chính.

Kiến trúc tổng thể

+----------+        +-----------------+        +----------+
| Cron Hub | -----> | Redis Queue     | -----> | 10 Agent |
| (hourly) |        | "fetch_jobs"    |        | Workers  |
+----------+        +-----------------+        +----------+
                          ^                          |
                          |                          v
                    +-----+-----+              +-----------+
                    | Postgres  | <----------- | Claude API|
                    | articles  |  summary     +-----------+
                    +-----------+
                          |
                          v
                    +-----------+
                    | Dashboard |
                    | (Grafana) |
                    +-----------+

Cron mỗi giờ push 10 job (mỗi job 1 source) vào Redis list. 10 worker process pull job, fetch RSS/HTML, parse article, gọi Claude tóm tắt, lưu Postgres. Dashboard Grafana đọc Postgres render bảng news brief.

Yêu cầu VPS

VPS 4 GB RAM / 2-4 vCPU đủ dùng. Bandwidth không cần cao, mỗi article HTML 100-300 KB. Cloud VPS TND gói khoảng giữa range 199k-3190k/tháng là vừa.

Cấu trúc source code

/srv/news-farm/
├── docker-compose.yml
├── .env
├── sources.yaml          # khai báo 10 nguồn
├── worker.py             # main worker
├── parser/               # 1 parser per source
│   ├── vnexpress.py
│   ├── tuoitre.py
│   ├── cafef.py
│   └── ...
├── claude_client.py
├── db.py
└── enqueue.py            # cron job push task

sources.yaml: khai báo nguồn

sources:
  - id: vnexpress
    rss: https://vnexpress.net/rss/tin-moi-nhat.rss
    parser: vnexpress
    rate_limit_per_min: 30
    enabled: true

  - id: tuoitre
    rss: https://tuoitre.vn/rss/tin-moi-nhat.rss
    parser: tuoitre
    rate_limit_per_min: 20
    enabled: true

  - id: cafef
    rss: https://cafef.vn/trang-chu.rss
    parser: cafef
    rate_limit_per_min: 30
    enabled: true

  - id: vietnamnet
    rss: https://vietnamnet.vn/rss/tin-moi.rss
    parser: vietnamnet
    rate_limit_per_min: 20
    enabled: true

  # ... thêm 6 nguồn nữa

Tách config khỏi code giúp PM/analyst tự thêm nguồn mà không sửa Python. Khi 1 nguồn down dài hạn, set enabled: false.

worker.py: main loop async

import asyncio
import json
import os
import logging
import hashlib
import yaml
import aiohttp
import redis.asyncio as redis
from anthropic import AsyncAnthropic

from db import save_article, exists_by_hash
from parser import get_parser

logging.basicConfig(level=logging.INFO)

R = redis.from_url(os.environ["REDIS_URL"])
CLAUDE = AsyncAnthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
MODEL = os.environ.get("CLAUDE_MODEL", "claude-haiku-4-5")

with open("sources.yaml") as f:
    SOURCES = {s["id"]: s for s in yaml.safe_load(f)["sources"]}

QUEUE = "fetch_jobs"

PROMPT_TEMPLATE = """Bạn là biên tập viên tin tức.

Cho bài báo tiếng Việt sau, hãy trả về JSON với cấu trúc:
{{
  "title": "tiêu đề gốc",
  "summary": "tóm tắt 3-4 câu (50-80 từ tiếng Việt)",
  "category": "1 trong: kinh-te, cong-nghe, chinh-tri, the-thao, giai-tri, suc-khoe, khac",
  "sentiment": "1 trong: positive, neutral, negative",
  "key_entities": ["entity1", "entity2"],
  "newsworthiness": 1-10
}}

Bài báo:
TIEU DE: {title}
NOI DUNG: {content}

Chỉ trả về JSON, không giải thích."""

async def call_claude(title: str, content: str) -> dict:
    content = content[:8000]  # giới hạn để tránh tốn token
    prompt = PROMPT_TEMPLATE.format(title=title, content=content)
    msg = await CLAUDE.messages.create(
        model=MODEL,
        max_tokens=600,
        messages=[{"role": "user", "content": prompt}],
    )
    raw = msg.content[0].text.strip()
    if raw.startswith("```"):
        raw = raw.strip("`").lstrip("json").strip()
    return json.loads(raw)

async def process_article(source_id: str, url: str, http: aiohttp.ClientSession):
    url_hash = hashlib.sha256(url.encode()).hexdigest()
    if await exists_by_hash(url_hash):
        return
    parser = get_parser(SOURCES[source_id]["parser"])
    try:
        async with http.get(url, timeout=20) as r:
            html = await r.text()
        article = parser.parse(html)
        if not article or not article.get("content"):
            return
        analysis = await call_claude(article["title"], article["content"])
        await save_article(
            source_id=source_id,
            url=url,
            url_hash=url_hash,
            title=article["title"],
            content=article["content"],
            published_at=article.get("published_at"),
            **analysis,
        )
        logging.info(f"[{source_id}] saved: {article['title'][:60]}")
    except Exception as e:
        logging.exception(f"[{source_id}] {url}: {e}")

async def fetch_source(source_id: str, http: aiohttp.ClientSession):
    src = SOURCES[source_id]
    async with http.get(src["rss"], timeout=15) as r:
        rss_text = await r.text()
    import feedparser
    feed = feedparser.parse(rss_text)
    sem = asyncio.Semaphore(3)  # 3 article song song trong 1 source
    async def bounded(url):
        async with sem:
            await process_article(source_id, url, http)
    await asyncio.gather(*[bounded(e.link) for e in feed.entries[:30]])

async def worker(worker_id: int):
    async with aiohttp.ClientSession() as http:
        while True:
            job = await R.brpop(QUEUE, timeout=10)
            if not job:
                continue
            _, source_id = job
            source_id = source_id.decode()
            logging.info(f"worker-{worker_id} picked {source_id}")
            try:
                await fetch_source(source_id, http)
            except Exception as e:
                logging.exception(f"worker-{worker_id} failed {source_id}: {e}")

async def main():
    n = int(os.environ.get("WORKER_CONCURRENCY", "10"))
    await asyncio.gather(*[worker(i) for i in range(n)])

if __name__ == "__main__":
    asyncio.run(main())

enqueue.py: cron mỗi giờ

import os, yaml
import redis

R = redis.from_url(os.environ["REDIS_URL"])
with open("sources.yaml") as f:
    sources = yaml.safe_load(f)["sources"]

for s in sources:
    if s["enabled"]:
        R.lpush("fetch_jobs", s["id"])
        print(f"queued {s['id']}")

Crontab: 0 * * * * cd /srv/news-farm && .venv/bin/python enqueue.py

parser/vnexpress.py: ví dụ 1 parser

from bs4 import BeautifulSoup
from datetime import datetime

def parse(html: str) -> dict:
    soup = BeautifulSoup(html, "lxml")
    title_el = soup.select_one("h1.title-detail") or soup.select_one("h1")
    if not title_el:
        return {}
    title = title_el.get_text(strip=True)

    content_parts = []
    for p in soup.select("article.fck_detail p"):
        text = p.get_text(strip=True)
        if text:
            content_parts.append(text)
    content = "n".join(content_parts)

    date_el = soup.select_one("span.date")
    published_at = date_el.get_text(strip=True) if date_el else None

    return {
        "title": title,
        "content": content,
        "published_at": published_at,
    }

Mỗi nguồn 1 file parser tương tự, selector khác. Dùng class CSS đặc trưng của từng site. Khi site đổi layout, chỉ sửa 1 file parser, các nguồn khác vẫn chạy.

db.py: lưu Postgres

import asyncpg
import os
import json

POOL = None

async def init_pool():
    global POOL
    POOL = await asyncpg.create_pool(os.environ["DATABASE_URL"], max_size=10)

async def exists_by_hash(h: str) -> bool:
    async with POOL.acquire() as c:
        r = await c.fetchval("SELECT 1 FROM articles WHERE url_hash=$1", h)
    return r is not None

async def save_article(**kw):
    async with POOL.acquire() as c:
        await c.execute("""
            INSERT INTO articles
            (source_id, url, url_hash, title, content, published_at,
             summary, category, sentiment, key_entities, newsworthiness)
            VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
            ON CONFLICT (url_hash) DO NOTHING
        """,
        kw["source_id"], kw["url"], kw["url_hash"], kw["title"], kw["content"],
        kw.get("published_at"), kw["summary"], kw["category"], kw["sentiment"],
        json.dumps(kw["key_entities"]), kw["newsworthiness"])

Schema bảng articles: id, source_id, url, url_hash UNIQUE, title, content, summary, category, sentiment, key_entities JSONB, newsworthiness INT, created_at, published_at. Index trên url_hash và created_at + category cho query nhanh.

Dashboard Grafana

Connect Grafana tới Postgres, tạo dashboard "News Brief" với panel:

• Bảng "Top 20 news 1 giờ qua" sort theo newsworthiness DESC.
• Pie chart phân loại category.
• Time series số bài / source / giờ.
• Bảng sentiment negative cao - để PR/marketing alert.
• Heat map giờ đăng bài / source - biết khi nào nên check.

PM mở Grafana mỗi sáng 9h, đọc top 20 trong 5 phút, sau đó share Slack nếu có tin nóng.

Cost ước tính

• 500-1000 article/ngày, mỗi bài ~2k input + 500 output token Claude Haiku.
• Haiku rất rẻ (~1 USD/M input, 5 USD/M output) -> ~1.5 USD/ngày = ~45 USD/tháng.
• VPS 4 GB: ~500k VND/tháng.
• Tổng < 1.5 triệu/tháng cho 1 agent farm 10 source xử lý 30k bài/tháng.

So sánh: dịch vụ news monitoring SaaS như Meltwater, Mention, Brand24 từ 300-3000 USD/tháng. Self-host agent farm rẻ hơn 10-50x cho cùng feature cốt lõi (chưa tính social monitoring).

Bảo mật và lưu ý pháp lý

• Tôn trọng robots.txt của từng nguồn. Một số trang cấm crawl, đừng bypass.
• User-Agent rõ ràng (đặt email contact để webmaster liên hệ nếu cần).
• Rate limit chặt: 1-3 req/giây/source, không DDoS.
• Crawl content public, không scrape paywall.
• Tóm tắt + đường link gốc khi share - tôn trọng bản quyền.
• Không tái xuất bản nguyên văn (vi phạm bản quyền), chỉ summary nội bộ.

Mở rộng và biến thể

• Thêm Twitter/X agent: dùng twscrape pull tweet keyword.
• Thêm Reddit/HackerNews agent: dùng API chính thức.
• Thêm Telegram channel scraper bằng telethon.
• Translation agent: dịch article tiếng Trung/Anh sang Việt trước khi summary.
• Trend detection: chạy embedding + clustering hằng đêm để phát hiện topic đang nóng.
• Alert: khi sentiment negative + entity là brand bạn -> Slack ngay lập tức.

FAQ

Vì sao chọn Claude Haiku thay vì Sonnet/Opus?

Task tóm tắt + phân loại không cần reasoning sâu, Haiku đủ tốt và rẻ hơn 10-30 lần. Nếu yêu cầu nhiều ngữ cảnh hoặc cross-reference giữa các bài, nâng lên Sonnet. Opus chỉ cần khi viết bài tổng hợp deep insight cho leadership. Mỗi tier cost x10, suy nghĩ kỹ trước khi nâng.

Có rủi ro Claude trả về JSON không hợp lệ không?

Có, tỉ lệ 1-3% với Haiku. Mitigation: dùng response_format JSON mode khi available, fallback try/except json.loads + retry 1 lần. Hoặc dùng tool use với schema chặt - Claude bắt buộc trả đúng schema. Hoặc cấu trúc prompt yêu cầu chỉ JSON và parse với regex extract.

Khi nào nên dùng framework như CrewAI, AutoGen?

Khi agent cần "trao đổi" với nhau qua nhiều turn để giải task phức tạp (research, plan, execute, review). Use case scrape news thì 1 agent / 1 task là đủ, framework chỉ thêm boilerplate. CrewAI/AutoGen phát huy khi build assistant đa năng, agent business analysis, planner-executor pattern.

Có cách nào tránh dedupe bị bỏ sót khi URL khác nhưng nội dung giống?

Có. Bên cạnh url_hash, tính simhash của title + first 200 char nội dung. So sánh simhash với 100 bài gần nhất cùng category, nếu Hamming distance < 5 -> coi là duplicate. Hoặc dùng embedding + cosine similarity > 0.92. Phổ biến với báo lá cải lấy lại nội dung nhau.

Có thể chạy farm trên Cloudflare Worker thay vì VPS không?

Có nhưng giới hạn CPU time và RAM của Worker (free 10ms, paid 50ms) làm việc fetch + parse + LLM khó fit. Phù hợp hơn là dùng VPS cho worker chính, Cloudflare Worker chỉ làm edge proxy hoặc webhook receiver. Hoặc dùng Cloudflare Queues + Workers paid plan cho workload nhỏ.