Hướng Dẫn Sử Dụng Interrupt và Command trong LangGraph

Chào mừng bạn đến với hướng dẫn này về cách sử dụng interrupt và command trong LangGraph để xây dựng quy trình làm việc tương tác với sự can thiệp của con người. Nếu bạn mới bắt đầu với LangGraph hoặc muốn có một cái nhìn trực quan, hãy xem video hướng dẫn trên YouTube trước: Xem Video.

Trong bài viết này, chúng ta sẽ khám phá cách xây dựng một quy trình đơn giản, tạm dừng để nhận sự chấp thuận của người dùng và định tuyến động dựa trên quyết định đó. Điều này rất hữu ích cho các tình huống cần giám sát của con người, chẳng hạn như phê duyệt triển khai hoặc xem xét các quyết định do AI tạo ra. Chúng ta sẽ đi qua từng bước một, với các khối mã mà bạn có thể sao chép và làm theo trong Jupyter Notebook hoặc môi trường Python của riêng bạn. Cuối cùng, bạn sẽ hiểu cách triển khai interrupt để tạm dừng thực thi và command để định tuyến động.

LangGraph là một thư viện mạnh mẽ cho việc xây dựng các ứng dụng đa tác nhân có trạng thái với LLMs. Ở đây, chúng ta sẽ tập trung vào các tính năng interrupt và command, tận dụng checkpointing để lưu và phục hồi trạng thái một cách liền mạch.

Các Khái Niệm Chính

Trước khi đi vào mã, hãy cùng nhau điểm qua các khái niệm cơ bản. Interrupt cho phép bạn tạm dừng thực thi đồ thị và chờ đợi đầu vào từ con người, trong khi command cho phép kiểm soát động luồng đồ thị. Checkpointing là rất quan trọng vì nó giữ lại trạng thái trong những lần tạm dừng này.

Tính năng interrupt đơn giản hóa các tác nhân có sự can thiệp của con người bằng cách sử dụng lớp lưu trữ của LangGraph. Nó cho phép bạn phê duyệt hoặc từ chối các bước, chẳng hạn như các quyết định của LLM hoặc các cuộc gọi hàm, và điều hướng đồ thị tương ứng. Bạn cũng có thể xem xét và chỉnh sửa trạng thái đồ thị, như cập nhật phản hồi hoặc tài liệu, hoặc phê duyệt các cuộc gọi công cụ để giám sát.

Command, ngược lại, cung cấp giao tiếp phong phú giữa các nút. Chúng hỗ trợ định tuyến động mà không cần các cạnh được xác định trước, chuyển giao trong các thiết lập đa tác nhân, và kiểm soát luồng nâng cao bằng cách cập nhật trạng thái và chỉ định các đường dẫn.

Tổng Quan Quy Trình Làm Việc

Quy trình làm việc mẫu của chúng ta trình bày một nhiệm vụ cần sự phê duyệt, tạm dừng để nhận quyết định của người dùng, và sau đó định tuyến đến hoàn thành hoặc hủy bỏ. Điều này thể hiện một quy trình làm việc thực tế có sự can thiệp của con người: bắt đầu với một nhiệm vụ như "Triển khai tính năng mới lên môi trường sản xuất," tạm dừng để phê duyệt, và tiếp tục dựa trên việc người dùng nói "phê duyệt" hay "từ chối."

Bây giờ, hãy cùng xây dựng nó. Bạn cần cài đặt LangGraph—chạy pip install langgraph nếu bạn chưa làm điều đó. Chúng ta cũng sẽ sử dụng một số tiện ích nhập liệu và IPython để hiển thị, nhưng phần cốt lõi là pure Python.

Định Nghĩa Trạng Thái

Nền tảng của bất kỳ quy trình làm việc nào trong LangGraph là trạng thái. Đây là một từ điển chia sẻ mà tất cả các nút có thể truy cập và sửa đổi. Đối với quy trình của chúng ta, nó theo dõi nhiệm vụ, quyết định của người dùng, và trạng thái cuối cùng.

from typing_extensions import TypedDict

class WorkflowState(TypedDict):
    task: str
    # Quyết định của người dùng ('phê duyệt' hoặc 'từ chối') sẽ được lưu trữ tại đây sau khi tạm dừng.
    user_decision: str
    # Trạng thái cuối cùng của quy trình làm việc của chúng ta.
    status: str

Đoạn mã TypedDict này đảm bảo tính an toàn về kiểu khi chúng ta chuyền trạng thái xung quanh.

Các Hàm Nút

Các nút là các khối xây dựng—các hàm thực hiện các hành động và cập nhật trạng thái. Chúng ta có bốn hàm: một để nhận sự phê duyệt (với interrupt), một để định tuyến (với command), và hai hàm cuối để hoàn thành hoặc hủy bỏ.

Hàm get_approval tạm dừng đồ thị, in nhiệm vụ ra và chờ đầu vào. Hàm router quyết định bước tiếp theo một cách động. Các hàm còn lại chỉ đơn giản hoàn tất trạng thái.

from langgraph.types import interrupt, Command

def get_approval(state: WorkflowState):
    """
    Nút này sử dụng 'interrupt' để tạm dừng đồ thị.
    Nó chờ một người cung cấp quyết định trước khi đồ thị có thể tiếp tục.
    """
    print('--- ⏸️ TẠM DỪNG ĐỂ PHÊ DUYỆT ---')
    print(f"Nhiệm vụ: '{state['task']}'")
    decision = interrupt("Vui lòng nhập 'phê duyệt' hoặc 'từ chối' để tiếp tục.")
    print(f"--- ▶️ TIẾP TỤC VỚI QUYẾT ĐỊNH: '{decision}' ---")
    return {'user_decision': decision}

def router(state: WorkflowState) -> Command:
    """
    Nút này sử dụng 'Command' để định tuyến động thực thi của đồ thị.
    Dựa trên quyết định của người dùng, nó quyết định nút nào sẽ chạy tiếp theo.
    """
    print('--- 🔀 ĐỊNH TUYẾN ---')
    decision = state.get('user_decision', '').strip().lower()

    if decision == 'phê duyệt':
        print("Quyết định: ✅ Đã phê duyệt -> Định tuyến đến 'complete_task'")
        return Command(goto='complete_task')
    else:
        print("Quyết định: ❌ Đã từ chối -> Định tuyến đến 'cancel_task'")
        return Command(goto='cancel_task')

def complete_task(state: WorkflowState):
    """Nút cuối cho khi nhiệm vụ được phê duyệt."""
    print('--- 🎉 NHIỆM VỤ ĐÃ HOÀN THÀNH ---')
    return {'status': 'done'}

def cancel_task(state: WorkflowState):
    """Nút cuối cho khi nhiệm vụ bị từ chối."""
    print('--- 🗑️ NHIỆM VỤ ĐÃ HỦY BỎ ---')
    return {'status': 'canceled'}

Sao chép những đoạn mã này vào tập tin của bạn. Lưu ý cách interrupt tạm dừng thực thi, và Command(goto=...) xử lý định tuyến mà không cần các cạnh cố định.

Xây Dựng và Hiển Thị Đồ Thị

Bây giờ, hãy xây dựng đồ thị. Chúng ta sử dụng một trình xây dựng StateGraph, thêm các nút, và định nghĩa các cạnh. Quan trọng là chúng ta kích hoạt checkpointing với một bộ lưu trữ trong bộ nhớ để các interrupt hoạt động. Router không cần các cạnh rõ ràng—các command xử lý điều đó tại thời gian chạy.

Để hiển thị, chúng ta sẽ tạo một sơ đồ, nhưng hãy nhớ rằng các phần động sẽ không hiển thị tĩnh.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from IPython.display import Image, display

# Khởi tạo một bộ lưu trữ trong bộ nhớ. Điều này là cần thiết cho 'interrupt' hoạt động,
# vì nó cần lưu trạng thái của đồ thị khi tạm dừng.
memory = InMemorySaver()

# Tạo trình xây dựng đồ thị.
builder = StateGraph(WorkflowState)

# Thêm các hàm như là các nút vào đồ thị.
builder.add_node('get_approval', get_approval)
builder.add_node('router', router)
builder.add_node('complete_task', complete_task)
builder.add_node('cancel_task', cancel_task)

# Định nghĩa cấu trúc của đồ thị (các cạnh của nó).
builder.add_edge(START, 'get_approval')
builder.add_edge('get_approval', 'router')

# LƯU Ý: Chúng ta KHÔNG cần thêm các cạnh điều kiện từ nút 'router'.
# Đối tượng 'Command' được trả về bởi router xử lý định tuyến một cách động.

builder.add_edge('complete_task', END)
builder.add_edge('cancel_task', END)

# Biên dịch đồ thị, kích hoạt bộ kiểm tra.
graph = builder.compile(checkpointer=memory)

# Bạn có thể hiển thị cấu trúc đồ thị.
try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception as e:
    print(f'Không thể hiển thị đồ thị: {e}')

Chạy đoạn mã này để xây dựng đồ thị của bạn. Nếu bạn đang trong Jupyter, bạn sẽ thấy sơ đồ; nếu không, nó có thể chỉ in ra một lỗi nếu thiếu phụ thuộc.

Thực Thi Quy Trình Làm Việc

Cuối cùng, hãy chạy nó. Chúng ta sẽ mô phỏng hai kịch bản: phê duyệt và từ chối. Mỗi cái sử dụng một ID luồng duy nhất để theo dõi trạng thái độc lập. Chúng ta truyền phát các sự kiện để xem tiến trình, tạm dừng tại interrupt, và tiếp tục với một quyết định.

Đầu tiên, chạy phê duyệt:

# --- Chạy 1: Phê duyệt nhiệm vụ ---
print('\n' + '=' * 50 + '\n🚀 BẮT ĐẦU CHẠY 1: PHÊ DUYỆT\n' + '=' * 50)

# Cần một 'thread_id' để theo dõi trạng thái của một lần chạy duy nhất.
thread = {'configurable': {'thread_id': 'run-1'}}
initial_task = {'task': 'Triển khai tính năng mới lên môi trường sản xuất'}

# Bắt đầu đồ thị. Nó sẽ chạy cho đến khi gặp 'interrupt' trong nút 'get_approval'.
for event in graph.stream(initial_task, thread, stream_mode='values', debug=True):
    print(f'\n[SỰ KIỆN TRUYỀN PHÁT]:\n{event}\n')

# Tại thời điểm này, đồ thị đã tạm dừng. Hãy tiếp tục nó với quyết định của người dùng.
print("\n... Tiếp tục Chạy 1 với 'phê duyệt' ...\n")

for event in graph.stream(Command(resume='phê duyệt'), thread, stream_mode='values', debug=True):
    print(f'\n[SỰ KIỆN TRUYỀN PHÁT]:\n{event}\n')

Đoạn mã này sẽ tạm dừng, chờ đợi "phê duyệt," sau đó hoàn tất nhiệm vụ.

Bây giờ, chạy từ chối:

# --- Chạy 2: Từ chối nhiệm vụ ---
print('\n' + '=' * 50 + '\n🚀 BẮT ĐẦU CHẠY 2: TỪ CHỐI\n' + '=' * 50)

# Sử dụng một thread_id mới cho lần chạy thứ hai, độc lập.
thread2 = {'configurable': {'thread_id': 'run-2'}}

# Bắt đầu lần chạy thứ hai.
for event in graph.stream(initial_task, thread2, stream_mode='values', debug=True):
    print(f'\n[SỰ KIỆN TRUYỀN PHÁT]:\n{event}\n')

# Tiếp tục lần chạy thứ hai, nhưng lần này với quyết định 'từ chối'.
print("\n... Tiếp tục Chạy 2 với 'từ chối' ...\n")
for event in graph.stream(Command(resume='từ chối'), thread2, stream_mode='values', debug=True):
    print(f'\n[SỰ KIỆN TRUYỀN PHÁT]:\n{event}\n')

Đoạn mã này sẽ định tuyến đến việc hủy bỏ.

Kết Luận

Bạn đã xây dựng một quy trình làm việc có sự can thiệp của con người với LangGraph! Các interrupt xử lý việc tạm dừng để nhận đầu vào, các command cho phép định tuyến động, và checkpointing giữ mọi thứ có trạng thái. Hãy thử nghiệm bằng cách thay đổi nhiệm vụ hoặc thêm nhiều nút hơn. Để biết thêm các ứng dụng nâng cao, hãy kiểm tra tài liệu của LangGraph. Nếu bài viết này hữu ích, hãy thích video và chia sẻ ý kiến của bạn trong phần bình luận!