English version: The Agent Loop: The 40 Lines Every AI Agent Is Built On. Đây là phần 1 của series sáu bài về AI agent trong production. Phần 2: Agent Harness.

Agent đầu tiên tôi đưa cho người dùng thật là một bot xử lý tài liệu cho một team vận hành. Nó đọc hợp đồng scan, bóc ngày tháng và các bên ký kết, đối chiếu với database rồi viết tóm tắt. Demo mất hai tuần. Làm cho nó chạy ngoan ngoãn suốt ba tháng thì mất gần hết phần còn lại của năm, và hầu như không có giờ nào trong số đó dành cho prompt. Thời gian đi vào cái vòng lặp bao quanh model: khi nào dừng, làm gì khi tool lỗi, được tiêu bao nhiêu tiền, và làm sao tái hiện lại một lần chạy khi có người hỏi "sao nó lại làm thế?"

Vòng lặp đó là chủ đề của bài này. Hiểu nó kỹ thì mọi framework (LangGraph, OpenAI Agents SDK, Claude Agent SDK, CrewAI, hay cái gì đó ra mắt quý sau) chỉ còn là lớp tiện ích bọc quanh một thứ bạn có thể tự viết trong một buổi chiều. Không hiểu nó thì framework sẽ giấu đi đúng những phần sẽ làm bạn đau.

Agent thực chất là gì#

Một lần chat completion là một lệnh gọi: prompt vào, text ra. Agent là một model được phép yêu cầu làm việc gì đó, nhìn thấy kết quả, rồi tự quyết định bước tiếp theo. Toàn bộ "ma thuật" nằm ở một vòng lặp:

1. Gửi hội thoại cho model, kèm danh sách tool nó được phép gọi.
2. Nếu model trả lời bằng text thuần, xong.
3. Nếu model trả lời bằng một tool call, chạy tool, nối kết quả vào hội thoại, quay lại bước 1.

Hết. Bài "Building effective agents" của Anthropic cuối 2024 nói đúng ý này nhưng cẩn thận hơn: phần lớn hệ thống thành công là các pattern đơn giản, ghép được với nhau, và một agent tự chủ hoàn toàn chẳng qua là vòng lặp trên với model tự quyết số lần đi vòng. Paper ReAct năm 2022 đã mô tả đúng hình dạng này từ trước khi mọi tooling hiện nay tồn tại.

Thứ biến nó thành agent chứ không phải workflow là ai nắm control flow. Trong workflow, code của bạn quyết định thứ tự các bước và model điền vào chỗ trống. Trong agent, model quyết định thứ tự. Khác biệt duy nhất đó là lý do agent xử lý được bài toán mở, và cũng là lý do nó có thể đốt bốn mươi đô chỉ để chạy ls trong vòng lặp.

Khi nào không nên làm agent#

Trước khi vào code, đây là bộ lọc tôi áp cho mọi yêu cầu kiểu "làm cái này thành agent đi", vì vòng lặp rất quyến rũ mà đa số bài toán không cần nó.

Nếu bạn viết được chuỗi các bước lên bảng trắng, hãy viết chúng thành code và gọi model bên trong từng bước. Đó là workflow: rẻ hơn, nhanh hơn, test được bằng unit test bình thường, và hỏng ở những chỗ đoán trước được. Pipeline hợp đồng của chúng tôi rốt cuộc thành đúng như vậy: trích xuất, validate, đối chiếu, tóm tắt, bốn lần gọi model theo thứ tự cố định, còn agent loop chỉ giữ lại cho 5% tài liệu rớt validate và cần điều tra thêm.

Dùng vòng lặp khi con đường đến câu trả lời phụ thuộc vào thứ bạn phát hiện dọc đường: debug, nghiên cứu trải trên nhiều file, làm sạch dữ liệu mà mỗi dataset hỏng một kiểu, bất cứ việc gì mà một chuyên gia sẽ nói "để tôi mò thử đã". Cái giá của agent là phương sai. Mỗi lần chạy tốn thời gian, tiền và số bước khác nhau, và sẽ có lần hỏng theo cách bạn không lường. Bạn chấp nhận phương sai đó để đổi lấy việc phủ được những case không liệt kê nổi. Nếu liệt kê được thì bạn không cần phương sai.

Điểm cân bằng thật thà là workflow có agent bên trong: một khung ngoài cố định, và một vòng lặp có giới hạn ở một hai điểm thực sự cần thăm dò. Phần lớn thứ chúng tôi ship trông như vậy.

Vòng lặp trong 40 dòng#

Đây là phiên bản nhỏ nhất mà tôi vẫn gọi là agent. Dùng thẳng Anthropic SDK, không framework. Cần pip install anthropic và biến môi trường ANTHROPIC_API_KEY.

# agent_loop.py
import json
import os
import subprocess
from pathlib import Path

import anthropic

MODEL = os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-5-5")
WORKDIR = Path(os.environ.get("AGENT_WORKDIR", ".")).resolve()
client = anthropic.Anthropic()

SYSTEM = (
    "You are a careful engineer working inside a sandboxed repository. "
    "Use tools to inspect files and run commands. Stop and answer when the task is done."
)

TOOLS = [
    {
        "name": "read_file",
        "description": "Read a UTF-8 text file relative to the working directory.",
        "input_schema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"],
        },
    },
    {
        "name": "run",
        "description": "Run a shell command in the working directory. Returns stdout, stderr and exit code.",
        "input_schema": {
            "type": "object",
            "properties": {"command": {"type": "string"}},
            "required": ["command"],
        },
    },
]


def read_file(path: str) -> str:
    target = (WORKDIR / path).resolve()
    if WORKDIR not in target.parents and target != WORKDIR:
        return "error: path escapes the working directory"
    return target.read_text(encoding="utf-8")


def run(command: str) -> str:
    proc = subprocess.run(
        command, shell=True, cwd=WORKDIR, capture_output=True, text=True, timeout=60
    )
    return json.dumps({"exit": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr})


def execute(name: str, args: dict) -> str:
    try:
        if name == "read_file":
            return read_file(**args)
        if name == "run":
            return run(**args)
        return f"error: unknown tool {name}"
    except Exception as exc:  # model cần NHÌN THẤY lỗi, không phải crash vì lỗi
        return f"error: {type(exc).__name__}: {exc}"


def agent(task: str, max_steps: int = 20, token_budget: int = 300_000) -> str:
    messages = [{"role": "user", "content": task}]
    spent, step = 0, 0

    while step < max_steps:
        step += 1
        response = client.messages.create(
            model=MODEL, max_tokens=4096, system=SYSTEM, tools=TOOLS, messages=messages
        )
        spent += response.usage.input_tokens + response.usage.output_tokens
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return "".join(b.text for b in response.content if b.type == "text")

        results = []
        for block in response.content:
            if block.type == "tool_use":
                print(f"[{step}] {block.name} {json.dumps(block.input)[:120]}")
                output = execute(block.name, block.input)
                results.append(
                    {"type": "tool_result", "tool_use_id": block.id, "content": output[:12_000]}
                )
        messages.append({"role": "user", "content": results})

        if spent > token_budget and step < max_steps:
            messages.append({"role": "user", "content": "Budget exhausted. Summarise what you found and stop."})
            max_steps = step + 1  # đúng một lần gọi model nữa rồi dừng

    raise RuntimeError(f"gave up after {step} steps")


if __name__ == "__main__":
    import sys
    print(agent(" ".join(sys.argv[1:]) or "List the files here and tell me what this project does."))

Chạy thử trong bất kỳ repo nào:

export ANTHROPIC_API_KEY=sk-ant-...
python agent_loop.py "Tìm test chậm nhất trong repo này và giải thích vì sao nó chậm"

Đọc lại hàm agent() một lần nữa, chậm thôi, vì mọi agent production bạn từng đụng vào đều là hàm này cộng thêm trang trí. Model nhận hội thoại và danh sách tool. Nếu nó dừng vì bất kỳ lý do gì khác tool_use, ta trả text của nó về. Ngược lại ta chạy từng tool call trong response, gói output thành các block tool_result gắn với id của call, nối vào thành một lượt user, rồi đi vòng tiếp.

Có hai chi tiết dễ bỏ sót và cả hai đều quan trọng. Một, một response có thể chứa nhiều tool call; bạn phải trả lời tất cả, trong cùng một message user, không thì API từ chối request kế tiếp. Hai, tool result là một chuỗi do bạn kiểm soát. Thứ bạn nhét vào chuỗi đó là điều duy nhất model sẽ biết về chuyện gì đã xảy ra. Đó là cần gạt đầu tiên của chất lượng agent, và phần lớn team chưa bao giờ chạm vào.

Vòng lặp vỡ ở đâu#

Bản đồ chơi phía trên chạy được. Nó cũng chính là phiên bản sẽ làm bạn mất mặt trước khách hàng, vì bốn lý do tôi đã tự nếm.

Nó không biết khi nào dừng. Model vốn lạc quan. Giao một task mơ hồ, nó sẽ gọi tool mãi sau khi việc hữu ích đã xong, hoặc tuyên bố thắng lợi ngay sau kết quả dang dở đầu tiên. max_steps và token_budget trong code không phải thứ có cũng được. Chúng là ranh giới giữa chi phí có trần và một hoá đơn. Thực tế tôi đặt giới hạn bước theo loại task: agent tra cứu được 6, agent code được 40 đến 60, cao hơn nữa thì cắt thành nhiều session (bài harness sẽ nói kỹ).

Lỗi tool biến thành bão retry. Nếu run raise thay vì trả về chuỗi, vòng lặp chết và người dùng thấy stack trace. Nếu nó trả về mỗi chữ "error", model sẽ thử lại cùng lệnh đó ba bốn lần với vài thay đổi vụn vặt. Chỗ sửa nằm ở execute(): bắt hết, và trả về thông điệp lỗi nói cho model biết cần làm khác đi thế nào. "Command timed out after 60s; try a narrower command or add a limit" cho ra bước tiếp theo rất khác so với "error".

Output của tool làm ngập context. Một lệnh cat lên file log 2 MB là phần còn lại của context window bay màu, kèm theo token budget. Cắt output[:12_000] thô thiển nhưng cần thiết. Bản tốt hơn giữ đầu và đuôi, nói rõ đã bỏ bao nhiêu byte. Bản tốt nhất cho model những tool không thể sinh ra output vô hạn ngay từ đầu (grep có limit, read_file theo khoảng dòng).

Bạn không giải thích được nó đã làm gì. Dòng print trong vòng lặp chỉ là thế thân cho thứ bạn thật sự cần: một trace được lưu lại của mọi lần gọi model, mọi tool call và kết quả, kèm timestamp và số token. Khi người dùng hỏi vì sao agent sửa nhầm file, "model quyết định thế" không phải câu trả lời. Trace mới là. Xây nó từ ngày đầu; gắn thêm sau sự cố là một trải nghiệm khổ sở.

Bốn thứ tôi không bao giờ ship thiếu#

Qua vài agent chạy production, bốn thứ này thành bắt buộc. Chúng ứng thẳng với bốn kiểu hỏng ở trên.

Chính sách dừng tường minh, không phải tự phát. Model tự dừng là một điều kiện kết thúc. Các điều kiện còn lại: giới hạn bước, ngân sách token hoặc tiền, giới hạn thời gian thực, và một tín hiệu "xong" mà model phải chủ động yêu cầu (một tool finish với tham số summary hoạt động rất tốt và cho bạn kết quả cuối có cấu trúc thay vì text tự do).

Tool result biết dạy. Mỗi chuỗi lỗi là một prompt. Viết nó cho model: nói sai ở đâu, input hợp lệ là gì, nên thử gì thay thế. Việc này rẻ hơn bất kỳ lượng system-prompt engineering nào, và hiệu quả thấy ngay.

Output có trần cho mỗi tool call. Giới hạn cứng theo byte, cộng thêm tool được thiết kế để trả về tóm tắt hoặc lát cắt thay vì nguyên artifact. Một tool search trả về 20 kết quả đầu kèm số dòng ăn đứt read_file lên một file mà model phải cuộn mãi.

Trace replay được. Log nguyên mảng messages trước mỗi lần gọi model, response thô, từng lần gọi tool và output, token usage và latency. Lưu dạng JSON lines theo run id. Có file đó bạn replay được một lần chạy với phiên bản model mới, xây eval set từ lỗi thật, và trả lời câu "vì sao" mà không phải đoán.

Đây là vòng lặp sau khi có đủ bốn thứ. Vẫn ngắn, và đó là mục đích.

# agent_loop_v2.py — vẫn vòng lặp đó, thêm những kiểm soát sống sót được production
import json, os, time, uuid
from pathlib import Path
import anthropic

MODEL = os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-5-5")
client = anthropic.Anthropic()
TRACE_DIR = Path("traces"); TRACE_DIR.mkdir(exist_ok=True)

FINISH_TOOL = {
    "name": "finish",
    "description": "Call this exactly once when the task is complete. Put the final answer in `summary`.",
    "input_schema": {
        "type": "object",
        "properties": {"summary": {"type": "string"}, "confidence": {"type": "number"}},
        "required": ["summary"],
    },
}


class Budget:
    def __init__(self, steps=30, tokens=400_000, seconds=600):
        self.steps, self.tokens, self.seconds = steps, tokens, seconds
        self.used_steps = self.used_tokens = 0
        self.started = time.monotonic()

    def charge(self, usage):
        self.used_steps += 1
        self.used_tokens += usage.input_tokens + usage.output_tokens

    def exceeded(self):
        if self.used_steps >= self.steps: return "step limit"
        if self.used_tokens >= self.tokens: return "token budget"
        if time.monotonic() - self.started >= self.seconds: return "time limit"
        return None


def clip(text: str, limit: int = 12_000) -> str:
    if len(text) <= limit: return text
    head, tail = text[: limit // 2], text[-limit // 2 :]
    return f"{head}\n... [{len(text) - limit} characters omitted] ...\n{tail}"


def run_agent(task, tools, execute, system, budget=None):
    budget = budget or Budget()
    run_id = uuid.uuid4().hex[:8]
    trace = (TRACE_DIR / f"{run_id}.jsonl").open("a")
    log = lambda kind, **data: trace.write(json.dumps({"t": time.time(), "kind": kind, **data}) + "\n")

    messages = [{"role": "user", "content": task}]
    all_tools = [*tools, FINISH_TOOL]
    grace_turn = False  # True khi hết budget: model được đúng một lượt cuối để gọi `finish`
    log("start", task=task, model=MODEL)

    while True:
        response = client.messages.create(
            model=MODEL, max_tokens=4096, system=system, tools=all_tools, messages=messages
        )
        budget.charge(response.usage)
        log("model", stop=response.stop_reason, usage=response.usage.model_dump(),
            content=[b.model_dump() for b in response.content])
        messages.append({"role": "assistant", "content": response.content})

        calls = [b for b in response.content if b.type == "tool_use"]
        for call in calls:
            if call.name == "finish":
                log("finish", **call.input)
                return {"status": "done", "run_id": run_id, **call.input}

        if grace_turn:
            log("stopped", reason="budget")
            return {"status": "stopped", "reason": "budget exhausted", "run_id": run_id}

        if not calls:
            # Model nói chuyện thay vì kết thúc. Nhắc nó dùng tool finish để ta có kết quả có cấu trúc.
            messages.append({"role": "user", "content": "If the task is complete, call `finish`. Otherwise continue."})
        else:
            results = []
            for call in calls:
                started = time.monotonic()
                output = clip(execute(call.name, call.input))
                log("tool", name=call.name, input=call.input, output=output,
                    ms=int((time.monotonic() - started) * 1000))
                results.append({"type": "tool_result", "tool_use_id": call.id, "content": output})
            messages.append({"role": "user", "content": results})

        reason = budget.exceeded()
        if reason:
            log("budget", reason=reason)
            messages.append({"role": "user", "content": f"Stopping: {reason}. Call `finish` with what you have."})
            grace_turn = True

Tool finish đáng được nói thêm một câu. Để model kết thúc bằng cách đơn giản là không gọi tool nào thì tiện, nhưng bạn chỉ nhận được text không cấu trúc và không có tín hiệu gì về độ tự tin. Biến việc hoàn thành thành một tool call tường minh thì điểm kết thúc của mọi lần chạy trở thành dữ liệu: một trường summary để hiển thị, một con số confidence để định tuyến (dưới 0.6 thì đẩy cho người), và một chỗ sạch sẽ để gắn validation trước khi chấp nhận kết quả.

Vài điều tôi từng tin và giờ không còn tin#

"Context window to hơn nghĩa là khỏi lo output của tool." Model hiện nay nhận một triệu token. Bạn vẫn không nuôi nổi chuyện lấp đầy nó ở mỗi bước, vì mỗi lần gọi bạn trả tiền cho cả cửa sổ, và chất lượng tụt từ rất lâu trước khi chạm giới hạn cứng. Vòng lặp làm chi phí dồn tích: bước 30 đọc lại toàn bộ bước 1 đến 29.

"Gọi tool song song là tối ưu hoá." Nó cũng là bẫy về tính đúng đắn. Nếu model xin write_file và run_tests trong cùng một lượt, bạn phải quyết định thứ tự và liệu cái thứ hai có nhìn thấy cái thứ nhất không. Tôi thực thi tuần tự theo đúng thứ tự model phát ra, trừ khi tool được đánh dấu read-only.

"Không bao giờ cho model thấy lỗi thô." Ngược lại mới đúng. Lọc secret thì có. Nhưng nguyên văn stderr của một build fail chính là thông tin model cần, còn câu thân thiện "something went wrong" thì vô dụng với nó.

"Vòng lặp là phần dễ, prompt mới là phần khó." Đảo lại. Một system prompt tầm tầm với điều kiện dừng tốt, lỗi tool tốt và có trace sẽ thắng một prompt tuyệt đẹp nằm trong vòng lặp chạy mãi không debug nổi.

Đi tiếp từ đây#

Mọi thứ làm agent đáng tin nằm quanh vòng lặp chứ không nằm trong model: tool và quyền của chúng, các file nói cho model biết về dự án của bạn, cách context được cắt tỉa giữa các bước, checkpoint để một lần chạy sống qua crash, và eval để biết một thay đổi có giúp ích hay không. Ngành đã chốt một từ cho cả lớp đó: harness. Đó là bài tiếp theo, và là nơi phần kỹ thuật thật sự nằm.

Nếu chỉ mang về một điều từ bài này: tự viết bản 40 dòng, một lần, với tool của chính bạn, trước khi nhận framework về. Bạn sẽ hiểu từng cái nút framework lộ ra, và biết nên nghi ngờ default nào của nó.

Tài liệu tham khảo#

• Anthropic, Building effective agents (12/2024)
• Yao et al., ReAct: Synergizing Reasoning and Acting in Language Models (2022)
• Anthropic, Tool use with the Messages API