Giới thiệu: Context Engineering -- Kỷ Nguyên Mới của Phát Triển Ứng Dụng AI
Năm 2026 này thực sự là một năm đặc biệt đối với ai làm về AI. LLM giờ đã mạnh lắm rồi -- nhưng bạn biết không, thách thức lớn nhất bây giờ không phải ở model mạnh hay yếu, mà ở cách chúng ta "feed" thông tin cho nó. Đó chính là lý do Context Engineering (hay thiết kế ngữ cảnh, nghe fancy nhỉ?) giờ đã trở thành kỹ năng cốt lõi mà mọi kỹ sư AI đều cần nắm vững.
Context Engineering là kỷ luật xây dựng các hệ thống động (dynamic systems) có khả năng cung cấp cho LLM mọi thứ nó cần -- đúng thông tin, đúng định dạng, đúng thời điểm -- để hoàn thành nhiệm vụ một cách chính xác nhất. Để dễ hình dung, bạn cứ nghĩ LLM là CPU còn context window là RAM -- đúng kiểu đó luôn. Giống như một chương trình cần dữ liệu phù hợp nạp vào RAM để CPU xử lý, LLM cần ngữ cảnh được thiết kế cẩn thận để tạo ra kết quả chất lượng cao.
Gartner có báo cáo gần đây (khá ấn tượng) là đến năm 2026, 40% ứng dụng doanh nghiệp sẽ tích hợp AI agent. Con số này cho thấy nhu cầu cấp thiết về một phương pháp luận có hệ thống để quản lý ngữ cảnh cho ứng dụng AI trong production. Context Engineering chính là câu trả lời.
Định nghĩa: Context Engineering là nghệ thuật và khoa học thiết kế, xây dựng, và tối ưu hóa toàn bộ chuỗi thông tin (information pipeline) được cung cấp cho mô hình AI, bao gồm system prompt, dữ liệu truy xuất, bộ nhớ hội thoại, mô tả công cụ, và cấu trúc đầu ra -- tất cả hoạt động như một hệ thống thống nhất.
Bài viết này sẽ đi sâu vào từng thành phần của Context Engineering, kèm theo code thực tế, và một mẫu kiến trúc production hoàn chỉnh mà bạn có thể áp dụng ngay vào dự án của mình.
Context Engineering vs Prompt Engineering: Sự Khác Biệt Cốt Lõi
Okay, trước khi lao vào code, mình cần làm rõ hai khái niệm này vì nhiều người hay nhầm lắm: Prompt Engineering và Context Engineering -- chúng khác nhau đấy nhé!
Prompt Engineering: Điểm Khởi Đầu
Prompt Engineering thì đơn giản hơn -- chủ yếu là viết và chỉnh prompt sao cho hay thôi. Nó quan trọng đấy, nhưng thật ra chỉ giải quyết được một mảnh nhỏ của bài toán. Prompt Engineering giống như việc viết một câu hỏi hay cho một kỳ thi -- bạn tối ưu cách đặt câu hỏi, nhưng không kiểm soát được kiến thức mà người trả lời có.
Context Engineering: Bức Tranh Toàn Cảnh
Còn Context Engineering thì rộng hơn nhiều, nó bao trùm cả một hệ thống thông tin. Không chỉ quan tâm đến cách đặt câu hỏi, mà còn bao gồm:
- Dữ liệu nào cần được truy xuất và từ nguồn nào
- Bộ nhớ hội thoại nào cần được giữ lại và khi nào cần nén hoặc loại bỏ
- Công cụ nào cần được mô tả để LLM biết cách sử dụng
- Cấu trúc đầu ra nào cần được yêu cầu để hệ thống downstream có thể xử lý
- Chiến lược tối ưu chi phí khi vận hành trong production
Để so sánh cho dễ hiểu: Prompt Engineering giống như bạn viết một email hay, còn Context Engineering thì bạn đang thiết kế cả hệ thống giao tiếp của công ty -- từ quy trình, kênh truyền thông, lưu trữ, đến metrics tracking. Khác nhau xa đúng không?
Tại Sao Prompt Engineering Đơn Thuần Không Đủ?
Khi đưa lên production thực tế, có vài thách thức mà prompt engineering một mình không "cân" nổi:
- Dữ liệu thay đổi liên tục: Giá sản phẩm hôm nay khác ngày mai, chính sách tuần này sửa tuần sau -- bạn không thể cứ hard-code vào prompt được.
- Giới hạn context window: Dù các model mới nhất hỗ trợ hàng triệu token, việc nhồi nhét mọi thứ vào context vừa tốn kém vừa giảm hiệu suất.
- Đa dạng tác vụ: Một AI agent có thể cần gọi hàng chục công cụ khác nhau -- liệt kê tất cả trong prompt? Không khả thi.
- Yêu cầu về tính nhất quán: Hệ thống phải nhớ ngữ cảnh qua nhiều lượt hội thoại và nhiều phiên làm việc.
- Chi phí vận hành: Mỗi token input đều tốn tiền -- cần tối ưu hóa thông minh.
5 Thành Phần Cốt Lõi của Context Engineering
Một hệ thống Context Engineering đàng hoàng cần có 5 thành phần chính -- thiếu cái nào cũng thấy "kì kì".
1. System Prompts (Lệnh Hệ Thống)
System prompt là cái nền, nó định nghĩa AI của bạn "là ai", làm gì, và được phép làm gì. Tuy là thành phần "tĩnh" nhất nhưng lại quan trọng nhất -- coi như là "personality" của agent vậy.
2. RAG (Retrieval-Augmented Generation)
RAG cho phép LLM truy xuất thông tin bên ngoài -- từ một hệ thống chỉ biết kiến thức được train, giờ nó có thể kéo data real-time về dùng. Năm 2026, RAG 2.0 với semantic filtering và multi-hop retrieval đã trở thành tiêu chuẩn.
3. Memory (Bộ Nhớ)
Bộ nhớ cho phép AI agent duy trì ngữ cảnh qua nhiều lượt tương tác, bao gồm bộ nhớ ngắn hạn (conversation buffer) và bộ nhớ dài hạn (persistent memory).
4. Tools (Công Cụ)
Mô tả công cụ cho phép LLM biết những hành động nào nó có thể thực hiện -- từ tìm kiếm cơ sở dữ liệu, gọi API, đến thực thi mã nguồn. Quản lý tool description tốt hay dở ảnh hưởng rất nhiều đến chất lượng đầu ra.
5. Structured Outputs (Đầu Ra Có Cấu Trúc)
Đầu ra có cấu trúc đảm bảo LLM trả về dữ liệu theo định dạng mà hệ thống downstream có thể xử lý -- JSON schema, function calls, hoặc các format tùy chỉnh.
Xây Dựng Hệ Thống System Prompt Hiệu Quả
System prompt là thành phần đầu tiên bạn cần làm tốt. Mình có mấy nguyên tắc hay ho này, bạn tham khảo nhé:
Nguyên Tắc Thiết Kế System Prompt
- Rõ ràng về vai trò: Xác định chính xác AI agent là ai và làm gì
- Có ranh giới: Định nghĩa rõ những gì agent được và không được làm
- Có cấu trúc: Sử dụng markdown, XML tags, hoặc các dấu phân cách rõ ràng
- Có ví dụ: Cung cấp few-shot examples cho các tình huống phổ biến
- Có thể cập nhật: Thiết kế để dễ dàng thay đổi mà không ảnh hưởng toàn bộ hệ thống
Dưới đây là ví dụ triển khai hệ thống system prompt có cấu trúc trong Python:
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime
@dataclass
class SystemPromptConfig:
"""Cấu hình system prompt có cấu trúc cho AI agent."""
role: str
objective: str
constraints: List[str] = field(default_factory=list)
tools_available: List[str] = field(default_factory=list)
output_format: str = "text"
language: str = "vi"
few_shot_examples: List[dict] = field(default_factory=list)
version: str = "1.0"
def build_prompt(self) -> str:
"""Xây dựng system prompt từ cấu hình."""
sections = []
# Phần vai trò
sections.append(f"# Vai Trò\nBạn là {self.role}.")
# Phần mục tiêu
sections.append(f"# Mục Tiêu\n{self.objective}")
# Phần ràng buộc
if self.constraints:
constraints_text = "\n".join(
f"- {c}" for c in self.constraints
)
sections.append(f"# Ràng Buộc\n{constraints_text}")
# Phần công cụ
if self.tools_available:
tools_text = "\n".join(
f"- {t}" for t in self.tools_available
)
sections.append(f"# Công Cụ Có Sẵn\n{tools_text}")
# Phần định dạng đầu ra
sections.append(
f"# Định Dạng Đầu Ra\n"
f"Trả lời bằng tiếng {'Việt' if self.language == 'vi' else 'Anh'}. "
f"Định dạng: {self.output_format}"
)
# Phần ví dụ
if self.few_shot_examples:
examples_text = ""
for i, ex in enumerate(self.few_shot_examples, 1):
examples_text += (
f"\n## Ví dụ {i}\n"
f"**Input:** {ex['input']}\n"
f"**Output:** {ex['output']}\n"
)
sections.append(f"# Ví Dụ Minh Họa{examples_text}")
# Metadata
sections.append(
f"# Metadata\n"
f"- Phiên bản prompt: {self.version}\n"
f"- Ngày cập nhật: {datetime.now().strftime('%Y-%m-%d')}"
)
return "\n\n".join(sections)
# Ví dụ sử dụng
customer_support_config = SystemPromptConfig(
role="chuyên viên hỗ trợ khách hàng cao cấp của công ty TechVN",
objective=(
"Hỗ trợ khách hàng giải quyết vấn đề kỹ thuật, trả lời "
"câu hỏi về sản phẩm, và chuyển tiếp đến bộ phận phù hợp "
"khi cần thiết."
),
constraints=[
"Không được chia sẻ thông tin nội bộ công ty",
"Luôn xác nhận danh tính khách hàng trước khi truy cập tài khoản",
"Không được đưa ra cam kết về hoàn tiền mà không có sự phê duyệt",
"Ghi nhận mọi khiếu nại vào hệ thống ticket",
],
tools_available=[
"search_knowledge_base: Tìm kiếm tài liệu hỗ trợ",
"lookup_customer: Tra cứu thông tin khách hàng",
"create_ticket: Tạo ticket hỗ trợ mới",
"escalate_to_human: Chuyển tiếp đến nhân viên",
],
output_format="JSON với các trường: response, action, confidence",
few_shot_examples=[
{
"input": "Tôi không đăng nhập được vào tài khoản",
"output": '{"response": "Tôi hiểu bạn đang gặp khó khăn khi đăng nhập. Để hỗ trợ bạn, tôi cần xác nhận email đăng ký tài khoản của bạn.", "action": "ask_verification", "confidence": 0.95}'
}
]
)
system_prompt = customer_support_config.build_prompt()
print(system_prompt)Làm kiểu này có nhiều ưu điểm lắm: system prompt giờ là code bình thường -- git commit được, test được, deploy cũng chuẩn. Không còn cảnh copy-paste prompt loạn xị ngầu nữa. Khi cần thay đổi hành vi của agent, bạn chỉ cần cập nhật cấu hình thay vì sửa trực tiếp chuỗi văn bản dài.
Tích Hợp RAG vào Pipeline Context
RAG thì chắc ai cũng biết rồi nhỉ? Nhưng năm nay, RAG đã "lên đời" khá nhiều -- gọi là RAG 2.0 cho oai. Nó tập trung vào semantic filtering, multi-hop retrieval, và adaptive chunking. Nghe phức tạp nhưng xài thì thấy hiệu quả thật.
RAG 2.0: Những Cải Tiến Quan Trọng
So với RAG "cũ kỹ" ngày trước, bản 2.0 này xịn hơn hẳn:
- Semantic Chunking: Thay vì cứ cắt đều từng 500 ký tự một cách máy móc, giờ nó phân tích ngữ nghĩa rồi mới cắt -- cho ra đoạn văn bản mạch lạc hơn nhiều.
- Multi-hop Retrieval: Khả năng truy xuất thông tin qua nhiều bước, kết nối các mảnh kiến thức liên quan từ nhiều nguồn khác nhau. Câu hỏi phức tạp? Không vấn đề.
- Contextual Compression: Nén thông tin truy xuất được để chỉ giữ lại phần liên quan nhất, giảm thiểu nhiễu trong context window.
- Hybrid Search: Kết hợp tìm kiếm theo từ khóa (BM25) với tìm kiếm ngữ nghĩa (vector similarity) -- kiểu "best of both worlds".
Triển Khai RAG 2.0 với LangChain
Dưới đây là ví dụ triển khai pipeline RAG 2.0 hoàn chỉnh với semantic chunking và multi-hop retrieval:
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from langchain.prompts import ChatPromptTemplate
from typing import List, Dict, Any
import numpy as np
class SemanticChunker:
"""Semantic chunking dựa trên độ tương đồng giữa các câu."""
def __init__(self, embeddings, similarity_threshold: float = 0.75):
self.embeddings = embeddings
self.similarity_threshold = similarity_threshold
def chunk_document(self, text: str) -> List[str]:
"""Chia văn bản thành các chunk dựa trên ngữ nghĩa."""
sentences = [s.strip() for s in text.split('.') if s.strip()]
if len(sentences) <= 1:
return [text]
sentence_embeddings = self.embeddings.embed_documents(sentences)
chunks = []
current_chunk = [sentences[0]]
for i in range(1, len(sentences)):
sim = self._cosine_similarity(
sentence_embeddings[i],
sentence_embeddings[i - 1]
)
if sim >= self.similarity_threshold:
current_chunk.append(sentences[i])
else:
chunks.append('. '.join(current_chunk) + '.')
current_chunk = [sentences[i]]
if current_chunk:
chunks.append('. '.join(current_chunk) + '.')
return chunks
@staticmethod
def _cosine_similarity(vec_a, vec_b) -> float:
a = np.array(vec_a)
b = np.array(vec_b)
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
class RAGPipeline:
"""Pipeline RAG 2.0 với semantic chunking và multi-hop retrieval."""
def __init__(
self,
collection_name: str = "production_docs",
model_name: str = "gpt-4o",
embedding_model: str = "text-embedding-3-large"
):
self.embeddings = OpenAIEmbeddings(model=embedding_model)
self.llm = ChatOpenAI(model=model_name, temperature=0)
self.chunker = SemanticChunker(self.embeddings)
self.vectorstore = Chroma(
collection_name=collection_name,
embedding_function=self.embeddings,
persist_directory="./chroma_db"
)
def ingest_documents(self, documents: List[Dict[str, str]]):
"""Nạp tài liệu với semantic chunking."""
all_docs = []
for doc in documents:
chunks = self.chunker.chunk_document(doc["content"])
for i, chunk in enumerate(chunks):
all_docs.append(Document(
page_content=chunk,
metadata={
"source": doc.get("source", "unknown"),
"title": doc.get("title", ""),
"chunk_index": i,
"total_chunks": len(chunks),
}
))
self.vectorstore.add_documents(all_docs)
def multi_hop_retrieve(
self, query: str, hops: int = 2, k: int = 4
) -> List[Document]:
"""Multi-hop retrieval: truy xuất thông tin qua nhiều bước."""
all_retrieved = []
current_query = query
for hop in range(hops):
docs = self.vectorstore.similarity_search(current_query, k=k)
all_retrieved.extend(docs)
if hop < hops - 1:
context = "\n".join([d.page_content for d in docs])
refinement_prompt = ChatPromptTemplate.from_template(
"Dựa trên câu hỏi gốc: {query}\n\n"
"Và thông tin đã tìm được:\n{context}\n\n"
"Hãy tạo một câu hỏi bổ sung để tìm thêm "
"thông tin cần thiết."
)
refined = self.llm.invoke(
refinement_prompt.format(query=query, context=context)
)
current_query = refined.content
# Loại bỏ trùng lặp
seen = set()
unique_docs = []
for doc in all_retrieved:
if doc.page_content not in seen:
seen.add(doc.page_content)
unique_docs.append(doc)
return unique_docs
def query(self, question: str, use_multi_hop: bool = True) -> str:
"""Truy vấn RAG pipeline."""
if use_multi_hop:
docs = self.multi_hop_retrieve(question)
else:
docs = self.vectorstore.similarity_search(question, k=4)
prompt = ChatPromptTemplate.from_template(
"Dựa trên ngữ cảnh sau đây, hãy trả lời câu hỏi.\n\n"
"Ngữ cảnh:\n{context}\n\n"
"Câu hỏi: {question}\n\n"
"Trả lời chi tiết và chính xác."
)
context = "\n---\n".join([doc.page_content for doc in docs])
response = self.llm.invoke(
prompt.format(context=context, question=question)
)
return response.content
# Ví dụ sử dụng
rag = RAGPipeline()
documents = [
{
"title": "Chính sách bảo hành",
"content": "Sản phẩm được bảo hành 24 tháng kể từ ngày mua. "
"Khách hàng cần giữ hóa đơn để được bảo hành. "
"Bảo hành không áp dụng cho hư hỏng do người dùng.",
"source": "policy_v3.pdf"
}
]
rag.ingest_documents(documents)
answer = rag.query("Thời gian bảo hành sản phẩm là bao lâu?")
print(answer)Pipeline RAG 2.0 ở trên minh họa hai điểm chính: semantic chunking đảm bảo các đoạn văn bản có tính mạch lạc về ngữ nghĩa, và multi-hop retrieval cho phép hệ thống kết nối thông tin từ nhiều nguồn khác nhau. Kết hợp cả hai, bạn sẽ thấy chất lượng câu trả lời cải thiện rõ rệt.
Quản Lý Bộ Nhớ (Memory) cho AI Agent
Memory là cái giúp AI nhớ được context của cuộc nói chuyện và học hỏi từ các tương tác trước. Quản lý memory tốt hay dở chính là cái phân biệt giữa một chatbot bình thường với một AI assistant thực sự thông minh.
Phân Loại Bộ Nhớ
Hệ thống bộ nhớ cho AI agent chia thành hai loại chính:
- Bộ nhớ ngắn hạn (Short-term Memory): Lưu ngữ cảnh của cuộc hội thoại hiện tại. Đây là conversation buffer chứa các tin nhắn gần đây, thường bị giới hạn bởi kích thước context window.
- Bộ nhớ dài hạn (Long-term Memory): Lưu trữ thông tin persist qua nhiều phiên -- sở thích người dùng, lịch sử tương tác, kiến thức đã học. Thường được lưu trong vector database hoặc cơ sở dữ liệu quan hệ.
Kỹ Thuật Nén Bộ Nhớ (Memory Compression)
Vấn đề là khi chat dài, context window sẽ đầy. Lúc đó có 2 cách xử lý chính:
- Message Trimming: Cắt mấy tin nhắn cũ đi, chỉ giữ lại N tin gần nhất. Đơn giản nhưng hơi "nguy" vì có thể mất thông tin quan trọng từ đầu cuộc hội thoại (trade-off mà).
- Summarization (Tóm tắt): Dùng LLM tóm tắt phần hội thoại cũ thành một đoạn ngắn, rồi thay thế các tin nhắn chi tiết bằng bản tóm tắt. Hiệu quả hơn nhưng tốn thêm chi phí API call.
Dưới đây là triển khai hệ thống quản lý bộ nhớ hoàn chỉnh:
from openai import OpenAI
from typing import List, Dict, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime
import json
import tiktoken
@dataclass
class Message:
role: str
content: str
timestamp: datetime = field(default_factory=datetime.now)
token_count: int = 0
class MemoryManager:
"""Quản lý bộ nhớ cho AI Agent với nén và tóm tắt."""
def __init__(
self,
model: str = "gpt-4o",
max_context_tokens: int = 8000,
summary_threshold: int = 6000
):
self.client = OpenAI()
self.model = model
self.max_context_tokens = max_context_tokens
self.summary_threshold = summary_threshold
self.encoding = tiktoken.encoding_for_model(model)
self.conversation_buffer: List[Message] = []
self.conversation_summary: Optional[str] = None
self.user_preferences: Dict[str, Any] = {}
self.important_facts: List[str] = []
def _count_tokens(self, text: str) -> int:
return len(self.encoding.encode(text))
def _get_total_tokens(self) -> int:
total = 0
if self.conversation_summary:
total += self._count_tokens(self.conversation_summary)
for msg in self.conversation_buffer:
total += msg.token_count
return total
def add_message(self, role: str, content: str):
"""Thêm tin nhắn mới vào bộ nhớ."""
token_count = self._count_tokens(content)
message = Message(
role=role, content=content, token_count=token_count
)
self.conversation_buffer.append(message)
if self._get_total_tokens() > self.summary_threshold:
self._compress_memory()
def _compress_memory(self):
"""Nén bộ nhớ bằng kỹ thuật tóm tắt."""
if len(self.conversation_buffer) < 4:
return
split_point = len(self.conversation_buffer) // 2
messages_to_summarize = self.conversation_buffer[:split_point]
conversation_text = "\n".join(
f"{m.role}: {m.content}"
for m in messages_to_summarize
)
summary_prompt = (
"Hãy tóm tắt cuộc hội thoại sau đây, giữ lại tất cả "
"thông tin quan trọng và ngữ cảnh cần thiết.\n\n"
f"Tóm tắt trước đó: {self.conversation_summary or 'Không có'}"
f"\n\nHội thoại:\n{conversation_text}"
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "Bạn là chuyên gia tóm tắt hội thoại."},
{"role": "user", "content": summary_prompt}
],
max_tokens=500,
temperature=0
)
self.conversation_summary = response.choices[0].message.content
self.conversation_buffer = self.conversation_buffer[split_point:]
def get_context_messages(self) -> List[Dict[str, str]]:
"""Lấy danh sách tin nhắn cho API call."""
messages = []
if self.conversation_summary:
messages.append({
"role": "system",
"content": f"Tóm tắt hội thoại trước: {self.conversation_summary}"
})
if self.important_facts:
facts = "; ".join(self.important_facts[-5:])
messages.append({
"role": "system",
"content": f"Thông tin quan trọng: {facts}"
})
for msg in self.conversation_buffer:
messages.append({"role": msg.role, "content": msg.content})
return messages
# Ví dụ sử dụng
memory = MemoryManager(max_context_tokens=8000)
memory.add_message("user", "Xin chào, tôi cần tư vấn mua laptop.")
memory.add_message("assistant", "Chào bạn! Bạn dùng laptop cho mục đích gì?")
memory.add_message("user", "Tôi thích lập trình Python và machine learning.")
context = memory.get_context_messages()
print(f"Số tin nhắn trong context: {len(context)}")Chiến Lược Bộ Nhớ Theo Tình Huống
Không phải ứng dụng nào cũng cần cùng một chiến lược bộ nhớ. Dưới đây là vài gợi ý dựa trên kinh nghiệm thực tế:
- Chatbot hỗ trợ khách hàng: Ưu tiên message trimming (giữ 10-20 tin gần nhất) kết hợp bộ nhớ dài hạn để nhớ lịch sử khiếu nại.
- AI coding assistant: Ưu tiên summarization vì ngữ cảnh mã nguồn rất quan trọng nhưng thường dài, cần nén thông minh.
- AI agent tự động: Kết hợp cả hai, với bộ nhớ episodic (nhớ các tác vụ đã hoàn thành) và semantic (nhớ kiến thức đã học).
Quản Lý Tool và Context Control
Khi hệ thống AI phức tạp lên, LLM cần biết xài nhiều tool. Vấn đề là khi có quá nhiều tool, nhét tất cả description vào context thì vừa tốn tiền vừa khiến model bị "lú".
Vấn Đề Tool Confusion
Tool confusion xảy ra khi LLM nhầm lẫn giữa các công cụ có chức năng tương tự, hoặc không chọn đúng công cụ phù hợp nhất. Đây là vấn đề khá nghiêm trọng trong production -- một lệnh gọi sai tool có thể dẫn đến mất dữ liệu hoặc kết quả sai hoàn toàn.
Áp Dụng RAG cho Tool Selection
Có một kỹ thuật khá hay trong năm 2026 là dùng RAG để chọn tool -- nghe lạ nhưng hiệu quả thật sự. Thay vì liệt kê tất cả công cụ trong prompt, hệ thống dùng semantic search để chỉ chọn những tool liên quan nhất. Theo research gần đây, cách này tăng độ chính xác lên 3 lần so với cách list hết tool ra.
from openai import OpenAI
from typing import List, Dict, Any
import numpy as np
class ToolRegistry:
"""Registry quản lý công cụ với RAG-based tool selection."""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
self.tools: Dict[str, Dict[str, Any]] = {}
self.tool_embeddings: Dict[str, List[float]] = {}
def register_tool(
self,
name: str,
description: str,
parameters: Dict[str, Any],
examples: List[str] = None,
category: str = "general"
):
"""Đăng ký công cụ mới với mô tả chi tiết."""
extended_desc = (
f"Công cụ: {name}\n"
f"Mô tả: {description}\n"
f"Danh mục: {category}"
)
if examples:
extended_desc += f"\nVí dụ: {'; '.join(examples)}"
response = self.client.embeddings.create(
model="text-embedding-3-large",
input=extended_desc
)
embedding = response.data[0].embedding
self.tools[name] = {
"name": name,
"description": description,
"parameters": parameters,
"category": category,
}
self.tool_embeddings[name] = embedding
def select_tools(
self, query: str, max_tools: int = 5
) -> List[Dict[str, Any]]:
"""Chọn công cụ phù hợp nhất dựa trên query."""
response = self.client.embeddings.create(
model="text-embedding-3-large",
input=query
)
query_embedding = response.data[0].embedding
similarities = {}
for name, tool_emb in self.tool_embeddings.items():
a, b = np.array(query_embedding), np.array(tool_emb)
sim = float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
similarities[name] = sim
sorted_tools = sorted(
similarities.items(), key=lambda x: x[1], reverse=True
)[:max_tools]
selected = []
for name, score in sorted_tools:
if score > 0.3:
tool = self.tools[name]
selected.append({
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["parameters"],
}
})
return selected
# Ví dụ sử dụng
registry = ToolRegistry()
registry.register_tool(
name="search_products",
description="Tìm kiếm sản phẩm trong catalog theo tên hoặc danh mục",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Từ khóa tìm kiếm"},
"category": {"type": "string", "description": "Danh mục"}
},
"required": ["query"]
},
examples=["Tìm laptop dưới 20 triệu", "Sản phẩm gaming nào đang giảm giá?"],
category="product"
)
registry.register_tool(
name="check_order_status",
description="Kiểm tra trạng thái đơn hàng theo mã đơn",
parameters={
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "Mã đơn hàng"}
},
"required": ["order_id"]
},
examples=["Đơn hàng #12345 đang ở đâu?", "Kiểm tra trạng thái giao hàng"],
category="order"
)
# Dynamic tool selection
tools = registry.select_tools("Tôi muốn tìm laptop gaming")
print(f"Đã chọn {len(tools)} công cụ phù hợp")Model Context Protocol (MCP)
Một xu hướng đáng chú ý năm 2026 là Model Context Protocol (MCP), hiện đã được đặt dưới sự quản lý của Linux Foundation. MCP chuẩn hóa cách các ứng dụng AI kết nối và tương tác với nguồn dữ liệu, công cụ, và dịch vụ bên ngoài. Cụ thể, với MCP bạn có thể:
- Định nghĩa tool interface theo chuẩn chung, tương thích với mọi LLM provider
- Chia sẻ và tái sử dụng tool definitions giữa các dự án
- Tự động khám phá và kết nối công cụ mới mà không cần sửa code ứng dụng
- Quản lý quyền truy cập và bảo mật cho từng công cụ
Nói đơn giản, MCP đóng vai trò như HTTP cho thế giới AI -- một giao thức chuẩn cho phép các thành phần khác nhau giao tiếp một cách thống nhất và đáng tin cậy.
Giám Sát và Tối Ưu Context trong Production
Lên production rồi thì bạn phải monitor liên tục -- không có chuyện "deploy xong thì nghỉ" đâu nhé. Chi phí API, độ trễ, và chất lượng phản hồi đều phụ thuộc vào cách bạn quản lý context.
Các Chỉ Số Giám Sát Quan Trọng
- Context Utilization Rate: Tỷ lệ sử dụng context window. Quá cao (>90%) có nguy cơ mất thông tin; quá thấp (<30%) lãng phí cơ hội cung cấp ngữ cảnh.
- Token Cost per Request: Chi phí token trung bình cho mỗi yêu cầu, cần theo dõi theo từng loại tác vụ.
- RAG Retrieval Precision: Tỷ lệ tài liệu truy xuất thực sự liên quan đến câu hỏi.
- Tool Selection Accuracy: Tỷ lệ chọn đúng công cụ ngay lần đầu.
- Memory Compression Ratio: Tỷ lệ nén bộ nhớ -- quá cao thì mất thông tin, quá thấp thì không tiết kiệm đủ token.
Tối Ưu Chi Phí với Plan-and-Execute Pattern
Một trong những chiến lược tối ưu chi phí hiệu quả nhất là Plan-and-Execute pattern. Thay vì gửi toàn bộ context phức tạp mỗi lần gọi LLM, pattern này chia tác vụ thành hai giai đoạn:
- Plan (Lập kế hoạch): Dùng model lớn (ví dụ GPT-4o) một lần duy nhất để phân tích yêu cầu và lập kế hoạch chi tiết.
- Execute (Thực thi): Dùng model nhỏ hơn, rẻ hơn (ví dụ GPT-4o-mini) để thực hiện từng bước trong kế hoạch.
Pattern này có thể giảm tới 90% chi phí mà vẫn ra kết quả tốt -- tiết kiệm được khá nhiều, nhất là khi hệ thống xử lý hàng nghìn request mỗi ngày.
Triển Khai Hệ Thống Giám Sát Context
from dataclasses import dataclass
from typing import Dict, List, Any
from datetime import datetime
import json
@dataclass
class ContextMetrics:
"""Đo lường hiệu suất context cho mỗi request."""
request_id: str
timestamp: datetime
input_tokens: int = 0
output_tokens: int = 0
context_window_size: int = 128000
system_prompt_tokens: int = 0
rag_context_tokens: int = 0
memory_tokens: int = 0
tool_description_tokens: int = 0
rag_docs_retrieved: int = 0
rag_docs_relevant: int = 0
tool_calls_made: int = 0
tool_calls_successful: int = 0
latency_ms: float = 0
estimated_cost_usd: float = 0
@property
def context_utilization(self) -> float:
return self.input_tokens / self.context_window_size
@property
def rag_precision(self) -> float:
if self.rag_docs_retrieved == 0:
return 0.0
return self.rag_docs_relevant / self.rag_docs_retrieved
class ContextMonitor:
"""Giám sát và tối ưu context trong production."""
def __init__(
self,
cost_per_input_token: float = 0.0000025,
cost_per_output_token: float = 0.00001,
alert_thresholds: Dict[str, float] = None
):
self.cost_per_input = cost_per_input_token
self.cost_per_output = cost_per_output_token
self.metrics_history: List[ContextMetrics] = []
self.alerts: List[Dict[str, Any]] = []
self.thresholds = alert_thresholds or {
"context_utilization_high": 0.85,
"rag_precision_low": 0.5,
"cost_per_request_high": 0.05,
"latency_high_ms": 5000,
}
def record_request(self, metrics: ContextMetrics):
"""Ghi nhận metrics cho một request."""
metrics.estimated_cost_usd = (
metrics.input_tokens * self.cost_per_input
+ metrics.output_tokens * self.cost_per_output
)
self.metrics_history.append(metrics)
self._check_alerts(metrics)
def _check_alerts(self, metrics: ContextMetrics):
"""Kiểm tra và tạo cảnh báo nếu vượt ngưỡng."""
if metrics.context_utilization > self.thresholds["context_utilization_high"]:
self.alerts.append({
"type": "HIGH_CONTEXT_UTILIZATION",
"message": f"Utilization {metrics.context_utilization:.1%}",
"request_id": metrics.request_id,
})
if (metrics.rag_precision < self.thresholds["rag_precision_low"]
and metrics.rag_docs_retrieved > 0):
self.alerts.append({
"type": "LOW_RAG_PRECISION",
"message": f"RAG precision {metrics.rag_precision:.1%}",
"request_id": metrics.request_id,
})
def get_daily_report(self) -> Dict[str, Any]:
"""Tạo báo cáo tổng hợp hàng ngày."""
if not self.metrics_history:
return {"error": "Không có dữ liệu"}
total_cost = sum(m.estimated_cost_usd for m in self.metrics_history)
avg_latency = (
sum(m.latency_ms for m in self.metrics_history)
/ len(self.metrics_history)
)
return {
"total_requests": len(self.metrics_history),
"total_cost_usd": round(total_cost, 4),
"avg_latency_ms": round(avg_latency, 1),
"alerts_count": len(self.alerts),
}
# Ví dụ sử dụng
monitor = ContextMonitor()
metrics = ContextMetrics(
request_id="req_001",
timestamp=datetime.now(),
input_tokens=4500,
output_tokens=800,
system_prompt_tokens=500,
rag_context_tokens=2000,
memory_tokens=1000,
tool_description_tokens=500,
rag_docs_retrieved=5,
rag_docs_relevant=3,
tool_calls_made=2,
tool_calls_successful=2,
latency_ms=1200,
)
monitor.record_request(metrics)
report = monitor.get_daily_report()
print(json.dumps(report, indent=2, ensure_ascii=False))Chiến Lược Tối Ưu Token
Ngoài Plan-and-Execute, còn có vài chiến lược tối ưu token khác khá hay:
- Dynamic System Prompt: Chỉ bao gồm phần system prompt liên quan đến tác vụ hiện tại, thay vì một prompt khổng lồ bao gồm mọi tình huống.
- Lazy Tool Loading: Chỉ nạp mô tả công cụ khi có dấu hiệu người dùng cần sử dụng công cụ đó -- đừng load trước hết.
- Tiered Caching: Cache kết quả RAG cho các câu hỏi phổ biến, giảm chi phí embedding và vector search.
- Prompt Compression: Sử dụng các kỹ thuật nén prompt (ví dụ LLMLingua) để giảm số token mà không mất thông tin quan trọng.
Mẫu Kiến Trúc Production Hoàn Chỉnh
Okay, giờ mình gom tất cả những gì đã nói ở trên vào một kiến trúc production hoàn chỉnh. Đây là mẫu khá ổn mà bạn có thể tham khảo ngay.
Tổng Quan Kiến Trúc
Kiến trúc được chia thành 4 layer chính:
- Input Layer: Tiếp nhận và tiền xử lý yêu cầu từ người dùng
- Context Assembly Layer: Lắp ráp ngữ cảnh từ nhiều nguồn (system prompt, RAG, memory, tools)
- LLM Execution Layer: Gọi LLM với context đã được tối ưu
- Output & Monitoring Layer: Xử lý kết quả, cập nhật bộ nhớ, ghi nhận metrics
from openai import OpenAI
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
import tiktoken
class TaskComplexity(Enum):
SIMPLE = "simple"
MODERATE = "moderate"
COMPLEX = "complex"
@dataclass
class ContextBudget:
"""Ngân sách token cho mỗi thành phần context."""
total_budget: int = 16000
system_prompt_ratio: float = 0.10
rag_context_ratio: float = 0.35
memory_ratio: float = 0.20
tool_descriptions_ratio: float = 0.15
user_message_ratio: float = 0.15
output_reserve_ratio: float = 0.05
def get_budget(self, component: str) -> int:
ratio = getattr(self, f"{component}_ratio", 0)
return int(self.total_budget * ratio)
class ProductionContextEngine:
"""Context Engine cho ứng dụng AI production."""
def __init__(
self,
primary_model: str = "gpt-4o",
secondary_model: str = "gpt-4o-mini",
budget: Optional[ContextBudget] = None,
):
self.client = OpenAI()
self.primary_model = primary_model
self.secondary_model = secondary_model
self.budget = budget or ContextBudget()
self.encoding = tiktoken.encoding_for_model(primary_model)
def _classify_complexity(self, query: str) -> TaskComplexity:
"""Phân loại độ phức tạp của tác vụ."""
complex_keywords = [
"so sánh", "phân tích", "giải thích chi tiết",
"làm thế nào", "quy trình"
]
query_lower = query.lower()
matches = sum(1 for kw in complex_keywords if kw in query_lower)
if matches >= 2:
return TaskComplexity.COMPLEX
elif matches >= 1:
return TaskComplexity.MODERATE
return TaskComplexity.SIMPLE
def _select_model(self, complexity: TaskComplexity) -> str:
"""Plan-and-Execute: Chọn model dựa trên độ phức tạp."""
if complexity == TaskComplexity.COMPLEX:
return self.primary_model
return self.secondary_model
def _truncate_to_budget(self, text: str, component: str) -> str:
"""Cắt văn bản để vừa với ngân sách token."""
budget = self.budget.get_budget(component)
tokens = self.encoding.encode(text)
if len(tokens) <= budget:
return text
return self.encoding.decode(tokens[:budget])
def assemble_context(
self,
user_message: str,
system_prompt: str = "",
rag_context: str = "",
memory_messages: List[Dict[str, str]] = None,
tools: List[Dict] = None
) -> Dict[str, Any]:
"""Lắp ráp ngữ cảnh hoàn chỉnh từ tất cả thành phần."""
messages = []
token_usage = {}
# 1. System Prompt
if system_prompt:
sp = self._truncate_to_budget(system_prompt, "system_prompt")
messages.append({"role": "system", "content": sp})
token_usage["system_prompt"] = len(self.encoding.encode(sp))
# 2. RAG Context
if rag_context:
rc = self._truncate_to_budget(rag_context, "rag_context")
messages.append({
"role": "system",
"content": f"Thông tin tham khảo:\n{rc}"
})
token_usage["rag"] = len(self.encoding.encode(rc))
# 3. Memory
if memory_messages:
messages.extend(memory_messages[-10:])
# 4. User message
messages.append({"role": "user", "content": user_message})
return {
"messages": messages,
"tools": tools or [],
"token_usage": token_usage
}
def execute(
self,
user_message: str,
system_prompt: str = "Bạn là trợ lý AI thông minh.",
rag_context: str = "",
memory_messages: List[Dict[str, str]] = None,
tools: List[Dict] = None
) -> Dict[str, Any]:
"""Thực thi hoàn chỉnh với context tối ưu."""
start = datetime.now()
complexity = self._classify_complexity(user_message)
model = self._select_model(complexity)
context = self.assemble_context(
user_message, system_prompt, rag_context,
memory_messages, tools
)
api_params = {
"model": model,
"messages": context["messages"],
"temperature": 0 if complexity == TaskComplexity.SIMPLE else 0.3,
}
if context["tools"]:
api_params["tools"] = context["tools"]
response = self.client.chat.completions.create(**api_params)
elapsed = (datetime.now() - start).total_seconds()
return {
"content": response.choices[0].message.content,
"model_used": model,
"complexity": complexity.value,
"usage": {
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens,
},
"latency_seconds": elapsed,
}
# Ví dụ sử dụng
engine = ProductionContextEngine(
budget=ContextBudget(total_budget=16000)
)
result = engine.execute(
user_message="Giải thích chi tiết cách tối ưu context window cho AI agent",
system_prompt="Bạn là chuyên gia AI với kinh nghiệm production.",
memory_messages=[
{"role": "user", "content": "Tôi đang xây dựng AI chatbot"},
{"role": "assistant", "content": "Tuyệt vời! Bạn cần hỗ trợ gì?"}
]
)
print(f"Model: {result['model_used']}")
print(f"Complexity: {result['complexity']}")
print(f"Response: {result['content'][:200]}...")Giải Thích Kiến Trúc
Giải thích thêm vài điểm quan trọng trong kiến trúc này:
- Context Budget Management: Mỗi thành phần được phân bổ một tỷ lệ token cố định, đảm bảo không có thành phần nào "nuốt" hết context window.
- Plan-and-Execute: Hệ thống tự động phân loại độ phức tạp và chọn model phù hợp. Câu hỏi đơn giản dùng model nhỏ rẻ hơn -- tiết kiệm đáng kể.
- Dynamic Tool Selection: Thay vì nạp tất cả tool definitions, hệ thống chỉ chọn tools liên quan thông qua semantic search.
- Observability: Mọi request được ghi nhận chi tiết về token usage, latency, và budget utilization.
Các Mẫu Thiết Kế Bổ Sung
Ngoài kiến trúc chính, có một số pattern bổ sung đáng lưu ý:
- Circuit Breaker cho Context: Khi context assembly gặp lỗi (ví dụ RAG service không khả dụng), hệ thống tiếp tục hoạt động với context không đầy đủ thay vì sập hoàn toàn. Graceful degradation là chìa khóa.
- A/B Testing cho Context Strategies: So sánh hiệu quả của các chiến lược khác nhau -- ví dụ semantic chunking vs fixed-size chunking -- để biết cái nào thực sự tốt hơn cho use case của bạn.
- Context Caching Layer: Cache toàn bộ assembled context cho các câu hỏi lặp lại, giảm thời gian xử lý và chi phí.
Tổng Hợp Các Nguyên Tắc Context Engineering
Tóm lại, sau khi "dạo" qua toàn bộ bài viết, mình rút ra được mấy nguyên tắc then chốt này:
Nguyên Tắc 1: Context Là Tài Nguyên Có Hạn
Luôn nhớ: context window = RAM của LLM. Giống như quản lý bộ nhớ trong lập trình hệ thống, bạn cần phân bổ, giám sát, và tối ưu cách sử dụng context. Đừng lãng phí token cho thông tin không liên quan.
Nguyên Tắc 2: Động Hóa Mọi Thứ
Context tĩnh = context chết. Nghe hơi khắc nghiệt nhưng đúng vậy. Mọi thành phần -- từ system prompt đến tool descriptions -- nên được lắp ráp động dựa trên ngữ cảnh cụ thể của từng request.
Nguyên Tắc 3: Đo Lường Trước Khi Tối Ưu
Không thể tối ưu những gì không đo lường được. Thiết lập hệ thống monitoring từ ngày đầu tiên, theo dõi token usage, chi phí, và chất lượng phản hồi.
Nguyên Tắc 4: Thiết Kế Cho Failure
Trong production, mọi thứ đều có thể fail. Context Engine cần có khả năng graceful degradation -- tiếp tục hoạt động ở mức chấp nhận được ngay cả khi một số thành phần gặp sự cố.
Nguyên Tắc 5: Lấy Chi Phí Làm Ràng Buộc Thiết Kế
Chi phí API là thực tế không thể tránh khỏi. Thay vì coi đây là vấn đề cần giải quyết sau, hãy đặt nó làm ràng buộc thiết kế ngay từ đầu. Bạn sẽ cảm ơn chính mình sau này.
Kết Luận
Context Engineering không phải là trend nhất thời -- nó là một sự chuyển đổi tư duy hoàn toàn trong cách xây dựng ứng dụng AI. Năm 2026, khi 40% ứng dụng doanh nghiệp tích hợp AI agent (theo Gartner), khả năng thiết kế và vận hành hệ thống context hiệu quả sẽ là yếu tố phân biệt giữa thành công và thất bại.
Hãy nhớ rằng:
- LLM là CPU, context window là RAM -- quản lý tốt context chính là quản lý tốt tài nguyên quý giá nhất của hệ thống AI.
- Context Engineering có 5 thành phần cốt lõi: System Prompts, RAG, Memory, Tools, và Structured Outputs -- tất cả phải hoạt động hài hòa.
- RAG 2.0 với semantic chunking và multi-hop retrieval là tiêu chuẩn mới.
- RAG for tool selection có thể cải thiện độ chính xác lên 3 lần.
- Plan-and-Execute pattern có thể giảm chi phí lên đến 90% mà không hy sinh chất lượng.
- MCP (Model Context Protocol) dưới sự quản lý của Linux Foundation đang chuẩn hóa cách AI tương tác với thế giới bên ngoài.
- Giám sát liên tục với các metrics rõ ràng là nền tảng cho mọi thứ.
Bước Tiếp Theo
Muốn bắt đầu thì làm theo roadmap này (từng bước một, đừng vội):
- Audit hệ thống hiện tại: Đo lường context utilization và chi phí token của ứng dụng AI đang vận hành.
- Triển khai monitoring: Thiết lập hệ thống giám sát context metrics trước khi tối ưu bất kỳ thành phần nào.
- Áp dụng RAG 2.0: Nâng cấp pipeline RAG với semantic chunking và multi-hop retrieval.
- Triển khai dynamic tool selection: Áp dụng RAG cho việc chọn công cụ thay vì liệt kê tĩnh.
- Tối ưu chi phí: Triển khai Plan-and-Execute pattern, bắt đầu với phân loại tác vụ đơn giản.
- Khám phá MCP: Tìm hiểu và thử nghiệm Model Context Protocol cho việc chuẩn hóa tool integration.
Context Engineering đang phát triển rất nhanh -- những gì ở đây là nền tảng vững chắc, nhưng cứ vài tháng lại có technique mới xuất hiện. Điều quan trọng nhất? Bắt tay vào làm thôi -- áp dụng từng cái một, đo lường kết quả, rồi cải tiến dần.
Lời cuối: Trong kỷ nguyên AI, xây dựng ứng dụng AI production không chỉ là về mô hình mạnh mẽ nhất, mà là về ngữ cảnh thông minh nhất. Hãy đầu tư vào Context Engineering -- đó là khoản đầu tư có lợi suất cao nhất cho tương lai ứng dụng AI của bạn.