在Golang项目中使用vLLM 实现高并发LLM推理指南

package main

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"

    pb "your_vllm_protobuf_package" // 替换为你的vLLM protobuf包
)

const (
    address     = "localhost:50051" // 替换为你的vLLM服务器地址
    numWorkers  = 10                // worker数量
    requestQueueSize = 100          // 请求队列大小
)

type InferenceRequest struct {
    Prompt string
    Response chan string
}

func main() {
    // 1. 连接到gRPC服务器
    conn, err := grpc.Dial(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        log.Fatalf("did not connect: %v", err)
    }
    defer conn.Close()

    client := pb.NewVLLMServiceClient(conn)

    // 2. 创建请求队列
    requestQueue := make(chan InferenceRequest, requestQueueSize)

    // 3. 启动worker pool
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go worker(client, requestQueue, &wg)
    }

    // 4. 模拟发送推理请求
    go func() {
        for i := 0; i < 20; i++ {
            prompt := fmt.Sprintf("Translate to French: Hello, world! (%d)", i)
            responseChan := make(chan string, 1) // Buffered channel to prevent blocking
            requestQueue <- InferenceRequest{Prompt: prompt, Response: responseChan}

            // 异步接收结果，防止阻塞发送
            go func(i int, responseChan chan string) {
                select {
                case result := <-responseChan:
                    fmt.Printf("Request %d: %s\n", i, result)
                case <-time.After(5 * time.Second): // 超时处理
                    fmt.Printf("Request %d: Timeout\n", i)
                }
            }(i, responseChan)
            time.Sleep(100 * time.Millisecond) // 模拟请求间隔
        }
        close(requestQueue) // 关闭请求队列，通知worker退出
    }()

    // 5. 等待所有worker完成
    wg.Wait()
    fmt.Println("All requests processed.")
}

func worker(client pb.VLLMServiceClient, requestQueue <-chan InferenceRequest, wg *sync.WaitGroup) {
    defer wg.Done()

    for req := range requestQueue {
        // 6. 调用vLLM进行推理
        ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) // 超时控制
        defer cancel()

        r, err := client.Generate(ctx, &pb.GenerateRequest{Prompt: req.Prompt})
        if err != nil {
            log.Printf("could not generate: %v", err)
            req.Response <- "Error: " + err.Error() // 发送错误信息
            continue
        }
        req.Response <- r.GetText() // 发送推理结果
        close(req.Response) // 关闭channel
    }
    fmt.Println("Worker exiting.")
}