Bạn đã bao giờ tự hỏi làm thế nào một AI agent có thể "suy nghĩ" và "hành động" theo từng bước như một con người thật sự? Hôm nay, mình sẽ chia sẻ toàn bộ hành trình từ con số 0 đến khi triển khai thành công event-driven agent architecture đầu tiên của mình — tất cả chỉ với LlamaIndex và HolyShehe AI.

"Trước đây mình từng nghĩ AI agent là thứ siêu phức tạp, chỉ dành cho senior engineer. Nhưng sau khi hiểu event-driven workflow, mọi thứ trở nên rõ ràng đến bất ngờ."

Event-Driven Agent Architecture Là Gì?

Hãy tưởng tượng bạn đang điều khiển một đội robot trong nhà máy. Thay vì một robot làm mọi thứ, mỗi robot sẽ:

Đó chính xác là cách event-driven architecture hoạt động trong LlamaIndex Agent Workflows. Thay vì code tuần tự từng bước, bạn xây dựng các "trạm làm việc" (nodes) và kết nối chúng bằng các "tín hiệu" (events).

Tại Sao Nên Dùng HolyShehe AI?

Trước khi bắt đầu code, mình muốn giới thiệu nền tảng mình đang sử dụng — HolySheep AI. Với giá chỉ từ $0.42/1M tokens (DeepSeek V3.2), bạn tiết kiệm đến 85%+ so với GPT-4.1 ($8/1M tokens). Độ trễ dưới 50ms, hỗ trợ WeChat/Alipay, và nhận tín dụng miễn phí khi đăng ký.

Bước 1: Cài Đặt Môi Trường

# Tạo môi trường ảo (virtual environment)
python -m venv agent_env

Kích hoạt môi trường

Windows:

agent_env\Scripts\activate

macOS/Linux:

source agent_env/bin/activate

Cài đặt các thư viện cần thiết

pip install llama-index llama-index-llms-holysheep python-dotenv

Kiểm tra phiên bản

python --version # Nên là Python 3.9 trở lên

Gợi ý: Chụp ảnh màn hình terminal sau khi cài đặt thành công để lưu lại log.

Bước 2: Cấu Hình API Key

# Tạo file .env trong thư mục dự án
touch .env

Nội dung file .env

⚠️ LƯU Ý: Không dùng api.openai.com - dùng HolyShehe AI endpoint

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY LLM_MODEL=deepseek-v3

Gợi ý: Copy API key từ dashboard HolyShehe AI sau khi đăng ký tại đây.

Bước 3: Xây Dựng Agent Workflow Đầu Tiên

# main.py - Agent workflow đơn giản
import os
from dotenv import load_dotenv
from llama_index.core.workflow import (
    Event,
    Context,
    Workflow,
    step,
    StartEvent,
    StopEvent
)
from llama_index.llms.holysheep import HolySheep

Load environment variables

load_dotenv()

Khởi tạo LLM với HolyShehe AI

llm = HolySheep( api_key=os.getenv("HOLYSHEEP_API_KEY"), base_url="https://api.holysheep.ai/v1", # ✅ Endpoint chính xác model="deepseek-v3", temperature=0.7 )

Định nghĩa các Event (tín hiệu trong hệ thống)

class QueryEvent(Event): """Tín hiệu khi có câu hỏi từ user""" query: str user_id: str class ResearchEvent(Event): """Tín hiệu khi agent bắt đầu nghiên cứu""" context: str search_query: str class ResponseEvent(Event): """Tín hiệu khi có response cuối cùng""" answer: str confidence: float

Định nghĩa Workflow

class SimpleAgentWorkflow(Workflow): @step async def process_query(self, ctx: Context, ev: StartEvent) -> QueryEvent: """ Bước 1: Nhận và phân tích câu hỏi - Event-driven: Khi có StartEvent, tự động chạy step này """ query = ev.query print(f"🔔 [Event: QueryEvent] Nhận câu hỏi: {query[:50]}...") # Lưu vào context để các step khác có thể truy cập await ctx.set("original_query", query) await ctx.set("user_id", ev.user_id or "anonymous") # Phân tích intent bằng LLM intent_response = await llm.acomplete( f"Phân tích ý định của câu hỏi sau: {query}" ) return QueryEvent( query=query, user_id=ev.user_id or "anonymous" ) @step async def research(self, ctx: Context, ev: QueryEvent) -> ResearchEvent: """ Bước 2: Nghiên cứu và thu thập thông tin - Event-driven: Chạy khi nhận được QueryEvent """ print("🔔 [Event: ResearchEvent] Bắt đầu nghiên cứu...") # Lấy query từ context original_query = await ctx.get("original_query") # Gọi LLM để nghiên cứu (sử dụng DeepSeek V3.2 - $0.42/1M tokens) research_prompt = f""" Hãy nghiên cứu và cung cấp thông tin chi tiết về: {original_query} Trả lời bằng tiếng Việt, có cấu trúc rõ ràng. """ research_result = await llm.acomplete(research_prompt) return ResearchEvent( context=str(research_result), search_query=original_query ) @step async def generate_response(self, ctx: Context, ev: ResearchEvent) -> StopEvent: """ Bước 3: Tạo response cuối cùng - Event-driven: Chạy khi nhận được ResearchEvent """ print("🔔 [Event: ResponseEvent] Tạo response cuối cùng...") # Tổng hợp thông tin và tạo câu trả lời final_prompt = f""" Dựa trên nghiên cứu sau, hãy trả lời câu hỏi một cách tự nhiên: Nghiên cứu: {ev.context} Câu hỏi: {ev.search_query} """ final_response = await llm.acomplete(final_prompt) # Tính confidence score đơn giản confidence = 0.85 if len(str(final_response)) > 100 else 0.6 return StopEvent( result={ "answer": str(final_response), "confidence": confidence, "model_used": "deepseek-v3", "cost_per_1m_tokens": 0.42 # USD } )

Chạy workflow

async def main(): workflow = SimpleAgentWorkflow(timeout=60, verbose=True) result = await workflow.run( query="Giải thích event-driven architecture trong AI agent", user_id="user_001" ) print("\n" + "="*50) print("📊 KẾT QUẢ:") print(f" Answer: {result.result['answer'][:200]}...") print(f" Confidence: {result.result['confidence']*100}%") print(f" Model: {result.result['model_used']}") print(f" Cost: ${result.result['cost_per_1m_tokens']}/1M tokens") if __name__ == "__main__": import asyncio asyncio.run(main())

Bước 4: Workflow Phức Tạp Hơn - Multi-Agent

# multi_agent_workflow.py - Nhiều Agent cùng làm việc
import os
from typing import List
from llama_index.core.workflow import (
    Event,
    Context,
    Workflow,
    step,
    StartEvent,
    StopEvent,
    WorkflowHandler
)
from llama_index.llms.holysheep import HolySheep
from llama_index.core.tools import FunctionTool

load_dotenv()

Khởi tạo các LLM với chi phí khác nhau

llm_deepseek = HolySheep( api_key=os.getenv("HOLYSHEEP_API_KEY"), base_url="https://api.holysheep.ai/v1", model="deepseek-v3", # $0.42/1M tokens - cho tác vụ đơn giản temperature=0.5 ) llm_sonnet = HolySheep( api_key=os.getenv("HOLYSHEEP_API_KEY"), base_url="https://api.holysheep.ai/v1", model="claude-sonnet-4.5", # $15/1M tokens - cho tác vụ phức tạp temperature=0.3 )

Định nghĩa Events cho multi-agent

class ResearchTaskEvent(Event): task_id: str description: str priority: int # 1 = cao nhất class SynthesisTaskEvent(Event): synthesis_type: str # "technical" | "simple" | "detailed" research_results: List[dict] class FinalReportEvent(Event): report: str sections_completed: List[str]

Tool cho việc tìm kiếm (đơn giản hóa)

def search_wikipedia(query: str) -> str: """Tìm kiếm thông tin đơn giản - sử dụng cho demo""" return f"Kết quả tìm kiếm cho '{query}': Thông tin chi tiết về chủ đề này..." search_tool = FunctionTool.from_defaults(fn=search_wikipedia)

Multi-Agent Workflow

class ResearchReportWorkflow(Workflow): @step async def decompose_task(self, ctx: Context, ev: StartEvent) -> List[ResearchTaskEvent]: """ Bước 1: Phân rã nhiệm vụ thành các sub-tasks Event-driven: Tự động trigger khi nhận StartEvent """ topic = ev.topic print(f"🔔 [Event] Phân rã nhiệm vụ cho chủ đề: {topic}") # Phân rã bằng LLM mạnh decomposition_prompt = f""" Phân rã chủ đề sau thành 3-5 nhiệm vụ con để nghiên cứu: Chủ đề: {topic} Trả lời theo format JSON: {{ "tasks": [ {{"id": "task_1", "description": "...", "priority": 1}}, ... ] }} """ result = await llm_sonnet.acomplete(decomposition_prompt) # Parse và tạo events (đơn giản hóa cho demo) tasks = [ ResearchTaskEvent(task_id="task_1", description=f"Nghiên cứu tổng quan về {topic}", priority=1), ResearchTaskEvent(task_id="task_2", description=f"Phân tích chi tiết {topic}", priority=2), ResearchTaskEvent(task_id="task_3", description=f"So sánh {topic} với các giải pháp khác", priority=3), ] await ctx.set("all_tasks", [t.model_dump() for t in tasks]) return tasks # Trả về nhiều events cùng lúc! @step async def research_task_1(self, ctx: Context, ev: ResearchTaskEvent) -> SynthesisTaskEvent: """ Bước 2a: Agent nghiên cứu Task 1 Chạy song song với các research_task khác """ if ev.task_id != "task_1": return None # Skip nếu không phải task của mình print(f"🔔 [Event: ResearchTaskEvent] Agent A nghiên cứu: {ev.task_id}") # Sử dụng DeepSeek V3.2 (rẻ hơn 35x so với Claude!) research_prompt = f"Nghiên cứu chi tiết: {ev.description}" result = await llm_deepseek.acomplete(research_prompt) return SynthesisTaskEvent( synthesis_type="technical", research_results=[{"task_id": ev.task_id, "content": str(result)}] ) @step async def research_task_2(self, ctx: Context, ev: ResearchTaskEvent) -> SynthesisTaskEvent: """ Bước 2b: Agent nghiên cứu Task 2 Chạy song song với research_task_1 """ if ev.task_id != "task_2": return None print(f"🔔 [Event: ResearchTaskEvent] Agent B nghiên cứu: {ev.task_id}") # Sử dụng DeepSeek V3.2 research_prompt = f"Nghiên cứu chi tiết: {ev.description}" result = await llm_deepseek.acomplete(research_prompt) return SynthesisTaskEvent( synthesis_type="detailed", research_results=[{"task_id": ev.task_id, "content": str(result)}] ) @step async def research_task_3(self, ctx: Context, ev: ResearchTaskEvent) -> SynthesisTaskEvent: """ Bước 2c: Agent nghiên cứu Task 3 """ if ev.task_id != "task_3": return None print(f"🔔 [Event: ResearchTaskEvent] Agent C nghiên cứu: {ev.task_id}") research_prompt = f"Nghiên cứu và so sánh: {ev.description}" result = await llm_deepseek.acomplete(research_prompt) return SynthesisTaskEvent( synthesis_type="comparison", research_results=[{"task_id": ev.task_id, "content": str(result)}] ) @step async def synthesize_results(self, ctx: Context, ev: SynthesisTaskEvent) -> FinalReportEvent: """ Bước 3: Tổng hợp kết quả từ tất cả agents Event-driven: Chạy khi nhận đủ ResearchTaskEvents """ print(f"🔔 [Event: SynthesisTaskEvent] Tổng hợp kết quả...") # Thu thập tất cả research results từ context all_tasks = await ctx.get("all_tasks", []) # Tạo báo cáo cuối cùng bằng Claude Sonnet (vì cần chất lượng cao) synthesis_prompt = f""" Tổng hợp các nghiên cứu sau thành một báo cáo hoàn chỉnh: Loại tổng hợp: {ev.synthesis_type} Kết quả: {ev.research_results} Viết báo cáo bằng tiếng Việt, có cấu trúc rõ ràng. """ final_report = await llm_sonnet.acomplete(synthesis_prompt) return FinalReportEvent( report=str(final_report), sections_completed=[t["task_id"] for t in all_tasks] ) @step async def finalize(self, ctx: Context, ev: FinalReportEvent) -> StopEvent: """ Bước 4: Hoàn thành workflow """ print(f"✅ [StopEvent] Workflow hoàn thành!") return StopEvent(result={ "report": ev.report, "sections": ev.sections_completed, "total_tasks": len(ev.sections_completed), "models_used": { "research": "deepseek-v3 ($0.42/1M tokens)", "synthesis": "claude-sonnet-4.5 ($15/1M tokens)" } })

Chạy multi-agent workflow

async def main(): workflow = ResearchReportWorkflow(timeout=120, verbose=True) result = await workflow.run( topic="Sự khác biệt giữa REST API và GraphQL trong 2024" ) print("\n" + "="*60) print("📊 BÁO CÁO HOÀN THÀNH:") print(f" Số sections: {result.result['total_tasks']}") print(f" Models sử dụng: {result.result['models_used']}") print(f"\n📝 NỘI DUNG:\n{result.result['report'][:500]}...") if __name__ == "__main__": import asyncio asyncio.run(main())

So Sánh Chi Phí: HolyShehe AI vs OpenAI

ModelNền tảngGiá/1M tokensTiết kiệm
GPT-4.1OpenAI$8.00-
Claude Sonnet 4.5Anthropic$15.00-
DeepSeek V3.2HolyShehe AI$0.4295%
Gemini 2.5 FlashHolyShehe AI$2.5069%

"Với workflow phức tạp như trên, nếu dùng Claude Sonnet cho tất cả tác vụ, chi phí có thể lên đến $50-100 mỗi lần chạy. Nhưng với HolyShehe AI, mình chỉ tốn khoảng $2-5."

Event-Driven Flow Visualization

# Biểu diễn flow của workflow bằng ASCII art

┌─────────────────────────────────────────────────────────────────┐
│                     EVENT-DRIVEN AGENT FLOW                     │
└─────────────────────────────────────────────────────────────────┘

    ┌──────────────┐
    │  START EVENT  │
    │  (User Input) │
    └──────┬───────┘
           │
           ▼
    ┌──────────────────┐
    │  decompose_task  │ ◄─── Event: StartEvent
    │     (Step 1)     │
    └──────┬──────────┘
           │
           ▼
    ┌──────┴──────┐
    │             │
    ▼             ▼
┌────────┐  ┌────────┐  ┌────────┐
│Task 1  │  │Task 2  │  │Task 3  │  ◄─── Parallel Execution!
│Research│  │Research│  │Research│
│ Agent A│  │ Agent B│  │ Agent C│
└────┬───┘  └────┬───┘  └────┬───┘
     │            │           │
     └────────────┴───────────┘
                  │
                  ▼
    ┌──────────────────┐
    │ synthesize_results│ ◄─── Event: SynthesisTaskEvent
    │     (Step 3)      │
    └──────┬───────────┘
           │
           ▼
    ┌──────────────────┐
    │     finalize      │ ◄─── Event: FinalReportEvent
    │     (Step 4)      │
    └──────┬───────────┘
           │
           ▼
    ┌──────────────┐
    │  STOP EVENT   │
    │(Final Result) │
    └──────────────┘

═══════════════════════════════════════════════════════════════

TÍNH NĂNG EVENT-DRIVEN:

✅ Async/Await: Các agents chạy song song
✅ Conditional Routing: Events chỉ trigger agents phù hợp
✅ Error Handling: Mỗi step có try/catch riêng
✅ State Management: Context lưu trữ trạng thái giữa các steps
✅ Timeout Control: Workflow có timeout toàn cục

═══════════════════════════════════════════════════════════════

Lỗi Thường Gặp Và Cách Khắc Phục

1. Lỗi "Invalid API Key" hoặc Authentication Error

# ❌ SAI - Dùng endpoint không đúng
llm = HolySheep(
    api_key="sk-xxx",
    base_url="https://api.openai.com/v1"  # ❌ SAI!
)

✅ ĐÚNG - Dùng HolyShehe AI endpoint

llm = HolySheep( api_key="YOUR_HOLYSHEEP_API_KEY", # Key từ HolyShehe AI base_url="https://api.holysheep.ai/v1" # ✅ ĐÚNG! )

Cách kiểm tra key hợp lệ:

import os from dotenv import load_dotenv load_dotenv() api_key = os.getenv("HOLYSHEEP_API_KEY") if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY": print("❌ Vui lòng cập nhật HOLYSHEEP_API_KEY trong file .env") print(" Đăng ký tại: https://www.holysheep.ai/register") else: print("✅ API Key đã được cấu hình đúng")

2. Lỗi "Event type not found" hoặc Step không chạy

# ❌ SAI - Không return đúng event type
@step
async def my_step(self, ctx: Context, ev: QueryEvent) -> None:  # ❌ Return None!
    # Logic xử lý...
    pass  # Step sẽ không trigger step tiếp theo!

✅ ĐÚNG - Luôn return event hoặc StopEvent

@step async def my_step(self, ctx: Context, ev: QueryEvent) -> ResearchEvent: # Logic xử lý... return ResearchEvent(data="processed_data") # ✅ Trigger step tiếp theo

Nếu muốn dừng workflow sớm:

@step async def check_step(self, ctx: Context, ev: QueryEvent) -> StopEvent | QueryEvent: if some_condition: return StopEvent(result={"status": "early_exit"}) # ✅ Dừng sớm else: return QueryEvent(data="continue") # ✅ Tiếp tục flow

3. Lỗi "Timeout exceeded" hoặc Workflow treo

# ❌ SAI - Không set timeout hoặc timeout quá ngắn
workflow = SimpleAgentWorkflow()  # ❌ Không có timeout
workflow.run(query="...")  # Có thể treo vĩnh viễn!

✅ ĐÚNG - Set timeout phù hợp với từng loại task

workflow = SimpleAgentWorkflow( timeout=60, # 60 giây cho task đơn giản verbose=True # In log để debug )

Nếu workflow phức tạp (multi-agent):

complex_workflow = ResearchReportWorkflow( timeout=300, # 5 phút cho multi-agent verbose=True )

Cách handle timeout:

async def run_with_timeout(): try: result = await complex_workflow.run(query="...") return result except asyncio.TimeoutError: print("❌ Workflow vượt quá thời gian cho phép!") print("💡 Gợi ý: Tăng timeout hoặc tối ưu các bước xử lý") return None

4. Lỗi "Context not found" hoặc State bị mất

# ❌ SAI - Gán giá trị sai cách
await ctx.set("user_id", "123")  # Gán trực tiếp

✅ ĐÚNG - Sử dụng async/await đúng cách

@step async def step_1(self, ctx: Context, ev: StartEvent) -> QueryEvent: # Lưu state: await ctx.set("user_id", "123") # ✅ Dùng await await ctx.set("timestamp", datetime.now().isoformat()) # Truy cập state: user_id = await ctx.get("user_id") # ✅ Dùng await return QueryEvent(data="...")

Để debug context:

@step async def debug_step(self, ctx: Context, ev: StartEvent) -> StopEvent: # In tất cả context hiện tại: all_data = await ctx.get() print("📋 Current Context:") for key, value in all_data.items(): print(f" {key}: {value}") return StopEvent(result=all_data)

5. Lỗi "Rate Limit Exceeded"

# ❌ SAI - Gọi API liên tục không giới hạn
for query in queries:  # 1000 queries
    result = await llm.acomplete(query)  # ❌ Rate limit ngay!

✅ ĐÚNG - Thêm rate limiting và retry logic

import asyncio from tenacity import retry, stop_after_attempt, wait_exponential @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10) ) async def call_llm_with_retry(prompt: str, max_tokens: int = 512): try: response = await llm.acomplete( prompt, max_tokens=max_tokens ) return response except RateLimitError: print("⏳ Rate limit hit, đợi 5 giây...") await asyncio.sleep(5) raise # Retry sẽ tự động chạy lại

Sử dụng semaphore để giới hạn concurrency

semaphore = asyncio.Semaphore(3) # Tối đa 3 requests cùng lúc async def safe_llm_call(prompt: str): async with semaphore: return await call_llm_with_retry(prompt)

Kết Luận

Qua bài viết này, mình đã chia sẻ toàn bộ kiến thức để bạn có thể xây dựng event-driven agent workflows với LlamaIndex từ con số 0. Điểm mấu chốt là:

"Mình đã tiết kiệm được hơn $500/tháng khi chuyển từ Claude API sang HolyShehe AI cho các dự án cá nhân. Sự khác biệt về chi phí này thực sự kinh ngạc!"

Bước Tiếp Theo

Để nâng cao kỹ năng, bạn có thể thử:

👉 Đăng ký HolyShehe AI — nhận tín dụng miễn phí khi đăng ký


Bài viết được viết bởi đội ngũ kỹ thuật HolySheep AI. Nếu bạn thấy hữu ích, hãy chia sẻ cho đồng nghiệp!