Hướng Dẫn Triển Khai AG-UI và LangGraph Streaming

🎯 Mục Đích

Hướng dẫn này trình bày cách đạt được streaming sự kiện thời gian thực từ các quy trình AI đến giao diện người dùng (UI) bằng cách sử dụng giao thức AG-UI với thực thi LangGraph StateGraph. Phương pháp này cung cấp độ trễ dưới 100ms để phản hồi người dùng trực tiếp trong các hoạt động AI phức tạp.

🚀 Ứng Dụng Demo

Một triển khai hoàn chỉnh của kiến trúc này có sẵn tại:
https://github.com/cimulink/ai-workflow-engine

Kho lưu trữ bao gồm:

Triển khai máy chủ thuần túy LangGraph + AG-UI
Giao diện React với streaming thời gian thực
Ví dụ quy trình xử lý tài liệu
Hướng dẫn cài đặt và triển khai đầy đủ

📋 Bối Cảnh Giao Thức AG-UI

AG-UI định nghĩa một tập hợp các loại sự kiện tiêu chuẩn cho tương tác thời gian thực giữa đại lý và người dùng. Mục tiêu của chúng tôi là thích ứng toàn bộ quy trình LangGraph để tạo ra những sự kiện đã được định nghĩa trước và streaming chúng từ backend đến frontend để phản hồi người dùng theo thời gian thực.

Các Loại Tin Nhắn AG-UI

AG-UI định nghĩa một số danh mục sự kiện cho các khía cạnh khác nhau của giao tiếp đại lý:

Sự Kiện Vòng Đời

RUN_STARTED, RUN_FINISHED, RUN_ERROR
STEP_STARTED, STEP_FINISHED

Sự Kiện Tin Nhắn Văn Bản

TEXT_MESSAGE_START, TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_END

Sự Kiện Gọi Công Cụ

TOOL_CALL_START, TOOL_CALL_ARGS, TOOL_CALL_END

Sự Kiện Quản Lý Trạng Thái

STATE_SNAPSHOT, STATE_DELTA, MESSAGES_SNAPSHOT

Sự Kiện Đặc Biệt

RAW, CUSTOM

🏗️ Tổng Quan Kiến Trúc

Hệ thống kết hợp ba thành phần chính:

LangGraph StateGraph - Quản lý điều phối quy trình làm việc với các nút và cạnh điều kiện
Giao thức AG-UI - Định nghĩa các loại sự kiện cho giao tiếp UI thời gian thực
Streaming HTTP - Sử dụng Server-Sent Events (SSE) để streaming tương thích với trình duyệt

Kiến Trúc Cấp Cao

Tổng Quan Dòng Dữ Liệu

🔧 Các Thành Phần Kỹ Thuật Cốt Lõi

1. asyncio.Queue: Trung Tâm Quản Lý Sự Kiện

asyncio.Queue hoạt động như một bộ đệm an toàn giữa việc thực thi nút LangGraph và streaming HTTP. Khi các nút LangGraph hoàn thành, chúng sẽ đặt sự kiện vào hàng đợi. Điểm cuối streaming HTTP liên tục đọc từ hàng đợi và gửi sự kiện đến frontend.

Lợi ích chính:

An toàn đa luồng: Nhiều nút có thể phát ra sự kiện đồng thời
Backpressure: Ngăn ngừa tràn bộ nhớ trong quá trình xử lý nặng
Bảo toàn thứ tự: Các sự kiện duy trì thứ tự thời gian
Không chặn: Việc sản xuất và tiêu thụ sự kiện diễn ra độc lập

2. yield Keyword: Mô Hình Streaming Python

Từ khóa yield của Python biến đổi các hàm thông thường thành các trình tạo async. Thay vì trả về tất cả kết quả cùng một lúc, hàm sẽ phát ra sự kiện một cách tuần tự khi chúng được sản xuất. Điều này tạo ra một pipeline streaming tiết kiệm bộ nhớ, nơi các sự kiện được xử lý ngay lập tức thay vì được đệm.

Lợi ích Streaming:

Tiết kiệm bộ nhớ: Các sự kiện được phát ra khi sản xuất, không được lưu trữ
Thời gian thực: Không có độ trễ đệm giữa sản xuất và tiêu thụ sự kiện
Đánh giá lười biếng: Thực thi tạm dừng cho đến khi sự kiện tiếp theo được yêu cầu
Kiểm soát dòng chảy tự nhiên: Người tiêu dùng kiểm soát tốc độ xử lý

3. Custom AGUIStreamingCheckpointer: Trạng Thái + Sự Kiện

Điều này mở rộng SqliteSaver checkpointer tích hợp của LangGraph để tự động phát ra các sự kiện AG-UI mỗi khi trạng thái quy trình làm việc thay đổi. Mỗi khi một nút hoàn thành và LangGraph lưu một điểm kiểm tra, checkpointer tùy chỉnh cũng phát ra một sự kiện AG-UI tương ứng.

Lợi ích Checkpointer:

Sự kiện tự động: Không cần phát sự kiện bằng tay trong các nút
Tính nhất quán trạng thái: Các sự kiện luôn phản ánh trạng thái thực tế của LangGraph
Lưu trữ thống nhất: Cơ sở dữ liệu và streaming làm việc cùng nhau
Hỗ trợ phục hồi: Các quy trình làm việc thất bại có thể tiếp tục với toàn bộ lịch sử sự kiện

4. Streaming HTTP: Server-Sent Events (SSE)

FastAPI phục vụ các sự kiện AG-UI bằng định dạng Server-Sent Events. SSE là một tiêu chuẩn web cho phép các máy chủ đẩy dữ liệu đến trình duyệt qua một kết nối HTTP duy nhất. Mỗi sự kiện được định dạng là data: {json}\n\n và được gửi ngay lập tức đến khách hàng.

Lợi ích SSE so với WebSocket:

Giao thức đơn giản hơn: HTTP tiêu chuẩn, không cần nâng cấp kết nối
Tự động kết nối lại: Trình duyệt tự động xử lý kết nối lại
Thân thiện với tường lửa: Sử dụng các cổng HTTP tiêu chuẩn, hoạt động qua proxy
Một chiều tối ưu: Hoàn hảo cho streaming sự kiện (không cần hai chiều)
Gỡ lỗi tốt hơn: Các công cụ HTTP tiêu chuẩn hoạt động (curl, Postman)

🔄 Dòng Chảy Cuối Đến Cuối

Sơ Đồ Chuỗi Hoàn Chỉnh

Dòng Chảy Bước Từng Bước

Yêu cầu từ Frontend: Thành phần React gửi nội dung tài liệu qua hook useAgent
Thiết lập Streaming HTTP: FastAPI tạo trình tạo async cho streaming sự kiện
Thực Thi LangGraph: Các nút quy trình làm việc thực thi với định tuyến điều kiện
Tạo Sự Kiện: Checkpointer tùy chỉnh phát ra các sự kiện AG-UI khi có thay đổi trạng thái
Quản Lý Hàng Đợi: Các sự kiện chảy qua asyncio.Queue đến phản hồi HTTP
Xử Lý Frontend: Trình duyệt nhận các sự kiện SSE và cập nhật UI theo thời gian thực

Hình Ảnh Dòng Chảy Sự Kiện

✅ Ưu và Nhược Điểm

Ưu Điểm

Trải Nghiệm Người Dùng Thời Gian Thực: Phản hồi tiến độ ngay lập tức, phản hồi nhạy bén với người dùng
LangGraph Thực Sự: Bảo tồn điều phối StateGraph, các cạnh điều kiện, điểm kiểm tra
Có thể mở rộng: AsyncIO xử lý các luồng đồng thời, theo dõi nguồn sự kiện
Thân Thiện Với Nhà Phát Triển: Gỡ lỗi HTTP tiêu chuẩn, sự kiện an toàn kiểu, các mẫu React quen thuộc

Nhược Điểm

Phức Tạp: Cần hiểu các trình tạo async, bảo trì checkpointer tùy chỉnh
Sử Dụng Tài Nguyên: Kết nối liên tục, tăng trưởng cơ sở dữ liệu
Xử Lý Lỗi: Gián đoạn streaming, lỗi một phần, logic phục hồi kết nối
Kiểm Tra: Streaming async khó kiểm tra đơn vị hơn, xem xét thời gian, điều kiện đua

🎯 Chiến Lược Triển Khai

Lộ Trình Triển Khai Tiến Bộ

Mô Hình Thiết Kế Sự Kiện

Chiến Lược Xử Lý Lỗi

Giảm Độ Phức Tạp: Phát ra sự kiện lỗi, trả về kết quả một phần
Phục Hồi Kết Nối: Kết nối lại tự động từ frontend với việc giảm dần theo cấp số nhân
Gộp Sự Kiện: Giảm thiểu băng thông mạng trong các khoảng thời gian khối lượng lớn

🚀 Khi Nào Sử Dụng Kiến Trúc Này

Hoàn Hảo Cho:

Xử Lý Tài Liệu: Phân tích nhiều bước với xem xét của con người
Pipeline Dữ Liệu: Theo dõi tiến độ ETL thời gian thực
Đại Lý AI: Quy trình làm việc hội thoại với việc sử dụng công cụ
Công Việc Dài: Các quy trình >30 giây cần cập nhật tiến độ

Xem Xét Các Thay Thế:

Yêu Cầu/Phản Hồi Đơn Giản: Các cuộc gọi API đơn lẻ
Xử Lý Gộp: Không yêu cầu thời gian thực
Hạn Chế Tài Nguyên: Bộ nhớ/băng thông hạn chế

💡 Những Điều Chính

Kiến trúc này cung cấp giao diện quy trình làm việc AI thời gian thực sẵn sàng sản xuất bằng cách kết hợp:

Điều phối LangGraph (StateGraph, nút, cạnh)
Giao thức streaming AG-UI (sự kiện thời gian thực)
Streaming HTTP SSE (tương thích với trình duyệt, gỡ lỗi đơn giản)
asyncio.Queue + yield (pipeline sự kiện tiết kiệm bộ nhớ)

Yếu Tố Thành Công: Thiết kế sự kiện để mang lại giá trị cho người dùng, triển khai xử lý lỗi mạnh mẽ, giám sát hiệu suất, kiểm tra hành vi streaming, lập kế hoạch cho việc mở rộng ngang.

Độ phức tạp là hợp lý khi trải nghiệm người dùng và tính minh bạch quy trình làm việc là rất quan trọng. Bắt đầu đơn giản và thêm các khả năng streaming khi tương tác thời gian thực trở nên cần thiết.