1. Định nghĩa vai trò (Roles) và nhiệm vụ (Tasks) trong Crew
Để xây dựng hệ thống đa Agent hợp tác, bạn cần định nghĩa rõ ràng từng thành viên trong nhóm (Crew). Mỗi Agent sẽ đóng một vai trò chuyên biệt (Role) và thực hiện một nhiệm vụ cụ thể (Task).
Bước này chúng ta sẽ tạo file cấu hình mô tả các Agent và Task bằng Python. File này sẽ là "bản thiết kế" cho hệ thống, nơi bạn khai báo chuyên môn, mục tiêu và đầu ra mong đợi của từng Agent.
Tạo file /app/crew_config.py với nội dung hoàn chỉnh sau:
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
# Khai báo công cụ tìm kiếm (Tool) để Agent có thể truy cập internet
search_tool = SerperDevTool()
# --- Định nghĩa Các Agent (Roles) ---
researcher = Agent(
role='Nhà nghiên cứu thị trường chuyên nghiệp',
goal='Tìm kiếm thông tin cập nhật và sâu sắc về chủ đề được yêu cầu',
backstory='Bạn là một chuyên gia phân tích dữ liệu với kinh nghiệm 10 năm, am hiểu thị trường toàn cầu và có khả năng tổng hợp thông tin từ nhiều nguồn.',
tools=[search_tool],
verbose=True
)
writer = Agent(
role='Biên tập viên nội dung cao cấp',
goal='Viết báo cáo chi tiết, dễ hiểu và có chiều sâu dựa trên dữ liệu nghiên cứu',
backstory='Bạn là một tác giả nổi tiếng với khả năng biến các dữ liệu khô khan thành những câu chuyện hấp dẫn, logic và thuyết phục.',
verbose=True
)
# --- Định nghĩa Các Nhiệm vụ (Tasks) ---
research_task = Task(
description='Nghiên cứu về chủ đề: "Xu hướng AI Agent trong năm 2024". Tìm kiếm ít nhất 3 nguồn tin cậy, phân tích xu hướng chính và các công nghệ nền tảng.',
agent=researcher,
expected_output='Danh sách các xu hướng chính kèm theo trích dẫn nguồn và phân tích ngắn gọn.'
)
writing_task = Task(
description='Dựa trên kết quả nghiên cứu từ Task 1, viết một bài báo cáo tổng hợp dài 500 từ. Bài viết cần có cấu trúc: Giới thiệu, Phân tích xu hướng, Kết luận.',
agent=writer,
expected_output='Một bài báo cáo hoàn chỉnh, được trình bày rõ ràng, không có lỗi chính tả.'
)
# --- Khởi tạo Crew (Nhóm) ---
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
verbose=2,
process=Process.sequential # Quy trình: Researcher làm xong -> Writer mới bắt đầu
)
# Chạy Crew
if __name__ == '__main__':
result = crew.kickoff()
print("\n\n################################")
print(f"BÁO CÁO CUỐI CÙNG:\n{result}")
Kết quả mong đợi: Khi chạy file này, hệ thống sẽ khởi tạo 2 Agent. Agent Researcher sẽ thực hiện tìm kiếm (nếu có API key), sau đó truyền kết quả cho Agent Writer để tổng hợp báo cáo.
2. Cấu hình quy trình phối hợp (Process) và Chạy thử nghiệm
Trong CrewAI, tham số process quyết định cách các Agent tương tác với nhau. Có hai chế độ chính: sequential (tuyến tính) và hierarchical (hierarchical manager tự phân công).
Chúng ta sẽ bắt đầu với chế độ sequential để đảm bảo tính ổn định. Ở chế độ này, Task 1 phải hoàn thành trước khi Task 2 được thực thi, mô phỏng quy trình làm việc thực tế trong một đội nhóm.
Trước khi chạy code, bạn cần cài đặt các thư viện phụ thuộc và cấu hình biến môi trường cho API Key (Serper.dev cho tính năng tìm kiếm).
Cài đặt các gói cần thiết vào môi trường Python hiện tại:
pip install crewai crewai-tools python-dotenv
Kết quả mong đợi: Các thư viện được cài đặt thành công, không có lỗi dependency.
Tạo file cấu hình biến môi trường /app/.env để lưu trữ API Key. Bạn cần đăng ký tại serper.dev để lấy key miễn phí cho mục đích test:
SERPER_API_KEY=YOUR_SERPER_API_KEY_HERE
OPENAI_API_KEY=YOUR_OPENAI_API_KEY_HERE
Kết quả mong đợi: File .env được tạo, chứa các key cần thiết để Agent truy cập LLM và công cụ tìm kiếm.
Chạy thử nghiệm kịch bản bằng cách thực thi file cấu hình đã tạo:
cd /app && python crew_config.py
Kết quả mong đợi: Console sẽ in ra log chi tiết từng bước. Bạn sẽ thấy dòng "Researcher is working on..." sau đó là "Writer is working on...", cuối cùng là "BÁO CÁO CUỐI CÙNG" với nội dung hoàn chỉnh.
3. Tích hợp công cụ (Tools) bên ngoài để tăng cường khả năng
Để Agent không chỉ dựa vào kiến thức huấn luyện sẵn có (cutoff date) mà có thể xử lý dữ liệu thời gian thực, chúng ta cần tích hợp các công cụ bên ngoài (Tools).
Trong ví dụ này, chúng ta sẽ mở rộng khả năng của Agent bằng cách thêm công cụ truy vấn Web (Serper) và công cụ đọc file (FileReadTool). Điều này giúp Agent có thể phân tích tài liệu PDF hoặc văn bản được cung cấp.
Cập nhật lại file /app/crew_config.py để bao gồm thêm công cụ đọc file và một Agent mới có khả năng phân tích tài liệu:
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileReadTool
from pydantic import BaseModel
# Khai báo công cụ
search_tool = SerperDevTool()
file_read_tool = FileReadTool(file_path='/app/sample_data.txt')
# --- Định nghĩa Các Agent ---
researcher = Agent(
role='Nhà nghiên cứu thị trường',
goal='Nghiên cứu thông tin từ internet và file dữ liệu nội bộ',
backstory='Chuyên gia phân tích dữ liệu, am hiểu thị trường và có khả năng đọc hiểu tài liệu kỹ thuật.',
tools=[search_tool, file_read_tool],
verbose=True
)
writer = Agent(
role='Biên tập viên nội dung',
goal='Tổng hợp dữ liệu từ internet và file nội bộ thành báo cáo',
backstory='Tác giả nổi tiếng, giỏi trong việc tổng hợp đa nguồn thông tin.',
verbose=True
)
# --- Định nghĩa Các Nhiệm vụ ---
# Task 1: Kết hợp tìm kiếm web và đọc file
research_task = Task(
description='1. Tìm kiếm thông tin mới nhất về "AI Agent trends 2024". \n2. Đọc nội dung file sample_data.txt để lấy thông tin bổ sung về dự án nội bộ. \n3. Tổng hợp cả hai nguồn dữ liệu.',
agent=researcher,
expected_output='Bảng so sánh giữa xu hướng thị trường và dữ liệu dự án nội bộ.'
)
writing_task = Task(
description='Viết báo cáo chiến lược dựa trên kết quả nghiên cứu ở Task 1.',
agent=writer,
expected_output='Báo cáo chiến lược dài 500 từ.'
)
# --- Khởi tạo Crew ---
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
verbose=2,
process=Process.sequential
)
if __name__ == '__main__':
result = crew.kickoff()
print(f"\nKẾT QUẢ TÍCH HỢP TOOLS:\n{result}")
Để test công cụ đọc file, tạo một file mẫu /app/sample_data.txt:
Dữ liệu dự án nội bộ: Dự án Alpha đang phát triển module xử lý ngôn ngữ tự nhiên với tốc độ 2x so với thị trường.
Kết quả mong đợi: Khi chạy lại script, Agent Researcher sẽ thực hiện hành động "Search" và "Read File". Kết quả trả về sẽ bao gồm cả thông tin từ Google và nội dung trong file text.
4. Chạy thử nghiệm kịch bản đa Agent giải quyết bài toán phức tạp
Bây giờ chúng ta sẽ nâng cấp kịch bản lên mức phức tạp hơn: Sử dụng quy trình hierarchical. Trong quy trình này, CrewAI tự động tạo ra một "Manager Agent" để phân công công việc, giám sát và tổng hợp kết quả từ các Agent con.
Quy trình này mô phỏng môi trường làm việc thực tế, nơi có một trưởng nhóm điều phối các thành viên chuyên môn khác nhau để giải quyết một vấn đề không rõ ràng ngay từ đầu.
Tạo file mới /app/complex_crew.py với cấu trúc 3 Agent và quy trình Hierarchical:
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, CodeInterpreterTool
# Công cụ
search_tool = SerperDevTool()
code_tool = CodeInterpreterTool()
# --- Định nghĩa 3 Agent chuyên biệt ---
tech_researcher = Agent(
role='Kỹ sư nghiên cứu công nghệ',
goal='Đánh giá khả năng kỹ thuật và công nghệ nền tảng',
backstory='Chuyên gia kỹ thuật sâu, am hiểu kiến trúc hệ thống và các framework AI.',
tools=[search_tool],
verbose=True
)
business_analyst = Agent(
role='Chuyên gia phân tích kinh doanh',
goal='Đánh giá tiềm năng thị trường và mô hình kinh doanh',
backstory='Chuyên gia chiến lược, am hiểu thị trường, ROI và phân tích đối thủ.',
tools=[search_tool],
verbose=True
)
risk_assessor = Agent(
role='Chuyên gia đánh giá rủi ro',
goal='Phát hiện các rủi ro pháp lý, kỹ thuật và thị trường',
backstory='Chuyên gia tuân thủ và quản trị rủi ro, có tư duy phản biện sắc bén.',
tools=[search_tool],
verbose=True
)
# --- Định nghĩa Nhiệm vụ chung ---
# Manager sẽ tự động phân công các task này cho đúng người dựa trên role
main_task = Task(
description='Phân tích toàn diện về việc triển khai hệ thống AI Agent trên Kubernetes cho một startup. \nYêu cầu: \n1. Đánh giá kỹ thuật (độ phức tạp, chi phí hạ tầng).\n2. Đánh giá thị trường (nhu cầu, đối thủ).\n3. Đánh giá rủi ro (an ninh, pháp lý).',
expected_output='Báo cáo chiến lược toàn diện bao gồm phân tích kỹ thuật, kinh doanh và rủi ro, kèm theo khuyến nghị triển khai.',
# Không gán agent cụ thể, để Manager phân công
)
# --- Khởi tạo Crew với Process Hierarchical ---
crew = Crew(
agents=[tech_researcher, business_analyst, risk_assessor],
tasks=[main_task],
verbose=2,
process=Process.hierarchical, # Sử dụng Manager để phân công
manager_llm="gpt-4o" # Sử dụng LLM mạnh hơn cho vai trò Manager
)
if __name__ == '__main__':
print("Bắt đầu quy trình phân tích đa chiều với Manager...")
result = crew.kickoff()
print("\n\nBÁO CÁO TỪ QUẢN LÝ:")
print(result)
Chạy kịch bản phức tạp này:
cd /app && python complex_crew.py
Kết quả mong đợi: Log sẽ hiển thị sự xuất hiện của "Manager Agent". Bạn sẽ thấy Manager phân tích yêu cầu, sau đó tạo ra các sub-tasks và phân công cho Tech Researcher, Business Analyst và Risk Assessor. Cuối cùng, Manager tổng hợp các báo cáo riêng lẻ thành một báo cáo duy nhất.
5. Kiểm tra và xác minh kết quả (Verification)
Sau khi chạy xong các kịch bản, bạn cần xác minh rằng hệ thống hoạt động đúng như thiết kế. Việc kiểm tra này đảm bảo các Agent đã thực sự trao đổi thông tin và công cụ đã được kích hoạt.
Kiểm tra log đầu ra của file /app/crew_config.py (kịch bản cơ bản):
grep -E "Researcher|Writer" /app/crew_config.py | head -n 20
Kiểm tra xem file log có chứa các hành động cụ thể của công cụ (Tool Execution):
grep "Tool" /app/crew_config.py
Xác minh kết quả cuối cùng bằng cách kiểm tra độ dài của báo cáo (đảm bảo Writer đã viết đủ nội dung):
python -c "from crew_config import crew; result = crew.kickoff(); print(f'Dộ dài báo cáo: {len(result)} ký tự')"
Kết quả mong đợi: Độ dài báo cáo phải lớn hơn 500 ký tự. Nếu nhỏ hơn, cần điều chỉnh lại prompt trong Task để yêu cầu nội dung dài hơn.
Kiểm tra kịch bản Hierarchical (phức tạp):
grep "Manager" /app/complex_crew.py
Xác minh xem Manager có thực sự phân công không bằng cách quan sát log:
python complex_crew.py 2>&1 | grep "Delegating"
Kết quả mong đợi: Bạn sẽ thấy các dòng log bắt đầu bằng "Delegating task to..." chứng tỏ Manager đang phân công công việc cho các Agent con. Nếu không thấy dòng này, quy trình đang chạy ở chế độ tuyến tính hoặc Manager không được kích hoạt đúng cách.
Điều hướng series:
Mục lục: Series: Xây dựng nền tảng AI Agent tự động hóa với LangGraph, CrewAI và Kubernetes
« Phần 2: Xây dựng Agent đơn giản với LangGraph và LLM
Phần 4: Đóng gói ứng dụng AI Agent thành Docker Image »