On this page16 sections
Xây Dựng AI Agent Từ Đầu Với Claude API (Có Full Code)
Hầu hết các tutorial về AI Agent đều dừng lại ở mức "chatbot có system prompt". Bài này không như vậy. Chúng ta sẽ xây dựng một agent thực sự — có vòng lặp ReAct, gọi tool thực tế, xử lý lỗi, streaming, và cả async execution. Đây là kiến trúc bạn có thể đưa vào production.
1. AI Agent là gì và tại sao ReAct?
Một AI Agent không phải là chatbot. Sự khác biệt cốt lõi:
ChatbotAI AgentNhận prompt → trả lờiNhận mục tiêu → lập kế hoạch → thực thiStateless (thường)Có vòng lặp quan sát–hành độngChỉ dùng kiến thức nội tạiGọi được tool ngoài (API, DB, file...)Một lượtNhiều bước đến khi xong mục tiêu
Pattern ReAct (Reasoning + Acting) là backbone của hầu hết production agents:
Người dùng → [Câu hỏi / Mục tiêu]
↓
Claude suy nghĩ (Reasoning)
↓
Chọn tool cần dùng (Acting)
↓
Thực thi tool → Nhận kết quả
↓
Quan sát kết quả (Observing)
↓
Lặp lại... hoặc trả lời cuối cùngClaude API hỗ trợ pattern này cực kỳ tự nhiên thông qua cơ chế Tool Use (Function Calling).
2. Cài Đặt Môi Trường
bash
# Tạo virtual environment
python -m venv agent-env
source agent-env/bin/activate # Windows: agent-env\Scripts\activate
# Cài dependencies
pip install anthropic python-dotenv aiohttp tenacityTạo file .env ở root project:
bash
# .env
ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxCấu trúc project:
claude-agent/
├── .env
├── agent.py # Agent chính
├── tools.py # Định nghĩa và thực thi tools
├── memory.py # Quản lý memory
└── main.py # Entry point3. Kết Nối Claude API Cơ Bản
Trước khi xây agent, hãy xác nhận kết nối hoạt động:
python
# agent.py
import os
import anthropic
from dotenv import load_dotenv
load_dotenv()
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
def ask_claude(prompt: str, system: str = None) -> str:
"""Gọi Claude với một prompt đơn giản."""
kwargs = {
"model": "claude-sonnet-4-5",
"max_tokens": 1024,
"messages": [{"role": "user", "content": prompt}]
}
if system:
kwargs["system"] = system
response = client.messages.create(**kwargs)
return response.content[0].text
# Test nhanh
if __name__ == "__main__":
result = ask_claude("Tính 15 * 24 + 100. Chỉ trả lời con số.")
print(f"Kết quả: {result}")Chạy thử:
bash
python agent.py
# Kết quả: 4604. Định Nghĩa Tools Cho Agent
Tools là "bàn tay" của agent — không có tools, Claude chỉ có thể suy nghĩ, không thể hành động.
4.1 Schema của Tools (JSON)
Claude đọc JSON schema này để biết tool làm gì và cần tham số gì:
python
# tools.py
import math
import json
import os
from datetime import datetime
# ---- Định nghĩa schema cho Claude ----
TOOL_SCHEMAS = [
{
"name": "calculator",
"description": (
"Thực hiện các phép tính toán học. "
"Dùng khi cần tính số, không tự tính trong đầu."
),
"input_schema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Biểu thức toán học, ví dụ: '15 * 24 + 100' hoặc 'sqrt(144)'"
}
},
"required": ["expression"]
}
},
{
"name": "web_search",
"description": (
"Tìm kiếm thông tin trên web. "
"Dùng khi cần thông tin hiện tại hoặc sự kiện mới nhất."
),
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Từ khóa hoặc câu hỏi cần tìm kiếm"
},
"max_results": {
"type": "integer",
"description": "Số kết quả tối đa (mặc định: 3)",
"default": 3
}
},
"required": ["query"]
}
},
{
"name": "save_to_file",
"description": "Lưu nội dung văn bản vào file cục bộ.",
"input_schema": {
"type": "object",
"properties": {
"filename": {
"type": "string",
"description": "Tên file, ví dụ: 'report.txt' hoặc 'output/result.md'"
},
"content": {
"type": "string",
"description": "Nội dung cần lưu vào file"
},
"mode": {
"type": "string",
"enum": ["write", "append"],
"description": "Ghi đè (write) hay thêm vào cuối (append). Mặc định: write"
}
},
"required": ["filename", "content"]
}
},
{
"name": "get_current_time",
"description": "Lấy ngày giờ hiện tại theo múi giờ được chỉ định.",
"input_schema": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "Múi giờ, ví dụ: 'Asia/Ho_Chi_Minh', 'UTC'. Mặc định: UTC"
}
}
}
}
]4.2 Hàm Thực Thi Tools
python
# tools.py (tiếp theo)
# ---- Các hàm thực thi ----
def calculator(expression: str) -> str:
"""Tính toán biểu thức toán học một cách an toàn."""
try:
# Chỉ cho phép các hàm toán học an toàn
safe_globals = {
"__builtins__": {},
**{k: v for k, v in math.__dict__.items() if not k.startswith("__")}
}
result = eval(expression, safe_globals, {})
return json.dumps({"result": result, "expression": expression})
except ZeroDivisionError:
return json.dumps({"error": "Không thể chia cho 0"})
except Exception as e:
return json.dumps({"error": f"Biểu thức không hợp lệ: {str(e)}"})
def web_search(query: str, max_results: int = 3) -> str:
"""
Tìm kiếm web (mô phỏng — trong production thay bằng SerpAPI/Tavily/Brave).
"""
# Trong production: tích hợp API thật ở đây
# import httpx
# response = httpx.get("https://api.tavily.com/search", params={"q": query, "api_key": ...})
simulated_results = [
{
"title": f"Kết quả 1 về: {query}",
"url": f"https://example.com/result-1",
"snippet": f"Thông tin chi tiết về {query} từ nguồn uy tín. Nội dung được cập nhật năm 2025."
},
{
"title": f"Kết quả 2 về: {query}",
"url": f"https://example.com/result-2",
"snippet": f"Phân tích chuyên sâu về {query} với dữ liệu mới nhất."
}
]
return json.dumps({
"query": query,
"results": simulated_results[:max_results],
"note": "Đây là kết quả mô phỏng. Tích hợp API thật để dùng production."
}, ensure_ascii=False)
def save_to_file(filename: str, content: str, mode: str = "write") -> str:
"""Lưu nội dung vào file."""
try:
# Tạo thư mục nếu chưa tồn tại
dir_name = os.path.dirname(filename)
if dir_name:
os.makedirs(dir_name, exist_ok=True)
file_mode = "w" if mode == "write" else "a"
with open(filename, file_mode, encoding="utf-8") as f:
f.write(content)
return json.dumps({
"success": True,
"filename": filename,
"bytes_written": len(content.encode("utf-8")),
"mode": mode
})
except Exception as e:
return json.dumps({"success": False, "error": str(e)})
def get_current_time(timezone: str = "UTC") -> str:
"""Lấy thời gian hiện tại."""
now = datetime.utcnow()
return json.dumps({
"datetime": now.strftime("%Y-%m-%d %H:%M:%S"),
"timezone": timezone,
"note": "Cài pytz để hỗ trợ múi giờ thực sự"
})
# ---- Tool Dispatcher ----
TOOL_REGISTRY = {
"calculator": calculator,
"web_search": web_search,
"save_to_file": save_to_file,
"get_current_time": get_current_time,
}
def execute_tool(tool_name: str, tool_input: dict) -> str:
"""Thực thi tool theo tên và input."""
if tool_name not in TOOL_REGISTRY:
return json.dumps({"error": f"Tool không tồn tại: {tool_name}"})
handler = TOOL_REGISTRY[tool_name]
try:
return handler(**tool_input)
except TypeError as e:
return json.dumps({"error": f"Tham số không hợp lệ: {str(e)}"})
except Exception as e:
return json.dumps({"error": f"Lỗi thực thi: {str(e)}"})5. Xây Dựng Vòng Lặp ReAct
Đây là phần cốt lõi. Vòng lặp ReAct hoạt động như sau:
1. Gửi query + danh sách tools lên Claude
2. Claude trả về: tool_use (muốn gọi tool) HOẶC end_turn (có câu trả lời)
3. Nếu tool_use → thực thi tool → gửi kết quả lại cho Claude
4. Lặp lại đến khi end_turnpython
# agent.py
import json
from typing import Optional
from tools import TOOL_SCHEMAS, execute_tool
SYSTEM_PROMPT = """Bạn là một AI Agent thông minh với khả năng sử dụng tools.
Nguyên tắc làm việc:
- Suy nghĩ từng bước trước khi hành động
- Dùng tool khi cần dữ liệu thực tế hoặc tính toán phức tạp
- Không tự bịa kết quả tool — hãy gọi tool thật
- Tổng hợp kết quả rõ ràng, súc tích ở câu trả lời cuối
Ngôn ngữ: Trả lời bằng tiếng Việt trừ khi được yêu cầu khác."""
def run_agent(
user_query: str,
max_iterations: int = 10,
verbose: bool = True
) -> str:
"""
Chạy agent với vòng lặp ReAct.
Args:
user_query: Câu hỏi/yêu cầu của người dùng
max_iterations: Số vòng lặp tối đa (bảo vệ khỏi vòng lặp vô hạn)
verbose: In log chi tiết ra console
Returns:
Câu trả lời cuối cùng của agent
"""
if verbose:
print(f"\n{'='*60}")
print(f"🧑 Người dùng: {user_query}")
print(f"{'='*60}")
messages = [{"role": "user", "content": user_query}]
for iteration in range(1, max_iterations + 1):
if verbose:
print(f"\n[Vòng lặp {iteration}]")
# --- Gọi Claude ---
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=TOOL_SCHEMAS,
messages=messages
)
stop_reason = response.stop_reason
if verbose:
print(f" Stop reason: {stop_reason}")
# --- Claude đã có câu trả lời cuối ---
if stop_reason == "end_turn":
final_answer = ""
for block in response.content:
if hasattr(block, "text"):
final_answer += block.text
if verbose:
print(f"\n{'='*60}")
print(f"🤖 Agent: {final_answer}")
print(f"{'='*60}")
return final_answer
# --- Claude muốn gọi tool ---
if stop_reason == "tool_use":
# Lưu response của assistant vào history
messages.append({
"role": "assistant",
"content": response.content
})
# Xử lý từng tool call
tool_results = []
for block in response.content:
if block.type == "tool_use":
tool_name = block.name
tool_input = block.input
if verbose:
print(f" 🔧 Gọi tool: {tool_name}")
print(f" Input: {json.dumps(tool_input, ensure_ascii=False)}")
# Thực thi tool
result = execute_tool(tool_name, tool_input)
if verbose:
print(f" Kết quả: {result}")
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
# Gửi kết quả tool lại cho Claude
messages.append({
"role": "user",
"content": tool_results
})
else:
# Stop reason bất thường (max_tokens, etc.)
if verbose:
print(f" ⚠️ Stop reason không mong đợi: {stop_reason}")
break
return "⚠️ Agent đã đạt giới hạn vòng lặp mà chưa có câu trả lời cuối."6. Streaming Response
Với streaming, người dùng thấy câu trả lời hiện ra từng chữ thay vì chờ toàn bộ:
python
# streaming_agent.py
def run_agent_streaming(user_query: str) -> str:
"""Agent với streaming — hiển thị response theo thời gian thực."""
messages = [{"role": "user", "content": user_query}]
full_response = ""
print(f"\n🤖 Agent: ", end="", flush=True)
with client.messages.stream(
model="claude-sonnet-4-5",
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=TOOL_SCHEMAS,
messages=messages
) as stream:
for event in stream:
# Stream text delta
if hasattr(event, "type"):
if event.type == "content_block_delta":
if hasattr(event.delta, "text"):
text_chunk = event.delta.text
print(text_chunk, end="", flush=True)
full_response += text_chunk
print() # Xuống dòng sau khi xong
# Nếu cần xử lý tool use sau streaming
final_message = stream.get_final_message()
if final_message.stop_reason == "tool_use":
# Xử lý tool calls tương tự như ở run_agent()
# ... (logic giống phần trên)
pass
return full_response7. Async Agent Với Tool Song Song
Khi agent cần gọi nhiều tools cùng lúc, async giúp giảm latency đáng kể:
python
# async_agent.py
import asyncio
import aiohttp
import anthropic
async_client = anthropic.AsyncAnthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
async def execute_tool_async(tool_name: str, tool_input: dict) -> tuple[str, str]:
"""
Thực thi tool bất đồng bộ.
Trả về (tool_use_id, result) — nhưng ở đây ta trả (tool_name, result) để debug.
"""
# Giả lập I/O bất đồng bộ (gọi API, đọc DB...)
await asyncio.sleep(0.1)
result = execute_tool(tool_name, tool_input)
return tool_name, result
async def run_agent_async(user_query: str, max_iterations: int = 10) -> str:
"""Agent bất đồng bộ — các tool calls được thực thi song song."""
messages = [{"role": "user", "content": user_query}]
for iteration in range(1, max_iterations + 1):
response = await async_client.messages.create(
model="claude-sonnet-4-5",
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=TOOL_SCHEMAS,
messages=messages
)
if response.stop_reason == "end_turn":
return "".join(
block.text for block in response.content
if hasattr(block, "text")
)
if response.stop_reason == "tool_use":
messages.append({"role": "assistant", "content": response.content})
# Thu thập tất cả tool calls
tool_calls = [
(block.id, block.name, block.input)
for block in response.content
if block.type == "tool_use"
]
# Thực thi TẤT CẢ tools song song!
tasks = [
execute_tool_async(name, inp)
for _, name, inp in tool_calls
]
results_raw = await asyncio.gather(*tasks)
# Ghép kết quả
tool_results = [
{
"type": "tool_result",
"tool_use_id": tool_id,
"content": result
}
for (tool_id, _, _), (_, result)
in zip(tool_calls, results_raw)
]
messages.append({"role": "user", "content": tool_results})
return "Agent đạt giới hạn vòng lặp."
# Chạy async agent
if __name__ == "__main__":
result = asyncio.run(run_agent_async(
"Tính căn bậc hai của 256, đồng thời tìm kiếm thông tin về Python asyncio, "
"và cho tôi biết thời gian hiện tại."
))
print(result)Tại sao async quan trọng? Giả sử agent cần gọi 3 tools, mỗi cái mất 500ms:
Tuần tự: 500 + 500 + 500 = 1500ms
Song song (async): max(500, 500, 500) = ~500ms ⚡
8. Thêm Memory Cho Agent
Agent mặc định là stateless — mỗi cuộc gọi run_agent() bắt đầu lại từ đầu. Memory giúp agent nhớ ngữ cảnh qua nhiều lượt:
python
# memory.py
from dataclasses import dataclass, field
from typing import Any
import json
from datetime import datetime
@dataclass
class ConversationMemory:
"""Quản lý lịch sử hội thoại cho agent."""
history: list[dict] = field(default_factory=list)
max_tokens_estimate: int = 100_000 # Giới hạn để tránh vượt context window
def add_user(self, content: str):
self.history.append({"role": "user", "content": content})
def add_assistant(self, content: Any):
"""content có thể là string hoặc list[ContentBlock]"""
if isinstance(content, str):
self.history.append({"role": "assistant", "content": content})
else:
self.history.append({"role": "assistant", "content": content})
def add_tool_results(self, results: list[dict]):
self.history.append({"role": "user", "content": results})
def get_history(self) -> list[dict]:
return self.history.copy()
def clear(self):
self.history.clear()
def summarize_if_needed(self):
"""
Trong production: khi history quá dài, gọi Claude để tóm tắt
các tin nhắn cũ và thay thế bằng bản tóm tắt.
"""
# Ước tính token đơn giản (4 chars ~ 1 token)
total_chars = sum(
len(str(msg.get("content", "")))
for msg in self.history
)
estimated_tokens = total_chars // 4
if estimated_tokens > self.max_tokens_estimate * 0.8:
print("⚠️ History gần đầy — nên implement summarization")
class AgentWithMemory:
"""Agent có bộ nhớ hội thoại liên tục."""
def __init__(self):
self.memory = ConversationMemory()
self.turn_count = 0
def chat(self, user_message: str, verbose: bool = False) -> str:
"""Gửi tin nhắn và nhận phản hồi, giữ nguyên context."""
self.turn_count += 1
self.memory.add_user(user_message)
if verbose:
print(f"\n[Lượt {self.turn_count}] 🧑: {user_message}")
max_iterations = 10
for _ in range(max_iterations):
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=TOOL_SCHEMAS,
messages=self.memory.get_history()
)
if response.stop_reason == "end_turn":
answer = "".join(
block.text for block in response.content
if hasattr(block, "text")
)
self.memory.add_assistant(answer)
if verbose:
print(f"🤖: {answer}")
return answer
if response.stop_reason == "tool_use":
self.memory.add_assistant(response.content)
tool_results = []
for block in response.content:
if block.type == "tool_use":
if verbose:
print(f" 🔧 Tool: {block.name}({block.input})")
result = execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
self.memory.add_tool_results(tool_results)
self.memory.summarize_if_needed()
return "Đã đạt giới hạn vòng lặp."
def reset(self):
"""Reset toàn bộ memory."""
self.memory.clear()
self.turn_count = 0
print("✅ Memory đã được xóa.")Dùng agent có memory:
python
agent = AgentWithMemory()
# Lượt 1
print(agent.chat("Ngân sách của tôi là 50 triệu VNĐ. Tính lãi suất 7%/năm trong 5 năm."))
# Lượt 2 — Claude nhớ 50 triệu và 7% từ lượt trước!
print(agent.chat("Bây giờ tính cho 10 năm với cùng con số đó."))
# Lượt 3
print(agent.chat("Lưu hai kết quả vừa tính vào file 'dau_tu.txt'"))9. Error Handling & Retry
Production agent cần chịu được lỗi mạng, rate limit, và tool failure:
python
# robust_agent.py
import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import anthropic
class RobustAgent:
"""Agent với error handling và retry tự động."""
def __init__(self):
self.client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
@retry(
retry=retry_if_exception_type((
anthropic.APIConnectionError,
anthropic.RateLimitError,
anthropic.InternalServerError,
)),
wait=wait_exponential(multiplier=1, min=2, max=30),
stop=stop_after_attempt(3)
)
def _call_claude(self, messages: list, **kwargs) -> anthropic.types.Message:
"""Gọi Claude với retry tự động khi gặp lỗi tạm thời."""
return self.client.messages.create(
model="claude-sonnet-4-5",
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=TOOL_SCHEMAS,
messages=messages,
**kwargs
)
def _safe_execute_tool(self, tool_name: str, tool_input: dict) -> str:
"""Thực thi tool với error isolation — một tool lỗi không ảnh hưởng cả agent."""
try:
result = execute_tool(tool_name, tool_input)
return result
except Exception as e:
error_msg = f"Tool '{tool_name}' gặp lỗi: {str(e)}"
print(f" ❌ {error_msg}")
return json.dumps({"error": error_msg})
def run(self, query: str, verbose: bool = True) -> str:
messages = [{"role": "user", "content": query}]
for iteration in range(10):
try:
response = self._call_claude(messages)
except anthropic.APIError as e:
return f"Lỗi API sau 3 lần thử: {str(e)}"
except Exception as e:
return f"Lỗi không mong đợi: {str(e)}"
if response.stop_reason == "end_turn":
return "".join(
b.text for b in response.content if hasattr(b, "text")
)
if response.stop_reason == "tool_use":
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = self._safe_execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
messages.append({"role": "user", "content": tool_results})
return "Đã đạt giới hạn vòng lặp."10. Chạy Thử và Kết Luận
File main.py tổng hợp:
python
# main.py
from agent import run_agent
from memory import AgentWithMemory
from robust_agent import RobustAgent
def demo_basic():
print("\n" + "="*60)
print("DEMO 1: Agent Cơ Bản")
print("="*60)
run_agent(
"Tính lãi kép của 100 triệu VNĐ với lãi suất 8%/năm trong 15 năm, "
"rồi lưu kết quả vào file 'tinh_lai_khang.txt'",
verbose=True
)
def demo_memory():
print("\n" + "="*60)
print("DEMO 2: Agent Có Memory")
print("="*60)
agent = AgentWithMemory()
r1 = agent.chat("Vốn đầu tư ban đầu của tôi là 200 triệu VNĐ.", verbose=True)
r2 = agent.chat("Tính lợi nhuận sau 5 năm với lãi suất 10%/năm.", verbose=True)
r3 = agent.chat("Nếu rút 50 triệu sau năm thứ 3, thì năm 5 còn bao nhiêu?", verbose=True)
def demo_robust():
print("\n" + "="*60)
print("DEMO 3: Agent Robust (Error Handling)")
print("="*60)
agent = RobustAgent()
result = agent.run(
"Tìm thông tin về xu hướng AI năm 2025 và tóm tắt các điểm chính.",
verbose=True
)
print(result)
if __name__ == "__main__":
demo_basic()
demo_memory()
demo_robust()Chạy:
bash
python main.pyTóm Tắt Kiến Trúc
┌─────────────────────────────────────────────────┐
│ AI AGENT │
│ │
│ ┌─────────┐ ┌──────────┐ ┌─────────────┐ │
│ │ Memory │◄───│ ReAct │───►│ Tool │ │
│ │ Manager │ │ Loop │ │ Dispatcher │ │
│ └─────────┘ └────┬─────┘ └──────┬──────┘ │
│ │ │ │
│ ┌──────▼──────┐ ┌──────▼──────┐ │
│ │ Claude API │ │ Tools │ │
│ │ (Sonnet 4.5)│ │ calculator │ │
│ └─────────────┘ │ web_search │ │
│ │ save_file │ │
│ │ get_time │ │
│ └─────────────┘ │
└─────────────────────────────────────────────────┘Những gì bạn đã xây dựng:
Tính năngTrạng tháiVòng lặp ReAct cơ bản✅Custom tools (calculator, search, file I/O)✅Streaming response✅Async + parallel tool execution✅Conversation memory✅Error handling & auto-retry✅
Bước tiếp theo:
Thay web_search bằng API thật: Tích hợp Tavily, Brave Search, hoặc SerpAPI để search thật.
Persistent memory: Lưu conversation_history vào Redis hoặc PostgreSQL giữa các session.
Structured output: Dùng Pydantic để validate tool inputs/outputs chặt chẽ hơn.
Observability: Tích hợp LangSmith hoặc Langfuse để trace từng bước agent trong production.
Multi-agent: Nhiều agents chuyên biệt phối hợp với nhau (orchestrator + worker agents).
Bài viết này lấy cảm hứng từ tutorial của Dextra Labs và mở rộng thêm các tính năng production-ready như async, memory management, và robust error handling.