当前位置：首页 > news >正文

LLM 服务网关的性能优化：连接池、批量推理与请求合并的工程方案

news 2026/7/15 21:32:42

LLM 服务网关的性能优化：连接池、批量推理与请求合并的工程方案

一、$2300 一个月的 API 账单是怎么来的

一个面向开发者的 AI 编程助手，月均调用 GPT-4 API 约 120 万次。最初的实现是每个用户的每次请求直接新建一个 HTTP 连接调用 OpenAI API。这导致了两个问题：

成本爆炸：每个请求独立发送，大量短小的 prompt（补全 50 个 token 也需要携带完整的 system prompt）造成了严重的 Token 浪费
延迟飙升：TLS 握手的频繁重复建立（每次 200-300ms），高峰期短连接耗尽端口，连接等待时间超过推理时间

排查发现：120 万次请求中，38% 的请求体 Token 消耗在重复的 system prompt 上，15% 的端到端延迟花在连接建立上。解决方向：引入 LLM 服务网关，集中管理连接、请求和缓存。

二、LLM 服务网关的三层优化

flowchart TD A[客户端请求] --> B[网关接入层] B --> C{请求分类} C -->|高频短请求| D[请求合并队列] C -->|长推理请求| E[直通连接池] C -->|缓存命中| F[直接返回] D --> G[批量推理: 合并多个短请求] G --> H[LLM Provider] E --> I[连接池: 复用 TLS 连接] I --> H F --> J[Token 缓存] H --> K[结果拆分与分发] K --> L[返回各客户端]

2.1 连接池：消除 TLS 握手开销

// LLM 连接池：复用 HTTP 连接，减少 TLS 握手 type LLMConnectionPool struct { client *http.Client semaphore chan struct{} // 并发控制信号量 maxConns int } func NewLLMConnectionPool(maxConns int) *LLMConnectionPool { return &LLMConnectionPool{ client: &http.Client{ Transport: &http.Transport{ MaxIdleConns: maxConns, MaxIdleConnsPerHost: maxConns, IdleConnTimeout: 90 * time.Second, TLSHandshakeTimeout: 10 * time.Second, DisableKeepAlives: false, // 启用 Keep-Alive }, Timeout: 60 * time.Second, }, semaphore: make(chan struct{}, maxConns), maxConns: maxConns, } } func (p *LLMConnectionPool) Do(ctx context.Context, req *http.Request) (*http.Response, error) { // 并发控制：超过最大连接数时阻塞等待 select { case p.semaphore <- struct{}{}: defer func() { <-p.semaphore }() case <-ctx.Done(): return nil, ctx.Err() } resp, err := p.client.Do(req.WithContext(ctx)) if err != nil { return nil, fmt.Errorf("LLM API 调用失败: %w", err) } if resp.StatusCode == 429 { // 限流时读取 Retry-After 头并等待 retryAfter := resp.Header.Get("Retry-After") waitTime, _ := strconv.Atoi(retryAfter) if waitTime == 0 { waitTime = 5 // 默认等待 5 秒 } resp.Body.Close() select { case <-time.After(time.Duration(waitTime) * time.Second): return p.Do(ctx, req) // 重试 case <-ctx.Done(): return nil, ctx.Err() } } return resp, nil }

启用连接池后，TLS 握手只在连接建立时发生一次。实测 1000 QPS 场景下，P99 延迟从 850ms 降到 420ms，其中 430ms 的收益来自消除重复的 TLS 握手。

2.2 请求合并：批量推理降低 Token 浪费

// 请求合并器：将短请求批量发送给 LLM，减少 system prompt 重复 type RequestBatcher struct { batchSize int maxWaitTime time.Duration queue chan MergedRequest llmClient LLMClient } type MergedRequest struct { Prompts []string Results chan []string // 每个请求一个结果 ErrCh chan error } func (b *RequestBatcher) Submit(ctx context.Context, prompt string) (string, error) { // 如果请求很长（> 500 tokens），直接发送，不合入批量 if estimateTokens(prompt) > 500 { return b.llmClient.Complete(ctx, prompt) } resultCh := make(chan []string, 1) errCh := make(chan error, 1) select { case b.queue <- MergedRequest{Prompts: []string{prompt}, Results: resultCh, ErrCh: errCh}: case <-ctx.Done(): return "", ctx.Err() } select { case results := <-resultCh: return results[0], nil case err := <-errCh: return "", err case <-ctx.Done(): return "", ctx.Err() } } func (b *RequestBatcher) Run(ctx context.Context) { timer := time.NewTimer(b.maxWaitTime) defer timer.Stop() batch := make([]MergedRequest, 0, b.batchSize) for { timer.Reset(b.maxWaitTime) select { case req := <-b.queue: batch = append(batch, req) if len(batch) >= b.batchSize { b.processBatch(ctx, batch) batch = batch[:0] } case <-timer.C: if len(batch) > 0 { b.processBatch(ctx, batch) batch = batch[:0] } case <-ctx.Done(): if len(batch) > 0 { b.processBatch(context.Background(), batch) } return } } }

批量合并的核心收益不是"减少 API 调用次数"（API 调用次数相同），而是"一个请求携带多个子任务，system prompt 只计算一次 Token 费用"。在代码补全场景中，10 个用户的补全请求合并为一个 API 调用，system prompt 的 Token 费用降低到原来的 1/10。

2.3 响应拆分与分发

批量请求的响应需要正确拆分回各个原始请求：

func (b *RequestBatcher) processBatch(ctx context.Context, batch []MergedRequest) { allPrompts := make([]string, 0) separator := "\n---SEPARATOR---\n" for _, req := range batch { allPrompts = append(allPrompts, req.Prompts...) } combinedPrompt := strings.Join(allPrompts, separator) resp, err := b.llmClient.Complete(ctx, combinedPrompt) if err != nil { for _, req := range batch { req.ErrCh <- err } return } // 按分隔符拆分响应 parts := strings.Split(resp, separator) idx := 0 for _, req := range batch { if idx >= len(parts) { req.ErrCh <- fmt.Errorf("批量响应的分隔数量不匹配") continue } count := len(req.Prompts) req.Results <- parts[idx : idx+count] idx += count } }

三、Token 缓存的语义哈希

// 语义级 Token 缓存：相同意图的请求复用之前的结果 type SemanticCache struct { store *redis.Client embedder *EmbeddingClient threshold float64 // 相似度阈值 0.0-1.0 } func (c *SemanticCache) GetOrCompute( ctx context.Context, prompt string, compute func(context.Context) (string, error), ) (string, error) { // 1. 精确匹配：Prompot 完全相同的直接返回 exactKey := fmt.Sprintf("llm_cache:exact:%x", sha256.Sum256([]byte(prompt))) if val, err := c.store.Get(ctx, exactKey).Result(); err == nil { return val, nil } // 2. 语义匹配：通过 Embedding 向量查找相似 Prompt 的结果 embedding, err := c.embedder.Embed(ctx, prompt) if err == nil { similar, err := c.findSimilar(ctx, embedding) if err == nil && similar.Score > c.threshold { return similar.Result, nil } } // 3. 计算新结果 result, err := compute(ctx) if err != nil { return "", err } // 4. 写入缓存 pipe := c.store.Pipeline() pipe.Set(ctx, exactKey, result, 1*time.Hour) if err == nil { pipe.Set(ctx, fmt.Sprintf("llm_cache:vec:%x", sha256.Sum256(embedding)), result, 1*time.Hour, ) } pipe.Exec(ctx) return result, nil }