Câu Chuyện Thực Tế: Hệ Thống RAG Cho Thương Mại Điện Tử
Tôi đã triển khai một hệ thống RAG cho startup thương mại điện tử với 50,000 sản phẩm. Khách hàng cần hỏi về tồn kho, giá cả, chính sách đổi trả bằng tiếng Việt tự nhiên. Vấn đề? LangChain mặc định không hiểu database nội bộ của họ. Giải pháp: Custom Tools — và tôi sẽ chia sẻ toàn bộ quy trình.
Trong bài viết này, tôi sẽ hướng dẫn bạn xây dựng custom tools với LangChain, tích hợp API từ
HolySheep AI — nền tảng với độ trễ dưới 50ms và chi phí chỉ từ $0.42/MTok.
1. Tổng Quan Kiến Trúc Custom Tool Trong LangChain
Custom tool là các hàm Python mà LLM có thể gọi để thực hiện tác vụ cụ thể: truy vấn database, gọi API bên thứ ba, xử lý file, hoặc tính toán phức tạp.
┌──────────────┐ ┌─────────────────┐ ┌──────────────────┐
│ User Input │ ──► │ LangChain │ ──► │ Custom Tools │
│ (Tiếng Việt)│ │ Agent │ │ • Query DB │
└──────────────┘ └─────────────────┘ │ • Search API │
│ • Calculator │
└──────────────────┘
2. Cài Đặt Môi Trường
# Cài đặt các thư viện cần thiết
pip install langchain langchain-core langchain-community \
langchain-openai python-dotenv sqlalchemy pymysql
Kiểm tra phiên bản
python -c "import langchain; print(f'LangChain: {langchain.__version__}')"
3. Tạo Custom Tool Đầu Tiên Với HolySheep AI
Dưới đây là code hoàn chỉnh để tạo custom tool truy vấn sản phẩm:
import os
from dotenv import load_dotenv
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub
Load environment variables
load_dotenv()
Cấu hình HolySheep AI - KHÔNG dùng OpenAI/Anthropic endpoint
os.environ["OPENAI_API_KEY"] = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
Khởi tạo model với HolySheep AI
llm = ChatOpenAI(
model="gpt-4.1", # $8/MTok - tiết kiệm 85%+
temperature=0.7,
api_key=os.environ["OPENAI_API_KEY"],
base_url=os.environ["OPENAI_API_BASE"]
)
@tool
def query_product_stock(product_id: str) -> str:
"""
Truy vấn tồn kho sản phẩm theo mã sản phẩm.
Args:
product_id: Mã sản phẩm (ví dụ: "SKU-001")
Returns:
Thông tin tồn kho chi tiết
"""
# Demo database - thay bằng truy vấn thực tế
products = {
"SKU-001": {"name": "Áo thun nam", "stock": 150, "price": 299000},
"SKU-002": {"name": "Quần jeans nữ", "stock": 45, "price": 599000},
"SKU-003": {"name": "Giày thể thao", "stock": 0, "price": 1200000},
}
if product_id in products:
p = products[product_id]
status = "Còn hàng" if p["stock"] > 0 else "Hết hàng"
return f"Sản phẩm: {p['name']} | Giá: {p['price']:,}đ | Tồn kho: {p['stock']} | {status}"
return f"Không tìm thấy sản phẩm: {product_id}"
@tool
def calculate_discount(original_price: float, discount_percent: float) -> str:
"""
Tính giá sau khi giảm giá.
Args:
original_price: Giá gốc (VND)
discount_percent: Phần trăm giảm giá (0-100)
Returns:
Giá sau giảm và số tiền tiết kiệm
"""
if discount_percent < 0 or discount_percent > 100:
return "Lỗi: Phần trăm giảm giá phải từ 0 đến 100"
discounted_price = original_price * (1 - discount_percent / 100)
savings = original_price - discounted_price
return (f"Giá gốc: {original_price:,.0f}đ | "
f"Giảm: {discount_percent}% | "
f"Giá mới: {discounted_price:,.0f}đ | "
f"Tiết kiệm: {savings:,.0f}đ")
Đăng ký tools vào agent
tools = [query_product_stock, calculate_discount]
Pull prompt từ hub
prompt = hub.pull("hwchase17/react")
Tạo agent
agent = create_react_agent(llm, tools, prompt)
Tạo agent executor
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
Test với câu hỏi tiếng Việt
result = agent_executor.invoke({
"input": "Cho tôi biết sản phẩm SKU-001 còn hàng không, và nếu có thì giá sau khi giảm 20% là bao nhiêu?"
})
print(result["output"])
4. Custom Tool Với Database Thực Tế
Đây là ví dụ nâng cao kết nối với MySQL database sản phẩm:
import os
from typing import List, Optional
from pydantic import BaseModel, Field
from langchain.tools import tool
import sqlalchemy as sa
from sqlalchemy.orm import Session
Cấu hình kết nối database
DATABASE_URL = os.getenv(
"DATABASE_URL",
"mysql+pymysql://user:password@localhost:3306/ecommerce"
)
Schema cho dữ liệu trả về
class ProductInfo(BaseModel):
product_id: str
name: str
category: str
price: float
stock: int
rating: float = Field(ge=0, le=5)
def get_db_connection():
"""Tạo kết nối database mới"""
engine = sa.create_engine(DATABASE_URL)
return engine.connect()
@tool
def search_products(
keyword: str,
category: Optional[str] = None,
max_price: Optional[float] = None,
limit: int = 5
) -> str:
"""
Tìm kiếm sản phẩm trong database theo từ khóa.
Args:
keyword: Từ khóa tìm kiếm (tên sản phẩm)
category: Lọc theo danh mục (tùy chọn)
max_price: Giá tối đa (VND, tùy chọn)
limit: Số lượng kết quả tối đa (mặc định: 5)
Returns:
Danh sách sản phẩm phù hợp
"""
conn = get_db_connection()
query = """
SELECT product_id, name, category, price, stock, rating
FROM products
WHERE name LIKE :keyword
"""
params = {"keyword": f"%{keyword}%"}
if category:
query += " AND category = :category"
params["category"] = category
if max_price:
query += " AND price <= :max_price"
params["max_price"] = max_price
query += f" ORDER BY rating DESC LIMIT {limit}"
try:
result = conn.execute(sa.text(query), params)
rows = result.fetchall()
if not rows:
return f"Không tìm thấy sản phẩm nào cho từ khóa: '{keyword}'"
output = f"Tìm thấy {len(rows)} sản phẩm cho '{keyword}':\n\n"
for row in rows:
status = "✅ Còn hàng" if row[4] > 0 else "❌ Hết hàng"
output += f"• {row[1]} [{row[0]}]\n"
output += f" Danh mục: {row[2]} | Giá: {row[3]:,.0f}đ\n"
output += f" Đánh giá: ⭐{row[5]} | {status}\n\n"
return output
finally:
conn.close()
@tool
def check_order_status(order_id: str) -> str:
"""
Kiểm tra trạng thái đơn hàng.
Args:
order_id: Mã đơn hàng (ví dụ: "ORD-2024-001")
Returns:
Thông tin trạng thái đơn hàng
"""
conn = get_db_connection()
query = """
SELECT o.order_id, o.customer_name, o.total_amount,
o.status, o.created_at, p.name as payment_method
FROM orders o
LEFT JOIN payment_methods p ON o.payment_id = p.id
WHERE o.order_id = :order_id
"""
try:
result = conn.execute(sa.text(query), {"order_id": order_id})
row = result.fetchone()
if not row:
return f"Không tìm thấy đơn hàng: {order_id}"
status_icons = {
"pending": "⏳ Chờ xác nhận",
"confirmed": "✅ Đã xác nhận",
"shipping": "🚚 Đang giao",
"delivered": "📦 Đã giao",
"cancelled": "❌ Đã hủy"
}
status_display = status_icons.get(row[3], row[3])
return (f"Mã đơn: {row[0]}\n"
f"Khách hàng: {row[1]}\n"
f"Tổng tiền: {row[2]:,.0f}đ\n"
f"Thanh toán: {row[5]}\n"
f"Trạng thái: {status_display}\n"
f"Ngày đặt: {row[4].strftime('%d/%m/%Y %H:%M')}")
finally:
conn.close()
Sử dụng trong agent
tools_ecommerce = [search_products, check_order_status]
5. Tích Hợp RAG Với Custom Tool
Kết hợp vector search với custom tool để tạo hệ thống trả lời thông minh:
import os
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.tools import tool
Cấu hình HolySheep cho embeddings
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
embeddings = OpenAIEmbeddings(
model="text-embedding-3-small",
api_key=os.environ["OPENAI_API_KEY"],
base_url=os.environ["OPENAI_API_BASE"]
)
@tool
def rag_product_search(query: str, top_k: int = 3) -> str:
"""
Tìm kiếm thông tin sản phẩm sử dụng RAG (Retrieval-Augmented Generation).
Kết hợp vector search với LLM để trả lời câu hỏi phức tạp.
Args:
query: Câu hỏi về sản phẩm
top_k: Số lượng kết quả retrieval (mặc định: 3)
Returns:
Câu trả lời được tăng cường với ngữ cảnh từ database
"""
# Khởi tạo vector store (thay bằng FAISS hoặc Pinecone thực tế)
vectorstore = Chroma(
persist_directory="./product_vectors",
embedding_function=embeddings
)
# Retrieval: Tìm documents liên quan
retriever = vectorstore.as_retriever(search_kwargs={"k": top_k})
docs = retriever.get_relevant_documents(query)
if not docs:
return "Không tìm thấy thông tin liên quan trong cơ sở tri thức."
# Tạo context từ documents
context = "\n\n".join([doc.page_content for doc in docs])
# Sử dụng LLM để tạo câu trả lời với context
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4.1",
api_key=os.environ["OPENAI_API_KEY"],
base_url=os.environ["OPENAI_API_BASE"]
)
prompt = f"""Dựa trên thông tin sản phẩm sau, hãy trả lời câu hỏi của khách hàng bằng tiếng Việt tự nhiên.
Thông tin sản phẩm:
{context}
Câu hỏi: {query}
Nếu thông tin không đủ, hãy nói rõ và gợi ý khách hàng hỏi cụ thể hơn.
"""
response = llm.invoke(prompt)
return response.content
Tích hợp vào agent
tools_full = [query_product_stock, calculate_discount, search_products, rag_product_search]
6. Đăng Ký Tool Động Với Tool Registry
from typing import Dict, List, Type
from langchain.tools import BaseTool
from pydantic import BaseModel
class ToolRegistry:
"""Registry để quản lý và đăng ký tools động"""
def __init__(self):
self._tools: Dict[str, BaseTool] = {}
def register(self, name: str, tool: BaseTool) -> None:
"""Đăng ký một tool mới"""
self._tools[name] = tool
print(f"✅ Registered tool: {name}")
def unregister(self, name: str) -> None:
"""Hủy đăng ký tool"""
if name in self._tools:
del self._tools[name]
print(f"🗑️ Unregistered tool: {name}")
def get_tool(self, name: str) -> BaseTool:
"""Lấy tool theo tên"""
if name not in self._tools:
raise ValueError(f"Tool '{name}' not found in registry")
return self._tools[name]
def list_tools(self) -> List[str]:
"""Liệt kê tất cả tools đã đăng ký"""
return list(self._tools.keys())
def get_all_tools(self) -> List[BaseTool]:
"""Lấy tất cả tools dưới dạng list"""
return list(self._tools.values())
Sử dụng registry
registry = ToolRegistry()
Đăng ký tools
registry.register("query_stock", query_product_stock)
registry.register("calculate_price", calculate_discount)
registry.register("search", search_products)
registry.register("rag", rag_product_search)
print(f"Tổng số tools: {len(registry.list_tools())}")
print(f"Danh sách: {registry.list_tools()}")
7. Bảng Giá So Sánh Khi Sử Dụng Custom Tools
Khi triển khai custom tools với LangChain, việc chọn đúng provider ảnh hưởng lớn đến chi phí:
| Model | Provider | Giá/MTok | Độ trễ TB | Tiết kiệm |
|---|
| GPT-4.1 | HolySheep AI | $8.00 | <50ms | 85%+ |
| GPT-4.1 | OpenAI | $60.00 | ~200ms | — |
| Claude Sonnet 4.5 | HolySheep AI | $15.00 | <50ms | 80%+ |
| DeepSeek V3.2 | HolySheep AI | $0.42 | <50ms | 95%+ |
Với dự án của tôi (khoảng 1 triệu tokens/tháng), chuyển từ OpenAI sang HolySheep tiết kiệm **$52,000/năm**.
Lỗi Thường Gặp Và Cách Khắc Phục
Lỗi 1: "API Key Not Valid" Hoặc Authentication Error
**Nguyên nhân:** API key không đúng hoặc chưa được set đúng environment variable.
# ❌ SAI - Key nằm trong code
os.environ["OPENAI_API_KEY"] = "sk-abc123..."
✅ ĐÚNG - Load từ .env hoặc environment
from dotenv import load_dotenv
load_dotenv()
Hoặc export trực tiếp
export HOLYSHEEP_API_KEY="YOUR_ACTUAL_KEY"
Verify key trước khi sử dụng
import os
api_key = os.getenv("HOLYSHEEP_API_KEY")
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError("Vui lòng set HOLYSHEEP_API_KEY trong file .env")
llm = ChatOpenAI(
model="gpt-4.1",
api_key=api_key,
base_url="https://api.holysheep.ai/v1" # Phải chỉ định rõ
)
Lỗi 2: "Tool Input Parsing Error" - LLM Không Gọi Đúng Tool
**Nguyên nhân:** Docstring không đúng format hoặc thiếu type hints.
# ❌ SAI - Thiếu format chuẩn
@tool
def get_weather(city):
return f"Thời tiết ở {city}"
✅ ĐÚNG - Format đầy đủ với docstring chuẩn
@tool
def get_weather(city: str) -> str:
"""
Lấy thông tin thời tiết hiện tại của một thành phố.
Args:
city: Tên thành phố cần tra cứu (VD: "Hà Nội", "TP.HCM")
Returns:
Thông tin thời tiết bao gồm nhiệt độ, độ ẩm, tình trạng
"""
# Implement actual weather API call here
return f"Thời tiết {city}: 32°C, nắng"
Kiểm tra tool schema
print(get_weather.args_schema.schema())
Lỗi 3: "Context Length Exceeded" Khi Tool Trả Về Quá Nhiều Data
**Nguyên nhân:** Kết quả từ database quá lớn, vượt context window.
# ❌ SAI - Trả về toàn bộ data
@tool
def get_all_products() -> str:
result = db.query("SELECT * FROM products") # 50,000 rows!
return str(result) # Overflow!
✅ ĐÚNG - Giới hạn và phân trang
@tool
def get_all_products(page: int = 1, page_size: int = 10) -> str:
"""
Lấy danh sách sản phẩm với phân trang.
Args:
page: Số trang (bắt đầu từ 1)
page_size: Số sản phẩm mỗi trang (tối đa: 50)
"""
page_size = min(page_size, 50) # Hard limit
offset = (page - 1) * page_size
query = f"""
SELECT product_id, name, price, stock
FROM products
ORDER BY created_at DESC
LIMIT {page_size} OFFSET {offset}
"""
total = db.execute("SELECT COUNT(*) FROM products").fetchone()[0]
results = db.execute(query).fetchall()
output = f"Trang {page}/{-(-total // page_size)} (Tổng: {total} sản phẩm)\n\n"
for p in results:
output += f"• {p[1]} - {p[2]:,}đ\n"
return output
Sử dụng streaming cho response dài
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
callbacks=[StreamingStdOutCallbackHandler()] # Stream thay vì đợi full response
)
Lỗi 4: "Rate Limit Exceeded" Khi Gọi Tool Liên Tục
**Nguyên nhân:** Gọi API quá nhiều lần trong thời gian ngắn.
# ❌ SAI - Không có rate limiting
def process_batch(items):
results = []
for item in items: # 100 items
result = tool.invoke(item) # 100 API calls liên tục!
results.append(result)
return results
✅ ĐÚNG - Batch request và retry logic
import time
from functools import wraps
def rate_limit(max_calls=10, period=60):
"""Decorator giới hạn số lần gọi API"""
calls = []
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
now = time.time()
calls[:] = [t for t in calls if now - t < period]
if len(calls) >= max_calls:
sleep_time = period - (now - calls[0])
print(f"Rate limit reached. Sleeping {sleep_time:.1f}s...")
time.sleep(sleep_time)
calls.append(now)
return func(*args, **kwargs)
return wrapper
return decorator
@rate_limit(max_calls=50, period=60)
def call_with_limit(tool_name, *args, **kwargs):
"""Gọi tool với rate limiting"""
tool = registry.get_tool(tool_name)
return tool.invoke(*args, **kwargs)
Hoặc sử dụng caching để giảm số lần gọi
from functools import lru_cache
@lru_cache(maxsize=100)
def cached_product_query(product_id: str) -> str:
"""Cache kết quả truy vấn sản phẩm"""
return query_product_stock.invoke({"product_id": product_id})
Kết Luận
Custom tools trong LangChain là chìa khóa để xây dựng AI agents thông minh, có khả năng tương tác với thế giới thực. Qua bài viết này, bạn đã học được:
- Cách định nghĩa và đăng ký custom tools với decorators
- Kết nối với database MySQL/PostgreSQL
- Tích hợp RAG cho vector search
- Xây dựng Tool Registry để quản lý động
- Xử lý các lỗi thường gặp
Với HolySheep AI, bạn không chỉ tiết kiệm **85%+ chi phí** mà còn được hưởng độ trễ dưới **50ms** — lý tưởng cho ứng dụng production cần real-time response.
👉
Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký
Tài nguyên liên quan
Bài viết liên quan