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ẽ:
- Nhận tín hiệu (event) khi có đơn hàng mới
- Xử lý công việc riêng của mình
- Gửi tín hiệu cho robot tiếp theo
- Đội robot hoạt động song song, hiệu quả cao hơn nhiều
Đó 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
| Model | Nền tảng | Giá/1M tokens | Tiết kiệm |
|---|---|---|---|
| GPT-4.1 | OpenAI | $8.00 | - |
| Claude Sonnet 4.5 | Anthropic | $15.00 | - |
| DeepSeek V3.2 | HolyShehe AI | $0.42 | 95% |
| Gemini 2.5 Flash | HolyShehe AI | $2.50 | 69% |
"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à:
- Event-driven architecture giúp code linh hoạt, dễ mở rộng
- Multi-agent workflows tận dụng parallelism để tăng tốc độ
- HolyShehe AI giúp tiết kiệm đến 85%+ chi phí với DeepSeek V3.2 chỉ $0.42/1M tokens
- Độ trễ dưới 50ms, hỗ trợ WeChat/Alipay, và tín dụng miễn phí khi đăng ký
"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ử:
- Tích hợp thêm RAG (Retrieval-Augmented Generation) vào workflow
- Thêm human-in-the-loop để approve trước khi thực hiện action
- Xây dựng monitoring dashboard để theo dõi token usage
👉 Đă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!