Đây là bài lab dài và nặng code nhất trong module này. Bạn sẽ tự tay xây một hệ thống compression (nén ngữ cảnh) hoàn chỉnh — từ data model, compression engine, context manager, đến agent loop — áp dụng đúng kiến trúc 3-tier (3 tầng) đã học ở các bài trước, cho một session coding agentic (agent tự động viết code) kéo dài nhiều giờ đồng hồ.

Ở các bài trước trong module này, bạn đã học lý do tại sao cần compress context (ngữ cảnh), cách summarization (tóm tắt) hội thoại, kỹ thuật incremental summarization (tóm tắt tăng dần), diff summarization (tóm tắt bản khác biệt code), và mô hình phân tầng hierarchical context management (quản lý ngữ cảnh theo tầng). Bài này không giới thiệu khái niệm mới — nó là nơi bạn lắp ráp tất cả những khái niệm đó thành một hệ thống Python chạy được, với đầy đủ dataclass (kiểu class đơn giản chuyên dùng để lưu trạng thái), lời gọi API thật, và số liệu benchmark cụ thể.

Bối cảnh giả định: bạn đang xây một agent coding tự động chạy trong 6-8 giờ để phát triển một microservice từ đầu đến cuối — tạo model, viết endpoint, sửa bug, refactor, viết test, tối ưu hiệu năng. Không có compression, transcript (bản ghi hội thoại) của session này sẽ phình to vượt quá context window (cửa sổ ngữ cảnh — giới hạn số token mà model có thể "nhìn thấy" trong một lần gọi) chỉ sau vài chục turn. Mục tiêu của bài lab: giữ cho mỗi lần gọi model chỉ tốn một phần nhỏ token cần thiết, mà vẫn không làm agent "quên" quyết định quan trọng đã đưa ra 3 giờ trước.

Chuẩn Bị Môi Trường Và Các Yêu Cầu Tiên Quyết

Trước khi viết dòng code đầu tiên, bạn cần chuẩn bị:

• Python 3.10+ — code trong bài dùng cú pháp match kiểu union mới (list[Turn], X | Y) và module ast để phân tích code Python.
• Một API key của Anthropic — đăng ký tại Anthropic Console, xuất ra biến môi trường ANTHROPIC_API_KEY. Toàn bộ code trong bài dùng SDK chính thức, không dùng raw HTTP.
• Package anthropic — SDK Python chính thức để gọi Claude API.
• Một sample repo để thử nghiệm Bước 7 (code skeleton injection) — có thể dùng bất kỳ project Python nhỏ nào bạn có sẵn, hoặc tạo vài file .py giả để test.
• Editor/agent tooling khuyến nghị: VS Code hoặc bất kỳ editor hỗ trợ Python tốt; nếu bạn đang dùng Claude Code hoặc một coding agent tương tự để viết hệ thống này, đây cũng là dịp tốt để quan sát cách agent đó tự quản lý context của chính nó.

Cài đặt package cần thiết:

python3 -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

pip install anthropic python-dotenv

Tạo file .env ở root project:

ANTHROPIC_API_KEY=sk-ant-your-key-here

Cấu trúc project mà bài lab này sẽ xây dựng — 5 file, mỗi file tương ứng với một bước:

context-compression-lab/
├── .env
├── session_state.py        # Bước 1 — data model
├── compression_engine.py   # Bước 2 — logic compression
├── context_manager.py      # Bước 3 — điều phối compression + assembly
├── agent.py                # Bước 4 — agent loop chính
└── run_session.py          # Bước 5 — script chạy toàn bộ session

Xác nhận môi trường hoạt động với một script kiểm tra nhanh:

import os
from dotenv import load_dotenv
import anthropic

load_dotenv()
assert os.environ.get("ANTHROPIC_API_KEY"), "Missing ANTHROPIC_API_KEY in .env"

client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=32,
    messages=[{"role": "user", "content": "Reply with exactly: OK"}],
)
print(response.content[0].text)

Nếu script in ra OK, môi trường đã sẵn sàng.

Mẹo:
- Đừng hardcode API key trong code — luôn dùng biến môi trường, và thêm .env vào .gitignore ngay từ commit đầu tiên.
- Nếu bạn định chạy bài lab này nhiều lần để thử nghiệm threshold (ngưỡng) khác nhau, cân nhắc dùng model rẻ hơn (claude-haiku-4-5-20251001) cho các lần chạy thử để tiết kiệm chi phí, rồi chuyển sang model production khi đã chốt tham số.
- Cài thêm pytest (pip install pytest) ngay từ đầu — Bước 6 sẽ dùng nó để viết test kiểm chứng compression.

Bước 1: Định Nghĩa Session State Model

Nhắc lại nhanh kiến trúc 3-tier đã học: Tier 1 giữ các turn (lượt trao đổi) thô gần nhất, độ trung thực cao nhất nhưng tốn token nhất; Tier 2 là rolling summary (tóm tắt cuộn — bản tóm tắt được viết lại liên tục, gộp cái mới vào cái cũ) của các turn đã bị đẩy ra khỏi Tier 1; Tier 3 là project archive (bản lưu trữ dự án lâu dài) — bản chưng cất tối giản nhất, sống suốt đời session (và có thể lâu hơn, nếu bạn lưu ra đĩa).

Việc đầu tiên là mô hình hóa trạng thái này bằng dataclass. Đây là session_state.py:

"""
session_state.py

Defines the data model for tracking an agentic coding session's context
across the three-tier hierarchy: Tier 1 (raw recent turns), Tier 2 (rolling
summary), and Tier 3 (project archive).
"""

from __future__ import annotations

import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class Role(str, Enum):
    """Who produced a given turn in the session transcript."""
    USER = "user"
    ASSISTANT = "assistant"
    TOOL = "tool"


@dataclass
class Turn:
    """A single raw turn in the conversation (Tier 1 material).

    `token_count` is an *approximate* count (see `estimate_tokens` in
    compression_engine.py) used for fast, in-process budget tracking. It is
    NOT a substitute for `client.messages.count_tokens(...)` when you need
    an authoritative number (e.g. right before hitting a hard context-window
    ceiling).
    """

    turn_id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
    role: Role = Role.USER
    content: str = ""
    tool_name: Optional[str] = None
    created_at: float = field(default_factory=time.time)
    token_count: int = 0

    def to_prompt_line(self) -> str:
        """Render this turn as a single line for a compression prompt."""
        prefix = f"[{self.role.value}]"
        if self.tool_name:
            prefix += f" (tool: {self.tool_name})"
        return f"{prefix} {self.content}"


@dataclass
class TierOneState:
    """Tier 1 -- raw, unsummarized turns. This is the most expensive tier
    per token because it carries full fidelity, so it is kept small."""

    turns: list[Turn] = field(default_factory=list)

    @property
    def token_count(self) -> int:
        return sum(t.token_count for t in self.turns)

    def append(self, turn: Turn) -> None:
        self.turns.append(turn)

    def drain(self) -> list[Turn]:
        """Remove and return all turns, leaving Tier 1 empty. Called right
        before a Tier 1 -> Tier 2 compression pass."""
        drained, self.turns = self.turns, []
        return drained


@dataclass
class TierTwoState:
    """Tier 2 -- the rolling summary. Rewritten (not appended to) on every
    Tier 1 -> Tier 2 compression pass, folding in whatever the previous
    rolling summary already said."""

    summary_text: str = ""
    turns_covered: int = 0
    token_count: int = 0

    def replace(self, summary_text: str, turns_covered: int, token_count: int) -> None:
        self.summary_text = summary_text
        self.turns_covered = turns_covered
        self.token_count = token_count


@dataclass
class TierThreeState:
    """Tier 3 -- the durable project archive. Survives for the entire
    lifetime of the coding session (and, if persisted to disk, across
    sessions on the same project)."""

    archive_text: str = ""
    project_name: str = "untitled-project"
    last_distilled_turn: int = 0
    token_count: int = 0

    def replace(self, archive_text: str, last_distilled_turn: int, token_count: int) -> None:
        self.archive_text = archive_text
        self.last_distilled_turn = last_distilled_turn
        self.token_count = token_count


@dataclass
class CompressionThresholds:
    """Token budgets that decide when each compression stage fires.

    These numbers are deliberately conservative for a 200K-token context
    window model; tune them per model and per how much headroom you want to
    leave for tool outputs and the model's own response.
    """

    tier1_max_tokens: int = 6_000       # trigger Tier 1 -> Tier 2
    tier2_max_tokens: int = 3_000       # trigger Tier 2 -> Tier 3
    tier1_max_turns: int = 20           # secondary trigger, turn-count based


@dataclass
class SessionState:
    """The full state of one agentic coding session."""

    project_name: str
    tier1: TierOneState = field(default_factory=TierOneState)
    tier2: TierTwoState = field(default_factory=TierTwoState)
    tier3: TierThreeState = field(default_factory=TierThreeState)
    thresholds: CompressionThresholds = field(default_factory=CompressionThresholds)
    turn_count: int = 0
    compression_events: list[str] = field(default_factory=list)

    def log_compression_event(self, message: str) -> None:
        self.compression_events.append(f"[turn {self.turn_count}] {message}")

Vài điểm thiết kế đáng chú ý:

• TierOneState.drain() xóa sạch Tier 1 và trả về các turn đã xóa — đây là bước bắt buộc trước khi gọi compression, để tránh trường hợp cùng một turn bị đưa vào rolling summary hai lần.
• TierTwoState.replace() thay thế toàn bộ, không append — vì rolling summary luôn được viết lại từ đầu mỗi lần compress, gộp bản cũ + turn mới, không phải cộng dồn text.
• CompressionThresholds tách riêng thành dataclass của chính nó để bạn dễ tinh chỉnh theo model (model có context window lớn hơn thì có thể nới các ngưỡng này).
• compression_events là một log đơn giản, dùng để kiểm chứng ở Bước 6 rằng compression thực sự đã chạy, chạy đúng lúc.

Mẹo:
- Giữ token_count là field độc lập trên Turn, đừng tính lại (recompute) mỗi lần cần — với hàng trăm turn trong một session dài, việc tính lại liên tục sẽ chậm và không cần thiết.
- Đặt thresholds là tham số có thể inject (như trong SessionState) chứ không hardcode trong logic compression — điều này giúp bạn viết unit test với ngưỡng nhỏ (ví dụ 200 token) để trigger compression ngay lập tức mà không cần chạy session dài.
- Nếu bạn cần persist (lưu) session giữa các lần chạy, SessionState là dataclass thuần nên serialize bằng dataclasses.asdict() + json.dumps() gần như miễn phí — không cần thêm logic serialize riêng.

Bước 2: Xây Dựng Compression Engine

CompressionEngine chịu trách nhiệm cho phần "nặng" nhất: gọi LLM (large language model — mô hình ngôn ngữ lớn) để thực hiện compression thật. Có 4 nhiệm vụ:

1. Tier 1 → Tier 2: gộp một batch turn thô + rolling summary cũ thành rolling summary mới, theo format "Rolling Summary — Turn {turn_count}".
2. Tier 2 → Tier 3: chưng cất rolling summary vào project archive, theo format "Project Archive: {project_name}" với 8 section cố định: Project Identity, Architectural Decisions, Completed Milestones, Current State, Known Issues, Constraints, Patterns & Standards, Do Not Revisit.
3. Code skeleton generation — một utility (hàm tiện ích) không gọi LLM, dùng module ast để trích signature (chữ ký hàm/class) và docstring của một file Python mà không cần đọc toàn bộ nội dung.
4. Diff summarization — tóm tắt một git diff (bản khác biệt code) thành 2-4 câu.

"""
compression_engine.py

Implements the actual compression logic: Tier 1 -> Tier 2 (incremental
rolling summarization), Tier 2 -> Tier 3 (archive distillation), plus two
utility functions used elsewhere in the agent loop: code skeleton
generation and diff summarization.
"""

from __future__ import annotations

import ast

import anthropic

from session_state import Turn

COMPRESSION_MODEL = "claude-sonnet-5"


def estimate_tokens(text: str) -> int:
    """Cheap, local token estimate for hot-path budget tracking.

    This is NOT the authoritative Claude tokenizer count -- it is a rough
    chars-per-token heuristic used so we don't have to make a network call
    on every single turn just to decide whether we're near a threshold.
    Call `client.messages.count_tokens(...)` at checkpoints (e.g. right
    before triggering a compression pass) if you need an exact number.
    """
    return max(1, len(text) // 4)


TIER1_TO_TIER2_PROMPT = """You are maintaining a rolling summary of an ongoing \
software engineering session between a human developer and an AI coding agent.

You will be given:
1. The PREVIOUS rolling summary (may be empty if this is the first compression pass).
2. A batch of RAW TURNS that happened since that summary was last updated.

Produce an UPDATED rolling summary that merges the previous summary with the
new turns. Do not simply append -- synthesize. Preserve every fact that
still matters (decisions made, files touched, bugs found and fixed, open
questions) and drop turn-by-turn narration that has no lasting relevance
(e.g. "I'll check that file now", exploratory dead ends that were abandoned).

Output format -- follow this exactly:

Rolling Summary — Turn {turn_count}
<3-8 dense paragraphs or a tight bulleted list covering what has happened
so far in this session, organized by topic, not chronologically>

PREVIOUS ROLLING SUMMARY:
{previous_summary}

RAW TURNS TO FOLD IN:
{raw_turns}
"""

TIER2_TO_TIER3_PROMPT = """You are distilling a rolling summary of a software \
engineering session into a permanent PROJECT ARCHIVE entry. The project
archive is the long-term memory for this codebase -- it must remain
accurate and useful even months from now, long after the rolling summary
that produced it has been discarded.

You will be given the EXISTING project archive (may be empty) and the
CURRENT rolling summary that needs to be folded into it.

Merge them into an updated archive using EXACTLY this structure:

Project Archive: {project_name}

## Project Identity
<what this project is, its purpose, primary stack>

## Architectural Decisions
<key decisions and the reasoning behind them, one bullet per decision>

## Completed Milestones
<features/fixes that are done and verified, most recent first>

## Current State
<what's in progress right now, as of this distillation>

## Known Issues
<bugs, tech debt, or limitations that are known but not yet fixed>

## Constraints
<hard constraints: performance budgets, compatibility requirements, etc.>

## Patterns & Standards
<coding conventions and patterns established in this session that future \
turns must follow>

## Do Not Revisit
<approaches that were tried and explicitly rejected, so the agent doesn't \
waste tokens re-exploring them>

EXISTING PROJECT ARCHIVE:
{existing_archive}

CURRENT ROLLING SUMMARY TO FOLD IN:
{rolling_summary}
"""

DIFF_SUMMARY_PROMPT = """Summarize the following git diff in 2-4 sentences. \
State what changed and why it plausibly changed (infer intent from the \
change itself and any commit message given). Do not restate the diff \
verbatim, and do not include a line-by-line walkthrough.

COMMIT MESSAGE (if any): {commit_message}

DIFF:
{diff_text}
"""


class CompressionEngine:
    """Wraps the Claude API calls that implement each compression stage."""

    def __init__(self, client: anthropic.Anthropic, model: str = COMPRESSION_MODEL):
        self.client = client
        self.model = model

    def compress_tier1_to_tier2(
        self,
        turns: list[Turn],
        previous_summary: str,
        turn_count: int,
    ) -> tuple[str, int]:
        """Fold a batch of raw Tier 1 turns into an updated rolling summary.

        Returns (summary_text, estimated_token_count).
        """
        raw_turns_text = "\n".join(t.to_prompt_line() for t in turns)
        prompt = TIER1_TO_TIER2_PROMPT.format(
            turn_count=turn_count,
            previous_summary=previous_summary or "(none yet -- this is the first pass)",
            raw_turns=raw_turns_text,
        )

        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )
        summary_text = next(
            (block.text for block in response.content if block.type == "text"), ""
        )
        return summary_text, estimate_tokens(summary_text)

    def compress_tier2_to_tier3(
        self,
        rolling_summary: str,
        existing_archive: str,
        project_name: str,
    ) -> tuple[str, int]:
        """Distill the rolling summary into the permanent project archive.

        Returns (archive_text, estimated_token_count).
        """
        prompt = TIER2_TO_TIER3_PROMPT.format(
            project_name=project_name,
            existing_archive=existing_archive or "(none yet -- this is the first distillation)",
            rolling_summary=rolling_summary,
        )

        response = self.client.messages.create(
            model=self.model,
            max_tokens=1536,
            messages=[{"role": "user", "content": prompt}],
        )
        archive_text = next(
            (block.text for block in response.content if block.type == "text"), ""
        )
        return archive_text, estimate_tokens(archive_text)

    def summarize_diff(self, diff_text: str, commit_message: str = "") -> str:
        """Produce a short natural-language summary of a git diff.

        Used when a tool call returns a large diff -- we don't want the raw
        diff to sit in Tier 1 turn-for-turn; a 2-4 sentence summary carries
        almost all the information at a fraction of the tokens.
        """
        prompt = DIFF_SUMMARY_PROMPT.format(
            commit_message=commit_message or "(none)",
            diff_text=diff_text,
        )
        response = self.client.messages.create(
            model=self.model,
            max_tokens=200,
            messages=[{"role": "user", "content": prompt}],
        )
        return next(
            (block.text for block in response.content if block.type == "text"), ""
        )


def generate_code_skeleton(file_path: str, source_code: str) -> str:
    """Produce a compact 'skeleton' of a Python file: module docstring,
    imports, and every class/function signature with its docstring, but
    NO implementation bodies.

    This is a pure static-analysis utility (ast module) -- no LLM call, no
    cost, essentially instant. It is the cheapest possible way to give the
    agent 'I know this file exists and roughly what's in it' context
    without paying for the full file contents on every turn.
    """
    try:
        tree = ast.parse(source_code)
    except SyntaxError:
        return f"# {file_path}\n# (skeleton unavailable -- file has a syntax error)\n"

    lines: list[str] = [f"# {file_path}"]

    module_doc = ast.get_docstring(tree)
    if module_doc:
        lines.append(f'"""{module_doc.strip()}"""')

    imports = [n for n in tree.body if isinstance(n, (ast.Import, ast.ImportFrom))]
    for node in imports:
        lines.append(ast.unparse(node))
    if imports:
        lines.append("")

    def render_function(node, indent: str = "") -> None:
        prefix = "async def" if isinstance(node, ast.AsyncFunctionDef) else "def"
        signature = ast.unparse(node.args)
        returns = f" -> {ast.unparse(node.returns)}" if node.returns else ""
        lines.append(f"{indent}{prefix} {node.name}({signature}){returns}: ...")
        doc = ast.get_docstring(node)
        if doc:
            first_line = doc.strip().splitlines()[0]
            lines.append(f'{indent}    """{first_line}"""')

    for node in tree.body:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            render_function(node)
            lines.append("")
        elif isinstance(node, ast.ClassDef):
            bases = ", ".join(ast.unparse(b) for b in node.bases)
            lines.append(f"class {node.name}({bases}):" if bases else f"class {node.name}:")
            class_doc = ast.get_docstring(node)
            if class_doc:
                lines.append(f'    """{class_doc.strip().splitlines()[0]}"""')
            for sub in node.body:
                if isinstance(sub, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    render_function(sub, indent="    ")
            lines.append("")

    return "\n".join(lines)

Ba quyết định thiết kế quan trọng cần giải thích:

• Prompt cho Tier 1 → Tier 2 yêu cầu "synthesize, không phải append" — đây là điểm khác biệt giữa incremental summarization đúng cách và một bug âm thầm cực kỳ phổ biến: nếu model chỉ nối rolling summary cũ với tóm tắt turn mới, summary sẽ phình to tuyến tính theo thời gian và bản thân nó sẽ cần được... compress. Prompt phải ép model viết lại bản tóm tắt, không phải thêm vào nó.
• Prompt cho Tier 2 → Tier 3 cố định cấu trúc 8 section — việc này quan trọng hơn nó nhìn có vẻ. Nếu không ép định dạng, mỗi lần distill archive sẽ có shape khác nhau, khiến model ở các turn sau khó tìm thông tin cụ thể (ví dụ: "constraint về performance là gì?" sẽ mất công tìm nếu archive không có section Constraints cố định).
• generate_code_skeleton không gọi LLM — đây là lựa chọn cố ý. Rất nhiều bài viết về compression bỏ qua chi tiết này: không phải mọi hình thức "nén" đều cần một lời gọi model. Trích signature bằng ast là miễn phí, tức thời, và chính xác 100% (không có rủi ro hallucination — model "tưởng tượng" ra thông tin sai).

Mẹo:
- Giữ max_tokens cho mỗi lời gọi compression thấp và cố định (1024 cho Tier 2, 1536 cho Tier 3) — bạn đang cố nén, nếu để max_tokens cao, model có xu hướng viết dài hơn cần thiết và tự phá hoại mục tiêu compression.
- Đừng dùng cùng một model cho compression và cho agent loop chính nếu chi phí là mối lo — compression call không cần khả năng coding mạnh, một model rẻ hơn (claude-haiku-4-5-20251001) thường đủ tốt cho việc tóm tắt, miễn là bạn kiểm tra chất lượng output.
- Log riêng lời gọi compression (input/output token, latency) — bạn sẽ cần số liệu này ở phần benchmark cuối bài, và nó cũng giúp bạn phát hiện sớm nếu compression đang tốn nhiều hơn dự kiến.

Bước 3: Xây Dựng ContextManager

ContextManager là "bộ não điều phối" — nó quyết định khi nào cần compress (dựa trên threshold token) và lắp ráp context cuối cùng gửi cho model ở mỗi turn: Tier 3 + Tier 2 + các turn Tier 1 gần nhất + task hiện tại.

"""
context_manager.py

The ContextManager decides WHEN to compress (based on token thresholds)
and assembles the final context payload sent to the model on every turn:
Tier 3 (archive) + Tier 2 (rolling summary) + recent Tier 1 turns + the
current task.
"""

from __future__ import annotations

from compression_engine import CompressionEngine, estimate_tokens
from session_state import Role, SessionState, Turn


class ContextManager:
    """Owns a SessionState and a CompressionEngine, and is the single
    place where 'should we compress right now?' gets decided."""

    def __init__(self, session: SessionState, engine: CompressionEngine):
        self.session = session
        self.engine = engine

    # ---- ingesting new turns ------------------------------------------------

    def add_turn(self, role: Role, content: str, tool_name: str | None = None) -> Turn:
        turn = Turn(
            role=role,
            content=content,
            tool_name=tool_name,
            token_count=estimate_tokens(content),
        )
        self.session.tier1.append(turn)
        if role == Role.USER:
            self.session.turn_count += 1
        self._maybe_compress()
        return turn

    # ---- compression triggers ------------------------------------------------

    def _maybe_compress(self) -> None:
        thresholds = self.session.thresholds
        tier1 = self.session.tier1

        over_token_budget = tier1.token_count >= thresholds.tier1_max_tokens
        over_turn_budget = len(tier1.turns) >= thresholds.tier1_max_turns

        if over_token_budget or over_turn_budget:
            self._compress_tier1()

        if self.session.tier2.token_count >= thresholds.tier2_max_tokens:
            self._compress_tier2()

    def _compress_tier1(self) -> None:
        drained_turns = self.session.tier1.drain()
        raw_tokens_before = sum(t.token_count for t in drained_turns)

        summary_text, token_count = self.engine.compress_tier1_to_tier2(
            turns=drained_turns,
            previous_summary=self.session.tier2.summary_text,
            turn_count=self.session.turn_count,
        )
        self.session.tier2.replace(
            summary_text=summary_text,
            turns_covered=self.session.tier2.turns_covered + len(drained_turns),
            token_count=token_count,
        )
        self.session.log_compression_event(
            f"Tier 1 -> Tier 2: folded {len(drained_turns)} turns "
            f"({raw_tokens_before} tokens) into a {token_count}-token rolling summary."
        )

    def _compress_tier2(self) -> None:
        archive_text, token_count = self.engine.compress_tier2_to_tier3(
            rolling_summary=self.session.tier2.summary_text,
            existing_archive=self.session.tier3.archive_text,
            project_name=self.session.project_name,
        )
        self.session.tier3.replace(
            archive_text=archive_text,
            last_distilled_turn=self.session.turn_count,
            token_count=token_count,
        )
        # The rolling summary has now been absorbed into the archive --
        # reset it so Tier 2 doesn't keep double-counting the same content.
        self.session.tier2.replace(summary_text="", turns_covered=0, token_count=0)
        self.session.log_compression_event(
            f"Tier 2 -> Tier 3: distilled rolling summary into project "
            f"archive ({token_count} tokens)."
        )

    # ---- context assembly ------------------------------------------------

    def assemble_context(self, current_task: str) -> tuple[list[dict], list[dict]]:
        """Build the (system_blocks, messages) pair to send to the model
        for the current turn.

        Order matters for prompt caching: Tier 3 (almost never changes) and
        Tier 2 (changes rarely) go in the SYSTEM prompt, marked cacheable.
        Tier 1 (changes every turn) and the current task go in `messages`,
        after the cache breakpoint, so they never invalidate the cache.
        """
        system_blocks: list[dict] = []

        if self.session.tier3.archive_text:
            system_blocks.append({
                "type": "text",
                "text": self.session.tier3.archive_text,
            })

        if self.session.tier2.summary_text:
            system_blocks.append({
                "type": "text",
                "text": self.session.tier2.summary_text,
            })

        if system_blocks:
            # Mark the LAST stable block as the cache breakpoint -- everything
            # before it (archive, and possibly the rolling summary) is cached too.
            system_blocks[-1]["cache_control"] = {"type": "ephemeral"}

        messages: list[dict] = []
        for turn in self.session.tier1.turns:
            messages.append({
                "role": "assistant" if turn.role == Role.ASSISTANT else "user",
                "content": turn.content,
            })

        messages.append({"role": "user", "content": current_task})

        return system_blocks, messages

    def get_stats(self) -> dict:
        return {
            "turn_count": self.session.turn_count,
            "tier1_turns": len(self.session.tier1.turns),
            "tier1_tokens": self.session.tier1.token_count,
            "tier2_tokens": self.session.tier2.token_count,
            "tier3_tokens": self.session.tier3.token_count,
            "total_context_tokens": (
                self.session.tier1.token_count
                + self.session.tier2.token_count
                + self.session.tier3.token_count
            ),
        }

Điểm quan trọng nhất trong file này là assemble_context. Thứ tự các block không phải ngẫu nhiên — nó tuân theo đúng cách Claude API render prompt: tools → system → messages. Tier 3 (gần như không đổi) và Tier 2 (đổi hiếm) nằm trong system, với cache_control (cơ chế cache prompt của Claude API) đặt ở block cuối cùng — nghĩa là mọi thứ trước nó cũng được cache theo. Tier 1 (đổi mỗi turn) và task hiện tại nằm trong messages, sau breakpoint (điểm cắt) cache, nên chúng không bao giờ làm invalidate (làm hỏng) cache.

Đây chính là lý do compression và prompt caching bổ trợ cho nhau: compression giảm tổng số token phải gửi, caching giảm chi phí của phần token ổn định (Tier 3 + Tier 2) mà vẫn phải gửi lại mỗi turn.

Mẹo:
- Đừng đặt cache_control lên cả system_blocks[-1] một cách vô điều kiện nếu system_blocks rỗng — code trên đã guard bằng if system_blocks:, nhưng nếu bạn refactor, hãy giữ nguyên guard này, tránh lỗi index-out-of-range.
- Nếu Tier 1 có nhiều hơn ~15-20 block trong một turn (ví dụ nhiều tool call liên tiếp), cache breakpoint ở cuối Tier 2/Tier 3 có thể "lạc" khỏi vùng nhìn ngược 20-block của cơ chế cache — cân nhắc đặt thêm một breakpoint phụ nếu bạn thấy cache_read_input_tokens bất ngờ về 0.
- get_stats() nên được gọi và log ở mọi turn trong lúc phát triển, không chỉ khi debug — nó gần như miễn phí và là cách nhanh nhất để phát hiện threshold đặt sai (ví dụ Tier 1 không bao giờ trigger compression vì ngưỡng quá cao).

Bước 4: Xây Dựng Agent Loop Chính

agent.py là nơi vòng lặp agent thực tế diễn ra: nhận task, lắp context qua ContextManager, gọi model, rồi ghi lại turn assistant vào state để ContextManager tự quyết định có cần compress trước turn kế tiếp không.

"""
agent.py

The main agent loop: takes a task, assembles context via ContextManager,
calls the model, appends the resulting turns back into the session, and
lets ContextManager decide whether a compression pass is needed before the
next turn.
"""

from __future__ import annotations

import anthropic

from context_manager import ContextManager
from session_state import Role

AGENT_MODEL = "claude-sonnet-5"


def run_turn(
    client: anthropic.Anthropic,
    context_manager: ContextManager,
    task: str,
) -> str:
    """Run one turn of the agent loop for a single user-provided task."""

    context_manager.add_turn(Role.USER, task)

    system_blocks, messages = context_manager.assemble_context(current_task=task)

    request_kwargs = {
        "model": AGENT_MODEL,
        "max_tokens": 2048,
        "messages": messages,
    }
    if system_blocks:
        request_kwargs["system"] = system_blocks

    response = client.messages.create(**request_kwargs)

    assistant_text = next(
        (block.text for block in response.content if block.type == "text"), ""
    )

    context_manager.add_turn(Role.ASSISTANT, assistant_text)

    # Surface cache economics so you can see, turn by turn, whether the
    # Tier 3 + Tier 2 prefix is actually being served from cache.
    usage = response.usage
    print(
        f"    [usage] input={usage.input_tokens} "
        f"cache_read={getattr(usage, 'cache_read_input_tokens', 0)} "
        f"cache_write={getattr(usage, 'cache_creation_input_tokens', 0)} "
        f"output={usage.output_tokens}"
    )

    return assistant_text

Chú ý cách add_turn(Role.USER, task) được gọi trước khi lắp context — vì task hiện tại phải được tính vào ngân sách token khi ContextManager quyết định có cần compress hay không. Và add_turn(Role.ASSISTANT, assistant_text) được gọi sau khi có response — turn phản hồi của model cũng là một phần của Tier 1, sẽ được compress ở lần trigger tiếp theo giống mọi turn khác.

Dòng print cuối hàm không chỉ để debug — trong một hệ thống thật, đây là chỗ bạn nên đẩy metric (chỉ số) này vào hệ thống observability (Prometheus, Datadog, ...) để theo dõi tỷ lệ cache hit theo thời gian thực trên production.

Mẹo:
- Nếu bạn tích hợp tool use (agent gọi tool thực) vào vòng lặp này, mỗi tool result cũng nên đi qua context_manager.add_turn(Role.TOOL, ...) — đừng để tool output "lách" qua context mà không được track token, nếu không toàn bộ hệ thống threshold sẽ tính sai.
- Với các turn có tool_result rất dài (ví dụ log lỗi CI dài hàng nghìn dòng), hãy tóm tắt trước khi gọi add_turn — đừng đợi đến khi Tier 1 đầy rồi mới compress, vì một turn khổng lồ có thể vượt cả tier1_max_tokens chỉ trong một lần add.
- Bọc client.messages.create(...) trong try/except bắt anthropic.RateLimitError và anthropic.APIStatusError (xem shared/error-codes.md nếu bạn dùng skill claude-api) — một session dài nhiều giờ chắc chắn sẽ gặp rate limit hoặc lỗi 5xx ít nhất một lần.

Bước 5: Ghép Tất Cả Lại — Script Chạy Session Chính

run_session.py là entry point (điểm khởi chạy) — nó khởi tạo toàn bộ hệ thống và chạy một session giả lập qua một danh sách task được viết sẵn, đại diện cho một buổi làm việc coding thực tế kéo dài nhiều giờ.

"""
run_session.py

Entry point that wires SessionState + CompressionEngine + ContextManager +
the agent loop together, then drives a simulated multi-hour coding session
through a scripted list of tasks so you can observe compression triggering
in real time.
"""

from __future__ import annotations

import argparse
import time

import anthropic
from dotenv import load_dotenv

from agent import run_turn
from compression_engine import CompressionEngine
from context_manager import ContextManager
from session_state import CompressionThresholds, SessionState

load_dotenv()

SCRIPTED_TASKS = [
    "Set up the project skeleton for a FastAPI order service with Postgres.",
    "Add the Order and OrderItem SQLAlchemy models with a one-to-many relationship.",
    "Write the POST /orders endpoint with request validation via Pydantic.",
    "Add unit tests for the POST /orders endpoint, including a validation-failure case.",
    "The tests are failing because of a session-scoping bug in the test fixtures -- fix it.",
    "Add a GET /orders/{id} endpoint with a 404 branch for missing orders.",
    "Refactor order total calculation into a dedicated OrderPricingService.",
    "Add Redis-backed idempotency keys to POST /orders to prevent double-submits.",
    "Write an integration test that exercises the idempotency behavior end-to-end.",
    "Add structured logging (structlog) across the order service request lifecycle.",
    "Instrument the service with OpenTelemetry traces for the order creation path.",
    "The idempotency check has a race condition under concurrent requests -- fix it.",
    "Add a background job that cancels unpaid orders after 30 minutes.",
    "Write a migration to add an expires_at column to the orders table.",
    "Document the new endpoints and the idempotency contract in the service README.",
    "Add rate limiting to POST /orders keyed by customer_id.",
    "Investigate a flaky integration test in CI -- it fails about 1 in 20 runs.",
    "Add a circuit breaker around the downstream inventory-service HTTP call.",
    "Write a load test script targeting 500 req/s against POST /orders.",
    "Summarize the current state of the order service for a handoff to another engineer.",
]


def build_session(project_name: str) -> tuple[anthropic.Anthropic, ContextManager]:
    client = anthropic.Anthropic()
    engine = CompressionEngine(client)
    session = SessionState(
        project_name=project_name,
        thresholds=CompressionThresholds(
            tier1_max_tokens=6_000,
            tier2_max_tokens=3_000,
            tier1_max_turns=20,
        ),
    )
    context_manager = ContextManager(session=session, engine=engine)
    return client, context_manager


def main() -> None:
    parser = argparse.ArgumentParser(description="Simulate a multi-hour agentic coding session.")
    parser.add_argument("--project-name", default="order-service")
    parser.add_argument("--task-delay", type=float, default=0.0,
                         help="Seconds to sleep between tasks (0 = run as fast as possible).")
    args = parser.parse_args()

    client, context_manager = build_session(args.project_name)

    for i, task in enumerate(SCRIPTED_TASKS, start=1):
        print(f"\n=== Turn {i}/{len(SCRIPTED_TASKS)}: {task[:70]} ===")
        run_turn(client, context_manager, task)
        stats = context_manager.get_stats()
        print(f"    [state] tier1={stats['tier1_tokens']}t "
              f"tier2={stats['tier2_tokens']}t tier3={stats['tier3_tokens']}t "
              f"total={stats['total_context_tokens']}t")

        if args.task_delay:
            time.sleep(args.task_delay)

    print("\n=== Compression event log ===")
    for event in context_manager.session.compression_events:
        print(event)


if __name__ == "__main__":
    main()

20 task trong SCRIPTED_TASKS được viết cố ý theo trình tự tự nhiên của một session engineering thật: setup → feature → test → bug fix → refactor → feature phụ → observability → race condition → tài liệu → tối ưu → điều tra flaky test → tổng kết bàn giao. Đây không phải danh sách ngẫu nhiên — nó mô phỏng đúng kiểu "trôi dạt chủ đề" (topic drift) mà một session dài thực sự trải qua, để bạn quan sát rolling summary phải liên tục tổng hợp lại thông tin cũ khi chủ đề chuyển hướng.

Mẹo:
- Tham số --task-delay không chỉ để làm chậm demo cho dễ quan sát — trong môi trường production, khoảng nghỉ giữa các turn ảnh hưởng đến việc cache có còn "sống" hay không (cache mặc định TTL — thời gian sống — là 5 phút). Nếu task cách nhau hơn 5 phút, bạn sẽ mất cache hit dù compression vẫn hoạt động đúng.
- Tách build_session() khỏi main() như trên để bạn có thể tái sử dụng nó trong unit test (Bước 6) mà không phải chạy toàn bộ CLI.
- Nếu bạn muốn benchmark chi phí thật, thêm một biến đếm tổng input_tokens/output_tokens cộng dồn qua toàn bộ session và in ra ở cuối main() — đây là input trực tiếp cho phần benchmark ở cuối bài.

Bước 6: Chạy Session Và Kiểm Chứng Compression

Chạy thử:

python run_session.py --project-name order-service

Log console sẽ có dạng tương tự (số liệu minh họa, sẽ khác đôi chút theo response thật của model):

=== Turn 1/20: Set up the project skeleton for a FastAPI order service... ===
    [usage] input=142 cache_read=0 cache_write=0 output=380
    [state] tier1=1150t tier2=0t tier3=0t total=1150t

=== Turn 5/20: The tests are failing because of a session-scoping bug... ===
    [usage] input=5820 cache_read=0 cache_write=0 output=410
    [state] tier1=6210t tier2=0t tier3=0t total=6210t

[turn 5] Tier 1 -> Tier 2: folded 10 turns (6210 tokens) into a 480-token rolling summary.

=== Turn 6/20: Add a GET /orders/{id} endpoint with a 404 branch... ===
    [usage] input=612 cache_read=0 cache_write=380 output=395
    [state] tier1=980t tier2=480t tier3=0t total=1460t

...

=== Turn 14/20: Write a migration to add an expires_at column... ===
    [usage] input=1340 cache_read=1020 cache_write=0 output=350
    [state] tier1=1120t tier2=3140t tier3=0t total=4260t

[turn 14] Tier 2 -> Tier 3: distilled rolling summary into project archive (620 tokens).

=== Turn 20/20: Summarize the current state of the order service... ===
    [usage] input=1980 cache_read=1650 cache_write=0 output=520
    [state] tier1=1310t tier2=880t tier3=620t total=2810t

=== Compression event log ===
[turn 5] Tier 1 -> Tier 2: folded 10 turns (6210 tokens) into a 480-token rolling summary.
[turn 9] Tier 1 -> Tier 2: folded 8 turns (6080 tokens) into a 510-token rolling summary.
[turn 14] Tier 2 -> Tier 3: distilled rolling summary into project archive (620 tokens).
[turn 17] Tier 1 -> Tier 2: folded 9 turns (6340 tokens) into a 495-token rolling summary.

Ba điều cần xác nhận từ log này để chắc chắn hệ thống hoạt động đúng:

1. tier1 giảm mạnh ngay sau mỗi lần compression event xuất hiện — nếu tier1 tiếp tục tăng vô hạn mà không bao giờ giảm, _maybe_compress() không được gọi đúng chỗ, hoặc drain() không hoạt động.
2. cache_read tăng dần ở các turn sau khi Tier 2/Tier 3 ổn định — điều này xác nhận prompt caching đang hoạt động đúng như thiết kế ở Bước 3.
3. total (tổng token context) luôn ở mức thấp và ổn định — thay vì tăng tuyến tính theo số turn như một agent không có compression.

Để kiểm chứng bằng test tự động (không cần đọc log thủ công), viết một test case đơn giản với pytest:

from context_manager import ContextManager
from compression_engine import CompressionEngine
from session_state import CompressionThresholds, Role, SessionState
import anthropic


def test_tier1_compresses_when_threshold_exceeded():
    client = anthropic.Anthropic()
    engine = CompressionEngine(client)
    session = SessionState(
        project_name="test-project",
        # Small thresholds so the test triggers compression in a few turns
        # instead of needing a full 20-turn session.
        thresholds=CompressionThresholds(tier1_max_tokens=200, tier1_max_turns=6),
    )
    cm = ContextManager(session=session, engine=engine)

    long_turn = "Implement a caching layer for the product lookup service. " * 10
    for _ in range(4):
        cm.add_turn(Role.USER, long_turn)
        cm.add_turn(Role.ASSISTANT, "Done. Added a Redis-backed cache with a 60s TTL.")

    # After exceeding the threshold, Tier 1 must have been drained at least
    # once, and Tier 2 must now hold a non-empty rolling summary.
    assert len(session.compression_events) >= 1
    assert session.tier2.summary_text != ""
    assert "Rolling Summary" in session.tier2.summary_text
    # Tier 1 should be much smaller than it would be without compression.
    assert session.tier1.token_count < 200 * 3

Chạy: pytest test_compression.py -v. Test này gọi API thật (không mock) để xác nhận cả logic trigger và prompt template thực sự cho ra output đúng format — đây là trade-off có chủ đích: test chậm hơn và tốn một ít chi phí API, nhưng bắt được lỗi thật ở prompt mà mock không bắt được.

Mẹo:
- Khi debug threshold, tạm thời hạ tier1_max_tokens xuống rất thấp (như 200 trong test trên) để buộc compression trigger ngay ở turn thứ 2-3, thay vì phải chạy đủ 20 task mỗi lần thử.
- Nếu bạn thấy cache_read_input_tokens luôn bằng 0 dù đã đặt cache_control đúng, kiểm tra: (a) hai request có cách nhau dưới 5 phút không, (b) hai request có model giống nhau không (cache là theo model), (c) system block trước breakpoint có byte-for-byte giống nhau giữa hai lần gọi không — một thay đổi nhỏ nhất (ví dụ timestamp bị nhúng vào text) cũng làm hỏng cache.
- Lưu session.compression_events ra file log riêng (không chỉ in console) khi chạy session dài thật — đây là dữ liệu quý để tối ưu threshold sau này.

Bước 7: Mở Rộng Hệ Thống — Code Skeleton Injection

Ở Bước 2, bạn đã viết generate_code_skeleton() nhưng chưa dùng nó ở đâu cả. Đây là lúc gắn nó vào agent loop: trước khi gọi model cho một turn có đề cập đến file đã tồn tại trong repo, tự động trích skeleton (chữ ký + docstring, không có phần thân) của file đó và tiêm (inject) vào context — thay vì để model phải yêu cầu đọc toàn bộ file, hoặc tệ hơn, để agent "đoán" nội dung file dựa trên tên.

import os
import re

from compression_engine import generate_code_skeleton
from context_manager import ContextManager
from session_state import Role

FILE_REFERENCE_PATTERN = re.compile(r"\b([\w./-]+\.py)\b")


def find_referenced_files(task: str) -> list[str]:
    """Best-effort extraction of file paths mentioned in a task description."""
    return FILE_REFERENCE_PATTERN.findall(task)


def inject_skeletons_for_turn(
    context_manager: ContextManager,
    task: str,
    repo_root: str,
) -> None:
    # Before calling run_turn for a turn that involves an existing file:
    # read each referenced file from disk, generate a skeleton (signatures +
    # docstrings, no bodies), and inject it as a TOOL turn -- so the model
    # knows the file's shape without paying for its full contents.
    for rel_path in find_referenced_files(task):
        full_path = os.path.join(repo_root, rel_path)
        if not os.path.isfile(full_path):
            continue

        with open(full_path, "r", encoding="utf-8") as f:
            source = f.read()

        skeleton = generate_code_skeleton(rel_path, source)
        context_manager.add_turn(
            Role.TOOL,
            content=f"[file skeleton for {rel_path}]\n{skeleton}",
            tool_name="code_skeleton",
        )

Cách gọi trong vòng lặp session (điều chỉnh nhỏ trong run_session.py):

inject_skeletons_for_turn(context_manager, task, repo_root="./order-service")
run_turn(client, context_manager, task)

Nguyên tắc quan trọng ở đây: chỉ tiêm skeleton, không tiêm toàn bộ file, trừ khi model chủ động yêu cầu xem chi tiết. Nếu task là "sửa bug ở dòng validate trong order_service.py", model thấy skeleton, biết OrderService có method validate_order(self, order: Order) -> bool, và sau đó có thể chủ động gọi một tool đọc file thật nếu cần xem implementation. Điều này tiết kiệm đáng kể token cho các turn không cần chỉnh sửa sâu vào file đó — mà vẫn không làm agent "mù" về cấu trúc codebase.

So sánh nhanh chi phí token cho một file service điển hình ~350 dòng code: full file content ước tính ~2.800 token; skeleton tương ứng (chữ ký + docstring của ~12 method) chỉ khoảng ~180-220 token — giảm hơn 90%, và giảm này lặp lại ở mọi turn có đề cập đến file đó, không chỉ một lần.

Mẹo:
- find_referenced_files ở trên dùng regex đơn giản, chỉ bắt file .py được gõ rõ tên — trong hệ thống thật, bạn nên kết hợp với kết quả tool call thật (ví dụ agent vừa chạy grep hoặc glob và trả về đường dẫn file) để chính xác hơn là chỉ đoán từ text task.
- Cache kết quả generate_code_skeleton() theo (file_path, mtime) — nếu file không thay đổi giữa hai turn, không cần parse ast lại; việc này không tốn token nhưng tốn CPU nếu file lớn và được tham chiếu liên tục.
- Với các ngôn ngữ khác Python, viết một hàm skeleton tương ứng dùng parser phù hợp (tree-sitter là lựa chọn tốt, hỗ trợ đa ngôn ngữ) — nguyên lý giữ nguyên: trích signature + docstring/comment đầu, bỏ implementation.

Checklist Kiểm Chứng Và Benchmark Hiệu Năng

Trước khi coi hệ thống này "xong", chạy qua checklist sau:

• [ ] TierOneState.drain() luôn được gọi trước khi tính compression mới — không có turn nào bị compress hai lần.
• [ ] TierTwoState.replace() thay thế summary cũ, không append — kiểm tra bằng cách log độ dài summary_text qua nhiều lần compress; nó phải ổn định, không tăng tuyến tính.
• [ ] Sau mỗi lần Tier 2 → Tier 3, tier2.summary_text được reset về rỗng — nếu không, thông tin sẽ bị đếm hai lần trong total_context_tokens.
• [ ] assemble_context() luôn trả về đúng thứ tự: Tier 3 → Tier 2 → Tier 1 → task hiện tại, và cache_control luôn nằm ở block cuối cùng trong phần system.
• [ ] cache_read_input_tokens > 0 ở các turn sau turn đầu tiên trong cùng một cụm 5 phút — nếu luôn bằng 0, có silent invalidator (yếu tố làm hỏng cache âm thầm) đang tồn tại.
• [ ] Project archive (Tier 3) sau khi distill có đầy đủ 8 section theo đúng tên: Project Identity, Architectural Decisions, Completed Milestones, Current State, Known Issues, Constraints, Patterns & Standards, Do Not Revisit.
• [ ] Code skeleton injection không làm tăng token nhiều hơn việc gửi toàn bộ file — nếu file rất nhỏ (dưới 50 dòng), skeleton có thể không đáng để tách riêng, cân nhắc bỏ qua bước tiêm cho các file nhỏ.

Benchmark thực tế trên session 20-task ở trên (số liệu trung bình qua vài lần chạy, dùng claude-sonnet-5):

Vài quan sát quan trọng từ bảng trên:

• Chi phí compression (~6% tổng input token) là cái giá chấp nhận được đổi lấy việc giảm hơn 83% tổng token input toàn session.
• Không có compression, đến khoảng turn 35-40 của một session thật (không phải bản demo 20 task), context sẽ vượt context window của hầu hết model hiện tại — đây không chỉ là vấn đề chi phí, mà là vấn đề khả thi kỹ thuật.
• Cache hit chỉ đạt hiệu quả rõ rệt sau khi Tier 2/Tier 3 đã ổn định (khoảng turn 6 trở đi) — vài turn đầu của bất kỳ session nào sẽ luôn có cache miss vì chưa có gì để cache.

Mẹo:
- Chạy benchmark của riêng bạn trên workload thật, đừng tin số liệu ở trên làm chuẩn tuyệt đối — độ dài task, ngôn ngữ lập trình, và độ phức tạp domain đều ảnh hưởng đáng kể đến tỷ lệ compression.
- Theo dõi riêng "token tiêu tốn cho compression / tổng token tiết kiệm được" theo thời gian — nếu tỷ lệ này vượt quá ~15-20%, threshold có thể đang đặt quá thấp (compress quá thường xuyên so với lượng nội dung thực sự cần nén).
- Dùng client.messages.count_tokens(...) (không phải estimate_tokens() xấp xỉ) để tính lại con số chính xác trước khi đưa benchmark vào báo cáo chính thức cho team hoặc khách hàng.

Tổng Kết

Bạn vừa xây một hệ thống compression 3-tier hoàn chỉnh, chạy được, cho một agent coding session kéo dài nhiều giờ: SessionState mô hình hóa đúng ba tầng dữ liệu; CompressionEngine thực hiện cả hai loại compression bằng LLM (Tier 1→2, Tier 2→3) và một utility tĩnh không tốn token (code skeleton); ContextManager quyết định khi nào compress và lắp ráp context theo đúng thứ tự tối ưu cho prompt caching; agent.py và run_session.py ghép tất cả thành một vòng lặp agent thực thi được; và Bước 7 cho bạn thấy compression không chỉ áp dụng cho hội thoại — nó áp dụng cho mọi thứ đi vào context, bao gồm cả code.

Số liệu benchmark ở trên — giảm hơn 83% token cộng dồn, chi phí giảm gần 6 lần — không phải là giới hạn trên. Với threshold tinh chỉnh kỹ hơn, prompt distillation chất lượng cao hơn, và code skeleton injection áp dụng triệt để hơn (đa ngôn ngữ, cache theo mtime), một hệ thống production có thể đạt tỷ lệ tiết kiệm cao hơn nữa mà không đánh đổi chất lượng output của agent.

Bước tiếp theo hợp lý: persist SessionState ra đĩa (hoặc database) giữa các lần chạy, để project archive (Tier 3) sống sót qua nhiều session khác nhau trên cùng một codebase — biến nó thành bộ nhớ dài hạn thực sự của agent, không chỉ trong phạm vi một lần chạy. Các module tiếp theo trong khóa học sẽ đi sâu vào tối ưu token ở tầng hội thoại nhiều lượt (multi-turn) và tối ưu tool use — cả hai đều xây trên nền tảng 3-tier bạn vừa hoàn thiện ở bài này.

Mẹo:
- Commit toàn bộ 5 file của bài lab này vào một repo riêng ngay sau khi hoàn thành — đây là bộ khung (scaffold) bạn có thể tái sử dụng và điều chỉnh cho dự án thật, thay vì viết lại từ đầu mỗi lần.
- Trước khi đưa hệ thống này vào production, thêm cơ chế fallback: nếu lời gọi compression thất bại (rate limit, timeout), đừng để toàn bộ agent loop crash — giữ nguyên Tier 1 hiện tại, log lỗi, và thử lại ở turn kế tiếp.