LangGraph là gì? Framework orchestration agent AI có state

LangGraph là gì? Framework orchestration AI mã nguồn mở từ LangChain cho phép xây dựng agent AI dạng graph — nơi bạn kết hợp deterministic code và LLM-driven steps trong cùng một workflow có state, persistence và fault tolerance.

Khác với simple prompt chain, LangGraph biến agent thành state machine có kiểm soát tuyệt đối: lưu checkpoint, rollback, human-in-the-loop, memory dài hạn và multi-agent coordination.

Bài viết giải thích LangGraph hoạt động thế nào, so sánh với LangChain Agents, AutoGen/CrewAI, và hướng dẫn build agent cơ bản.

LangGraph hoạt động như thế nào?

LangGraph lấy ý tưởng từ Pregel / Apache Beam và áp dụng vào LLM. Thay vì chain tuyến tính, bạn định nghĩa:

• Nodes: Hàm xử lý (LLM call, tool execution, human review).
• Edges: Luồng điều hướng — conditional routing dựa trên state.
• State: Object chứa toàn bộ context (messages, metadata, kết quả tool).

Cơ chế execution:

1. Graph nhận State đầu vào → node đầu tiên chạy.
2. Mỗi node đọc state, xử lý, ghi lại state mới.
3. Edge quyết định node tiếp theo dựa trên condition (ví dụ: tool trả lỗi → human review node).
4. Khi gặp END, graph trả final state.

Điểm mạnh nhất: Durable execution. LangGraph lưu checkpoint sau mỗi node → agent có thể dừng, resume, rollback, hoặc bị interrupt (human-in-the-loop) mà mất state.

LangGraph vs LangChain Agents vs AutoGen

Chọn LangGraph khi bạn cần production-grade agent với fault tolerance, audit trail, và kiểm soát fine-grained. Chọn LangChain Agents khi cần prototype nhanh. Chọn AutoGen khi muốn multi-agent tự động thảo luận.

Build agent cơ bản với LangGraph

Cài đặt:

pip install langgraph langchain-openai

Định nghĩa graph:

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 1. Định nghĩa state
class AgentState(TypedDict):
    messages: Annotated[list, "messages"]
    next: str

# 2. Tool
@tool
def search_web(query: str) -> str:
    """Search web for current information."""
    return f"Results for: {query}"

tools = [search_web]
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# 3. Nodes
def agent(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

def should_continue(state: AgentState):
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tools"
    return END

# 4. Build graph
builder = StateGraph(AgentState)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
builder.add_edge("tools", "agent")

graph = builder.compile()

# 5. Run
result = graph.invoke({"messages": [("user", "Tìm giá ETH hôm nay")]})
print(result["messages"][-1].content)

Thêm persistence với SQLite:

from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver("agent_memory.db")
graph = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "user-123"}}
result = graph.invoke({"messages": [...]}, config)

# Resume sau 1 giờ:
result2 = graph.invoke(None, config)

Pattern production phổ biến

• Supervisor: Một LLM phân loại intent → route đến sub-agent chuyên biệt (coding, search, writing).
• Human-in-the-loop: Sau tool call risky (ví dụ: send email, delete record) → interrupt, chờ approve.
• Subgraphs: Mỗi domain (support, sales, engineering) là subgraph riêng, coordinator nối chúng.
• Time travel: Debug bằng cách rollback state về bất kỳ checkpoint nào.

Memory dài hạn và Cross-thread persistence

LangGraph hỗ trợ 2 loại memory:

• Short-term (thread-scoped): Lưu trong checkpointer, gắn với thread_id. Resume conversation bất kỳ lúc nào.
• Long-term (cross-thread): Dùng Store interface (Redis, Postgres, in-memory) để lưu facts, preferences, summaries dùng chung nhiều thread.

Ví dụ dùng InMemoryStore cho long-term memory:

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# Trong node, ghi fact:
def extract_facts(state, store):
    # Gọi LLM trích xuất facts từ conversation
    facts = {"user_preference": "likes technical details"}
    store.put(("user_123", "facts"), "preferences", facts)
    return {}

# Trong node khác, đọc facts:
def personalize(state, store):
    facts = store.get(("user_123", "facts"), "preferences")
    # Dùng facts để tùy chỉnh response
    return {}

Kết hợp short + long-term tạo ra agent nhớ vĩnh viễn profile user qua nhiều phiên.

LangGraph Studio và Debugging

LangGraph Studio là GUI trực quan hóa graph execution — xem state, transitions, timing và logs của từng node. Khác với LangSmith (observability platform), Studio chạy local và miễn phí, phù hợp cho development.

Tính năng chính:

• Visualize graph: Xem nodes, edges, và execution path màu sắc (green=success, red=error, yellow=interrupt).
• Inspect state: Click bất kỳ node để xem input/output state đầy đủ — debug LLM prompt, tool args, return values.
• Replay: Re-run graph từ bất kỳ checkpoint nào mà không cần gọi lại LLM đầu vào.
• Human-in-the-loop UI: Approve, edit, hoặc reject tool call/interrupt trực tiếp trên Studio — không cần thêm code.

Để bật Studio:

from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver("agent_memory.db")
graph = builder.compile(checkpointer=memory)

# LangGraph CLI sẽ auto-detect và expose Studio trên localhost
# langgraph dev

Tham khảo thêm

• LangGraph Docs — Tài liệu chính thức
• GitHub LangGraph — Source code & examples
• LangGraph Concepts — Giải thích kiến trúc state machine