Skip to content

Commit 61ebb49

Browse files
rootclaude
andcommitted
perf(go): 强化 prompt_cache 同会话粘号,提升多轮缓存命中
Codex 多轮依赖 prompt_cache_key + 同账号;候选池 top-N 可能漏掉粘号账号。 - 按 prompt_cache_key / X-Grok-Conv-Id 注入 sticky 账号到候选窗口 - fingerprint 优先 prompt_cache_key 与会话头 - 配合既有 affinity,减少切号导致的缓存冷启动 Co-Authored-By: Claude <noreply@anthropic.com>
1 parent 4e2c2b9 commit 61ebb49

3 files changed

Lines changed: 145 additions & 5 deletions

File tree

‎internal/proxy/chat.go‎

Lines changed: 43 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import (
88
"errors"
99
"fmt"
1010
"io"
11+
"net/http"
1112
"strings"
1213
"time"
1314

@@ -663,11 +664,23 @@ func upstreamAccounts(chain []pool.Candidate) []grok.Account {
663664
}
664665

665666
func ChatFingerprint(request ChatRequest) string {
666-
for _, key := range []string{"conversation_id", "conversation", "thread_id", "session_id", "prompt_cache_key"} {
667+
if request.Raw == nil {
668+
return ""
669+
}
670+
// Explicit sticky keys first (Codex / OpenAI Responses multi-turn).
671+
for _, key := range []string{"prompt_cache_key", "conversation_id", "conversation", "thread_id", "session_id"} {
667672
if value, _ := request.Raw[key].(string); strings.TrimSpace(value) != "" {
668673
return "chat:" + strings.TrimSpace(request.Model) + ":" + key + ":" + strings.TrimSpace(value)
669674
}
670675
}
676+
// Nested metadata (Anthropic / some relays).
677+
if meta, _ := request.Raw["metadata"].(map[string]any); meta != nil {
678+
for _, key := range []string{"prompt_cache_key", "session_id", "sessionId", "thread_id", "conversation_id", "user_id"} {
679+
if value, _ := meta[key].(string); strings.TrimSpace(value) != "" {
680+
return "chat:" + strings.TrimSpace(request.Model) + ":meta:" + key + ":" + strings.TrimSpace(value)
681+
}
682+
}
683+
}
671684
messages, ok := request.Raw["messages"].([]any)
672685
if !ok || len(messages) == 0 {
673686
return ""
@@ -680,6 +693,35 @@ func ChatFingerprint(request ChatRequest) string {
680693
return "chat:" + strings.TrimSpace(request.Model) + ":messages:" + hex.EncodeToString(sum[:16])
681694
}
682695

696+
// ChatFingerprintFromHeaders picks sticky keys from common client/proxy headers
697+
// (Codex X-Grok-Conv-Id, session/thread headers).
698+
func ChatFingerprintFromHeaders(headers http.Header, model string) string {
699+
if headers == nil {
700+
return ""
701+
}
702+
get := func(names ...string) string {
703+
for _, name := range names {
704+
if v := strings.TrimSpace(headers.Get(name)); v != "" {
705+
return v
706+
}
707+
}
708+
return ""
709+
}
710+
if v := get("X-Grok-Conv-Id", "x-grok-conv-id", "X-Grok2API-Conv-Id"); v != "" {
711+
return "chat:" + strings.TrimSpace(model) + ":conv:" + v
712+
}
713+
if v := get("X-Session-Id", "x-session-id", "Session-Id"); v != "" {
714+
return "chat:" + strings.TrimSpace(model) + ":session:" + v
715+
}
716+
if v := get("X-Thread-Id", "x-thread-id", "Thread-Id"); v != "" {
717+
return "chat:" + strings.TrimSpace(model) + ":thread:" + v
718+
}
719+
if v := get("X-Prompt-Cache-Key", "x-prompt-cache-key"); v != "" {
720+
return "chat:" + strings.TrimSpace(model) + ":prompt_cache_key:" + v
721+
}
722+
return ""
723+
}
724+
683725
func preferAffinity(ctx context.Context, candidates []pool.Candidate, store AffinityStore, fingerprint string) {
684726
accountID, err := store.GetAffinity(ctx, fingerprint)
685727
if err != nil || accountID == "" {

‎internal/server/server.go‎

Lines changed: 66 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -479,7 +479,7 @@ func serveChatCompletions(w http.ResponseWriter, r *http.Request, options Option
479479
writeJSON(w, http.StatusBadRequest, map[string]any{"detail": err.Error()})
480480
return
481481
}
482-
candidates, err := listCandidates(r.Context(), options)
482+
candidates, err := listCandidatesForRequest(r.Context(), options, chatReq, r.Header)
483483
if err != nil {
484484
writeJSON(w, http.StatusInternalServerError, map[string]any{"detail": err.Error()})
485485
return
@@ -763,7 +763,7 @@ func serveMessages(w http.ResponseWriter, r *http.Request, options Options) {
763763
}
764764
allowedTools := allowedAnthropicToolNames(body)
765765
chatReq := proxy.ChatRequest{Model: model, Stream: false, Raw: body}
766-
candidates, err := listCandidates(r.Context(), options)
766+
candidates, err := listCandidatesForRequest(r.Context(), options, chatReq, r.Header)
767767
if err != nil {
768768
writeAnthropicError(w, http.StatusInternalServerError, err.Error(), "api_error")
769769
return
@@ -960,15 +960,22 @@ func serveResponses(w http.ResponseWriter, r *http.Request, options Options) {
960960
writeOpenAIError(w, http.StatusBadRequest, "input must contain at least one message", "invalid_request_error")
961961
return
962962
}
963-
candidates, err := listCandidates(r.Context(), options)
963+
chatReq := proxy.ChatRequest{Model: model, Stream: stream, Raw: body}
964+
// Preserve client prompt_cache_key on chatReq for sticky fingerprint (stripped later for upstream).
965+
if pck := strings.TrimSpace(stringValue(raw["prompt_cache_key"])); pck != "" {
966+
if chatReq.Raw == nil {
967+
chatReq.Raw = map[string]any{}
968+
}
969+
chatReq.Raw["prompt_cache_key"] = pck
970+
}
971+
candidates, err := listCandidatesForRequest(r.Context(), options, chatReq, r.Header)
964972
if err != nil {
965973
writeOpenAIError(w, http.StatusInternalServerError, err.Error(), "server_error")
966974
return
967975
}
968976
service := proxy.ChatService{Catalog: modelCatalog(options), Client: upstreamClient(options), PickObserver: options.PickObserver, AffinityStore: options.AffinityStore}
969977
started := time.Now()
970978
responseID := responses.NewResponseID()
971-
chatReq := proxy.ChatRequest{Model: model, Stream: stream, Raw: body}
972979
respPolicy := historycompact.ResolveOutboundToolPolicy(
973980
"openai_responses",
974981
r.UserAgent(),
@@ -2433,6 +2440,25 @@ func upstreamClient(options Options) *grok.Client {
24332440
return &grok.Client{BaseURL: options.Config.UpstreamBase}
24342441
}
24352442

2443+
func listCandidatesForRequest(ctx context.Context, options Options, chatReq proxy.ChatRequest, headers http.Header) ([]pool.Candidate, error) {
2444+
candidates, err := listCandidates(ctx, options)
2445+
if err != nil {
2446+
return nil, err
2447+
}
2448+
fp := proxy.ChatFingerprint(chatReq)
2449+
if fp == "" {
2450+
fp = proxy.ChatFingerprintFromHeaders(headers, chatReq.Model)
2451+
}
2452+
// If still empty, try raw prompt_cache_key alone.
2453+
if fp == "" && chatReq.Raw != nil {
2454+
if pck, _ := chatReq.Raw["prompt_cache_key"].(string); strings.TrimSpace(pck) != "" {
2455+
fp = "chat:" + strings.TrimSpace(chatReq.Model) + ":prompt_cache_key:" + strings.TrimSpace(pck)
2456+
}
2457+
}
2458+
sticky := stickyAccountID(ctx, options, fp)
2459+
return ensureStickyCandidate(ctx, options, candidates, sticky), nil
2460+
}
2461+
24362462
func listCandidates(ctx context.Context, options Options) ([]pool.Candidate, error) {
24372463
if len(options.Candidates) > 0 {
24382464
out := make([]pool.Candidate, len(options.Candidates))
@@ -2445,6 +2471,42 @@ func listCandidates(ctx context.Context, options Options) ([]pool.Candidate, err
24452471
return options.Store.ListPoolCandidates(ctx)
24462472
}
24472473

2474+
// ensureStickyCandidate injects an affinity-bound account into the pick window
2475+
// even when it is outside the top-N least_used scan. Critical for prompt-cache
2476+
// multi-turn (Codex prompt_cache_key / X-Grok-Conv-Id) high hit rates.
2477+
func ensureStickyCandidate(ctx context.Context, options Options, candidates []pool.Candidate, stickyID string) []pool.Candidate {
2478+
stickyID = strings.TrimSpace(stickyID)
2479+
if stickyID == "" || options.Store == nil {
2480+
return candidates
2481+
}
2482+
for _, c := range candidates {
2483+
if c.ID == stickyID {
2484+
return candidates
2485+
}
2486+
}
2487+
extra, err := options.Store.GetPoolCandidate(ctx, stickyID)
2488+
if err != nil || extra == nil {
2489+
return candidates
2490+
}
2491+
// Put sticky first so prepareChain affinity boost is redundant but safe.
2492+
out := make([]pool.Candidate, 0, len(candidates)+1)
2493+
out = append(out, *extra)
2494+
out = append(out, candidates...)
2495+
return out
2496+
}
2497+
2498+
func stickyAccountID(ctx context.Context, options Options, fingerprint string) string {
2499+
fingerprint = strings.TrimSpace(fingerprint)
2500+
if fingerprint == "" || options.AffinityStore == nil {
2501+
return ""
2502+
}
2503+
id, err := options.AffinityStore.GetAffinity(ctx, fingerprint)
2504+
if err != nil {
2505+
return ""
2506+
}
2507+
return strings.TrimSpace(id)
2508+
}
2509+
24482510
func modelCatalog(options Options) *models.Catalog {
24492511
if options.Models != nil {
24502512
return options.Models

‎internal/store/postgres/pool.go‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,42 @@ var (
2121

2222
const candidateCacheTTL = 400 * time.Millisecond
2323

24+
// GetPoolCandidate loads one account as a pick candidate (sticky TTFT / prompt-cache path).
25+
func (c *Connector) GetPoolCandidate(ctx context.Context, accountID string) (*pool.Candidate, error) {
26+
accountID = strings.TrimSpace(accountID)
27+
if accountID == "" {
28+
return nil, nil
29+
}
30+
row := c.Pool.QueryRow(ctx, `
31+
SELECT a.id, a.payload, a.email, a.user_id, a.team_id, a.expires_at,
32+
COALESCE(ap.enabled, true), COALESCE(ap.disabled_for_quota, false),
33+
ap.cooldown_until, COALESCE(ap.blocked_models, '{}'::jsonb),
34+
COALESCE(ap.request_count, 0), COALESCE(ap.weight, 1)
35+
FROM accounts a
36+
LEFT JOIN account_pool ap ON ap.account_id = a.id
37+
WHERE a.id = $1
38+
LIMIT 1`, accountID)
39+
var candidate pool.Candidate
40+
var payloadBytes, blockedBytes []byte
41+
var email, userID, teamID *string
42+
var expiresAt, cooldownUntil *time.Time
43+
if err := row.Scan(&candidate.ID, &payloadBytes, &email, &userID, &teamID, &expiresAt, &candidate.Enabled, &candidate.DisabledForQuota, &cooldownUntil, &blockedBytes, &candidate.RequestCount, &candidate.Weight); err != nil {
44+
return nil, err
45+
}
46+
payload := decodeMap(payloadBytes)
47+
candidate.Token, _ = firstString(payload, "key", "access_token", "token")
48+
candidate.Email = stringValue(email, stringFromMap(payload, "email"))
49+
candidate.UserID = stringValue(userID, firstMapString(payload, "user_id", "principal_id"))
50+
candidate.TeamID = stringValue(teamID, stringFromMap(payload, "team_id"))
51+
candidate.ExpiresAt = expiresAt
52+
candidate.CooldownUntil = cooldownUntil
53+
candidate.BlockedModels = decodeMap(blockedBytes)
54+
if strings.TrimSpace(candidate.Token) == "" {
55+
return nil, nil
56+
}
57+
return &candidate, nil
58+
}
59+
2460
func (c *Connector) ListPoolCandidates(ctx context.Context) ([]pool.Candidate, error) {
2561
candidateCacheMu.Lock()
2662
if time.Since(candidateCacheAt) < candidateCacheTTL && len(candidateCacheData) > 0 {

0 commit comments

Comments
 (0)