diff --git a/backend/internal/proxy/gateway.go b/backend/internal/proxy/gateway.go index e6a2590..e2550de 100644 --- a/backend/internal/proxy/gateway.go +++ b/backend/internal/proxy/gateway.go @@ -41,6 +41,11 @@ type Gateway struct { modelDAO *dao.ModelDAO channelSvc *channel.Service usageRec *usage.Recorder + + // 原始请求/响应记录开关（系统配置 log_raw_requests，带 TTL 缓存避免每次查库）。 + rawLogMu sync.Mutex + rawLogVal bool + rawLogSet time.Time } func NewGateway(ctx context.Context, cfg *config.Config, db *gorm.DB, wg *sync.WaitGroup, userDAO *dao.UserDAO, apiKeyDAO *dao.ApiKeyDAO, usageDAO *dao.UsageDAO, dailyDAO *dao.DailyUsageDAO) *Gateway { @@ -79,6 +84,22 @@ func (g *Gateway) SetUsageRecorder(r *usage.Recorder) { g.usageRec = r } +// rawLogEnabled 读取系统配置 log_raw_requests（10s TTL 缓存），决定是否记录原始请求/响应。 +func (g *Gateway) rawLogEnabled() bool { + g.rawLogMu.Lock() + defer g.rawLogMu.Unlock() + if time.Since(g.rawLogSet) < 10*time.Second { + return g.rawLogVal + } + var sc store.SystemConfig + g.rawLogVal = false + if err := g.db.Where("key = ?", "log_raw_requests").First(&sc).Error; err == nil { + g.rawLogVal = strings.TrimSpace(sc.Value) == "true" + } + g.rawLogSet = time.Now() + return g.rawLogVal +} + // generateRequestID 生成请求级唯一 ID，用于用量明细关联与排障。 func generateRequestID() string { b := make([]byte, 12) @@ -98,6 +119,10 @@ type Request struct { UserID uint64 KeyID uint64 RequestID string + + CaptureRaw bool // 原始请求/响应记录（管理员 + 系统开关开启） + + rawBuf *strings.Builder // 上游原始响应累积器（仅 CaptureRaw 时非 nil） } // ParseRequest parses the incoming request and extracts key fields @@ -109,6 +134,7 @@ func (g *Gateway) ParseRequest(c *gin.Context, protocol string) (*Request, error apiKey, _ := c.Get("api_key") userID, _ := c.Get("user_id") + userRole, _ := c.Get("user_role") req := &Request{ Protocol: protocol, @@ -124,6 +150,11 @@ func (g *Gateway) ParseRequest(c *gin.Context, protocol string) (*Request, error req.APIKey = ak } + // 原始请求/响应记录：仅管理员 且 系统开关 log_raw_requests 开启。 + if role, _ := userRole.(string); role == store.RoleAdmin && g.rawLogEnabled() { + req.CaptureRaw = true + } + // Parse model and stream based on protocol switch protocol { case "chat": @@ -160,6 +191,12 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) { return } + // 原始请求/响应捕获：仅管理员 + 系统开关开启（req.CaptureRaw 已在 ParseRequest 判定）。 + // 客户端原始请求体即 req.Body；上游原始响应由 stream/bufferResponse 累积进 rawBuf。 + if req.CaptureRaw { + req.rawBuf = &strings.Builder{} + } + cands := g.channelSvc.Candidates(req.Model) // 内存健康过滤：连续失败进入 cooldown 的渠道不再尝试（渠道级健康自愈靠冷却过期）。 cands = g.channelSvc.FilterHealthy(cands) @@ -245,6 +282,9 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) { body, _ := io.ReadAll(resp.Body) resp.Body.Close() log.Printf("Upstream error: status=%d body=%s", resp.StatusCode, string(body)) + if req.rawBuf != nil { + req.rawBuf.Write(body) + } g.recordUsage(req, cand, ch, usage.Event{ IsError: true, ErrorCode: fmt.Sprintf("upstream_%d", resp.StatusCode), @@ -264,9 +304,9 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) { // Stream or buffer response；tok 从上游响应（SSE usage 块或非流式 JSON）提取。 var tok convert.TokenUsage if req.Stream { - tok = g.streamResponse(c, resp, req.Protocol, targetFormat) + tok = g.streamResponse(c, resp, req.Protocol, targetFormat, req.rawBuf) } else { - tok = g.bufferResponse(c, resp, req.Protocol, targetFormat) + tok = g.bufferResponse(c, resp, req.Protocol, targetFormat, req.rawBuf) } resp.Body.Close() // 成功记录：用量 + 定价计费。 @@ -320,6 +360,13 @@ func (g *Gateway) recordUsage(req *Request, cand *channel.Candidate, ch *store.C ev.CompletionTokens = tok.OutputTokens ev.CacheReadTokens = tok.CacheReadTokens ev.CacheCreationTokens = tok.CacheCreationTokens + // 原始请求/响应（仅管理员+开关开启时捕获）。 + if req.CaptureRaw { + ev.RawRequest = string(req.Body) + if req.rawBuf != nil { + ev.RawResponse = req.rawBuf.String() + } + } // 定价与成本（价格按每百万 token 的 USD 单价）。 // 成本口径：非缓存输入 × 输入价 + 缓存读 × 缓存价 + 缓存写与输出 × 输出价。 if ev.ModelID != 0 { @@ -385,7 +432,8 @@ func (g *Gateway) setHeaders(req *http.Request, ch *store.Channel, apiKey string // streamResponse 流式响应：按 \n\n 分块零缓冲转发；跨协议时逐行转换。 // 返回从上游 SSE usage 块累计的 token 用量（按上游协议解析）。 -func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string) convert.TokenUsage { +// capture 非 nil 时把上游原始行累积进去（原始响应记录）。 +func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string, capture *strings.Builder) convert.TokenUsage { w := c.Writer c.Header("Content-Type", "text/event-stream") c.Header("Cache-Control", "no-cache") @@ -429,6 +477,11 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt } } + // 原始响应捕获（仅管理员+开关开启时启用）。 + if capture != nil { + capture.Write(buf) + } + // 先解析用量（data: {...} 行），再决定转发内容。 for _, data := range sseDataPayloads(buf) { accum.Feed(data, upstreamProto) @@ -493,13 +546,18 @@ func sseDataPayloads(chunk []byte) [][]byte { return out } -func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string) convert.TokenUsage { +func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string, capture *strings.Builder) convert.TokenUsage { body, err := io.ReadAll(resp.Body) if err != nil { g.writeError(c, http.StatusBadGateway, "failed to read response") return convert.TokenUsage{} } + // 原始响应捕获（仅管理员+开关开启时启用）。 + if capture != nil { + capture.Write(body) + } + // 用量从上游原始响应体提取（先于转换，转换会改字段名）。 tok, _ := convert.ExtractUsageJSON(body, upstreamProto) diff --git a/backend/internal/store/models.go b/backend/internal/store/models.go index fc7ed26..5c02b39 100644 --- a/backend/internal/store/models.go +++ b/backend/internal/store/models.go @@ -174,6 +174,8 @@ type UsageLog struct { LatencyMS int `json:"latency_ms"` Status string `gorm:"size:16;not null" json:"status"` ErrorCode *string `json:"error_code,omitempty"` + RawRequest string `gorm:"type:text" json:"raw_request,omitempty"` // 客户端原始请求体（未转换；仅管理员+开关开启时记录） + RawResponse string `gorm:"type:text" json:"raw_response,omitempty"` // 上游原始响应（未转换；流式为全部 SSE 事件） CreatedAt time.Time `gorm:"index" json:"created_at"` } diff --git a/backend/internal/usage/recorder.go b/backend/internal/usage/recorder.go index 074512e..b2d4e81 100644 --- a/backend/internal/usage/recorder.go +++ b/backend/internal/usage/recorder.go @@ -32,6 +32,8 @@ type Event struct { CacheReadPrice float64 TraceID string // TraceID for distributed tracing ModelID uint64 // Model ID from channel-model binding + RawRequest string // 客户端原始请求体（仅管理员+开关开启时记录） + RawResponse string // 上游原始响应（未转换；流式为全部 SSE 事件） } // Recorder handles async usage recording @@ -153,6 +155,8 @@ func (r *Recorder) flush(events []Event) { ErrorCode: errCode, RequestID: e.RequestID, TraceID: e.TraceID, + RawRequest: e.RawRequest, + RawResponse: e.RawResponse, } logs = append(logs, log) } diff --git a/backend/middleware/auth_llm.go b/backend/middleware/auth_llm.go index f33cc51..df64de1 100644 --- a/backend/middleware/auth_llm.go +++ b/backend/middleware/auth_llm.go @@ -40,8 +40,16 @@ func AuthLLM(db *gorm.DB) gin.HandlerFunc { return } + // 附带用户角色（判断是否管理员，供原始请求/响应记录等管理能力使用）。 + var user store.User + if err := db.First(&user, apiKey.UserID).Error; err != nil { + unauthorized(c, "无效的API密钥") + return + } + c.Set("api_key", &apiKey) c.Set("user_id", apiKey.UserID) + c.Set("user_role", user.Role) c.Next() } } diff --git a/doc/flow.md b/doc/flow.md new file mode 100644 index 0000000..c76eb57 --- /dev/null +++ b/doc/flow.md @@ -0,0 +1,276 @@ +# 网关调用流程示意图 + +> 对应实现：`backend/router/setRouter.go`（路由注册）、`backend/middleware/auth_llm.go`（鉴权）、 +> `backend/internal/proxy/{gateway.go,handlers.go,convert/*}`（网关与三协议互转）、 +> `backend/internal/channel/{channel.go,health.go}`（渠道路由与健康）、 +> `backend/internal/usage/recorder.go`（用量异步落库）。 + +## 0. 总览 + +``` + ┌────────────────────────────────────────────────┐ + │ Gin Router (/v1) │ + │ ┌──────────────────────────────────────────┐ │ + 客户端 ───────────▶│ │ middleware.AuthLLM (密钥鉴权, 401拦截) │ │ + Bearer sk-ot-… │ └──────────────────────────────────────────┘ │ + │ ┌───────┬────────┬─────────┬─────────┐ │ + │ │ chat │messages│responses│ models │ │ + │ │Handle │Handle │ Handle │ Handle │ │ + │ │Chat │Messages│Responses│ Models │ │ + │ └───┬───┴───┬────┴────┬────┴─────────┘ │ + │ └───────┴────┬────┘ │ + │ ParseRequest │ + │ (model/stream) │ + │ │ │ + │ Dispatch ◀── Candidates/Pick │ + │ (转换+故障转移+用量记录) │ + └───────────────────┼──────────────────────────────┘ + │ + ▼ + 上游 /v1/* (chat|messages|responses) +``` + +## 1. 请求入口与鉴权 + +```mermaid +sequenceDiagram + autonumber + participant C as 客户端 + participant R as Gin /v1 路由 + participant A as AuthLLM + participant DB as SQLite(APIKey) + participant H as HandleXxx + + C->>R: POST /v1/chat/completions 等 + Note over R: /v1 组挂 middleware.AuthLLM + R->>A: 进入中间件 + A->>A: 提取 Bearer token（兼容无前缀直传） + alt 未携带 Authorization + A-->>C: 401 「未提供认证信息」 + else token 长度 < 12 或 prefix 不匹配 + A-->>C: 401 「无效的API密钥」 + else prefix 命中 + A->>DB: SELECT * WHERE key_prefix=? AND status=active + A->>A: sha256(token) == KeyHash ? + alt 哈希不一致 + A-->>C: 401 「无效的API密钥」 + else 校验通过 + A->>H: c.Set(api_key, user_id) → 放行 + end + end +``` + +> 关键点：`key_prefix` 取 `sk-ot-` 后前 12 位（`api.go` 创建密钥时 `keyValue[:12]`）， +> `auth_llm.go` 用同一常量 `keyPrefixLen=12` 切片，避免越界 panic。 + +## 2. 三协议主调用流程 + +```mermaid +sequenceDiagram + autonumber + participant C as 客户端 + participant H as HandleChat/Messages/Responses + participant P as ParseRequest + participant G as Dispatch + participant S as ChannelService + participant U as usage.Recorder + participant UP as 上游(OpenRouter等) + + C->>H: 请求体 (model, stream, messages/input…) + H->>P: ParseRequest(protocol) + P->>P: 读 body → 解析 model / stream + P-->>H: Request{Model, Stream, Protocol, Body, …} + H->>G: Dispatch(req) + + G->>S: Candidates(req.Model) + S-->>G: []Candidate{Channel, Binding?} + G->>S: FilterHealthy(cands) + G->>S: Pick(cands) → 加权随机选定一个候选 + Note over G,S: 绑定优先（携带 upstream_model 映射）；
无绑定回退到权重最低的健康备用渠道 + + loop 故障转移（候选耗尽前） + G->>G: conversionTarget(ch, proto) → 渠道首选协议 + alt 客户端协议 ≠ 渠道协议 + G->>G: ConvertRequest(body, from, to) 转换请求体 + Note over G: 走 convert 包（chat/messages/responses 互转） + end + alt 有 Binding.UpstreamModel + G->>G: rewriteModel(body, upstreamModel) 别名映射 + end + G->>UP: POST {base}/v1/{path} (按渠道协议拼 URL/头) + alt 连接失败 或 429/5xx + S->>S: RecordFailure(ch) → 连续2次 degraded 熔断 + G->>U: Record(error 事件, error_code) + Note over G: continue → 换下一个候选渠道 + else 4xx + G-->>C: 透传上游错误体 (不重试) + else 2xx + S->>S: RecordSuccess(ch) + alt stream=true + G->>G: streamResponse → 逐块转发 + 累计 usage + else + G->>G: bufferResponse → 整体转发 + 提取 usage + end + G->>G: recordUsage (按模型定价计算 cost) + G->>U: Record(成功事件, tokens, cost) + G-->>C: 响应 + end + end + Note over G,C: 全部候选失败 → 502/503 +``` + +> **故障转移规则**（对齐参考实现 `doProxy`）： +> - 连接错误、429、5xx → 可重试，换下一个候选； +> - 4xx（如 400 参数错误）→ 透传上游错误体，不重试； +> - 无可用渠道（全部不健康/无绑定且无备用）→ 502/503 + `error_code=no_channel`。 + +## 3. 路由选择细节 + +```mermaid +flowchart TD + A[客户端 model 名] --> B{存在启用模型行?} + B -- 是 --> C{有绑定且渠道健康?} + C -- 是 --> D[候选 = 绑定该模型的渠道
排序 priority ASC, weight DESC, id ASC] + C -- 否 --> E + B -- 否 --> E[候选 = 全部启用渠道
取权重最低的健康备用渠道] + D --> F[FilterHealthy 内存熔断过滤] + E --> F + F --> G[Pick 加权随机选中一个] + G --> H[Dispatch 开始尝试] + H --> I{尝试成功?} + I -- 失败可重试 --> J[RecordFailure + 换下一个] + J --> H + I -- 成功 --> K[RecordSuccess + 响应 + 记账] + J -. 全部耗尽 .-> L[502/503] +``` + +## 4. 跨协议转换（client ↔ 渠道原生协议） + +```mermaid +flowchart LR + subgraph 客户端协议 + CHAT[/"chat
chat/completions"/] + MSG[/"messages
(Anthropic)"/] + RESP[/"responses
(OpenAI)"/] + end + subgraph 中间模型 + MID["Chat 形状
(标准中间模型)"] + end + subgraph 渠道协议 + UCHAT[/"chat"/] + UMSG[/"messages"/] + URESP[/"responses"/] + end + + CHAT -->|直通| UCHAT + MSG -->|messagesToChat| MID -->|chatToMessages| UMSG + MSG -->|messagesToChat| MID -->|chatToResponses| URESP + RESP -->|responsesToChat| MID -->|chatToResponses| URESP + RESP -->|responsesToChat| MID -->|chatToMessages| UMSG +``` + +> 转换入口：`convert.ConvertRequest`（请求体）、`convert.ConvertResponse`（非流式响应）、 +> `convert.NewStreamTransformer`（流式 SSE 逐行转换）。跨两跳时经 Chat 中转（如 +> `responses→messages` = `responsesToChatReq` + `chatToMessagesReq`）。 + +## 5. 流式 / 非流式响应与用量提取 + +```mermaid +sequenceDiagram + autonumber + participant G as Dispatch + participant S as streamResponse + participant B as bufferResponse + participant ACC as StreamUsageAccum + participant W as 客户端 Writer + participant UP as 上游 + + alt stream=true + G->>S: streamResponse(resp, clientProto, upstreamProto) + S->>UP: 按 \n\n 读块 (bufio) + loop 每个 SSE 块 + S->>ACC: sseDataPayloads(chunk) → Feed(data, upstreamProto) + Note over ACC: 逐协议累计 usage 字段 + alt 跨协议 + S->>S: NewStreamTransformer(upstream→client).line(chunk) + end + S->>W: 写块 + Flush + alt 遇到流结束标记 + Note over S: chat: data:[DONE]
messages: message_stop
responses: response.completed + S-->>G: 返回累计 TokenUsage + end + end + else stream=false + G->>B: bufferResponse(resp, clientProto, upstreamProto) + B->>B: io.ReadAll + B->>B: ExtractUsageJSON(body, upstreamProto) + alt 跨协议 + B->>B: ConvertResponse(body, upstream→client) + else 直通 + B->>B: CleanJSON(body) 去空白/SSE注释前缀 + end + B-->>G: 返回 TokenUsage + end + G->>G: recordUsage(req, cand, ch, ev, tok) + Note over G: 定价 cost = (非缓存输入×输入价 + 缓存读×缓存价
+ 缓存写×输出价 + 输出×输出价) / 1e6 + G->>G: usageRec.Record(Event) +``` + +## 6. 用量异步落库 + +```mermaid +sequenceDiagram + autonumber + participant G as Gateway + participant R as usage.Recorder + participant U as UsageDAO + participant D as DailyUsageDAO + participant DB as SQLite + + G->>R: Record(Event) 每次请求(成功/错误/取消) + Note over R: 缓冲 channel (10000), 每 5s 或满 100 条 flush + R->>U: BatchCreate(UsageLog[]) + R->>D: UpsertDailyUsage(UsageDaily) 按(user_id,model_id,date)增量累加 + U->>DB: INSERT usage_logs + D->>DB: ON CONFLICT 累加 requests/input/output/cache/cost +``` + +> `usage_dailies` 用 `gorm.Expr("requests + ?")` 增量累加而非覆盖，保证多次 flush 不互相清零。 + +## 7. 渠道健康与熔断 + +```mermaid +flowchart TD + A[请求失败] --> B[RecordFailure: consecutive++] + B --> C{consecutive >= 2?} + C -- 是 --> D[status=degraded + 5min cooldown] + C -- 否 --> E[仅计数] + D --> F{后续请求 Candidates} + F --> G{FilterHealthy 该渠道} + G -- degraded/cooldown 未过期 --> H[排除, 走其他渠道/备用] + G -- healthy --> I[参与选择] + D -. 冷却过期 .-> J[复位 healthy] + J --> I + K[健康检查周期探测成功] --> L[RecordSuccess: 复位 healthy] +``` + +> 说明：`health.go` 的 `StartPeriodicCheck`（默认 5min）会探测各渠道 `/models`， +> 成功调 `RecordSuccess` 复位；失败调 `RecordFailure` 进入熔断计数。 + +## 8. 关键代码锚点 + +| 环节 | 位置 | +|---|---| +| /v1 路由注册 + AuthLLM | `router/setRouter.go:146` | +| 密钥鉴权 | `middleware/auth_llm.go` | +| 请求解析 | `proxy/gateway.go:104 ParseRequest` | +| 主调度 + 故障转移 | `proxy/gateway.go:157 Dispatch` | +| 流式转发 + 结束检测 | `proxy/gateway.go:388 streamResponse` | +| 非流式转发 | `proxy/gateway.go:496 bufferResponse` | +| 用量记账 | `proxy/gateway.go:302 recordUsage` | +| 候选构建 | `channel/channel.go:51 Candidates` | +| 内存健康过滤 | `channel/channel.go:252 FilterHealthy` | +| 加权选择 | `channel/channel.go:222 Pick` | +| 失败熔断 | `channel/channel.go:139 RecordFailure` | +| 三协议互转 | `proxy/convert/{convert.go,json_chat.go,json_responses.go,stream_transform.go}` | +| 用量异步落库 | `usage/recorder.go:114 flush` | diff --git a/frontend/src/views/dashboard/SystemConfig.vue b/frontend/src/views/dashboard/SystemConfig.vue index 259bcea..82dd13b 100644 --- a/frontend/src/views/dashboard/SystemConfig.vue +++ b/frontend/src/views/dashboard/SystemConfig.vue @@ -13,6 +13,7 @@ const saving = ref(false) const registrationEnabled = ref(true) const passwordLoginEnabled = ref(true) +const logRawRequests = ref(false) async function load() { loading.value = true @@ -23,6 +24,9 @@ async function load() { ]) registrationEnabled.value = regRes.data.data.enabled passwordLoginEnabled.value = pwdRes.data.data.enabled + // 原始请求/响应记录开关（通用配置键 log_raw_requests） + const cfgRes = await request.get('/admin/config') + logRawRequests.value = cfgRes.data?.data?.log_raw_requests === 'true' } catch (e) { setToast(errMsg(e), 'error') } finally { @@ -56,6 +60,19 @@ async function savePasswordLogin(enabled: boolean) { } } +async function saveLogRawRequests(enabled: boolean) { + saving.value = true + try { + await request.put('/admin/config', { log_raw_requests: enabled ? 'true' : 'false' }) + logRawRequests.value = enabled + setToast(enabled ? '已开启原始请求/响应记录' : '已关闭原始请求/响应记录', 'success') + } catch (e) { + setToast(errMsg(e), 'error') + } finally { + saving.value = false + } +} + onMounted(load) @@ -116,6 +133,30 @@ onMounted(load) + + +
+
+
+

记录原始请求/响应

+

仅对管理员账号生效：在用量明细中保存每次请求的客户端原始请求体与上游原始响应体（流式含全部 SSE 事件），用于排障。会显著增加存储。

+
+ +
+