Tool trong CrewAI: mở rộng khả năng agent

Ở bài tổng quan về CrewAI, mình đã giới thiệu cách framework này tổ chức nhiều AI agent làm việc cùng nhau. Agent có role, goal, backstory. Task có mô tả và kết quả mong đợi. Nhưng agent chỉ thực sự mạnh khi được trang bị thêm tool.

Tool cho phép agent tương tác với thế giới bên ngoài: tìm kiếm Google, đọc file, truy vấn database, gọi API. Bài này mình sẽ đi qua hệ thống tool trong CrewAI, từ tool có sẵn đến cách tự viết tool riêng.

Tool trong CrewAI là gì

Tool là một hàm (function) mà agent có thể gọi trong quá trình thực thi task. Mỗi tool có tên, mô tả, và logic xử lý. Agent đọc mô tả của tool để quyết định khi nào cần dùng, truyền tham số gì.

Nôm na thì tool giống như “tay chân” của agent. Agent là bộ não suy nghĩ, còn tool giúp nó thực hiện hành động cụ thể. Một agent nghiên cứu thị trường cần tool tìm kiếm web. Một agent xử lý dữ liệu cần tool đọc file CSV.

CrewAI hỗ trợ tool từ hai nguồn chính:

• CrewAI Toolkit (gói crewai-tools): bộ tool có sẵn, cài thêm là dùng được
• Custom tool: tự viết bằng Python, tuỳ nhu cầu dự án

Mỗi tool đều có cơ chế xử lý lỗi (error handling) và cache tích hợp sẵn. Nếu tool gọi API bên ngoài, kết quả được cache lại để lần sau không cần gọi lại, tiết kiệm thời gian và chi phí.

Tool có sẵn trong CrewAI

Để dùng tool có sẵn, cài thêm gói mở rộng:

# Cài CrewAI kèm bộ tool mở rộng
pip install 'crewai[tools]'

Sau đó import tool cần dùng từ crewai_tools. Dưới đây là một số tool phổ biến nhất:

Tìm kiếm và duyệt web

• SerperDevTool: tìm kiếm Google qua Serper API. Cần đăng ký API key tại serper.dev
• WebsiteSearchTool: tìm kiếm nội dung trong một website cụ thể (dạng RAG)
• ScrapeWebsiteTool: lấy toàn bộ nội dung một trang web
• FirecrawlScrapeWebsiteTool: scrape web qua Firecrawl, hỗ trợ JavaScript rendering

Đọc file và thư mục

• FileReadTool: đọc nội dung file (text, JSON, YAML…)
• DirectoryReadTool: đọc cấu trúc thư mục và nội dung các file bên trong
• CSVSearchTool: tìm kiếm trong file CSV
• PDFSearchTool: tìm kiếm trong file PDF
• DOCXSearchTool: tìm kiếm trong file Word

Code và dữ liệu

• CodeInterpreterTool: chạy code Python trực tiếp
• CodeDocsSearchTool: tìm kiếm trong tài liệu code
• JSONSearchTool: tìm kiếm trong file JSON
• PGSearchTool: truy vấn database PostgreSQL
• GithubSearchTool: tìm kiếm trong repository GitHub

Multimedia

• DALL-E Tool: tạo hình ảnh bằng DALL-E API
• YoutubeVideoSearchTool: tìm kiếm nội dung trong video YouTube
• YoutubeChannelSearchTool: tìm kiếm trong channel YouTube

💡 Danh sách tool liên tục được cập nhật. Bạn kiểm tra đầy đủ tại GitHub crewAI-tools để xem tool mới nhất.

Tạo custom tool với @tool decorator

Cách nhanh nhất để tạo tool riêng là dùng decorator @tool. Viết một hàm Python bình thường, thêm decorator lên trên, CrewAI sẽ tự động đóng gói thành tool.

from crewai.tools import tool
# Tạo tool lấy thời tiết
@tool("Kiểm tra thời tiết")
def check_weather(city: str) -> str:
    """Lấy thông tin thời tiết hiện tại của một thành phố."""
    # Giả lập kết quả (thực tế sẽ gọi API)
    return f"Thời tiết tại {city}: 28°C, trời nắng"

Chỗ này cần lưu ý vài điểm:

• Chuỗi truyền vào @tool("Kiểm tra thời tiết") là tên tool. Agent sẽ thấy tên này khi chọn tool
• Docstring (phần trong ba dấu nháy) là mô tả tool. Đây là phần quan trọng nhất vì LLM dựa vào mô tả để quyết định có dùng tool hay không
• Tham số hàm (ở đây là city: str) sẽ trở thành input mà agent truyền vào
• Giá trị return là kết quả mà agent nhận được

Một ví dụ khác, tạo tool tính toán:

from crewai.tools import tool
# Tool nhân hai số
@tool("Phép nhân")
def multiply(a: int, b: int) -> str:
    """Nhân hai số với nhau và trả về kết quả."""
    result = a * b
    return f"Kết quả: {a} x {b} = {result}"

Custom cache cho tool

Tool tạo bằng @tool decorator cũng hỗ trợ custom cache. Bạn gán thêm thuộc tính cache_function để kiểm soát khi nào cache kết quả:

# Hàm quyết định có cache kết quả hay không
def should_cache(args, result):
    # Chỉ cache nếu kết quả là số chẵn
    return result % 2 == 0
multiply.cache_function = should_cache

Tạo tool nâng cao với BaseTool

Khi tool phức tạp hơn (cần validation input, nhiều config, kết nối database…), bạn nên kế thừa class BaseTool. Cách này cho phép kiểm soát chi tiết hơn.

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type
# Định nghĩa schema cho input
class StockPriceInput(BaseModel):
    """Schema cho tool tra cứu giá cổ phiếu."""
    symbol: str = Field(..., description="Mã cổ phiếu, ví dụ: AAPL, GOOGL")
# Tạo tool bằng cách kế thừa BaseTool
class StockPriceTool(BaseTool):
    name: str = "Tra cứu giá cổ phiếu"
    description: str = "Lấy giá cổ phiếu hiện tại theo mã ticker."
    args_schema: Type[BaseModel] = StockPriceInput
def _run(self, symbol: str) -> str:
        # Logic tra cứu giá (giả lập)
        prices = {"AAPL": 178.50, "GOOGL": 141.20, "MSFT": 415.30}
        price = prices.get(symbol.upper(), None)
        if price:
            return f"{symbol.upper()}: ${price}"
        return f"Không tìm thấy mã {symbol}"

Điểm khác biệt so với @tool decorator:

• args_schema: dùng Pydantic model để validate input. Nếu agent truyền sai kiểu dữ liệu, Pydantic sẽ báo lỗi trước khi tool chạy
• _run(): method chứa logic xử lý chính. Bạn override method này
• Có thể thêm thuộc tính config (API key, endpoint URL…) vào class

Tool bất đồng bộ (async)

CrewAI cũng hỗ trợ tool bất đồng bộ (async). Thay vì def _run, bạn dùng async def _run:

import asyncio
from crewai.tools import BaseTool
class AsyncAPITool(BaseTool):
    name: str = "Gọi API bất đồng bộ"
    description: str = "Gọi API không chặn luồng chính."
async def _run(self, endpoint: str) -> str:
        # Giả lập gọi API async
        await asyncio.sleep(1)
        return f"Kết quả từ {endpoint}"

ℹ️ CrewAI tự động phát hiện tool sync hay async. Bạn không cần thay đổi cách gán tool cho agent.

Gán tool cho agent

Sau khi có tool (dù có sẵn hay tự viết), bạn gán vào agent qua tham số tools. Mỗi agent nhận một danh sách tool:

from crewai import Agent
from crewai_tools import SerperDevTool, FileReadTool
# Khởi tạo tool
search_tool = SerperDevTool()
file_tool = FileReadTool()
# Agent nghiên cứu: dùng tool tìm kiếm
researcher = Agent(
    role="Chuyên viên nghiên cứu",
    goal="Tìm thông tin mới nhất về công nghệ AI",
    backstory="Bạn là chuyên gia phân tích với 10 năm kinh nghiệm.",
    tools=[search_tool],  # Gán tool tìm kiếm
    verbose=True
)
# Agent viết bài: dùng tool đọc file
writer = Agent(
    role="Biên tập viên",
    goal="Viết bài blog chất lượng cao",
    backstory="Bạn có khả năng viết lách xuất sắc.",
    tools=[file_tool],  # Gán tool đọc file
    verbose=True
)

Nguyên tắc gán tool

• Gán đúng tool cho đúng agent. Agent nghiên cứu cần search tool, không cần image tool
• Mỗi agent có thể nhận nhiều tool. Truyền dạng list: tools=[tool1, tool2, tool3]
• Nếu không gán tool nào, agent chỉ dựa vào kiến thức của LLM để trả lời
• Tool cũng có thể gán ở cấp task (qua tham số tools trong Task). Tool ở task sẽ chỉ khả dụng khi agent xử lý task đó

⚠️ Mô tả tool (description) cực kỳ quan trọng. LLM dựa vào mô tả để quyết định dùng tool nào. Mô tả mơ hồ sẽ khiến agent chọn sai tool hoặc không dùng tool khi cần.

Ví dụ: tool tìm kiếm + tool đọc file

Dưới đây là ví dụ hoàn chỉnh: hai agent cộng tác viết blog. Agent đầu tiên nghiên cứu trên web, agent thứ hai đọc bài mẫu từ thư mục rồi viết bài mới.

import os
from crewai import Agent, Task, Crew
from crewai_tools import (
    SerperDevTool,
    WebsiteSearchTool,
    DirectoryReadTool,
    FileReadTool
)
# Cấu hình API key
os.environ["SERPER_API_KEY"] = "your-serper-key"
os.environ["OPENAI_API_KEY"] = "your-openai-key"
# Khởi tạo tool
search_tool = SerperDevTool()           # Tìm kiếm Google
web_rag_tool = WebsiteSearchTool()      # Tìm trong website
docs_tool = DirectoryReadTool(          # Đọc thư mục bài mẫu
    directory='./blog-posts'
)
file_tool = FileReadTool()              # Đọc file cụ thể
# === AGENT 1: Nghiên cứu ===
researcher = Agent(
    role="Chuyên viên nghiên cứu thị trường",
    goal="Phân tích xu hướng mới nhất trong ngành AI",
    backstory="Bạn theo dõi ngành AI từ 2018, nắm rõ các công ty lớn và startup nổi bật.",
    tools=[search_tool, web_rag_tool],
    verbose=True
)
# === AGENT 2: Viết bài ===
writer = Agent(
    role="Content Writer",
    goal="Viết bài blog hấp dẫn về AI",
    backstory="Bạn có 5 năm kinh nghiệm viết blog công nghệ, biết cách kể chuyện cho người không chuyên.",
    tools=[docs_tool, file_tool],
    verbose=True
)
# === TASK 1: Nghiên cứu ===
research_task = Task(
    description="Tìm 3 xu hướng AI nổi bật nhất hiện tại. Tập trung vào ứng dụng thực tế, không lý thuyết.",
    expected_output="Báo cáo 3 xu hướng, mỗi xu hướng gồm: tên, mô tả ngắn, công ty/sản phẩm tiêu biểu.",
    agent=researcher
)
# === TASK 2: Viết bài ===
write_task = Task(
    description=(
        "Dựa trên báo cáo nghiên cứu, viết bài blog 4 đoạn. "
        "Tham khảo phong cách từ các bài trong thư mục ./blog-posts."
    ),
    expected_output="Bài blog markdown, 4 đoạn, dễ hiểu, có ví dụ cụ thể.",
    agent=writer,
    output_file="blog-posts/new_post.md"  # Lưu kết quả ra file
)
# === Ghép thành Crew ===
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    verbose=True
)
# Chạy
result = crew.kickoff()
print(result)

Luồng hoạt động

1. researcher nhận research_task, dùng SerperDevTool tìm kiếm Google, dùng WebsiteSearchTool đào sâu vào các trang kết quả
2. Kết quả nghiên cứu được truyền làm context cho task tiếp theo
3. writer nhận write_task, dùng DirectoryReadTool đọc các bài mẫu để tham khảo phong cách, rồi viết bài mới
4. Bài viết được lưu vào blog-posts/new_post.md

💡 Nếu bạn muốn chạy thử trên VPS, đảm bảo cài Python 3.10+ và có API key cho OpenAI + Serper.

Câu hỏi thường gặp

Bài tiếp theo trong serie, mình sẽ nói về Crew: cách ghép agent + task + tool thành một đội ngũ hoàn chỉnh và điều phối chúng chạy theo quy trình tuần tự hoặc phân cấp.