Bài 28: Xây Agent đầu tiên với LangGraph

Sau bài này bạn sẽ:

• ✅ Hiểu ReAct pattern hoạt động thế nào trong LangGraph
• ✅ Xây agent bằng StateGraph thủ công — hiểu rõ từng node và edge
• ✅ Dùng create_react_agent để tạo agent nhanh cho production
• ✅ Stream output từng node và từng token LLM
• ✅ Persist conversation history qua nhiều turn bằng MemorySaver
• ✅ Nhận diện và tránh các pitfall phổ biến khi dùng tool với LangGraph

Yêu cầu: Đã đọc bài 27 (State, Node, Edge). Đã có API key cho OpenAI hoặc model tương thích tool calling.

pip install langgraph>=0.2 langchain-openai>=0.2 langchain-core>=0.3

ReAct (Yao et al., arXiv:2210.03629) là pattern kết hợp Reasoning và Acting trong một vòng lặp:

1. Reason: LLM nhìn vào message history, quyết định cần gọi tool nào (hoặc không cần).
2. Act: Ứng dụng thực thi tool, thu được kết quả.
3. Observe: Kết quả tool được đưa trở lại message history dưới dạng ToolMessage.
4. Quay lại bước 1 cho đến khi LLM không còn yêu cầu gọi tool nào nữa.

Trong LangGraph, pattern này được biểu diễn bằng 2 node và 1 conditional edge:

START
  │
  ▼
[agent] ──── (có tool_calls?) ─── YES ──► [tools]
  ▲                                            │
  │                                            │
  └────────────────────────────────────────────┘
  │
 NO
  ▼
 END

• agent node: gọi LLM với toàn bộ message history.
• tools node: execute tất cả tool_calls trong message cuối, gắn ToolMessage vào state.
• Conditional edge từ agent: nếu message cuối có tool_calls → đi tools; không có → kết thúc.

Điểm khác biệt so với vòng lặp agent thủ công ở bài 25: LangGraph quản lý toàn bộ cycle bằng graph structure — bạn không cần viết while True, không cần tự append message vào list.

Dùng @tool decorator từ langchain_core.tools (đã học ở bài 25). Ở đây ta định nghĩa 2 tool đơn giản để demo:

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Lấy thời tiết hiện tại của một thành phố."""
    # Mock — production thay bằng API call thực
    return f"Trời nắng 30°C ở {city}"

@tool
def calculate(expression: str) -> str:
    """Tính biểu thức toán học đơn giản. Chỉ dùng cho demo."""
    try:
        # Cảnh báo: eval() trong production phải được sandbox
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Lỗi tính toán: {e}"

tools = [get_weather, calculate]

Hai điểm cần chú ý:

• Docstring là phần LLM đọc để quyết định có gọi tool không. Viết rõ ràng, cụ thể.
• calculate dùng eval() — chỉ hợp lệ cho demo/prototype. Production phải dùng thư viện như simpleeval hoặc sandbox riêng.

Cách này build graph từng bước để thấy rõ cơ chế bên trong:

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI

# Bind tools vào LLM — bước này bắt buộc để LLM biết tool schema
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)

# Định nghĩa state schema
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# Node 1: gọi LLM
def call_model(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Node 2: execute tools (prebuilt, tự parse tool_calls)
tool_node = ToolNode(tools)

# Conditional edge: tiếp tục gọi tool hay kết thúc?
def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

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

graph = builder.compile()

add_messages reducer

Annotated[list[BaseMessage], add_messages] khai báo rằng field messages không bị ghi đè mà được merge khi update. Khi node trả về {"messages": [response]}, LangGraph tự append response vào list hiện tại thay vì replace toàn bộ list.

ToolNode(tools)

Prebuilt node từ langgraph.prebuilt. Nhận state, đọc tool_calls trong message cuối, execute từng tool call song song (bằng asyncio nếu dùng async), đóng gói kết quả thành ToolMessage và trả về {"messages": [...]}.

Bạn không cần tự viết logic parse tool_calls hay tạo ToolMessage — ToolNode đã xử lý.

should_continue

Conditional routing function: nhận state, trả về string tên node tiếp theo (hoặc END). Phải match với dict mapping trong add_conditional_edges. Trả về sai tên → KeyError lúc runtime.

Cycle tools → agent

builder.add_edge("tools", "agent") tạo vòng lặp: sau khi execute tool, graph quay về agent node để LLM nhìn vào ToolMessage vừa được thêm và quyết định bước tiếp theo.

Luồng message trong state

Turn 1:
  state.messages = [HumanMessage("Thời tiết Hà Nội? Và tính 12 * 7")]

After agent node:
  state.messages = [..., AIMessage(tool_calls=[get_weather, calculate])]

After tools node:
  state.messages = [..., ToolMessage("Trời nắng 30°C ở Hà Nội"), ToolMessage("84")]

After agent node (lần 2):
  state.messages = [..., AIMessage("Thời tiết Hà Nội đang nắng 30°C. 12 × 7 = 84.")]
  → last_message.tool_calls == [] → should_continue trả END

result = graph.invoke({
    "messages": [HumanMessage("Thời tiết Hà Nội thế nào? Và tính 12 * 7 giúp tôi.")]
})

for msg in result["messages"]:
    msg.pretty_print()

Output mong đợi:

================================ Human Message =================================
Thời tiết Hà Nội thế nào? Và tính 12 * 7 giúp tôi.

================================== Ai Message ==================================
Tool Calls:
  get_weather (call_xxx)
   Call ID: call_xxx
   Args: {"city": "Hà Nội"}
  calculate (call_yyy)
   Call ID: call_yyy
   Args: {"expression": "12 * 7"}

================================= Tool Message =================================
Name: get_weather
Trời nắng 30°C ở Hà Nội

================================= Tool Message =================================
Name: calculate
84

================================== Ai Message ==================================
Thời tiết tại Hà Nội hiện đang nắng, nhiệt độ 30°C. Kết quả của 12 × 7 là 84.

LLM gọi cả 2 tool trong 1 lượt (parallel tool calls, hỗ trợ từ GPT-4 series và nhiều model hiện đại). ToolNode execute song song rồi trả về 2 ToolMessage cùng lúc.

create_react_agent là factory function trong langgraph.prebuilt. Nó build graph giống hệt Cách 1 nhưng bạn không cần viết boilerplate:

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    tools=[get_weather, calculate],
    prompt="Bạn là trợ lý hữu ích. Trả lời ngắn gọn bằng tiếng Việt.",
)

result = agent.invoke({
    "messages": [HumanMessage("Thời tiết Đà Nẵng?")]
})

Các parameter hay dùng của create_react_agent (LangGraph 0.2.x):

• model: LLM đã hoặc chưa bind tools — factory tự gọi bind_tools nếu chưa.
• tools: list tool hoặc ToolNode có sẵn.
• prompt: system message (string hoặc SystemMessage object).
• state_schema: custom TypedDict nếu cần thêm field vào state ngoài messages.
• checkpointer: gắn persistence — xem mục 9.
• interrupt_before / interrupt_after: danh sách node name để dừng và chờ human review — dùng trong bài 30.
• pre_model_hook / post_model_hook: callable chạy trước/sau khi invoke LLM — dùng để trim message history khi quá dài.

Khi nào dùng Cách 1 vs Cách 2:

• Cách 1 (StateGraph thủ công): khi cần custom node logic phức tạp, custom state schema, thêm node ngoài agent+tools, hoặc cần kiểm soát chính xác flow.
• Cách 2 (create_react_agent): khi pattern đủ là agent gọi tool → lấy kết quả → trả lời. Phần lớn use case production rơi vào đây.

Stream từng node update: mỗi chunk là dict {node_name: state_update}.

for chunk in agent.stream({"messages": [HumanMessage("Thời tiết Hà Nội?")]}):
    # chunk = {"agent": {"messages": [...]}} hoặc {"tools": {"messages": [...]}}
    node_name, update = next(iter(chunk.items()))
    print(f"[{node_name}]", update["messages"][-1].pretty_repr())

Stream token LLM (cần version="v2"):

import asyncio

async def stream_tokens():
    async for event in agent.astream_events(
        {"messages": [HumanMessage("Tính 99 * 88")]},
        version="v2",
    ):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                print(content, end="", flush=True)

asyncio.run(stream_tokens())

Event types trong astream_events hay dùng:

• on_chat_model_stream: token LLM sinh ra từng phần.
• on_tool_start: bắt đầu execute một tool call cụ thể.
• on_tool_end: tool call hoàn thành, có output.
• on_chain_end: graph kết thúc.

Lưu ý: khi graph có conditional edge, một số chunk stream sẽ có key __interrupt__ nếu graph bị pause (xem bài 30). Bỏ qua các chunk đó khi chỉ cần output bình thường.

Mặc định mỗi lần invoke là một session độc lập — agent không nhớ turn trước. Để persist history qua nhiều turn, gắn checkpointer:

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o-mini"),
    tools=[get_weather, calculate],
    checkpointer=memory,
)

# thread_id phân biệt các conversation độc lập
config = {"configurable": {"thread_id": "user-123"}}

# Turn 1
agent.invoke(
    {"messages": [HumanMessage("Thời tiết Hà Nội?")]},
    config=config,
)

# Turn 2 — agent nhớ turn trước
result = agent.invoke(
    {"messages": [HumanMessage("Tôi vừa hỏi về thành phố nào?")]},
    config=config,
)
print(result["messages"][-1].content)
# → "Bạn vừa hỏi về Hà Nội."

Cơ chế hoạt động: trước mỗi invoke, LangGraph load state từ checkpointer theo thread_id. Sau khi hoàn thành, save state mới lại. Field messages được merge qua add_messages reducer nên history giữ nguyên qua các turn.

Checkpointer options

• MemorySaver: in-memory, mất khi process restart. Dùng cho dev/test.
• SqliteSaver (langgraph-checkpoint-sqlite): lưu vào SQLite file. Dùng cho single-server production đơn giản.
• PostgresSaver / AsyncPostgresSaver (langgraph-checkpoint-postgres): lưu vào PostgreSQL. Dùng cho multi-instance production.

# SQLite checkpointer (cần pip install langgraph-checkpoint-sqlite)
from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    agent = create_react_agent(llm, tools, checkpointer=checkpointer)
    # ... invoke ...

Quản lý history dài

Khi history tích lũy qua nhiều turn, context window sẽ đầy. Giải pháp với LangGraph 0.2+: dùng pre_model_hook để trim message list trước khi gọi LLM:

from langchain_core.messages import trim_messages

def trim_hook(state: dict) -> dict:
    trimmed = trim_messages(
        state["messages"],
        max_tokens=4096,
        strategy="last",  # giữ N token cuối
        token_counter=ChatOpenAI(model="gpt-4o-mini"),
        include_system=True,
    )
    return {"messages": trimmed}

agent = create_react_agent(
    model=llm,
    tools=tools,
    pre_model_hook=trim_hook,
    checkpointer=memory,
)

LangGraph có cơ chế bảo vệ tránh vòng lặp vô tận: recursion limit, mặc định là 25 step. Sau 25 lần graph traverse node, GraphRecursionError được raise.

from langgraph.errors import GraphRecursionError

try:
    result = graph.invoke(
        {"messages": [HumanMessage("...")]},
        config={"recursion_limit": 50},  # override mặc định
    )
except GraphRecursionError as e:
    print(f"Graph vượt giới hạn step: {e}")

Lưu ý về cách đếm: mỗi lần một node được execute tính là 1 step. Với ReAct agent có 3 tool calls, số step tối thiểu là:

• 1 (agent node lần 1) + 1 (tools node) + 1 (agent node lần 2) = 3 step.
• Thêm START edge không tính vào step count.

Khi gặp GraphRecursionError không phải do bug: tăng recursion_limit hoặc review lại prompt để LLM không gọi tool lặp không cần thiết.

LangGraph cung cấp một số cách xem cấu trúc graph:

ASCII (không cần dependency):

print(agent.get_graph().draw_ascii())

        +-----------+
        | __start__ |
        +-----------+
              *
              *
              *
          +-------+
          | agent |
          +-------+
         *         .
        *           .
       *             .
+-------+          +---------+
| tools |          | __end__ |
+-------+          +---------+
    *
    *
    *
+-------+
| agent |  ← (cycle back)
+-------+

PNG (cần pygraphviz hoặc Mermaid):

# Cần pip install pygraphviz
from IPython.display import Image
Image(agent.get_graph().draw_png())

# Mermaid syntax (không cần dependency nặng)
print(agent.get_graph().draw_mermaid())

Visualize hữu ích khi debug graph phức tạp nhiều node — kiểm tra edge routing có đúng ý định không.

Pattern chuẩn cho chatbot multi-turn với checkpointer:

def chat(agent, thread_id: str, user_input: str) -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        {"messages": [HumanMessage(user_input)]},
        config=config,
    )
    return result["messages"][-1].content

# Sử dụng
agent_with_memory = create_react_agent(
    model=ChatOpenAI(model="gpt-4o-mini"),
    tools=[get_weather, calculate],
    checkpointer=MemorySaver(),
)

print(chat(agent_with_memory, "session-1", "Thời tiết Hà Nội?"))
print(chat(agent_with_memory, "session-1", "Còn Hồ Chí Minh?"))
print(chat(agent_with_memory, "session-1", "So sánh hai thành phố đó?"))
# Agent nhớ cả 2 kết quả weather từ turn trước

Mỗi thread_id là một conversation độc lập — trạng thái không bị trộn lẫn giữa các user. Dùng user ID hoặc session ID làm thread_id trong production.

Kiểm tra state hiện tại của một thread mà không invoke:

state = agent_with_memory.get_state({"configurable": {"thread_id": "session-1"}})
print(f"Số message hiện tại: {len(state.values['messages'])}")

1. Quên bind_tools vào LLM

# SAI — LLM không biết tools tồn tại
llm = ChatOpenAI(model="gpt-4o-mini")

# ĐÚNG
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)

# Hoặc để create_react_agent tự bind (cũng ổn)
agent = create_react_agent(model=ChatOpenAI(model="gpt-4o-mini"), tools=tools)

Triệu chứng: LLM trả lời text mà không gọi tool, dù câu hỏi rõ ràng cần tool.

2. should_continue return string không khớp mapping

# SAI — "tool" (thiếu s) không có trong dict
builder.add_conditional_edges("agent", should_continue, {
    "tool": "tools",  # KeyError lúc runtime
    END: END,
})

# ĐÚNG
def should_continue(state):
    return "tools" if state["messages"][-1].tool_calls else END

builder.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    END: END,
})

3. Tool exception không được xử lý

# Không xử lý lỗi — khi tool fail, graph raise exception và dừng
@tool
def risky_tool(query: str) -> str:
    """Gọi API bên ngoài."""
    return requests.get(f"...{query}").json()["result"]  # có thể raise

# Tốt hơn — wrap try/except
@tool
def risky_tool(query: str) -> str:
    """Gọi API bên ngoài."""
    try:
        return requests.get(f"...{query}").json()["result"]
    except Exception as e:
        return f"Không thể lấy dữ liệu: {e}"

Hoặc dùng handle_tool_error=True khi tạo ToolNode: ToolNode(tools, handle_tool_error=True) — sẽ catch exception và trả về error message thay vì raise.

4. LLM liên tục gọi tool không dừng

Nguyên nhân: docstring tool không rõ ràng, LLM không đủ thông tin để tổng hợp câu trả lời, hoặc tool luôn trả về kết quả kích thích LLM gọi tiếp. Xử lý:

• Set recursion_limit hợp lý (10-20 với task đơn giản).
• Cải thiện docstring tool và system prompt.
• Monitor số lượng tool calls mỗi run qua stream events.

5. Nhầm giữa messages bị ghi đè và được merge

# State KHÔNG có add_messages reducer
class BadState(TypedDict):
    messages: list[BaseMessage]  # sẽ bị ghi đè mỗi node update

# State ĐÚNG
class GoodState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]  # merge

Triệu chứng: sau mỗi node, chỉ còn message cuối cùng trong list, mất toàn bộ history.

Bài tập 1: File assistant (không có side-effect write)

Định nghĩa 2 tool:

• list_files(directory: str) -> str: gọi os.listdir, trả chuỗi tên file.
• read_file(path: str) -> str: đọc nội dung file text, trả 500 ký tự đầu.

Build agent và hỏi: "Trong thư mục /tmp có những file gì? Đọc nội dung file đầu tiên."

Bài tập 2: Multi-turn với checkpointer

Tạo agent MemorySaver. Chạy 3 turn liên tiếp:

1. "Thời tiết Hà Nội?"
2. "Còn Đà Nẵng?"
3. "Thành phố nào nóng hơn theo những gì tôi vừa hỏi?"

Turn 3 phải cho ra câu trả lời đúng dựa vào context 2 turn trước.

Bài tập 3: Custom state field

Thêm field tool_call_count: int vào state schema. Tăng counter mỗi khi tools node chạy. Sau khi agent hoàn thành, in ra tổng số tool calls đã thực hiện.

Ghi chú về tool có side-effect

Tool ghi file (write_file), gửi email, thực thi code có ảnh hưởng thật lên hệ thống. Với những tool này, nên dừng graph để người dùng xác nhận trước khi execute — pattern này được trình bày trong bài 30 (human-in-the-loop).

ReAct agent trong LangGraph cần 2 node (agent, tools) và 1 conditional edge. ToolNode từ prebuilt tự execute tool_calls và tạo ToolMessage. add_messages reducer đảm bảo history được merge thay vì ghi đè.

create_react_agent build graph tương đương nhưng ít boilerplate hơn, phù hợp cho production. Gắn MemorySaver để persist state qua nhiều turn; dùng thread_id để tách biệt conversation của từng user.

Các điểm cần nhớ: bind_tools bắt buộc, should_continue phải trả string khớp mapping, tool nên wrap try/except, recursion_limit mặc định 25 step.

Bài 29: Conditional edge — branching logic