LangGraph 条件分支：动态路由与循环控制流设计

Chào các bạn, mình là Minh — Tech Lead tại một startup AI tại Việt Nam. Hôm nay mình muốn chia sẻ hành trình 6 tháng xây dựng hệ thống AI agent phức tạp với LangGraph, từ những thử thách đầu tiên với control flow đến cách chúng tôi tối ưu chi phí với HolySheep AI.

Vì sao cần conditional branching trong LangGraph?

Khi xây dựng AI agent cho hệ thống chăm sóc khách hàng thông minh, mình gặp ngay bài toán: cùng một input nhưng cần xử lý theo nhiều nhánh khác nhau. Đơn giản nhất là phân loại intent:

• Khách hỏi về sản phẩm → Nhánh product_info
• Khách muốn đổi trả → Nhánh return_policy
• Khách phàn nàn → Nhánh escalation
• Không xác định được → Nhánh clarification

Với LangGraph, mình sử dụng StateGraph với conditional edges để handle chính xác luồng này. Đây là kiến trúc mà chúng tôi đã build thành công với hơn 50,000 requests mỗi ngày.

Cài đặt môi trường và kết nối HolySheep AI

Trước khi đi vào code, mình muốn nhấn mạnh: chúng tôi đã migrate từ OpenAI sang HolySheep AI và tiết kiệm được 85% chi phí. Tỷ giá chỉ ¥1 = $1, latency trung bình dưới 50ms. Đăng ký ngay để nhận tín dụng miễn phí khi bắt đầu.

# Cài đặt dependencies cần thiết
pip install langgraph langchain-core langchain-holysheep

Cấu hình HolySheep AI API
import os
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["HOLYSHEEP_BASE_URL"] = "https://api.holysheep.ai/v1"

Xây dựng Customer Support Agent với Conditional Routing

Dưới đây là kiến trúc hoàn chỉnh mà chúng tôi đang sử dụng production. Mình sẽ giải thích từng phần:

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.constants import Send
from langchain_holysheep import ChatHolySheep
import operator

Định nghĩa state schema cho agent
class CustomerSupportState(TypedDict):
    user_input: str
    intent: str
    extracted_entities: dict
    conversation_history: list
    current_node: str
    response: str
    escalation_needed: bool

Khởi tạo LLM với HolySheep AI
llm = ChatHolySheep(
    model="gpt-4.1",  # $8/MTok - tiết kiệm 85% so với OpenAI
    base_url="https://api.holysheep.ai/v1",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    temperature=0.3
)

Node: Intent Classification
def classify_intent(state: CustomerSupportState) -> CustomerSupportState:
    """Phân loại intent từ user input"""
    messages = [
        {"role": "system", "content": """Bạn là classifier. Phân loại intent:
        - product_info: hỏi về sản phẩm, giá, tính năng
        - return_policy: yêu cầu đổi trả, hoàn tiền
        - escalation: phàn nàn, khiếu nại, tình trạng khẩn cấp
        - general: câu hỏi chung khác
        
        Trả về JSON: {"intent": "xxx", "confidence": 0.xx}"""},
        {"role": "user", "content": state["user_input"]}
    ]
    
    response = llm.invoke(messages)
    import json
    result = json.loads(response.content)
    
    return {
        "intent": result["intent"],
        "current_node": "classifier"
    }

Node: Xử lý thông tin sản phẩm
def handle_product_info(state: CustomerSupportState) -> CustomerSupportState:
    """Trả lời câu hỏi về sản phẩm"""
    messages = [
        {"role": "system", "content": "Bạn là tư vấn sản phẩm. Cung cấp thông tin chi tiết."},
        {"role": "user", "content": state["user_input"]}
    ]
    response = llm.invoke(messages)
    
    return {
        "response": response.content,
        "current_node": "product_info"
    }

Node: Xử lý đổi trả
def handle_return(state: CustomerSupportState) -> CustomerSupportState:
    """Xử lý yêu cầu đổi trả"""
    messages = [
        {"role": "system", "content": """Bạn là agent đổi trả. Hỏi thông tin:
        1. Số order
        2. Lý do đổi trả
        3. Yêu cầu: đổi size/màu hay hoàn tiền"""},
        {"role": "user", "content": state["user_input"]}
    ]
    response = llm.invoke(messages)
    
    return {
        "response": response.content,
        "current_node": "return_policy"
    }

Node: Escalation handler
def handle_escalation(state: CustomerSupportState) -> CustomerSupportState:
    """Chuyển đến human agent"""
    return {
        "response": "Đang kết nối với agent...",
        "escalation_needed": True,
        "current_node": "escalation"
    }

Định nghĩa Conditional Edges - Trái tim của routing logic

Đây là phần quan trọng nhất — cách mình thiết kế dynamic routing. Mình đã thử nghiệm nhiều pattern và cuối cùng chọn cách này vì clean và maintainable:

# Hàm route chính - điều hướng based on intent
def route_based_on_intent(state: CustomerSupportState) -> Literal[
    "product_info", "return_policy", "escalation", "general"
]:
    """
    Dynamic routing: Chuyển đến node phù hợp dựa trên intent classification
    """
    intent = state.get("intent", "general").lower()
    
    routing_map = {
        "product_info": "product_info",
        "return_policy": "return_policy",
        "escalation": "escalation"
    }
    
    return routing_map.get(intent, "general")

Xây dựng graph
workflow = StateGraph(CustomerSupportState)

Thêm tất cả nodes
workflow.add_node("classifier", classify_intent)
workflow.add_node("product_info", handle_product_info)
workflow.add_node("return_policy", handle_return)
workflow.add_node("escalation", handle_escalation)
workflow.add_node("general", lambda s: {**s, "current_node": "general"})

Conditional edges: classifier -> các node khác
workflow.add_conditional_edges(
    "classifier",
    route_based_on_intent,
    {
        "product_info": "product_info",
        "return_policy": "return_policy",
        "escalation": "escalation",
        "general": "general"
    }
)

Tất cả nodes đều kết thúc tại END
workflow.add_edge("product_info", END)
workflow.add_edge("return_policy", END)
workflow.add_edge("escalation", END)
workflow.add_edge("general", END)

Set entry point
workflow.set_entry_point("classifier")

Compile graph
customer_support_graph = workflow.compile()

Test với các input khác nhau
test_inputs = [
    {"user_input": "Sản phẩm A có mấy màu?", "conversation_history": []},
    {"user_input": "Tôi muốn đổi sang size L", "conversation_history": []},
    {"user_input": "Giao hàng chậm quá, tôi phàn nàn!", "conversation_history": []}
]

for inp in test_inputs:
    result = customer_support_graph.invoke(inp)
    print(f"Intent: {result['intent']} -> Node: {result['current_node']}")

Triển khai Loop Control - Xử lý multi-turn conversation

Trong thực tế, một conversation không chỉ đơn giản là request-response. Mình cần loop để handle clarification liên tiếp. Đây là pattern mà team mình đã optimize:

from typing import Union

class MultiTurnSupportState(TypedDict):
    messages: list
    iteration_count: int
    max_iterations: int
    needs_clarification: bool
    final_response: str

def should_continue_loop(state: MultiTurnSupportState) -> Union[str, Literal["__end__"]]:
    """Quyết định tiếp tục loop hay kết thúc"""
    
    # Điều kiện dừng
    if state["iteration_count"] >= state["max_iterations"]:
        return "__end__"
    
    if not state.get("needs_clarification", False):
        return "__end__"
    
    # Tiếp tục loop
    return "continue"

def clarification_node(state: MultiTurnSupportState) -> MultiTurnSupportState:
    """Node xử lý clarification - hỏi thêm thông tin"""
    messages = state["messages"]
    
    prompt = f"""Dựa trên lịch sử hội thoại, xác định:
    1. Còn thiếu thông tin gì?
    2. Có cần hỏi thêm user không?
    
    Lịch sử: {messages[-3:]}"""
    
    response = llm.invoke([{"role": "user", "content": prompt}])
    
    import json
    try:
        info = json.loads(response.content)
        needs_more = info.get("needs_clarification", False)
    except:
        needs_more = False
    
    return {
        "messages": messages + [{"role": "assistant", "content": response.content}],
        "iteration_count": state["iteration_count"] + 1,
        "needs_clarification": needs_more,
        "final_response": response.content if not needs_more else ""
    }

def final_response_node(state: MultiTurnSupportState) -> MultiTurnSupportState:
    """Tổng hợp và trả lời final"""
    summary_prompt = f"""Tổng hợp thông tin từ các messages và đưa ra câu trả lời final:
    
    Messages: {state['messages']}"""
    
    response = llm.invoke([{"role": "user", "content": summary_prompt}])
    
    return {
        "final_response": response.content,
        "needs_clarification": False
    }

Xây dựng loop graph
loop_graph = StateGraph(MultiTurnSupportState)

loop_graph.add_node("clarification", clarification_node)
loop_graph.add_node("final_response", final_response_node)

Conditional edge để control loop
loop_graph.add_conditional_edges(
    "clarification",
    should_continue_loop,
    {
        "continue": "clarification",  # Loop back
        "__end__": "final_response"
    }
)

loop_graph.add_edge("final_response", END)
loop_graph.set_entry_point("clarification")

compiled_loop_graph = loop_graph.compile()

Run với max 3 iterations
result = compiled_loop_graph.invoke({
    "messages": [{"role": "user", "content": "Tôi muốn đổi đồ"}],
    "iteration_count": 0,
    "max_iterations": 3,
    "needs_clarification": True,
    "final_response": ""
})

print(f"Hoàn thành sau {result['iteration_count']} lần loop")
print(f"Response: {result['final_response'][:200]}...")

So sánh chi phí: OpenAI vs HolySheep AI

Bảng dưới đây là thực tế chúng tôi đã tính toán sau khi migrate:

Model	OpenAI	HolySheep AI	Tiết kiệm
GPT-4.1	$30/MTok	$8/MTok	73%
Claude Sonnet 4.5	$45/MTok	$15/MTok	67%
Gemini 2.5 Flash	$10/MTok	$2.50/MTok	75%
DeepSeek V3.2	$2.80/MTok	$0.42/MTok	85%

Với 1 triệu tokens/month sử dụng GPT-4.1, chúng tôi tiết kiệm được $22,000 mỗi tháng!

Lỗi thường gặp và cách khắc phục

1. Lỗi "State key not found" khi truy cập state

Nguyên nhân: Không định nghĩa đầy đủ các keys trong TypedDict hoặc return sai keys từ node.

# ❌ SAI: Thiếu keys hoặc return sai format
def bad_node(state):
    return {"new_field": "value"}  # Lost original state!

✅ ĐÚNG: Merge với state hiện tại
def good_node(state: CustomerSupportState) -> CustomerSupportState:
    return {
        **state,  # Giữ nguyên tất cả keys cũ
        "new_field": "value",
        "updated": True
    }

✅ Cách khác: Sử dụng update
def another_good_node(state: CustomerSupportState) -> CustomerSupportState:
    state["new_field"] = "value"
    return state

2. Lỗi "Conditional edge function returned unknown node"

Nguyên nhân: Hàm route trả về node name không tồn tại trong graph.

# ❌ SAI: Return giá trị không có trong mapping
def bad_router(state):
    intent = state["intent"]
    if intent == "order":
        return "track_order"  # Node này chưa được add!
    return "general"

✅ ĐÚNG: Luôn đảm bảo node tồn tại
workflow.add_node("track_order", track_order_node)

def good_router(state) -> Literal["track_order", "general", "escalation"]:
    intent = state["intent"]
    if intent == "order":
        return "track_order"  # Đã được add ở trên
    return "general"

3. Lỗi "Maximum iterations exceeded" - Infinite loop

Nguyên nhân: Logic loop không có điều kiện dừng hoặc điều kiện không bao giờ satisfied.

# ❌ NGUY HIỂM: Không có max_iterations
def unsafe_loop_condition(state):
    return state.get("needs_response", True)  # Có thể never becomes False!

✅ AN TOÀN: Luôn có max iterations
MAX_ITERATIONS = 5

def safe_loop_condition(state: MultiTurnSupportState):
    # Điều kiện 1: Đã đạt max iterations
    if state.get("iteration_count", 0) >= MAX_ITERATIONS:
        print(f"⚠️ Max iterations reached: {MAX_ITERATIONS}")
        return "__end__"
    
    # Điều kiện 2: Hoàn thành mục tiêu
    if not state.get("needs_clarification", True):
        return "__end__"
    
    # Điều kiện 3: Loop tiếp
    return "continue"

Thêm logging để debug
def monitored_node(state):
    print(f"📍 Node called at iteration {state.get('iteration_count', 0)}")
    result = process_node(state)
    print(f"✅ Node completed, needs_clarification={result.get('needs_clarification')}")
    return result

4. Lỗi API Authentication với HolySheep

Nguyên nhân: API key không đúng format hoặc chưa được set đúng cách.

# ❌ SAI: Không set environment hoặc sai key
import os
os.environ["HOLYSHEEP_API_KEY"] = "sk-xxx"  # Commented out!
llm = ChatHolySheep(model="gpt-4.1")

✅ ĐÚNG: Set environment trước khi khởi tạo
import os

Cách 1: Set trực tiếp
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"

Cách 2: Qua constructor
llm = ChatHolySheep(
    model="gpt-4.1",
    base_url="https://api.holysheep.ai/v1",  # ⚠️ PHẢI đúng URL này!
    api_key="YOUR_HOLYSHEEP_API_KEY",
    timeout=30
)

Verify connection
try:
    response = llm.invoke([{"role": "user", "content": "test"}])
    print("✅ Kết nối HolySheep AI thành công!")
except Exception as e:
    print(f"❌ Lỗi: {e}")
    print("Kiểm tra lại API key tại: https://www.holysheep.ai/register")

Kinh nghiệm thực chiến từ production

Qua 6 tháng vận hành hệ thống LangGraph với HolySheep AI, đây là những bài học mà mình muốn chia sẻ:

• Always set max_iterations: Không bao giờ tin tưởng 100% vào logic điều kiện. Chúng tôi đã có lần production bị infinite loop tốn 10,000 requests chỉ vì một edge case.
• Log everything: Thêm logging vào mọi conditional edge. Khi có bug, bạn sẽ thank mình vì đã làm điều này.
• Test với production-like data: Unit test với happy path là không đủ. Chúng tôi build được bộ test với 100 edge cases trước khi deploy.
• Monitor latency: HolySheep AI cho chúng tôi latency trung bình 47ms, nhưng chúng tôi vẫn set alert nếu >200ms.
• Cost tracking: Set up budget alert. Với $8/MTok cho GPT-

  Tài nguyên liên quan

  • 📚 Hướng dẫn AI API
  • 💰 Xem giá
  • 📖 Tài liệu nhà phát triển
  • 🚀 Đăng ký miễn phí

  Bài viết liên quan

  • Hướng dẫn toàn tập: Xây dựng báo cáo tài chính tự động với A
  • AI套利策略：跨交易所价差检测与自动交易的完整指南
  • LangGraph 监控与调试：Agent执行轨迹可视化分析