From 18c2385c615cc1fc9d6fbb538288acfde368bfd9 Mon Sep 17 00:00:00 2001 From: dev Date: Tue, 25 Aug 2026 08:24:08 +0800 Subject: [PATCH] feat: add per-source TTFB and tokens/s metrics to status page - Req: add FirstByteMs field (ms to first byte, tracked for streaming) - Stat: add FirstByteSum for aggregation - SourceAverages(): new method computing per-source avg TTFB and tokens/s from the in-memory ring (300s window) - SourceStatus: add AvgFirstByteMs and AvgTokPerS fields - pumpStream: record FirstByteMs after first SSE chunk sent to client - singleChat/singleChatAuto: set FirstByteMs = LatMs (non-streaming) - handleStatusAPI: populate the new SourceStatus fields from SourceAverages() - WebUI source table: two new columns showing TTFB (s) and Tokens/s --- .../ses_fcda7f5e1ffe9g78A7lVoLMxn7.json | 10 +++ deploy.sh | 11 +-- internal/gateway/chat.go | 13 ++- internal/gateway/server.go | 5 ++ internal/gateway/stats.go | 83 +++++++++++++++++-- internal/gateway/ui/index.html | 10 ++- internal/provider/registry.go | 6 ++ 7 files changed, 118 insertions(+), 20 deletions(-) create mode 100644 .omo/run-continuation/ses_fcda7f5e1ffe9g78A7lVoLMxn7.json diff --git a/.omo/run-continuation/ses_fcda7f5e1ffe9g78A7lVoLMxn7.json b/.omo/run-continuation/ses_fcda7f5e1ffe9g78A7lVoLMxn7.json new file mode 100644 index 0000000..d86ef64 --- /dev/null +++ b/.omo/run-continuation/ses_fcda7f5e1ffe9g78A7lVoLMxn7.json @@ -0,0 +1,10 @@ +{ + "sessionID": "ses_fcda7f5e1ffe9g78A7lVoLMxn7", + "updatedAt": "2026-08-24T11:19:43.949Z", + "sources": { + "background-task": { + "state": "idle", + "updatedAt": "2026-08-24T11:19:43.949Z" + } + } +} \ No newline at end of file diff --git a/deploy.sh b/deploy.sh index ae09242..e8546ff 100755 --- a/deploy.sh +++ b/deploy.sh @@ -144,9 +144,9 @@ atomic_replace_binary() { sync_adapters() { log "同步适配器" - # 只从内嵌适配器目录同步（完整集 10 个适配器，含最新 usage 透传修复）。 - # 注意：仓库根的 adapters/ 是运行时覆盖目录，可能不全（缺 opencode.lua 等）， - # 不可作为部署源。 + # 只覆盖内置适配器（与 internal/lua/adapters 同名的文件）， + # 保留目录里其它运行时上传的 .lua —— WebUI 上传的自定义适配器 + # 必须跨部署存活，清空会静默移除线上源依赖的适配器。 SRC_ADAPTERS="$REPO_DIR/internal/lua/adapters" [[ -d "$SRC_ADAPTERS" ]] || fail "内嵌适配器目录不存在: $SRC_ADAPTERS" @@ -158,14 +158,11 @@ sync_adapters() { log " 旧适配器已备份到 $BACKUP_DIR" fi - # 清空目标目录中的 .lua 文件（保留 .bak.* 归档），再复制全部新适配器。 - # 不依赖 rsync（部署机可能未安装），纯 shell 保证可移植。 - find "$TARGET_ADAPTERS" -maxdepth 1 -name '*.lua' -not -name '*.bak.*' -delete cp -f "$SRC_ADAPTERS/"*.lua "$TARGET_ADAPTERS/" chmod 0644 "$TARGET_ADAPTERS/"*.lua # 校验：每个适配器必须包含 usage 透传修复（opencode/openai/deepseek 等应有 'uses'） - local required_files=(openai deepseek anthropic gemini ollama opencode github groq kimicode mistral) + local required_files=(openai deepseek anthropic gemini ollama opencode github groq kimicode mistral sensenova agentrouter) for f in "${required_files[@]}"; do [[ -f "$TARGET_ADAPTERS/$f.lua" ]] || warn " 缺少适配器: $f.lua" done diff --git a/internal/gateway/chat.go b/internal/gateway/chat.go index 1f51824..429d0df 100644 --- a/internal/gateway/chat.go +++ b/internal/gateway/chat.go @@ -613,6 +613,9 @@ func (g *Gateway) singleChat(w http.ResponseWriter, ctx context.Context, cands [ recordChatUsage(rec, req, resp) rec.Source = usedSrc rec.Model = usedModel + // Non-streaming: the whole response arrives at once, so TTFB equals + // the total latency. + rec.FirstByteMs = rec.LatMs g.writeRec(rec) writeChatCompletion(w, resp, effective) } @@ -668,7 +671,7 @@ func mergeUsage(prev, cur *types.TokenUsage) *types.TokenUsage { // OpenAI-standard final usage chunk (empty choices) and [DONE]. modelName // follows writeChatCompletion's rule (requested id for direct routes, exact // slot model for AUTO). -func (g *Gateway) pumpStream(w http.ResponseWriter, rec *Req, chunks <-chan types.UnifiedChunk, modelName string) { +func (g *Gateway) pumpStream(w http.ResponseWriter, rec *Req, chunks <-chan types.UnifiedChunk, modelName string, t0 time.Time) { w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") @@ -697,6 +700,9 @@ func (g *Gateway) pumpStream(w http.ResponseWriter, rec *Req, chunks <-chan type }) { return } + // First SSE byte sent to the client: record time-to-first-byte for the + // source's status-page latency average. + rec.FirstByteMs = time.Since(t0).Milliseconds() var lastUsage *types.TokenUsage lastFinish := "" for ck := range chunks { @@ -787,7 +793,7 @@ func (g *Gateway) streamChat(w http.ResponseWriter, ctx context.Context, cands [ // discarded it. rec.Source = usedSrc rec.Prompt = estimatePromptTokens(req) - g.pumpStream(w, rec, chunks, effective) + g.pumpStream(w, rec, chunks, effective, t0) } // singleChatAuto runs a non-streaming AUTO request down the chain (see @@ -811,6 +817,7 @@ func (g *Gateway) singleChatAuto(w http.ResponseWriter, ctx context.Context, cha recordChatUsage(rec, req, resp) rec.Source = usedSrc rec.Model = usedModel + rec.FirstByteMs = rec.LatMs g.writeRec(rec) writeChatCompletion(w, resp, usedModel) } @@ -838,7 +845,7 @@ func (g *Gateway) streamChatAuto(w http.ResponseWriter, ctx context.Context, cha } rec.Source = usedSrc rec.Prompt = estimatePromptTokens(req) - g.pumpStream(w, rec, chunks, usedModel) + g.pumpStream(w, rec, chunks, usedModel, t0) } func (g *Gateway) handleImage(w http.ResponseWriter, r *http.Request) { diff --git a/internal/gateway/server.go b/internal/gateway/server.go index fe6a21a..c8f8575 100644 --- a/internal/gateway/server.go +++ b/internal/gateway/server.go @@ -469,11 +469,16 @@ func (g *Gateway) handleStatusAPI(w http.ResponseWriter, r *http.Request) { if reqRole(r.Context()) == "admin" { sts := g.core.Registry().Status() recent := g.stats.SourceRecent(300) + avgs := g.stats.SourceAverages(300) for i := range sts { if v, ok := recent[sts[i].Name]; ok { sts[i].RecentOK = v[0] sts[i].RecentErr = v[1] } + if a, ok := avgs[sts[i].Name]; ok { + sts[i].AvgFirstByteMs = a.AvgFirstByteMs + sts[i].AvgTokPerS = a.AvgTokPerS + } } resp["sources"] = sts resp["adapters"] = g.core.ListAdapters() diff --git a/internal/gateway/stats.go b/internal/gateway/stats.go index 5bc6f31..789e833 100644 --- a/internal/gateway/stats.go +++ b/internal/gateway/stats.go @@ -26,6 +26,10 @@ type Req struct { Compl int64 `json:"completion_tokens"` // LatMs total handling time ms LatMs int64 `json:"latency_ms"` + // FirstByteMs time-to-first-byte for streaming (ms from request start + // to the first SSE chunk sent to the client); for non-streaming it + // equals LatMs. 0 when unmeasured (legacy records). + FirstByteMs int64 `json:"first_byte_ms,omitempty"` OK bool `json:"ok"` // Status http status code Status int `json:"status"` @@ -35,14 +39,15 @@ type Req struct { // Stat aggregates counters for one dimension row. type Stat struct { - Reqs int64 `json:"reqs"` - OK int64 `json:"ok"` - Err int64 `json:"err"` - Tokens int64 `json:"tokens"` - Prompt int64 `json:"prompt_tokens"` - Compl int64 `json:"completion_tokens"` - LatSum int64 `json:"latency_sum_ms"` - LatMax int64 `json:"latency_max_ms"` + Reqs int64 `json:"reqs"` + OK int64 `json:"ok"` + Err int64 `json:"err"` + Tokens int64 `json:"tokens"` + Prompt int64 `json:"prompt_tokens"` + Compl int64 `json:"completion_tokens"` + LatSum int64 `json:"latency_sum_ms"` + LatMax int64 `json:"latency_max_ms"` + FirstByteSum int64 `json:"first_byte_sum_ms,omitempty"` } type agrRow struct { @@ -130,6 +135,9 @@ func incStatus(a *Stat, name string, r Req) { if r.LatMs > a.LatMax { a.LatMax = r.LatMs } + if r.FirstByteMs > 0 { + a.FirstByteSum += r.FirstByteMs + } } func (s *Stats) LoadAudit(path string) { @@ -452,6 +460,65 @@ func (s *Stats) SourceRecent(windowSec int64) map[string][2]int64 { return out } +// SourceAvg carries per-source performance averages for the status page. +type SourceAvg struct { + // AvgFirstByteMs is the mean time-to-first-byte over successful + // requests in the window (0 when no measured samples). + AvgFirstByteMs int64 `json:"avg_first_byte_ms"` + // AvgTokPerS is the aggregate completion throughput: + // sum(completion_tokens) / sum(latency_seconds) (0 when no samples). + AvgTokPerS int64 `json:"avg_tok_per_s"` + // Samples is the number of successful requests the averages cover. + Samples int64 `json:"samples"` +} + +// SourceAverages computes TTFB and tokens/s averages per source from the +// in-memory ring within the window (unix seconds). Only successful chat/ +// stream rows count; image and failed rows are skipped. +func (s *Stats) SourceAverages(windowSec int64) map[string]SourceAvg { + s.mu.Lock() + defer s.mu.Unlock() + cut := time.Now().Unix() - windowSec + type acc struct { + fbSum, latSum, complSum, n int64 + } + accs := map[string]*acc{} + for _, r := range s.recs { + if !r.OK || r.Source == "" || r.Time/1000 < cut { + continue + } + if r.Type != "chat" && r.Type != "stream" { + continue + } + if r.LatMs <= 0 { + continue + } + a := accs[r.Source] + if a == nil { + a = &acc{} + accs[r.Source] = a + } + a.latSum += r.LatMs + a.complSum += r.Compl + a.fbSum += r.FirstByteMs + if r.FirstByteMs > 0 { + a.n++ + } + } + out := make(map[string]SourceAvg, len(accs)) + for src, a := range accs { + avg := SourceAvg{Samples: a.n} + if a.n > 0 { + avg.AvgFirstByteMs = a.fbSum / a.n + } + if a.latSum > 0 && a.complSum > 0 { + avg.AvgTokPerS = a.complSum * 1000 / a.latSum + } + out[src] = avg + } + return out +} + // Snapshot returns the whole dashboard payload; when key != "" the records // and aggregate views are restricted to that gateway key. func (s *Stats) Snapshot(limit int, key string) map[string]interface{} { diff --git a/internal/gateway/ui/index.html b/internal/gateway/ui/index.html index 272fe29..8ffb2ef 100644 --- a/internal/gateway/ui/index.html +++ b/internal/gateway/ui/index.html @@ -774,6 +774,8 @@ tURL: "地址", tConn: "连接", tConc: "并发", + tAvgLat: "首字延迟", + tTokSpd: "Token/s", online: "在线", offline: "退避 / 不可用", adTitle: "已加载适配器", @@ -971,6 +973,8 @@ tURL: "URL", tConn: "Status", tConc: "Concurrency", + tAvgLat: "TTFB", + tTokSpd: "Tokens/s", online: "online", offline: "backoff / down", adTitle: "Loaded adapters", @@ -1374,7 +1378,9 @@ ${esc(x.name)}${esc(x.adapter)}
${(x.models || []).map((m) => `${esc(m)}`).join("")}
${esc(x.base_url || "")} - ${x.max_concurrent}`, + ${x.max_concurrent} + ${x.avg_first_byte_ms ? (x.avg_first_byte_ms / 1000).toFixed(1) + "s" : "—"} + ${x.avg_tok_per_s ? x.avg_tok_per_s + " tok/s" : "—"}`, ) .join(""); } catch (e) { @@ -1422,7 +1428,7 @@ ${ s.sources ? `

${t("srcTitle")} (${s.sources.length})

-
${t("tConn")}${t("tName")}${t("tAdapter")}${t("tModels")}${t("tURL")}${t("tConc")}
+
${t("tConn")}${t("tName")}${t("tAdapter")}${t("tModels")}${t("tURL")}${t("tConc")}${t("tAvgLat")}${t("tTokSpd")}

状态列每 5 秒自动刷新

` : "" diff --git a/internal/provider/registry.go b/internal/provider/registry.go index 6cae1af..6d49efb 100644 --- a/internal/provider/registry.go +++ b/internal/provider/registry.go @@ -217,6 +217,12 @@ type SourceStatus struct { // probe was rate-limited by the upstream). RecentOK int64 `json:"recent_ok,omitempty"` RecentErr int64 `json:"recent_err,omitempty"` + // AvgFirstByteMs is the mean time-to-first-byte (ms) for recently + // successful requests served by this source (0 when unmeasured). + AvgFirstByteMs int64 `json:"avg_first_byte_ms,omitempty"` + // AvgTokPerS is the aggregate completion throughput (tokens/s) for + // recently successful requests served by this source (0 when no data). + AvgTokPerS int64 `json:"avg_tok_per_s,omitempty"` } // ProbeAll runs a live reachability check for every provider (in parallel).