refactor(initialize,util): 重构缓存头设置与 Anthropic 缓存处理

1. 抽离重复的缓存头设置逻辑为公共函数setCacheHeaders
2. 新增Anthropic显式缓存块追踪功能，支持标准缓存头上报
3. 优化缓存状态检测与Age头生成逻辑
4. 统一多接口的缓存头设置流程

Author: xiaozhou26

initialize/files.go

@@ -392,8 +392,7 @@ func (h *Handler) chatWithFiles(c *gin.Context) {
 
 	// Cache breakdown is known before streaming starts, so set these headers
 	// before the first chunk is flushed (works for both stream and non-stream).
-	c.Header("X-Cache-Creation-Tokens", fmt.Sprintf("%d", cacheCreation))
-	c.Header("X-Cache-Read-Tokens", fmt.Sprintf("%d", cacheRead))
+	setCacheHeaders(c, promptHash, cacheCreation, cacheRead)
 
 	result := duckgo.Handler(c, response, translated_request, req.Stream, stats)
 

initialize/handlers.go

@@ -105,8 +105,7 @@ func (h *Handler) duckduckgo(c *gin.Context) {
 
 	// Cache breakdown is known before streaming starts, so set these headers
 	// before the first chunk is flushed (works for both stream and non-stream).
-	c.Header("X-Cache-Creation-Tokens", fmt.Sprintf("%d", cacheCreation))
-	c.Header("X-Cache-Read-Tokens", fmt.Sprintf("%d", cacheRead))
+	setCacheHeaders(c, promptHash, cacheCreation, cacheRead)
 
 	result := duckgo.Handler(c, response, translated_request, original_request.Stream, stats)
 
@@ -144,6 +143,19 @@ func messagesText(messages []officialtypes.ApiMessage) string {
 	return sb.String()
 }
 
+// setCacheHeaders sets both the existing custom cache headers and the standard
+// HTTP cache headers (X-Cache / Age) that external cache-statistics software
+// (Varnish, Squid, CDNs, monitoring) can read.
+func setCacheHeaders(c *gin.Context, promptHash string, cacheCreation, cacheRead int) {
+	c.Header("X-Cache-Creation-Tokens", fmt.Sprintf("%d", cacheCreation))
+	c.Header("X-Cache-Read-Tokens", fmt.Sprintf("%d", cacheRead))
+	xCache, age := util.CacheHeaders(promptHash)
+	c.Header("X-Cache", xCache)
+	if xCache == "HIT" {
+		c.Header("Age", fmt.Sprintf("%d", age))
+	}
+}
+
 func (h *Handler) responses(c *gin.Context) {
 	var responseRequest officialtypes.ResponseAPIRequest
 	err := c.BindJSON(&responseRequest)
@@ -173,6 +185,10 @@ func (h *Handler) responses(c *gin.Context) {
 	cacheCreation, cacheRead := util.RecordCache(promptHash, inputTokens)
 	cachedTokens := cacheRead
 
+	// Cache breakdown is known before streaming starts, so set these headers
+	// before the first chunk is flushed (works for both stream and non-stream).
+	setCacheHeaders(c, promptHash, cacheCreation, cacheRead)
+
 	translatedRequest, response, err := h.startDuckDuckGoRequest(chatRequest)
 	if err != nil {
 		c.JSON(500, gin.H{
@@ -189,8 +205,7 @@ func (h *Handler) responses(c *gin.Context) {
 	}
 
 	// Cache breakdown is known before streaming; set before first flush.
-	c.Header("X-Cache-Creation-Tokens", fmt.Sprintf("%d", cacheCreation))
-	c.Header("X-Cache-Read-Tokens", fmt.Sprintf("%d", cacheRead))
+	setCacheHeaders(c, promptHash, cacheCreation, cacheRead)
 
 	start := time.Now()
 	stats := duckgo.HandlerStats{

initialize/messages.go

@@ -55,9 +55,12 @@ func (h *Handler) messagesHandler(c *gin.Context) {
 
 	// Input token counting + prompt-cache simulation.
 	inputTokens := util.CountMessagesTokens(apiReq.Messages)
+	cacheCreation, cacheRead := util.RecordAnthropicCache(req)
 	promptHash := util.HashPrompt(messagesText(apiReq.Messages))
-	cacheCreation, cacheRead := util.RecordCache(promptHash, inputTokens)
-	cachedTokens := cacheRead
+	if cacheCreation == 0 && cacheRead == 0 {
+		// Fallback to prompt hash simulation if request has no explicit cache_control blocks
+		cacheCreation, cacheRead = util.RecordCache(promptHash, inputTokens)
+	}
 
 	translated, response, err := h.startDuckDuckGoRequest(apiReq)
 	if err != nil {
@@ -77,11 +80,10 @@ func (h *Handler) messagesHandler(c *gin.Context) {
 	}
 
 	// Cache breakdown is known before streaming; set before first flush.
-	c.Header("X-Cache-Creation-Tokens", fmt.Sprintf("%d", cacheCreation))
-	c.Header("X-Cache-Read-Tokens", fmt.Sprintf("%d", cacheRead))
+	setCacheHeaders(c, promptHash, cacheCreation, cacheRead)
 
 	start := time.Now()
-	result := handleAnthropicStream(c, response.Body, req.Model, req.Stream, start, inputTokens, cachedTokens, effort)
+	result := handleAnthropicStream(c, response.Body, req.Model, req.Stream, start, inputTokens, cacheCreation, cacheRead, effort)
 
 	// Timing headers (only delivered for non-stream; for stream the same values
 	// live in the message_delta event).
@@ -93,13 +95,13 @@ func (h *Handler) messagesHandler(c *gin.Context) {
 	}
 
 	if !req.Stream {
-		c.JSON(200, buildAnthropicResponse(req.Model, result, inputTokens, cachedTokens, effort))
+		c.JSON(200, buildAnthropicResponse(req.Model, result, inputTokens, cacheCreation, cacheRead, effort))
 	}
 }
 
 // handleAnthropicStream reads DuckDuckGo's text SSE and emits Anthropic SSE events.
 // For non-stream it accumulates the text and returns the result.
-func handleAnthropicStream(c *gin.Context, body io.ReadCloser, model string, stream bool, start time.Time, inputTokens, cachedTokens int, effort string) anthropicStreamResult {
+func handleAnthropicStream(c *gin.Context, body io.ReadCloser, model string, stream bool, start time.Time, inputTokens, cacheCreation, cacheRead int, effort string) anthropicStreamResult {
 	defer body.Close()
 
 	reader := bufio.NewReader(body)
@@ -124,7 +126,11 @@ func handleAnthropicStream(c *gin.Context, body io.ReadCloser, model string, str
 				Role:    "assistant",
 				Model:   model,
 				Content: []anthropic.ContentBlock{},
-				Usage:   anthropic.AnthropicUsage{InputTokens: inputTokens},
+				Usage: anthropic.AnthropicUsage{
+					InputTokens:              inputTokens,
+					CacheCreationInputTokens: cacheCreation,
+					CacheReadInputTokens:     cacheRead,
+				},
 			},
 		})
 		writeEvent(c, "content_block_start", anthropic.ContentBlockStartEvent{
@@ -210,13 +216,12 @@ func writeEvent(c *gin.Context, eventType string, payload interface{}) {
 }
 
 // buildAnthropicResponse builds the non-stream MessagesResponse.
-func buildAnthropicResponse(model string, r anthropicStreamResult, inputTokens, cachedTokens int, effort string) anthropic.MessagesResponse {
+func buildAnthropicResponse(model string, r anthropicStreamResult, inputTokens, cacheCreation, cacheRead int, effort string) anthropic.MessagesResponse {
 	usage := anthropic.AnthropicUsage{
-		InputTokens:  inputTokens,
-		OutputTokens: r.outputTokens,
-	}
-	if cachedTokens > 0 {
-		usage.CacheReadInputTokens = cachedTokens
+		InputTokens:              inputTokens,
+		OutputTokens:             r.outputTokens,
+		CacheCreationInputTokens: cacheCreation,
+		CacheReadInputTokens:     cacheRead,
 	}
 	return anthropic.MessagesResponse{
 		ID:         "msg_" + util.RandomHexadecimalString(),

typings/anthropic/messages.go

@@ -46,8 +46,9 @@ type ContentBlock struct {
 	ToolUseID string          `json:"tool_use_id,omitempty"` // tool_result
 	Content   json.RawMessage `json:"content,omitempty"`     // tool_result
 	IsError   bool            `json:"is_error,omitempty"`
-	Thinking  string          `json:"thinking,omitempty"` // thinking block
-	Signature string          `json:"signature,omitempty"`
+	Thinking     string          `json:"thinking,omitempty"` // thinking block
+	Signature    string          `json:"signature,omitempty"`
+	CacheControl json.RawMessage `json:"cache_control,omitempty"`
 }
 
 // ImageSource describes a base64 image in Anthropic format.

util/cache_tracker.go

@@ -0,0 +1,110 @@
+package util
+
+import (
+	"encoding/json"
+	"sync"
+	"time"
+
+	anthropic "aurora/typings/anthropic"
+)
+
+type cacheBlock struct {
+	fp     string
+	tokens int
+}
+
+type cacheTracker struct {
+	mu   sync.Mutex
+	seen map[string]time.Time
+	ttl  time.Duration
+}
+
+var globalCacheTracker = &cacheTracker{
+	seen: make(map[string]time.Time),
+	ttl:  5 * time.Minute,
+}
+
+func cacheFingerprint(role, blockType, text string) string {
+	return role + "|" + blockType + "|" + text
+}
+
+func (t *cacheTracker) RecordAnthropic(req anthropic.MessagesRequest) (creationTokens, readTokens int) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	now := time.Now()
+	for fp, seenAt := range t.seen {
+		if now.Sub(seenAt) > t.ttl {
+			delete(t.seen, fp)
+		}
+	}
+
+	blocks := collectCacheBlocks(req)
+	if len(blocks) == 0 {
+		return 0, 0
+	}
+
+	for _, b := range blocks {
+		if _, ok := t.seen[b.fp]; ok {
+			readTokens += b.tokens
+		} else {
+			creationTokens += b.tokens
+			t.seen[b.fp] = now
+		}
+	}
+	return creationTokens, readTokens
+}
+
+func collectCacheBlocks(req anthropic.MessagesRequest) []cacheBlock {
+	var blocks []cacheBlock
+
+	// system blocks
+	if len(req.System) > 0 {
+		if req.System[0] == '[' {
+			var rawBlocks []map[string]interface{}
+			if err := json.Unmarshal(req.System, &rawBlocks); err == nil {
+				for _, m := range rawBlocks {
+					if _, hasCache := m["cache_control"]; hasCache {
+						text, _ := m["text"].(string)
+						fp := cacheFingerprint("system", "text", text)
+						tokens := CountToken(text)
+						if tokens < 1 && len(text) > 0 {
+							tokens = 1
+						}
+						blocks = append(blocks, cacheBlock{fp, tokens})
+					}
+				}
+			}
+		}
+	}
+
+	// message blocks
+	for _, msg := range req.Messages {
+		if len(msg.Content) == 0 || msg.Content[0] != '[' {
+			continue
+		}
+		var rawBlocks []map[string]interface{}
+		if err := json.Unmarshal(msg.Content, &rawBlocks); err != nil {
+			continue
+		}
+		for _, m := range rawBlocks {
+			if _, hasCache := m["cache_control"]; hasCache {
+				text, _ := m["text"].(string)
+				typ, _ := m["type"].(string)
+				fp := cacheFingerprint(msg.Role, typ, text)
+				tokens := CountToken(text)
+				if tokens < 1 && len(text) > 0 {
+					tokens = 1
+				}
+				blocks = append(blocks, cacheBlock{fp, tokens})
+			}
+		}
+	}
+
+	return blocks
+}
+
+// RecordAnthropicCache parses real cache_control blocks in Anthropic requests.
+func RecordAnthropicCache(req anthropic.MessagesRequest) (creationTokens, readTokens int) {
+	return globalCacheTracker.RecordAnthropic(req)
+}

util/util.go

@@ -90,6 +90,7 @@ const cacheTTL = 5 * time.Minute
 
 type cacheEntry struct {
 	tokens    int
+	createdAt time.Time // when the entry was created (for Age header)
 	expiresAt time.Time
 }
 
@@ -119,10 +120,39 @@ func RecordCache(promptHash string, tokens int) (cacheCreation int, cacheRead in
 	if e, ok := cacheStore[promptHash]; ok && now.Before(e.expiresAt) {
 		return 0, e.tokens
 	}
-	cacheStore[promptHash] = cacheEntry{tokens: tokens, expiresAt: now.Add(cacheTTL)}
+	cacheStore[promptHash] = cacheEntry{tokens: tokens, createdAt: now, expiresAt: now.Add(cacheTTL)}
 	return tokens, 0
 }
 
+// CacheStatus reports whether a prompt hash is currently cached (HIT vs MISS)
+// and how many seconds ago the entry was created (for the Age header).
+// Returns (hit bool, ageSeconds int). ageSeconds is 0 when not hit.
+func CacheStatus(promptHash string) (hit bool, ageSeconds int) {
+	if promptHash == "" {
+		return false, 0
+	}
+	cacheMu.Lock()
+	defer cacheMu.Unlock()
+
+	now := time.Now()
+	if e, ok := cacheStore[promptHash]; ok && now.Before(e.expiresAt) {
+		return true, int(now.Sub(e.createdAt).Seconds())
+	}
+	return false, 0
+}
+
+// CacheHeaders returns the standard HTTP cache headers so external
+// cache-statistics/monitoring software (Varnish, Squid, CDNs) can read them.
+//   - X-Cache: "HIT" or "MISS"
+//   - Age: <seconds> (omitted on MISS)
+func CacheHeaders(promptHash string) (xCache string, ageSeconds int) {
+	hit, age := CacheStatus(promptHash)
+	if hit {
+		return "HIT", age
+	}
+	return "MISS", 0
+}
+
 // HashPrompt builds a stable hash for prompt cache keying.
 func HashPrompt(s string) string {
 	h := sha256.Sum256([]byte(s))