Bốn bài đầu series này nói về concept: agent là gì, tools, control loop, memory. Bài này làm thật. Không framework, không abstraction, chỉ Anthropic SDK và khoảng 100 dòng Python.
Khi bài viết xong, bạn sẽ có một agent chạy được trên máy: đọc file, ghi file, liệt kê thư mục, chạy shell command. Giao cho nó task “đếm tổng số dòng code Python trong một repo”, nó sẽ tự đi tìm, tự đọc, tự cộng, rồi report lại. Không cần hướng dẫn từng bước.
Setup
pip install anthropic
export ANTHROPIC_API_KEY="sk-ant-..."Chỉ một dependency. Không LangChain, không CrewAI, không vector DB. Mục tiêu bài này là thấy rõ cơ chế, không bị framework che khuất.
Model dùng trong bài: claude-sonnet-4-6. Nếu bạn muốn tiết kiệm hơn khi dev, claude-haiku-4-5 cũng chạy được với tools, nhưng reasoning kém hơn với task phức tạp.
Phần 1: Định nghĩa tools
Agent có 4 tools: đọc file, ghi file, liệt kê thư mục, và chạy shell. Mỗi tool là một dict với name, description, và input_schema theo chuẩn JSON Schema.
TOOLS = [
{
"name": "read_file",
"description": (
"Read the full text content of a file. "
"Returns the content as a string. "
"Use this to inspect source code, config files, or any text file."
),
"input_schema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Absolute or relative path to the file",
}
},
"required": ["path"],
},
},
{
"name": "write_file",
"description": (
"Write text content to a file, creating it if it does not exist "
"and overwriting it if it does. Use this to save results or create new files."
),
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Path to write to"},
"content": {"type": "string", "description": "Text content to write"},
},
"required": ["path", "content"],
},
},
{
"name": "list_dir",
"description": (
"List all entries (files and subdirectories) in a directory. "
"Returns a newline-separated list of names. "
"Does not recurse into subdirectories."
),
"input_schema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the directory",
}
},
"required": ["path"],
},
},
{
"name": "run_shell",
"description": (
"Run a shell command and return stdout + stderr combined. "
"Use for counting lines, grepping, or lightweight data processing. "
"Do NOT use for destructive operations like rm, mv, or anything that modifies "
"files outside the target directory."
),
"input_schema": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "Shell command to execute",
}
},
"required": ["command"],
},
},
]Một điểm quan trọng trong description của run_shell: tôi viết rõ “Do NOT use for destructive operations”. LLM đọc description khi quyết định dùng tool. Description mờ nhạt nghĩa là LLM dự đoán sai intent. Sẽ đào sâu hơn về tool design ở bài 11.
Phần 2: Tool dispatcher
Dispatcher nhận tên tool và args từ LLM, gọi function Python tương ứng, trả về string kết quả.
import os
import subprocess
def dispatch_tool(name: str, args: dict) -> str:
"""Execute a tool call and return the result as a string.
Always returns a string, never raises. Errors are wrapped so LLM
can read them and decide whether to retry or report to the user.
"""
try:
if name == "read_file":
path = args["path"]
with open(path, "r", encoding="utf-8") as f:
return f.read()
elif name == "write_file":
path = args["path"]
content = args["content"]
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return f"Wrote {len(content)} chars to {path}"
elif name == "list_dir":
path = args["path"]
entries = sorted(os.listdir(path))
return "\n".join(entries) if entries else "(empty directory)"
elif name == "run_shell":
command = args["command"]
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=30,
)
output = result.stdout + result.stderr
return output.strip() or "(no output)"
else:
return f"Unknown tool: {name}"
except Exception as e:
# Wrap error as string so LLM sees it in tool_result
return f"ERROR: {type(e).__name__}: {e}"Hai điểm thiết kế quan trọng ở đây:
Luôn trả về string, không bao giờ raise. Nếu read_file nhận path không tồn tại, dispatcher trả về "ERROR: FileNotFoundError: ..." thay vì raise exception lên loop. LLM nhìn thấy error này trong tool_result, có thể quyết định thử path khác hoặc báo lại cho user. Nếu raise lên, loop crash và user không hiểu tại sao.
timeout=30 trên run_shell. Không có timeout, một command như find / -name "*.py" có thể chạy nhiều phút. Agent đứng chờ, user không biết chuyện gì. 30 giây đủ cho hầu hết lệnh đọc. Nếu task cần lâu hơn, tăng timeout có chủ đích, không để mặc định vô hạn.
Phần 3: Control loop
Đây là phần trung tâm. Loop nhận user input, gọi LLM, chạy tools khi cần, lặp lại cho đến khi LLM trả về end_turn.
from anthropic import Anthropic
client = Anthropic()
def run_agent(user_input: str, max_iterations: int = 15) -> str:
"""Run the agent loop until LLM signals end_turn or max_iterations is hit."""
messages = [{"role": "user", "content": user_input}]
for iteration in range(max_iterations):
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=4096,
system=(
"You are a file system assistant. You have access to tools to read files, "
"write files, list directories, and run shell commands. "
"Complete the user's task step by step. "
"When you are done, summarize what you did and the final result."
),
tools=TOOLS,
messages=messages,
)
# Append the full assistant response to history
messages.append({"role": "assistant", "content": response.content})
# LLM finished without calling any tool
if response.stop_reason == "end_turn":
# Extract the final text block
for block in response.content:
if hasattr(block, "text"):
return block.text
return "(no text response)"
# LLM wants to call one or more tools
if response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = dispatch_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})
continue
# Unexpected stop reason
return f"Unexpected stop_reason: {response.stop_reason}"
return f"Agent stopped after {max_iterations} iterations without completing the task."Vì sao max_iterations=15 chứ không phải 100? Bài 1 đã nói: max_iterations là budget, không phải safety net. 15 iteration × ~3000 token/iteration với Claude Sonnet 4.6 vào khoảng $0.10-0.15 cho một task. Đủ để làm hầu hết task file system. Nếu agent chạm 15 mà chưa xong, task đó có thể cần thiết kế lại, không phải tăng limit.
Phần 4: Entry point và demo
import sys
def main():
if len(sys.argv) > 1:
task = " ".join(sys.argv[1:])
else:
task = input("Task: ").strip()
if not task:
print("No task provided.")
return
print(f"\nRunning agent on: {task}\n{'=' * 60}")
result = run_agent(task)
print(result)
if __name__ == "__main__":
main()Chạy thử với task thật:
python agent.py "Đọc tất cả file .py trong thư mục hiện tại, đếm tổng số dòng code (không tính dòng trống và comment), rồi ghi kết quả vào report.txt"Ví dụ session thật (tóm tắt):
Running agent on: Đọc tất cả file .py trong thư mục hiện tại...
============================================================
[Iteration 1] LLM gọi list_dir(".")
[Iteration 2] LLM gọi read_file("agent.py")
[Iteration 3] LLM gọi run_shell("wc -l *.py")
[Iteration 4] LLM gọi write_file("report.txt", "...")
[Iteration 5] LLM trả về end_turn
Kết quả:
- agent.py: 98 dòng (82 dòng code, 16 dòng comment/blank)
- Tổng: 82 dòng code thực
- Đã ghi kết quả vào report.txtBốn tool call, năm iteration. Agent tự quyết định flow: list trước để biết có gì, đọc file, dùng shell để đếm nhanh, ghi kết quả, báo cáo. Không có instruction nào về thứ tự. Đây là điểm khác biệt cốt lõi so với workflow tuyến tính đã nói ở bài 1.
Pitfall: shell tool và command injection
run_shell với shell=True là con dao hai lưỡi. Nếu agent nhận user input không sanitized rồi nhúng vào command, bạn có bug nghiêm trọng.
Ví dụ nguy hiểm:
# User nhập: "report.txt; rm -rf ~"
task = "Đọc file report.txt; rm -rf ~"
# LLM có thể tạo command: "cat report.txt; rm -rf ~"
# shell=True sẽ chạy cả hai lệnhLLM thường không làm điều này nếu system prompt rõ ràng, nhưng “thường không” không phải “không bao giờ”. Đặc biệt nếu task đến từ user không tin cậy (ví dụ: external webhook, untrusted input), risk này là thật.
Ba lớp phòng thủ tối thiểu:
Lớp 1: Allowlist command prefix. Chỉ cho phép một số command cụ thể:
ALLOWED_COMMANDS = {"wc", "find", "grep", "cat", "ls", "echo", "head", "tail"}
def run_shell_safe(command: str) -> str:
first_word = command.strip().split()[0]
if first_word not in ALLOWED_COMMANDS:
return f"ERROR: Command '{first_word}' not in allowlist"
# ...Lớp 2: Không dùng shell=True khi có thể. Truyền list thay vì string cho subprocess.run:
# Thay vì:
subprocess.run(command, shell=True, ...)
# Dùng:
import shlex
subprocess.run(shlex.split(command), shell=False, ...)shlex.split sẽ không diễn giải ;, &&, || như operator mà coi là literal string.
Lớp 3: Sandbox. Chạy toàn bộ agent trong container hoặc VM. Sẽ đào sâu ở bài 12.
Với agent nội bộ (chỉ chạy trên máy của bạn, task từ bản thân) thì lớp 1 là đủ. Với agent nhận input từ user ngoài: cần cả ba.
Toàn bộ file
Ghép các phần lại, đây là agent.py đầy đủ dưới 120 dòng:
"""agent.py -- minimal file-system agent with Anthropic SDK.
Usage:
python agent.py "Count lines of Python code in current directory"
python agent.py # interactive prompt
"""
import os
import subprocess
import sys
from anthropic import Anthropic
client = Anthropic()
# --- Tool definitions ---
TOOLS = [
{
"name": "read_file",
"description": (
"Read the full text content of a file. "
"Returns the content as a string."
),
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Path to the file"}
},
"required": ["path"],
},
},
{
"name": "write_file",
"description": "Write text content to a file, overwriting if it exists.",
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string"},
"content": {"type": "string"},
},
"required": ["path", "content"],
},
},
{
"name": "list_dir",
"description": "List entries in a directory (non-recursive).",
"input_schema": {
"type": "object",
"properties": {"path": {"type": "string"}},
"required": ["path"],
},
},
{
"name": "run_shell",
"description": (
"Run a shell command and return stdout + stderr. "
"Do NOT use for destructive operations (rm, mv, etc.)."
),
"input_schema": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"],
},
},
]
# --- Tool dispatcher ---
def dispatch_tool(name: str, args: dict) -> str:
try:
if name == "read_file":
with open(args["path"], "r", encoding="utf-8") as f:
return f.read()
elif name == "write_file":
os.makedirs(os.path.dirname(args["path"]) or ".", exist_ok=True)
with open(args["path"], "w", encoding="utf-8") as f:
f.write(args["content"])
return f"Wrote {len(args['content'])} chars to {args['path']}"
elif name == "list_dir":
entries = sorted(os.listdir(args["path"]))
return "\n".join(entries) or "(empty)"
elif name == "run_shell":
r = subprocess.run(
args["command"], shell=True, capture_output=True,
text=True, timeout=30
)
return (r.stdout + r.stderr).strip() or "(no output)"
return f"Unknown tool: {name}"
except Exception as e:
return f"ERROR: {type(e).__name__}: {e}"
# --- Agent loop ---
def run_agent(user_input: str, max_iterations: int = 15) -> str:
messages = [{"role": "user", "content": user_input}]
for _ in range(max_iterations):
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=4096,
system=(
"You are a file system assistant. Complete the user's task "
"using the available tools. Summarize the final result clearly."
),
tools=TOOLS,
messages=messages,
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason == "end_turn":
for block in response.content:
if hasattr(block, "text"):
return block.text
return "(no text)"
if response.stop_reason == "tool_use":
results = [
{
"type": "tool_result",
"tool_use_id": b.id,
"content": dispatch_tool(b.name, b.input),
}
for b in response.content
if b.type == "tool_use"
]
messages.append({"role": "user", "content": results})
return f"Stopped after {max_iterations} iterations."
# --- Entry point ---
def main():
task = " ".join(sys.argv[1:]) if len(sys.argv) > 1 else input("Task: ").strip()
if not task:
print("No task.")
return
print(f"\nTask: {task}\n{'=' * 60}")
print(run_agent(task))
if __name__ == "__main__":
main()Đây là bản sạch để clone và chạy thử. Không cần thêm gì ngoài pip install anthropic và API key.
Cheatsheet
| Phần | Vai trò | Điểm hay sai |
|---|---|---|
TOOLS list | Schema expose cho LLM | Description mờ, LLM đoán sai intent |
dispatch_tool | Chạy tool, trả string | Raise exception thay vì wrap error |
messages list | Conversation history (memory) | Không append tool_result, LLM mất context |
stop_reason == "tool_use" | LLM muốn gọi tool | Quên handle, loop thoát sớm |
stop_reason == "end_turn" | LLM tự khai báo xong | Tin tuyệt đối, không verify task thật xong |
max_iterations | Token budget | Để quá cao, đốt tiền khi agent stuck |
shell=True | Tiện nhưng nguy hiểm | Command injection nếu input từ user ngoài |
Lời kết
Bài này kết thúc Part 1 của series. Từ bài 1 đến bài 5, chúng ta đi từ định nghĩa agent, qua từng thành phần riêng lẻ, đến một agent chạy được trên máy. Khoảng 100 dòng Python, không framework nào.
Bước tiếp theo quan trọng: copy code này về, chạy với một task thật của bạn, xem agent fail ở đâu. Đó là cách nhanh nhất để biến concept thành intuition.
Part 2 bắt đầu ở bài 6: ReAct, thought-action-observation cycle. Agent ở bài 5 phản xạ: nhận task, chọn tool, chạy. ReAct thêm một bước: nghĩ thành lời trước khi hành động. Nghe có vẻ overhead, nhưng đây là kỹ thuật làm agent giải quyết được class bài toán phức tạp hơn hẳn. Hẹn gặp ở bài 6.