Posted inAI

LlamaIndex Workflows: Xây dựng AI Agent Event-driven ‘chuẩn’ production

Artificial Intelligence tutorial - IT technology blog
Artificial Intelligence tutorial - IT technology blog

Thay đổi tư duy: Từ Pipeline ‘cứng’ sang Event-driven ‘mềm’

Làm RAG cơ bản, anh em thường quen với mô hình Sequential Pipeline (dữ liệu đi theo đường thẳng). Quy trình thường là: Nhận query -> Search tài liệu -> Tổng hợp câu trả lời. Flow này chạy tốt cho các task đơn giản. Tuy nhiên, khi Agent cần tự check lỗi hoặc lặp lại bước X cho đến khi đạt yêu cầu, Pipeline truyền thống sẽ nhanh chóng biến thành một “mớ bòng bong” if-else.

Mình từng tốn cả tuần chỉ để quản lý logic điều hướng trong LangChain đời đầu. Code lúc đó cực kỳ khó bảo trì và dễ gãy khi logic phình to. LlamaIndex Workflows giải quyết triệt để vấn đề này bằng cách chuyển từ dạng “chuỗi” sang “sự kiện”. Thay vì ép Agent đi theo đường thẳng, chúng ta chỉ cần định nghĩa: “Khi sự kiện X xảy ra, hãy kích hoạt bước Y”.

So sánh các phương pháp xây dựng AI Agent

Để anh em dễ hình dung tại sao Workflows lại đáng giá, hãy nhìn vào bảng so sánh nhanh dưới đây:

1. Linear Chains (Chuỗi tuyến tính)

  • Thực tế: Giống như một dây chuyền sản xuất chỉ chạy tiến, không chạy lùi.
  • Hạn chế: Không thể quay lại bước trước hoặc xử lý các rẽ nhánh phức tạp mà không làm rối code.

2. State Machines (DAG – LangGraph)

  • Thực tế: Kiểm soát trạng thái (state) cực tốt nhưng thiết lập khá rườm rà.
  • Hạn chế: Việc định nghĩa node và edge khiến sơ đồ flow trông như mạng nhện khi hệ thống lớn dần.

3. LlamaIndex Workflows (Event-driven)

  • Thực tế: Các bước (steps) hoàn toàn độc lập, giao tiếp qua tín hiệu (events).
  • Điểm cộng: Code rất “Pythonic” nhờ async/await và decorator. Bạn có thể thêm bớt tính năng mà không sợ làm hỏng toàn bộ flow hiện có.

Khi nào nên dùng Workflows?

Qua triển khai thực tế, mình rút ra 3 kịch bản mà Workflows phát huy sức mạnh tốt nhất:

  • Tự sửa lỗi (Self-Correction): Agent tự chạy unit test cho code vừa viết, nếu lỗi thì tự sửa lại (vòng lặp).
  • Xử lý song song (Parallelism): Cùng lúc lấy dữ liệu từ 3-4 nguồn khác nhau để đối chiếu thông tin nhanh chóng.
  • Chờ phản hồi (Human-in-the-loop): Agent gửi email cho sếp duyệt nội dung, chờ sếp OK mới thực hiện bước tiếp theo.

Hướng dẫn triển khai chi tiết

Đầu tiên, hãy đảm bảo anh em đang dùng llama-index bản v0.10.20 trở lên để có hỗ trợ đầy đủ nhất.

pip install llama-index llama-index-core

Bước 1: Định nghĩa các Sự kiện (Events)

Event là phương tiện vận chuyển dữ liệu. Chúng ta dùng Pydantic để định nghĩa cấu trúc dữ liệu đi kèm một cách tường minh.

from llama_index.core.workflow import Event

class SearchEvent(Event):
    query: str

class RefineEvent(Event):
    initial_answer: str
    context: str

Bước 2: Xây dựng cấu trúc Workflow

Mỗi hàm được gắn tag @step sẽ hoạt động như một micro-service nhỏ trong Agent của bạn.

from llama_index.core.workflow import Workflow, StartEvent, StopEvent, step
from llama_index.llms.openai import OpenAI

class ResearchAgent(Workflow):
    llm = OpenAI(model="gpt-4o")

    @step
    async def search_step(self, ev: StartEvent) -> SearchEvent:
        user_query = ev.get("query")
        print(f"Đang quét dữ liệu cho: {user_query}")
        return SearchEvent(query=user_query)

    @step
    async def process_step(self, ev: SearchEvent) -> StopEvent:
        # Giả lập kết quả từ LLM
        result = f"Dữ liệu phân tích cho {ev.query}: Thị trường AI đang tăng trưởng 30% mỗi năm."
        return StopEvent(result=result)

Bước 3: Thực thi

Chạy workflow cực kỳ đơn giản. Bạn chỉ cần gọi hàm run() và chờ kết quả trả về.

async def main():
    agent = ResearchAgent(timeout=60, verbose=True)
    result = await agent.run(query="Thị trường AI 2025")
    print(f"Kết quả: {result}")

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

Kỹ thuật nâng cao: Cơ chế Tự phản biện (Reflection)

Một pattern mình thường áp dụng cho production là cho Agent tự kiểm tra output của chính nó. Nếu câu trả lời chưa đạt chuẩn, nó sẽ tự phát một RefineEvent để yêu cầu xử lý lại.

@step
async def critic_step(self, ev: ProcessEvent) -> RefineEvent | StopEvent:
    score = check_quality(ev.answer) 
    if score > 0.8:
        return StopEvent(result=ev.answer)
    return RefineEvent(feedback="Thông tin hơi sơ sài, cần thêm số liệu cụ thể.")

Áp dụng chiêu này vào hệ thống hỗ trợ viết code, mình đã giúp tỷ lệ code chạy được ngay từ lần đầu tăng từ 60% lên hơn 85%.

Kinh nghiệm thực chiến (Best Practices)

  1. Dùng Context (ctx): Ngoài việc truyền dữ liệu qua Event, hãy dùng ctx để lưu các biến dùng chung như UserID hoặc SessionID.
  2. Trực quan hóa flow: Hãy dùng workflow.draw("flow.html"). Tin mình đi, khi flow có trên 10 bước, bạn sẽ không muốn debug bằng mắt đâu.
  3. Chặn vòng lặp vô tận: Luôn set timeout. Tránh trường hợp hai Event gọi nhau liên tục làm cháy tài khoản API của bạn.
  4. Naming Convention: Đừng đặt tên Event1, Event2. Hãy đặt là ValidationFailedEvent để nhìn log là biết ngay Agent đang gặp vấn đề gì.

Kết luận

LlamaIndex Workflows không chỉ là một tính năng, nó là cách tư duy mới để xây dựng AI chuyên nghiệp. Việc tách biệt logic thành các bước độc lập giúp hệ thống của bạn linh hoạt và dễ bảo trì hơn nhiều. Nếu bạn đang cảm thấy bế tắc với những chuỗi code dài dằng dặc, hãy thử chuyển sang hướng sự kiện. Trải nghiệm bảo trì code chắc chắn sẽ nhẹ đầu hơn nhiều!

Anh em gặp khó khăn khi điều hướng Event hay cần code mẫu cho kịch bản phức tạp hơn? Cứ để lại comment, mình sẽ hỗ trợ nhé!

Share:
Tags: