Overview and Setup

Đây là bài lab thực hành tổng hợp (capstone lab) của toàn module. Khác với các bài trước — nơi mỗi kỹ thuật tối ưu token được trình bày riêng lẻ — bài này sẽ ghép toàn bộ các kỹ thuật đó lại vào một agentic workflow (luồng xử lý của agent) thực tế: một code review agent dùng LangGraph để tự động review pull request.

Chúng ta sẽ đi theo đúng quy trình mà một kỹ sư senior làm khi được giao nhiệm vụ "giảm chi phí AI agent xuống": đo baseline (đường mốc gốc) trước, áp từng optimization (tối ưu hóa) một, đo lại sau mỗi bước, và cuối cùng tổng hợp số liệu để chứng minh hiệu quả. Đây chính là quy trình bạn sẽ áp dụng ngoài thực tế khi sếp hỏi "sao token bill (hoá đơn token) tháng này cao vậy?".

Bài toán cụ thể: Agent nhận vào một pull request gồm nhiều file thay đổi (diff), phải review từng file, tìm bug/code smell/vấn đề bảo mật, rồi tổng hợp thành một báo cáo review cuối cùng. Đây là use case rất phổ biến — nhiều team engineering ở Việt Nam hiện đã có bot review PR tự động dùng LLM, và bài toán tối ưu token cho loại agent này là bài toán có thật, không phải ví dụ hàn lâm.

Setup môi trường:

python3 -m venv .venv
source .venv/bin/activate

pip install langgraph langchain-anthropic tiktoken python-dotenv

export ANTHROPIC_API_KEY="sk-ant-your-key-here"

Chuẩn bị một bộ dữ liệu test cố định — quan trọng: dùng đúng một bộ PR mẫu cho mọi lần benchmark (đo lường hiệu năng), để việc so sánh baseline và optimized (đã tối ưu) là công bằng (apples-to-apples). Trong bài lab này, ta giả định một PR mẫu gồm 6 file thay đổi, tổng cộng khoảng 340 dòng diff — kích cỡ PR trung bình khá phổ biến trong thực tế (không quá nhỏ để optimization không có ý nghĩa, không quá lớn để vượt quá phạm vi bài lab).

SAMPLE_PR_FILES = [
    {"path": "src/auth/login_handler.py", "diff_lines": 62, "has_issue": True},
    {"path": "src/auth/token_refresh.py", "diff_lines": 48, "has_issue": True},
    {"path": "src/utils/validators.py", "diff_lines": 35, "has_issue": False},
    {"path": "src/api/routes/user.py", "diff_lines": 71, "has_issue": True},
    {"path": "src/db/migrations/0042_add_index.sql", "diff_lines": 18, "has_issue": False},
    {"path": "tests/test_login_handler.py", "diff_lines": 106, "has_issue": False},
]

Chú ý cột has_issue — ta cố tình biết trước 3/6 file có vấn đề thật (bug hoặc security issue được cài sẵn) và 3/6 file "sạch". Việc này giúp bài lab kiểm chứng được không chỉ token giảm mà chất lượng review không bị giảm — một optimization mà làm agent bỏ sót bug thì vô giá trị, dù có tiết kiệm token đến đâu.

Mẹo

• Luôn cố định (pin) bộ fixture test trước khi bắt đầu optimize. Nếu dữ liệu test thay đổi giữa các lần đo, bạn sẽ không biết token giảm là do optimization hay do PR lần sau đơn giản hơn.
• Dùng tiktoken hoặc token counter API của provider để đếm token thực tế thay vì ước lượng bằng số từ — sai lệch giữa ước lượng và số đo thực có thể lên tới 20-30%.
• Ghi log token usage vào file JSON/CSV ngay từ bước đầu (baseline), đừng để tới cuối mới nghĩ tới việc lưu số liệu — bạn sẽ cần so sánh xuyên suốt cả bài lab.

Step 0: Establish the Baseline

Trước khi tối ưu bất cứ thứ gì, phải có một baseline (đường mốc gốc) đáng tin cậy. Baseline ở đây là một agent "viết theo cách tự nhiên nhất" — đúng như cách một kỹ sư mới bắt đầu với LangGraph, chưa nghĩ đến việc tối ưu token, sẽ viết ra.

Đặc điểm của baseline agent (rất phổ biến trong thực tế):
- System prompt (chỉ dẫn hệ thống) dài, chung chung, cố gắng "dạy" mô hình mọi thứ về coding convention, security, style guide... trong một khối văn bản.
- Toàn bộ lịch sử hội thoại (conversation history) được giữ nguyên và truyền lại qua từng node trong graph — không có khái niệm "quên" ngữ cảnh không cần thiết.
- Không có structured output (đầu ra có cấu trúc) — model tự do viết review dưới dạng văn xuôi dài dòng.
- Không cache, không early exit, không parallel — mọi file được xử lý tuần tự, đầy đủ, không phân biệt file có vấn đề hay không.

Baseline Workflow Code

from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from typing import TypedDict, List

BASELINE_SYSTEM_PROMPT = """
You are an extremely thorough and experienced senior software engineer
acting as a code reviewer for our engineering organization. Your job is
to carefully review every single pull request that comes through our
CI/CD pipeline, paying close attention to code quality, maintainability,
readability, performance, and security. You should think like a staff
engineer with 15 years of experience across backend systems, frontend
applications, and infrastructure code.

When reviewing code, please consider the following extensive checklist
of things to look for, in no particular order of priority, because all
of them matter equally and none should be skipped even if the file is
short or simple:

1. Naming conventions — are variable, function, and class names
   descriptive, consistent with the rest of the codebase, and free of
   abbreviations that might confuse future maintainers?
2. Error handling — are exceptions caught appropriately? Is there
   silent failure anywhere? Are error messages descriptive and
   actionable for whoever reads the logs later?
3. Security concerns — SQL injection, XSS, hardcoded secrets,
   insecure deserialization, path traversal, insufficient input
   validation, broken authentication or session handling...
   [... the real prompt continues for ~40 more lines covering
   performance, testing, documentation, style, architecture,
   backward compatibility, and team-specific conventions ...]

Please write your review in a detailed, narrative style, explaining
your reasoning for every comment you make, even for minor stylistic
suggestions, so that junior engineers on the team can learn from your
feedback. Do not be terse. Elaborate on the "why" behind each finding.
"""

class ReviewState(TypedDict):
    pr_files: List[dict]
    history: List[dict]        # accumulates across every file — never trimmed
    findings: List[str]        # raw narrative text, one long block per file
    final_report: str

llm = ChatAnthropic(model="claude-sonnet-5", temperature=0)

def review_node(state: ReviewState) -> ReviewState:
    for file in state["pr_files"]:
        # The FULL history (every prior file's review + full diff) is
        # re-sent on every turn — this is the main token sink.
        messages = [{"role": "system", "content": BASELINE_SYSTEM_PROMPT}]
        messages.extend(state["history"])
        messages.append({
            "role": "user",
            "content": f"Please review this file in full detail:\n\n"
                        f"File: {file['path']}\n"
                        f"Diff:\n{file['diff_lines']} lines of changes here...\n\n"
                        f"Write a comprehensive narrative review covering every "
                        f"item in your checklist, even if there is nothing to "
                        f"report for that item."
        })
        response = llm.invoke(messages)
        state["history"].append({"role": "user", "content": messages[-1]["content"]})
        state["history"].append({"role": "assistant", "content": response.content})
        state["findings"].append(response.content)
    return state

def summarize_node(state: ReviewState) -> ReviewState:
    # Summary step re-sends the ENTIRE accumulated history one more time
    messages = [{"role": "system", "content": BASELINE_SYSTEM_PROMPT}]
    messages.extend(state["history"])
    messages.append({"role": "user", "content": "Now write a final combined report."})
    state["final_report"] = llm.invoke(messages).content
    return state

graph = StateGraph(ReviewState)
graph.add_node("review", review_node)
graph.add_node("summarize", summarize_node)
graph.add_edge("review", "summarize")
graph.add_edge("summarize", END)
graph.set_entry_point("review")
baseline_app = graph.compile()

Nhìn vào code trên, ba "thủ phạm" đốt token lộ rõ ngay: (1) system prompt 1.847 token bị gửi lại mỗi lần gọi model, (2) history phình to liên tục vì không bao giờ bị cắt, và (3) không có structured output nên mỗi file được review bằng một đoạn văn dài dòng, không kiểm soát được độ dài đầu ra.

Benchmark: Run Baseline and Record

Chạy baseline trên bộ fixture 6 file và ghi lại số liệu:

import time
import json

def count_tokens_in_messages(messages, model="claude-sonnet-5"):
    # Using Anthropic's token counting endpoint for accuracy
    from anthropic import Anthropic
    client = Anthropic()
    return client.messages.count_tokens(
        model=model, messages=messages
    ).input_tokens

def run_benchmark(app, initial_state, label):
    start = time.perf_counter()
    result = app.invoke(initial_state)
    elapsed = time.perf_counter() - start

    # Sum input + output tokens across every LLM call made during the run
    # (in practice, wrap llm.invoke with a callback/handler to capture this)
    metrics = {
        "label": label,
        "total_input_tokens": result.get("_input_tokens_sum", 0),
        "total_output_tokens": result.get("_output_tokens_sum", 0),
        "latency_seconds": round(elapsed, 2),
    }
    with open(f"benchmark_{label}.json", "w") as f:
        json.dump(metrics, f, indent=2)
    return metrics

initial_state = {"pr_files": SAMPLE_PR_FILES, "history": [], "findings": [], "final_report": ""}
baseline_metrics = run_benchmark(baseline_app, initial_state, "baseline")
print(baseline_metrics)

Kết quả benchmark baseline điển hình cho bộ 6 file trên (đây là số liệu tham chiếu xuyên suốt bài lab):

Đây là con số baseline ta sẽ dùng làm mẫu số cho mọi phần trăm giảm token ở các phần sau.

Mẹo

• Chạy baseline benchmark ít nhất 3 lần và lấy trung bình — LLM output length có thể dao động ±10% giữa các lần chạy dù input giống nhau, do temperature và tính không xác định (non-determinism) của model.
• Lưu lại toàn bộ payload request (không chỉ số token) của baseline vào file, để sau này debug khi nghi ngờ optimization làm sai lệch hành vi agent.
• Đừng benchmark trên PR quá nhỏ (1-2 file) — hiệu ứng của caching, scoped session sẽ không thể hiện rõ, dễ dẫn đến kết luận sai về mức độ hiệu quả của optimization.

Optimization 1: Focused System Prompt (Target: -15% tokens)

Optimization đầu tiên và dễ nhất: cắt bớt system prompt. System prompt của baseline dài 1.847 token nhưng thực chất chỉ có một phần nhỏ nội dung là thứ model thực sự cần để làm tốt việc review — phần còn lại là văn phong "dạy dỗ" không cần thiết cho một model đã được huấn luyện sẵn kiến thức coding.

Nguyên tắc khi viết lại system prompt: liệt kê checklist ngắn gọn dạng bullet, bỏ phần giải thích "tại sao" (model không cần lý do, chỉ cần chỉ dẫn), bỏ yêu cầu "viết narrative dài dòng" (đây chính là thứ khiến output token cũng phình to theo).

OPTIMIZED_SYSTEM_PROMPT = """
You are a senior code reviewer. Review the given diff for:
- Bugs and logic errors
- Security issues (injection, secrets, auth, input validation)
- Error handling gaps
- Naming/readability issues that would confuse maintainers

Rules:
- Only report actual findings. If none, say "No issues found."
- Be terse: one line per finding, format "[SEVERITY] file:line — issue".
- Severities: HIGH, MEDIUM, LOW.
- Do not explain reasoning or restate the code. Do not praise good code.
"""

Điểm mấu chốt không nằm ở việc cắt số từ, mà ở thay đổi hành vi đầu ra của model: baseline yêu cầu "explain your reasoning for every comment... elaborate on the why" — chính câu này khiến output dài ra rất nhiều lần so với input. Prompt tối ưu ra lệnh ngược lại: "Do not explain reasoning". Đây là lý do vì sao optimization 1 tác động lên cả input và output token, không chỉ input.

So sánh nhanh trên 1 file mẫu (file login_handler.py, 62 dòng diff):

Áp dụng cho toàn bộ 7 lời gọi LLM (6 review + 1 summarize) trong workflow, tổng token giảm từ 34.500 xuống khoảng 29.300 — tương đương -15%, đúng target đề ra.

Mẹo

• Test system prompt mới trên các case "khó" trước (file có bug thật) để đảm bảo việc cắt gọn không làm model bỏ sót vấn đề — đo recall (độ bao phủ phát hiện lỗi), không chỉ đo token.
• Câu lệnh "Do not explain reasoning" cực kỳ hiệu quả để giảm output token nhưng có thể khiến báo cáo khó hiểu với người đọc không rành code — cân nhắc thêm bước "expand on demand" nếu người dùng cuối cần giải thích chi tiết hơn.
• Giữ system prompt trong một file riêng (không hardcode trong node function) để dễ A/B test nhiều phiên bản prompt mà không phải sửa logic graph.

Optimization 2: File-by-File Scoped Sessions (Target: -25% additional)

Đây là optimization có tác động lớn nhất trong toàn bài lab. Nhìn lại baseline: mỗi lần review một file mới, toàn bộ history — bao gồm diff và review của mọi file trước đó — bị gửi lại cho model. Với PR 6 file, lần review file thứ 6 phải gửi lại gần như toàn bộ nội dung của 5 file trước, dù nội dung đó hoàn toàn không liên quan đến việc đánh giá file thứ 6.

Giải pháp: mỗi file được review trong một scoped session (session có phạm vi giới hạn) độc lập — không kế thừa lịch sử từ file trước. Model chỉ cần system prompt + diff của đúng file đang xét để đưa ra nhận định, vì việc review một file Python không phụ thuộc vào việc file SQL trước đó có vấn đề gì.

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-5", temperature=0)

def review_single_file_scoped(file: dict, system_prompt: str) -> str:
    """
    Each call is a fresh, isolated session — no accumulated history.
    Only the system prompt + this file's diff are sent, regardless of
    how many files were reviewed before or after this one.
    """
    messages = [
        {"role": "system", "content": system_prompt},
        {
            "role": "user",
            "content": f"File: {file['path']}\nDiff ({file['diff_lines']} lines):\n"
                        f"{file['diff_content']}"
        },
    ]
    response = llm.invoke(messages)
    return response.content

def scoped_review_loop(pr_files: list, system_prompt: str) -> list:
    # No accumulating history object at all — each iteration is independent.
    # This is the key structural change vs. baseline_agent.py's review_node.
    findings = []
    for file in pr_files:
        result = review_single_file_scoped(file, system_prompt)
        findings.append({"path": file["path"], "review": result})
    return findings

Lưu ý điểm khác biệt cấu trúc so với baseline: hàm scoped_review_loop không còn tham số history đi qua từng vòng lặp — mỗi lời gọi review_single_file_scoped là một transaction độc lập hoàn toàn. Về mặt LangGraph, điều này thường được triển khai bằng cách mỗi file là một sub-graph invocation riêng, hoặc dùng map pattern thay vì node tuần tự chia sẻ state.

Có một câu hỏi hợp lý sẽ nảy ra: "Vậy summarize step lấy thông tin từ đâu nếu không có history?" — câu trả lời nằm ở Optimization 3 (structured output): thay vì gửi lại toàn bộ narrative dài dòng, ta chỉ cần gửi danh sách finding đã được rút gọn cho bước summarize, chứ không phải toàn bộ transcript hội thoại.

Tác động đo được: vì history không còn tích lũy, chi phí trung bình cho mỗi file review giữ nguyên phẳng (flat) thay vì tăng dần theo cấp số. Trên bộ 6 file, tổng input token giảm thêm khoảng 25% so với con số sau Optimization 1: từ ~29.300 xuống khoảng ~22.700.

Mẹo

• Scoped session hiệu quả nhất khi các sub-task (ở đây là mỗi file) độc lập về ngữ nghĩa — nếu bài toán của bạn có file A phụ thuộc logic vào file B (ví dụ B định nghĩa interface mà A implement), cần truyền một bản tóm tắt tối thiểu (không phải toàn bộ), không phải cắt hoàn toàn liên kết.
• Kiểm tra kỹ trường hợp review chất lượng có giảm không khi mất context xuyên file — ví dụ, agent baseline có thể phát hiện "biến này đặt tên không nhất quán với file khác trong PR", điều mà scoped session baseline sẽ không thấy được. Nếu use case của bạn cần loại phát hiện xuyên file này, hãy thêm một node review riêng ở cấp PR (không phải cấp file) chạy sau scoped loop.
• Đặt tên rõ cho pattern này trong code review nội bộ team — "scoped session" hay "stateless-per-item loop" — để tránh nhầm với cache hoặc memory khác trong hệ thống.

Optimization 3: Structured Output with Compressed Handoffs (Target: -10% additional)

Optimization 2 giải quyết vấn đề input token phình to do history, nhưng vẫn còn một vấn đề: output của mỗi lần review vẫn là văn xuôi tự do, và bước summarize cuối cùng vẫn cần nhận đủ thông tin từ 6 file để viết báo cáo tổng hợp — nếu 6 "review" đó là 6 đoạn văn dài, ta lại quay về bài toán cũ ở quy mô nhỏ hơn.

Giải pháp: bắt model trả về structured output (đầu ra có cấu trúc) — JSON hoặc Pydantic schema — cho mỗi file, sau đó handoff (chuyển giao) giữa các bước trong graph chỉ truyền structured data đã được compress (nén gọn), không truyền lại prose.

from pydantic import BaseModel, Field
from typing import Literal, List

class Finding(BaseModel):
    severity: Literal["HIGH", "MEDIUM", "LOW"]
    line: int
    issue: str = Field(max_length=120)  # hard cap forces conciseness

class FileReviewResult(BaseModel):
    path: str
    findings: List[Finding]
    clean: bool  # True when findings is empty — used in Optimization 5

structured_llm = llm.with_structured_output(FileReviewResult)

def review_single_file_structured(file: dict, system_prompt: str) -> FileReviewResult:
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": f"File: {file['path']}\nDiff:\n{file['diff_content']}"},
    ]
    return structured_llm.invoke(messages)

def build_compressed_handoff(results: List[FileReviewResult]) -> str:
    """
    Instead of forwarding full narrative text to the summarize step,
    forward a compact serialized form — this is the "handoff" payload.
    """
    lines = []
    for r in results:
        if r.clean:
            continue  # clean files contribute nothing to the handoff
        for f in r.findings:
            lines.append(f"[{f.severity}] {r.path}:{f.line} — {f.issue}")
    return "\n".join(lines)

Hai cơ chế cộng hưởng ở đây: (1) max_length=120 trên field issue buộc model tự nén nội dung ngay từ lúc sinh ra — không cần bước tóm tắt riêng sau đó; (2) build_compressed_handoff chỉ chuyển tiếp finding thực sự có vấn đề (clean=False), loại bỏ hoàn toàn nội dung của các file sạch trong payload gửi cho bước summarize.

Bước summarize giờ nhận một payload rất gọn — vài chục dòng text có cấu trúc, thay vì hàng nghìn token narrative:

def summarize_from_handoff(handoff_text: str, system_prompt: str) -> str:
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": f"Findings:\n{handoff_text}\n\nWrite a 1-paragraph PR summary."},
    ]
    return llm.invoke(messages).content

Tác động: output token cho mỗi file giảm mạnh vì schema ép cô đọng, và input token của bước summarize giảm gần như toàn bộ. Từ mốc ~22.700 sau Optimization 2, tổng token giảm thêm ~10%, xuống khoảng ~20.400.

Mẹo

• Structured output với max_length là cách "ép" model tự tóm tắt tại nguồn — rẻ hơn nhiều so với gọi thêm một lần LLM riêng để tóm tắt lại review dài.
• Khi dùng with_structured_output, luôn có fallback xử lý trường hợp model trả về JSON không hợp lệ (schema validation error) — thêm retry với instruction nhắc lại format, tránh crash cả pipeline vì 1 file lỗi.
• Compressed handoff nên dùng format dòng đơn giản (line-based) như ví dụ trên, không nên dùng JSON đầy đủ lại cho bước tiếp theo — JSON có nhiều ký tự cấu trúc ({, ", :) tiêu tốn token oan uổng khi chỉ cần truyền qua nội bộ pipeline.

Optimization 4: Tool Result Caching (Target: -5% on repeated runs)

Trong thực tế, agentic workflow này thường không chạy đúng một lần — pipeline CI/CD có thể trigger review lại nhiều lần cho cùng một PR (mỗi lần push thêm commit nhỏ, mỗi lần rerun do flaky test, hoặc do reviewer bấm "re-run check"). Nếu file A không đổi giữa 2 lần chạy, việc review lại từ đầu là lãng phí token thuần túy.

Giải pháp: cache (đệm) kết quả review theo content hash (mã băm nội dung) của diff, không theo tên file hay PR ID — vì tên file/PR không đổi nhưng nội dung diff mới đổi mới là thứ quyết định kết quả review có cần chạy lại hay không.

import hashlib
import json
import os

CACHE_DIR = ".review_cache"
os.makedirs(CACHE_DIR, exist_ok=True)

def _content_hash(diff_content: str) -> str:
    return hashlib.sha256(diff_content.encode("utf-8")).hexdigest()[:16]

def get_cached_review(diff_content: str) -> FileReviewResult | None:
    key = _content_hash(diff_content)
    cache_path = os.path.join(CACHE_DIR, f"{key}.json")
    if os.path.exists(cache_path):
        with open(cache_path) as f:
            data = json.load(f)
        return FileReviewResult(**data)
    return None

def set_cached_review(diff_content: str, result: FileReviewResult) -> None:
    key = _content_hash(diff_content)
    cache_path = os.path.join(CACHE_DIR, f"{key}.json")
    with open(cache_path, "w") as f:
        json.dump(result.model_dump(), f)

def review_single_file_cached(file: dict, system_prompt: str) -> FileReviewResult:
    cached = get_cached_review(file["diff_content"])
    if cached is not None:
        return cached  # zero LLM tokens spent — cache hit
    result = review_single_file_structured(file, system_prompt)
    set_cached_review(file["diff_content"], result)
    return result

Điểm quan trọng cần nhấn mạnh trong bài lab: cache theo content hash của diff, không theo file path. Nếu file login_handler.py không thay đổi giữa lần push commit thứ 1 và thứ 2 của cùng PR, hash của diff content giống nhau tuyệt đối, cache hit, tiết kiệm 100% token cho file đó ở lần chạy thứ 2. Ngược lại, nếu file đổi tên nhưng nội dung diff giống file cũ (trường hợp hiếm nhưng có thể xảy ra khi refactor), cache vẫn đúng vì key không phụ thuộc path.

Con số "-5%" trong target không tính trên một lần chạy đơn (vì lần đầu chạy không có gì để cache hit), mà tính trung bình trên nhiều lần chạy CI thực tế của cùng một PR — theo dữ liệu quan sát thực tế, khoảng 40-60% file trong một PR không đổi giữa 2 lần CI run liên tiếp (do người dùng chỉ sửa 1-2 file rồi push lại), nên cache mang lại mức giảm trung bình ổn định khoảng 5% trên tổng chi phí vòng đời của một PR.

Mẹo

• Đặt TTL (thời gian sống) cho cache — code review logic của bạn có thể thay đổi (system prompt update, model version update) và bạn không muốn serve lại review cũ dựa trên logic đã lỗi thời. Thêm prompt version vào cache key để tự động invalidate khi đổi prompt.
• Cache theo content hash hoạt động tốt nhất với các bước có tính deterministic (tất định) — nếu bạn dùng temperature > 0, việc "review lại vẫn cho cùng kết quả" không còn đúng về mặt logic, cache lúc này chỉ nên coi là "tiết kiệm chi phí có đánh đổi", không phải "đảm bảo đúng kết quả mới nhất".
• Trong production, dùng cache layer chung (Redis, DynamoDB) thay vì file JSON cục bộ như ví dụ trên — ví dụ trên chỉ để minh họa cơ chế, không phù hợp khi có nhiều CI runner chạy song song.

Optimization 5: Early Termination on Empty Findings (Target: -3% average)

Nhớ lại field clean: bool được định nghĩa ở Optimization 3 — đây chính là điều kiện cho optimization tiếp theo. Baseline luôn bắt model viết đủ "checklist đầy đủ" cho mọi file, kể cả file hoàn toàn sạch (không có bug, ví dụ file migration SQL chỉ thêm index). Việc bắt model giải thích dài dòng "tại sao file này không có vấn đề gì" là lãng phí token thuần túy — không mang lại giá trị.

Early termination (dừng sớm) ở đây áp dụng tại 2 cấp:

Cấp 1 — early exit trong structured output: schema đã ép model trả clean: true kèm findings: [] ngay khi không có gì để báo cáo, không cần viết thêm câu giải thích.

Cấp 2 — early exit ở cấp workflow: nếu một loại file được biết trước là rủi ro thấp (ví dụ file test, file migration đơn giản, file config), có thể dùng một pass rẻ hơn (model nhỏ hơn hoặc heuristic) để lọc trước, chỉ đưa file "khả năng có vấn đề" vào review đầy đủ bằng model lớn.

def quick_risk_screen(file: dict) -> bool:
    """
    Cheap heuristic pre-filter — runs BEFORE calling the LLM at all.
    Returns True if the file likely needs a full review.
    Uses a lightweight local check, not an LLM call, to avoid spending
    any tokens on files that are almost certainly low-risk.
    """
    low_risk_patterns = ("test_", "_test.py", ".sql", "README", ".md")
    if any(file["path"].endswith(p) or file["path"].split("/")[-1].startswith(p)
           for p in low_risk_patterns):
        return False  # skip full review — route to lightweight pass
    return True

def early_terminating_review(file: dict, system_prompt: str) -> FileReviewResult:
    if not quick_risk_screen(file):
        # Lightweight review: a much shorter prompt, no full checklist needed
        return FileReviewResult(path=file["path"], findings=[], clean=True)
    return review_single_file_cached(file, system_prompt)

Cần lưu ý một rủi ro rõ ràng: quick_risk_screen là heuristic, và heuristic có thể sai — file test cũng có thể chứa bug thật (ví dụ test bị viết sai logic dẫn đến false positive/negative). Vì vậy trong bài lab này, ta chỉ áp dụng early termination cấp workflow cho các loại file có rủi ro thực sự thấp và đã được team đồng thuận (ví dụ file migration SQL đơn giản, file markdown), không áp dụng bừa cho mọi loại file "nhìn có vẻ an toàn".

Với bộ fixture 6 file, tests/test_login_handler.py và src/db/migrations/0042_add_index.sql là 2 file rơi vào early exit path — tiết kiệm được 2/7 lời gọi LLM đầy đủ. Tính trung bình trên nhiều loại PR khác nhau (một số PR có nhiều file test/config hơn, một số ít hơn), mức giảm trung bình đạt khoảng -3% so với mốc sau Optimization 4.

Mẹo

• Early termination hiệu quả nhất khi kết hợp với dữ liệu thống kê thật — trước khi quyết định loại file nào được "miễn review đầy đủ", hãy phân tích lịch sử bug đã từng lọt qua review trong 6-12 tháng gần nhất theo loại file, đừng chỉ dựa vào cảm tính.
• Luôn có escape hatch (lối thoát) cho phép reviewer con người yêu cầu "review đầy đủ file này dù nó rơi vào nhóm low-risk" — early termination không nên là quyết định không thể đảo ngược.
• Log lại rõ ràng file nào bị skip và vì lý do gì (clean=True, reason="pre-filtered" khác với clean=True, reason="llm_reviewed_no_issues") — nếu không phân biệt, sau này khó audit khi có bug lọt lưới.

Optimization 6: Parallel Execution of Independent Reviews (Latency Reduction)

Năm optimization trước tập trung vào giảm số token. Optimization thứ 6 khác biệt: nó không giảm token (thậm chí có thể giữ nguyên tổng token), mà giảm latency (độ trễ) — thời gian chờ đến khi có kết quả cuối cùng. Đây là điểm quan trọng cần làm rõ với các bên liên quan (đặc biệt Product Manager): token cost và latency là hai trục tối ưu khác nhau, không phải lúc nào giảm cái này cũng giảm cái kia.

Nhờ Optimization 2 (scoped sessions), mỗi file review giờ đã hoàn toàn độc lập — không còn phụ thuộc lẫn nhau qua history. Điều này mở ra khả năng chạy các review song song (parallel) thay vì tuần tự.

import asyncio

async def review_single_file_async(file: dict, system_prompt: str) -> FileReviewResult:
    if not quick_risk_screen(file):
        return FileReviewResult(path=file["path"], findings=[], clean=True)
    cached = get_cached_review(file["diff_content"])
    if cached is not None:
        return cached
    result = await structured_llm.ainvoke([
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": f"File: {file['path']}\nDiff:\n{file['diff_content']}"},
    ])
    set_cached_review(file["diff_content"], result)
    return result

async def parallel_review_loop(pr_files: list, system_prompt: str) -> list:
    # All independent file reviews fire concurrently instead of one-by-one.
    # Token count is roughly unchanged — this optimizes wall-clock time, not tokens.
    tasks = [review_single_file_async(f, system_prompt) for f in pr_files]
    return await asyncio.gather(*tasks)

Với 6 file, baseline tuần tự tốn khoảng 42 giây (mỗi lời gọi LLM ~6-7 giây). Chạy song song với asyncio.gather, thời gian tổng gần bằng thời gian của lời gọi chậm nhất, không phải tổng của tất cả — trong thực nghiệm, latency giảm từ 42 giây xuống còn khoảng 8-10 giây (giới hạn bởi rate limit của API và file có diff dài nhất).

Cần lưu ý đánh đổi: chạy song song làm tăng peak concurrency (số request đồng thời) tới API — nếu có rate limit chặt, cần thêm semaphore giới hạn số request đồng thời:

async def parallel_review_loop_bounded(pr_files: list, system_prompt: str, max_concurrent: int = 4) -> list:
    semaphore = asyncio.Semaphore(max_concurrent)

    async def bounded_review(file):
        async with semaphore:
            return await review_single_file_async(file, system_prompt)

    tasks = [bounded_review(f) for f in pr_files]
    return await asyncio.gather(*tasks)

Mẹo

• Không nhầm lẫn "giảm token" với "giảm latency" khi báo cáo kết quả optimization cho stakeholder — Optimization 6 gần như không đổi tổng token (có thể tăng nhẹ do overhead của mỗi request riêng lẻ), nhưng cải thiện latency rất đáng kể, đây là giá trị khác cần được ghi nhận riêng.
• Luôn giới hạn concurrency (max_concurrent) theo rate limit thực tế của API key — chạy song song không kiểm soát dễ dẫn đến lỗi 429 (Too Many Requests) và làm workflow thất bại hoàn toàn thay vì chỉ chậm.
• Parallel execution chỉ an toàn khi các task thực sự độc lập (đã đảm bảo từ Optimization 2) — nếu có bất kỳ dependency ngầm nào giữa các file (ví dụ thứ tự review ảnh hưởng đến logic nào đó), phải giải quyết dependency trước khi parallelize, không thể "vá" bằng cách chạy song song rồi hy vọng đúng.

Putting It All Together: The Fully Optimized Workflow

Giờ ghép toàn bộ 6 optimization vào một workflow hoàn chỉnh, thay thế baseline_agent.py ban đầu:

import asyncio
from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class OptimizedReviewState(TypedDict):
    pr_files: List[dict]
    file_results: List[FileReviewResult]
    handoff_text: str
    final_report: str

async def optimized_review_node(state: OptimizedReviewState) -> OptimizedReviewState:
    # Opt 2 (scoped) + Opt 4 (cache) + Opt 5 (early exit) + Opt 6 (parallel)
    # all converge inside this single node.
    results = await parallel_review_loop_bounded(
        state["pr_files"], OPTIMIZED_SYSTEM_PROMPT, max_concurrent=4
    )
    state["file_results"] = results
    return state

def compress_handoff_node(state: OptimizedReviewState) -> OptimizedReviewState:
    # Opt 3: compressed handoff — clean files contribute nothing
    state["handoff_text"] = build_compressed_handoff(state["file_results"])
    return state

def summarize_node(state: OptimizedReviewState) -> OptimizedReviewState:
    # Opt 1: focused prompt, Opt 3: compact input, no accumulated history at all
    if not state["handoff_text"].strip():
        state["final_report"] = "No issues found across all reviewed files."
        return state
    state["final_report"] = summarize_from_handoff(
        state["handoff_text"], OPTIMIZED_SYSTEM_PROMPT
    )
    return state

graph = StateGraph(OptimizedReviewState)
graph.add_node("review", optimized_review_node)
graph.add_node("compress", compress_handoff_node)
graph.add_node("summarize", summarize_node)
graph.add_edge("review", "compress")
graph.add_edge("compress", "summarize")
graph.add_edge("summarize", END)
graph.set_entry_point("review")
optimized_app = graph.compile()

async def run_optimized(initial_state):
    return await optimized_app.ainvoke(initial_state)

So với baseline_agent.py, cấu trúc graph thực chất đơn giản hơn, không phức tạp hơn: 3 node rõ ràng (review → compress → summarize), không còn khối history cồng kềnh đi qua từng node. Đây là điểm hay thường bị bỏ qua khi nói về tối ưu hóa: tối ưu token tốt thường đi kèm với code dễ đọc hơn, không phải khó đọc hơn — vì nó buộc bạn loại bỏ trạng thái ngầm (implicit state) không cần thiết.

Mẹo

• Khi ghép nhiều optimization, test lại toàn bộ integration test (không chỉ unit test từng optimization riêng) — có optimization tương tác bất ngờ với optimization khác (ví dụ cache key cần bao gồm cả effect của early-exit heuristic, nếu không hai file khác nội dung nhưng cùng bị early-exit có thể collide sai — thực ra ví dụ này an toàn vì early-exit không qua cache path, nhưng đây là loại lỗi cần soát kỹ).
• Giữ mỗi optimization sau một feature flag riêng trong giai đoạn đầu triển khai production — cho phép rollback từng phần nếu một optimization gây vấn đề chất lượng, không phải rollback toàn bộ workflow.
• Viết lại toàn bộ docstring/comment giải thích rõ optimization nào tương ứng với đoạn code nào (như ví dụ trên) — 6 tháng sau, người maintain code (có thể chính là bạn) sẽ cảm ơn vì không phải suy luận lại từ đầu.

Final Measurement: Comparing Baseline vs. Optimized

Chạy benchmark cuối cùng trên đúng bộ fixture 6 file, so sánh trực tiếp với baseline ở Step 0:

Con số cuối, làm tròn thận trọng, đạt mức giảm ~40-45% tổng token so với baseline — đạt và vượt nhẹ target 40% đề ra ban đầu, với biên an toàn hợp lý cho các PR có cấu trúc khác đi ít nhiều so với fixture mẫu.

Chi phí tính bằng USD (giả định giá $3/1M input token, $15/1M output token — mức giá tham khảo cho model tier cao cấp tại thời điểm viết bài, cần thay bằng giá thực tế theo provider bạn dùng):

Với team chạy trung bình 200 lần review PR/ngày (con số thực tế cho một tổ chức engineering cỡ vừa), mức tiết kiệm này tương đương khoảng $15,8/ngày, hay ~$474/tháng — không phải số khổng lồ cho một team, nhưng đủ để minh chứng rõ ràng cho việc scale optimization pattern này ra toàn bộ các agent khác trong hệ thống, nơi hiệu ứng cộng gộp sẽ lớn hơn nhiều.

Latency: từ 42 giây (baseline, tuần tự) xuống khoảng 8-10 giây (optimized, song song) — cải thiện độ trễ ~75-80%, một trục giá trị hoàn toàn tách biệt với tiết kiệm chi phí, như đã nói ở Optimization 6.

Mẹo

• Luôn trình bày cả 2 loại số liệu — % giảm và số USD tuyệt đối — khi báo cáo kết quả optimization. % giảm ấn tượng (45%) nhưng số USD tuyệt đối nhỏ (một vài trăm đô/tháng) có thể không đủ thuyết phục để đầu tư thêm thời gian, trừ khi bạn cho thấy rõ hiệu ứng scale khi áp dụng cho nhiều agent khác.
• Đo lại chất lượng review (không chỉ token) sau khi optimize toàn bộ — chạy lại trên bộ fixture có has_issue=True đã biết trước, xác nhận agent tối ưu vẫn phát hiện đúng 3/3 file có vấn đề như baseline đã phát hiện.
• Giữ lại báo cáo benchmark của từng bước (không chỉ số cuối) — khi cần debug "tại sao token tăng trở lại" trong tương lai (ví dụ do prompt bị sửa lại, hoặc cache bị vô hiệu do đổi hạ tầng), có breakdown theo từng optimization sẽ giúp khoanh vùng vấn đề nhanh hơn nhiều.

Exercise Variations for Different Personas

Bài lab trên là khung chuẩn — mỗi vai trò trong team nên tùy biến bài tập theo mục tiêu công việc của mình.

For Software Engineers

• Tự viết thêm Optimization 7: thử nghiệm dùng một model nhỏ hơn/rẻ hơn (ví dụ Claude Haiku) cho bước quick_risk_screen và cho các file có clean=True dự kiến, chỉ dùng model lớn cho file có khả năng cao có vấn đề — đo xem việc "phân tầng model" (model tiering) này giảm thêm được bao nhiêu % so với việc dùng một model duy nhất cho mọi bước.
• Thử đo lại toàn bộ pipeline khi PR có 20 file thay đổi thay vì 6 — kiểm tra xem % giảm token có còn giữ nguyên ~40% hay tăng lên (giả thuyết: PR lớn hơn thường có % giảm cao hơn vì hiệu ứng history tích lũy ở baseline sẽ nghiêm trọng hơn theo cấp số nhân).
• Refactor optimized_workflow.py để dùng LangGraph's Send API cho map-reduce pattern chính thức thay vì asyncio.gather thủ công, rồi so sánh độ phức tạp code và khả năng quan sát (observability) giữa hai cách triển khai.

For QA Engineers

• Thiết kế một bộ test case chuyên biệt để xác minh early termination (Optimization 5) không gây false negative — cụ thể, tạo một file test giả (test_*.py) có chứa bug logic thật, chạy qua pipeline optimized, và xác nhận có nên gắn cờ cảnh báo (hoặc điều chỉnh heuristic) nếu agent bỏ sót nó do bị early-exit.
• Viết test đo độ ổn định (consistency) của structured output qua nhiều lần chạy — với cùng một diff input, chạy 10 lần, đo tỷ lệ schema validation pass và độ lệch giữa các finding được báo cáo (do temperature > 0 hoặc do model version thay đổi).
• Xây dựng bộ regression test riêng cho cache: xác minh cache trả đúng kết quả cũ khi diff content giống, và cache không trả nhầm kết quả cũ khi diff content chỉ thay đổi 1 dòng nhỏ (kiểm tra content hash có nhạy đủ để phân biệt các thay đổi nhỏ).

For Product Managers

• Xây một dashboard đơn giản (có thể dùng spreadsheet) theo dõi 3 chỉ số theo thời gian cho agent review PR: tổng token/tháng, chi phí USD/tháng, và tỷ lệ file được early-exit — dùng dữ liệu này để quyết định khi nào cần đầu tư thêm optimization và khi nào chi phí đã đủ thấp để chuyển ưu tiên sang latency hoặc chất lượng.
• Thu thập phản hồi từ đội ngũ engineer thực tế dùng bot review này: liệu báo cáo "terse" (cô đọng, do Optimization 1) có làm giảm giá trị học hỏi cho junior engineer so với báo cáo "narrative" chi tiết của baseline hay không — đây là trade-off giữa chi phí và giá trị đào tạo, cần quyết định ở tầm sản phẩm, không phải tầm kỹ thuật đơn thuần.
• Định nghĩa rõ SLA (thỏa thuận mức dịch vụ) mong muốn cho latency của bot review (ví dụ "phải trả kết quả trong 15 giây để không làm chậm CI pipeline") và dùng số liệu Optimization 6 để xác nhận optimized workflow có đáp ứng SLA đó hay chưa, trước khi rollout ra toàn bộ repository của công ty.

Mẹo

• Ba persona trên nên chạy bài lab cùng trên một bộ fixture rồi họp lại so sánh góc nhìn — kinh nghiệm thực tế cho thấy Software Engineer thường tối ưu quá tay về token mà bỏ quên tác động tới trải nghiệm người đọc báo cáo, còn Product Manager thường chỉ nhìn số USD mà bỏ quên rủi ro chất lượng — làm chung sẽ cân bằng được cả ba góc nhìn.
• Nếu team bạn không có người chuyên trách QA cho agentic system, ít nhất Software Engineer nên tự đóng vai QA cho phần "kiểm tra early termination không gây false negative" — đây là rủi ro dễ bị bỏ qua nhất trong toàn bộ 6 optimization.

Troubleshooting Common Issues in the Optimization Process

Trong quá trình làm lại bài lab này (hoặc áp dụng vào workflow thật của bạn), một số vấn đề thường gặp:

Token giảm nhưng chất lượng review giảm theo (false negative tăng)
Nguyên nhân phổ biến nhất: system prompt bị cắt quá tay ở Optimization 1, loại bỏ luôn cả phần checklist quan trọng (ví dụ bỏ mất "security issues" khỏi checklist khi rút gọn). Cách khắc phục: giữ lại đủ hạng mục kiểm tra (categories), chỉ cắt phần giải thích (explanations) — đây là sự khác biệt cốt lõi giữa "cắt nội dung" và "cắt văn phong".

Scoped session (Optimization 2) làm mất phát hiện lỗi xuyên file
Ví dụ: file A định nghĩa hàm, file B gọi hàm đó sai tham số — baseline (có full history) có thể phát hiện, scoped session độc lập từng file thì không. Khắc phục: thêm một node review riêng ở cấp "toàn PR" chạy sau vòng scoped loop, chỉ nhận input là danh sách tên hàm/class được thay đổi (không phải toàn bộ diff), để giữ chi phí thấp mà vẫn bắt được lỗi xuyên file ở mức tối thiểu cần thiết.

Cache trả kết quả cũ dù đã thay đổi system prompt
Đây là lỗi cực kỳ dễ mắc: content hash chỉ dựa trên diff, không bao gồm version của system prompt hay model. Khắc phục: luôn ghép hash(diff_content + prompt_version + model_name) làm cache key, không chỉ hash riêng diff content.

Structured output liên tục fail validation (schema error)
Thường do model trả về text lẫn cả markdown code fence (json ...) hoặc thêm ký tự thừa quanh JSON. Khắc phục: dùng đúng API structured output/tool-calling chính thức của provider (như with_structured_output trong ví dụ) thay vì tự parse JSON từ raw text — API chính thức đảm bảo model tuân thủ schema chặt hơn nhiều so với prompt-based JSON.

Parallel execution (Optimization 6) gây lỗi rate limit hàng loạt
Chạy asyncio.gather không giới hạn concurrency trên PR có 30-40 file dễ dàng đập thẳng vào rate limit của API. Khắc phục: luôn dùng Semaphore như minh họa ở Optimization 6, và thêm retry với exponential backoff (thời gian chờ tăng dần) cho các request bị 429.

Benchmark trước/sau không nhất quán, kết luận optimization sai
Nguyên nhân hay gặp nhất: quên cố định fixture, hoặc benchmark baseline và optimized ở hai thời điểm cách nhau (model provider có thể âm thầm update model phía sau, thay đổi hành vi output). Khắc phục: luôn chạy A/B trong cùng một buổi, cùng model version, cùng fixture, và chạy nhiều lần lấy trung bình như đã nêu ở Step 0.

Mẹo

• Khi gặp bug lạ trong optimized workflow mà baseline không có, luôn nghĩ đến câu hỏi "optimization nào mới thêm state ngầm (hash, cache key, schema) mà mình chưa kiểm chứng đầy đủ" — hầu hết bug trong bài lab loại này nằm ở lớp state quản lý optimization, không nằm ở logic nghiệp vụ review.
• Giữ một "kill switch" đơn giản (biến môi trường DISABLE_ALL_OPTIMIZATIONS=true) để có thể quay lại chạy baseline logic ngay lập tức khi nghi ngờ optimization gây lỗi production — không cần revert code, chỉ cần đổi flag.

Summary

Bài lab này đã đi qua toàn bộ quy trình tối ưu token cho một agentic workflow thực tế, từ đo baseline (34.500 token, 1 lần review PR 6 file) đến workflow đã tối ưu (~19.000 token, giảm ~45%), thông qua 6 optimization độc lập nhưng cộng hưởng lẫn nhau: focused system prompt, scoped sessions, structured/compressed handoff, tool result caching, early termination, và parallel execution.

Bài học quan trọng nhất xuyên suốt: tối ưu token không phải là một kỹ thuật đơn lẻ, mà là một quy trình có phương pháp — đo trước, đổi một biến, đo lại, xác nhận chất lượng không giảm, rồi mới chuyển sang biến tiếp theo. Áp dụng ngẫu nhiên nhiều optimization cùng lúc mà không đo từng bước sẽ khiến bạn không biết cái nào thực sự hiệu quả, và không biết cái nào (nếu có) đang âm thầm làm giảm chất lượng output.

Điểm thứ hai cần nhớ: không phải mọi optimization đều nhằm giảm token — Optimization 6 (parallel execution) là minh chứng rằng latency và cost là hai trục giá trị riêng biệt, và một agentic system trưởng thành cần được đo lường, báo cáo, và tối ưu trên cả hai trục, tùy theo ưu tiên của từng giai đoạn sản phẩm.

Với pattern đã thực hành trong bài lab này — baseline benchmark, focused prompt, scoped session, structured handoff, caching, early exit, parallel execution — bạn đã có đủ bộ công cụ để áp dụng vào bất kỳ agentic workflow nào khác trong tổ chức của mình, không chỉ riêng code review agent.