LangGraph State Graph: Hướng Dẫn Toàn Diện Cho Người Mới Bắt Đầu

Bạn đã bao giờ tự hỏi làm sao để xây dựng một ứng dụng AI có thể "suy nghĩ" theo nhiều bước, nhớ được những gì đã làm trước đó, và đưa ra quyết định thông minh chưa? Trong bài viết này, mình sẽ hướng dẫn bạn từ con số 0 để hiểu và sử dụng LangGraph State Graph - một công cụ mạnh mẽ giúp bạn xây dựng ứng dụng AI với luồng xử lý phức tạp.

LangGraph Là Gì? Tại Sao Nó Quan Trọng?

Khi mới tìm hiểu về AI, nhiều người thường nghĩ AI chỉ đơn giản là "hỏi một câu, nhận một câu trả lời". Nhưng thực tế phức tạp hơn nhiều. LangGraph là thư viện mở rộng của LangChain, giúp bạn xây dựng đồ thị trạng thái (State Graph) - tức là ứng dụng AI có thể:

• Chuyển đổi giữa nhiều "trạng thái" khác nhau
• Nhớ và chia sẻ dữ liệu giữa các bước xử lý
• Quyết định bước tiếp theo dựa trên kết quả hiện tại
• Xử lý nhiều tác vụ phức tạp theo luồng có trật tự

So Sánh Chi Phí: HolySheep AI vs Các Nhà Cung Cấp Khác

Trước khi bắt đầu code, mình muốn chia sẻ về đăng ký tại đây để bạn có thể tiết kiệm đáng kể chi phí khi sử dụng API. Bảng so sánh giá dưới đây cho thấy rõ sự chênh lệch:

Model	Giá/1M Tokens	Tiết kiệm
GPT-4.1	$8.00	-
Claude Sonnet 4.5	$15.00	-
Gemini 2.5 Flash	$2.50	-
DeepSeek V3.2	$0.42	Tiết kiệm 85%+

Với HolySheep AI, bạn không chỉ tiết kiệm chi phí mà còn được hỗ trợ thanh toán qua WeChat và Alipay - rất tiện lợi cho người dùng Việt Nam. Độ trễ trung bình chỉ dưới 50ms, giúp ứng dụng của bạn phản hồi cực nhanh.

Cài Đặt Môi Trường

Đầu tiên, bạn cần cài đặt các thư viện cần thiết. Mở terminal và chạy lệnh sau:

pip install langgraph langchain-core langchain-holysheep python-dotenv

Sau đó, tạo file .env để lưu trữ API key của bạn:

# File: .env
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY

Xây Dựng Ứng Dụng Đầu Tiên Với LangGraph

Mình sẽ hướng dẫn bạn xây dựng một ứng dụng chatbot có khả năng:

1. Chào hỏi người dùng
2. Hỏi tên người dùng
3. Tùy theo câu trả lời mà chuyển sang trạng thái phù hợp

Bước 1: Định Nghĩa Trạng Thái (State)

State là nơi lưu trữ tất cả dữ liệu mà ứng dụng cần nhớ. Mình sẽ tạo một state đơn giản:

from typing import TypedDict, Annotated
import operator

class ChatState(TypedDict):
    """Định nghĩa trạng thái cho ứng dụng chat"""
    messages: list  # Lưu lịch sử hội thoại
    user_name: str  # Lưu tên người dùng
    current_step: str  # Theo dõi bước hiện tại

Bước 2: Tạo Các Node (Nút Xử Lý)

Node là các hàm xử lý logic. Mỗi node nhận state hiện tại, xử lý và trả về state mới:

from langchain_core.messages import HumanMessage, AIMessage
from dotenv import load_dotenv
import os

load_dotenv()

Kết nối với HolySheep AI
from langchain_holysheep import ChatHolySheep

llm = ChatHolySheep(
    model="deepseek-v3.2",
    base_url="https://api.holysheep.ai/v1",
    api_key=os.getenv("HOLYSHEEP_API_KEY")
)

def greet_node(state: ChatState) -> ChatState:
    """Node chào hỏi người dùng"""
    greeting = "Xin chào! Mình là trợ lý AI. Bạn tên gì vậy?"
    
    # Cập nhật state
    state["messages"].append(AIMessage(content=greeting))
    state["current_step"] = "waiting_name"
    
    return state

def process_name_node(state: ChatState) -> ChatState:
    """Node xử lý tên người dùng"""
    # Lấy tin nhắn cuối cùng của người dùng
    last_message = state["messages"][-1].content
    
    # Gọi LLM để phân tích
    prompt = f"Người dùng trả lời: '{last_message}'. Hãy trích xuất tên của họ."
    
    response = llm.invoke(prompt)
    extracted_name = response.content.strip()
    
    # Cập nhật state với tên người dùng
    state["user_name"] = extracted_name
    state["messages"].append(AIMessage(content=f"Rất vui được gặp bạn, {extracted_name}!"))
    state["current_step"] = "conversation"
    
    return state

def chat_node(state: ChatState) -> ChatState:
    """Node trò chuyện thông thường"""
    # Lấy lịch sử hội thoại
    conversation_history = "\n".join([
        f"{'AI' if isinstance(m, AIMessage) else 'User'}: {m.content}" 
        for m in state["messages"]
    ])
    
    response = llm.invoke(conversation_history)
    state["messages"].append(AIMessage(content=response.content))
    
    return state

Bước 3: Xây Dựng Đồ Thị (Graph)

Graph kết nối các node lại với nhau và định nghĩa luồng di chuyển:

from langgraph.graph import StateGraph, START, END

Tạo đồ thị
workflow = StateGraph(ChatState)

Thêm các node
workflow.add_node("greet", greet_node)
workflow.add_node("process_name", process_name_node)
workflow.add_node("chat", chat_node)

Định nghĩa các cạnh (edges) - luồng xử lý
workflow.add_edge(START, "greet")
workflow.add_edge("greet", "process_name")

Cạnh có điều kiện: quyết định bước tiếp theo
def should_continue(state: ChatState) -> str:
    if state["current_step"] == "waiting_name":
        return "process_name"
    elif state["current_step"] == "conversation":
        return "chat"
    return END

workflow.add_conditional_edges(
    "process_name",
    should_continue,
    {
        "chat": "chat",
        END: END
    }
)

Tiếp tục chat sau mỗi lượt
workflow.add_conditional_edges(
    "chat",
    lambda state: "chat" if len(state["messages"]) < 10 else END,
    {"chat": "chat", END: END}
)

Compile đồ thị
app = workflow.compile()

print("Đồ thị đã được tạo thành công!")

Bước 4: Chạy Ứng Dụng

# Chạy ứng dụng
initial_state = ChatState(
    messages=[],
    user_name="",
    current_step="start"
)

Stream kết quả để xem từng bước
print("=== Bắt đầu hội thoại ===\n")

for event in app.stream({"messages": [HumanMessage(content="Tôi tên Minh")]}, 
                        stream_mode="values"):
    if "messages" in event:
        for msg in event["messages"]:
            if isinstance(msg, AIMessage):
                print(f"🤖 AI: {msg.content}\n")
            elif isinstance(msg, HumanMessage):
                print(f"👤 User: {msg.content}\n")

Ứng Dụng Thực Tế: Hệ Thống Hỗ Trợ Khách Hàng

Sau khi hiểu cách hoạt động cơ bản, mình sẽ hướng dẫn bạn xây dựng một hệ thống hỗ trợ khách hàng hoàn chỉnh hơn:

from enum import Enum
from typing import Union

class SupportState(TypedDict):
    """State cho hệ thống hỗ trợ khách hàng"""
    messages: list
    issue_type: str
    issue_description: str
    resolution_status: str
    escalation_needed: bool

class IssueType(Enum):
    """Các loại vấn đề được hỗ trợ"""
    BILLING = "billing"
    TECHNICAL = "technical"
    GENERAL = "general"
    ESCALATE = "escalate"

def triage_node(state: SupportState) -> SupportState:
    """Node phân loại vấn đề"""
    last_message = state["messages"][-1].content
    
    prompt = f"""Phân loại vấn đề sau thành 1 trong 4 loại:
    - billing: Vấn đề về thanh toán, hóa đơn
    - technical: Vấn đề kỹ thuật, lỗi phần mềm
    - general: Câu hỏi chung
    - escalate: Cần chuyển lên cấp cao hơn
    
    Vấn đề: {last_message}
    
    Chỉ trả lời: billing/technical/general/escalate"""
    
    response = llm.invoke(prompt)
    issue_type = response.content.strip().lower()
    
    state["issue_type"] = issue_type
    state["messages"].append(AIMessage(
        content=f"Tôi đã hiểu vấn đề của bạn. Để tôi chuyển đến bộ phận phù hợp..."
    ))
    
    return state

def billing_handler(state: SupportState) -> SupportState:
    """Xử lý vấn đề thanh toán"""
    state["resolution_status"] = "in_progress"
    state["messages"].append(AIMessage(
        content="Tôi đang kiểm tra thông tin thanh toán của bạn. Vui lòng đợi một chút..."
    ))
    
    # Giả lập xử lý
    state["messages"].append(AIMessage(
        content="Tôi đã tìm thấy vấn đề. Bạn có thể kiểm tra lại thông tin thẻ không?"
    ))
    
    return state

def technical_handler(state: SupportState) -> SupportState:
    """Xử lý vấn đề kỹ thuật"""
    state["resolution_status"] = "in_progress"
    state["messages"].append(AIMessage(
        content="Đội ngũ kỹ thuật đang xem xét vấn đề của bạn..."
    ))
    
    return state

def escalate_handler(state: SupportState) -> SupportState:
    """Chuyển vấn đề lên cấp cao hơn"""
    state["escalation_needed"] = True
    state["messages"].append(AIMessage(
        content="Vấn đề của bạn khá phức tạp. Tôi sẽ chuyển đến chuyên viên hỗ trợ cao cấp."
    ))
    
    return state

Xây dựng workflow cho hệ thống hỗ trợ
support_graph = StateGraph(SupportState)

support_graph.add_node("triage", triage_node)
support_graph.add_node("billing", billing_handler)
support_graph.add_node("technical", technical_handler)
support_graph.add_node("escalate", escalate_handler)

support_graph.add_edge(START, "triage")

Định tuyến dựa trên loại vấn đề
def route_issue(state: SupportState) -> str:
    issue_type = state.get("issue_type", "general")
    
    routes = {
        "billing": "billing",
        "technical": "technical", 
        "escalate": "escalate"
    }
    
    return routes.get(issue_type, "escalate")

support_graph.add_conditional_edges(
    "triage",
    route_issue,
    {
        "billing": "billing",
        "technical": "technical",
        "escalate": "escalate"
    }
)

support_graph.add_edge("billing", END)
support_graph.add_edge("technical", END)
support_graph.add_edge("escalate", END)

support_app = support_graph.compile()

print("Hệ thống hỗ trợ khách hàng đã sẵn sàng!")

Lỗi Thường Gặp và Cách Khắc Phục

Trong quá trình làm việc với LangGraph, mình đã gặp nhiều lỗi phổ biến. Dưới đây là cách mình xử lý từng lỗi:

1. Lỗi "API Key Not Found" hoặc Authentication Error

# ❌ SAI: Key không đúng hoặc chưa load
llm = ChatHolySheep(
    model="deepseek-v3.2",
    api_key="YOUR_HOLYSHEEP_API_KEY"  # Key này là ví dụ
)

✅ ĐÚNG: Load từ biến môi trường
from dotenv import load_dotenv
import os

load_dotenv()  # Load file .env

llm = ChatHolySheep(
    model="deepseek-v3.2",
    base_url="https://api.holysheep.ai/v1",
    api_key=os.environ.get("HOLYSHEEP_API_KEY")  # Đọc từ biến môi trường
)

Kiểm tra key có tồn tại không
if not os.environ.get("HOLYSHEEP_API_KEY"):
    raise ValueError("Vui lòng thiết lập HOLYSHEEP_API_KEY trong file .env")

2. Lỗi "State Key Not Found" khi truy cập state

# ❌ SAI: Truy cập key không tồn tại
def bad_node(state: ChatState) -> ChatState:
    user_city = state["city"]  # Key "city" không có trong State!
    return state

✅ ĐÚNG: Kiểm tra key trước khi truy cập
def good_node(state: ChatState) -> ChatState:
    # Cách 1: Dùng get() với default value
    user_city = state.get("city", "Không xác định")
    
    # Cách 2: Kiểm tra tồn tại
    if "city" not in state:
        state["city"] = "Hồ Chí Minh"  # Set giá trị mặc định
    
    return state

3. Lỗi "Infinite Loop" - Vòng lặp vô hạn

# ❌ SAI: Không có điều kiện dừng
workflow.add_conditional_edges(
    "chat_node",
    lambda state: "chat_node",  # Luôn quay lại → vô hạn!
    {"chat_node": "chat_node"}
)

✅ ĐÚNG: Thêm điều kiện dừng rõ ràng
MAX_TURNS = 10

workflow.add_conditional_edges(
    "chat_node",
    lambda state: (
        "chat_node" 
        if len(state.get("messages", [])) < MAX_TURNS 
        else END  # Dừng sau 10 lượt
    ),
    {"chat_node": "chat_node", END: END}
)

Hoặc theo nội dung tin nhắn
STOP_WORDS = ["tạm biệt", "cảm ơn bạn", "goodbye"]

workflow.add_conditional_edges(
    "chat_node",
    lambda state: (
        "chat_node" 
        if not any(word in state["messages"][-1].content.lower() 
                   for word in STOP_WORDS) 
        else END
    ),
    {"chat_node": "chat_node", END: END}
)

4. Lỗi Type khi định nghĩa State

# ❌ SAI: Thiếu type hints
class BadState(TypedDict):
    messages  # Thiếu type!

✅ ĐÚNG: Đầy đủ type hints
class GoodState(TypedDict):
    messages: list[str]  # Python 3.9+
    count: int
    is_active: bool
    user_data: dict | None  # Union type

Với Python 3.8, dùng:
from typing import List, Optional, Dict, Union
class CompatibleState(TypedDict):
    messages: List[str]
    count: int
    is_active: bool
    user_data: Optional[Dict]

Mẹo Tối Ưu Hiệu Suất

• Sử dụng checkpointing: Lưu trạng thái để có thể tiếp tục từ điểm dừng
• Tối ưu prompts: Viết prompts ngắn gọn, rõ ràng để giảm token sử dụng
• Cache responses: Lưu lại các câu trả lời thường dùng
• Batch processing: Xử lý nhiều request cùng lúc khi có thể

Kết Luận

LangGraph State Graph là công cụ cực kỳ mạnh mẽ để xây dựng ứng dụng AI phức tạp. Qua bài viết này, bạn đã học được cách:

• Định nghĩa State và các Node
• Xây dựng đồ thị với các cạnh có điều kiện
• Tạo ứng dụng hội thoại đa bước
• Xây dựng hệ thống hỗ trợ khách hàng tự động
• Xử lý các lỗi phổ biến

Với chi phí chỉ $0.42/1M tokens cho DeepSeek V3.2 và độ trễ dưới 50ms, HolySheep AI là lựa chọn tối ưu cho developers Việt Nam. Thanh toán qua WeChat/Alipay cũng rất thuận tiện.

👉 Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký