Bài 2: Building Agents với LangGraph

Tổng quan

Bài này trang bị kỹ năng xây dựng agent bằng LangGraph: biểu diễn agent như một state graph, xây dựng tool execution loop, biết khi nào dừng và phục hồi khi tool lỗi, tách biệt lập kế hoạch khỏi hành động, giữ trạng thái qua nhiều lượt hội thoại, và chèn con người vào vòng lặp khi cần phê duyệt hành động nhạy cảm.

---

1. State, Nodes, Edges

LangGraph biểu diễn agent như một đồ thị có hướng: mỗi bước xử lý là một node, luồng di chuyển giữa các bước là edge, và toàn bộ dữ liệu được chia sẻ qua các node là state.

graph LR
    START((START)) --> AGENT[Agent Node]
    AGENT -->|có tool call| TOOLS[Tool Node]
    AGENT -->|không có tool call| END((END))
    TOOLS --> AGENT

State

State là một TypedDict (hoặc Pydantic model) mô tả dữ liệu được truyền và cập nhật xuyên suốt graph.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # tự động append thay vì overwrite
    step_count: int

Annotated[list, add_messages] là gì?

Mặc định, khi một node trả về {"messages": [...]}, LangGraph sẽ ghi đè giá trị cũ. Dùng add_messages (reducer function) để thay vào đó nối thêm message mới vào danh sách - cần thiết để giữ lịch sử hội thoại qua nhiều bước.

Nodes

Node là một hàm Python nhận state hiện tại, trả về phần state cần cập nhật.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

def agent_node(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response], "step_count": state["step_count"] + 1}

Edges

Edge nối 2 node theo một chiều cố định. Conditional edge chọn node tiếp theo dựa trên logic (thường dựa trên state hiện tại).

from langgraph.graph import StateGraph, START, END

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

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})

---

2. Tool Nodes & Tool Execution Loop

Định nghĩa tools

from langchain_core.tools import tool

@tool
def search_vietnamese_news(query: str) -> str:
    """Tìm kiếm tin tức tiếng Việt mới nhất theo từ khoá."""
    # Gọi API thực tế ở đây
    return f"Kết quả tìm kiếm cho '{query}': ..."

@tool
def get_stock_price(symbol: str) -> str:
    """Lấy giá cổ phiếu hiện tại theo mã chứng khoán (vd: VNM, FPT)."""
    return f"{symbol}: 68,500 VNĐ (+1.2%)"

tools = [search_vietnamese_news, get_stock_price]
llm_with_tools = llm.bind_tools(tools)

Tool Node - dùng ToolNode có sẵn

LangGraph cung cấp ToolNode để tự động thực thi tool calls mà không cần viết logic parse JSON thủ công (như ở Module I, Bài 1).

from langgraph.prebuilt import ToolNode

tool_node = ToolNode(tools)

def agent_node(state: AgentState) -> dict:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

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

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")  # Vòng lặp: sau khi chạy tool, quay lại agent

app = graph.compile()

Chạy thử

result = app.invoke({
    "messages": [{"role": "user", "content": "Giá cổ phiếu FPT hôm nay bao nhiêu?"}],
    "step_count": 0,
})
print(result["messages"][-1].content)

Đây chính là ReAct loop (Bài 1) được biểu diễn bằng graph: agent → tools → agent → ... → END cho đến khi model không còn gọi tool nữa.

Luôn giới hạn số vòng lặp

Nếu model liên tục gọi tool (vd: do tool trả về lỗi mà model không nhận ra), graph có thể lặp vô hạn. Thêm giới hạn qua recursion_limit khi invoke: app.invoke(input, config={"recursion_limit": 15}).

---

3. Loop Termination & Error Recovery

Agent biết dừng khi nào?

Đây là câu hỏi tưởng đơn giản nhưng dễ bị bỏ qua - agent không dừng đúng lúc vừa tốn chi phí vừa có thể gây hại nếu tiếp tục hành động ngoài ý muốn.

Chiến lược	Cách hoạt động	Khi nào dùng
Model tự quyết định	Model ngừng gọi tool, trả text - cách mặc định ở Bài 2 mục 2	Luôn có, là cơ chế cơ bản nhất
Max steps	Giới hạn cứng số vòng lặp (recursion_limit)	Luôn thêm làm lưới an toàn
Explicit stop tool	Agent phải gọi tool final_answer/submit để hoàn thành	Task cần xác nhận rõ ràng đã xong (vd: coding agent)
Goal verification	Một lời gọi LLM riêng kiểm tra xem mục tiêu đã đạt chưa	Task quan trọng, cần double-check trước khi trả kết quả
Cost/token budget	Dừng khi vượt ngân sách token/chi phí	Production, cần kiểm soát chi phí chặt

@tool
def final_answer(answer: str) -> str:
    """Nộp câu trả lời cuối cùng. Chỉ gọi tool này khi đã hoàn thành nhiệm vụ."""
    return answer

def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if not last_message.tool_calls:
        return END
    # Nếu tool call là final_answer, dừng graph dù về mặt kỹ thuật vẫn là tool call
    if last_message.tool_calls[0]["name"] == "final_answer":
        return END
    return "tools"

Đưa lỗi tool ngược lại cho model, đừng giấu

Khi tool thất bại, đừng raise exception làm crash graph - trả lỗi dưới dạng tool message để model tự điều chỉnh cách tiếp cận (đã nói ở Bài 4 mục 1, nhắc lại vì đây là phần cốt lõi của vòng lặp).

def tool_node_with_recovery(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    results = []
    for call in last_message.tool_calls:
        try:
            output = execute_tool(call)
        except Exception as e:
            # Lỗi trở thành 1 quan sát bình thường, không phải crash
            output = f"LỖI: {e}. Hãy thử cách tiếp cận khác."
        results.append({"role": "tool", "tool_call_id": call["id"], "content": str(output)})
    return {"messages": results}

Model hiện đại tự điều chỉnh tốt khi thấy lỗi

Đừng che giấu lỗi khỏi model bằng cách retry âm thầm hoặc trả về giá trị mặc định. Model đủ mạnh để đọc thông báo lỗi và tự thử hướng khác - đây chính là bản chất "tự sửa lỗi" của vòng lặp ReAct.

Phát hiện vòng lặp

Nếu agent lặp lại cùng 1 hành động nhiều lần liên tiếp, đó là dấu hiệu agent bị kẹt - cần can thiệp thay vì để recursion_limit âm thầm cắt ngang.

def detect_repetition(state: AgentState, window: int = 4) -> bool:
    """Kiểm tra xem N tool call gần nhất có bị lặp lại giống hệt không."""
    recent_calls = [
        m.tool_calls[0] for m in state["messages"][-window:]
        if hasattr(m, "tool_calls") and m.tool_calls
    ]
    if len(recent_calls) < window:
        return False
    return len(set(str(c) for c in recent_calls)) == 1  # Tất cả giống nhau

def agent_node_with_loop_check(state: AgentState) -> dict:
    if detect_repetition(state):
        return {"messages": [{
            "role": "system",
            "content": "Bạn đang lặp lại cùng 1 hành động. Hãy thử cách tiếp cận hoàn toàn khác.",
        }]}
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

---

4. Plan/Act Separation

Vấn đề: hành động trước khi hiểu rõ

Lỗi phổ biến nhất của agent là lao vào hành động trước khi hiểu đủ vấn đề - sửa nhầm chỗ, edit file không cần thiết, hoặc bỏ sót ngữ cảnh quan trọng. Cline (VS Code agent) giải quyết bằng cách tách rõ 2 chế độ.

Chế độ	Được phép làm	Không được phép
Plan Mode	Đọc file, tìm kiếm codebase, phân tích, tạo kế hoạch từng bước	Sửa file, chạy lệnh, tạo/xoá file
Act Mode	Toàn bộ tool, thực thi theo kế hoạch đã duyệt	-

Plan Mode buộc agent hiểu trước khi hành động; user review và duyệt kế hoạch trước khi chuyển sang Act Mode - nơi mọi hành động có side effect mới được phép chạy.

Implement bằng LangGraph: giới hạn tool theo phase

from typing import TypedDict, Annotated, Literal
from langgraph.graph.message import add_messages

class PlanActState(TypedDict):
    messages: Annotated[list, add_messages]
    mode: Literal["plan", "act"]
    plan: str

READ_ONLY_TOOLS = [read_file, search_codebase, list_directory]
ALL_TOOLS = READ_ONLY_TOOLS + [edit_file, run_command, delete_file]

def plan_node(state: PlanActState) -> dict:
    """Chỉ dùng tool đọc - không có side effect."""
    llm_plan = llm.bind_tools(READ_ONLY_TOOLS)
    response = llm_plan.invoke(
        [{"role": "system", "content": "Phân tích yêu cầu và tạo kế hoạch từng bước. "
                                        "KHÔNG sửa file hay chạy lệnh - chỉ đọc và phân tích."}]
        + state["messages"]
    )
    return {"messages": [response]}

def act_node(state: PlanActState) -> dict:
    """Toàn quyền tool, chỉ chạy sau khi user duyệt plan."""
    llm_act = llm.bind_tools(ALL_TOOLS)
    response = llm_act.invoke(
        [{"role": "system", "content": f"Thực thi theo kế hoạch đã duyệt:\n{state['plan']}"}]
        + state["messages"]
    )
    return {"messages": [response]}

def route_by_mode(state: PlanActState) -> str:
    return "plan" if state["mode"] == "plan" else "act"

Kết hợp với HITL (mục 6) - interrupt_before giữa 2 phase để user review kế hoạch trước khi cho phép chuyển sang Act Mode - đây là ứng dụng cụ thể của HITL cho toàn bộ 1 giai đoạn, không chỉ 1 tool call.

Tách Plan/Act là quyết định kiến trúc, không chỉ prompt

Chỉ dặn model "hãy lập kế hoạch trước" trong system prompt (implicit planning, Bài 1) không đủ tin cậy - model vẫn có thể lẫn lộn giữa suy nghĩ và hành động. Giới hạn tool khả dụng theo từng node như trên mới đảm bảo được ranh giới cứng giữa 2 giai đoạn.

---

5. Memory & State Persistence

Checkpointer - lưu state giữa các lượt gọi

Mặc định, state chỉ tồn tại trong một lần invoke(). Để agent "nhớ" hội thoại qua nhiều lượt (nhiều lần gọi invoke khác nhau), dùng checkpointer.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-123"}}

# Lượt 1
app.invoke({"messages": [{"role": "user", "content": "Tôi tên Minh."}]}, config=config)

# Lượt 2 - agent vẫn nhớ "Minh" nhờ cùng thread_id
result = app.invoke({"messages": [{"role": "user", "content": "Tôi tên gì?"}]}, config=config)
print(result["messages"][-1].content)  # "Bạn tên Minh."

thread_id phân biệt các cuộc hội thoại độc lập - mỗi user/session dùng một thread_id riêng.

Persistent checkpointer cho production

MemorySaver chỉ lưu trong RAM, mất khi restart. Production cần checkpointer bền vững:

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)

Checkpointer khác Memory ở Bài 3

Checkpointer lưu toàn bộ state của graph (để resume/replay execution) - đây là cơ chế kỹ thuật của LangGraph. Bài 3 (Agent Memory) nói về thiết kế những gì agent nên nhớ (short-term, long-term, episodic...) - một bài toán ở tầng cao hơn, thường xây trên nền checkpointer này.

---

6. Human-in-the-Loop

Với hành động nhạy cảm (xoá dữ liệu, gửi email, giao dịch tài chính), cần con người phê duyệt trước khi agent thực thi.

Interrupt trước khi gọi tool

app = graph.compile(checkpointer=checkpointer, interrupt_before=["tools"])

config = {"configurable": {"thread_id": "session-1"}}
result = app.invoke(
    {"messages": [{"role": "user", "content": "Huỷ đơn hàng #4521 giúp tôi."}]},
    config=config,
)
# Graph dừng lại TRƯỚC khi chạy tool node

# Kiểm tra tool call agent định thực hiện
pending_call = result["messages"][-1].tool_calls[0]
print(f"Agent muốn gọi: {pending_call['name']}({pending_call['args']})")

# Con người phê duyệt → resume graph bằng cách invoke với input=None
if user_approves:
    final_result = app.invoke(None, config=config)
else:
    # Từ chối: inject message huỷ bỏ rồi tiếp tục
    pass

Chỉnh sửa state trước khi resume

# Con người có thể sửa state trước khi cho graph tiếp tục
app.update_state(config, {"messages": [{"role": "user", "content": "Chỉ huỷ nếu chưa giao hàng."}]})
final_result = app.invoke(None, config=config)

HITL cần checkpointer

interrupt_before / interrupt_after chỉ hoạt động khi graph được compile với checkpointer - vì graph cần lưu state tại điểm dừng để resume sau này.

---

7. Hands-on: Vietnamese Personal Assistant

Kết hợp toàn bộ: tool calling + memory + HITL cho một trợ lý cá nhân tiếng Việt.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# ====== STATE ======
class AssistantState(TypedDict):
    messages: Annotated[list, add_messages]

# ====== TOOLS ======
@tool
def check_calendar(date: str) -> str:
    """Kiểm tra lịch làm việc theo ngày (định dạng YYYY-MM-DD)."""
    return f"Ngày {date}: 14h họp team, 16h30 gọi khách hàng."

@tool
def send_reminder(message: str, time: str) -> str:
    """Đặt lời nhắc. Hành động này cần con người phê duyệt trước khi gửi."""
    return f"Đã đặt lời nhắc '{message}' lúc {time}."

@tool
def search_restaurant(location: str, cuisine: str = "") -> str:
    """Tìm nhà hàng theo khu vực và loại món ăn."""
    return f"3 nhà hàng {cuisine} gần {location}: Quán A, Quán B, Quán C."

tools = [check_calendar, send_reminder, search_restaurant]

# ====== LLM ======
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
llm_with_tools = llm.bind_tools(tools)

SYSTEM_PROMPT = """Bạn là trợ lý cá nhân tiếng Việt, thân thiện và ngắn gọn.
Dùng tool khi cần thông tin lịch, đặt lời nhắc, hoặc tìm nhà hàng."""

# ====== NODES ======
def agent_node(state: AssistantState) -> dict:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def should_continue(state: AssistantState) -> str:
    last_message = state["messages"][-1]
    return "tools" if last_message.tool_calls else END

# ====== BUILD GRAPH ======
graph = StateGraph(AssistantState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

checkpointer = MemorySaver()
assistant = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["tools"],  # Phê duyệt trước khi gọi bất kỳ tool nào
)

# ====== CHẠY ======
config = {"configurable": {"thread_id": "user-minh"}}

result = assistant.invoke(
    {"messages": [{"role": "user", "content": "Chiều nay tôi có lịch gì không?"}]},
    config=config,
)

# Graph dừng trước tool call - hiển thị cho user xác nhận
last_msg = result["messages"][-1]
if last_msg.tool_calls:
    call = last_msg.tool_calls[0]
    print(f"Trợ lý muốn: {call['name']}({call['args']}) - Đồng ý? (y/n)")
    # Giả sử user đồng ý:
    final = assistant.invoke(None, config=config)
    print(final["messages"][-1].content)

---

Tóm tắt

graph TD
    A[Cần xây agent?] --> B[Định nghĩa State - TypedDict]
    B --> C[Viết Agent Node - gọi LLM + tools]
    C --> D[Viết Tool Node - ToolNode có sẵn]
    D --> E{Cần nhớ qua nhiều lượt?}
    E -->|Có| F[Thêm Checkpointer + thread_id]
    E -->|Không| G[Compile graph không checkpointer]
    F --> H{Có hành động nhạy cảm?}
    H -->|Có| I[interrupt_before=tools]
    H -->|Không| J[Chạy trực tiếp]

Thành phần	Vai trò	API chính
State	Dữ liệu chia sẻ giữa các node	TypedDict + Annotated[list, add_messages]
Node	Đơn vị xử lý (agent, tool)	Hàm nhận state, trả dict cập nhật
Conditional edge	Rẽ nhánh dựa trên state	add_conditional_edges
ToolNode	Tự động thực thi tool calls	langgraph.prebuilt.ToolNode
Loop termination	Model tự dừng, max steps, explicit stop tool, goal verification	recursion_limit, tool final_answer
Error recovery	Trả lỗi tool về dạng model đọc được, không raise	try/except trong tool node
Plan/Act separation	Giới hạn tool theo phase để tránh hành động trước khi hiểu rõ	Node riêng + tool set riêng cho mỗi mode
Checkpointer	Lưu state qua nhiều lượt invoke	MemorySaver, SqliteSaver
HITL	Dừng để con người phê duyệt	interrupt_before / interrupt_after