Một AI agent chạy production không thể "tiêu token vô tội vạ" như lúc bạn demo trên local. Nếu không có giới hạn rõ ràng, một agent bị kẹt trong loop reasoning, một prompt bị leak vào context quá dài, hay một user vô tình yêu cầu agent xử lý cả một repository — bất kỳ tình huống nào trong số này cũng có thể khiến bill API tăng vọt trong vài giờ. Bài này đi sâu vào cách thiết kế token budget (ngân sách token) ở ba tầng — per-task, per-session, per-project — và các guardrail (cơ chế chặn/bảo vệ) để ngăn agent vượt ngân sách trước khi hóa đơn cuối tháng khiến bạn giật mình.

Đây không phải là bài lý thuyết. Mọi phần đều có code Python minh họa để bạn có thể áp dụng ngay vào hệ thống agent hiện tại — từ việc tính max_tokens cho một lệnh gọi API, đến việc dựng rate limiter dùng Redis cho toàn bộ project.

Ba Cấp Độ Chi Tiết Của Token Budget (Granularity)

Trước khi viết một dòng code nào, cần phân biệt rõ ba tầng ngân sách, vì mỗi tầng giải quyết một loại rủi ro khác nhau và cần cơ chế enforcement (thực thi giới hạn) khác nhau.

Per-task budget là giới hạn token cho một đơn vị công việc cụ thể — một lệnh gọi tool, một vòng reasoning, hoặc một yêu cầu đơn lẻ của người dùng. Đây là tầng "hạt mịn" nhất, có tác dụng ngăn một task đơn lẻ phình to bất thường (ví dụ agent cố đọc toàn bộ file log 50MB để trả lời một câu hỏi đơn giản).

Per-session budget kiểm soát tổng lượng token tiêu tốn trong một phiên làm việc — tức là toàn bộ cuộc hội thoại từ lúc user mở agent đến lúc đóng lại (hoặc timeout). Session budget quan trọng vì context window (giới hạn ngữ cảnh model có thể xử lý trong một lượt gọi) tích lũy dần: một cuộc hội thoại dài 40 lượt trao đổi có thể tiêu tốn gấp 20 lần một cuộc hội thoại 5 lượt, dù mỗi lượt đều "bình thường".

Per-project budget là tầng cao nhất, áp dụng cho toàn bộ một project, team, hoặc tenant trong một khoảng thời gian (thường là theo ngày hoặc theo tháng). Đây là công cụ để tài chính và engineering leadership kiểm soát chi phí tổng thể, tương tự như hạn mức chi tiêu (spending cap) trên thẻ tín dụng doanh nghiệp.

Ba tầng này không độc lập — chúng lồng vào nhau. Per-project budget là tổng của nhiều per-session budget, và mỗi per-session budget lại là tổng của nhiều per-task budget. Thiết kế đúng nghĩa là bạn phải enforce (thực thi) được ở cả ba tầng đồng thời, vì chỉ chặn ở tầng project thì đã quá muộn — tiền đã bị tiêu hết trước khi bạn kịp phản ứng.

Mẹo

• Luôn thiết kế budget theo thứ tự từ nhỏ đến lớn: per-task trước (phản ứng nhanh nhất, latency thấp nhất), rồi mới per-session và per-project (bảo vệ toàn cục).
• Đừng chỉ theo dõi token — hãy gắn cost (chi phí) thực tế bằng USD vào mỗi budget, vì giá token input/output/cached khác nhau tùy model, và stakeholder không đọc số token, họ đọc con số tiền.
• Ghi log token usage theo từng tầng ngay từ ngày đầu triển khai, dù chưa cần enforcement — dữ liệu lịch sử là nền tảng để đặt ngưỡng budget hợp lý sau này (xem phần Baseline ở dưới).

Thiết Kế Per-Task Token Budget

Bước 1: Baseline các Task của bạn

Trước khi đặt ra một con số giới hạn, bạn cần biết task của mình thường tiêu tốn bao nhiêu token trong điều kiện bình thường. Đặt budget mà không có baseline (đường cơ sở) giống như đặt tốc độ giới hạn trên đường mà chưa từng đo tốc độ xe chạy qua đó — bạn sẽ hoặc chặn nhầm các task hợp lệ, hoặc đặt ngưỡng quá cao để nó vô nghĩa.

Cách làm thực tế: chạy lại tối thiểu 50-100 task điển hình (representative) trong môi trường staging, ghi lại token input, token output, và token cache-read/cache-write cho từng task, rồi tính phân phối (distribution) — không chỉ trung bình (mean) mà cả percentile p50, p90, p99.

import statistics
from dataclasses import dataclass, field

@dataclass
class TaskUsageRecord:
    task_type: str
    input_tokens: int
    output_tokens: int
    cached_tokens: int = 0

def summarize_baseline(records: list[TaskUsageRecord], task_type: str) -> dict:
    """Compute percentile baselines for one task type, used to set budgets."""
    subset = [r for r in records if r.task_type == task_type]
    totals = sorted(r.input_tokens + r.output_tokens for r in subset)

    def percentile(data: list[int], pct: float) -> int:
        if not data:
            return 0
        idx = int(len(data) * pct)
        return data[min(idx, len(data) - 1)]

    return {
        "task_type": task_type,
        "sample_size": len(subset),
        "p50_total_tokens": percentile(totals, 0.50),
        "p90_total_tokens": percentile(totals, 0.90),
        "p99_total_tokens": percentile(totals, 0.99),
        "mean_total_tokens": round(statistics.mean(totals)) if totals else 0,
    }

Quy tắc thực dụng: đặt soft budget (ngưỡng cảnh báo) ở khoảng p90, và hard budget (ngưỡng chặn cứng) ở khoảng p99 × 1.3-1.5. Điều này cho phép các task hợp lệ nhưng phức tạp hơn bình thường vẫn hoàn thành, đồng thời bắt được các trường hợp bất thường thật sự (runaway task).

Bước 2: Triển khai Enforcement cho Budget

Có baseline rồi, bước tiếp theo là code hóa nó thành cấu hình có thể enforce được. Một pattern phổ biến là định nghĩa budget theo loại task, sau đó dùng một decorator hoặc middleware để kiểm tra trước và sau mỗi lệnh gọi LLM (Large Language Model).

from dataclasses import dataclass
from enum import Enum

class BudgetAction(Enum):
    ALLOW = "allow"
    WARN = "warn"
    BLOCK = "block"

@dataclass
class TaskBudgetConfig:
    soft_limit: int   # trigger warning, log, but allow continuation
    hard_limit: int   # trigger BLOCK before the call is even made

TASK_BUDGETS: dict[str, TaskBudgetConfig] = {
    "code_review_single_file": TaskBudgetConfig(soft_limit=6_000, hard_limit=9_000),
    "code_review_pull_request": TaskBudgetConfig(soft_limit=18_000, hard_limit=28_000),
    "customer_support_reply": TaskBudgetConfig(soft_limit=3_000, hard_limit=5_000),
    "test_generation": TaskBudgetConfig(soft_limit=10_000, hard_limit=16_000),
}

class TaskBudgetExceeded(Exception):
    def __init__(self, task_type: str, projected: int, limit: int):
        super().__init__(
            f"Task '{task_type}' projected to use {projected} tokens, "
            f"exceeding hard limit of {limit}."
        )
        self.task_type = task_type
        self.projected = projected
        self.limit = limit

def check_task_budget(
    task_type: str,
    accumulated_tokens: int,
    projected_next_call_tokens: int,
) -> BudgetAction:
    """Call this BEFORE issuing the next LLM call inside a task."""
    config = TASK_BUDGETS.get(task_type)
    if config is None:
        return BudgetAction.ALLOW  # unknown task type: no budget configured yet

    projected_total = accumulated_tokens + projected_next_call_tokens

    if projected_total >= config.hard_limit:
        raise TaskBudgetExceeded(task_type, projected_total, config.hard_limit)
    if projected_total >= config.soft_limit:
        return BudgetAction.WARN
    return BudgetAction.ALLOW

Ví dụ áp dụng trong vòng lặp agent — mỗi lần agent gọi tool hoặc gọi lại LLM, ta cộng dồn token đã dùng và kiểm tra trước khi thực hiện lệnh gọi tiếp theo:

def run_agent_step(task_type: str, agent_state: dict, next_call_estimate: int):
    action = check_task_budget(
        task_type=task_type,
        accumulated_tokens=agent_state["tokens_used"],
        projected_next_call_tokens=next_call_estimate,
    )
    if action == BudgetAction.WARN:
        logger.warning(
            "task=%s approaching soft budget (used=%d, next=%d)",
            task_type, agent_state["tokens_used"], next_call_estimate,
        )
    # BudgetAction.ALLOW or WARN both proceed; hard limit raises before this point.
    return execute_llm_call(agent_state)

Điểm mấu chốt: kiểm tra budget trước khi gọi API, không phải sau. Kiểm tra sau khi tiêu tiền rồi thì chỉ còn ý nghĩa ghi log, không còn ý nghĩa ngăn chặn.

Bước 3: Tham số max_tokens như một Hard Stop

max_tokens là tham số bắt buộc trong hầu hết API của LLM, quy định số token output tối đa mà model được sinh ra. Đây là hard stop (điểm chặn cứng) rẻ nhất và đáng tin cậy nhất bạn có — vì nó được thực thi ngay tại provider, không phụ thuộc vào logic enforcement phía client của bạn.

Công thức tính max_tokens hợp lý dựa trên budget còn lại của task:

def compute_max_tokens(
    task_budget: int,
    input_tokens_used: int,
    safety_margin_tokens: int = 200,
) -> int:
    """
    Derive a hard-stop max_tokens value from the remaining task budget.

    Example: a task with a 3,000-token total budget that has already
    consumed 1,500 input tokens should cap output generation so the
    combined input+output never exceeds the budget.
    """
    remaining = task_budget - input_tokens_used - safety_margin_tokens
    # Never allow a negative or absurdly small ceiling; clamp to a sane floor
    # so the model can still produce a minimally useful response.
    return max(remaining, 256)

max_output = compute_max_tokens(task_budget=3_000, input_tokens_used=1_500)
print(max_output)  # -> 1300

response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=max_output,
    messages=conversation_history,
)

Lưu ý ba điểm quan trọng khi dùng max_tokens như guardrail:

1. max_tokens không tính input token. Nó chỉ giới hạn phần output. Muốn kiểm soát tổng ngân sách (input + output), bạn phải tự trừ input đã dùng ra khỏi budget, như trong hàm compute_max_tokens trên.
2. Model có thể dừng sớm hơn max_tokens nếu nó tự nhận thấy đã hoàn thành câu trả lời (stop_reason: "end_turn") — đây là hành vi bình thường, không phải lỗi.
3. Khi model bị cắt giữa câu do đạt max_tokens (stop_reason: "max_tokens"), output có thể không hoàn chỉnh về cấu trúc (ví dụ JSON bị cắt dở). Luôn kiểm tra stop_reason trong response và xử lý riêng trường hợp này — đừng parse thẳng output như thể nó luôn hoàn chỉnh.

response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=max_output,
    messages=conversation_history,
)

if response.stop_reason == "max_tokens":
    logger.warning("Response truncated by max_tokens hard stop; task=%s", task_type)
    # Decide: retry with a higher (but still budget-bounded) ceiling,
    # or return a partial/degraded response to the caller.

Mẹo

• Luôn để lại một safety_margin_tokens (thường 100-300 token) khi tính max_tokens — model không tự dừng chính xác ở ranh giới budget, nên cần đệm an toàn.
• Với các task cần output có cấu trúc (JSON, code block), đặt max_tokens dư dả hơn baseline p99 một chút, vì bị cắt giữa cấu trúc gây lỗi parse tốn công xử lý hơn là vượt budget một ít.
• Log stop_reason cho mọi lệnh gọi — đây là tín hiệu rẻ nhất để phát hiện budget đang quá chật cho loại task đó.

Thiết Kế Per-Session Token Budget

Kiến trúc Session Budget

Session budget khác per-task budget ở một điểm cốt lõi: nó phải theo dõi trạng thái tích lũy xuyên suốt nhiều lượt gọi, thường được lưu trong một session store (Redis, database, hoặc in-memory cache tùy quy mô hệ thống). Thiết kế phổ biến là một SessionBudgetTracker gắn với session ID, cập nhật sau mỗi lượt tương tác.

import time
from dataclasses import dataclass, field

@dataclass
class SessionBudget:
    session_id: str
    max_tokens_total: int
    warn_threshold_ratio: float = 0.75
    tokens_used: int = 0
    turn_count: int = 0
    started_at: float = field(default_factory=time.time)

    @property
    def remaining_tokens(self) -> int:
        return max(self.max_tokens_total - self.tokens_used, 0)

    @property
    def usage_ratio(self) -> float:
        return self.tokens_used / self.max_tokens_total

    def is_over_warn_threshold(self) -> bool:
        return self.usage_ratio >= self.warn_threshold_ratio

    def is_exhausted(self) -> bool:
        return self.tokens_used >= self.max_tokens_total


class SessionBudgetStore:
    """Minimal in-memory store; swap for Redis in multi-instance deployments."""

    def __init__(self):
        self._sessions: dict[str, SessionBudget] = {}

    def get_or_create(self, session_id: str, default_budget: int) -> SessionBudget:
        if session_id not in self._sessions:
            self._sessions[session_id] = SessionBudget(
                session_id=session_id, max_tokens_total=default_budget
            )
        return self._sessions[session_id]

    def record_usage(self, session_id: str, input_tokens: int, output_tokens: int):
        session = self._sessions[session_id]
        session.tokens_used += input_tokens + output_tokens
        session.turn_count += 1

Với hệ thống chạy nhiều instance (multi-instance), in-memory store không dùng được vì mỗi instance có state riêng — bạn cần một store tập trung (Redis là lựa chọn phổ biến nhất, sẽ nói kỹ ở phần per-project). Nguyên tắc thiết kế giữ nguyên, chỉ khác cách persist.

Khi session chạm is_exhausted(), hành vi hệ thống thường là một trong ba: (1) từ chối lượt tương tác tiếp theo và yêu cầu user mở session mới, (2) tự động chuyển sang model rẻ hơn cho phần còn lại của session, hoặc (3) buộc phải compress context trước khi tiếp tục — đây là kỹ thuật ở phần tiếp theo.

Compression Context ở các Ngưỡng Budget

Khi session tiến gần đến giới hạn, một trong những cách hiệu quả nhất để "câu giờ" thêm mà không cắt ngang trải nghiệm người dùng là compress context (nén ngữ cảnh) — tóm tắt lại phần lịch sử hội thoại cũ để giải phóng chỗ cho token mới, thay vì để context window tự nhiên đầy lên và bị cắt cứng.

Chiến lược phổ biến: đặt nhiều ngưỡng (threshold) ứng với các hành động khác nhau, càng gần giới hạn thì càng nén mạnh.

from enum import Enum

class CompressionLevel(Enum):
    NONE = "none"
    LIGHT = "light"       # summarize turns older than the last 10
    AGGRESSIVE = "aggressive"  # summarize everything except the last 3 turns
    EMERGENCY = "emergency"    # keep only system prompt + last user turn

def choose_compression_level(usage_ratio: float) -> CompressionLevel:
    if usage_ratio < 0.60:
        return CompressionLevel.NONE
    if usage_ratio < 0.80:
        return CompressionLevel.LIGHT
    if usage_ratio < 0.95:
        return CompressionLevel.AGGRESSIVE
    return CompressionLevel.EMERGENCY


def compress_conversation(history: list[dict], level: CompressionLevel, summarizer_fn) -> list[dict]:
    """
    Reduce token footprint of conversation history based on compression level.
    summarizer_fn: a cheap/fast LLM call (or extractive heuristic) that condenses
    a list of turns into a short summary turn.
    """
    if level == CompressionLevel.NONE:
        return history

    if level == CompressionLevel.LIGHT:
        keep_recent, to_summarize = history[-10:], history[:-10]
    elif level == CompressionLevel.AGGRESSIVE:
        keep_recent, to_summarize = history[-3:], history[:-3]
    else:  # EMERGENCY
        keep_recent, to_summarize = history[-1:], history[:-1]

    if not to_summarize:
        return history

    summary_text = summarizer_fn(to_summarize)
    summary_turn = {
        "role": "system",
        "content": f"[Compressed summary of {len(to_summarize)} earlier turns]: {summary_text}",
    }
    return [summary_turn] + keep_recent

Tích hợp vào vòng lặp session:

def prepare_next_turn(session: SessionBudget, history: list[dict], summarizer_fn) -> list[dict]:
    level = choose_compression_level(session.usage_ratio)
    if level != CompressionLevel.NONE:
        logger.info(
            "session=%s usage_ratio=%.2f -> applying %s compression",
            session.session_id, session.usage_ratio, level.value,
        )
    return compress_conversation(history, level, summarizer_fn)

Một lưu ý quan trọng: compression tự nó tiêu token (bạn phải gọi một LLM để tóm tắt, hoặc chạy một thuật toán extractive). Vì vậy compression chỉ có lợi khi phần token tiết kiệm được ở các lượt sau lớn hơn chi phí tóm tắt — với các session ngắn (dưới 15-20 lượt), thường không đáng để compress; với session dài hàng chục đến hàng trăm lượt, compression gần như là bắt buộc.

Mẹo

• Đừng compress toàn bộ lịch sử một cách "phẳng" — luôn giữ nguyên system prompt và vài lượt gần nhất, vì đó là phần model cần chính xác nhất để duy trì mạch hội thoại.
• Cache lại summary đã tạo — nếu session tiếp tục kéo dài, bạn không cần tóm tắt lại từ đầu, chỉ cần tóm tắt phần mới thêm vào summary cũ.
• Thông báo cho người dùng (qua UI) khi context bị compress, đặc biệt ở mức AGGRESSIVE/EMERGENCY — người dùng có quyền biết agent "đã quên" một phần chi tiết cũ.

Thiết Kế Per-Project Token Budget

Mô hình Phân bổ Budget

Ở tầng project, câu hỏi không còn là "task này tốn bao nhiêu" mà là "tổ chức sẵn sàng chi bao nhiêu cho agent này trong một khoảng thời gian, và phân chia thế nào giữa các team/feature/user". Mô hình phân bổ (allocation model) phổ biến nhất trong thực tế là budget theo thời gian + theo đơn vị tổ chức, ví dụ: 50 triệu token/ngày cho toàn project, trong đó tối đa 60% dành cho tính năng production-facing, 30% cho batch job nội bộ, 10% dự phòng (buffer) cho spike bất thường.

from dataclasses import dataclass

@dataclass
class ProjectBudgetAllocation:
    project_id: str
    daily_token_cap: int
    allocation_ratios: dict[str, float]  # e.g. {"production": 0.6, "batch": 0.3, "buffer": 0.1}

    def cap_for_category(self, category: str) -> int:
        ratio = self.allocation_ratios.get(category, 0.0)
        return int(self.daily_token_cap * ratio)


allocation = ProjectBudgetAllocation(
    project_id="support-agent-v2",
    daily_token_cap=50_000_000,
    allocation_ratios={"production": 0.6, "batch": 0.3, "buffer": 0.1},
)

print(allocation.cap_for_category("production"))  # -> 30,000,000

Mô hình này giúp trả lời trực tiếp câu hỏi mà product manager hoặc finance thường hỏi: "Nếu feature X đột ngột viral, nó có thể ăn hết budget của feature Y không?" — câu trả lời phải là không, nhờ có ratio phân bổ theo category.

Enforcement Budget với Rate Limiting dùng Redis

Ở quy mô project (nhiều request/giây, nhiều instance chạy song song), bạn cần một cơ chế đếm và chặn tập trung, có khả năng chịu tải cao và có độ trễ thấp. Redis là lựa chọn phổ biến nhất trong ngành cho use case này, nhờ các lệnh atomic (nguyên tử) như INCRBY và khả năng đặt TTL (time-to-live) để tự động reset counter theo ngày.

import redis
import time
from datetime import datetime, timezone

class RedisTokenBudgetLimiter:
    """
    Token-based rate limiter backed by Redis, enforcing a daily cap
    per project/category. Uses atomic INCRBY to stay correct under
    concurrent access from multiple app instances.
    """

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def _key(self, project_id: str, category: str) -> str:
        day = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        return f"token_budget:{project_id}:{category}:{day}"

    def try_consume(self, project_id: str, category: str, tokens: int, cap: int) -> bool:
        """
        Atomically increment usage and check against the cap.
        Returns True if the consumption is allowed, False if it would
        exceed the cap (in which case the increment is rolled back).
        """
        key = self._key(project_id, category)
        pipe = self.redis.pipeline()
        pipe.incrby(key, tokens)
        pipe.ttl(key)
        new_total, ttl = pipe.execute()

        # First write for the day: set expiry so the counter resets automatically.
        if ttl == -1:
            self.redis.expire(key, 60 * 60 * 26)  # 26h buffer past midnight UTC

        if new_total > cap:
            # Roll back the increment so a rejected call doesn't still count.
            self.redis.decrby(key, tokens)
            return False
        return True

    def current_usage(self, project_id: str, category: str) -> int:
        value = self.redis.get(self._key(project_id, category))
        return int(value) if value else 0


limiter = RedisTokenBudgetLimiter(redis.Redis(host="localhost", port=6379, db=0))

estimated_tokens = 4_200
allowed = limiter.try_consume(
    project_id="support-agent-v2",
    category="production",
    tokens=estimated_tokens,
    cap=allocation.cap_for_category("production"),
)

if not allowed:
    raise RuntimeError("Daily token budget for category 'production' exceeded.")

Vài chi tiết triển khai quan trọng cần lưu ý:

• Ước lượng trước, đối soát sau (estimate-then-reconcile). Bạn chưa biết chính xác output sẽ dài bao nhiêu token trước khi gọi API, nên try_consume nên chạy với số ước lượng (dựa trên max_tokens đã set, hoặc trung bình lịch sử), sau đó gọi lại một lần điều chỉnh (adjust) bằng số thực tế lấy từ response (usage.input_tokens, usage.output_tokens).
• Rollback khi từ chối. Nếu không rollback phần đã INCRBY, một request bị chặn vẫn "chiếm chỗ" trong budget — đây là lỗi rất dễ mắc phải khi code enforcement.
• Chọn TTL dư ra một chút (ví dụ 26 giờ thay vì đúng 24 giờ) để tránh race condition ở đúng thời điểm chuyển ngày làm mất dữ liệu counter đang dùng.

def adjust_after_call(limiter, project_id, category, estimated, actual):
    """Reconcile the estimate with actual usage reported by the API response."""
    diff = actual - estimated
    if diff != 0:
        key = limiter._key(project_id, category)
        limiter.redis.incrby(key, diff)

Mẹo

• Với hệ thống multi-region, cân nhắc dùng Redis Cluster hoặc một instance Redis tập trung riêng cho budget counter — đừng để độ trễ cross-region ảnh hưởng đến logic enforcement của mọi request.
• Luôn có một endpoint/dashboard đọc current_usage() theo thời gian thực — team vận hành cần thấy được "còn bao nhiêu ngân sách" mà không phải đọi log.
• Đặt cảnh báo (alert) ở 80% và 95% của mỗi cap, không chỉ chặn ở 100% — cảnh báo sớm cho phép con người can thiệp trước khi tính năng bị tự động khóa.

Các Pattern Guardrail cho Lỗi Thường Gặp

Phát hiện Vòng lặp Vô hạn (Infinite Loop)

Một trong những nguyên nhân "đốt token" tệ nhất không phải là task lớn, mà là agent bị kẹt trong một vòng lặp — gọi lại cùng một tool với tham số gần giống nhau, hoặc lặp lại cùng một suy luận (reasoning) mà không tiến triển. Đây là lỗi rất khó phát hiện bằng mắt vì mỗi lượt gọi riêng lẻ nhìn "hợp lý".

Cách phát hiện thực dụng: theo dõi chữ ký (signature) của các lệnh gọi tool gần nhất, và chặn khi phát hiện lặp lại quá nhiều lần trong một khoảng thời gian ngắn.

import hashlib
import json
from collections import deque

class LoopDetector:
    """
    Detects repetitive tool-call patterns within a sliding window,
    a common symptom of an agent stuck reasoning in circles.
    """

    def __init__(self, window_size: int = 8, max_repeats: int = 3):
        self.window_size = window_size
        self.max_repeats = max_repeats
        self.recent_signatures: deque[str] = deque(maxlen=window_size)

    @staticmethod
    def _signature(tool_name: str, tool_args: dict) -> str:
        normalized = json.dumps(tool_args, sort_keys=True)
        raw = f"{tool_name}:{normalized}"
        return hashlib.sha256(raw.encode()).hexdigest()

    def record_and_check(self, tool_name: str, tool_args: dict) -> bool:
        """Returns True if a loop is detected, False otherwise."""
        sig = self._signature(tool_name, tool_args)
        self.recent_signatures.append(sig)
        count = list(self.recent_signatures).count(sig)
        return count >= self.max_repeats


class NoProgressDetector:
    """
    Complements LoopDetector: flags cases where tool calls differ slightly
    but the agent's own output text is not meaningfully changing.
    """

    def __init__(self, similarity_threshold: float = 0.92, window_size: int = 5):
        self.similarity_threshold = similarity_threshold
        self.recent_outputs: deque[str] = deque(maxlen=window_size)

    @staticmethod
    def _jaccard_similarity(a: str, b: str) -> float:
        set_a, set_b = set(a.split()), set(b.split())
        if not set_a or not set_b:
            return 0.0
        return len(set_a & set_b) / len(set_a | set_b)

    def record_and_check(self, output_text: str) -> bool:
        for previous in self.recent_outputs:
            if self._jaccard_similarity(previous, output_text) >= self.similarity_threshold:
                self.recent_outputs.append(output_text)
                return True
        self.recent_outputs.append(output_text)
        return False

Tích hợp cả hai vào agent loop:

loop_detector = LoopDetector(window_size=8, max_repeats=3)
progress_detector = NoProgressDetector()

def agent_step_with_loop_guard(tool_name, tool_args, output_text):
    if loop_detector.record_and_check(tool_name, tool_args):
        raise RuntimeError(
            f"Loop detected: tool '{tool_name}' called with identical args "
            f"{loop_detector.max_repeats}+ times in the last "
            f"{loop_detector.window_size} steps. Aborting task."
        )
    if progress_detector.record_and_check(output_text):
        raise RuntimeError("No-progress detected: agent output is not changing meaningfully.")

Ngoài detection dựa trên pattern, luôn kết hợp thêm một giới hạn số bước tuyệt đối (absolute step cap) — ví dụ tối đa 25 bước cho một task — như lưới an toàn cuối cùng, vì không có detector nào bắt được 100% các dạng loop (đặc biệt là loop "trôi dạt" chậm, mỗi bước hơi khác nhau nhưng vẫn không tiến triển thật).

Guardrail cho Độ dài Output

Ngoài loop, một nguồn "đốt token" phổ biến khác là output quá dài so với nhu cầu thực tế — model liệt kê chi tiết dư thừa, lặp lại thông tin, hoặc sinh ra nội dung boilerplate không cần thiết. max_tokens chặn được trường hợp cực đoan, nhưng không giải quyết được trường hợp output "hợp lệ về hình thức" nhưng dài không cần thiết.

@dataclass
class OutputLengthPolicy:
    task_type: str
    expected_max_chars: int
    hard_max_chars: int

OUTPUT_POLICIES: dict[str, OutputLengthPolicy] = {
    "customer_support_reply": OutputLengthPolicy(
        task_type="customer_support_reply", expected_max_chars=800, hard_max_chars=2_000
    ),
    "code_review_comment": OutputLengthPolicy(
        task_type="code_review_comment", expected_max_chars=1_500, hard_max_chars=4_000
    ),
}

def enforce_output_length(task_type: str, output_text: str) -> str:
    policy = OUTPUT_POLICIES.get(task_type)
    if policy is None:
        return output_text

    length = len(output_text)
    if length > policy.hard_max_chars:
        logger.error(
            "task=%s output length=%d exceeds hard_max_chars=%d; truncating.",
            task_type, length, policy.hard_max_chars,
        )
        return output_text[: policy.hard_max_chars] + "\n\n[Output truncated: exceeded length guardrail]"

    if length > policy.expected_max_chars:
        logger.warning(
            "task=%s output length=%d exceeds expected_max_chars=%d.",
            task_type, length, policy.expected_max_chars,
        )
    return output_text

Truncate cứng ở tầng ứng dụng (application layer) là biện pháp cuối, không phải biện pháp chính — vì nó có thể cắt output ở vị trí không hợp lý (giữa một khối JSON, giữa một câu). Biện pháp chính vẫn nên là định hướng model qua prompt — ví dụ dùng system prompt kiểu "Trả lời trong tối đa 3 đoạn, không lặp lại thông tin đã có trong câu hỏi" — kết hợp với max_tokens làm lưới chặn thứ hai, và enforce_output_length làm lưới chặn cuối cùng.

Một pattern nâng cao hơn là guardrail thích ứng (adaptive): nếu bạn phát hiện một loại task liên tục vượt expected_max_chars, đó là tín hiệu cho thấy prompt hoặc baseline cần điều chỉnh, không phải chỉ đơn thuần "chặn output dài hơn" — hãy đưa dữ liệu này về đội ngũ prompt engineering để cải thiện gốc rễ.

Mẹo

• Đặt hard_max_chars gấp khoảng 2.5-3 lần expected_max_chars — đủ rộng để không chặn nhầm các câu trả lời hợp lệ nhưng phức tạp, đủ hẹp để bắt được các trường hợp bất thường rõ ràng.
• Với loop detection, luôn log lại toàn bộ lịch sử tool call dẫn đến việc phát hiện loop — dữ liệu này cực kỳ quý để debug lại lỗi logic trong agent, không chỉ để chặn chi phí.
• Kết hợp guardrail độ dài output với guardrail loop — nhiều trường hợp loop vô hạn thực chất được ngụy trang dưới dạng "agent liên tục sinh output dài dần" chứ không hẳn là gọi lại đúng tool với đúng tham số.

Trực quan hóa và Báo cáo Budget

Enforcement chỉ giải quyết được nửa vấn đề — nửa còn lại là làm sao con người (engineering lead, product manager, finance) có thể nhìn thấy tình trạng budget mà không phải đọc log thô. Một dashboard budget tốt nên trả lời được ba câu hỏi cơ bản: đang dùng bao nhiêu so với giới hạn, xu hướng (trend) đang tăng hay giảm, và ai/cái gì đang tiêu tốn nhiều nhất.

Cấu trúc dữ liệu tối thiểu cho báo cáo, tổng hợp từ ba tầng budget đã nói ở trên:

from dataclasses import dataclass

@dataclass
class BudgetReportRow:
    scope: str          # "task" | "session" | "project"
    identifier: str     # task_type, session_id, or project_id
    tokens_used: int
    tokens_limit: int
    estimated_cost_usd: float

    @property
    def usage_pct(self) -> float:
        return round(100 * self.tokens_used / self.tokens_limit, 1) if self.tokens_limit else 0.0

def build_daily_report(rows: list[BudgetReportRow]) -> list[dict]:
    return [
        {
            "scope": r.scope,
            "identifier": r.identifier,
            "usage_pct": r.usage_pct,
            "tokens_used": r.tokens_used,
            "cost_usd": round(r.estimated_cost_usd, 2),
            "status": "critical" if r.usage_pct >= 95 else "warning" if r.usage_pct >= 75 else "ok",
        }
        for r in rows
    ]

Đưa estimated_cost_usd vào ngay từ dataclass, không chỉ token thô — vì trong buổi họp với finance hoặc leadership không kỹ thuật, "đã dùng 82% ngân sách 12 triệu đô mỗi tháng" có tác dụng thuyết phục hơn hẳn "đã dùng 41 triệu trong 50 triệu token".

Về mặt trực quan hóa, ba loại biểu đồ hữu ích nhất trong thực tế vận hành:

• Biểu đồ cột theo category (production/batch/buffer) — trả lời câu hỏi "ai đang tiêu nhiều nhất".
• Biểu đồ đường theo thời gian (7-30 ngày) — trả lời câu hỏi "xu hướng đang đi lên hay ổn định".
• Gauge/progress bar cho từng session/project đang active — trả lời câu hỏi "cái nào sắp chạm giới hạn ngay bây giờ".

Báo cáo nên được đẩy tự động (qua Slack, email, hoặc dashboard nội bộ) theo lịch cố định — hàng ngày cho tầng project, theo thời gian thực (real-time) cho tầng session khi usage vượt 75%. Đừng chờ đến cuối tháng mới nhìn vào bill — đó là lúc đã quá muộn để điều chỉnh.

Mẹo

• Luôn hiển thị cả số tuyệt đối và tỉ lệ phần trăm — 75% của một budget nhỏ có thể ít đáng lo hơn 40% của một budget lớn về số tiền thực tế.
• Gắn báo cáo với alerting (cảnh báo tự động qua Slack/email/PagerDuty) ở các ngưỡng 75%/90%/100%, đừng chỉ dựa vào việc có người mở dashboard ra xem.
• Giữ lại dữ liệu lịch sử tối thiểu 90 ngày — đây là nguồn dữ liệu để tái tính baseline định kỳ (budget hợp lý hôm nay có thể sai sau 3-6 tháng khi traffic hoặc loại task thay đổi).

Tổng kết

Token budget không phải là một con số cố định viết một lần rồi để đó — nó là một hệ thống ba tầng (per-task, per-session, per-project) liên tục được baseline, enforce, và điều chỉnh dựa trên dữ liệu thực tế. Per-task budget dùng baseline theo percentile và max_tokens như hard stop để chặn ngay tại từng lệnh gọi. Per-session budget cần theo dõi trạng thái tích lũy và dùng context compression để kéo dài session mà không cắt ngang trải nghiệm. Per-project budget cần một cơ chế enforcement tập trung, chịu được tải cao — Redis với atomic increment là lựa chọn kiểm chứng trong thực tế — cùng một mô hình phân bổ rõ ràng theo category.

Song song với ba tầng budget, guardrail cho các lỗi vận hành phổ biến — loop vô hạn, output quá dài — là lớp bảo vệ bổ sung không thể thiếu, vì không phải mọi lỗi tốn token đều xuất phát từ việc thiếu budget; nhiều trường hợp xuất phát từ lỗi logic của agent mà budget đơn thuần chỉ giới hạn thiệt hại, không ngăn được nguyên nhân gốc. Cuối cùng, mọi cơ chế enforcement chỉ phát huy hết giá trị khi đi kèm reporting và visualization rõ ràng — biến số token trừu tượng thành thông tin mà cả kỹ sư và người ra quyết định kinh doanh đều hiểu và hành động được.