【AIaaS 全栈架构师】第 20 篇：推理部署实战——用 Go 构建 vLLM 推理服务客户端

本系列定位：面向全栈架构师的 AIaaS 系统化教程。本篇为模块三"模型推理与部署优化"的第七篇（收官篇），把前面学到的推理引擎理论落地到 Go 工程实践。技术栈以 Go 为主。

本篇你将学到

  • 掌握 Go 调用 vLLM OpenAI 兼容 API 的完整实现（/v1/chat/completions）
  • 理解流式响应 SSE 协议，并用 bufio.Scanner 正确解析 data: 行
  • 学会用 gRPC 调用推理服务，处理双向流式推理
  • 实现健康检查、指数退避重试、熔断保护
  • 掌握连接池管理、超时控制、上下文传递等生产级技巧

学完本篇，你将拥有一个可直接用于生产环境的 Go 推理客户端库，能集成到任何 AIaaS 平台的应用层。

一、从理论到工程：为什么用 Go 写推理客户端

1.1 模块三的回顾

模块三我们走过了完整的推理优化理论：

  • 第 14 篇：推理引擎全景（vLLM / TGI / TensorRT-LLM / Triton）
  • 第 15 篇：KV Cache 与 PagedAttention
  • 第 16 篇：Continuous Batching 与调度优化
  • 第 17 篇：量化技术（GPTQ / AWQ / GGUF）
  • 第 18 篇：Speculative Decoding
  • 第 19 篇：模型格式转换全链路

这些知识解决了"如何让推理引擎本身更快"。但用户和上层应用并不直接和引擎对话——他们调用的是 API。在 API 和引擎之间，需要一个客户端层。

1.2 为什么客户端层用 Go

在第 6 篇我们讨论过 Go 在 AI 基础设施中的战略定位。推理客户端用 Go 的理由：

优势 说明
并发模型轻量 goroutine 让高并发请求管理极简，一个网关轻松处理上万 QPS
标准库强大 net/http、bufio、context、sync 覆盖 90% 客户端需求
编译为单二进制 部署简单，无 Python 解释器和依赖地狱
内存占用低 相比 Python 客户端，Go 客户端常驻内存小一个数量级
gRPC 一等公民 protobuf + gRPC 在 Go 中是原生体验
生态成熟 Prometheus、OpenTelemetry、etcd 等基础设施都有 Go SDK

本篇所有代码仅依赖 Go 标准库（除 gRPC 章节需要 google.golang.org/grpc），不引入任何重型框架。

1.3 vLLM 的对外接口

vLLM 启动后默认监听 OpenAI 兼容的 HTTP API：

python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Llama-3-8B-Instruct \
    --port 8000

主要端点：

端点 方法 用途
/v1/chat/completions POST 对话补全（主流）
/v1/completions POST 文本补全（无对话模板）
/v1/embeddings POST 文本向量（需 Embedding 模型）
/v1/models GET 列出已加载模型
/health GET 健康检查

我们重点实现 /v1/chat/completions，它支持流式和非流式两种模式。

二、非流式调用：基础客户端

2.1 数据结构定义

先定义与 OpenAI API 对齐的请求和响应结构：

package vllmclient

// ChatMessage 表示一条对话消息
type ChatMessage struct {
    Role    string `json:"role"`              // "system" / "user" / "assistant"
    Content string `json:"content"`
}

// ChatRequest 对应 POST /v1/chat/completions 的请求体
type ChatRequest struct {
    Model          string        `json:"model"`
    Messages       []ChatMessage `json:"messages"`
    Temperature    float64       `json:"temperature,omitempty"`
    TopP           float64       `json:"top_p,omitempty"`
    MaxTokens      int           `json:"max_tokens,omitempty"`
    Stream         bool          `json:"stream,omitempty"`
    Stop           []string      `json:"stop,omitempty"`
    FrequencyPenalty float64     `json:"frequency_penalty,omitempty"`
    PresencePenalty  float64     `json:"presence_penalty,omitempty"`
    // vLLM 扩展字段
    Seed           *int          `json:"seed,omitempty"`
}

// ChatChoice 表示一个候选输出
type ChatChoice struct {
    Index        int         `json:"index"`
    Message      ChatMessage `json:"message"`
    FinishReason string      `json:"finish_reason"` // "stop" / "length" / "tool_calls"
}

// Usage 表示 Token 计量（第 23 篇会深入讨论）
type Usage struct {
    PromptTokens     int `json:"prompt_tokens"`
    CompletionTokens int `json:"completion_tokens"`
    TotalTokens      int `json:"total_tokens"`
}

// ChatResponse 对应响应体
type ChatResponse struct {
    ID      string        `json:"id"`
    Object  string        `json:"object"`    // "chat.completion"
    Created int64         `json:"created"`   // Unix 时间戳
    Model   string        `json:"model"`
    Choices []ChatChoice  `json:"choices"`
    Usage   Usage         `json:"usage"`
}

2.2 客户端结构体

import (
    "context"
    "fmt"
    "net/http"
    "time"
)

// Client 是 vLLM 推理服务的客户端
type Client struct {
    baseURL    string          // 如 "http://localhost:8000"
    apiKey     string          // Bearer Token（vLLM 默认不校验，但网关层可能需要）
    httpClient *http.Client    // 复用连接池
    timeout    time.Duration   // 总超时
}

// Option 函数式选项模式
type Option func(*Client)

func WithAPIKey(key string) Option {
    return func(c *Client) { c.apiKey = key }
}

func WithTimeout(d time.Duration) Option {
    return func(c *Client) { c.timeout = d }
}

func WithHTTPClient(h *http.Client) Option {
    return func(c *Client) { c.httpClient = h }
}

// NewClient 创建客户端
func NewClient(baseURL string, opts ...Option) *Client {
    c := &Client{
        baseURL: baseURL,
        timeout: 60 * time.Second,
    }
    for _, opt := range opts {
        opt(c)
    }
    // 未显式提供 httpClient,创建默认的(带连接池)
    if c.httpClient == nil {
        c.httpClient = defaultHTTPClient(c.timeout)
    }
    return c
}

2.3 连接池配置

这是生产环境最容易被忽视的细节。默认 http.Transport 的连接池参数偏保守，高并发下会成为瓶颈：

import "crypto/tls"

func defaultHTTPClient(timeout time.Duration) *http.Client {
    transport := &http.Transport{
        // 连接池大小:每个 host 最多保持 100 个空闲连接
        MaxIdleConns:        100,
        MaxIdleConnsPerHost: 100,  // 关键!默认只有 2
        IdleConnTimeout:     90 * time.Second,
        // 握手超时
        TLSHandshakeTimeout: 10 * time.Second,
        // 响应头超时(防止慢攻击)
        ResponseHeaderTimeout: 30 * time.Second,
        // 启用 HTTP/2(vLLM 支持)
        ForceAttemptHTTP2: true,
        TLSClientConfig: &tls.Config{
            InsecureSkipVerify: false,
        },
    }
    return &http.Client{
        Transport: transport,
        Timeout:   timeout,
    }
}

为什么 MaxIdleConnsPerHost 默认只有 2？ 这是 Go 标准库为了通用性的保守默认值。对推理客户端这种"几乎只调用一个后端"的场景，2 个连接远远不够——高并发时会不断新建连接，性能急剧下降。

2.4 非流式 Chat 方法

import (
    "bytes"
    "encoding/json"
)

// Chat 发起一次非流式对话补全
func (c *Client) Chat(ctx context.Context, req *ChatRequest) (*ChatResponse, error) {
    // 1. 强制非流式
    req.Stream = false

    // 2. 序列化请求体
    body, err := json.Marshal(req)
    if err != nil {
        return nil, fmt.Errorf("marshal request: %w", err)
    }

    // 3. 构造 HTTP 请求
    httpReq, err := http.NewRequestWithContext(ctx,
        http.MethodPost, c.baseURL+"/v1/chat/completions", bytes.NewReader(body))
    if err != nil {
        return nil, fmt.Errorf("new request: %w", err)
    }
    httpReq.Header.Set("Content-Type", "application/json")
    if c.apiKey != "" {
        httpReq.Header.Set("Authorization", "Bearer "+c.apiKey)
    }

    // 4. 发送请求
    resp, err := c.httpClient.Do(httpReq)
    if err != nil {
        return nil, fmt.Errorf("http do: %w", err)
    }
    defer resp.Body.Close()

    // 5. 检查状态码
    if resp.StatusCode != http.StatusOK {
        return nil, parseError(resp)
    }

    // 6. 解析响应
    var chatResp ChatResponse
    if err := json.NewDecoder(resp.Body).Decode(&chatResp); err != nil {
        return nil, fmt.Errorf("decode response: %w", err)
    }
    return &chatResp, nil
}

// APIError 表示 vLLM 返回的错误
type APIError struct {
    StatusCode int
    Message    string
    Type       string
}

func (e *APIError) Error() string {
    return fmt.Sprintf("vllm api error: %d %s", e.StatusCode, e.Message)
}

func parseError(resp *http.Response) error {
    var errBody struct {
        Error struct {
            Message string `json:"message"`
            Type    string `json:"type"`
        } `json:"error"`
    }
    _ = json.NewDecoder(resp.Body).Decode(&errBody)
    return &APIError{
        StatusCode: resp.StatusCode,
        Message:    errBody.Error.Message,
        Type:       errBody.Error.Type,
    }
}

2.5 使用示例

func main() {
    client := vllmclient.NewClient("http://localhost:8000",
        vllmclient.WithTimeout(30*time.Second),
    )

    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    resp, err := client.Chat(ctx, &vllmclient.ChatRequest{
        Model: "meta-llama/Llama-3-8B-Instruct",
        Messages: []vllmclient.ChatMessage{
            {Role: "system", Content: "你是一个简洁的助手。"},
            {Role: "user", Content: "用一句话解释什么是 KV Cache。"},
        },
        Temperature: 0.7,
        MaxTokens:   100,
    })
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("回答:", resp.Choices[0].Message.Content)
    fmt.Printf("Token 用量: prompt=%d, completion=%d\n",
        resp.Usage.PromptTokens, resp.Usage.CompletionTokens)
}

三、流式响应：SSE 协议解析

3.1 为什么需要流式

非流式调用的问题：用户要等几十秒才能看到第一个字。对于生成 500 Token 的回答，TTFT（Time To First Token，首 Token 延迟）如果走非流式，等于总生成时间——体验极差。

流式调用让 Token 一个个返回，用户几乎瞬间看到输出开始：

vLLM Go 客户端 用户/前端 vLLM Go 客户端 用户/前端 生成完成 发起对话请求(stream=true) POST /v1/chat/completions (stream=true) data: {chunk 1, "你"} 转发 "你" data: {chunk 2, "好"} 转发 "好" data: {chunk 3, "!"} 转发 "!" data: [DONE] 流结束

3.2 SSE 协议格式

Server-Sent Events 是流式的传输协议。vLLM 的响应格式：

data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"你"}}],...}

data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"好"}}],...}

data: [DONE]

关键点：

  • 每条消息以 data: 开头
  • 消息之间用两个换行（\n\n）分隔
  • 流结束标志是 data: [DONE]
  • 每个 chunk 是一个 JSON 对象，其中 choices[0].delta.content 是增量 Token

3.3 流式数据结构

// ChatStreamChunk 表示流式响应的一个 chunk
type ChatStreamChunk struct {
    ID      string             `json:"id"`
    Object  string             `json:"object"`  // "chat.completion.chunk"
    Created int64              `json:"created"`
    Model   string             `json:"model"`
    Choices []ChatStreamChoice `json:"choices"`
}

type ChatStreamChoice struct {
    Index        int           `json:"index"`
    Delta        ChatDelta     `json:"delta"`
    FinishReason *string       `json:"finish_reason"` // 流末才非 nil
}

type ChatDelta struct {
    Role    string `json:"role,omitempty"`    // 第一个 chunk 才有
    Content string `json:"content,omitempty"` // 增量 Token
}

3.4 流式方法实现

这里用回调模式（callback），简单且内存友好：

// StreamHandler 是处理每个 chunk 的回调
// 接收 delta 内容,返回 error 可以中断流
type StreamHandler func(chunk *ChatStreamChunk) error

// ChatStream 发起流式对话补全
func (c *Client) ChatStream(ctx context.Context, req *ChatRequest, handler StreamHandler) error {
    req.Stream = true

    body, err := json.Marshal(req)
    if err != nil {
        return fmt.Errorf("marshal request: %w", err)
    }

    httpReq, err := http.NewRequestWithContext(ctx,
        http.MethodPost, c.baseURL+"/v1/chat/completions", bytes.NewReader(body))
    if err != nil {
        return fmt.Errorf("new request: %w", err)
    }
    httpReq.Header.Set("Content-Type", "application/json")
    httpReq.Header.Set("Accept", "text/event-stream")  // 声明接收 SSE
    if c.apiKey != "" {
        httpReq.Header.Set("Authorization", "Bearer "+c.apiKey)
    }

    resp, err := c.httpClient.Do(httpReq)
    if err != nil {
        return fmt.Errorf("http do: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return parseError(resp)
    }

    // 用 bufio.Scanner 逐行读取
    scanner := newSSEScanner(resp.Body)
    for scanner.Scan() {
        select {
        case <-ctx.Done():
            return ctx.Err()  // 上下文取消
        default:
        }

        line := scanner.Text()
        // SSE 行格式: "data: {...}"
        if len(line) < 6 || line[:6] != "data: " {
            continue  // 跳过空行和注释行
        }
        data := line[6:]

        // 流结束标志
        if data == "[DONE]" {
            return nil
        }

        // 解析 chunk JSON
        var chunk ChatStreamChunk
        if err := json.Unmarshal([]byte(data), &chunk); err != nil {
            continue  // 容忍单个 chunk 解析失败
        }

        // 回调
        if err := handler(&chunk); err != nil {
            return err  // 回调要求中断
        }
    }
    return scanner.Err()
}

3.5 bufio.Scanner 的缓冲区陷阱

这是流式实现的最大坑。默认 bufio.Scanner 的缓冲区上限是 64KB。vLLM 的单条 SSE 消息通常很小，但某些情况下（logprobs、长停止词、首 chunk 含完整 system message）可能超过 64KB，导致 Scanner 报 bufio.ErrTooLong。

解决方案：增大缓冲区：

import "bufio"

func newSSEScanner(r io.Reader) *bufio.Scanner {
    scanner := bufio.NewScanner(r)
    // 单行最大 1MB(防止超长 chunk)
    buf := make([]byte, 0, 64*1024)
    scanner.Buffer(buf, 1024*1024)
    // 按行分割(默认即可,但显式指定更清晰)
    scanner.Split(bufio.ScanLines)
    return scanner
}

3.6 使用示例

func main() {
    client := vllmclient.NewClient("http://localhost:8000")

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    var fullText strings.Builder
    err := client.ChatStream(ctx, &vllmclient.ChatRequest{
        Model: "meta-llama/Llama-3-8B-Instruct",
        Messages: []vllmclient.ChatMessage{
            {Role: "user", Content: "写一首关于秋天的五言绝句。"},
        },
        Temperature: 0.8,
        MaxTokens:   200,
    }, func(chunk *vllmclient.ChatStreamChunk) error {
        // 每个 delta 推送一次到前端(假设有 webSocket)
        delta := chunk.Choices[0].Delta.Content
        fmt.Print(delta)  // 实时打印
        fullText.WriteString(delta)
        return nil
    })
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println()
    fmt.Println("完整文本:", fullText.String())
}

3.7 channel 模式（替代回调）

回调模式简单，但有些场景（如 select 多路复用）更适合 channel：

// ChunkChan 返回一个 channel,持续推送 chunk
// 当流结束或出错时,channel 关闭
func (c *Client) ChatStreamChan(ctx context.Context, req *ChatRequest) (<-chan *ChatStreamChunk, error) {
    req.Stream = true

    body, _ := json.Marshal(req)
    httpReq, _ := http.NewRequestWithContext(ctx,
        http.MethodPost, c.baseURL+"/v1/chat/completions", bytes.NewReader(body))
    httpReq.Header.Set("Content-Type", "application/json")
    httpReq.Header.Set("Accept", "text/event-stream")

    resp, err := c.httpClient.Do(httpReq)
    if err != nil {
        return nil, err
    }
    if resp.StatusCode != http.StatusOK {
        defer resp.Body.Close()
        return nil, parseError(resp)
    }

    ch := make(chan *ChatStreamChunk, 16)  // 带缓冲
    go func() {
        defer close(ch)
        defer resp.Body.Close()

        scanner := newSSEScanner(resp.Body)
        for scanner.Scan() {
            line := scanner.Text()
            if len(line) < 6 || line[:6] != "data: " {
                continue
            }
            data := line[6:]
            if data == "[DONE]" {
                return
            }
            var chunk ChatStreamChunk
            if err := json.Unmarshal([]byte(data), &chunk); err != nil {
                continue
            }
            select {
            case ch <- &chunk:
            case <-ctx.Done():
                return
            }
        }
    }()
    return ch, nil
}

使用：

ch, err := client.ChatStreamChan(ctx, req)
if err != nil { log.Fatal(err) }
for chunk := range ch {
    fmt.Print(chunk.Choices[0].Delta.Content)
}

四、健康检查与服务发现

4.1 vLLM 健康检查端点

vLLM 提供了 /health 端点：

  • 模型未加载：返回 503 Service Unavailable
  • 模型已加载、就绪：返回 200 OK
// HealthCheck 检查 vLLM 服务是否就绪
func (c *Client) HealthCheck(ctx context.Context) error {
    httpReq, err := http.NewRequestWithContext(ctx,
        http.MethodGet, c.baseURL+"/health", nil)
    if err != nil {
        return err
    }

    resp, err := c.httpClient.Do(httpReq)
    if err != nil {
        return fmt.Errorf("health check: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("health check: status %d", resp.StatusCode)
    }
    return nil
}

4.2 模型就绪检查

/health 只表明服务存活，不代表模型已加载完成。vLLM 加载大模型可能耗时几分钟。更可靠的就绪检查是查询 /v1/models：

// ModelInfo 表示一个已加载模型
type ModelInfo struct {
    ID      string `json:"id"`
    Object  string `json:"object"`
    Created int64  `json:"created"`
    OwnedBy string `json:"owned_by"`
}

// IsModelReady 检查指定模型是否已加载完成
func (c *Client) IsModelReady(ctx context.Context, modelID string) error {
    httpReq, _ := http.NewRequestWithContext(ctx,
        http.MethodGet, c.baseURL+"/v1/models", nil)

    resp, err := c.httpClient.Do(httpReq)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    if resp.StatusCode != http.StatusOK {
        return fmt.Errorf("list models: status %d", resp.StatusCode)
    }

    var list struct {
        Data []ModelInfo `json:"data"`
    }
    if err := json.NewDecoder(resp.Body).Decode(&list); err != nil {
        return err
    }
    for _, m := range list.Data {
        if m.ID == modelID {
            return nil  // 找到模型
        }
    }
    return fmt.Errorf("model %s not loaded", modelID)
}

// WaitForModel 轮询等待模型就绪
func (c *Client) WaitForModel(ctx context.Context, modelID string, timeout time.Duration) error {
    deadline := time.Now().Add(timeout)
    ticker := time.NewTicker(2 * time.Second)
    defer ticker.Stop()

    for {
        if time.Now().After(deadline) {
            return fmt.Errorf("timeout waiting for model %s", modelID)
        }
        if err := c.IsModelReady(ctx, modelID); err == nil {
            return nil
        }
        select {
        case <-ticker.C:
            continue
        case <-ctx.Done():
            return ctx.Err()
        }
    }
}

五、重试与熔断

5.1 指数退避重试

推理服务在高负载时会偶发 503（队列满）或网络抖动。简单重试机制：

// RetryConfig 重试配置
type RetryConfig struct {
    MaxAttempts  int           // 最大尝试次数(含首次)
    InitialDelay time.Duration // 首次重试延迟
    MaxDelay     time.Duration // 最大延迟
    Multiplier   float64       // 退避倍数
}

func DefaultRetryConfig() *RetryConfig {
    return &RetryConfig{
        MaxAttempts:  3,
        InitialDelay: 500 * time.Millisecond,
        MaxDelay:     5 * time.Second,
        Multiplier:   2.0,
    }
}

// shouldRetry 判断错误是否可重试
func shouldRetry(err error) bool {
    var apiErr *APIError
    if errors.As(err, &apiErr) {
        // 5xx 和 429 可重试
        if apiErr.StatusCode >= 500 || apiErr.StatusCode == 429 {
            return true
        }
    }
    // 网络错误可重试
    var netErr net.Error
    return errors.As(err, &netErr)
}

// ChatWithRetry 带重试的 Chat
func (c *Client) ChatWithRetry(ctx context.Context, req *ChatRequest, cfg *RetryConfig) (*ChatResponse, error) {
    if cfg == nil {
        cfg = DefaultRetryConfig()
    }

    var lastErr error
    delay := cfg.InitialDelay

    for attempt := 1; attempt <= cfg.MaxAttempts; attempt++ {
        resp, err := c.Chat(ctx, req)
        if err == nil {
            return resp, nil
        }
        lastErr = err

        if !shouldRetry(err) {
            return nil, err  // 不可重试错误立即返回
        }
        if attempt == cfg.MaxAttempts {
            break
        }

        // 指数退避等待
        timer := time.NewTimer(delay)
        select {
        case <-timer.C:
        case <-ctx.Done():
            timer.Stop()
            return nil, ctx.Err()
        }
        timer.Stop()

        // 计算下次延迟
        delay = time.Duration(float64(delay) * cfg.Multiplier)
        if delay > cfg.MaxDelay {
            delay = cfg.MaxDelay
        }
    }
    return nil, fmt.Errorf("after %d attempts: %w", cfg.MaxAttempts, lastErr)
}

5.2 流式重试的特殊性

流式调用不能简单重试——如果已经向用户发送了部分 Token，重试会导致内容重复或跳变。处理策略：

  1. 首 Token 前可重试：还没向下游发送任何内容时，重试无副作用
  2. 首 Token 后不重试：直接返回错误，由上层决定是否告知用户重试
// ChatStreamWithRetry 仅在首 Token 前重试
func (c *Client) ChatStreamWithRetry(ctx context.Context, req *ChatRequest,
    handler StreamHandler, cfg *RetryConfig) error {

    if cfg == nil {
        cfg = DefaultRetryConfig()
    }
    delay := cfg.InitialDelay
    var lastErr error

    for attempt := 1; attempt <= cfg.MaxAttempts; attempt++ {
        firstChunkSent := false
        err := c.ChatStream(ctx, req, func(chunk *ChatStreamChunk) error {
            // 首个有内容的 chunk 标记,之后不再可重试
            if chunk.Choices[0].Delta.Content != "" {
                firstChunkSent = true
            }
            return handler(chunk)
        })

        if err == nil {
            return nil
        }
        lastErr = err

        // 一旦发送过 chunk,不可重试
        if firstChunkSent {
            return err
        }
        if !shouldRetry(err) || attempt == cfg.MaxAttempts {
            return err
        }

        // 退避等待
        select {
        case <-time.After(delay):
        case <-ctx.Done():
            return ctx.Err()
        }
        delay = time.Duration(float64(delay) * cfg.Multiplier)
        if delay > cfg.MaxDelay {
            delay = cfg.MaxDelay
        }
    }
    return lastErr
}

六、gRPC 推理客户端

6.1 为什么需要 gRPC

HTTP/JSON 的优点是通用，但性能不是最优。在生产 AIaaS 平台中，内部服务间通信用 gRPC 更合适：

维度 HTTP/JSON gRPC/protobuf
序列化 JSON 文本解析（慢） protobuf 二进制（快 3-10 倍）
传输 HTTP/1.1（多数情况） HTTP/2 多路复用
流式 SSE 文本流 原生双向流
Schema 弱约定 强类型 .proto
跨语言 一般 一流

vLLM 本身不直接暴露 gRPC，但 AIaaS 平台通常会在 vLLM 之上包一层 gRPC 网关（用 Go 写），对外提供 gRPC 接口。这部分代码展示如何写客户端调用这样的网关。

6.2 protobuf 定义

先定义推理服务的 proto：

// inference.proto
syntax = "proto3";

package inference.v1;

option go_package = "aiacs/inference/v1;inferencev1";

service InferenceService {
  // 流式对话补全
  rpc ChatStream(ChatRequest) returns (stream ChatResponseChunk) {}
  // 非流式对话补全
  rpc Chat(ChatRequest) returns (ChatResponse) {}
}

message ChatMessage {
  string role = 1;
  string content = 2;
}

message ChatRequest {
  string model = 1;
  repeated ChatMessage messages = 2;
  float temperature = 3;
  int32 max_tokens = 4;
  bool stream = 5;
}

message ChatResponseChunk {
  string content = 1;        // 增量内容
  string finish_reason = 2;  // 流末才有
  Usage usage = 3;
}

message ChatResponse {
  repeated ChatMessage choices = 1;
  Usage usage = 2;
}

message Usage {
  int32 prompt_tokens = 1;
  int32 completion_tokens = 2;
}

生成 Go 代码：

protoc --go_out=. --go_opt=paths=source_relative \
    --go-grpc_out=. --go-grpc_opt=paths=source_relative \
    inference.proto

6.3 gRPC 客户端实现

package vllmclient

import (
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    pb "aiacs/inference/v1"
)

// GRPCClient 是 gRPC 推理客户端
type GRPCClient struct {
    conn   *grpc.ClientConn
    client pb.InferenceServiceClient
}

// NewGRPCClient 创建 gRPC 客户端
func NewGRPCClient(target string) (*GRPCClient, error) {
    // 生产环境应使用 credentials.NewTLS(...) 而非 insecure
    conn, err := grpc.Dial(target,
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        grpc.WithDefaultCallOptions(
            grpc.MaxCallRecvMsgSize(10*1024*1024),  // 10MB
        ),
    )
    if err != nil {
        return nil, fmt.Errorf("grpc dial: %w", err)
    }
    return &GRPCClient{
        conn:   conn,
        client: pb.NewInferenceServiceClient(conn),
    }, nil
}

func (g *GRPCClient) Close() error {
    return g.conn.Close()
}

// Chat 非流式
func (g *GRPCClient) Chat(ctx context.Context, req *pb.ChatRequest) (*pb.ChatResponse, error) {
    req.Stream = false
    return g.client.Chat(ctx, req)
}

// ChatStream 流式
// 返回一个 chunk channel,流结束时关闭
func (g *GRPCClient) ChatStream(ctx context.Context, req *pb.ChatRequest) (<-chan *pb.ChatResponseChunk, error) {
    req.Stream = true
    stream, err := g.client.ChatStream(ctx, req)
    if err != nil {
        return nil, err
    }
    ch := make(chan *pb.ChatResponseChunk, 16)
    go func() {
        defer close(ch)
        for {
            chunk, err := stream.Recv()
            if err != nil {
                if err == io.EOF {
                    return
                }
                // 传递错误(生产环境可用专用 error channel)
                log.Printf("grpc stream recv error: %v", err)
                return
            }
            select {
            case ch <- chunk:
            case <-ctx.Done():
                return
            }
        }
    }()
    return ch, nil
}

6.4 gRPC 使用示例

func main() {
    client, err := vllmclient.NewGRPCClient("inference-gateway:50051")
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    // 流式
    stream, err := client.ChatStream(ctx, &pb.ChatRequest{
        Model: "llama-3-8b",
        Messages: []*pb.ChatMessage{
            {Role: "user", Content: "讲个短笑话"},
        },
        Temperature: 0.9,
        MaxTokens:   100,
    })
    if err != nil {
        log.Fatal(err)
    }
    for chunk := range stream {
        fmt.Print(chunk.Content)
    }
}

七、客户端架构总览

推理后端

Go 客户端库

应用层

业务应用
(网关/对话助手等)

HTTP Client
(OpenAI 兼容)

gRPC Client
(protobuf)

连接池管理

重试 + 熔断

健康检查

vLLM
OpenAI API

gRPC 推理网关
(Go 实现)

关键设计原则：

  1. HTTP 与 gRPC 共享连接池和重试逻辑：通过接口抽象统一
  2. 健康检查独立于推理调用：定期后台检查，结果缓存
  3. 上下文全链路传递：所有方法首参都是 context.Context，支持超时和取消

八、并发请求与批量处理

8.1 并发发送多个请求

Go 的 goroutine 让并发调用推理服务极其简单：

// BatchChat 并发发送多个请求
// 返回结果顺序与输入一致
func (c *Client) BatchChat(ctx context.Context, reqs []*ChatRequest) ([]*ChatResponse, []error) {
    n := len(reqs)
    results := make([]*ChatResponse, n)
    errs := make([]error, n)

    var wg sync.WaitGroup
    wg.Add(n)

    // 用 semaphore 控制并发度(防止打爆后端)
    sem := make(chan struct{}, 10)  // 最大并发 10

    for i, req := range reqs {
        go func(idx int, r *ChatRequest) {
            defer wg.Done()
            sem <- struct{}{}
            defer func() { <-sem }()

            resp, err := c.Chat(ctx, r)
            results[idx] = resp
            errs[idx] = err
        }(i, req)
    }
    wg.Wait()
    return results, errs
}

8.2 注意：vLLM 的 Continuous Batching

我们在第 16 篇讨论过，vLLM 内部用 Continuous Batching 把多个并发请求合并到一个 batch 中处理。这意味着客户端并发发送的请求会自动获得 batching 加速。

但要注意：

  • 并发度不是越高越好：vLLM 有队列长度限制，超过会 429 拒绝
  • 流式请求的并发：每个流式请求会占用一个调度槽，过多流式请求会让 vLLM 调度负担加重
  • 合理并发度：通常是 GPU 数量 × 4~8，可通过压测确定

九、可观测性：指标与日志

9.1 关键指标

生产客户端必须暴露指标，便于监控和排障：

import (
    "sync/atomic"
    "time"
)

// Metrics 客户端运行指标
type Metrics struct {
    RequestCount    atomic.Int64  // 总请求数
    SuccessCount    atomic.Int64  // 成功数
    ErrorCount      atomic.Int64  // 错误数
    RetryCount      atomic.Int64  // 重试次数
    TotalTokensIn   atomic.Int64  // 输入 Token 总数
    TotalTokensOut  atomic.Int64  // 输出 Token 总数
    TotalLatencyMs  atomic.Int64  // 总延迟(毫秒)
}

// instrumentedClient 包装 Client,添加指标采集
type instrumentedClient struct {
    inner   *Client
    metrics *Metrics
}

func (c *instrumentedClient) Chat(ctx context.Context, req *ChatRequest) (*ChatResponse, error) {
    start := time.Now()
    c.metrics.RequestCount.Add(1)

    resp, err := c.inner.Chat(ctx, req)
    latency := time.Since(start).Milliseconds()
    c.metrics.TotalLatencyMs.Add(latency)

    if err != nil {
        c.metrics.ErrorCount.Add(1)
    } else {
        c.metrics.SuccessCount.Add(1)
        c.metrics.TotalTokensIn.Add(int64(resp.Usage.PromptTokens))
        c.metrics.TotalTokensOut.Add(int64(resp.Usage.CompletionTokens))
    }
    return resp, err
}

// TTFT (Time To First Token) 是流式场景的关键指标
// 在 ChatStream 回调中记录首个 chunk 的时间
func (c *instrumentedClient) ChatStream(ctx context.Context, req *ChatRequest, handler StreamHandler) error {
    start := time.Now()
    ttft := time.Duration(0)

    wrappedHandler := func(chunk *ChatStreamChunk) error {
        if ttft == 0 && chunk.Choices[0].Delta.Content != "" {
            ttft = time.Since(start)
            // 上报到 Prometheus
            // observeTTFT(ttft)
        }
        return handler(chunk)
    }
    return c.inner.ChatStream(ctx, req, wrappedHandler)
}

9.2 关键 SLO 指标

指标 含义 健康阈值
TTFT P95 首 Token 延迟 95 分位 < 500ms
TPOT 平均每 Token 延迟 < 50ms
错误率 非 2xx 响应占比 < 1%
重试率 触发重试的请求占比 < 5%
连接池利用率 活跃连接 / 最大连接 50-80%

十、完整代码与单元测试

10.1 测试用 mock server

用 Go 标准库写一个 vLLM mock，便于单元测试：

// mock_vllm_test.go
package vllmclient

import (
    "fmt"
    "net/http"
    "net/http/httptest"
    "strings"
    "testing"
)

func newMockServer() *httptest.Server {
    return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        switch r.URL.Path {
        case "/health":
            w.WriteHeader(http.StatusOK)
            return
        case "/v1/chat/completions":
            // 判断是否流式
            var req ChatRequest
            _ = json.NewDecoder(r.Body).Decode(&req)

            if req.Stream {
                w.Header().Set("Content-Type", "text/event-stream")
                flusher, _ := w.(http.Flusher)
                // 模拟流式输出 3 个 chunk
                for _, s := range []string{"你好", "世界", "!"} {
                    chunk := ChatStreamChunk{
                        ID: "test",
                        Choices: []ChatStreamChoice{{
                            Index: 0,
                            Delta: ChatDelta{Content: s},
                        }},
                    }
                    data, _ := json.Marshal(chunk)
                    fmt.Fprintf(w, "data: %s\n\n", data)
                    flusher.Flush()
                }
                fmt.Fprintf(w, "data: [DONE]\n\n")
                flusher.Flush()
            } else {
                resp := ChatResponse{
                    ID: "test",
                    Choices: []ChatChoice{{
                        Index: 0,
                        Message: ChatMessage{Role: "assistant", Content: "你好世界"},
                    }},
                    Usage: Usage{PromptTokens: 5, CompletionTokens: 3, TotalTokens: 8},
                }
                json.NewEncoder(w).Encode(resp)
            }
        default:
            w.WriteHeader(http.StatusNotFound)
        }
    }))
}

func TestChat(t *testing.T) {
    server := newMockServer()
    defer server.Close()

    client := NewClient(server.URL)
    resp, err := client.Chat(context.Background(), &ChatRequest{
        Model:    "test",
        Messages: []ChatMessage{{Role: "user", Content: "hi"}},
    })
    if err != nil {
        t.Fatal(err)
    }
    if resp.Choices[0].Message.Content != "你好世界" {
        t.Errorf("unexpected content: %s", resp.Choices[0].Message.Content)
    }
}

func TestChatStream(t *testing.T) {
    server := newMockServer()
    defer server.Close()

    client := NewClient(server.URL)
    var sb strings.Builder
    err := client.ChatStream(context.Background(), &ChatRequest{
        Model:    "test",
        Messages: []ChatMessage{{Role: "user", Content: "hi"}},
        Stream:   true,
    }, func(chunk *ChatStreamChunk) error {
        sb.WriteString(chunk.Choices[0].Delta.Content)
        return nil
    })
    if err != nil {
        t.Fatal(err)
    }
    if sb.String() != "你好世界!" {
        t.Errorf("unexpected stream: %q", sb.String())
    }
}

10.2 运行测试

go test -v -race ./...

-race 检测竞态条件，对并发代码尤其重要。

十一、生产部署 checklist

把客户端推向生产前，确认以下清单：

项 说明 本篇章节
连接池配置 MaxIdleConnsPerHost 调到 100+ 二章 2.3
超时设置 总超时 + 握手超时 + 响应头超时 二章 2.3
重试机制 指数退避,区分可重试错误 五章
健康检查 启动时 + 周期性 四章
流式缓冲区 Scanner buffer ≥ 1MB 三章 3.5
上下文传递 所有调用支持 context.Context 全篇
指标采集 QPS / 延迟 / TTFT / 错误率 九章
单元测试 含 mock server 的集成测试 十章
优雅关闭 客户端支持 Close(),等待进行中请求 gRPC 部分
熔断保护 连续失败时短路(本篇未实现,推荐 sony/gobreaker) -

十二、本篇代码与模块三收官

12.1 代码组织

本篇完整代码结构（实际项目）：

vllm-client/
├── client.go              # HTTP 客户端 + Chat/ChatStream
├── client_grpc.go         # gRPC 客户端
├── retry.go               # 重试机制
├── health.go              # 健康检查
├── metrics.go             # 指标采集
├── types.go               # 数据结构定义
├── errors.go              # 错误类型
├── inference.proto        # gRPC protobuf 定义
└── vllmclient_test.go     # 单元测试

12.2 模块三全貌回顾

至此，模块三"模型推理与部署优化"七篇文章全部完成：

篇目 主题 核心收获
第 14 篇 推理引擎全景 vLLM / TGI / TensorRT-LLM / Triton 的定位与差异
第 15 篇 KV Cache 与 PagedAttention 注意力的显存瓶颈与分页管理
第 16 篇 Continuous Batching 动态批处理让 GPU 满载
第 17 篇 量化技术 GPTQ / AWQ / GGUF 的精度效率权衡
第 18 篇 Speculative Decoding 投机解码用并行验证突破串行瓶颈
第 19 篇 模型格式转换 Safetensors / ONNX / GGUF / TRT Engine 的选型
第 20 篇 Go 推理客户端 把理论落地为生产级工程代码

从引擎内部原理到外部工程实践，模块三覆盖了推理优化的完整闭环。

本篇小结

知识点 核心内容
客户端层定位 应用层与推理引擎之间的中间件，处理协议适配、重试、连接池等
Go 选型理由 高并发、低内存、强标准库、gRPC 一等公民
OpenAI 兼容 API /v1/chat/completions 支持 stream 字段切换流式
SSE 协议 data: {...}\n\n 格式，[DONE] 结束，用 bufio.Scanner 解析
Scanner 缓冲陷阱 默认 64KB 上限，需手动调到 1MB 防止 ErrTooLong
连接池 MaxIdleConnsPerHost 必须显式调大（默认 2 不够）
重试机制 指数退避，区分可重试（5xx/429/网络错误）与不可重试（4xx）
流式重试 仅首 Token 前可重试，已发送 chunk 后不重试
健康检查 /health 查存活，/v1/models 查模型就绪
gRPC 推理 protobuf 二进制 + HTTP/2 多路复用，比 HTTP/JSON 高效
关键指标 TTFT（首 Token 延迟）、TPOT（每 Token 延迟）、错误率、重试率
Continuous Batching 协同 客户端并发请求会自动被 vLLM 合批处理

下篇预告

第 21 篇：Go AIaaS 平台架构总览——控制面与数据面的分离设计

模块三收官，模块四"Go 与 AIaaS 平台工程"开启。我们已经分别学过推理引擎内部（模块三）和 Go 推理客户端（本篇），但要构建一个完整的 AIaaS 平台，需要把这些组件编排起来。第 21 篇将从架构总览开始，建立控制面（Control Plane）与数据面（Data Plane）分离的设计思维，并梳理 Go 在 AIaaS 各层的角色定位，为后续 API 网关、限流计量、gRPC 服务等组件搭建脚手架。

如果本篇内容对你有帮助，欢迎点赞收藏！有任何疑问，欢迎在评论区交流。

# golang # 人工智能
Logo
加入AMD AI开发者计划！

免费领 150 小时云算力，进群参与显卡、AI PC 幸运抽奖

更多推荐

cover

大模型推理引擎架构深度解析：vLLM、SGLang与TensorRT-LLM的技术路线对决

avatar AMD开发者中国社区

本地部署大模型，服务器还是工作站？先把显存这笔账算清楚（浪潮 / 联想实战选型）

从一手交付经验出发，用 30 秒显存口算法、三场景决策框架、GPU 量化与 vLLM 调优细节，算清本地部署大模型"服务器 vs 工作站"的真正分界线，附成本测算模板与四条避坑经验。

avatar AMD开发者中国社区
cover

告别云API！本地AI编程神器Qwen3.6-27B部署全攻略：24G显存流畅运行，支持图像视频理解

avatar AMD开发者中国社区