Bạn không thể tối ưu thứ mà bạn không đo được. Trước khi nghĩ đến việc cắt giảm token, việc đầu tiên bắt buộc phải làm là dựng lên một hệ thống đo lường đáng tin cậy — để mọi con số "giảm 30% chi phí" sau này đều có căn cứ, không phải cảm tính.

Rất nhiều team bắt tay vào "tối ưu token" theo cách cảm tính: thấy bill OpenAI/Anthropic tháng này cao hơn tháng trước, đoán rằng agent đang gọi quá nhiều tool, rồi sửa vài chỗ "nhìn có vẻ tốn token" và hy vọng nó có tác dụng. Cách làm này gần như luôn thất bại, vì token footprint (lượng token tiêu thụ) của một hệ thống agentic bị chi phối bởi rất nhiều yếu tố chồng chéo — độ dài system prompt, số vòng lặp reasoning, kích thước tool output, chiến lược retry — mà bạn không thể tách bạch nếu không có dữ liệu đo lường theo từng chiều. Bài này xây dựng cho bạn một bộ công cụ đo lường hoàn chỉnh, từ mức đơn giản nhất (đọc số liệu provider có sẵn) đến mức nâng cao (dashboard tự xây, benchmark suite, alerting production) — để bất kỳ ai trong team, dù là engineer, QA hay product manager, đều có thể tự tin nói "trước khi tối ưu chúng ta tốn X token/tác vụ, sau khi tối ưu còn Y token/tác vụ".

Token Measurement Stack — Cần Gì và Tại Sao

Trước khi đi vào công cụ cụ thể, cần hiểu rõ "measurement stack" (chồng công cụ đo lường) gồm những tầng nào, vì mỗi tầng trả lời một câu hỏi khác nhau và không tầng nào thay thế được tầng khác.

Tầng thứ nhất là provider-native reporting — số liệu token mà chính nhà cung cấp LLM (OpenAI, Anthropic, Google) trả về trong mỗi response, hoặc hiển thị trên dashboard billing của họ. Tầng này cho bạn biết "tổng cộng đã tốn bao nhiêu token" nhưng thường không biết "token đó tốn cho việc gì" — tức là thiếu ngữ cảnh về workflow, task, hay user cụ thể.

Tầng thứ hai là observability platform chuyên cho agent, ví dụ LangSmith, Langfuse, Helicone. Các nền tảng này chặn (intercept) mọi lời gọi LLM trong ứng dụng của bạn, tự động gắn token count vào từng "run" (một lần thực thi agent hoặc một bước trong chain), và cho phép bạn truy vấn, nhóm, so sánh theo nhiều chiều: theo loại task, theo user, theo phiên bản prompt. Đây là tầng mang lại giá trị lớn nhất cho việc chẩn đoán, vì nó nối được token count với ngữ cảnh nghiệp vụ.

Tầng thứ ba là dashboard tự xây — khi bạn cần lưu trữ lâu dài, tùy biến báo cáo theo cách riêng của tổ chức, hoặc khi ngân sách không cho phép trả phí một observability platform bên thứ ba. Một dashboard SQLite + Python đơn giản đã đủ dùng cho phần lớn team ở quy mô vừa và nhỏ.

Tầng thứ tư là baseline benchmark — một bộ tác vụ chuẩn hóa, chạy lặp lại được, cho phép bạn so sánh "trước và sau" khi thay đổi prompt, model, hay kiến trúc agent. Không có benchmark, mọi tuyên bố "tối ưu giúp giảm token" đều là suy đoán vì bạn không kiểm soát được các biến nhiễu (input khác nhau giữa các lần đo).

Tầng cuối cùng là real-time monitoring cho production — vì hành vi của agent trong môi trường thật (input đa dạng, edge case, người dùng thật) khác hẳn với môi trường test, và bạn cần phát hiện bất thường (ví dụ agent bỗng lặp vòng vô hạn, tốn gấp 10 lần token bình thường) trong vài phút thay vì phát hiện qua bill cuối tháng.

Một sai lầm phổ biến là chỉ dừng ở tầng một (nhìn dashboard billing của provider) và cho rằng thế là đủ. Trên thực tế, dashboard billing chỉ cho bạn con số tổng — không giúp bạn trả lời được câu hỏi quan trọng nhất: "token nào đang bị lãng phí, ở bước nào, trong workflow nào?" Để trả lời câu hỏi đó, bạn cần tối thiểu tầng hai hoặc tầng ba.

Mẹo

Đừng cố xây cả 5 tầng ngay từ đầu. Với team mới bắt đầu tối ưu token, thứ tự ưu tiên hợp lý là: tầng một (miễn phí, có sẵn) → tầng bốn (benchmark, giúp bạn có "before picture" trước khi sửa gì cả) → tầng hai hoặc ba (chọn một, tùy ngân sách) → tầng năm (chỉ cần khi đã lên production ổn định).

Provider-Native Token Reporting — Bắt Đầu Từ Những Gì Bạn Đang Có

Mọi API của LLM provider lớn (OpenAI, Anthropic, Google Gemini) đều trả về số liệu sử dụng token ngay trong response — không cần tích hợp thêm gì cả. Với Anthropic API, mỗi response có field usage chứa input_tokens và output_tokens; với OpenAI, field tương tự nằm trong usage với prompt_tokens, completion_tokens, và total_tokens. Đây là nguồn dữ liệu rẻ nhất và chính xác nhất — vì nó là con số mà provider dùng để tính tiền bạn, không phải con số ước lượng qua tokenizer riêng của bạn (mà có thể sai lệch vài phần trăm so với tokenizer thật của model).

Vấn đề của tầng này là nó "mù ngữ cảnh" — nếu bạn chỉ log usage.input_tokens và usage.output_tokens vào một file text, bạn sẽ có một chuỗi số không biết gắn với workflow nào, task nào, người dùng nào. Giải pháp là metadata tagging (gắn thẻ metadata) ngay tại điểm gọi API, trước khi con số đó biến mất.

Dưới đây là một pattern thực tế để wrap lời gọi API, tự động ghi log token usage kèm metadata đủ để phân tích sau này:

import time
import json
import logging
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from anthropic import Anthropic

logger = logging.getLogger("token_usage")

client = Anthropic()


@dataclass
class TokenUsageRecord:
    timestamp: str
    workflow: str          # e.g. "customer_support_triage"
    task_type: str         # e.g. "classify_ticket"
    agent_step: str        # e.g. "planner", "tool_call", "final_answer"
    model: str
    input_tokens: int
    output_tokens: int
    cache_creation_tokens: int
    cache_read_tokens: int
    latency_ms: int
    request_id: str
    user_id: str | None = None
    session_id: str | None = None


def call_llm_with_tracking(
    *,
    workflow: str,
    task_type: str,
    agent_step: str,
    messages: list[dict],
    model: str = "claude-sonnet-5",
    max_tokens: int = 1024,
    system: str | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
) -> tuple[str, TokenUsageRecord]:
    """Wrap the Anthropic API call and emit a structured usage record.

    Every call in the codebase should go through this function instead
    of calling client.messages.create() directly — that's the only way
    to guarantee every token spend is attributable to a workflow.
    """
    started_at = time.monotonic()

    response = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        system=system,
        messages=messages,
    )

    latency_ms = int((time.monotonic() - started_at) * 1000)
    usage = response.usage

    record = TokenUsageRecord(
        timestamp=datetime.now(timezone.utc).isoformat(),
        workflow=workflow,
        task_type=task_type,
        agent_step=agent_step,
        model=model,
        input_tokens=usage.input_tokens,
        output_tokens=usage.output_tokens,
        cache_creation_tokens=getattr(usage, "cache_creation_input_tokens", 0) or 0,
        cache_read_tokens=getattr(usage, "cache_read_input_tokens", 0) or 0,
        latency_ms=latency_ms,
        request_id=response.id,
        user_id=user_id,
        session_id=session_id,
    )

    # Structured logging — one JSON line per call, easy to ship to
    # any log aggregator (CloudWatch, Datadog, ELK) later.
    logger.info(json.dumps(asdict(record)))

    return response.content[0].text, record

Điểm quan trọng nhất trong đoạn code trên không phải là việc gọi API — mà là ba trường workflow, task_type, và agent_step. Đây chính là "khóa phân tích" (analysis key) cho phép bạn sau này trả lời câu hỏi kiểu "workflow customer_support_triage tốn trung bình bao nhiêu token mỗi lần chạy?" hoặc "bước tool_call chiếm bao nhiêu phần trăm tổng token của agent?". Nếu không gắn ba trường này ngay từ đầu, bạn sẽ phải suy luận ngược từ log thô — vừa mất thời gian vừa dễ sai.

Một điểm dễ bị bỏ sót: hầu hết provider hiện đại (Anthropic prompt caching, OpenAI cache) đều tách riêng cache_creation_tokens và cache_read_tokens khỏi input_tokens thông thường. Nếu bạn không log riêng hai trường này, bạn sẽ đánh giá sai hiệu quả của caching — vì token đọc từ cache có giá rẻ hơn nhiều (thường 10% giá input token thường) nhưng vẫn được provider report là một dạng "input token" trong một số API cũ.

Mẹo

Log token usage dưới dạng JSON structured (mỗi field tách biệt), không phải text tự do. Ngay từ ngày đầu, hãy nhất quán chọn field name và định nghĩa thống nhất cho workflow/task_type/agent_step trong toàn bộ codebase — nếu 3 team khác nhau tự đặt tên field khác nhau, việc tổng hợp báo cáo sau này sẽ tốn rất nhiều công sức "dọn dữ liệu" trước khi phân tích được.

LangSmith — Nền Tảng Observability Chuyên Cho Agent

Khi hệ thống agentic của bạn có nhiều bước (multi-step), nhiều lần gọi tool, hoặc nhiều agent con phối hợp với nhau, việc tự log từng lời gọi API riêng lẻ (như cách ở phần trên) sẽ dần trở nên cồng kềnh. Đây là lúc một observability platform chuyên biệt như LangSmith (của LangChain) phát huy giá trị: nó tự động trace toàn bộ cây thực thi của agent — từ lời gọi LLM đầu tiên, qua các bước tool call, cho đến câu trả lời cuối — và gắn token count vào từng node trong cây đó mà bạn không cần tự viết code log.

Bước đầu tiên là thiết lập biến môi trường (environment variable) để LangChain tự động gửi trace lên LangSmith:

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="ls__your_api_key_here"
export LANGCHAIN_PROJECT="agent-token-optimization"
export LANGCHAIN_ENDPOINT="https://api.smith.langchain.com"

Điều đáng chú ý là sau khi thiết lập bốn biến môi trường trên, mọi lời gọi LangChain trong ứng dụng sẽ được trace tự động kèm dữ liệu token — không cần sửa một dòng code nghiệp vụ nào. Đây là khác biệt lớn nhất so với cách log tay ở phần trước: bạn không phải wrap từng lời gọi API, LangSmith tự chặn (instrument) toàn bộ chain, agent, và tool call thông qua callback system có sẵn của LangChain.

Sau khi đã có dữ liệu trace chạy trong vài ngày, bạn có thể truy vấn lại qua LangSmith SDK để phân tích. Ví dụ, lấy toàn bộ run trong 7 ngày gần nhất:

from datetime import datetime, timedelta, timezone
from langsmith import Client

client = Client()

project_name = "agent-token-optimization"
seven_days_ago = datetime.now(timezone.utc) - timedelta(days=7)

runs = list(
    client.list_runs(
        project_name=project_name,
        start_time=seven_days_ago,
        run_type="llm",  # chỉ lấy run gọi trực tiếp LLM, bỏ qua chain/tool wrapper
    )
)

print(f"Fetched {len(runs)} LLM runs from the last 7 days")

Tiếp theo, gộp (aggregate) token usage theo task_type — với điều kiện bạn đã gắn tag hoặc metadata task_type vào từng run khi gọi (LangSmith cho phép gắn tags và metadata tùy ý ở mọi lời gọi chain/agent):

from collections import defaultdict

usage_by_task: dict[str, dict[str, int]] = defaultdict(
    lambda: {"input_tokens": 0, "output_tokens": 0, "run_count": 0}
)

for run in runs:
    task_type = (run.extra or {}).get("metadata", {}).get("task_type", "unknown")
    usage = run.total_tokens or 0
    prompt_tokens = run.prompt_tokens or 0
    completion_tokens = run.completion_tokens or 0

    bucket = usage_by_task[task_type]
    bucket["input_tokens"] += prompt_tokens
    bucket["output_tokens"] += completion_tokens
    bucket["run_count"] += 1

print(f"{'Task Type':<30} {'Runs':>8} {'Avg Input':>12} {'Avg Output':>12}")
print("-" * 64)
for task_type, stats in sorted(
    usage_by_task.items(), key=lambda kv: kv[1]["input_tokens"], reverse=True
):
    avg_input = stats["input_tokens"] / stats["run_count"]
    avg_output = stats["output_tokens"] / stats["run_count"]
    print(
        f"{task_type:<30} {stats['run_count']:>8} "
        f"{avg_input:>12.0f} {avg_output:>12.0f}"
    )

Kết quả in ra một bảng tóm tắt hằng tuần, ví dụ:

Task Type                          Runs    Avg Input   Avg Output
----------------------------------------------------------------
document_summarization              412        8,240        420
customer_support_triage             891        2,150        180
code_review_agent                   156       15,600      1,240

Bảng này ngay lập tức chỉ ra chỗ đáng nghi: code_review_agent có avg input 15,600 token/lần — cao gấp 7 lần customer_support_triage. Nếu con số này không tương xứng với độ phức tạp thực tế của task, đó chính là ứng viên đầu tiên cần điều tra sâu (có thể do agent nhồi toàn bộ file diff thay vì chỉ đoạn code liên quan, hoặc system prompt quá dài).

Với QA và product manager không quen viết code Python, LangSmith cũng có UI web để xem trace trực quan (waterfall view từng bước, token count hiển thị ngay trên mỗi node) — không nhất thiết phải chạy script trên để có insight cơ bản; script này hữu ích khi cần tự động hóa báo cáo định kỳ hoặc tích hợp vào pipeline CI.

Mẹo

Gắn metadata={"task_type": ..., "workflow": ...} ngay tại điểm gọi chain/agent trong code (LangChain hỗ trợ config={"metadata": {...}, "tags": [...]} ở hầu hết API .invoke()/.run()), đừng đợi đến lúc phân tích mới cố suy luận từ tên chain. Nếu thiếu, mọi run sẽ rơi vào nhóm "unknown" và báo cáo sẽ vô dụng.

Xây Dashboard Token Tùy Biến — SQLite + Python

Không phải team nào cũng muốn (hoặc có ngân sách) trả phí cho một observability platform bên thứ ba, đặc biệt ở giai đoạn early-stage hoặc khi hệ thống chưa dùng LangChain. Một dashboard tự xây bằng SQLite + Python là lựa chọn cực kỳ thực tế: nhẹ, không cần server riêng, dễ backup (chỉ là một file .db), và đủ mạnh để phục vụ hàng triệu record log token trong nhiều năm.

Bắt đầu bằng schema lưu trữ:

CREATE TABLE IF NOT EXISTS token_sessions (
    session_id TEXT PRIMARY KEY,
    workflow TEXT NOT NULL,
    started_at TEXT NOT NULL,
    ended_at TEXT,
    status TEXT DEFAULT 'running'  -- running | completed | failed
);

CREATE TABLE IF NOT EXISTS token_events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id TEXT NOT NULL,
    task_type TEXT NOT NULL,
    agent_step TEXT NOT NULL,
    model TEXT NOT NULL,
    input_tokens INTEGER NOT NULL,
    output_tokens INTEGER NOT NULL,
    cache_read_tokens INTEGER DEFAULT 0,
    latency_ms INTEGER,
    created_at TEXT NOT NULL,
    FOREIGN KEY (session_id) REFERENCES token_sessions(session_id)
);

CREATE INDEX IF NOT EXISTS idx_events_session ON token_events(session_id);
CREATE INDEX IF NOT EXISTS idx_events_workflow ON token_sessions(workflow);

Việc tách hai bảng token_sessions (một "phiên" — tương ứng với một lần agent chạy từ đầu đến cuối) và token_events (từng lời gọi LLM riêng lẻ trong phiên đó) là mấu chốt để trả lời được cả câu hỏi cấp phiên ("một lần agent xử lý xong một ticket tốn tổng cộng bao nhiêu token") và câu hỏi cấp bước ("bước nào trong phiên đó tốn nhiều nhất").

Phần tiện lợi nhất khi tự xây dashboard là bạn có thể thiết kế API theo đúng cách team mình dùng agent. Đây là một context manager (bộ quản lý ngữ cảnh) cho phép tự động track một session, không cần nhớ gọi "mở" và "đóng" session ở đúng chỗ:

import sqlite3
import uuid
from contextlib import contextmanager
from datetime import datetime, timezone

DB_PATH = "token_footprint.db"


@contextmanager
def track_session(workflow: str):
    """Context manager for automatic session tracking.

    Usage:
        with track_session("customer_support_triage") as session:
            session.log_event(
                task_type="classify_ticket",
                agent_step="planner",
                model="claude-sonnet-5",
                input_tokens=1200,
                output_tokens=80,
            )
    """
    conn = sqlite3.connect(DB_PATH)
    session_id = str(uuid.uuid4())
    started_at = datetime.now(timezone.utc).isoformat()

    conn.execute(
        "INSERT INTO token_sessions (session_id, workflow, started_at, status) "
        "VALUES (?, ?, ?, 'running')",
        (session_id, workflow, started_at),
    )
    conn.commit()

    tracker = _SessionTracker(conn, session_id)

    try:
        yield tracker
        conn.execute(
            "UPDATE token_sessions SET status = 'completed', ended_at = ? "
            "WHERE session_id = ?",
            (datetime.now(timezone.utc).isoformat(), session_id),
        )
    except Exception:
        conn.execute(
            "UPDATE token_sessions SET status = 'failed', ended_at = ? "
            "WHERE session_id = ?",
            (datetime.now(timezone.utc).isoformat(), session_id),
        )
        raise
    finally:
        conn.commit()
        conn.close()


class _SessionTracker:
    def __init__(self, conn: sqlite3.Connection, session_id: str):
        self._conn = conn
        self._session_id = session_id

    def log_event(
        self,
        *,
        task_type: str,
        agent_step: str,
        model: str,
        input_tokens: int,
        output_tokens: int,
        cache_read_tokens: int = 0,
        latency_ms: int | None = None,
    ) -> None:
        self._conn.execute(
            """
            INSERT INTO token_events
                (session_id, task_type, agent_step, model,
                 input_tokens, output_tokens, cache_read_tokens,
                 latency_ms, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                self._session_id,
                task_type,
                agent_step,
                model,
                input_tokens,
                output_tokens,
                cache_read_tokens,
                latency_ms,
                datetime.now(timezone.utc).isoformat(),
            ),
        )
        self._conn.commit()

Cái hay của pattern context manager này là dù agent chạy thành công hay ném exception giữa đường, session vẫn được đóng đúng trạng thái (completed hoặc failed) nhờ khối try/except/finally — điều mà nếu code tay bằng cách gọi "start session" và "end session" riêng lẻ, bạn rất dễ quên gọi "end" khi có lỗi xảy ra, dẫn đến session bị treo ở trạng thái running vĩnh viễn và làm sai lệch báo cáo.

Sau khi đã có dữ liệu, truy vấn tổng hợp đơn giản bằng SQL thuần, ví dụ tính token trung bình mỗi workflow trong 30 ngày:

SELECT
    s.workflow,
    COUNT(DISTINCT s.session_id) AS total_sessions,
    SUM(e.input_tokens + e.output_tokens) AS total_tokens,
    ROUND(
        1.0 * SUM(e.input_tokens + e.output_tokens) / COUNT(DISTINCT s.session_id),
        0
    ) AS avg_tokens_per_session
FROM token_sessions s
JOIN token_events e ON e.session_id = s.session_id
WHERE s.started_at >= datetime('now', '-30 days')
GROUP BY s.workflow
ORDER BY avg_tokens_per_session DESC;

Với QA hoặc product manager không viết SQL/Python thường xuyên, bạn có thể mở file .db này bằng công cụ GUI như DB Browser for SQLite, hoặc kết nối trực tiếp vào các công cụ BI nhẹ (Metabase hỗ trợ SQLite qua plugin) để có dashboard trực quan mà không cần viết thêm code.

Mẹo

Đừng đợi có nhiều dữ liệu mới bắt đầu viết query phân tích — viết sẵn 3-5 query "core" (token theo workflow, token theo agent_step, session thất bại tốn token bao nhiêu so với thành công) ngay từ ngày đầu triển khai, lưu chúng thành file .sql trong repo. Khi cần báo cáo gấp cho stakeholder, bạn chỉ cần chạy lại thay vì viết mới dưới áp lực thời gian.

Xây Baseline Benchmark — Bức Tranh "Trước Khi Tối Ưu"

Đây là bước dễ bị bỏ qua nhất nhưng lại quan trọng nhất về mặt phương pháp: trước khi sửa bất kỳ điều gì (rút ngắn prompt, đổi model, thêm caching), bạn cần một baseline — một bộ số liệu "trước khi tối ưu" được đo trong điều kiện lặp lại được, để sau này so sánh công bằng. Không có baseline, bạn không thể phân biệt được "token giảm vì tối ưu thật" với "token giảm vì hôm nay người dùng hỏi câu ngắn hơn hôm qua".

Nguyên tắc cốt lõi để có baseline đáng tin cậy là benchmark task suite (bộ tác vụ chuẩn hóa) — một tập input cố định, đại diện cho các loại tác vụ thực tế mà agent của bạn xử lý, được chạy lại nhiều lần với cùng điều kiện. Việc định nghĩa bộ này cần có sự tham gia của cả engineer (biết agent xử lý gì) và QA/PM (biết use case thực tế của người dùng, tránh benchmark chỉ toàn case "dễ" mà bỏ qua case khó thường gặp trong thực tế).

from dataclasses import dataclass, field


@dataclass
class BenchmarkTask:
    task_id: str
    task_type: str
    description: str
    input_payload: dict
    expected_output_keywords: list[str] = field(default_factory=list)
    complexity: str = "medium"  # low | medium | high


BENCHMARK_SUITE: list[BenchmarkTask] = [
    BenchmarkTask(
        task_id="triage-001",
        task_type="customer_support_triage",
        description="Simple billing question, single intent, no attachments",
        input_payload={
            "ticket_text": "I was charged twice for my subscription this month.",
            "attachments": [],
        },
        expected_output_keywords=["billing", "refund", "duplicate"],
        complexity="low",
    ),
    BenchmarkTask(
        task_id="triage-002",
        task_type="customer_support_triage",
        description="Multi-intent ticket with a 3-message history and one attachment",
        input_payload={
            "ticket_text": "Also my export is broken AND I was overcharged, see screenshot.",
            "attachments": ["screenshot_export_error.png"],
            "conversation_history": ["...", "...", "..."],
        },
        expected_output_keywords=["export", "billing", "escalate"],
        complexity="high",
    ),
    BenchmarkTask(
        task_id="codereview-001",
        task_type="code_review_agent",
        description="Small diff, single file, 40 lines changed",
        input_payload={"diff_path": "fixtures/diffs/small_bugfix.diff"},
        expected_output_keywords=["approve", "suggestion"],
        complexity="low",
    ),
    BenchmarkTask(
        task_id="codereview-002",
        task_type="code_review_agent",
        description="Large diff spanning 6 files, includes a schema migration",
        input_payload={"diff_path": "fixtures/diffs/large_migration.diff"},
        expected_output_keywords=["migration", "breaking change", "rollback"],
        complexity="high",
    ),
]


def run_benchmark(suite: list[BenchmarkTask], agent_runner) -> list[dict]:
    """Run the full suite N times and record token usage per task.

    agent_runner is your actual agent entry point — the same function
    used in production, so the benchmark measures real behavior, not a
    simplified mock.
    """
    results = []
    for task in suite:
        for run_idx in range(3):  # chạy 3 lần mỗi task để lấy trung bình, giảm nhiễu
            output, usage_record = agent_runner(task.input_payload)
            results.append(
                {
                    "task_id": task.task_id,
                    "task_type": task.task_type,
                    "complexity": task.complexity,
                    "run_idx": run_idx,
                    "input_tokens": usage_record.input_tokens,
                    "output_tokens": usage_record.output_tokens,
                }
            )
    return results

Ba chi tiết thiết kế đáng chú ý trong benchmark suite này. Thứ nhất, mỗi task có trường complexity — vì trung bình gộp chung task đơn giản và task phức tạp sẽ cho một con số "trung bình" vô nghĩa, không phản ánh đúng cấu trúc chi phí thật. Thứ hai, mỗi task chạy lại 3 lần (run_idx) vì LLM có tính không xác định (non-deterministic) — kể cả cùng input, số token output có thể dao động 5-15% giữa các lần gọi do model chọn độ dài câu trả lời khác nhau; chạy một lần duy nhất rồi coi đó là "con số chính thức" là một lỗi phương pháp phổ biến. Thứ ba, benchmark gọi thẳng agent_runner — tức hàm agent thật đang chạy production, không phải một bản mock rút gọn; benchmark trên mock sẽ cho số liệu sai lệch hoàn toàn vì bỏ sót các bước tool call hoặc retry logic thật.

Sau khi có kết quả, việc tính baseline chỉ đơn giản là gộp theo task_type và lấy trung vị (median) thay vì trung bình (mean) — vì trung vị bền hơn với outlier (trường hợp agent bất ngờ lặp vòng dài do lỗi tạm thời):

import statistics

def summarize_baseline(results: list[dict]) -> dict:
    by_task_type: dict[str, list[int]] = {}
    for r in results:
        total = r["input_tokens"] + r["output_tokens"]
        by_task_type.setdefault(r["task_type"], []).append(total)

    return {
        task_type: {
            "median_tokens": statistics.median(totals),
            "p95_tokens": statistics.quantiles(totals, n=20)[18] if len(totals) >= 5 else max(totals),
            "sample_size": len(totals),
        }
        for task_type, totals in by_task_type.items()
    }

Baseline này chính là "before picture" — bạn lưu lại kết quả này (dưới dạng file JSON commit vào repo, hoặc ghi vào chính SQLite dashboard ở phần trước) và mỗi khi thử một thay đổi tối ưu (rút prompt, đổi model, thêm cache), bạn chạy lại đúng benchmark suite này và so sánh median/p95. Nếu không làm vậy, mọi khẳng định "tối ưu giúp giảm 20% token" chỉ là ước lượng chủ quan, không kiểm chứng được và rất dễ bị nghi ngờ khi trình bày với stakeholder.

Mẹo

Version benchmark suite của bạn giống như version code — mỗi khi thêm/sửa task trong suite, tăng version number và lưu lại baseline cũ. Nếu không, 6 tháng sau bạn sẽ không biết con số "baseline 8,200 token" đang so sánh với suite nào, dẫn đến so sánh sai giữa hai thời điểm dùng bộ task khác nhau.

Giám Sát Token Real-Time — Observability Cho Production

Benchmark và dashboard lịch sử giúp bạn hiểu hành vi trung bình, nhưng production có input đa dạng và bất ngờ hơn nhiều so với bất kỳ benchmark suite nào — vì vậy bạn cần một tầng giám sát real-time riêng, tích hợp vào observability stack đã có của tổ chức (Datadog, Grafana, CloudWatch) thay vì chỉ dựa vào dashboard tự xây chạy batch.

Nguyên tắc thiết kế: emit token usage như một metric có nhiều dimension (tag/label), không phải chỉ một con số tổng. Dưới đây là ví dụ dùng StatsD-style client (tương thích Datadog):

from datadog import statsd

def emit_token_metrics(
    *,
    workflow: str,
    task_type: str,
    model: str,
    input_tokens: int,
    output_tokens: int,
    latency_ms: int,
    status: str,  # success | error | timeout
) -> None:
    tags = [
        f"workflow:{workflow}",
        f"task_type:{task_type}",
        f"model:{model}",
        f"status:{status}",
    ]

    statsd.increment("agent.token.calls", tags=tags)
    statsd.histogram("agent.token.input", input_tokens, tags=tags)
    statsd.histogram("agent.token.output", output_tokens, tags=tags)
    statsd.histogram("agent.token.total", input_tokens + output_tokens, tags=tags)
    statsd.histogram("agent.latency_ms", latency_ms, tags=tags)

    # Cost estimate, useful for dollar-denominated dashboards and alerts
    estimated_cost_usd = _estimate_cost(model, input_tokens, output_tokens)
    statsd.histogram("agent.token.cost_usd", estimated_cost_usd, tags=tags)

Dùng histogram (chứ không phải gauge hay counter đơn thuần) là chủ đích — vì bạn sẽ cần xem phân phối (p50, p95, p99) chứ không chỉ trung bình. Một agent có p50 token thấp nhưng p99 rất cao là dấu hiệu điển hình của "long tail" — số ít request rơi vào vòng lặp dài hoặc input bất thường, và p99 chính là chỉ số cần theo dõi để bắt được vấn đề này trước khi nó ảnh hưởng đến ngân sách tổng.

Với alert, ngưỡng cần được điều chỉnh theo quy mô (adjust to your scale) — một ngưỡng cố định tuyệt đối (ví dụ "báo động nếu token/request > 10,000") sẽ vô nghĩa khi traffic tăng gấp đôi hoặc khi bạn có nhiều workflow với đặc điểm token rất khác nhau. Cách làm bền hơn là alert dựa trên độ lệch so với baseline động của từng workflow:

ALERT_THRESHOLDS = {
    "customer_support_triage": {
        "p95_tokens_warning": 4_000,     # baseline p95 là ~2,500
        "p95_tokens_critical": 8_000,
        "cost_per_hour_warning_usd": 15.0,
        "cost_per_hour_critical_usd": 40.0,
    },
    "code_review_agent": {
        "p95_tokens_warning": 25_000,    # baseline p95 là ~16,000, task này vốn tốn hơn
        "p95_tokens_critical": 50_000,
        "cost_per_hour_warning_usd": 30.0,
        "cost_per_hour_critical_usd": 80.0,
    },
}


def check_alert_conditions(workflow: str, current_p95: float, current_cost_per_hour: float) -> str | None:
    thresholds = ALERT_THRESHOLDS.get(workflow)
    if thresholds is None:
        return None

    if (
        current_p95 > thresholds["p95_tokens_critical"]
        or current_cost_per_hour > thresholds["cost_per_hour_critical_usd"]
    ):
        return "critical"
    if (
        current_p95 > thresholds["p95_tokens_warning"]
        or current_cost_per_hour > thresholds["cost_per_hour_warning_usd"]
    ):
        return "warning"
    return None

Chú ý rằng ngưỡng warning/critical của mỗi workflow được đặt riêng, dựa trên baseline p95 đã đo được từ benchmark ở phần trước (code_review_agent vốn tốn token nhiều hơn customer_support_triage một cách tự nhiên do bản chất công việc, nên ngưỡng cũng cao hơn tương ứng — không dùng một ngưỡng chung cho tất cả). Đây chính là lý do baseline benchmark ở phần trước không chỉ có giá trị "trước/sau tối ưu" mà còn là input trực tiếp để thiết lập alert threshold hợp lý cho production.

Với các dashboard trên Grafana hoặc CloudWatch, panel hữu ích nhất thường là: (1) token/giờ theo workflow (line chart, giúp thấy xu hướng), (2) chi phí ước tính/giờ toàn hệ thống (giúp product manager và tài chính theo dõi mà không cần hiểu chi tiết kỹ thuật), và (3) tỷ lệ request rơi vào nhóm "high token" (ví dụ vượt p95 baseline) — panel này giúp phát hiện sớm khi một thay đổi code (deploy mới) vô tình làm tăng token tiêu thụ trước khi nó phản ánh lên bill cuối tháng.

Mẹo

Luôn emit thêm tag status (success/error/timeout) cùng với metric token. Rất nhiều trường hợp token tăng bất thường không phải do prompt dài hơn, mà do agent gặp lỗi và tự động retry — nếu không tách theo status, bạn sẽ đổ oan cho "prompt" trong khi nguyên nhân thật là retry logic thiếu backoff.

Báo Cáo Token Footprint — Truyền Đạt Kết Quả Cho Stakeholder

Có dữ liệu tốt chưa đủ — bạn còn cần trình bày nó theo cách mà stakeholder không rành kỹ thuật (product manager, ban lãnh đạo, khách hàng nội bộ) hiểu được giá trị và đưa ra quyết định đúng (ví dụ phê duyệt ngân sách cho dự án tối ưu, hoặc chấp nhận đánh đổi độ trễ để giảm chi phí).

Một báo cáo token footprint hiệu quả nên có cấu trúc rõ ràng, đi từ tổng quan đến chi tiết, theo thứ tự sau:

1. Tóm tắt điều hành (executive summary) — 3-4 câu, không thuật ngữ kỹ thuật. Ví dụ: "Hệ thống agent hỗ trợ khách hàng hiện tiêu thụ trung bình 2,150 token/ticket, tương đương $0.018/ticket ở mức giá model hiện tại. Với 45,000 ticket/tháng, chi phí LLM ước tính $810/tháng. So với baseline 3 tháng trước ($1,240/tháng), chi phí đã giảm 35% nhờ áp dụng prompt caching — không ảnh hưởng đến chất lượng trả lời (đo qua benchmark)."

2. Bảng số liệu theo workflow — trình bày dưới dạng bảng dễ đọc, không phải dump SQL thô:

3. Biểu đồ xu hướng theo thời gian — line chart token/request hoặc chi phí/tháng theo tuần, đủ để stakeholder tự thấy xu hướng tăng/giảm mà không cần đọc số liệu thô. Nên đánh dấu (annotation) trên chart các mốc quan trọng: ngày deploy thay đổi model, ngày bật caching, ngày phát hiện bug gây token tăng bất thường — việc gắn nguyên nhân vào đúng thời điểm trên biểu đồ giúp câu chuyện dễ hiểu hơn nhiều so với chỉ nhìn đường biểu đồ trần trụi.

4. Phần "rủi ro và khuyến nghị" — đây là phần biến báo cáo từ "thông tin" thành "quyết định được đưa ra". Ví dụ: "Workflow code_review_agent có p99 token cao gấp 3.2 lần p50, cho thấy một số ít review (diff lớn/phức tạp) đang tốn chi phí không tương xứng. Khuyến nghị: giới hạn kích thước diff đưa vào agent, hoặc chia review lớn thành nhiều lượt nhỏ — ước tính giảm 20-25% chi phí của workflow này mà không ảnh hưởng chất lượng review."

Với đối tượng là engineer khác trong team, bạn có thể đính kèm phụ lục chi tiết hơn: breakdown theo agent_step (bước nào tốn token nhất), số liệu benchmark trước/sau khi có thay đổi cụ thể, và link tới dashboard/trace gốc để họ tự drill-down khi cần. Với đối tượng là QA, phần liên quan nhất thường là mối liên hệ giữa token usage và chất lượng output (ví dụ agent trả lời ngắn hơn có làm giảm độ chính xác không) — nên báo cáo cần có cả số liệu benchmark chất lượng đi kèm số liệu token, không chỉ báo cáo chi phí đơn thuần.

Một nguyên tắc quan trọng khi trình bày báo cáo: luôn nêu rõ khoảng thời gian đo, số lượng sample, và phương pháp tính (trung vị hay trung bình, có loại outlier không) ngay trong báo cáo — vì stakeholder có kinh nghiệm sẽ luôn hỏi "con số này đo trong bao lâu, dựa trên bao nhiêu request?" và nếu bạn không chuẩn bị trước câu trả lời, độ tin cậy của toàn bộ báo cáo sẽ bị nghi ngờ.

Mẹo

Luôn kèm một biểu đồ hoặc bảng "before/after" ngay đầu báo cáo cho bất kỳ dự án tối ưu token, thay vì chỉ báo cáo trạng thái hiện tại. Con số tuyệt đối ("2,150 token/ticket") ít thuyết phục hơn nhiều so với con số so sánh có ngữ cảnh ("giảm 35% so với 3 tháng trước, tương đương tiết kiệm $430/tháng") — đây là điều khiến stakeholder không rành kỹ thuật thực sự cảm nhận được giá trị công việc tối ưu token.

Những điểm chính

• Token measurement stack có 5 tầng bổ sung cho nhau: provider-native reporting, observability platform (LangSmith), dashboard tự xây, baseline benchmark, và real-time monitoring — mỗi tầng trả lời một câu hỏi khác nhau, không tầng nào thay thế được tầng khác.
• Số liệu token thô từ provider (usage.input_tokens, usage.output_tokens) chỉ có giá trị khi được gắn metadata (workflow, task_type, agent_step) ngay tại điểm gọi API — thiếu bước này, mọi phân tích sau đó đều phải suy luận ngược, tốn thời gian và dễ sai.
• LangSmith (và các platform tương tự như Langfuse) tự động trace token usage cho toàn bộ chain/agent qua vài biến môi trường, giúp bạn truy vấn và tổng hợp theo nhiều chiều mà không cần tự viết code log cho từng lời gọi.
• Dashboard SQLite + Python là lựa chọn thực tế cho team chưa có ngân sách dùng observability platform bên thứ ba; dùng context manager để tự động track session giúp tránh lỗi quên đóng session khi agent gặp exception.
• Baseline benchmark — một bộ task cố định chạy lặp lại nhiều lần, tổng hợp bằng trung vị và p95 — là điều kiện bắt buộc để mọi so sánh "trước/sau tối ưu" có căn cứ, thay vì chỉ là ước lượng cảm tính.
• Giám sát production nên emit metric dạng histogram với nhiều tag (workflow, task_type, model, status), và ngưỡng alert cần được thiết lập riêng theo baseline của từng workflow, không dùng một ngưỡng chung.
• Báo cáo token footprint cho stakeholder cần đi từ tóm tắt điều hành, đến bảng số liệu, biểu đồ xu hướng có annotation, và kết ở phần khuyến nghị hành động cụ thể — luôn kèm so sánh before/after và minh bạch về phương pháp đo để báo cáo đủ độ tin cậy.