Mở đầu: Khi Agent "im lặng" — Debug thực chiến với LangGraph
Tôi vẫn nhớ rõ buổi sáng tháng 3 năm 2025, hệ thống chatbot AI của khách hàng hoàn toàn "chết" lúc 9h47. Người dùng phản hồi lỗi ConnectionError: timeout after 30s, và console backend tràn ngập traceback rất khó đọc. Sau 4 tiếng đồng hồ debug, nguyên nhân gốc chỉ là một state node trong LangGraph bị loop vô hạn vì thiếu điều kiện dừng. Kể từ đó, tôi luôn đặt monitoring lên hàng đầu khi triển khai multi-agent system.
Bài viết này sẽ hướng dẫn bạn 监控与调试 LangGraph (Monitoring & Debugging LangGraph) từ cơ bản đến nâng cao: cách ghi log execution trajectory, visualize agent behavior, và xử lý các lỗi phổ biến. Toàn bộ ví dụ sử dụng HolySheep AI với chi phí chỉ bằng 1/6 so với OpenAI ($0.42/1M tokens cho DeepSeek V3.2).
1. Tại sao cần giám sát LangGraph Agent?
LangGraph tạo ra directed graph với nhiều node và edge. Khi agent phức tạp dần, việc trace execution path trở nên cực kỳ khó khăn:
- Loop không kiểm soát: Agent quay lại state trước đó và lặp vô hạn
- Latency bất thường: Một số bước mất 5-10 giây thay vì vài trăm ms
- Memory leak: ConversationHistory tích lũy không giới hạn
- Token explosion: Context window tràn vì không cắt truncate đúng lúc
Với HolyShehe AI, bạn có thể theo dõi chi phí theo thời gian thực — chỉ $0.42/1M tokens cho DeepSeek V3.2 và latency trung bình <50ms (theo benchmark tháng 1/2026).
2. Cài đặt môi trường và kết nối HolySheep AI
# Cài đặt dependencies cần thiết
pip install langgraph langchain-core langchain-holysheep
pip install langsmith plotly dash
Hoặc cài đặt đầy đủ hơn cho visualization
pip install "langgraph[instrumentation]" networkx matplotlib
Kiểm tra phiên bản
python -c "import langgraph; print(langgraph.__version__)"
Output mong đợi: 0.2.x hoặc cao hơn
# Cấu hình API key và biến môi trường
import os
QUAN TRỌNG: Không bao giờ hardcode API key trong production
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
Cấu hình base_url cho HolySheep AI
os.environ["HOLYSHEEP_BASE_URL"] = "https://api.holysheep.ai/v1"
Bật LangSmith để track execution (tùy chọn nhưng khuyến nghị)
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your_langsmith_key"
Verify kết nối
from langchain_holysheep import ChatHolySheep
chat = ChatHolySheep(model="deepseek-v3.2")
response = chat.invoke("Kiểm tra kết nối: Trả lời OK nếu nhận được")
print(f"Kết quả: {response.content}")3. Xây dựng Agent với built-in State Validation
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.prebuilt import ToolNode
import operator
Định nghĩa state schema với validation
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
step_count: int
total_tokens: int
last_error: str | None
Giới hạn quan trọng để tránh loop vô hạn
MAX_STEPS = 10
MAX_TOKENS_PER_REQUEST = 4000
def create_monitored_agent():
"""Tạo agent với monitoring và guardrails tích hợp"""
def should_continue(state: AgentState) -> str:
"""Quyết định tiếp tục hay dừng - ĐÂY LÀ NƠI THƯỜNG GÂY LỖI"""
messages = state["messages"]
last_message = messages[-1]
# Lỗi phổ biến #1: Không kiểm tra số bước → infinite loop
if state["step_count"] >= MAX_STEPS:
print(f"⚠️ Cảnh báo: Đạt giới hạn {MAX_STEPS} bước!")
return "end"
# Kiểm tra nếu agent đã kết thúc (có finish keyword)
if hasattr(last_message, "content"):
if "KẾT THÚC" in last_message.content.upper():
return "end"
return "continue"
def call_model(state: AgentState):
"""Gọi LLM qua HolySheep AI với retry logic"""
from langchain_holysheep import ChatHolySheep
from langchain_core.messages import HumanMessage
messages = state["messages"]
step = state["step_count"]
try:
# Sử dụng HolySheep với retry
llm = ChatHolySheep(model="deepseek-v3.2")
# Cắt bớt messages nếu quá dài (tránh token explosion)
if len(str(messages)) > MAX_TOKENS_PER_REQUEST * 4:
messages = messages[-10:] # Giữ 10 messages gần nhất
response = llm.invoke(messages)
# Cập nhật state với metrics
new_state = {
"messages": [response],
"step_count": step + 1,
"total_tokens": state["total_tokens"] + response.usage.total_tokens if hasattr(response, 'usage') else state["total_tokens"],
"last_error": None
}
print(f"📊 Bước {step + 1}: {response.content[:50]}...")
return new_state
except Exception as e:
error_msg = f"Error at step {step}: {type(e).__name__}: {str(e)}"
print(f"❌ {error_msg}")
return {
"messages": state["messages"],
"step_count": step,
"total_tokens": state["total_tokens"],
"last_error": error_msg
}
# Xây dựng graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {
"continue": "agent",
"end": END
})
return workflow.compile()
Khởi tạo agent
agent = create_monitored_agent()
Test với một câu hỏi
initial_state = {
"messages": [HumanMessage(content="Phân tích xu hướng AI 2026 và đưa ra 3 đề xuất")],
"step_count": 0,
"total_tokens": 0,
"last_error": None
}
result = agent.invoke(initial_state)
print(f"\n✅ Hoàn thành sau {result['step_count']} bước")
print(f"💰 Tổng tokens: {result['total_tokens']}")
print(f"💵 Chi phí ước tính: ${result['total_tokens'] / 1_000_000 * 0.42:.4f}")4. Visualize Execution Trajectory với Custom Dashboard
import json
from datetime import datetime
from pathlib import Path
class LangGraphDebugger:
"""Debug và visualize LangGraph execution trajectory"""
def __init__(self, log_dir: str = "./logs"):
self.log_dir = Path(log_dir)
self.log_dir.mkdir(exist_ok=True)
self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S")
self.trajectory = []
def log_step(self, node_name: str, state_before: dict, state_after: dict):
"""Ghi log mỗi bước execution"""
step_data = {
"timestamp": datetime.now().isoformat(),
"session_id": self.session_id,
"node": node_name,
"step": state_after.get("step_count", 0),
"state_before": {
"step_count": state_before.get("step_count"),
"message_count": len(state_before.get("messages", []))
},
"state_after": {
"step_count": state_after.get("step_count"),
"message_count": len(state_after.get("messages", [])),
"total_tokens": state_after.get("total_tokens", 0),
"last_error": state_after.get("last_error")
},
"latency_ms": (datetime.now().timestamp() -
datetime.fromisoformat(self.trajectory[-1]["timestamp"]).timestamp() * 1000
if self.trajectory else 0)
}
self.trajectory.append(step_data)
# In ra console với format dễ đọc
status = "✅" if not step_data["state_after"]["last_error"] else "❌"
print(f"{status} [{step_data['timestamp'][11:19]}] {node_name} | "
f"Step {step_data['step']} | Tokens: {step_data['state_after']['total_tokens']}")
def export_json(self, filename: str = None):
"""Export trajectory ra JSON để phân tích"""
if filename is None:
filename = f"trajectory_{self.session_id}.json"
filepath = self.log_dir / filename
with open(filepath, "w", encoding="utf-8") as f:
json.dump({
"session_id": self.session_id,
"total_steps": len(self.trajectory),
"trajectory": self.trajectory
}, f, indent=2, ensure_ascii=False)
print(f"📁 Đã export: {filepath}")
return str(filepath)
def generate_html_report(self):
"""Tạo HTML report với bảng timeline"""
html_content = f"""
LangGraph Debug Report - {self.session_id}
🔍 LangGraph Execution Report
Session ID: {self.session_id}
Tổng số bước: {len(self.trajectory)}
Thời gian: {self.trajectory[0]['timestamp']} → {self.trajectory[-1]['timestamp']}
Thời gian
Bước
Node
Messages
Tokens
Trạng thái
"""
for step in self.trajectory:
error_class = "error" if step["state_after"]["last_error"] else "success"
status = f"❌ {step['state_after']['last_error']}" if step["state_after"]["last_error"] else "✅ OK"
html_content += f"""
{step['timestamp'][11:19]}
{step['step']}
{step['node']}
{step['state_after']['message_count']}
{step['state_after']['total_tokens']}
{status}
"""
html_content += """
"""
filepath = self.log_dir / f"report_{self.session_id}.html"
with open(filepath, "w", encoding="utf-8") as f:
f.write(html_content)
print(f"📊 HTML report: {filepath}")
return str(filepath)
Sử dụng debugger
debugger = LangGraphDebugger()
Wrapper để tự động log mỗi bước
from functools import wraps
def with_debugging(agent_func):
@wraps(agent_func)
def wrapper(state):
node_name = "agent"
state_before = state.copy()
result = agent_func(state)
debugger.log_step(node_name, state_before, result)
return result
return wrapper
Áp dụng debugging vào agent
original_call_model = None # Sẽ được gán trong production
Chạy test và xem kết quả
debugger.generate_html_report()
debugger.export_json()5. Tích hợp LangSmith Tracing (Khuyến nghị cho Production)
# langsmith_integration.py
from langsmith.run_helpers import traceable
from langchain_core.callbacks import CallbackManager
from langchain_holysheep import ChatHolySheep
Cấu hình LangSmith với HolySheep
import os
os.environ["LANGSMITH_PROJECT"] = "langgraph-monitoring"
@traceable(
name="multi-step-agent",
tags=["production", "holy-sheep"],
metadata={"model": "deepseek-v3.2", "provider": "holysheep"}
)
def run_agent_with_tracing(user_input: str, context: dict = None):
"""
Agent được trace hoàn chỉnh với LangSmith.
Xem dashboard tại: https://smith.langchain.com/
"""
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
import operator
class State(TypedDict):
messages: Annotated[Sequence, operator.add]
iterations: int
# Khởi tạo LLM với HolySheep
llm = ChatHolySheep(
model="deepseek-v3.2",
temperature=0.7,
max_tokens=2000
)
def agent_node(state: State):
response = llm.invoke(state["messages"])
return {"messages": [response], "iterations": state["iterations"] + 1}
def should_continue(state: State):
if state["iterations"] >= 3:
return "end"
return "continue"
# Xây dựng graph
graph = StateGraph(State)
graph.add_node("agent", agent_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {
"continue": "agent",
"end": END
})
app = graph.compile()
result = app.invoke({
"messages": [HumanMessage(content=user_input)],
"iterations": 0
})
return result["messages"][-1].content
Test với input mẫu
if __name__ == "__main__":
result = run_agent_with_tracing(
"Phân tích: Ưu điểm của việc dùng LangGraph thay vì LangChain Chain?"
)
print(f"Response: {result[:200]}...")6. Real-time Monitoring với Prometheus + Grafana
# metrics_collector.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
from typing import Optional
import time
Định nghĩa metrics
AGENT_STEPS = Counter(
'langgraph_agent_steps_total',
'Total số bước agent thực thi',
['session_id', 'status']
)
AGENT_LATENCY = Histogram(
'langgraph_step_latency_seconds',
'Độ trễ mỗi bước agent',
['node_name'],
buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)
TOKEN_USAGE = Histogram(
'langgraph_token_usage',
'Số tokens sử dụng',
['model'],
buckets=[100, 500, 1000, 5000, 10000, 50000]
)
COST_ESTIMATE = Gauge(
'langgraph_cost_usd',
'Chi phí ước tính theo USD',
['model']
)
Pricing map cho HolySheep AI (cập nhật 01/2026)
HOLYSHEEP_PRICING = {
"gpt-4.1": 8.0, # $8/1M tokens
"claude-sonnet-4.5": 15.0, # $15/1M tokens
"gemini-2.5-flash": 2.50, # $2.50/1M tokens
"deepseek-v3.2": 0.42, # $0.42/1M tokens (RẺ NHẤT!)
}
class MetricsCollector:
"""Thu thập metrics cho LangGraph agent"""
def __init__(self, port: int = 9090):
self.port = port
self.current_session = None
self.step_count = 0
self.total_cost = 0.0
def start_server(self):
"""Khởi động Prometheus metrics server"""
start_http_server(self.port)
print(f"📊 Metrics server chạy tại http://localhost:{self.port}/metrics")
def record_step(self, session_id: str, node: str, latency: float,
tokens: int, model: str, success: bool = True):
"""Ghi lại metrics của một bước execution"""
self.step_count += 1
status = "success" if success else "error"
# Counter cho số bước
AGENT_STEPS.labels(session_id=session_id, status=status).inc()
# Histogram cho latency
AGENT_LATENCY.labels(node_name=node).observe(latency)
# Histogram cho tokens
TOKEN_USAGE.labels(model=model).observe(tokens)
# Tính chi phí dựa trên pricing
if model in HOLYSHEEP_PRICING:
cost_per_token = HOLYSHEEP_PRICING[model] / 1_000_000
step_cost = tokens * cost_per_token
self.total_cost += step_cost
COST_ESTIMATE.labels(model=model).set(self.total_cost)
def get_summary(self) -> dict:
"""Lấy tổng kết metrics"""
return {
"total_steps": self.step_count,
"total_cost_usd": round(self.total_cost, 6),
"avg