Phần 4: Active RAG & Strict Tool Calling: Kết Nối LLM Với Real-time Inventory API

← Chương trước: Phần 3: Tối Ưu Hóa Tìm Kiếm Lai Qdrant | Mục lục Series | Chương tiếp theo: Phần 5: Vòng Lặp Tự Phản Biện (Critique Loop) →

Answer-first: Active RAG kết nối mô hình ngôn ngữ với API tồn kho và định giá qua cơ chế Tool Calling với JSON Schema. Điều phối công cụ bằng Eino kết hợp circuit breaker Sony gobreaker và dataloader giúp kiểm tra 15 kho hàng dưới 4ms mà không gây sụt giảm hiệu năng backend.

📖 Bản tiếng Anh (English Edition): Part 4: Active RAG & Strict Tool Calling: Connecting LLMs to Real-Time Inventory APIs

1. Active RAG Đối Đầu Passive RAG: Yêu Cầu Sống Còn Về Trạng Thái Thực Trong E-Commerce

BLUF (Bottom Line Up Front): Passive RAG chỉ đọc dữ liệu tĩnh nên dễ trả về sản phẩm hết hàng hoặc sai giá. Active RAG khắc phục triệt để bằng cách chủ động gọi API tồn kho và định giá thời gian thực qua Tool Calling có JSON Schema, đảm bảo dữ liệu chính xác 100% trước khi phản hồi.

Trong các bài toán hỏi đáp tài liệu truyền thống, Passive RAG hoạt động theo một đường thẳng: nhận câu hỏi, tìm kiếm các đoạn văn bản vector gần nhất và đưa vào prompt để LLM tóm tắt. Tuy nhiên, trong thương mại điện tử, cách tiếp cận này lập tức thất bại:

• Dữ Liệu Tồn Kho Biến Động: Tồn kho tại các kho hàng chi nhánh thay đổi liên tục theo từng mili-giây. Một đoạn văn bản vector tĩnh ghi “Giày chạy bộ Nike Pegasus size 42 còn hàng” sẽ trở thành thông tin giả mạo ngay thời điểm có khách hàng khác hoàn tất thanh toán.
• Giá Cá Nhân Hóa Động: Mỗi khách hàng đăng nhập có thể có mức chiết khấu thành viên khác nhau (VIP Bạc, VIP Vàng, mã giảm giá sinh nhật). Không thể nhúng trước hàng triệu biến thể giá vào cơ sở dữ liệu vector.
• Điều Kiện Giao Hàng Hỏa Tốc: Trả lời các câu hỏi như “Sản phẩm này có thể giao đến Quận 1 trước 5 giờ chiều nay không?” đòi hỏi phải tra cứu trực tiếp vị trí kho hàng gần nhất và lịch trình của các đối tác vận chuyển.

Mô Hình Thực Thi Của Active RAG

Trong mô hình Active RAG, việc tìm kiếm vector ban đầu chỉ đóng vai trò đề xuất các ứng viên tiềm năng (Candidates). Sau đó, agent sẽ tự động kích hoạt các công cụ (Tools) để xác thực trạng thái vật lý thực tế:

sequenceDiagram
    autonumber
    actor Shopper as "Khách Hàng Mua Sắm"
    participant Orchestrator as "Cụm Điều Phối Go Eino"
    participant VectorEngine as "Qdrant Hybrid Engine"
    participant StockRedis as "Cụm Redis Tồn Kho Bitmap"
    participant PricingSvc as "Dịch Vụ Định Giá VIP"
    participant LLMReflection as "Mô Hình Phản Biện / SLM"

    Shopper->>Orchestrator: Tìm: "Áo khoác chạy bộ chống nước size L dưới 1.5 triệu"
    Orchestrator->>VectorEngine: Truy vấn 10 sản phẩm vector phù hợp nhất
    VectorEngine-->>Orchestrator: Trả về danh sách 10 mã SKU ứng viên
    
    rect rgb(240, 248, 255)
        Note over Orchestrator,PricingSvc: Giai Đoạn Gọi Công Cụ Đồng Thời (<4ms)
        par Gọi Công Cụ Qua Cơ Chế Dataloader
            Orchestrator->>StockRedis: Kiểm tra tồn kho Bitmap cho 10 SKUs
            Orchestrator->>PricingSvc: Tính giá chiết khấu theo hạng thành viên
        end
        StockRedis-->>Orchestrator: 8 SKUs còn hàng, 2 SKUs đã hết hàng
        PricingSvc-->>Orchestrator: Áp dụng giảm giá 15% cho thành viên VIP Vàng
    end
    
    Orchestrator->>LLMReflection: Truyền các sản phẩm đã xác thực thực tế
    LLMReflection-->>Orchestrator: Sinh thẻ sản phẩm và lời khuyên phù hợp
    Orchestrator-->>Shopper: Trả về kết quả tìm kiếm chính xác (P99: 42ms)

2. Định Nghĩa Hợp Đồng Công Cụ Bằng JSON Schema Nghiêm Ngặt

BLUF (Bottom Line Up Front): Việc cho phép LLM sinh tham số công cụ tự do dễ gây ra lỗi sai kiểu dữ liệu hoặc ảo giác tham số; thiết lập JSON Schema nghiêm ngặt đảm bảo 100% độ tin cậy khi giao tiếp giữa Agent và các microservices nội bộ.

Khi tích hợp mô hình ngôn ngữ với các hệ thống nghiệp vụ quan trọng, việc để mô hình tự do sinh JSON thường dẫn tới các thảm họa:

1. Ảo Giác Tham Số: Mô hình tự bịa ra mã kho không tồn tại (ví dụ: "warehouse_code": "KHO_SAO_HOA").
2. Lệch Kiểu Dữ Liệu: Truyền mã SKU dạng số nguyên thay vì chuỗi ký tự, hoặc truyền giá trị trần âm.
3. Tấn Công Prompt Injection: Khách hàng cố tình nhập các câu lệnh độc hại để ép agent gọi các công cụ quản trị (như xóa sản phẩm hoặc sửa giá về 0 đồng).

Để triệt tiêu các rủi ro này, mọi công cụ tìm kiếm trong CloudWeGo Eino đều phải đăng ký JSON Schema Nghiêm Ngặt:

package tools

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/cloudwego/eino/schema"
)

// WarehouseStockInput định nghĩa cấu trúc dữ liệu đầu vào chuẩn tắc
type WarehouseStockInput struct {
	SKUList       []string `json:"sku_list" jsonschema:"required,description=Danh sách tối đa 20 mã SKU cần kiểm tra"`
	DeliveryZone  string   `json:"delivery_zone" jsonschema:"required,enum=MIEN_BAC,enum=MIEN_TRUNG,enum=MIEN_NAM"`
}

// WarehouseStockOutput định nghĩa dữ liệu đầu ra từ dịch vụ kho
type WarehouseStockOutput struct {
	StockBySKU map[string]int  `json:"stock_by_sku"`
	Available  map[string]bool `json:"available"`
}

// NewStockVerificationTool khởi tạo công cụ Eino Tool an toàn kiểu dữ liệu
func NewStockVerificationTool(stockClient WarehouseServiceClient) schema.Tool {
	return &stockToolImplementation{client: stockClient}
}

type stockToolImplementation struct {
	client WarehouseServiceClient
}

func (t *stockToolImplementation) Info(ctx context.Context) (*schema.ToolInfo, error) {
	return &schema.ToolInfo{
		Name: "verify_warehouse_inventory",
		Desc: "Kiểm tra số lượng tồn kho vật lý thực tế của các mã SKU tại từng vùng miền.",
		ParamsOneOf: schema.NewParamsOneOfByParams(map[string]any{
			"type": "object",
			"properties": map[string]any{
				"sku_list": map[string]any{
					"type": "array",
					"items": map[string]any{"type": "string"},
					"minItems": 1,
					"maxItems": 20,
					"description": "Danh sách mã SKU sản phẩm",
				},
				"delivery_zone": map[string]any{
					"type": "string",
					"enum": []string{"MIEN_BAC", "MIEN_TRUNG", "MIEN_NAM"},
					"description": "Khu vực giao hàng của khách",
				},
			},
			"required": []string{"sku_list", "delivery_zone"},
		}),
	}, nil
}

func (t *stockToolImplementation) InvokableRun(ctx context.Context, argumentsInJSON string, opts ...schema.ToolOption) (string, error) {
	var input WarehouseStockInput
	if err := json.Unmarshal([]byte(argumentsInJSON), &input); err != nil {
		return "", fmt.Errorf("tham số JSON không hợp lệ: %w", err)
	}

	stockMap, err := t.client.BatchQueryStock(ctx, input.SKUList, input.DeliveryZone)
	if err != nil {
		return "", fmt.Errorf("truy vấn tồn kho thất bại: %w", err)
	}

	res := WarehouseStockOutput{
		StockBySKU: stockMap,
		Available:  make(map[string]bool),
	}
	for sku, count := range stockMap {
		res.Available[sku] = count > 0
	}

	bytes, _ := json.Marshal(res)
	return string(bytes), nil
}

type WarehouseServiceClient interface {
	BatchQueryStock(ctx context.Context, skus []string, zone string) (map[string]int, error)
}

3. Kiểm Tra Tồn Kho Kho Vật Lý Dưới 4ms Qua Redis Bitmaps

BLUF (Bottom Line Up Front): Thực hiện các câu truy vấn SQL quan hệ (SELECT quantity FROM inventory WHERE sku = ?) dưới tải 10,000 QPS sẽ làm cạn kiệt connection pool cơ sở dữ liệu; mã hóa trạng thái tồn kho thành các mảng Redis Bitmaps giúp kiểm tra 20 SKUs đồng thời trong dưới 4ms.

Trong bài toán tìm kiếm sản phẩm, mục tiêu kiểm tra tồn kho ban đầu là một câu hỏi nhị phân: Mã SKU này hiện có sẵn hàng để bán hay không? Việc gửi hàng chục nghìn truy vấn SQL đọc vào cơ sở dữ liệu giao dịch chính sẽ làm nghẽn toàn bộ hệ thống bán hàng.

Kiến Trúc Mảng Bitmaps Trong Redis Cluster

Thay vì lưu trữ dữ liệu dạng JSON hoặc bảng SQL, hệ thống mã hóa tồn kho thành các Redis Bitmaps:

• Mỗi kho hàng khu vực có một key Redis riêng: inventory:bitmap:MIEN_BAC:kho_tong.
• Mỗi mã SKU được gán một số nguyên 32-bit tăng dần (sku_offset).
• Một bit mang giá trị 1 nghĩa là còn hàng; bit mang giá trị 0 nghĩa là hết hàng.
• Khi có biến động tồn kho, hệ sinh thái CDC cập nhật bằng lệnh SETBIT trong thời gian dưới 100 micro-giây.

flowchart LR
    subgraph RedisBitmapsCluster ["Cụm Redis Lưu Trữ Tồn Kho Dạng Bit"]
        direction TB
        B1["Key: 'inv:MIEN_BAC' -> [ 1 | 1 | 0 | 1 | 0 | 0 | 1 ... ]"]
        B2["Key: 'inv:MIEN_NAM' -> [ 0 | 1 | 1 | 1 | 0 | 1 | 0 ... ]"]
    end

    subgraph BatchQueryExecution ["Truy Vấn Gom Nhóm Siêu Tốc (<3ms)"]
        SKUOffsets["Các Mã Cần Check: [Offset 2048, Offset 4096, Offset 8192]"]
        SKUOffsets --> RedisPipe["Gửi Lệnh Qua Redis Pipeline (BITFIELD / MGET)"]
        RedisPipe --> RedisBitmapsCluster
        RedisBitmapsCluster --> ReturnBits["Kết Quả Bit: [1 (Còn), 0 (Hết), 1 (Còn)]"]
    end

4. Cơ Chế Gom Nhóm Yêu Cầu Bằng Dataloader Trong Golang

BLUF (Bottom Line Up Front): Việc gửi từng lệnh RPC riêng lẻ cho mỗi sản phẩm ứng viên tạo ra hiện tượng khuếch đại mạng $N+1$; kỹ thuật Dataloader gom tất cả các yêu cầu kiểm tra SKU trong một cửa sổ 2ms thành một lệnh pipeline duy nhất, giảm 82% độ trễ mạng.

Để ngăn chặn việc mỗi sản phẩm ứng viên gửi một cuộc gọi mạng riêng biệt sang Redis hoặc Microservices, chúng ta áp dụng mẫu thiết kế Dataloader:

package dataloader

import (
	"context"
	"sync"
	"time"

	"github.com/redis/go-redis/v9"
)

// StockBatchLoader gom các yêu cầu kiểm tra tồn kho trong cửa sổ 2ms
type StockBatchLoader struct {
	redisClient *redis.Client
	mu          sync.Mutex
	requests    map[int64]chan bool
	waitDuration time.Duration
}

func NewStockBatchLoader(client *redis.Client) *StockBatchLoader {
	return &StockBatchLoader{
		redisClient:  client,
		requests:     make(map[int64]chan bool),
		waitDuration: 2 * time.Millisecond,
	}
}

// CheckStock kiểm tra tồn kho cho 1 SKU, tự động ghép vào mẻ gom nhóm
func (loader *StockBatchLoader) CheckStock(ctx context.Context, skuOffset int64) (bool, error) {
	loader.mu.Lock()
	ch, exists := loader.requests[skuOffset]
	if !exists {
		ch = make(chan bool, 1)
		loader.requests[skuOffset] = ch

		// Nếu là yêu cầu đầu tiên trong mẻ mới, khởi động timer
		if len(loader.requests) == 1 {
			go loader.executeBatchAfter(loader.waitDuration)
		}
	}
	loader.mu.Unlock()

	select {
	case inStock := <-ch:
		return inStock, nil
	case <-ctx.Done():
		return false, ctx.Err()
	}
}

func (loader *StockBatchLoader) executeBatchAfter(d time.Duration) {
	time.Sleep(d)

	loader.mu.Lock()
	batch := loader.requests
	loader.requests = make(map[int64]chan bool)
	loader.mu.Unlock()

	if len(batch) == 0 {
		return
	}

	// Gửi một lệnh duy nhất qua Redis Pipeline
	pipe := loader.redisClient.Pipeline()
	cmdMap := make(map[int64]*redis.IntCmd)

	for offset := range batch {
		cmdMap[offset] = pipe.GetBit(context.Background(), "inventory:bitmap:global", offset)
	}

	_, _ = pipe.Exec(context.Background())

	for offset, cmd := range cmdMap {
		val, err := cmd.Result()
		hasStock := err == nil && val == 1
		batch[offset] <- hasStock
	}
}

5. Tích Hợp Kháng Lỗi: Circuit Breaker Với Sony Gobreaker

BLUF (Bottom Line Up Front): Các microservices phụ trợ thường xuyên gặp sự cố nghẽn mạng trong giờ cao điểm flash sale; bọc các lệnh gọi công cụ bằng Sony gobreaker giúp ngắt kết nối tự động trong 5 lần lỗi liên tiếp, lập tức trả về dữ liệu thoái lui an toàn mà không làm treo thanh tìm kiếm.

Khi một cỗ máy tìm kiếm Agentic kết nối trực tiếp với nhiều microservices nội bộ (hệ thống tính giá, kho hàng, khuyến mại), nó phải đối mặt với nguy cơ Sụp Đổ Dây Chuyền (Cascading Outage). Nếu dịch vụ tính giá bị nghẽn và độ trễ tăng từ 5ms lên 10,000ms, toàn bộ các luồng worker của cỗ máy tìm kiếm sẽ bị treo theo:

stateDiagram-v2
    [*] --> ClosedState
    ClosedState --> OpenState: Số Lỗi Liên Tiếp > 5 HOẶC Tỷ Lệ Lỗi > 20%
    OpenState --> HalfOpenState: Hết Thời Gian Nghỉ (Sleep Window 5000ms)
    HalfOpenState --> ClosedState: 3 Yêu Cầu Thử Nghiệm Thành Công Liên Tục
    HalfOpenState --> OpenState: Bất Kỳ Yêu Cầu Nào Bị Lỗi
    
    note right of ClosedState
        Trạng Thái Bình Thường - Yêu cầu được gửi trực tiếp tới microservice
    end note
    note right of OpenState
        Trạng Thái Ngắt Mạch - Tự động thoái lui về giá mặc định
    end note
    note right of HalfOpenState
        Trạng Thái Thử Nghiệm - Cho phép một lượng nhỏ request đi qua
    end note

Triển Khai Circuit Breaker Cho Công Cụ Định Giá

package resilience

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/sony/gobreaker"
)

// ProtectedPricingService bọc lệnh gọi dịch vụ tính giá bằng Circuit Breaker
type ProtectedPricingService struct {
	breaker *gobreaker.CircuitBreaker
}

func NewProtectedPricingService() *ProtectedPricingService {
	settings := gobreaker.Settings{
		Name:        "PricingToolBreaker",
		MaxRequests: 3,                // Số request cho phép ở trạng thái Half-Open
		Interval:    10 * time.Second, // Chu kỳ xóa bộ đếm
		Timeout:     5 * time.Second,  // Thời gian mở mạch trước khi thử lại
		ReadyToTrip: func(counts gobreaker.Counts) bool {
			failureRatio := float64(counts.TotalFailures) / float64(counts.Requests)
			return counts.Requests >= 10 && failureRatio >= 0.20
		},
		OnStateChange: func(name string, from gobreaker.State, to gobreaker.State) {
			fmt.Printf("CircuitBreaker [%s] chuyển trạng thái từ %v sang %v\n", name, from, to)
		},
	}
	return &ProtectedPricingService{breaker: gobreaker.NewCircuitBreaker(settings)}
}

// CalculateDiscountWithFallback tính giá an toàn, tự động thoái lui khi đối tác gặp sự cố
func (s *ProtectedPricingService) CalculateDiscountWithFallback(ctx context.Context, sku string, basePrice float64) (float64, error) {
	res, err := s.breaker.Execute(func() (any, error) {
		// Ép trần thời gian gọi dịch vụ định giá tối đa 25ms
		childCtx, cancel := context.WithTimeout(ctx, 25*time.Millisecond)
		defer cancel()

		return mockPricingRPC(childCtx, sku, basePrice)
	})

	if err != nil {
		// Khi mạch bị ngắt (Circuit Open) hoặc timeout, thoái lui về giá niêm yết chuẩn
		if errors.Is(err, gobreaker.ErrOpenState) || errors.Is(err, gobreaker.ErrTooManyRequests) {
			return basePrice, nil // Thoái lui an toàn: hiển thị giá gốc không chiết khấu VIP
		}
		return basePrice, nil
	}

	return res.(float64), nil
}

func mockPricingRPC(ctx context.Context, sku string, basePrice float64) (float64, error) {
	select {
	case <-time.After(3 * time.Millisecond):
		return basePrice * 0.85, nil // Giảm 15% cho thành viên
	case <-ctx.Done():
		return 0, ctx.Err()
	}
}

6. Hồ Sơ Sự Cố Thực Tế: Thảm Họa Treo Hệ Thống Dây Chuyền Do Kho Hàng Bị Treo

BLUF (Bottom Line Up Front): Dịch vụ kho hàng bị khóa bảng dữ liệu do một câu lệnh báo cáo chậm trong đêm hội 11.11; do không có circuit breaker và timeout, cổng tìm kiếm bị giữ hơn 22,000 kết nối TCP treo, làm sập hoàn toàn hệ thống sàn thương mại điện tử trong 80 phút.

Tóm Tắt Sự Cố

• Thời Điểm: 00:00 - 01:20 UTC, Ngày 11 Tháng 11 Năm 2025 (Siêu Hội Mua Sắm 11.11).
• Hiện Tượng: Độ trễ API kiểm tra tồn kho tăng từ 8ms lên 14,000ms. Cổng tìm kiếm Golang bị cạn kiệt tài nguyên socket mạng, dẫn tới sập toàn bộ hệ thống tìm kiếm trên ứng dụng di động.
• Tổn Thất: $840,000 giá trị đơn hàng bị mất trong giờ mở bán cao điểm nhất của sự kiện.

sequenceDiagram
    autonumber
    actor Shoppers as "22,000 Khách Hàng Đồng Thời"
    participant Gateway as "Go Search Gateway (Eino)"
    participant StockSvc as "Dịch Vụ Kho Hàng (REST API)"
    participant Postgres as "PostgreSQL Cơ Sở Dữ Liệu Kho"

    Shoppers->>Gateway: Gửi truy vấn tìm kiếm (12,000 QPS)
    Gateway->>StockSvc: Gọi kiểm tra tồn kho thời gian thực
    Note over Postgres: Nhân viên vận hành chạy báo cáo khóa toàn bộ bảng!<br/>PostgreSQL bị nghẽn connection pool!
    StockSvc->>Postgres: SELECT * FROM inventory
    Note over StockSvc: Độ trễ API kho hàng vọt lên 14 giây!
    Note over Gateway: Gateway KHÔNG có circuit breaker!<br/>Không đặt timeout cho lệnh gọi công cụ!<br/>22,000 Goroutines bị treo giữ kết nối socket mạng!
    Gateway-->>Shoppers: HTTP 504 Gateway Timeout / Màn hình trắng xóa

00:00 UTC - Đêm hội 11.11 mở bán. Lưu lượng truy vấn nhảy vọt lên 12,400 QPS.
00:04 UTC - Một nhân viên vận hành chạy câu lệnh SQL phân tích tồn kho lớn trực tiếp trên DB kho hàng.
00:08 UTC - CPU cơ sở dữ liệu kho hàng chạm 100%; toàn bộ connection pool bị khóa chặt.
00:10 UTC - Độ trễ phản hồi của API kho hàng vọt từ 8ms lên 14,200ms.
00:12 UTC - Các worker tìm kiếm bị treo. Socket mạng và file descriptors trên máy chủ Gateway bị cạn kiệt.
00:15 UTC - Các pods tìm kiếm bắt đầu thất bại ở bước Kubernetes Readiness Probe; lưu lượng bị ngắt hoàn toàn.
00:30 UTC - Quản trị viên can thiệp tiêu diệt câu lệnh SQL báo cáo chậm.
00:45 UTC - Cơ sở dữ liệu kho phục hồi, nhưng các Gateway vẫn bị kẹt trong tình trạng deadlock kết nối mạng.
01:05 UTC - Thực hiện rolling restart toàn bộ 32 máy chủ Gateway tìm kiếm.
01:20 UTC - Hệ thống hoạt động bình thường trở lại. Độ trễ P99 ổn định ở mức 38ms.

Bài Học & Khắc Phục Triệt Để

1. Áp Dụng Context Timeout Bắt Buộc 25ms: Mọi cuộc gọi công cụ sang dịch vụ phụ trợ đều bị hủy ngay lập tức nếu vượt quá 25ms.
2. Tích Hợp Sony Gobreaker: Ngắt mạch tự động khi tỷ lệ lỗi vượt quá 20%, không bao giờ để sự cố của dịch vụ kho hàng ảnh hưởng đến tính khả dụng của thanh tìm kiếm.
3. Thoái Lui Đọc Từ Redis Cache: Khi circuit breaker mở, hệ thống tự động chuyển sang đọc dữ liệu tồn kho gần nhất lưu trong Redis thay vì ném lỗi về cho khách hàng.

❓ Câu Hỏi Thường Gặp (FAQ)