Giới thiệu

Hãy tưởng tượng nếu bạn có thể cung cấp dữ liệu từ bất kỳ trang web nào cho AI của mình, giúp bạn có được thông tin chính xác mà bạn cần. Điều này hoàn toàn khả thi với sự kết hợp của Scrapy và Zyte API.

Gặp gỡ Fab 👨‍💻
Fab là một lập trình viên có nhiều năm kinh nghiệm. Gần đây, anh bắt đầu tìm hiểu về tài chính và những cổ phiếu đầy hứa hẹn. Tuy nhiên, một vấn đề lớn đã nảy sinh: việc theo dõi tin tức hàng ngày, các thông cáo báo chí, và đọc qua 10 bài viết mỗi sáng là một công việc rất tốn thời gian và thủ công.

Vì vậy, Fab đã quyết định xây dựng một AI Agent giúp anh thực hiện điều đó - tự động thu thập, đọc và tóm tắt mọi thứ trong thời gian thực.

Đó chính là một pipeline RAG tùy chỉnh, được hỗ trợ bởi dữ liệu web sống động, không còn bị giới hạn bởi các tài liệu PDF tĩnh hay thông tin lỗi thời.

Tại sao lại cần làm điều này?

Bởi vì ngay cả những AI Agent thông minh nhất cũng chỉ tốt như dữ liệu mà nó có thể truy cập:

• Các LLM có giới hạn về kiến thức
• Dữ liệu theo thời gian thực, cụ thể cho từng lĩnh vực (như tài chính) là rất quan trọng cho việc ra quyết định

Bằng cách khai thác dữ liệu web sống, agent của Fab có thể theo kịp thông tin thế giới một cách chính xác và kịp thời - luôn luôn liên quan, luôn luôn sẵn sàng.

Nhưng hãy lưu ý ✋, việc tóm tắt hoặc trả lời không giống như việc thực hiện các hành động thực sự. Đó là điểm khác biệt giữa AI Agents và Agentic AI.

• AI Agents là các hệ thống phần mềm được thiết kế để tự động hóa các nhiệm vụ cụ thể, rõ ràng, như chatbot, công cụ phân loại email, hoặc trợ lý giọng nói, thường dựa trên các công cụ hoặc câu lệnh được xác định trước.
• Agentic AI, ngược lại, có một phạm vi tự chủ rộng hơn.

Những gì chúng ta sẽ khám phá ở đây về mặt kỹ thuật là một AI Agent, nhưng vì cả hai đều chia sẻ cùng một nền tảng, nó có thể phát triển thành Agentic AI.

Bộ công cụ của Fab 🛠️

• Scrapy → để trích xuất dữ liệu có cấu trúc
• Zyte API → để xử lý các trang web động và phức tạp
• DuckDuckGo + yfinance → để có thêm thông tin tìm kiếm và tài chính
• Agno → để điều phối quy trình làm việc đa agent
• GroqCloud → để suy luận LLM nhanh chóng

Kiến trúc 🏗️

Tại sao lại chọn Scrapy + Zyte API?

Bạn có thể thử thực hiện điều này chỉ với Scrapy và các proxy xoay vòng. Nhưng bất kỳ ai đã làm việc với việc thu thập dữ liệu quy mô lớn đều biết đến những khó khăn: bị chặn, captcha, yêu cầu thất bại.

Đó là lý do tại sao Zyte API tỏa sáng. Nó giúp giảm thiểu công việc nặng nhọc, vì vậy bạn không cần phải giám sát các scraper của mình, bạn chỉ cần nhận dữ liệu sạch và có cấu trúc.

Hãy nghĩ về nó như việc có một nhóm backend chuyên dụng đảm bảo rằng các spider của bạn không bao giờ bị kẹt.

Thu thập dữ liệu đúng cách! 📥

Thay vì thu thập mọi thứ, agent của Fab sẽ chỉ thu thập các URL trước... và sau đó lấy dữ liệu quan trọng dựa trên điểm xu hướng.

Để xử lý điều này một cách hiệu quả, Fab đã thiết kế một dự án Scrapy với một spider cơ sở và bốn spider chuyên biệt để thu thập:

1. Tin tức
2. Thông cáo báo chí
3. Biên bản
4. Bình luận

Spider cơ sở chịu trách nhiệm về việc thu thập dữ liệu cụ thể của từng trang bằng cách:

• Lấy URL và metadata
• Làm sạch và chuẩn hóa ngày tháng
• Tạo ID duy nhất từ các URL

import scrapy
from urllib.parse import urlparse, urlunparse
from datetime import datetime, timedelta
import hashlib

class BaseFinanceSpider(scrapy.Spider):
    name = "base_finance_spider"
    allowed_domains = ["finance-example.com"]

    def clean_url(self, url):
        """Chuẩn hóa các URL"""
        parsed = urlparse(url)
        return urlunparse(parsed._replace(query="", fragment=""))

    def create_id(self, url):
        """Tạo ID duy nhất từ URL đã được làm sạch"""
        return hashlib.sha256(self.clean_url(url).encode()).hexdigest()

    def convert_date(self, raw_date, now=None):
        """Chuyển đổi các ngày tương đối như 'Hôm qua' hay 'Hôm nay' sang ISO"""
        now = now or datetime.now()
        if "Hôm qua" in raw_date:
            return (now - timedelta(days=1)).isoformat()
        if "Hôm nay" in raw_date:
            return now.isoformat()
        return raw_date

Mỗi spider chuyên biệt sẽ kế thừa từ spider cơ sở và tập trung vào logic cụ thể của trang: điều hướng các trang và trích xuất thông tin quan trọng cho loại dữ liệu của nó.

Tại thời điểm này, ba spider chuyên biệt thu thập chỉ các URL và metadata, tạo ra một danh sách JSON cho mỗi loại dữ liệu. Bình luận là một ngoại lệ, chúng ta thu thập ngay lập tức. Hãy nghĩ về điều này như việc chuẩn bị một “danh sách việc cần làm” cho agent của Fab để xử lý sau này, giữ cho mọi thứ được tổ chức và hiệu quả.

Khi các mục được xuất ra, các Pipelines của Scrapy tự động xử lý các nhiệm vụ như chuẩn hóa URL và gán ID, loại bỏ trùng lặp, ẩn danh hóa, liên kết bình luận và lưu các mục vào JSON.

class UrlNormalizationPipeline:
    def process_item(self, item, spider):
        item['url'] = spider.clean_url(item.get('url'))
        item['id'] = spider.create_id(item['url'])
        return item

class DeduplicationPipeline:
    def __init__(self):
        self.seen_ids = set()
    def process_item(self, item, spider):
        if item['id'] in self.seen_ids:
            raise DropItem(f"Trùng lặp: {item['id']}")
        self.seen_ids.add(item['id'])
        return item

class AnonymizationPipeline:
    def process_item(self, item, spider):
        return item

class JsonFileExportPipeline:
    def process_item(self, item, spider):
        return item

Phân tích xu hướng 📈

Khi chúng ta đã thu thập xong các bài viết (tin tức, thông cáo báo chí, biên bản) và bình luận, bước tiếp theo là xác định các chủ đề nào thực sự đang nổi bật. Việc thu thập nội dung thô chỉ là một phần công việc, điều làm cho nó có giá trị là biết được sự chú ý đang hướng về đâu.

Chúng tôi đã xây dựng một Bộ tính toán Xu hướng. Nhiệm vụ của nó là kết nối tất cả các bài viết và bình luận mà chúng tôi đã thu thập, và sau đó gán cho mỗi bài viết một điểm xu hướng. Điểm số được dựa trên một vài tín hiệu đơn giản nhưng mạnh mẽ:

• Hoạt động bình luận – Các bài viết có nhiều bình luận hơn sẽ nhận được điểm cao hơn (có giới hạn, để một bài viết viral không làm lệch mọi thứ).
• Sự nhắc đến trong bình luận – Nếu mọi người đang thảo luận về một bài viết trong các bình luận của bài viết khác, đó là dấu hiệu của ảnh hưởng.
• Tính mới – Các bài viết gần đây sẽ nhận được điểm thưởng vì xu hướng phai nhạt nhanh chóng theo thời gian.
• Xác thực đa nguồn – Nếu cùng một chủ đề xuất hiện ở nhiều nguồn (như tin tức và thông cáo báo chí), nó có khả năng quan trọng.
• Chất lượng tương tác – Các bình luận dài hơn, suy nghĩ hơn sẽ có trọng số nhiều hơn so với các bình luận ngắn.

# Ví dụ về logic tính điểm
comment_score = 3 * min(article.get('comment_count', 0), 10)  
mention_score = 2 * min(article.get('comment_mentions', 0), 5)  
date_score = calculate_date_bonus(article.get('date'))  
source_score = 2 if len(article.get('sources', [])) > 1 else 0  
engagement_score = quality_from_comments(article.get('comments', []))  

trend_score = comment_score + mention_score + date_score + source_score + engagement_score

Xử lý các bài viết 📑

Bây giờ, đã đến lúc thoát khỏi các tín hiệu và thực sự lấy nội dung các bài viết. Đây là nơi agent của Fab sẽ lấy văn bản đầy đủ để có thể đọc, tóm tắt và thực hiện hành động - việc thu thập và xử lý thực sự bắt đầu ở đây.

Bước 1: Trích xuất thông minh với Zyte API

Thay vì thu thập dữ liệu một cách mù quáng, chúng ta sẽ chạy mỗi URL bài viết qua Zyte API. Nó sẽ thử nhiều chiến lược khác nhau:

• Kết xuất trình duyệt cho các trang phong phú.
• Phản hồi HTTP thay thế nếu lần đầu không thành công.
• Và nếu không có gì thành công → một đối tượng thay thế hợp lý ghi nhận rằng bài viết không thể được trích xuất (như tường phí, tường đăng nhập, v.v.).

Bước 2: Xử lý theo lô ⚡

Không tốt khi gửi 50 yêu cầu cùng một lúc, vì vậy agent của Fab sẽ thu thập các bài viết theo từng lô nhỏ. Điều này giữ cho mọi thứ ổn định, tránh giới hạn tỷ lệ và cho phép chúng tôi tiếp tục giữa chừng nếu có gì thất bại.

scraped_articles = process_articles_in_batches(urls, batch_size=3)

Bước 3: Bình luận + Ẩn danh

Khi các bài viết thô đã có, chúng tôi sẽ gán các bình luận liên quan của chúng (đã được thu thập trước đó) và ẩn danh hóa các tên người dùng. Bằng cách này, Fab có thể thấy các tín hiệu thảo luận mà không lo lắng về việc lộ thông tin cá nhân.

Bước 4: Tóm tắt với LLMs

Cuối cùng, mỗi bài viết sẽ được tóm tắt bằng cách sử dụng Groq + Llama 3.3, với các bình luận được đưa vào bối cảnh. Lời nhắc đảm bảo rằng Fab nhận được:

• Một thẻ loại nội dung rõ ràng ([Bài viết đầy đủ với bình luận], [Bài viết một phần], v.v.).
• Các điểm chính của bài viết.
• Những điểm nổi bật từ các bình luận của người dùng (sự đồng ý, tranh luận, cảm xúc).
• Một ghi chú nếu bài viết có vẻ không đầy đủ hoặc bị cắt ngắn.

summary = processor.summarize_article(article)

Biến các tóm tắt thô thành thứ hữu ích

Vậy là chúng tôi đã có các bài viết đã được làm sạch, tóm tắt ngắn gọn trong JSON. Điều này rất tốt, nhưng Fab không chỉ muốn một thư mục đầy những tóm tắt, mà là một agent có thể lý luận dựa trên chúng, kết hợp chúng với dữ liệu thị trường thời gian thực và cung cấp cho anh câu trả lời theo yêu cầu.

Đó là lý do Agno sẽ được sử dụng cho điều này. Agno là một framework để xây dựng các agent được hỗ trợ bởi LLM, nơi mọi thứ xoay quanh các công cụ. Chúng tôi sử dụng một số công cụ có sẵn, như yfinance cho dữ liệu thị trường hoặc DuckDuckGo cho các tìm kiếm nhanh, và chúng tôi sẽ tạo công cụ riêng của mình bằng cách sử dụng các bài viết đã được thu thập và tóm tắt.

Kết luận

Fab không chỉ có một công cụ thu thập dữ liệu hay một công cụ tóm tắt mà còn có một đồng hành tài chính luôn cập nhật, có ý thức về bối cảnh và dựa trên dữ liệu web thực tế.

Với quy trình này, những gì bắt đầu như một công việc thủ công, tốn thời gian đã được chuyển thành một hệ thống thông minh, liền mạch, chứng minh sức mạnh của AI Agents khi được kết hợp với dữ liệu web sống.