Nhảy đến nội dung chính

Streaming và Human-in-the-loop trong LangGraph

Đây là phần giải thích chi tiết và bổ sung kiến thức về streaming mode trong LangGraph:

1. Sự khác biệt giữa .invoke() và .stream()

.invoke():

  • Thực thi toàn bộ ứng dụng một lần

  • Chỉ trả về kết quả cuối cùng

  • Giống như xem một bộ phim từ đầu đến cuối không tạm dừng

.stream() (QUAN TRỌNG HƠN):

  • Thực thi từng bước (step-by-step)

  • Cho phép dừng/tạm ngưng quá trình thực thi

  • Có thể chèn các tương tác của con người vào giữa quá trình

  • Giống như xem phim có thể tạm dừng để thảo luận từng cảnh

Ví dụ minh họa:

# Cách cũ - không linh hoạt
final_result = agent.invoke({"input": "Câu hỏi phức tạp"})

# Cách mới - có kiểm soát từng bước
for step in agent.stream({"input": "Câu hỏi phức tạp"}):
    if need_human_review(step):
        human_feedback = get_human_input()
        agent.adjust(human_feedback)

2. Hai chế độ Streaming quan trọng

a. Chế độ values (Nên dùng)

  • Hiển thị toàn bộ state tại mỗi bước

  • Giống như xem toàn bộ bức tranh sau mỗi nét vẽ

  • Phù hợp cho debug và theo dõi toàn bộ tiến trình

b. Chế độ updates

  • Chỉ hiển thị phần thay đổi mới nhất

  • Giống như chỉ xem các nét vẽ mới được thêm vào

  • Tiết kiệm băng thông nhưng khó debug hơn

Bảng so sánh:

Feature values Mode updates Mode
Hiển thị Toàn bộ state Chỉ thay đổi
Dữ liệu trả về Nhiều hơn Ít hơn
Use case Debug ứng dụng Hiệu suất cao
Độ phức tạp Cao hơn Thấp hơn

3. Ứng dụng thực tế của .stream()

Human-in-the-loop workflow điển hình:

  1. Agent thực thi đến bước quan trọng

  2. Tự động dừng và chờ input từ người dùng

  3. Người dùng có thể:

    • Phê duyệt/ từ chối hành động tiếp theo

    • Điều chỉnh state của hệ thống

    • Cung cấp dữ liệu bổ sung

    • Gỡ lỗi trực tiếp

Ví dụ chatbot nâng cao:

for chunk in agent.stream(conversation, stream_mode="values"):
    if is_critical_step(chunk):
        show_analysis(chunk)  # Hiển thị phân tích cho người dùng
        decision = ask_for_approval()  # Xác nhận từ người dùng
        
        if not decision.approved:
            agent.adjust_trajectory(decision.feedback)  # Điều chỉnh hướng xử lý

4. Lời khuyên thực tế khi triển khai

  1. Ưu tiên dùng stream_mode="values" khi phát triển để dễ debug

  2. Chỉ chuyển sang "updates" khi ứng dụng đã ổn định và cần tối ưu hiệu suất

  3. Luôn thiết kế các "điểm kiểm soát" (checkpoints) cho human-in-the-loop tại:

    • Bước ra quyết định quan trọng

    • Khi cần xác nhận tính chính xác

    • Khi agent không chắc chắn

  4. Xử lý lỗi thông minh:

try:
    for chunk in agent.stream(...):
        process(chunk)
except HumanInterventionRequired as e:
    handle_exception(e)
    wait_for_human_decision()

5. Bổ sung kiến thức quan trọng

Tại sao .stream() mạnh mẽ hơn?

  • Cho phép xây dựng agent có khả năng tự sửa lỗi

  • Tạo điều kiện cho học tăng cường từ con người (RLHF)

  • Hỗ trợ xử lý tác vụ dài mà không bị timeout

  • Cho phép hiển thị tiến trình thời gian thực cho người dùng

Pattern hay gặp:

# Hybrid approach - Kết hợp cả invoke và stream khi cần
if is_simple_task(task):
    return agent.invoke(task)
else:
    return process_stream(agent.stream(task))

Hãy coi .stream() như "cửa sổ vào bộ não" của agent - bạn có thể quan sát và can thiệp vào quá trình xử lý theo cách chưa từng có với .invoke() thông thường!

Tác giả: Đỗ Ngọc Tú
Công Ty Phần Mềm VHTSoft

Lên đầu trang