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:

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

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

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

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

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

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

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

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

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