Tôi đã triển khai LangChain callback cho hệ thống production của mình và phát hiện ra rằng việc giám sát chi phí và độ trễ của từng LLM call là điều tối quan trọng. Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống monitoring hoàn chỉnh với HolySheep AI — nền tảng API giá rẻ với độ trễ dưới 50ms.
Tại sao cần Callback trong LangChain?
Trong các ứng dụng RAG, agent orchestration, hay multi-step reasoning, việc gọi LLM nhiều lần là thường trực. Không có callback mechanism, bạn sẽ không biết:
- Tổng chi phí token của cả chain là bao nhiêu?
- Độ trễ tại mỗi bước xử lý là bao nhiêu mili-giây?
- Token usage chi tiết (prompt/completion) ra sao?
Bảng so sánh chi phí LLM 2026
| Model | Giá Output ($/MTok) | 10M token/tháng |
|---|---|---|
| GPT-4.1 | $8.00 | $80.00 |
| Claude Sonnet 4.5 | $15.00 | $150.00 |
| Gemini 2.5 Flash | $2.50 | $25.00 |
| DeepSeek V3.2 | $0.42 | $4.20 |
Với HolySheep AI, bạn được hưởng tỷ giá ¥1=$1 và thanh toán qua WeChat/Alipay — tiết kiệm đến 85% so với các provider khác khi sử dụng DeepSeek V3.2.
Triển khai Callback Handler tùy chỉnh
Tôi đã xây dựng một callback handler hoàn chỉnh để ghi log chi phí và độ trễ real-time:
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult
from datetime import datetime
from typing import Optional, Dict, Any
import time
class CostTrackingCallback(BaseCallbackHandler):
"""Callback handler để tracking chi phí và độ trễ"""
def __init__(self):
super().__init__()
self.call_history: list[Dict[str, Any]] = []
self.total_cost = 0.0
self.total_tokens = 0
self.total_latency_ms = 0.0
# Bảng giá HolySheep AI 2026 (output)
self.pricing = {
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4.5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
"deepseek-v3.2": 0.42, # $0.42/MTok
}
def on_llm_start(
self, serialized: Dict[str, Any], prompts: list[str], **kwargs
):
self._start_time = time.perf_counter()
self._current_prompts = prompts
def on_llm_end(
self, response: LLMResult, **kwargs
):
end_time = time.perf_counter()
latency_ms = (end_time - self._start_time) * 1000
for generation in response.generations:
for gen in generation:
model = response.llm_output.get("model_name", "unknown")
token_usage = response.llm_output.get("token_usage", {})
completion_tokens = token_usage.get("completion_tokens", 0)
prompt_tokens = token_usage.get("prompt_tokens", 0)
# Tính chi phí dựa trên completion tokens
price_per_mtok = self.pricing.get(model, 0)
cost = (completion_tokens / 1_000_000) * price_per_mtok
record = {
"timestamp": datetime.now().isoformat(),
"model": model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens,
"latency_ms": round(latency_ms, 2),
"cost_usd": round(cost, 6)
}
self.call_history.append(record)
self.total_cost += cost
self.total_tokens += completion_tokens
self.total_latency_ms += latency_ms
def get_summary(self) -> Dict[str, Any]:
return {
"total_calls": len(self.call_history),
"total_cost_usd": round(self.total_cost, 6),
"total_tokens": self.total_tokens,
"avg_latency_ms": round(
self.total_latency_ms / len(self.call_history), 2
) if self.call_history else 0,
"calls": self.call_history
}
Kết nối HolySheep AI với LangChain
Đây là phần quan trọng — tôi sử dụng HolySheep AI với endpoint chuẩn OpenAI-compatible:
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
from langchain.prompts import PromptTemplate, ChatPromptTemplate
from langchain.chains import LLMChain
Cấu hình HolySheep AI - KHÔNG dùng api.openai.com
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Khởi tạo LLM với callback handler
callback_handler = CostTrackingCallback()
DeepSeek V3.2 - Chi phí thấp nhất, chất lượng cao
llm_deepseek = ChatOpenAI(
model="deepseek-v3.2",
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL,
temperature=0.7,
callbacks=[callback_handler]
)
Claude Sonnet 4.5 - Cho reasoning phức tạp
llm_claude = ChatOpenAI(
model="claude-sonnet-4.5",
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL,
temperature=0.3,
callbacks=[callback_handler]
)
Tạo chain đơn giản
prompt = ChatPromptTemplate.from_template(
"Giải thích {topic} theo cách đơn giản cho học sinh 10 tuổi."
)
chain = LLMChain(llm=llm_deepseek, prompt=prompt)
Chạy và theo dõi chi phí
result = chain.run(topic="Quang hợp")
summary = callback_handler.get_summary()
print(f"Tổng chi phí: ${summary['total_cost_usd']}")
print(f"Độ trễ TB: {summary['avg_latency_ms']}ms")
print(f"Số lần gọi: {summary['total_calls']}")
Callback cho Multi-Agent System
Trong production, tôi thường dùng multiple callbacks để tracking từng agent riêng biệt:
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import Tool
from langchain.prompts import MessagesPlaceholder
class AgentCostTracker(BaseCallbackHandler):
"""Tracker riêng cho từng agent"""
def __init__(self, agent_name: str):
self.agent_name = agent_name
self.cost = 0.0
self.latency = 0.0
def on_llm_end(self, response: LLMResult, **kwargs):
# Logic tracking riêng cho agent này
print(f"[{self.agent_name}] Hoàn thành: "
f"{response.llm_output.get('token_usage', {})}")
Setup multi-agent
research_agent = create_openai_functions_agent(
llm=llm_deepseek,
tools=[search_tool, browse_tool],
prompt=research_prompt
)
writing_agent = create_openai_functions_agent(
llm=llm_claude,
tools=[],
prompt=writing_prompt
)
Executor với callback riêng
research_executor = AgentExecutor.from_agent_and_tools(
agent=research_agent,
tools=[search_tool, browse_tool],
callbacks=[AgentCostTracker("researcher")]
)
writing_executor = AgentExecutor.from_agent_and_tools(
agent=writing_agent,
tools=[],
callbacks=[AgentCostTracker("writer")]
)
Orchestration với callback tổng hợp
orchestrator_callback = CostTrackingCallback()
Gọi tuần tự và tracking tổng chi phí
research_result = research_executor.invoke(
{"query": "AI trends 2026"},
callbacks=[orchestrator_callback]
)
writing_result = writing_executor.invoke(
{"context": research_result["output"]},
callbacks=[orchestrator_callback]
)
Xuất báo cáo chi phí
print(orchestrator_callback.get_summary())
Callback cho Streaming Response
Streaming là cần thiết cho UX tốt, nhưng tracking chi phí vẫn phải chính xác:
class StreamingCostCallback(BaseCallbackHandler):
"""Callback hỗ trợ streaming với tracking token"""
def __init__(self):
super().__init__()
self.current_tokens = 0
self.streaming_latency = []
def on_llm_start(self, serialized, prompts, **kwargs):
self.start_time = time.perf_counter()
self.current_tokens = 0
self._token_buffer = ""
def on_llm_new_token(self, token: str, **kwargs):
"""Gọi mỗi khi có token mới được stream"""
self._token_buffer += token
self.current_tokens += 1
def on_llm_end(self, response: LLMResult, **kwargs):
latency = (time.perf_counter() - self.start_time) * 1000
token_usage = response.llm_output.get("token_usage", {})
completion_tokens = token_usage.get("completion_tokens", 0)
model = response.llm_output.get("model_name", "unknown")
price = {"deepseek-v3.2": 0.42, "claude-sonnet-4.5": 15.0}.get(model, 0)
print(f"Streaming completed:")
print(f" - Tokens: {completion_tokens}")
print(f" - Latency: {latency:.2f}ms")
print(f" - Cost: ${(completion_tokens/1_000_000) * price:.6f}")
print(f" - Throughput: {completion_tokens/(latency/1000):.1f} tok/s")
Sử dụng với streaming
streaming_callback = StreamingCostCallback()
llm_stream = ChatOpenAI(
model="deepseek-v3.2",
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL,
streaming=True,
callbacks=[streaming_callback]
)
Response streaming
for chunk in llm_stream.stream("Viết đoạn văn 500 từ về AI..."):
print(chunk.content, end="", flush=True)
Tích hợp với Prometheus/Grafana
Để monitor trên production, tôi export metrics sang Prometheus:
from prometheus_client import Counter, Histogram, Gauge
import logging
Prometheus metrics
llm_call_counter = Counter(
'llm_calls_total',
'Total LLM API calls',
['model', 'status']
)
llm_cost_gauge = Gauge(
'llm_total_cost_usd',
'Total cost in USD'
)
llm_latency_histogram = Histogram(
'llm_latency_seconds',
'LLM response latency',
['model']
)
class PrometheusCallback(BaseCallbackHandler):
"""Export metrics sang Prometheus"""
def on_llm_end(self, response: LLMResult, **kwargs):
token_usage = response.llm_output.get("token_usage", {})
model = response.llm_output.get("model_name", "unknown")
completion_tokens = token_usage.get("completion_tokens", 0)
llm_call_counter.labels(model=model, status="success").inc()
price = {"deepseek-v3.2": 0.42, "claude-sonnet-4.5": 15.0}.get(model, 0)
cost = (completion_tokens / 1_000_000) * price
llm_cost_gauge.inc(cost)
llm_latency_histogram.labels(model=model).observe(
self._get_latency()
)
def on_llm_error(self, error: Exception, **kwargs):
llm_call_counter.labels(model="unknown", status="error").inc()
logging.error(f"LLM Error: {error}")
Khởi chạy với monitoring
prometheus_cb = PrometheusCallback()
llm_monitored = ChatOpenAI(
model="deepseek-v3.2",
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL,
callbacks=[prometheus_cb]
)
Lỗi thường gặp và cách khắc phục
1. Lỗi "Authentication Error" khi dùng HolySheep
Nguyên nhân: API key không đúng hoặc chưa đăng ký tài khoản.
# Sai - dùng endpoint OpenAI
base_url="https://api.openai.com/v1" # ❌ SAI
Đúng - dùng HolySheep endpoint
base_url="https://api.holysheep.ai/v1" # ✅ ĐÚNG
Kiểm tra API key
import os
assert os.getenv("HOLYSHEEP_API_KEY"), "Cần set HOLYSHEEP_API_KEY"
Test kết nối
from openai import OpenAI
client = OpenAI(
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
models = client.models.list()
print("Kết nối thành công:", models.data[0].id)
2. Callback không được gọi khi dùng .invoke()
Nguyên nhân: LangChain có 2 cách gọi — callback chỉ hoạt động với một số method nhất định.
# Sai - .run() không trigger callback trong một số trường hợp
result = chain.run(input="test") # ❌ Có thể không work
Đúng - dùng .invoke() với config
from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb:
result = chain.invoke(
{"topic": "AI"},
config={"callbacks": [callback_handler]} # ✅ Explicit callback
)
print(f"Tổng token: {cb.total_tokens}")
print(f"Tổng chi phí: ${cb.total_cost}")
Hoặc dùng async với ainvoke
result = await chain.ainvoke(
{"topic": "AI"},
config={"callbacks": [callback_handler]}
)
3. Token usage không được trả về
Nguyên nhân: Model không hỗ trợ usage tracking hoặc response format không đúng.
# Kiểm tra response có llm_output không
response = llm.invoke([HumanMessage(content="Test")])
if hasattr(response, 'additional_kwargs'):
# OpenAI format
usage = response.additional_kwargs.get('usage', {})
elif hasattr(response, 'usage'):
# Anthropic format
usage = response.usage
else:
usage = {}
print(f"Usage: {usage}")
Force include usage bằng cách set parameters
llm = ChatOpenAI(
model="deepseek-v3.2",
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL,
model_kwargs={"include_usage": True} # ✅ Force usage
)
Nếu vẫn không có, kiểm tra raw response
import openai
client = OpenAI(api_key=HOLYSHEEP_API_KEY, base_url=HOLYSHEEP_BASE_URL)
raw = client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": "Hi"}]
)
print(f"Raw usage: {raw.usage}")
4. Streaming callback không tracking được token cuối
Nguyên nhân: on_llm_end có thể không được gọi đúng lúc khi streaming.
class RobustStreamingCallback(BaseCallbackHandler):
def __init__(self):
self.tokens = []
self.start = None
def on_llm_start(self, *args, **kwargs):
self.start = time.time()
self.tokens = []
def on_llm_new_token(self, token: str, **kwargs):
self.tokens.append(token)
def on_llm_end(self, response: Any, **kwargs):
# Token count chính xác từ response cuối
if response.generations:
gen = response.generations[0][0]
print(f"Final token count: {len(gen.text)}")
def on_llm_error(self, error: Exception, **kwargs):
# Fallback nếu stream bị interrupt
print(f"Error - tokens received: {len(self.tokens)}")
raise error
Sử dụng
streaming_llm = ChatOpenAI(
model="deepseek-v3.2",
streaming=True,
callbacks=[RobustStreamingCallback()]
)
5. Cost tracking sai với multi-step chain
Nguyên nhân: Callback không được chain đúng cách qua các step.
# Dùng Callbacks với RunnableSequence
from langchain.schema import StrOutputParser
Chain với callback tại mỗi step
chain = prompt | llm | StrOutputParser()
Gắn callback vào chain cuối cùng
chain_with_callback = chain.with_config(
callbacks=[CostTrackingCallback()]
)
Chạy chain
result = chain_with_callback.invoke({"topic": "LangChain"})
Hoặc gắn callback vào từng LLM riêng
prompt_with_cb = prompt.with_config(callbacks=[cb1])
llm_with_cb = llm.with_config(callbacks=[cb2])
parser_with_cb = StrOutputParser().with_config(callbacks=[cb3])
full_chain = prompt_with_cb | llm_with_cb | parser_with_cb
result = full_chain.invoke({"topic": "LangChain"})
Kết luận
Qua thực chiến triển khai LangChain callback với HolySheep AI, tôi nhận thấy việc tracking chi phí và độ trễ là không thể thiếu. Với DeepSeek V3.2 chỉ $0.42/MTok và tỷ giá ¥1=$1, chi phí vận hành giảm đáng kể so với các provider khác.
Bạn có thể tải code mẫu đầy đủ từ GitHub repository hoặc đăng ký HolySheep AI để bắt đầu thử nghiệm ngay hôm nay.
👉 Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký