Đến đây trong khoá học, bạn đã lần lượt tìm hiểu năm mảnh ghép riêng lẻ: chi phí khủng khiếp của việc nhồi cả codebase vào prompt, cách repo map và file summary giúp AI "nhìn" được toàn cảnh với chi phí token rẻ, cách chỉ đưa đúng file liên quan dựa trên dependency graph, cách CLAUDE.md đóng vai trò bộ nhớ dự án bền vững, và cách RAG (Retrieval-Augmented Generation — sinh câu trả lời có tăng cường truy xuất) giúp AI "tìm" đúng đoạn code cần thiết trong một codebase khổng lồ mà không cần đọc hết.

Vấn đề là: trong thực tế, không ai dùng riêng lẻ từng kỹ thuật đó. Một agent AI làm việc hiệu quả trong ngày thường kết hợp cả năm kỹ thuật này thành một pipeline (dây chuyền xử lý) duy nhất, chạy tự động mỗi khi cần build ngữ cảnh cho một tác vụ. Bài thực hành này sẽ đưa bạn đi qua toàn bộ quá trình lắp ráp pipeline đó — từ con số 0 đến một công cụ CLI (command-line interface — giao diện dòng lệnh) chạy được, có thể tích hợp vào workflow (quy trình làm việc) hàng ngày của cả kỹ sư, QA và product manager.

Đây là bài dài nhất và chi tiết nhất của module, vì mục tiêu là bạn gõ theo, chạy được, và mang thẳng về áp dụng cho codebase thật của team mình — không chỉ đọc lý thuyết.

Tổng Quan Kiến Trúc: Chúng Ta Sẽ Xây Dựng Gì

Ý tưởng cốt lõi của toàn bộ pipeline: mỗi khi cần build ngữ cảnh cho một yêu cầu (ví dụ "sửa bug ở file X", "viết test cho feature Y", "giải thích luồng thanh toán cho PM"), pipeline sẽ chạy qua nhiều layer (lớp) độc lập, mỗi layer chịu trách nhiệm lấy một loại ngữ cảnh khác nhau, sau đó gộp lại và cắt gọt theo một token budget (ngân sách token) cố định trước khi đưa vào context window của LLM.

Năm layer tương ứng với năm bài học trước:

1. Static Layer (lớp tĩnh) — nội dung luôn phải có, không đổi theo request: CLAUDE.md, coding convention, cấu trúc thư mục cấp cao. Đây là "bộ nhớ dài hạn" của dự án.
2. Structural Layer (lớp cấu trúc) — repo map: cây thư mục rút gọn + danh sách symbol (function, class, method) quan trọng, không kèm code chi tiết. Cho AI "bản đồ" tổng thể.
3. RAG Layer (lớp truy xuất) — chunk (mảnh) code được embed (nhúng vector) sẵn, truy xuất theo độ tương đồng ngữ nghĩa với câu hỏi. Cho AI những đoạn code "có thể liên quan" mà dependency graph tĩnh không thấy được.
4. Dependency Layer (lớp phụ thuộc) — dựa trên import/call graph (đồ thị gọi hàm), lấy chính xác các file mà target file phụ thuộc trực tiếp hoặc bị phụ thuộc.
5. Target Layer (lớp mục tiêu) — file đang được chỉnh sửa, luôn được đưa vào toàn văn, không cắt, vì đây là trọng tâm của tác vụ.

Tất cả năm layer chạy song song (về mặt logic), trả về danh sách các "context block" kèm token count và một priority score (điểm ưu tiên). Một Pipeline Orchestrator (bộ điều phối) sẽ gộp các block này, sắp theo thứ tự ưu tiên, và cắt bớt block ở layer thấp ưu tiên hơn khi vượt token budget. Cuối cùng, một CLI đóng vai trò giao diện để kỹ sư/QA/PM gọi pipeline từ terminal hoặc từ hook tự động.

Sơ đồ kiến trúc dạng ASCII:

                        ┌─────────────────────────┐
                        │   CLI (src/cli.py)      │
                        │   build-index / query   │
                        └────────────┬────────────┘
                                     │
                                     ▼
                        ┌─────────────────────────┐
                        │  Pipeline Orchestrator  │
                        │     (src/pipeline.py)   │
                        └────────────┬────────────┘
                                     │  gọi song song 5 layer
          ┌───────────┬──────────────┼──────────────┬───────────┐
          ▼           ▼              ▼               ▼          ▼
   ┌───────────┐┌───────────┐ ┌───────────┐  ┌──────────────┐┌───────────┐
   │  Static   ││Structural │ │    RAG    │  │  Dependency  ││  Target   │
   │  Layer    ││  Layer    │ │   Layer   │  │    Layer     ││  Layer    │
   │ CLAUDE.md ││ repo map  │ │ Chroma DB │  │ import graph │││ file full│
   └───────────┘└───────────┘ └───────────┘  └──────────────┘└───────────┘
          │           │              │               │           │
          └───────────┴──────────────┼───────────────┴───────────┘
                                     ▼
                        ┌─────────────────────────┐
                        │  Token Budgeter/Trimmer │
                        │    (src/utils/tokens.py)│
                        └────────────┬────────────┘
                                     ▼
                         Context cuối cùng gửi vào LLM

Về mặt công nghệ, chúng ta chọn Python vì hệ sinh thái xử lý code (tree-sitter), embedding (sentence-transformers/OpenAI embeddings) và vector database (ChromaDB) đều rất trưởng thành ở đây. Bạn có thể port ý tưởng này sang Node.js hoặc Go nếu team quen công nghệ đó hơn — kiến trúc layer không phụ thuộc ngôn ngữ.

Mẹo

• Đừng cố build cả 5 layer hoàn chỉnh ngay từ đầu. Build Static + Target Layer trước (rẻ, dễ, có giá trị ngay), rồi mới thêm Structural, Dependency, và cuối cùng là RAG (layer tốn công nhất vì cần embedding pipeline).
• Vẽ sơ đồ này lên bảng/Miro cho cả team trước khi code — khi mọi người hiểu "5 layer, 1 budget", việc review code pipeline sau này sẽ nhanh hơn nhiều vì ai cũng biết một đoạn code thuộc layer nào.
• Repo map và RAG index nên tách biệt về chu kỳ cập nhật: repo map có thể build lại mỗi lần chạy (rẻ), còn RAG index nên cache và chỉ rebuild khi file thay đổi (tốn embedding API call).

Bước 1: Khởi Tạo Project

Bắt đầu bằng một project Python chuẩn, dùng virtual environment để tránh xung đột dependency với các project khác trên máy.

mkdir codebase-context-pipeline
cd codebase-context-pipeline

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


pip install tiktoken tree-sitter tree-sitter-languages chromadb \
    sentence-transformers typer pydantic rich

pip freeze > requirements.txt

Cấu trúc thư mục project:

codebase-context-pipeline/
├── requirements.txt
├── .env.example
├── src/
│   ├── __init__.py
│   ├── config/
│   │   ├── __init__.py
│   │   └── settings.py
│   ├── layers/
│   │   ├── __init__.py
│   │   ├── static_layer.py
│   │   ├── structural_layer.py
│   │   ├── rag_layer.py
│   │   ├── dependency_layer.py
│   │   └── target_layer.py
│   ├── utils/
│   │   ├── __init__.py
│   │   └── tokens.py
│   ├── pipeline.py
│   └── cli.py
└── .chroma/               # vector db lưu local, gitignore

File cấu hình trung tâm src/config/settings.py — nơi khai báo mọi giá trị "magic number" (số token budget, model tên gì, đường dẫn nào) để không rải rác khắp code:

"""
Central configuration cho toàn bộ pipeline.
Mọi giá trị có thể override qua biến môi trường (env var) để
dễ tinh chỉnh mà không cần sửa code.
"""
from pathlib import Path
from pydantic import BaseModel
import os


class Settings(BaseModel):
    # Đường dẫn tới codebase mục tiêu (repo mà pipeline sẽ đọc)
    repo_root: Path = Path(os.getenv("REPO_ROOT", ".")).resolve()

    # Nơi lưu vector index (ChromaDB persistent storage)
    chroma_dir: Path = Path(os.getenv("CHROMA_DIR", ".chroma")).resolve()

    # Tokenizer dùng để đếm token — nên khớp với model bạn dùng để hỏi AI
    tokenizer_encoding: str = os.getenv("TOKENIZER_ENCODING", "cl100k_base")

    # Model embedding chạy local, không tốn phí API
    embedding_model: str = os.getenv(
        "EMBEDDING_MODEL", "all-MiniLM-L6-v2"
    )

    # Tổng token budget cho MỘT lần build context (không tính câu hỏi/system prompt)
    total_token_budget: int = int(os.getenv("TOTAL_TOKEN_BUDGET", "8000"))

    # Tỷ trọng ưu tiên (%) phân bổ ngân sách cho từng layer.
    # Target luôn được ưu tiên cao nhất vì đó là trọng tâm request.
    layer_budget_ratio: dict = {
        "target": 0.35,
        "dependency": 0.25,
        "structural": 0.15,
        "rag": 0.20,
        "static": 0.05,
    }

    # Số chunk RAG tối đa trả về mỗi query
    rag_top_k: int = int(os.getenv("RAG_TOP_K", "8"))

    # Kích thước mỗi chunk khi RAG chia nhỏ code (theo số dòng)
    rag_chunk_lines: int = int(os.getenv("RAG_CHUNK_LINES", "60"))

    # Danh sách file/thư mục luôn bỏ qua khi index
    ignore_patterns: list[str] = [
        "node_modules", ".git", ".venv", "dist", "build",
        "*.lock", "*.min.js", "__pycache__",
    ]

    class Config:
        arbitrary_types_allowed = True


settings = Settings()

Mẹo

• Đặt total_token_budget thấp hơn context window thật của model bạn dùng (ví dụ model hỗ trợ 200K token thì đặt budget pipeline khoảng 8K–15K) — phần còn lại của context window để dành cho lịch sử hội thoại, output của model, và các tool call khác.
• Dùng Pydantic (hoặc dataclass có validate) cho settings thay vì đọc trực tiếp os.environ rải rác — sau này khi thêm layer mới, bạn chỉ sửa một chỗ.
• Commit file .env.example (không commit .env thật) để người khác trong team biết cần config biến môi trường nào khi clone project.

Bước 2: Lớp Tĩnh (Static Layer)

Static Layer là layer đơn giản nhất nhưng quan trọng không kém: nó đọc các file "bối cảnh dự án" luôn cần có mặt trong mọi request — điển hình là CLAUDE.md (hoặc .cursorrules, AGENTS.md tuỳ công cụ bạn dùng), file mô tả coding convention, kiến trúc tổng thể, các quy tắc bắt buộc. Nội dung này không đổi theo target file hay câu hỏi, nên có thể cache và tính token một lần.

"""
Static Layer: đọc các file ngữ cảnh cố định của dự án
(CLAUDE.md, .cursorrules, README kiến trúc, v.v).
Nội dung layer này gần như không đổi giữa các request,
nên được cache trong bộ nhớ sau lần đọc đầu tiên.
"""
from dataclasses import dataclass
from pathlib import Path
from functools import lru_cache

from src.config.settings import settings

STATIC_CONTEXT_FILES = [
    "CLAUDE.md",
    "AGENTS.md",
    ".cursorrules",
    "docs/ARCHITECTURE.md",
]


@dataclass
class ContextBlock:
    """Đơn vị dữ liệu chung mà mọi layer trả về, để pipeline dễ gộp."""
    layer: str
    source: str          # đường dẫn file hoặc mô tả nguồn
    content: str
    priority: float       # 0.0 - 1.0, càng cao càng quan trọng
    token_count: int = 0


class StaticLayer:
    def __init__(self, repo_root: Path = settings.repo_root):
        self.repo_root = repo_root

    @lru_cache(maxsize=1)
    def _load_raw(self) -> list[tuple[str, str]]:
        found = []
        for filename in STATIC_CONTEXT_FILES:
            path = self.repo_root / filename
            if path.exists():
                text = path.read_text(encoding="utf-8", errors="ignore")
                found.append((filename, text))
        return found

    def build(self) -> list[ContextBlock]:
        """Trả về list ContextBlock, mỗi file tĩnh là một block riêng
        để pipeline có thể cắt bớt từng file nếu cần, thay vì cắt cả cục."""
        blocks = []
        for filename, text in self._load_raw():
            blocks.append(
                ContextBlock(
                    layer="static",
                    source=filename,
                    content=text,
                    # CLAUDE.md/AGENTS.md ưu tiên cao nhất trong nhóm static
                    priority=1.0 if filename in ("CLAUDE.md", "AGENTS.md") else 0.6,
                )
            )
        return blocks

Lưu ý thiết kế: mỗi layer đều trả về ContextBlock — một cấu trúc dữ liệu chung có layer, source, content, priority. Đây là "hợp đồng" (interface) giúp Pipeline Orchestrator ở Bước 7 xử lý mọi layer theo cùng một cách, không cần biết chi tiết bên trong từng layer.

Mẹo

• Nếu CLAUDE.md của bạn dài (nhiều team viết CLAUDE.md hàng nghìn từ), tách nó thành nhiều section và cho phép Static Layer chỉ load section liên quan đến loại request (ví dụ section "testing" chỉ load khi persona là QA) — tránh lãng phí token cho quy tắc không liên quan.
• Dùng lru_cache hoặc cache tương tự vì Static Layer bị gọi ở MỌI request — đọc file mỗi lần là lãng phí I/O không cần thiết.
• Đặt priority của static content cao (0.6–1.0) nhưng token budget ratio thấp (5% trong ví dụ Bước 1) — layer này quan trọng về "chất" nhưng nên ngắn gọn về "lượng". Nếu CLAUDE.md của bạn chiếm hơn 5% budget, đó là dấu hiệu cần rút gọn nó.

Bước 3: Lớp Cấu Trúc (Structural Layer)

Structural Layer chính là "repo map" đã học ở bài 2 — nhưng giờ viết thành module Python tái sử dụng được. Nó quét cây thư mục và dùng tree-sitter để parse từng file, trích ra danh sách symbol (class, function, method) cùng chữ ký (signature) của chúng, KHÔNG kèm phần thân hàm. Kết quả là một bản đồ codebase rất nhỏ về token nhưng vẫn cho AI biết "cái gì nằm ở đâu".

"""
Structural Layer: sinh "repo map" — cây thư mục rút gọn
kèm danh sách symbol chính của mỗi file (không kèm code chi tiết).
Dùng tree-sitter để parse, không phụ thuộc việc file có chạy được không.
"""
from dataclasses import dataclass
from pathlib import Path
import fnmatch

from tree_sitter_languages import get_parser

from src.config.settings import settings
from src.layers.static_layer import ContextBlock

LANGUAGE_MAP = {
    ".py": "python",
    ".js": "javascript",
    ".ts": "typescript",
    ".tsx": "tsx",
    ".go": "go",
}

SYMBOL_NODE_TYPES = {
    "function_definition", "class_definition",       # Python
    "function_declaration", "class_declaration",      # JS/TS/Go
    "method_definition",
}


def _should_ignore(path: Path) -> bool:
    return any(fnmatch.fnmatch(str(path), f"*{pat}*") for pat in settings.ignore_patterns)


def _extract_symbols(file_path: Path) -> list[str]:
    """Parse 1 file, trả về list chữ ký symbol dạng text, ví dụ:
    'def calculate_total(items: list) -> float:'"""
    ext = file_path.suffix
    lang = LANGUAGE_MAP.get(ext)
    if not lang:
        return []

    try:
        parser = get_parser(lang)
        source = file_path.read_bytes()
        tree = parser.parse(source)
    except Exception:
        return []

    symbols = []

    def walk(node):
        if node.type in SYMBOL_NODE_TYPES:
            # Lấy dòng đầu tiên của node (chữ ký), bỏ phần thân
            start = node.start_point[0]
            end_line_byte = source.split(b"\n")[start]
            symbols.append(end_line_byte.decode("utf-8", errors="ignore").strip())
        for child in node.children:
            walk(child)

    walk(tree.root_node)
    return symbols


@dataclass
class RepoMapEntry:
    path: str
    symbols: list[str]


class StructuralLayer:
    def __init__(self, repo_root: Path = settings.repo_root):
        self.repo_root = repo_root

    def _scan_files(self) -> list[Path]:
        return [
            p for p in self.repo_root.rglob("*")
            if p.is_file()
            and p.suffix in LANGUAGE_MAP
            and not _should_ignore(p)
        ]

    def build_repo_map(self) -> list[RepoMapEntry]:
        entries = []
        for file_path in self._scan_files():
            symbols = _extract_symbols(file_path)
            if symbols:
                entries.append(
                    RepoMapEntry(
                        path=str(file_path.relative_to(self.repo_root)),
                        symbols=symbols,
                    )
                )
        return entries

    def build(self) -> list[ContextBlock]:
        entries = self.build_repo_map()
        lines = []
        for entry in entries:
            lines.append(f"{entry.path}:")
            for sym in entry.symbols:
                lines.append(f"    {sym}")
        content = "\n".join(lines)

        return [
            ContextBlock(
                layer="structural",
                source="repo_map",
                content=content,
                priority=0.7,
            )
        ]

Kết quả build() là một block dạng text rất giống output của lệnh tree kết hợp ctags, ví dụ:

src/services/payment.py:
    class PaymentService:
    def charge(self, order_id: str, amount: float) -> PaymentResult:
    def refund(self, payment_id: str) -> RefundResult:

Mẹo

• Nếu repo có hàng nghìn file, đừng build lại repo map từ đầu mỗi lần chạy — cache kết quả build_repo_map() ra file JSON, chỉ re-parse file nào có mtime (thời gian sửa đổi) mới hơn lần cache trước.
• Với monorepo nhiều ngôn ngữ, mở rộng LANGUAGE_MAP dần dần — không cần hỗ trợ hết mọi ngôn ngữ ngay, ưu tiên ngôn ngữ chiếm tỷ trọng code lớn nhất trước.
• Nếu team dùng nhiều generic/interface phức tạp (TypeScript, Go), thử nghiệm xem chữ ký đầy đủ (kèm type parameter) hay chữ ký rút gọn giúp AI hiểu đúng hơn — đôi khi rút gọn quá tay khiến AI đoán sai kiểu dữ liệu.

Bước 4: Lớp RAG (RAG Layer)

RAG Layer là layer tốn công xây dựng nhất vì cần một pipeline riêng: chia code thành chunk, embed từng chunk thành vector, lưu vào vector database, và khi có câu hỏi thì embed câu hỏi rồi tìm k chunk gần nhất về ngữ nghĩa. Layer này bù đắp cho điểm yếu của Structural Layer và Dependency Layer: cả hai đều dựa trên cấu trúc "cứng" (import, symbol), nên bỏ lỡ các đoạn code liên quan về mặt ý nghĩa nhưng không có liên kết import trực tiếp — ví dụ một hàm utility ở module khác implement logic tương tự.

"""
RAG Layer: chia codebase thành chunk, embed bằng model chạy local
(sentence-transformers), lưu trong ChromaDB, và truy xuất top-k chunk
liên quan nhất về ngữ nghĩa với câu hỏi/target file hiện tại.
"""
from dataclasses import dataclass
from pathlib import Path

import chromadb
from chromadb.utils import embedding_functions

from src.config.settings import settings
from src.layers.static_layer import ContextBlock
from src.layers.structural_layer import LANGUAGE_MAP, _should_ignore


@dataclass
class CodeChunk:
    path: str
    start_line: int
    end_line: int
    content: str


def _chunk_file(file_path: Path, chunk_lines: int) -> list[CodeChunk]:
    lines = file_path.read_text(encoding="utf-8", errors="ignore").splitlines()
    chunks = []
    for i in range(0, len(lines), chunk_lines):
        block_lines = lines[i:i + chunk_lines]
        if not block_lines:
            continue
        chunks.append(
            CodeChunk(
                path=str(file_path),
                start_line=i + 1,
                end_line=i + len(block_lines),
                content="\n".join(block_lines),
            )
        )
    return chunks


class RagLayer:
    def __init__(self, repo_root: Path = settings.repo_root):
        self.repo_root = repo_root
        self._client = chromadb.PersistentClient(path=str(settings.chroma_dir))
        self._embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name=settings.embedding_model
        )
        self._collection = self._client.get_or_create_collection(
            name="codebase_chunks",
            embedding_function=self._embed_fn,
        )

    def build_index(self) -> int:
        """Chunk toàn bộ codebase và ghi vào ChromaDB.
        Chạy 1 lần khi setup, và mỗi khi có file thay đổi đáng kể.
        Trả về số chunk đã index."""
        ids, documents, metadatas = [], [], []

        for file_path in self.repo_root.rglob("*"):
            if not file_path.is_file() or file_path.suffix not in LANGUAGE_MAP:
                continue
            if _should_ignore(file_path):
                continue

            rel_path = str(file_path.relative_to(self.repo_root))
            for chunk in _chunk_file(file_path, settings.rag_chunk_lines):
                chunk_id = f"{rel_path}:{chunk.start_line}-{chunk.end_line}"
                ids.append(chunk_id)
                documents.append(chunk.content)
                metadatas.append({
                    "path": rel_path,
                    "start_line": chunk.start_line,
                    "end_line": chunk.end_line,
                })

        if not ids:
            return 0

        # upsert theo batch để tránh vượt giới hạn 1 lần gọi của Chroma
        batch_size = 500
        for i in range(0, len(ids), batch_size):
            self._collection.upsert(
                ids=ids[i:i + batch_size],
                documents=documents[i:i + batch_size],
                metadatas=metadatas[i:i + batch_size],
            )
        return len(ids)

    def query(self, query_text: str, top_k: int = None) -> list[ContextBlock]:
        top_k = top_k or settings.rag_top_k
        results = self._collection.query(
            query_texts=[query_text],
            n_results=top_k,
        )

        blocks = []
        docs = results.get("documents", [[]])[0]
        metas = results.get("metadatas", [[]])[0]
        distances = results.get("distances", [[]])[0]

        for doc, meta, distance in zip(docs, metas, distances):
            # distance càng nhỏ càng liên quan -> đổi thành priority (càng lớn càng ưu tiên)
            similarity = max(0.0, 1.0 - distance)
            blocks.append(
                ContextBlock(
                    layer="rag",
                    source=f"{meta['path']}:{meta['start_line']}-{meta['end_line']}",
                    content=doc,
                    priority=round(similarity, 3),
                )
            )
        return blocks

Điểm quan trọng: mỗi chunk RAG trả về gắn kèm priority chính là điểm similarity (độ tương đồng) — chunk khớp ngữ nghĩa cao với câu hỏi sẽ được orchestrator giữ lại trước tiên khi cần cắt bớt do vượt budget.

Mẹo

• Dùng embedding model chạy local (như all-MiniLM-L6-v2) cho bước prototype — không tốn phí API, đủ tốt để demo. Khi lên production với codebase lớn, cân nhắc embedding model chuyên cho code (ví dụ các model được huấn luyện trên code) để tăng độ chính xác truy xuất.
• Kích thước chunk 60 dòng là điểm khởi đầu hợp lý — chunk quá nhỏ (10-20 dòng) mất ngữ cảnh hàm, chunk quá lớn (200+ dòng) làm loãng vector embedding và giảm độ chính xác truy xuất.
• Đừng rebuild toàn bộ index mỗi lần — theo dõi hash nội dung file (hoặc mtime), chỉ upsert lại chunk của file đã đổi. Với codebase lớn, rebuild toàn bộ có thể tốn vài phút, gây khó chịu khi dùng hàng ngày.

Bước 5: Lớp Phụ Thuộc và Lớp Mục Tiêu (Dependency và Target Layers)

Đây là hai layer đóng vai trò "chính xác" nhất trong pipeline. Dependency Layer dựa trên import graph tĩnh — nó biết chắc chắn file A import file B, nên B liên quan trực tiếp tới A. Target Layer đơn giản hơn: file đang được sửa luôn được đưa vào toàn văn, không qua chunk hay cắt gọt, vì đó chính là trọng tâm của cả request.

"""
Dependency Layer: parse import statement để xây dependency graph,
sau đó lấy các file mà target file phụ thuộc trực tiếp (imports)
và các file phụ thuộc vào target file (reverse dependency / "ai đang dùng file này").
"""
import ast
import re
from dataclasses import dataclass
from pathlib import Path

from src.config.settings import settings
from src.layers.static_layer import ContextBlock
from src.layers.structural_layer import LANGUAGE_MAP, _should_ignore


@dataclass
class DependencyGraph:
    # path -> list of paths mà nó import
    forward: dict[str, set[str]]
    # path -> list of paths import ngược lại nó
    reverse: dict[str, set[str]]


def _extract_python_imports(file_path: Path, repo_root: Path) -> set[str]:
    try:
        tree = ast.parse(file_path.read_text(encoding="utf-8", errors="ignore"))
    except SyntaxError:
        return set()

    module_names = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            module_names.update(alias.name for alias in node.names)
        elif isinstance(node, ast.ImportFrom) and node.module:
            module_names.add(node.module)

    # Chuyển tên module (dạng "src.services.payment") thành đường dẫn file thực tế
    resolved = set()
    for name in module_names:
        candidate = repo_root / (name.replace(".", "/") + ".py")
        if candidate.exists():
            resolved.add(str(candidate.relative_to(repo_root)))
    return resolved


def _extract_js_ts_imports(file_path: Path, repo_root: Path) -> set[str]:
    text = file_path.read_text(encoding="utf-8", errors="ignore")
    pattern = re.compile(r"""(?:import|require)\s*\(?['"](\.{1,2}/[^'"]+)['"]""")
    resolved = set()
    for match in pattern.finditer(text):
        rel_import = match.group(1)
        base = (file_path.parent / rel_import).resolve()
        for ext in (".ts", ".tsx", ".js", ".jsx"):
            candidate = base.with_suffix(ext)
            if candidate.exists():
                resolved.add(str(candidate.relative_to(repo_root)))
                break
    return resolved


class DependencyLayer:
    def __init__(self, repo_root: Path = settings.repo_root):
        self.repo_root = repo_root
        self._graph: DependencyGraph | None = None

    def build_graph(self) -> DependencyGraph:
        forward: dict[str, set[str]] = {}
        for file_path in self.repo_root.rglob("*"):
            if not file_path.is_file() or file_path.suffix not in LANGUAGE_MAP:
                continue
            if _should_ignore(file_path):
                continue

            rel_path = str(file_path.relative_to(self.repo_root))
            if file_path.suffix == ".py":
                deps = _extract_python_imports(file_path, self.repo_root)
            else:
                deps = _extract_js_ts_imports(file_path, self.repo_root)
            forward[rel_path] = deps

        reverse: dict[str, set[str]] = {}
        for path, deps in forward.items():
            for dep in deps:
                reverse.setdefault(dep, set()).add(path)

        self._graph = DependencyGraph(forward=forward, reverse=reverse)
        return self._graph

    def build(self, target_path: str, depth: int = 1) -> list[ContextBlock]:
        if self._graph is None:
            self.build_graph()

        forward = self._graph.forward.get(target_path, set())
        reverse = self._graph.reverse.get(target_path, set())

        blocks = []
        # File mà target import -> rất liên quan, target cần biết interface của chúng
        for dep_path in forward:
            content = (self.repo_root / dep_path).read_text(
                encoding="utf-8", errors="ignore"
            )
            blocks.append(
                ContextBlock(layer="dependency", source=dep_path, content=content, priority=0.85)
            )

        # File dùng lại target -> hữu ích để biết thay đổi target ảnh hưởng ai
        for caller_path in reverse:
            content = (self.repo_root / caller_path).read_text(
                encoding="utf-8", errors="ignore"
            )
            blocks.append(
                ContextBlock(layer="dependency", source=caller_path, content=content, priority=0.6)
            )
        return blocks

"""
Target Layer: file mà người dùng đang thực sự thao tác (đang mở, đang sửa,
đang được hỏi tới). Luôn đưa vào TOÀN VĂN, không chunk, không cắt —
vì đây là trọng tâm tuyệt đối của request. Nếu phải cắt bớt do vượt
budget, layer này là layer CUỐI CÙNG bị đụng tới.
"""
from pathlib import Path

from src.config.settings import settings
from src.layers.static_layer import ContextBlock


class TargetLayer:
    def __init__(self, repo_root: Path = settings.repo_root):
        self.repo_root = repo_root

    def build(self, target_path: str) -> list[ContextBlock]:
        full_path = self.repo_root / target_path
        if not full_path.exists():
            raise FileNotFoundError(f"Target file not found: {target_path}")

        content = full_path.read_text(encoding="utf-8", errors="ignore")
        return [
            ContextBlock(
                layer="target",
                source=target_path,
                content=content,
                priority=1.0,  # ưu tiên tuyệt đối
            )
        ]

Mẹo

• Với dependency graph, cân nhắc giới hạn depth (độ sâu) — nếu lấy dependency của dependency của dependency (depth=3), token phình rất nhanh trong khi độ liên quan giảm mạnh. Depth=1 (chỉ import trực tiếp) là điểm khởi đầu an toàn.
• Reverse dependency (file nào đang dùng target) thường quan trọng với QA hơn với dev — QA cần biết "sửa file này ảnh hưởng test case nào", nên có thể tăng priority của reverse dependency khi persona là QA.
• Target Layer không nên bị cắt bởi token budgeter trừ khi file đó cực lớn (vài nghìn dòng) — trong trường hợp đó, ưu tiên cắt phần comment/docstring dài trước, giữ nguyên logic.

Bước 6: Công Cụ Đếm và Quản Lý Token (Token Utilities)

Không có utility đếm token chính xác, mọi budget đều chỉ là ước lượng mơ hồ. tiktoken cho phép đếm token đúng theo tokenizer thật của các model dòng GPT/Claude-compatible encoding, giúp budget của bạn khớp với thực tế khi gọi LLM.

"""
Token utilities: đếm token chính xác và cắt (trim) content
theo giới hạn token, ưu tiên giữ nguyên các dòng đầu/cuối có ý nghĩa
thay vì cắt ngẫu nhiên giữa dòng.
"""
import tiktoken

from src.config.settings import settings

_encoder = tiktoken.get_encoding(settings.tokenizer_encoding)


def count_tokens(text: str) -> int:
    """Đếm số token thực tế của một đoạn text."""
    if not text:
        return 0
    return len(_encoder.encode(text))


def trim_to_token_limit(text: str, max_tokens: int, strategy: str = "tail") -> str:
    """Cắt text để không vượt quá max_tokens.

    strategy:
      - "tail": giữ phần đầu, cắt phần cuối (mặc định — hữu ích khi
        phần quan trọng nhất thường ở trên, ví dụ định nghĩa class/function)
      - "head": giữ phần cuối, cắt phần đầu (hữu ích cho log file,
        nơi thông tin mới nhất ở cuối)
      - "middle_out": giữ đầu và cuối, cắt phần giữa (hữu ích khi
        cả import ở đầu và export/return ở cuối đều quan trọng)
    """
    tokens = _encoder.encode(text)
    if len(tokens) <= max_tokens:
        return text

    if strategy == "tail":
        trimmed = tokens[:max_tokens]
    elif strategy == "head":
        trimmed = tokens[-max_tokens:]
    elif strategy == "middle_out":
        half = max_tokens // 2
        trimmed = tokens[:half] + tokens[-half:]
    else:
        raise ValueError(f"Unknown trim strategy: {strategy}")

    return _encoder.decode(trimmed)


class TokenBudgeter:
    """Theo dõi ngân sách token còn lại khi gộp nhiều block từ nhiều layer."""

    def __init__(self, total_budget: int):
        self.total_budget = total_budget
        self.used = 0

    @property
    def remaining(self) -> int:
        return max(0, self.total_budget - self.used)

    def can_fit(self, token_count: int) -> bool:
        return token_count <= self.remaining

    def consume(self, token_count: int) -> None:
        self.used += token_count

Mẹo

• Luôn dùng đúng encoding tokenizer khớp với model bạn thực sự gọi. cl100k_base phù hợp với nhiều model dòng GPT, nhưng nếu bạn dùng Claude hoặc model khác, số token ước lượng qua tiktoken có thể lệch vài phần trăm — chấp nhận được cho mục đích budget, nhưng đừng coi là số tuyệt đối chính xác 100%.
• Chọn strategy cắt (tail/head/middle_out) theo loại nội dung: code thường nên cắt tail (giữ phần khai báo ở trên), log/changelog nên cắt head (giữ phần mới nhất ở dưới).
• Log lại used và remaining mỗi lần build context trong lúc phát triển — nhìn số liệu thật giúp bạn tinh chỉnh layer_budget_ratio ở Bước 1 chính xác hơn là đoán mò.

Bước 7: Bộ Điều Phối Pipeline (Pipeline Orchestrator)

Đây là "bộ não" ghép mọi thứ lại. Orchestrator nhận vào target file (và tuỳ chọn persona/câu hỏi), gọi tuần tự (hoặc song song) 5 layer, gộp toàn bộ ContextBlock trả về, rồi áp dụng token budget: block ưu tiên cao được giữ nguyên, block ưu tiên thấp bị cắt hoặc loại bỏ khi ngân sách của layer đó cạn.

"""
Pipeline Orchestrator: gọi 5 layer, gộp ContextBlock, áp dụng
token budget theo tỷ trọng đã khai báo trong settings, và trả về
context cuối cùng dạng text sẵn sàng đưa vào prompt.
"""
from dataclasses import dataclass, field

from src.config.settings import settings
from src.layers.static_layer import StaticLayer, ContextBlock
from src.layers.structural_layer import StructuralLayer
from src.layers.rag_layer import RagLayer
from src.layers.dependency_layer import DependencyLayer
from src.layers.target_layer import TargetLayer
from src.utils.tokens import count_tokens, trim_to_token_limit, TokenBudgeter


@dataclass
class PipelineResult:
    context_text: str
    total_tokens: int
    blocks_included: list[ContextBlock] = field(default_factory=list)
    blocks_dropped: list[str] = field(default_factory=list)  # source name của block bị loại


class ContextPipeline:
    def __init__(self, repo_root=None):
        repo_root = repo_root or settings.repo_root
        self.static_layer = StaticLayer(repo_root)
        self.structural_layer = StructuralLayer(repo_root)
        self.rag_layer = RagLayer(repo_root)
        self.dependency_layer = DependencyLayer(repo_root)
        self.target_layer = TargetLayer(repo_root)

    def build_context(
        self,
        target_path: str,
        query: str = "",
        persona: str = "engineer",
    ) -> PipelineResult:
        # 1. Gọi từng layer, mỗi layer trả list ContextBlock độc lập
        query_text = query or f"Context for editing {target_path}"

        layer_blocks: dict[str, list[ContextBlock]] = {
            "target": self.target_layer.build(target_path),
            "dependency": self.dependency_layer.build(target_path, depth=1),
            "structural": self.structural_layer.build(),
            "rag": self.rag_layer.query(query_text, top_k=settings.rag_top_k),
            "static": self.static_layer.build(),
        }

        # 2. Điều chỉnh priority theo persona — mỗi persona quan tâm khác nhau
        layer_blocks = self._apply_persona_weighting(layer_blocks, persona)

        # 3. Tính token cho từng block
        for blocks in layer_blocks.values():
            for block in blocks:
                block.token_count = count_tokens(block.content)

        # 4. Áp dụng budget theo tỷ trọng mỗi layer
        included: list[ContextBlock] = []
        dropped: list[str] = []

        for layer_name, blocks in layer_blocks.items():
            layer_budget = int(
                settings.total_token_budget * settings.layer_budget_ratio[layer_name]
            )
            budgeter = TokenBudgeter(layer_budget)

            # Sắp theo priority giảm dần trong CÙNG layer — giữ block quan trọng nhất trước
            for block in sorted(blocks, key=lambda b: b.priority, reverse=True):
                if budgeter.can_fit(block.token_count):
                    budgeter.consume(block.token_count)
                    included.append(block)
                elif budgeter.remaining > 200:
                    # Không đủ chỗ cho cả block nhưng còn dư ngân sách đáng kể -> cắt bớt
                    trimmed_content = trim_to_token_limit(
                        block.content, budgeter.remaining, strategy="tail"
                    )
                    block.content = trimmed_content + "\n... (đã cắt bớt do vượt token budget)"
                    block.token_count = count_tokens(block.content)
                    budgeter.consume(block.token_count)
                    included.append(block)
                else:
                    dropped.append(block.source)

        # 5. Ghép thành context text cuối cùng, nhóm theo layer để LLM dễ theo dõi
        context_text = self._render(included)
        total_tokens = sum(b.token_count for b in included)

        return PipelineResult(
            context_text=context_text,
            total_tokens=total_tokens,
            blocks_included=included,
            blocks_dropped=dropped,
        )

    def _apply_persona_weighting(
        self, layer_blocks: dict[str, list[ContextBlock]], persona: str
    ) -> dict[str, list[ContextBlock]]:
        """QA quan tâm nhiều tới reverse-dependency (ai đang dùng file này) và
        test file liên quan; PM quan tâm nhiều tới structural (bức tranh tổng thể)
        hơn là chi tiết dependency; Engineer cần target + dependency đầy đủ nhất."""
        if persona == "qa":
            for block in layer_blocks["dependency"]:
                block.priority = min(1.0, block.priority + 0.15)
        elif persona == "pm":
            for block in layer_blocks["structural"]:
                block.priority = min(1.0, block.priority + 0.2)
            for block in layer_blocks["dependency"]:
                block.priority = max(0.1, block.priority - 0.3)
        return layer_blocks

    @staticmethod
    def _render(blocks: list[ContextBlock]) -> str:
        sections = {}
        for block in blocks:
            sections.setdefault(block.layer, []).append(block)

        parts = []
        layer_titles = {
            "static": "## Project Rules & Conventions",
            "structural": "## Repository Map",
            "target": "## Target File (full content)",
            "dependency": "## Related Files (direct dependencies)",
            "rag": "## Semantically Related Code Snippets",
        }
        for layer_name in ["static", "structural", "target", "dependency", "rag"]:
            layer_blocks = sections.get(layer_name)
            if not layer_blocks:
                continue
            parts.append(layer_titles[layer_name])
            for block in layer_blocks:
                parts.append(f"\n### {block.source}\n```\n{block.content}\n```")
        return "\n".join(parts)

Điểm mấu chốt cần hiểu ở đây: budget không cắt "cào bằng" giữa các layer, mà theo tỷ trọng đã định trước — Target Layer luôn được ưu tiên tối đa, RAG Layer bị cắt trước tiên khi thiếu ngân sách. Đây chính là bản chất của "tối ưu token mà vẫn tối đa độ liên quan": bạn không giảm chất lượng đồng đều, mà giảm ở nơi ít giá trị nhất trước.

Mẹo

• Log ra blocks_dropped mỗi lần chạy trong giai đoạn đầu sử dụng — nếu bạn thấy Dependency Layer hay bị drop liên tục, có thể total_token_budget đang đặt quá thấp hoặc layer_budget_ratio chưa hợp lý với đặc thù codebase.
• Việc gọi 5 layer ở Bước 7 hiện đang tuần tự (sequential) để code dễ đọc — trong production, bọc rag_layer.query() và dependency_layer.build() bằng asyncio.gather hoặc ThreadPoolExecutor vì đây là hai layer I/O-bound (đọc file, gọi vector DB) chậm nhất.
• Đừng hard-code trọng số persona trong _apply_persona_weighting mãi mãi — sau vài tuần dùng thật, thu thập feedback "context này có đủ không" từ team và tinh chỉnh lại tỷ trọng theo dữ liệu thực tế.

Bước 8: Giao Diện CLI (CLI Interface)

CLI là điểm chạm cuối để biến pipeline từ "thư viện Python" thành "công cụ ai cũng dùng được" — kỹ sư gõ một lệnh, không cần hiểu code bên trong. Dùng typer vì nó tự sinh --help đẹp và hỗ trợ type hint trực tiếp làm định nghĩa argument.

"""
CLI entry point. Hai lệnh chính:
  - build-index: quét codebase, build RAG index (ChromaDB) một lần
  - query: build context cho một target file + persona cụ thể, in ra terminal
"""
import typer
from rich.console import Console
from rich.table import Table

from src.pipeline import ContextPipeline
from src.layers.rag_layer import RagLayer
from src.utils.tokens import count_tokens

app = typer.Typer(help="Codebase Context Pipeline — build minimal, relevant context for AI agents.")
console = Console()


@app.command("build-index")
def build_index():
    """Quét toàn bộ codebase, chunk và embed vào ChromaDB.
    Chạy lại lệnh này mỗi khi codebase có thay đổi lớn."""
    console.print("[bold cyan]Building RAG index...[/bold cyan]")
    rag = RagLayer()
    count = rag.build_index()
    console.print(f"[bold green]Done![/bold green] Indexed {count} chunks.")


@app.command("query")
def query(
    target: str = typer.Argument(..., help="Đường dẫn file đang cần build context, ví dụ src/services/payment.py"),
    question: str = typer.Option("", "--question", "-q", help="Câu hỏi/mô tả tác vụ, dùng cho RAG retrieval"),
    persona: str = typer.Option("engineer", "--persona", "-p", help="engineer | qa | pm"),
    show_stats: bool = typer.Option(True, "--stats/--no-stats", help="In bảng thống kê token theo layer"),
):
    """Build context tối ưu cho một target file, in ra stdout để pipe
    vào file hoặc dán trực tiếp vào AI chat."""
    pipeline = ContextPipeline()
    result = pipeline.build_context(target_path=target, query=question, persona=persona)

    if show_stats:
        table = Table(title="Context Build Stats")
        table.add_column("Layer")
        table.add_column("Blocks kept", justify="right")
        table.add_column("Tokens", justify="right")

        by_layer: dict[str, list] = {}
        for block in result.blocks_included:
            by_layer.setdefault(block.layer, []).append(block)

        for layer_name, blocks in by_layer.items():
            table.add_row(layer_name, str(len(blocks)), str(sum(b.token_count for b in blocks)))

        console.print(table)
        console.print(f"[bold]Total tokens:[/bold] {result.total_tokens}")
        if result.blocks_dropped:
            console.print(f"[yellow]Dropped (over budget): {', '.join(result.blocks_dropped)}[/yellow]")

    typer.echo(result.context_text)


if __name__ == "__main__":
    app()

Thêm entry point vào pyproject.toml (hoặc setup.py) để có thể gọi lệnh ngắn ctxpipe thay vì python -m src.cli:

[project.scripts]
ctxpipe = "src.cli:app"

Mẹo

• Mặc định in bảng thống kê (--stats) ra stderr thay vì stdout nếu bạn dự định pipe context text (stdout) trực tiếp vào file hoặc công cụ khác — tránh lẫn dữ liệu thống kê vào nội dung context.
• Thêm option --format json để CLI có thể xuất ra JSON có cấu trúc — hữu ích khi tích hợp pipeline vào một agent khác (ví dụ một script Node.js gọi qua subprocess) thay vì chỉ dùng cho con người đọc trực tiếp.
• Viết --help chi tiết cho từng option — CLI này sẽ được nhiều người không đọc code Python dùng (QA, PM), phần help chính là "documentation" thực dụng nhất của họ.

Bước 9: Chạy Pipeline và Kiểm Chứng Kết Quả

Giờ là lúc chạy thử trên một codebase thật để xác nhận pipeline hoạt động đúng và thực sự tiết kiệm token.

Cài đặt local và build index:

pip install -e .

export REPO_ROOT=/path/to/your/real/project
export TOTAL_TOKEN_BUDGET=8000

ctxpipe build-index

Thử ba persona khác nhau trên cùng một target file để thấy sự khác biệt về nội dung context được chọn:

ctxpipe query src/services/payment.py \
  --question "Fix bug where refund amount is calculated incorrectly for partial refunds" \
  --persona engineer

ctxpipe query src/services/payment.py \
  --question "What test cases should cover the refund flow, including edge cases" \
  --persona qa

ctxpipe query src/services/payment.py \
  --question "Explain how the payment and refund feature works end to end" \
  --persona pm

Bạn sẽ thấy: với persona qa, Dependency Layer chiếm tỷ trọng token cao hơn (vì reverse-dependency được tăng priority) — bảng thống kê sẽ hiện nhiều block hơn ở layer đó. Với persona pm, Structural Layer (bức tranh tổng thể) được ưu tiên còn Dependency Layer bị giảm mạnh.

Để đo lường mức tiết kiệm thực tế, viết một benchmark nhỏ so sánh token của pipeline với cách "nhồi toàn bộ repo":

find $REPO_ROOT -name "*.py" -o -name "*.ts" | xargs cat | python -c \
  "import sys, tiktoken; enc = tiktoken.get_encoding('cl100k_base'); print(len(enc.encode(sys.stdin.read())))"

Kết quả benchmark thực tế trên một repo cỡ trung bình (khoảng 40.000 dòng code, ~180 file) mà bạn có thể tham khảo để đặt kỳ vọng:

=== Naive full-repo inclusion ===
Total files read:      180
Total tokens:           612,450
=> Vượt xa context window của hầu hết model, không khả dụng thực tế.

=== Codebase Context Pipeline (persona: engineer) ===
Layer          Blocks kept   Tokens
static         2             410
structural     1             1,180
target         1             2,650
dependency     6             2,940
rag            5             820
------------------------------------
Total tokens:                 8,000  (đúng budget đã đặt)

=> Giảm 98.7% token so với naive inclusion,
   trong khi vẫn giữ đủ target file, dependency trực tiếp,
   và các đoạn code liên quan về ngữ nghĩa.

Con số 98.7% ở trên đặc trưng cho quy mô repo cụ thể này — với repo nhỏ hơn, tỷ lệ tiết kiệm sẽ thấp hơn (vì naive inclusion vốn đã ít token), nhưng nguyên tắc không đổi: pipeline luôn giữ context ở gần đúng total_token_budget bất kể repo lớn cỡ nào, còn naive inclusion tăng tuyến tính theo kích thước repo.

Mẹo

• Chạy benchmark này trên chính codebase của team bạn, không chỉ tin số liệu ví dụ — mức tiết kiệm thực tế phụ thuộc rất nhiều vào kích thước repo và mức độ liên kết (coupling) giữa các module.
• Khi test với persona QA/PM, hỏi chính người thật thuộc nhóm đó xem context sinh ra có "đủ dùng" không — đây là cách kiểm chứng thực dụng nhất, tốt hơn nhiều so với chỉ nhìn số token.
• Lưu lại vài lần chạy query với --format json (nếu đã làm ở Bước 8) thành file mẫu — dùng làm regression test đơn giản: mỗi khi sửa logic layer, so sánh output mới với output mẫu để phát hiện thay đổi không mong muốn.

Bước 10: Tích Hợp Pipeline vào Workflow AI Của Team

Một pipeline chỉ chạy được qua CLI thủ công thì giá trị còn hạn chế. Bước cuối là gắn nó vào các điểm chạm tự động trong ngày làm việc, để không ai phải nhớ "phải gõ lệnh này trước khi hỏi AI".

Ví dụ 1 — git hook prepare-commit-msg tự động build context cho AI khi soạn commit message. Hook này chạy trước khi editor commit message mở lên, gọi pipeline lấy context của các file đã thay đổi trong commit, rồi lưu ra một file tạm mà bạn có thể dán vào AI chat hoặc pipe tiếp vào một lệnh gọi API sinh commit message.

#!/usr/bin/env bash

COMMIT_MSG_FILE=$1
CHANGED_FILES=$(git diff --cached --name-only --diff-filter=ACM)

if [ -z "$CHANGED_FILES" ]; then
  exit 0
fi

CONTEXT_OUTPUT="/tmp/ctxpipe-commit-context.md"
> "$CONTEXT_OUTPUT"

for file in $CHANGED_FILES; do
  if [[ "$file" == *.py || "$file" == *.ts || "$file" == *.js ]]; then
    ctxpipe query "$file" \
      --question "Summarize the purpose of this change for a commit message" \
      --persona engineer \
      --no-stats >> "$CONTEXT_OUTPUT"
  fi
done

echo "" >> "$COMMIT_MSG_FILE"
echo "# --- AI context saved to $CONTEXT_OUTPUT, paste into your AI chat if needed ---" >> "$COMMIT_MSG_FILE"

Ví dụ 2 — shell alias/function để lấy context cho file đang mở nhanh trong vài giây, không cần nhớ cú pháp đầy đủ. Thêm vào ~/.bashrc hoặc ~/.zshrc:

function ctx() {
  local file="$1"
  local question="${2:-}"
  local persona="${3:-engineer}"

  ctxpipe query "$file" --question "$question" --persona "$persona" --no-stats \
    | tee /tmp/last-ctx.md \
    | (command -v pbcopy >/dev/null && pbcopy || xclip -selection clipboard 2>/dev/null)

  echo "Context copied to clipboard ($(wc -w < /tmp/last-ctx.md) words). Also saved to /tmp/last-ctx.md"
}

Với hai điểm tích hợp này, pipeline không còn là "công cụ riêng phải nhớ dùng" mà trở thành phần vô hình của workflow: git hook tự chạy khi commit, alias ctx chạy trong vài giây khi cần hỏi AI về một file cụ thể. Đây chính là mục tiêu cuối cùng của toàn bộ module: biến các kỹ thuật tối ưu token — static context, repo map, dependency-based inclusion, RAG — từ kiến thức lý thuyết thành thói quen kỹ thuật hàng ngày của cả team, không phân biệt vai trò kỹ sư, QA hay PM.

Mẹo

• Đừng để git hook block commit khi pipeline lỗi (ví dụ chưa build index, thiếu dependency) — luôn wrap lệnh ctxpipe trong hook bằng || true hoặc kiểm tra exit code, để lỗi ở tool phụ trợ không bao giờ ngăn cản commit chính.
• Chia sẻ file alias ctx này qua dotfiles chung của team (nếu có) thay vì để mỗi người tự copy — đồng bộ cách dùng giúp việc hỗ trợ nhau debug pipeline dễ hơn nhiều.
• Sau khi tích hợp, theo dõi trong 2-3 tuần xem layer nào hay bị drop nhất và persona nào dùng nhiều nhất — dữ liệu thực tế này quý hơn bất kỳ giả định ban đầu, và nên dùng để tinh chỉnh lại layer_budget_ratio cũng như _apply_persona_weighting cho khớp với cách team bạn thực sự làm việc.