LangGraph mô tả agent như một đồ thị trạng thái: mỗi node là một hàm nhận state và trả về phần state cần cập nhật; edge quyết định node nào chạy tiếp. Khác với loop tự viết (tool use loop), LangGraph cho sẵn lưu trạng thái theo thread, dừng chờ người duyệt, resume, và stream từng bước.
Thành phần
| Thành phần | Vai trò | API |
|---|---|---|
| State | Dữ liệu chạy qua graph (TypedDict) | class State(TypedDict): ..., có sẵn MessagesState |
| Reducer | Cách gộp update vào state (ghi đè hay nối thêm) | Annotated[list, add_messages] |
| Node | Hàm state -> dict (gọi LLM, gọi tool, xử lý) |
builder.add_node("agent", call_model) |
| Edge | Chuyển cố định A → B | builder.add_edge("tools", "agent") |
| Conditional edge | Hàm router chọn node tiếp theo theo state | builder.add_conditional_edges("agent", route, {...}) |
| START / END | Điểm vào / kết thúc | from langgraph.graph import START, END |
| Checkpointer | Lưu state sau mỗi bước theo thread_id |
builder.compile(checkpointer=InMemorySaver()) |
| Interrupt | Dừng graph chờ input của người | interrupt(...), resume bằng Command(resume=...) |
Sơ đồ agent điển hình
flowchart LR S([START]) --> A["agent: gọi LLM"] A -->|"có tool_calls"| H{"cần duyệt?"} H -->|"read-only"| T[tools] H -->|"ghi cấu hình"| I["interrupt: chờ người duyệt"] I --> T T --> A A -->|"không có tool_calls"| E([END])
Ví dụ khung
from typing import Annotated, TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # reducer: nối thêm, không ghi đè
device: str
diff: str
def agent(state: State) -> dict:
... # gọi LLM, trả {"messages": [ai_message]}
def tools(state: State) -> dict:
... # chạy tool theo tool_calls của message cuối
def route(state: State) -> str:
last = state["messages"][-1]
return "tools" if getattr(last, "tool_calls", None) else END
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", tools)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route, ["tools", END])
builder.add_edge("tools", "agent")
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "ticket-4711"}}
graph.invoke({"messages": [("user", "Kiểm tra BGP trên pe1-hni")], "device": "pe1-hni"}, config)Checkpointer và thread
| Checkpointer | Dùng khi |
|---|---|
InMemorySaver (langgraph.checkpoint.memory) |
Dev, test; mất khi restart process |
SQLite (langgraph-checkpoint-sqlite) |
Một máy, cần giữ lịch sử |
Postgres (langgraph-checkpoint-postgres) |
Production, nhiều worker |
- Cùng
thread_id= tiếp tục hội thoại cũ (state đã lưu). Thread mới = bắt đầu sạch. graph.get_state(config)xem state hiện tại; checkpointer là điều kiện bắt buộc cho interrupt/resume.
Human-in-the-loop
from langgraph.types import Command, interrupt
def push_config(state: State) -> dict:
decision = interrupt({"question": "Đẩy config lên pe1-hni?", "diff": state["diff"]})
if decision != "approve":
return {"messages": [("system", "Người vận hành từ chối")]}
... # commit confirmed
# lần 1: graph dừng tại interrupt, trả payload trong kết quả
graph.invoke(inputs, config)
# sau khi người duyệt:
graph.invoke(Command(resume="approve"), config)Cách đơn giản hơn: builder.compile(checkpointer=..., interrupt_before=["tools"]) để dừng trước mọi lần chạy tool.
Khi nào dùng
- Dùng LangGraph khi cần: nhiều bước có rẽ nhánh, lưu/resume theo phiên, duyệt thủ công trước thao tác ghi, nhiều agent phối hợp.
- Không cần khi chỉ là hỏi-đáp + vài tool đọc: loop 40 dòng với SDK là đủ, ít dependency hơn.
- Nếu chỉ cần agent dùng tool trong terminal, cân nhắc viết MCP server cho Claude Code thay vì tự dựng agent.
Tham khảo: LangGraph docs