Bài 30: Human-in-the-loop — pause để con người duyệt

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

• Hiểu tại sao cần dừng graph và chờ người dùng xác nhận khi agent có tool side-effect.
• Thiết lập interrupt_before và checkpointer để pause/resume graph.
• Inspect state tại điểm pause, approve, modify args, hoặc reject tool call.
• Dùng interrupt() (LangGraph 0.2+) để kiểm soát interrupt theo điều kiện trong node.
• Biết khi nào dùng MemorySaver, SqliteSaver, PostgresSaver.
• Debug agent bằng time-travel — quay lại checkpoint cũ và thử lại với input khác.

Agent trong LangGraph thực thi tool call do LLM sinh ra. Với những tool chỉ đọc dữ liệu (search, calculate, fetch), nếu LLM hallucinate args thì cùng lắm kết quả sai — agent có thể thử lại. Với những tool có side-effect không thể hoàn tác, sai một lần là hậu quả thật:

• send_email(to, body) — email đã gửi, không thu hồi được.
• create_github_issue(repo, title, body) — issue đã tạo, lộ ra public.
• charge_card(amount) — tiền đã trừ.
• write_db(table, data) — bản ghi đã ghi vào production.
• delete_files(path) — file đã xóa.

Không thể tin tuyệt đối rằng LLM sẽ luôn sinh đúng args. Vì vậy với nhóm tool này, cần bước con người xem lại tool call trước khi thực sự execute. Đó là HITL (Human-in-the-loop).

Khi nào không cần HITL

Tool chỉ đọc (read-only), kết quả sai không gây hậu quả ngoài thế giới thực, hoặc agent đã được đánh giá kỹ ở môi trường test với tỷ lệ lỗi chấp nhận được — có thể bỏ qua HITL để giữ throughput cao. HITL tốt nhất áp dụng có chọn lọc, không phải mọi tool call.

LangGraph HITL dựa trên hai thành phần kết hợp:

1. Checkpointer: persist toàn bộ graph state (messages, intermediate values, metadata) sau mỗi step vào storage. Đây là điều kiện bắt buộc — nếu không có checkpointer, graph không thể pause và resume.
2. Interrupt: đánh dấu node cần dừng. Khi graph đến node đó, thay vì chạy tiếp, graph lưu state vào checkpointer và trả control về caller. Caller (web app, CLI) hiển thị thông tin cho user, nhận input, rồi invoke lại để graph chạy tiếp từ đúng điểm đó.

Graph không bị terminate — nó chỉ pause. State đầy đủ được lưu, bao gồm cả node tiếp theo cần chạy. Caller có thể chờ vài giây hoặc vài giờ, khi nào cũng có thể resume.

Luồng tổng quát:

1. invoke(input, config)
   → graph chạy đến trước node "tools" → pause, lưu state
   → caller nhận state (chứa tool_call LLM muốn execute)

2. [User xem tool_call, quyết định]
   → Approve: invoke(None, config)     → resume, execute tool
   → Modify:  update_state(config, ...) rồi invoke(None, config)
   → Reject:  update_state(fake ToolMessage) rồi invoke(None, config)

thread_id trong config là định danh conversation. Mỗi invoke dùng cùng thread_id thì LangGraph biết đây là cùng một luồng đang pause, không phải conversation mới.

pip install langgraph langchain-openai

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Gửi email tới địa chỉ chỉ định."""
    # Trong thực tế: gọi SMTP / SendGrid API
    print(f"[EMAIL SENT] To={to}, Subject={subject}")
    return f"Email đã gửi tới {to}"

@tool
def write_file(path: str, content: str) -> str:
    """Ghi nội dung vào file trên server."""
    # Trong thực tế: open(path, "w").write(content)
    print(f"[FILE WRITTEN] path={path}")
    return f"Đã ghi vào {path}"

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

agent = create_react_agent(
    model=llm,
    tools=[send_email, write_file],
    checkpointer=memory,
    interrupt_before=["tools"],  # pause trước node "tools"
)

config = {"configurable": {"thread_id": "thread-1"}}

interrupt_before=["tools"]: LangGraph sẽ dừng ngay trước khi node "tools" chạy. Tại điểm này, LLM đã sinh ra tool_calls bên trong AIMessage nhưng tool chưa được gọi — đây là thời điểm lý tưởng để user xem xét.

Cũng có interrupt_after=["agent"] — dừng sau khi node agent chạy xong (tức là sau khi LLM generate response, nhưng trước khi tools execute). Thực tế với create_react_agent, interrupt_before=["tools"] và interrupt_after=["agent"] dừng tại cùng thời điểm (giữa LLM output và tool execution), nên interrupt_before=["tools"] là lựa chọn rõ nghĩa hơn.

Bước 1: User gửi message — graph pause trước tools

from langchain_core.messages import HumanMessage

result = agent.invoke(
    {"messages": [HumanMessage(content="Gửi email cho [email protected] nói tôi nghỉ ốm hôm nay.")]},
    config=config,
)
# graph chạy node "agent" (LLM quyết định tool_call)
# → gặp interrupt_before=["tools"] → pause
# result là dict với messages, trong đó message cuối là AIMessage có tool_calls

Bước 2: Inspect state — xem tool sắp được gọi

state = agent.get_state(config)

print("Node tiếp theo:", state.next)
# → ('tools',)  — xác nhận graph đang dừng trước node tools

last_msg = state.values["messages"][-1]
print("Tool calls:", last_msg.tool_calls)
# → [{'name': 'send_email',
#      'args': {'to': '[email protected]',
#               'subject': 'Nghỉ ốm',
#               'body': 'Em xin nghỉ ốm hôm nay.'},
#      'id': 'call_abc123',
#      'type': 'tool_call'}]

Bước 3: Approve — invoke với None để resume

final = agent.invoke(None, config=config)
# → graph tiếp tục từ node "tools": execute send_email(...)
# → agent xử lý ToolMessage, generate final response
# → graph kết thúc

print(final["messages"][-1].content)
# → "Đã gửi email cho [email protected] thông báo bạn nghỉ ốm hôm nay."

Khi invoke với None, LangGraph load state từ checkpointer theo thread_id, biết node tiếp theo là "tools", và chạy tiếp từ đó. Input None ở đây không phải "không có input" mà là tín hiệu "resume từ interrupt hiện tại, không thêm message mới".

User có thể sửa args của tool_call trước khi execute — ví dụ sửa nội dung email, hoặc đổi path của file được ghi.

from langchain_core.messages import AIMessage

# Lấy state hiện tại
state = agent.get_state(config)
last_msg = state.values["messages"][-1]

# Tạo AIMessage mới với args đã sửa
modified_msg = AIMessage(
    content=last_msg.content,
    tool_calls=[{
        "name": "send_email",
        "args": {
            "to": "[email protected]",
            "subject": "Xin nghỉ ốm",
            "body": "Kính gửi anh/chị, em xin phép nghỉ ốm hôm nay. Trân trọng.",
        },
        "id": last_msg.tool_calls[0]["id"],  # giữ nguyên id
        "type": "tool_call",
    }],
)

# Cập nhật state — thay thế message cuối bằng message đã modify
agent.update_state(config, {"messages": [modified_msg]})

# Resume — graph sẽ execute tool với args mới
agent.invoke(None, config=config)

update_state thêm message vào danh sách (reducer add_messages trong state schema của create_react_agent). Khi truyền một AIMessage có cùng id với message cuối, reducer sẽ replace message cũ thay vì append — tránh duplicate.

Nếu state schema dùng custom reducer khác, hành vi có thể khác. Kiểm tra reducer của state trước khi dùng pattern này.

User từ chối thực hiện tool call. Không thể chỉ bỏ qua — graph đang chờ ToolMessage để biết kết quả của tool_call. Giải pháp: inject một ToolMessage giả với nội dung "bị từ chối", rồi resume. LLM sẽ thấy tool đã "chạy" với kết quả "bị cancel" và tự xử lý tiếp.

from langchain_core.messages import ToolMessage

state = agent.get_state(config)
last_msg = state.values["messages"][-1]

# ToolMessage giả — báo cho LLM biết tool không được execute
reject_msg = ToolMessage(
    content="Người dùng từ chối thực hiện tool này.",
    tool_call_id=last_msg.tool_calls[0]["id"],
)

# update_state với as_node="tools" để LangGraph biết message này "đến từ" node tools
agent.update_state(config, {"messages": [reject_msg]}, as_node="tools")

# Resume — graph skip node tools thực (vì đã có ToolMessage), chạy tiếp
final = agent.invoke(None, config=config)
print(final["messages"][-1].content)
# → LLM nhận ToolMessage với nội dung "từ chối", có thể trả lời:
#    "Hành động đã bị hủy theo yêu cầu của bạn."

as_node="tools": bắt buộc khi inject ToolMessage ở bên ngoài node. Nếu thiếu, LangGraph không biết update này "đến từ" node nào — graph state sẽ không tiến đúng theo edge, có thể gây lỗi hoặc vòng lặp vô tận.

Trường hợp LLM sinh nhiều tool_call trong một AIMessage, cần inject một ToolMessage cho mỗi tool_call_id:

# Khi last_msg.tool_calls có nhiều phần tử
reject_msgs = [
    ToolMessage(
        content="Người dùng từ chối.",
        tool_call_id=tc["id"],
    )
    for tc in last_msg.tool_calls
]
agent.update_state(config, {"messages": reject_msgs}, as_node="tools")
agent.invoke(None, config=config)

interrupt_before=["tools"] dừng tại node theo tên — không phân biệt được tool nào đang được gọi. Nếu agent dùng cả tool an toàn (search) và tool nguy hiểm (send_email), cả hai đều bị pause.

LangGraph 0.2+ thêm hàm interrupt(payload) có thể gọi từ bên trong node, cho phép kiểm soát interrupt theo logic:

from langgraph.types import interrupt
from langchain_core.messages import ToolMessage
from langgraph.graph import StateGraph, MessagesState, START, END

# Danh sách tool an toàn — không cần HITL
SAFE_TOOLS = {"search", "calculate", "fetch_weather"}

def review_and_execute_tools(state: MessagesState):
    """
    Node thay thế cho node tools mặc định.
    Với tool an toàn: execute ngay.
    Với tool side-effect: interrupt và chờ user approve/reject.
    """
    last_msg = state["messages"][-1]
    results = []

    for tc in last_msg.tool_calls:
        if tc["name"] in SAFE_TOOLS:
            # Auto-execute không cần HITL
            result = execute_tool_call(tc)
            results.append(ToolMessage(content=result, tool_call_id=tc["id"]))
        else:
            # Side-effect tool → interrupt với payload để frontend hiển thị
            approval = interrupt({
                "tool_name": tc["name"],
                "tool_args": tc["args"],
                "message": f"Tool '{tc['name']}' cần xác nhận trước khi chạy.",
            })
            # Sau khi resume, approval là giá trị user truyền vào Command(resume=...)
            if approval == "reject":
                results.append(ToolMessage(
                    content="Đã hủy theo yêu cầu người dùng.",
                    tool_call_id=tc["id"],
                ))
            else:
                result = execute_tool_call(tc)
                results.append(ToolMessage(content=result, tool_call_id=tc["id"]))

    return {"messages": results}

interrupt(payload) không trả về ngay — nó raise một exception nội bộ để LangGraph bắt và pause graph. Khi resume, hàm mới thực sự trả về giá trị từ Command(resume=...). Code sau interrupt() chỉ chạy sau khi resume.

So sánh interrupt_before và interrupt():

Khi graph pause do interrupt(), để resume cần dùng Command(resume=...) thay vì None:

from langgraph.types import Command

# User chọn approve
final = agent.invoke(Command(resume="approve"), config=config)

# User chọn reject
final = agent.invoke(Command(resume="reject"), config=config)

Giá trị truyền vào resume= là bất kỳ — string, dict, bool. Đây chính là giá trị mà interrupt(payload) trả về trong node sau khi resume. Quy ước cụ thể (dùng "approve"/"reject" hay {"action": "approve", "modified_args": {...}}) do code của bạn quy định.

Ví dụ truyền args đã modify từ user:

user_decision = {
    "action": "approve",
    "modified_args": {
        "to": "[email protected]",
        "subject": "Xin nghỉ",
        "body": "Kính gửi, em xin nghỉ hôm nay. Cảm ơn.",
    }
}
final = agent.invoke(Command(resume=user_decision), config=config)

Bên trong node, đọc giá trị trả về từ interrupt():

decision = interrupt({"tool_name": tc["name"], "args": tc["args"]})
# decision == user_decision dict ở trên

if decision["action"] == "approve":
    # Dùng args đã sửa thay vì args gốc từ LLM
    tc_to_run = {**tc, "args": decision.get("modified_args", tc["args"])}
    result = execute_tool_call(tc_to_run)

HITL thường được tích hợp trong web app theo mô hình async: frontend poll hoặc nhận webhook khi graph pause, hiển thị tool call info, nhận action của user rồi gọi endpoint resume.

# backend/main.py (FastAPI)
from fastapi import FastAPI
from pydantic import BaseModel
from langgraph.types import Command
from langchain_core.messages import HumanMessage

app = FastAPI()

class StartRequest(BaseModel):
    thread_id: str
    message: str

class ResumeRequest(BaseModel):
    thread_id: str
    action: str          # "approve" | "reject"
    modified_args: dict | None = None

@app.post("/start")
async def start_task(req: StartRequest):
    config = {"configurable": {"thread_id": req.thread_id}}
    result = agent.invoke(
        {"messages": [HumanMessage(content=req.message)]},
        config=config,
    )
    state = agent.get_state(config)

    if state.next == ("tools",):
        # Graph đang pause chờ HITL
        last_msg = state.values["messages"][-1]
        return {
            "status": "pending_approval",
            "tool_calls": last_msg.tool_calls,
        }
    # Graph đã hoàn thành không cần HITL
    return {
        "status": "completed",
        "reply": result["messages"][-1].content,
    }

@app.post("/resume")
async def resume_task(req: ResumeRequest):
    config = {"configurable": {"thread_id": req.thread_id}}

    if req.action == "approve":
        final = agent.invoke(None, config=config)
    elif req.action == "reject":
        state = agent.get_state(config)
        last_msg = state.values["messages"][-1]
        from langchain_core.messages import ToolMessage
        reject_msgs = [
            ToolMessage(content="Hủy theo yêu cầu.", tool_call_id=tc["id"])
            for tc in last_msg.tool_calls
        ]
        agent.update_state(config, {"messages": reject_msgs}, as_node="tools")
        final = agent.invoke(None, config=config)
    else:
        return {"error": "Unknown action"}

    return {
        "status": "completed",
        "reply": final["messages"][-1].content,
    }

Frontend chỉ cần hai API: POST /start để submit task và nhận pending_approval (hoặc completed ngay nếu không cần HITL), và POST /resume để truyền quyết định của user.

Trong production, cần checkpointer persist (Postgres/Redis) để state tồn tại qua nhiều worker và server restart. Phần 12 đề cập chi tiết.

Vì checkpointer lưu state sau mỗi step, bạn có thể duyệt toàn bộ lịch sử chạy và "replay" từ bất kỳ điểm nào với input khác. Tính năng này đặc biệt hữu ích khi debug agent đang ra quyết định sai.

# Lấy danh sách tất cả checkpoint theo thứ tự từ mới đến cũ
history = list(agent.get_state_history(config))

for i, checkpoint in enumerate(history):
    print(f"[{i}] next={checkpoint.next}, messages={len(checkpoint.values['messages'])}")
# [0] next=()          ← state cuối (graph đã kết thúc)
# [1] next=('agent',)  ← sau tools, trước agent generate final
# [2] next=('tools',)  ← điểm pause HITL
# [3] next=('agent',)  ← ngay sau khi user submit task
# ...

# Quay lại checkpoint khi graph đang chờ HITL (next=('tools',))
pause_checkpoint = history[2]

# Invoke từ checkpoint đó với input mới (hoặc None để resume y chang)
# Lưu ý: dùng pause_checkpoint.config, không phải config ban đầu
result = agent.invoke(
    {"messages": [HumanMessage(content="Thay vào đó hãy tạo một draft và cho tôi xem trước.")]},
    config=pause_checkpoint.config,
)

Khi invoke từ pause_checkpoint.config, LangGraph tạo một branch mới từ điểm đó. Thread gốc (thread-1) vẫn còn nguyên. Nếu muốn override thread gốc, dùng config ban đầu thay vì pause_checkpoint.config.

Time-travel chủ yếu dùng trong dev/debug, không phải production flow. Trong production, cần cẩn thận với vấn đề side-effect đã xảy ra trước điểm replay — ví dụ nếu email đã gửi ở step 5 và bạn replay từ step 3, email sẽ gửi lại.

HITL trong production yêu cầu state persist qua restart và chia sẻ được giữa nhiều server instance. MemorySaver không đáp ứng được cả hai yêu cầu này.

SqliteSaver

pip install langgraph-checkpoint-sqlite

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("./checkpoints.db") as saver:
    agent = create_react_agent(
        model=llm,
        tools=[send_email, write_file],
        checkpointer=saver,
        interrupt_before=["tools"],
    )
    # agent.invoke(...) sử dụng checkpoints.db

PostgresSaver

pip install langgraph-checkpoint-postgres psycopg2-binary

from langgraph.checkpoint.postgres import PostgresSaver
import psycopg2

conn = psycopg2.connect("postgresql://user:pass@localhost:5432/mydb")
saver = PostgresSaver(conn)
saver.setup()  # tạo bảng checkpoints nếu chưa có

agent = create_react_agent(
    model=llm,
    tools=[send_email, write_file],
    checkpointer=saver,
    interrupt_before=["tools"],
)

AsyncPostgresSaver cho FastAPI async

pip install langgraph-checkpoint-postgres asyncpg

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
import asyncpg

async def lifespan(app):
    conn = await asyncpg.connect("postgresql://user:pass@localhost:5432/mydb")
    saver = AsyncPostgresSaver(conn)
    await saver.setup()
    app.state.agent = create_react_agent(
        model=llm,
        tools=[send_email, write_file],
        checkpointer=saver,
        interrupt_before=["tools"],
    )
    yield
    await conn.close()

State trong checkpointer Postgres được lưu dưới dạng JSON serialized trong bảng checkpoints. Dữ liệu này bao gồm toàn bộ message history và intermediate state — cần tính đến storage size nếu conversation dài hoặc số thread lớn.

1. Không có checkpointer → interrupt không hoạt động

# Sai — không checkpointer
agent = create_react_agent(model=llm, tools=[send_email])
# interrupt_before không có ý nghĩa nếu không có nơi lưu state

# Đúng
memory = MemorySaver()
agent = create_react_agent(
    model=llm,
    tools=[send_email],
    checkpointer=memory,
    interrupt_before=["tools"],
)

2. Không có thread_id → mỗi invoke là conversation mới

# Sai — không thread_id → không thể resume
agent.invoke({"messages": [...]})  # không config

# Đúng — luôn truyền config với thread_id
config = {"configurable": {"thread_id": "user-123-task-456"}}
agent.invoke({"messages": [...]}, config=config)
agent.invoke(None, config=config)  # resume cùng thread

3. Invoke với input mới thay vì None khi resume

# Sai — truyền message mới khi đang pause → bắt đầu turn mới thay vì resume
result = agent.invoke(
    {"messages": [HumanMessage(content="OK proceed")]},  # ← không phải cách resume
    config=config,
)

# Đúng — None để resume, hoặc Command(resume=...) nếu dùng interrupt()
result = agent.invoke(None, config=config)
# hoặc
result = agent.invoke(Command(resume="approve"), config=config)

4. Thiếu as_node khi inject ToolMessage qua update_state

# Sai — LangGraph không biết update từ node nào
agent.update_state(config, {"messages": [reject_msg]})
# → graph không chuyển đúng edge, có thể loop hoặc lỗi

# Đúng
agent.update_state(config, {"messages": [reject_msg]}, as_node="tools")

5. interrupt_after thay vì interrupt_before cho HITL

interrupt_after=["tools"] dừng sau khi tool đã execute — quá muộn để từ chối. Tool side-effect đã chạy rồi. Luôn dùng interrupt_before=["tools"] để HITL có ý nghĩa.

6. Dùng MemorySaver trong production

Nếu server restart, mọi HITL đang pending (graph đang pause chờ approval) sẽ mất. User không thể approve/reject nữa. Bắt buộc dùng PostgresSaver hoặc SqliteSaver cho production.

• HITL cần thiết với tool có side-effect không thể hoàn tác (email, DB write, charge). Không cần cho tool read-only.
• Hai điều kiện bắt buộc: checkpointer (lưu state) + interrupt (đánh dấu điểm pause).
• interrupt_before=["tools"]: dừng trước node tools — cách đơn giản nhất, áp dụng cho tất cả tool.
• interrupt(payload) trong LangGraph 0.2+: dừng theo điều kiện bên trong node — phù hợp khi chỉ một số tool cần HITL.
• Resume bằng invoke(None, config) (approve) hoặc invoke(Command(resume=...), config) (khi dùng interrupt()).
• Modify args: update_state với AIMessage có cùng id → replace message cũ.
• Reject: inject ToolMessage giả qua update_state(..., as_node="tools") → invoke(None, config).
• Production: PostgresSaver để state survive restart và multi-worker. MemorySaver chỉ cho dev.
• Time-travel: get_state_history(config) để xem checkpoint, invoke từ checkpoint cũ để debug.

Bài 31: Multi-agent system intuition