Hướng dẫn Agent Swarm: Điều phối tác nhân AI với CrewAI

Agent Swarm là gì?

Agent swarm là một nhóm các tác nhân AI cùng hợp tác để hoàn thành một nhiệm vụ. Thay vì yêu cầu một mô hình tự nghiên cứu, viết, rà soát và hoàn thiện mọi thứ, một agent swarm chia nhỏ công việc cho nhiều tác nhân chuyên trách. Mỗi tác nhân có vai trò rõ ràng, và hệ thống sẽ điều phối để tạo ra kết quả cuối cùng tốt hơn.

Lợi thế nằm ở chuyên môn hóa. Một tác nhân tập trung nghiên cứu có thể chuyên tìm và kiểm chứng nguồn, tác nhân tập trung viết có thể chú trọng cấu trúc và độ rõ ràng, và một người rà soát có thể kiểm tra bản nháp so với nghiên cứu gốc mà không bị phân tán bởi chính phần viết vừa tạo. 

Một tác nhân điều phối sẽ kết nối các bước và quyết định bước tiếp theo. Trong bầy mà ta xây dựng, các vai trò này tương ứng trực tiếp với Researcher, Writer, Reviewer và Manager.

Dĩ nhiên điều này không miễn phí. Nhiều tác nhân hơn đồng nghĩa nhiều lượt gọi mô hình hơn, tốn thời gian hơn, và nhiều điểm có thể lỗi quy trình hơn; một đánh đổi mà ta sẽ quay lại vào cuối bài, sau khi bạn thấy chi phí thực tế để chạy.

CrewAI là gì?

CrewAI là một framework mã nguồn mở để xây dựng hệ thống đa tác nhân. Nó cung cấp cách rõ ràng để định nghĩa tác nhân, giao nhiệm vụ và chọn cách phối hợp, giúp bạn tập trung vào vai trò và quy trình làm việc thay vì phần “đường ống” kết nối.

Ba khái niệm cốt lõi đảm nhiệm phần lớn công việc. 

• Agents là những người “lao động” riêng lẻ, mỗi tác nhân được định nghĩa bởi vai trò, mục tiêu và hậu truyện định hình hành vi. 
• Tasks mô tả công việc cần làm và đầu ra mong đợi của từng nhiệm vụ. 
• Một crew kết nối tác nhân và nhiệm vụ rồi chạy chúng theo một quy trình đã chọn

Quy trình này có thể là tuần tự, nơi nhiệm vụ chạy theo thứ tự cố định, hoặc phân cấp, nơi một tác nhân quản lý ủy nhiệm công việc cho các tác nhân khác và thẩm định đầu ra. Ở đây ta sẽ dùng quy trình phân cấp, với Manager đảm nhận việc ủy nhiệm.

CrewAI cũng xử lý truy cập mô hình thông qua LiteLLM, một thư viện mã nguồn mở cung cấp giao diện thống nhất cho nhiều nhà cung cấp. Nhờ đó ta có thể định nghĩa một mô hình Gemini 3.5 Flash duy nhất và tái sử dụng cho cả bốn tác nhân, như ta sẽ thiết lập ở các bước bên dưới.

Xây dựng Agent Swarm với CrewAI và Gemini 3.5 Flash

Bắt đầu thiết lập nào.

1. Điều kiện tiên quyết

Trước khi bắt đầu, hãy đảm bảo bạn có:

• Đã cài đặt Python 3.11 trở lên
• Đã cài đặt Jupyter Notebook hoặc JupyterLab
• Hiểu biết cơ bản về Python và notebook
• Một tài khoản Google AI Studio
• Đã bật thanh toán, hoặc đã nạp tín dụng vào tài khoản Google của bạn để dùng Gemini
• Một tài khoản Olostep miễn phí

2. Cài đặt phụ thuộc

Khởi chạy Jupyter Notebook hoặc JupyterLab và mở một notebook với kernel Python mới nhất trong môi trường của bạn. Sau đó chạy cell sau để cài các gói cần thiết.

%pip install --quiet -U \
    crewai \
    crewai-tools \
    google-genai \
    litellm \
    olostep \
    nest-asyncio

Sau khi cài đặt xong, chạy cell tiếp theo để kiểm tra phiên bản các gói đã cài. Điều này giúp xác nhận mọi thứ đã được cài đúng.

import importlib.metadata as md

for pkg in [
    "crewai",
    "crewai-tools",
    "google-genai",
    "litellm",
    "olostep",
    "nest-asyncio",
]:
    print(f"{pkg:>16}  {md.version(pkg)}")

Bạn sẽ thấy đầu ra tương tự như sau:

         crewai  1.14.6
    crewai-tools  1.14.6
      google-genai  2.8.0
          litellm  1.87.1
          olostep  1.1.0
    nest-asyncio  1.6.0

3. Thiết lập biến môi trường

Bạn sẽ cần hai khóa API cho dự án này. 

Trước tiên, tạo khóa API Gemini từ Google AI Studio. Mở Google AI Studio, vào mục API key và tạo khóa mới cho dự án của bạn. Gemini có gói miễn phí, nhưng bị giới hạn mức sử dụng và tốc độ. Để kiểm thử và chạy Gemini ổn định hơn, hãy kết nối tài khoản thanh toán Google hoặc nạp tín dụng vào tài khoản Google để có thể tính phí khi vượt mức miễn phí.

Tiếp theo, tạo khóa API Olostep. Đăng ký Olostep, mở bảng điều khiển Olostep và tạo khóa API từ trang API keys. Chúng ta sẽ dùng khóa này để cấp cho tác nhân Researcher quyền truy cập web trực tiếp phục vụ tìm kiếm và trích xuất trang.

Khi đã có cả hai khóa, hãy lưu chúng dưới dạng biến môi trường. Gemini sẽ cung cấp sức mạnh cho các tác nhân CrewAI, trong khi Olostep cho phép tác nhân Researcher thu thập thông tin hiện tại từ web.

Sau khi tạo cả hai khóa, hãy lưu chúng thành biến môi trường trước khi chạy notebook.

PowerShell:

$env:GEMINI_API_KEY="your-gemini-key"
$env:OLOSTEP_API_KEY="your-olostep-key"

macOS/Linux:

export GEMINI_API_KEY="your-gemini-key"
export OLOSTEP_API_KEY="your-olostep-key"

Giờ hãy chạy cell sau để kiểm tra khóa API Gemini đã được nạp đúng chưa.

import os

api_key = os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY")

if not api_key:
    raise RuntimeError(
        "Set GEMINI_API_KEY or GOOGLE_API_KEY in your environment, "
        "then restart the notebook kernel."
    )
print(f"Gemini API key loaded")

Bạn sẽ thấy: 

Gemini API key loaded

4. Cấu hình Gemini 3.5 Flash qua LiteLLM

Trong dự án này, chúng ta sẽ dùng Gemini 3.5 Flash làm LLM dùng chung cho tất cả tác nhân CrewAI. Gemini 3.5 Flash được thiết kế cho khối lượng công việc nhanh, mang tính tác vụ và tập trung vào mã. Nó phù hợp vì các tác nhân của chúng ta cần suy luận nhiều bước, làm theo hướng dẫn, sử dụng công cụ và tạo đầu ra có cấu trúc.

Gemini 3.5 Flash cũng hỗ trợ cửa sổ ngữ cảnh lớn, đầu ra dài, tư duy và dùng công cụ, khiến nó là lựa chọn mạnh cho quy trình tác nhân. Vì một bầy tác nhân có thể gọi mô hình nhiều lần qua các tác nhân Researcher, Writer, Reviewer và Manager, dùng một model Flash giúp quy trình phản hồi nhanh mà vẫn giữ chất lượng suy luận và viết tốt.

Nếu bạn muốn so sánh mô hình với các LLM hàng đầu khác, tôi khuyên đọc các hướng dẫn về Gemini 3.5 Flash vs GPT-5.5 và Gemini 3.5 Flash vs Claude Opus 4.8.

Chúng ta sẽ truy cập Gemini qua LiteLLM. LiteLLM là thư viện mã nguồn mở cung cấp giao diện thống nhất để gọi nhiều nhà cung cấp LLM, gồm Gemini, OpenAI, Anthropic, Bedrock, Vertex AI và các bên khác. Trong thiết lập này, CrewAI dùng LiteLLM phía sau nên ta có thể truyền tên mô hình Gemini theo định dạng provider/model đơn giản.

Giờ ta sẽ tạo một LLM dùng chung của CrewAI và tái sử dụng cho tất cả tác nhân. Cách này giúp thiết lập đơn giản vì Researcher, Writer, Reviewer và Manager đều dùng cùng mô hình Gemini.

from crewai import LLM

GEMINI_MODEL = "gemini/gemini-3.5-flash"

gemini_llm = LLM(
    model=GEMINI_MODEL,
    api_key=os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY"),
    temperature=1.0,
)

print(f"LLM ready: {gemini_llm.model}")
print("Provider API: Google Gemini (via LiteLLM)")

Bạn sẽ thấy đầu ra tương tự như sau:

LLM ready: gemini-3.5-flash
Provider API: Google Gemini (via LiteLLM)

Điều này xác nhận CrewAI giờ có thể truy cập Gemini qua LiteLLM. Ở các bước tiếp theo, ta sẽ truyền đối tượng gemini_llm dùng chung này cho từng tác nhân.

5. Thêm truy cập web trực tiếp

Tác nhân Researcher cần truy cập thông tin hiện tại. Để làm điều này, ta sẽ tạo một công cụ tìm kiếm Olostep tùy chỉnh và cấp cho tác nhân Researcher sau.

Công cụ này có hai chế độ. Nó có thể chỉ trả về tóm tắt tìm kiếm (nhanh hơn), hoặc có thể trích xuất nội dung trang kết quả và trả về nội dung ở dạng markdown. Ta cũng thêm một hạn mức ngân sách tìm kiếm nhỏ để tác nhân không gọi công cụ web quá nhiều lần.

from pydantic import Field, PrivateAttr
from crewai.tools import BaseTool

try:
    from olostep import Olostep, Olostep_BaseError
except ImportError:
    raise ImportError(
        "The 'olostep' package is not installed. Run %pip install -U olostep."
    )

Trước tiên, nạp khóa API Olostep từ môi trường.

olostep_api_key = os.getenv("OLOSTEP_API_KEY")

if not olostep_api_key or olostep_api_key == "your-olostep-api-key-here":
    print(
        "The Olostep tool will fail at call time unless OLOSTEP_API_KEY "
        "is set in the environment."
    )
else:
    print("Olostep API key loaded.")

_olostep_client = Olostep(api_key=olostep_api_key) if olostep_api_key else None

Giờ hãy tạo công cụ CrewAI tùy chỉnh.

class OlostepSearchTool(BaseTool):
    """Search the web with a strict research-call budget."""

    name: str = "olostep_web_search"
    description: str = (
        "Search the live web. Set scrape=false for fast discovery and "
        "scrape=true for full-page evidence. You have at most 3 discovery "
        "searches without scraping and 6 external search calls total. After "
        "3 discovery searches, use scrape=true on the strongest query or "
        "domain. Each call returns at most 3 results. Inputs: query, scrape, "
        "limit, and optional include_domains."
    )

    default_limit: int = Field(default=3, ge=1, le=3)
    max_chars_per_page: int = Field(default=4000, ge=500, le=12000)

    _total_calls: int = PrivateAttr(default=0)
    _discovery_calls: int = PrivateAttr(default=0)
    _call_history: list[dict] = PrivateAttr(default_factory=list)

    def _run(
        self,
        query: str,
        scrape: bool = False,
        limit: int | None = None,
        include_domains: list[str] | None = None,
    ) -> str:
        if _olostep_client is None:
            return (
                "OLOSTEP_API_KEY is not set. Set it in the environment "
                "and restart the kernel."
            )

        if self._total_calls >= 6:
            return (
                "Research search budget exhausted: 6 external calls have "
                "already been used. Stop searching and complete the task "
                "with the collected sources."
            )

        if not scrape and self._discovery_calls >= 3:
            return (
                "Discovery-search limit reached. Do not run another "
                "scrape=False search. Use scrape=True on the strongest query "
                "or selected domains, or finish with the sources already found."
            )

        clean_query = query.strip()
        if not clean_query:
            return "Search query cannot be empty."

        result_limit = max(1, min(limit or self.default_limit, 3))
        kwargs = {"query": clean_query, "limit": result_limit}

        if scrape:
            kwargs["scrape_options"] = {
                "formats": ["markdown"],
                "timeout": 25,
            }

        if include_domains:
            kwargs["include_domains"] = include_domains

        self._total_calls += 1
        if not scrape:
            self._discovery_calls += 1

        call_record = {
            "number": self._total_calls,
            "query": clean_query,
            "scrape": scrape,
            "limit": result_limit,
            "domains": include_domains or [],
            "status": "started",
            "results": 0,
        }
        self._call_history.append(call_record)

        try:
            search = _olostep_client.searches.create(**kwargs)
        except Olostep_BaseError as error:
            call_record["status"] = "error"
            call_record["error"] = f"{type(error).__name__}: {error}"
            return (
                f"Olostep search error: {type(error).__name__}: {error}\n"
                f"Budget used: {self._total_calls}/6 total calls; "
                f"{self._discovery_calls}/3 discovery calls."
            )

        if not search.links:
            call_record["status"] = "no_results"
            return (
                f"No results found for query: {clean_query!r}\n"
                f"Budget used: {self._total_calls}/6 total calls; "
                f"{self._discovery_calls}/3 discovery calls."
            )

        call_record["status"] = "success"
        call_record["results"] = len(search.links)

        chunks = [
            f"## Web results for: {clean_query!r}",
            f"Page scraping: {'enabled' if scrape else 'disabled'}",
            (
                f"Budget used: {self._total_calls}/6 total calls; "
                f"{self._discovery_calls}/3 discovery calls."
            ),
        ]

        for index, link in enumerate(search.links, 1):
            url = link.get("url", "")
            title = link.get("title") or url or "Untitled result"
            description = link.get("description") or "No description provided."

            chunks.append(
                f"### {index}. {title}\n"
                f"URL: {url}\n"
                f"Summary: {description}"
            )

            markdown = link.get("markdown_content")
            if scrape and markdown:
                body = markdown.strip()
                if len(body) > self.max_chars_per_page:
                    body = body[: self.max_chars_per_page] + "\n...[truncated]"
                chunks.append(f"Page content:\n{body}")

        return "\n\n".join(chunks)

Cuối cùng, tạo một thể hiện của công cụ.

olostep_search_tool = OlostepSearchTool()
print(f"Tool ready: {olostep_search_tool.name}")

Bạn sẽ thấy đầu ra tương tự như sau:

Olostep API key loaded.
Tool ready: olostep_web_search

Điều này xác nhận công cụ tìm kiếm web Olostep đã sẵn sàng. Sau đó, ta sẽ gắn công cụ này cho tác nhân Researcher để thu thập thông tin hiện tại từ web.

6. Tạo các tác nhân

Giờ ta sẽ tạo các tác nhân cho bầy. Ta định nghĩa ba tác nhân làm việc và một tác nhân quản lý. Các tác nhân làm việc đảm nhiệm nhiệm vụ chính, trong khi tác nhân quản lý điều phối quy trình.

Tác nhân Researcher sẽ dùng công cụ Olostep để tìm thông tin hiện tại. Tác nhân Writer sẽ biến nghiên cứu thành bản nháp. Tác nhân Reviewer sẽ cải thiện bản nháp. Tác nhân Manager sẽ quyết định cách ủy nhiệm công việc trong crew phân cấp.

Tạo các tác nhân làm việc

Bắt đầu với ba tác nhân làm việc.

from crewai import Agent

researcher = Agent(
    role="Senior Research Analyst",
    goal=(
        "Find accurate, current, and well-sourced information about {topic}. "
        "Produce concrete, verifiable points for the writer."
    ),
    backstory=(
        "You are a meticulous research analyst. You decide whether each web "
        "search needs page scraping. Use scrape=False when titles, URLs, and "
        "summaries are enough. Use scrape=True when you need full-page evidence "
        "to verify a claim or understand a source in detail. Cite source URLs."
    ),
    llm=gemini_llm,
    tools=[olostep_search_tool],
    verbose=True,
    allow_delegation=False,
)

Tiếp theo, tạo tác nhân Writer.

writer = Agent(
    role="Blog Post Writer",
    goal=(
        "Turn the research into a complete, publication-ready educational "
        "article on {topic}. Include a strong title, an unlabeled italic "
        "summary, TL;DR, detailed body sections, conclusion, and FAQs. Add a "
        "table only when it makes a comparison or decision easier to understand."
    ),
    backstory=(
        "You are a seasoned technology educator and writer. You create "
        "practical, approachable articles similar in structure to high-quality "
        "DataCamp content, without copying its wording or branding. You use "
        "clear headings, short paragraphs, concrete examples, and sourced facts."
    ),
    llm=gemini_llm,
    verbose=True,
    allow_delegation=False,
)

Giờ tạo tác nhân Reviewer.

reviewer = Agent(
    role="Senior Editor",
    goal=(
        "Review the draft for factual accuracy, clarity, structure, and "
        "completeness. Ensure it follows the required article order and return "
        "a polished, publication-ready version."
    ),
    backstory=(
        "You are a detail-oriented senior editor for an educational technology "
        "publication. You verify claims, improve flow, remove repetition, and "
        "ensure the title is followed by an unlabeled italic summary and TL;DR. "
        "You keep a table only when it adds real value, remove any Key Takeaways "
        "section, and ensure Conclusion appears immediately before FAQs."
    ),
    llm=gemini_llm,
    verbose=True,
    allow_delegation=False,
)

Tạo tác nhân quản lý

Cuối cùng, tạo tác nhân Manager.

manager = Agent(
    role="Content Coordinator",
    goal=(
        "Coordinate research, writing, and review for {topic}. Delegate each "
        "task to the correct specialist and ensure the reviewed article is the "
        "final result."
    ),
    backstory=(
        "You are an experienced editorial lead. You keep the workflow simple: "
        "research first, writing second, and editorial review last."
    ),
    llm=gemini_llm,
    verbose=True,
    allow_delegation=True,
)

Tác nhân quản lý sẽ được cấu hình riêng khi ta xây crew phân cấp. Đó là lý do nó chưa được thêm vào danh sách tác nhân làm việc.

7. Định nghĩa tác vụ

Giờ ta sẽ định nghĩa các tác vụ cho bầy tác nhân. Những tác vụ này không được gán trực tiếp cho tác nhân làm việc. Thay vào đó, tác nhân Manager sẽ quyết định ai nên xử lý từng tác vụ.

Trong quy trình này, tác vụ đầu là nghiên cứu, thứ hai là viết, và thứ ba là rà soát. Mỗi tác vụ cũng bao gồm đầu ra mong đợi, để các tác nhân biết chính xác họ cần tạo gì.

from crewai import Task

research_task = Task(
    description=(
        "Delegate to the Senior Research Analyst. Research '{topic}' using at "
        "most 6 web-tool calls. Use no more than 3 calls with scrape=False, "
        "then use scrape=True only when detailed evidence is needed. Collect "
        "5-7 useful findings, source URLs, and common reader questions."
    ),
    expected_output=(
        "Clear markdown research notes with 5-7 sourced findings and 3-5 "
        "potential FAQ questions."
    ),
)

Tiếp theo, tạo tác vụ viết. Tác vụ này dùng tác vụ nghiên cứu làm ngữ cảnh, để Writer có thể xây bài từ phần nghiên cứu đã thu thập.

write_task = Task(
    description=(
        "Delegate to the Blog Post Writer. Write one complete educational "
        "markdown article on '{topic}', approximately 1,000-1,500 words. Start "
        "with one H1 title and an unlabeled italic summary. Then include TL;DR, "
        "an introduction, useful body sections, an optional table only when it "
        "improves understanding, a Conclusion, and FAQs. Use source links."
    ),
    expected_output=(
        "One continuous publication-ready markdown article with title, italic "
        "summary, TL;DR, body, Conclusion, and 3-5 FAQs."
    ),
    context=[research_task],
)

Giờ, tạo tác vụ rà soát. Tác vụ này dùng cả tác vụ nghiên cứu và viết làm ngữ cảnh, để Reviewer có thể đối chiếu bài với nghiên cứu gốc.

review_task = Task(
    description=(
        "Delegate to the Senior Editor. Review the article for accuracy, "
        "clarity, structure, and source support. Return only the corrected full "
        "article. Keep the same continuous article format and ensure FAQs are "
        "the final section."
    ),
    expected_output=(
        "Only the final reviewed markdown article, beginning with its H1 title "
        "and ending with the final FAQ answer."
    ),
    context=[research_task, write_task],
)

Cuối cùng, thêm cả ba tác vụ vào một danh sách.

tasks = [research_task, write_task, review_task]

print(f"Defined {len(tasks)} manager-delegated tasks.")

Bạn sẽ thấy:

Defined 3 manager-delegated tasks.

8. Xây dựng quy trình làm việc

Giờ ta sẽ kết nối các tác nhân và tác vụ thành một quy trình CrewAI thống nhất. Ta sẽ dùng quy trình phân cấp, trong đó tác nhân Manager điều phối các tác nhân làm việc.

Trong thiết lập này, Researcher, Writer và Reviewer được thêm làm tác nhân làm việc. Manager được truyền riêng qua manager_agent, vì nó chịu trách nhiệm ủy nhiệm và điều phối.

from crewai import Crew, Process

crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=tasks,
    manager_agent=manager,
    process=Process.hierarchical,
    verbose=True,
    memory=False,
)

Giờ hãy in chi tiết crew để xác nhận quy trình đã lắp ráp đúng.

print("Hierarchical crew assembled.")
print(f"  Workers : {[a.role for a in crew.agents]}")
print(f"  Manager : {crew.manager_agent.role}")
print(f"  Tasks   : {len(crew.tasks)}")
print(f"  Process : {crew.process}")

Bạn sẽ thấy đầu ra tương tự như sau:

Hierarchical crew assembled.
  Workers : ['Senior Research Analyst', 'Blog Post Writer', 'Senior Editor']
  Manager : Content Coordinator
  Tasks   : 3
  Process : Process.hierarchical

Điều này xác nhận bầy tác nhân phân cấp đã sẵn sàng. Ở bước tiếp theo, ta sẽ chạy crew với một chủ đề và tạo phản hồi cuối.

9. Chạy agent swarm

Giờ ta có thể chạy bầy tác nhân. Ta sẽ cung cấp một chủ đề và khởi động quy trình bằng kickoff_async().

Jupyter vốn đã chạy một vòng lặp sự kiện nền. Vì vậy, trước hết ta áp dụng nest_asyncio, để quy trình bất đồng bộ của CrewAI có thể chạy trong notebook.

try:
    import nest_asyncio

    nest_asyncio.apply()

    print("nest_asyncio applied: CrewAI can now run inside this kernel.")

except ModuleNotFoundError:
    print("nest_asyncio not installed. Run: pip install nest-asyncio")

Bạn sẽ thấy:

nest_asyncio applied: CrewAI can now run inside this kernel.

Tiếp theo, định nghĩa một chủ đề và tạo hàm trợ giúp để chạy crew.

from datetime import datetime

topic = "Learning Adaptive Thresholds for LLM Tool Use: When to Rely on Memory and When to Call a Tool"

async def run_crew(topic: str):
    """Run the CrewAI crew asynchronously."""
    print(f"Starting crew at {datetime.now().isoformat(timespec='seconds')}")
    print(f"Topic: {topic!r}\n")

    result = await crew.kickoff_async(inputs={"topic": topic})

    print(f"\nFinished at {datetime.now().isoformat(timespec='seconds')}")
    print(f"Result type: {type(result).__name__}")
    return result

result = await run_crew(topic)

Khi quy trình bắt đầu, CrewAI trước hết giao tác vụ cho tác nhân Manager. Manager sau đó ủy nhiệm công việc nghiên cứu cho Senior Research Analyst. Researcher sử dụng công cụ olostep_web_search để tìm kiếm thông tin trên web về ngưỡng thích ứng, việc dùng công cụ của LLM, bộ nhớ và quyết định gọi công cụ.

Trong nhật ký, Researcher dùng cả các lần gọi chỉ tìm kiếm và các lần trích xuất. Ngân sách tìm kiếm được dùng hết, đạt 6/6 lần gọi web tổng và 3/3 lần tìm kiếm khám phá. Một yêu cầu chỉ-tìm-kiếm bổ sung bị chặn vì đã chạm giới hạn tìm kiếm khám phá. Điều này cho thấy các quy tắc ngân sách trong công cụ Olostep tùy chỉnh của chúng ta hoạt động đúng.

Sau giai đoạn nghiên cứu, Manager tiếp tục quy trình bằng cách ủy nhiệm bước viết và rà soát. Các tác nhân tạo ghi chú nghiên cứu, soạn bản nháp bài viết, rồi hoàn thiện phản hồi cuối. Lượt chạy kết thúc thành công và trả về một đối tượng CrewOutput.

Bạn sẽ thấy đầu ra tương tự như sau:

Starting crew at 2026-06-05T13:54:07
Topic: 'Learning Adaptive Thresholds for LLM Tool Use: When to Rely on Memory and When to Call a Tool'

Crew Execution Started
Task Started
Agent: Content Coordinator
Agent: Senior Research Analyst
Tool: olostep_web_search
Budget used: 6/6 total calls; 3/3 discovery calls

Finished at 2026-06-05T13:59:29
Result type: CrewOutput

Điều này có nghĩa toàn bộ bầy đã chạy thành công. Manager điều phối quy trình, Researcher thu thập thông tin hiện tại, Writer tạo bản nháp, và Reviewer cải thiện đầu ra cuối.

10. Xem kết quả

Sau khi bầy chạy xong, ta có thể hiển thị bài viết cuối cùng do crew tạo ra. Giá trị result.raw chứa phản hồi cuối từ quy trình.

from IPython.display import Markdown, display

raw = result.raw

if "Blog Post" in raw:
    raw = raw.split("Blog Post", 1)[1].strip()

display(Markdown(raw))

Điều này sẽ hiển thị đầu ra cuối cùng dưới dạng markdown được định dạng trong notebook. 

Trong hầu hết trường hợp, blog được tạo sẽ có cấu trúc tốt, với nhiều tiêu đề, các phần rõ ràng, sơ đồ hoặc giải thích trực quan khi phù hợp, và phần kết luận thuyết phục. 

Sự kết hợp giữa nghiên cứu, viết và biên tập giúp tạo nội dung sẵn sàng xuất bản. Sau khi rà soát đầu ra, cá nhân tôi sẽ “bật đèn xanh” để xuất bản.

Tiếp theo, ta có thể in dấu vết thực thi. Điều này giúp hiểu bầy đã hoạt động thế nào, tác nhân nào tham gia, đã dùng bao nhiêu lần gọi web, và từng tác vụ được xử lý ra sao.

from collections import Counter

def role_name(value):
    """Return a readable role/name for CrewAI agent values."""
    if value is None:
        return "Not reported"
    if isinstance(value, str):
        return value
    return getattr(value, "role", None) or getattr(value, "name", None) or str(value)

print("=" * 70)
print("SWARM EXECUTION TRACE")
print("=" * 70)

# Hierarchical crew structure
print("\nCOORDINATION")
print(f"Process : {crew.process}")
print(f"Manager : {role_name(crew.manager_agent)}")
print(f"Workers : {', '.join(role_name(agent) for agent in crew.agents)}")
print(
    "Flow    : Manager receives each task, delegates specialist work, "
    "validates the response, and passes approved context to the next task."
)

Now print the web-tool usage recorded by the custom Olostep tool.
history = list(getattr(olostep_search_tool, "_call_history", []))
discovery_calls = sum(not item["scrape"] for item in history)
scrape_calls = sum(item["scrape"] for item in history)
successful_calls = sum(item["status"] == "success" for item in history)

print("\nWEB TOOL USAGE")
print(f"Total external calls : {len(history)}/6")
print(f"Without scraping     : {discovery_calls}/3")
print(f"With scraping        : {scrape_calls}")
print(f"Successful calls     : {successful_calls}")

if history:
    for item in history:
        mode = "scrape" if item["scrape"] else "search only"
        domains = ", ".join(item["domains"]) if item["domains"] else "all domains"
        print(
            f"  {item['number']}. {mode}; status={item['status']}; "
            f"results={item['results']}; scope={domains}"
        )
        print(f"     Query: {item['query']}")
else:
    print("  No Olostep calls were recorded.")

Finally, print the task-by-task trace.
print("\nTASK-BY-TASK TRACE")
agent_activity = Counter()

for index, task_out in enumerate(result.tasks_output, start=1):
    task = crew.tasks[index - 1] if index <= len(crew.tasks) else None
    reported_agent = role_name(getattr(task_out, "agent", None))
    processed_by = sorted(getattr(task, "processed_by_agents", set()) or [])
    participants = processed_by or [reported_agent]

    for participant in participants:
        agent_activity[role_name(participant)] += 1

    delegations = getattr(task, "delegations", 0) if task else 0
    used_tools = getattr(task, "used_tools", 0) if task else 0
    tool_errors = getattr(task, "tools_errors", 0) if task else 0
    start_time = getattr(task, "start_time", None) if task else None
    end_time = getattr(task, "end_time", None) if task else None
    duration = end_time - start_time if start_time and end_time else None

    print(f"\n--- Task {index} ---")
    print(f"Reported owner : {reported_agent}")
    print(f"Processed by   : {', '.join(map(role_name, participants))}")
    print(f"Delegations    : {delegations}")
    print(f"Tool uses      : {used_tools}")
    print(f"Tool errors    : {tool_errors}")
    if duration is not None:
        print(f"Duration       : {duration}")

    preview = getattr(task_out, "raw", "").strip().replace("\n", " ")
    print(f"Output preview : {preview[:300]}{'...' if len(preview) > 300 else ''}")

print("\nAGENT ACTIVITY")
for agent, count in agent_activity.most_common():
    print(f"{agent}: participated in {count} task(s)")

if not agent_activity:
    print("CrewAI did not expose per-agent processing metadata for this run.")

Bạn sẽ thấy đầu ra tương tự như sau:

Dấu vết này cho thấy bầy phân cấp hoạt động như kỳ vọng. Manager điều phối cả ba tác vụ, Researcher xử lý bước nghiên cứu, Writer tạo bản nháp, và Reviewer cải thiện phản hồi cuối.

Tuy vậy, điều này cũng cho thấy một hạn chế quan trọng: quy trình đa tác nhân có thể tốn kém. 

Trong lượt chạy này, quy trình dùng 6 lần gọi web bên ngoài và nhiều lần gọi LLM qua các tác nhân Manager, Researcher, Writer và Reviewer. Một lượt có thể tốn khoảng $0,28, và tạo hướng dẫn này tốn khoảng $2,78 trong quá trình thử nghiệm. Đó là mức cao cho một dự án hướng dẫn đơn giản.

Suy ngẫm cuối

Agent swarm và quy trình đa tác nhân trông rất ổn trên giấy. Ý tưởng có Researcher, Writer, Reviewer và Manager cùng làm việc nghe có vẻ mạnh mẽ, và trong hướng dẫn này, đầu ra cuối thực sự đủ tốt để xuất bản với vài chỉnh sửa nhỏ.

Nhưng trên thực tế, thiết lập này có thể tốn kém hơn, mất nhiều thời gian hơn và tạo thêm nhiều điểm có thể lỗi. Với tôi, một lượt tốn khoảng $0,28, và tạo hướng dẫn này tốn khoảng $2,78. Đó là con số lớn cho một dự án hướng dẫn đơn giản.

Với ứng dụng thực tế, tôi sẽ không phải lúc nào cũng dùng một bầy đa tác nhân đầy đủ. Tôi sẽ ưu tiên thiết lập mang tính lập trình hơn, nơi các tác vụ đơn giản được xử lý bằng mã hoặc quy tắc, và chỉ giao cho tác nhân những tác vụ phức tạp. Ta cũng có thể giảm chi phí bằng cách hạn chế gọi công cụ, dùng ít tác nhân hơn, rút ngắn prompt, và dùng quy trình tuyến tính thay vì phân cấp.

Tất nhiên, khi tối ưu quá nhiều, nó có thể không còn là một bầy tác nhân đúng nghĩa. Nó trở nên giống một quy trình AI dựa trên quy tắc hơn. Vì thế mục tiêu là cân bằng: dùng bầy tác nhân khi chúng thật sự mang lại giá trị, và giữ mọi thứ đơn giản khi không cần thiết.