diff --git a/server/internal/proxy/convert/convert_test.go b/server/internal/proxy/convert/convert_test.go index 4914fbc..fdfe812 100644 --- a/server/internal/proxy/convert/convert_test.go +++ b/server/internal/proxy/convert/convert_test.go @@ -1022,3 +1022,117 @@ func TestMessagesToChatReqToolOrder(t *testing.T) { t.Fatalf("msg[0] missing tool_calls: %s", out) } } + +// --------------------------------------------------------------------------- +// 流式工具调用:三个此前缺失的方向(chat→responses / responses→chat / messages→responses) + +func TestStreamChatToResponsesToolCall(t *testing.T) { + tf := newChatToResponses().line + out := feedLines(t, tf, []string{ + `data: {"id":"chatcmpl-1","object":"chat.completion.chunk","model":"gpt-4o","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}` + "\n\n", + `data: {"id":"chatcmpl-1","object":"chat.completion.chunk","model":"gpt-4o","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_9","type":"function","function":{"name":"get_weather","arguments":""}}]},"finish_reason":null}]}` + "\n\n", + `data: {"id":"chatcmpl-1","object":"chat.completion.chunk","model":"gpt-4o","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"city\":\"北京\"}"}}]},"finish_reason":null}]}` + "\n\n", + `data: {"id":"chatcmpl-1","object":"chat.completion.chunk","model":"gpt-4o","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}` + "\n\n", + `data: {"id":"chatcmpl-1","object":"chat.completion.chunk","model":"gpt-4o","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":5}}` + "\n\n", + "data: [DONE]\n\n", + }) + if !strings.Contains(out, "event: response.output_item.added") || !strings.Contains(out, `"name":"get_weather"`) { + t.Fatalf("missing function_call item added: %s", out) + } + if !strings.Contains(out, "event: response.function_call_arguments.delta") { + t.Fatalf("missing arguments delta: %s", out) + } + if !strings.Contains(out, "event: response.output_item.done") { + t.Fatalf("missing output_item.done: %s", out) + } + // completed 的 output 数组要携带完整参数,客户端不依赖事件也能拿到最终形态 + if !strings.Contains(out, `"call_id":"call_9"`) { + t.Fatalf("missing call_id: %s", out) + } + if !strings.Contains(out, `\"city\":\"北京\"`) { + t.Fatalf("missing accumulated arguments in completed output: %s", out) + } + if !strings.Contains(out, `"usage":{"completion_tokens":5,"prompt_tokens":10}`) { + t.Fatalf("missing usage in completed: %s", out) + } +} + +func TestStreamResponsesToChatToolCall(t *testing.T) { + tf := newResponsesToChat().line + out := feedLines(t, tf, []string{ + "event: response.created\n", + `data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-4o"}}` + "\n\n", + "event: response.output_item.added\n", + `data: {"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"get_weather","arguments":""}}` + "\n\n", + "event: response.function_call_arguments.delta\n", + `data: {"type":"response.function_call_arguments.delta","item_id":"fc_1","delta":"{\"city\":\"北京\"}"}` + "\n\n", + "event: response.completed\n", + `data: {"type":"response.completed","response":{"id":"resp_1","status":"completed","usage":{"input_tokens":7,"output_tokens":8}}}` + "\n\n", + }) + if !strings.Contains(out, `"tool_calls"`) || !strings.Contains(out, `"name":"get_weather"`) { + t.Fatalf("missing tool_calls chunk: %s", out) + } + if !strings.Contains(out, `"arguments":"{\"city\":\"北京\"}"`) { + t.Fatalf("missing arguments delta chunk: %s", out) + } + if !strings.Contains(out, `"finish_reason":"tool_calls"`) { + t.Fatalf("finish_reason should be tool_calls: %s", out) + } + if !strings.Contains(out, `"usage":{"input_tokens":7,"output_tokens":8}`) { + t.Fatalf("missing usage chunk: %s", out) + } + if !strings.Contains(out, "data: [DONE]") { + t.Fatalf("missing [DONE]: %s", out) + } +} + +func TestStreamMessagesToResponsesToolCall(t *testing.T) { + tf := newMessagesToResponses().line + out := feedLines(t, tf, []string{ + "event: message_start\n", + `data: {"type":"message_start","message":{"id":"msg_1","model":"claude-sonnet-5"}}` + "\n\n", + "event: content_block_start\n", + `data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_1","name":"get_weather"}}` + "\n\n", + "event: content_block_delta\n", + `data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"city\":\"北京\"}"}}` + "\n\n", + "event: content_block_stop\n", + `data: {"type":"content_block_stop","index":0}` + "\n\n", + "event: message_delta\n", + `data: {"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"input_tokens":10,"output_tokens":5}}` + "\n\n", + "event: message_stop\n", + `data: {"type":"message_stop"}` + "\n\n", + }) + if !strings.Contains(out, "event: response.output_item.added") || !strings.Contains(out, `"call_id":"toolu_1"`) { + t.Fatalf("missing function_call item added: %s", out) + } + if !strings.Contains(out, "event: response.function_call_arguments.delta") { + t.Fatalf("missing arguments delta: %s", out) + } + if !strings.Contains(out, "event: response.output_item.done") { + t.Fatalf("missing output_item.done: %s", out) + } + if !strings.Contains(out, `"call_id":"toolu_1"`) || !strings.Contains(out, `"arguments":"{\"city\":\"北京\"}"`) { + t.Fatalf("completed output array missing function_call with full arguments: %s", out) + } + if !strings.Contains(out, `"usage":{"input_tokens":10,"output_tokens":5}`) { + t.Fatalf("missing usage in completed: %s", out) + } +} + +func TestStreamResponsesToChatTextStillWorks(t *testing.T) { + tf := newResponsesToChat().line + out := feedLines(t, tf, []string{ + "event: response.created\n", + `data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-4o"}}` + "\n\n", + "event: response.output_text.delta\n", + `data: {"type":"response.output_text.delta","delta":"hi"}` + "\n\n", + "event: response.completed\n", + `data: {"type":"response.completed","response":{"id":"resp_1","status":"completed","usage":{"input_tokens":1,"output_tokens":1}}}` + "\n\n", + }) + if !strings.Contains(out, `"content":"hi"`) { + t.Fatalf("missing text chunk: %s", out) + } + if !strings.Contains(out, `"finish_reason":"stop"`) { + t.Fatalf("missing finish chunk: %s", out) + } +} diff --git a/server/internal/proxy/convert/stream.go b/server/internal/proxy/convert/stream.go index fe2732a..07af4f7 100644 --- a/server/internal/proxy/convert/stream.go +++ b/server/internal/proxy/convert/stream.go @@ -2,6 +2,7 @@ package convert import ( "encoding/json" + "strconv" "strings" ) @@ -443,9 +444,17 @@ type messagesToResponses struct { model string usage any done bool + + // 工具调用:messages tool_use block index → function_call 输出条目。 + // items 按首次出现顺序累积,response.completed 携带完整 output 数组。 + items []map[string]any + toolItem map[int]map[string]any + textSB strings.Builder } -func newMessagesToResponses() *messagesToResponses { return &messagesToResponses{} } +func newMessagesToResponses() *messagesToResponses { + return &messagesToResponses{toolItem: map[int]map[string]any{}} +} func (t *messagesToResponses) line(line []byte) []byte { isData, data, done := t.parseLine(line) @@ -479,21 +488,89 @@ func (t *messagesToResponses) line(line []byte) []byte { "id": "resp_" + strings.TrimPrefix(rid, "msg_"), "object": "response", "model": t.model, "status": "in_progress", }, })) + case "content_block_start": + cb, _ := m["content_block"].(map[string]any) + if cb == nil || cb["type"] != "tool_use" { + return nil + } + blockIdx, _ := m["index"].(float64) + toolID, _ := cb["id"].(string) + name, _ := cb["name"].(string) + item := map[string]any{ + "type": "function_call", "id": "fc_" + strconv.Itoa(len(t.items)), "call_id": toolID, + "name": name, "arguments": "", "status": "in_progress", + } + t.toolItem[int(blockIdx)] = item + oi := len(t.items) + t.items = append(t.items, item) + out = append(out, eventLine("response.output_item.added", map[string]any{ + "type": "response.output_item.added", "output_index": oi, "item": item, + })) case "content_block_delta": delta, _ := m["delta"].(map[string]any) - text, _ := delta["text"].(string) - if text != "" { - out = append(out, eventLine("response.output_text.delta", map[string]any{ - "type": "response.output_text.delta", "delta": text, "item_id": "msg_1", "output_index": 0, "content_index": 0, + deltaType, _ := delta["type"].(string) + if deltaType == "input_json_delta" { + blockIdx, _ := m["index"].(float64) + item, ok := t.toolItem[int(blockIdx)] + if !ok { + return nil + } + partial, _ := delta["partial_json"].(string) + if partial == "" { + return nil + } + item["arguments"] = item["arguments"].(string) + partial + out = append(out, eventLine("response.function_call_arguments.delta", map[string]any{ + "type": "response.function_call_arguments.delta", "item_id": item["id"], + "output_index": itemIndexByID(t.items, item["id"].(string)), "delta": partial, })) + return joinLines(out) } + text, _ := delta["text"].(string) + if text == "" { + return nil + } + t.textSB.WriteString(text) + if len(t.items) == 0 || t.items[0]["type"] != "message" { + t.items = append([]map[string]any{{ + "type": "message", "id": "msg_1", "role": "assistant", "status": "in_progress", "content": []any{}, + }}, t.items...) + } + out = append(out, eventLine("response.output_text.delta", map[string]any{ + "type": "response.output_text.delta", "delta": text, "item_id": "msg_1", "output_index": 0, "content_index": 0, + })) + case "content_block_stop": + blockIdx, _ := m["index"].(float64) + item, ok := t.toolItem[int(blockIdx)] + if !ok { + return nil + } + out = append(out, eventLine("response.output_item.done", map[string]any{ + "type": "response.output_item.done", "output_index": itemIndexByID(t.items, item["id"].(string)), + "item": functionCallDone(item), + })) case "message_stop": if !t.done { t.done = true + // 文本条目收尾(纯工具调用流可能没有文本条目) + if len(t.items) > 0 && t.items[0]["type"] == "message" { + out = append(out, eventLine("response.output_item.done", map[string]any{ + "type": "response.output_item.done", "output_index": 0, + "item": messageItemDone("msg_1", t.textSB.String()), + })) + } + output := make([]any, 0, len(t.items)) + for _, item := range t.items { + if item["type"] == "message" { + output = append(output, messageItemDone("msg_1", t.textSB.String())) + continue + } + output = append(output, functionCallDone(item)) + } out = append(out, eventLine("response.completed", map[string]any{ "type": "response.completed", "response": map[string]any{ - "id": "resp_stream", "object": "response", "model": t.model, "status": "completed", "usage": t.usage, + "id": "resp_stream", "object": "response", "model": t.model, "status": "completed", "usage": t.usage, "output": output, }, })) } @@ -507,9 +584,16 @@ func (t *messagesToResponses) line(line []byte) []byte { type responsesToChat struct { sseState id, model string + + // 工具调用:responses function_call 条目 → chat tool_calls index(并行调用各占一个 index) + toolIdx map[string]int + nextTool int + anyTool bool } -func newResponsesToChat() *responsesToChat { return &responsesToChat{} } +func newResponsesToChat() *responsesToChat { + return &responsesToChat{toolIdx: map[string]int{}} +} func (t *responsesToChat) line(line []byte) []byte { isData, data, done := t.parseLine(line) @@ -538,19 +622,61 @@ func (t *responsesToChat) line(line []byte) []byte { })) case "response.output_text.delta": delta, _ := m["delta"].(string) - if delta != "" { - out = append(out, dataLine(map[string]any{ - "id": "chatcmpl-" + strings.TrimPrefix(t.id, "resp_"), "object": "chat.completion.chunk", "model": t.model, - "choices": []any{map[string]any{"index": 0, "delta": map[string]any{"content": delta}, "finish_reason": nil}}, - })) + if delta == "" { + return nil } - case "response.completed": out = append(out, dataLine(map[string]any{ "id": "chatcmpl-" + strings.TrimPrefix(t.id, "resp_"), "object": "chat.completion.chunk", "model": t.model, - "choices": []any{map[string]any{"index": 0, "delta": map[string]any{}, "finish_reason": "stop"}}, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{"content": delta}, "finish_reason": nil}}, })) - if u, ok := m["response"].(map[string]any); ok { - if usage, ok := u["usage"]; ok { + case "response.output_item.added": + item, _ := m["item"].(map[string]any) + if item == nil || item["type"] != "function_call" { + return nil + } + itemID, _ := item["id"].(string) + callID, _ := item["call_id"].(string) + if callID == "" { + callID = itemID + } + name, _ := item["name"].(string) + idx := t.nextTool + t.nextTool++ + t.toolIdx[itemID] = idx + t.anyTool = true + out = append(out, dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "resp_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{ + "tool_calls": []any{map[string]any{"index": idx, "id": callID, "type": "function", "function": map[string]any{"name": name, "arguments": ""}}}, + }, "finish_reason": nil}}, + })) + case "response.function_call_arguments.delta": + itemID, _ := m["item_id"].(string) + idx, ok := t.toolIdx[itemID] + if !ok { + return nil + } + delta, _ := m["delta"].(string) + if delta == "" { + return nil + } + out = append(out, dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "resp_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{ + "tool_calls": []any{map[string]any{"index": idx, "function": map[string]any{"arguments": delta}}}, + }, "finish_reason": nil}}, + })) + case "response.completed": + stop := "stop" + if t.anyTool { + stop = "tool_calls" + } + out = append(out, dataLine(map[string]any{ + "id": "chatcmpl-" + strings.TrimPrefix(t.id, "resp_"), "object": "chat.completion.chunk", "model": t.model, + "choices": []any{map[string]any{"index": 0, "delta": map[string]any{}, "finish_reason": stop}}, + })) + if resp, ok := m["response"].(map[string]any); ok { + if usage, ok := resp["usage"]; ok { out = append(out, dataLine(map[string]any{ "id": "chatcmpl-" + strings.TrimPrefix(t.id, "resp_"), "object": "chat.completion.chunk", "model": t.model, "choices": []any{}, "usage": usage, @@ -571,9 +697,17 @@ type chatToResponses struct { usage any finishSeen bool done bool + + // 工具调用:chat tool_calls index → function_call 输出条目(arguments 就地累积)。 + // items 按首次出现顺序累积,response.completed 携带完整 output 数组。 + items []map[string]any + toolItem map[int]map[string]any + textSB strings.Builder } -func newChatToResponses() *chatToResponses { return &chatToResponses{} } +func newChatToResponses() *chatToResponses { + return &chatToResponses{toolItem: map[int]map[string]any{}} +} func (t *chatToResponses) line(line []byte) []byte { isData, data, done := t.parseLine(line) @@ -584,12 +718,7 @@ func (t *chatToResponses) line(line []byte) []byte { // 流结束兜底:finish 后 usage 未随块到达时在此补发 completed if !t.done { t.done = true - return eventLine("response.completed", map[string]any{ - "type": "response.completed", - "response": map[string]any{ - "id": "resp_stream", "object": "response", "model": t.model, "status": "completed", "usage": t.usage, - }, - }) + return t.completed() } return nil } @@ -616,24 +745,122 @@ func (t *chatToResponses) line(line []byte) []byte { var out [][]byte if role, _ := delta["role"].(string); role == "assistant" { out = append(out, eventLine("response.created", map[string]any{ - "type": "response.created", + "type": "response.created", "response": map[string]any{"id": "resp_stream", "object": "response", "model": t.model, "status": "in_progress"}, })) } if content, _ := delta["content"].(string); content != "" { + t.textSB.WriteString(content) + if len(t.items) == 0 || t.items[0]["type"] != "message" { + // 文本条目插入队首,保持 output_index 与 items 位置一致 + t.items = append([]map[string]any{{ + "type": "message", "id": "msg_1", "role": "assistant", "status": "in_progress", "content": []any{}, + }}, t.items...) + } out = append(out, eventLine("response.output_text.delta", map[string]any{ "type": "response.output_text.delta", "delta": content, "item_id": "msg_1", "output_index": 0, "content_index": 0, })) } + // 工具调用:新 index 开 function_call 条目(output_item.added),参数分段经 + // function_call_arguments.delta 透出(与 OpenAI 原生 Responses 流的事件形状一致) + if tcs, ok := delta["tool_calls"].([]any); ok { + for _, tc := range tcs { + call, ok := tc.(map[string]any) + if !ok { + continue + } + ci, _ := call["index"].(float64) + ti := int(ci) + fn, _ := call["function"].(map[string]any) + name, _ := fn["name"].(string) + args, _ := fn["arguments"].(string) + item, seen := t.toolItem[ti] + if !seen { + callID, _ := call["id"].(string) + item = map[string]any{ + "type": "function_call", "id": "fc_" + strconv.Itoa(len(t.items)), "call_id": callID, + "name": name, "arguments": "", "status": "in_progress", + } + t.toolItem[ti] = item + oi := len(t.items) + t.items = append(t.items, item) + out = append(out, eventLine("response.output_item.added", map[string]any{ + "type": "response.output_item.added", "output_index": oi, "item": item, + })) + } + if args != "" { + item["arguments"] = item["arguments"].(string) + args + out = append(out, eventLine("response.function_call_arguments.delta", map[string]any{ + "type": "response.function_call_arguments.delta", "item_id": item["id"], + "output_index": itemIndexByID(t.items, item["id"].(string)), "delta": args, + })) + } + } + } // 上游 usage 块(choices 为空)通常晚于 finish_reason:此时再发 completed,携带 usage if _, hasUsage := m["usage"]; hasUsage && t.finishSeen && !t.done { t.done = true - out = append(out, eventLine("response.completed", map[string]any{ - "type": "response.completed", - "response": map[string]any{ - "id": "resp_stream", "object": "response", "model": t.model, "status": "completed", "usage": t.usage, - }, - })) + out = append(out, t.completed()) } return joinLines(out) } + +// itemIndexByID 返回条目在 items 中的位置(output_index)。未知 id 返回 0。 +func itemIndexByID(items []map[string]any, id string) int { + for i, it := range items { + if s, _ := it["id"].(string); s == id { + return i + } + } + return 0 +} + +// messageItemDone 文本条目的终态(流式期间 content 逐段累积,done 时一次性补齐)。 +func messageItemDone(id, text string) map[string]any { + return map[string]any{ + "type": "message", "id": id, "role": "assistant", "status": "completed", + "content": []any{map[string]any{"type": "output_text", "text": text}}, + } +} + +// completed 收尾:先给每个条目发 output_item.done,再发带完整 output 数组的 response.completed。 +func (t *chatToResponses) completed() []byte { + var out [][]byte + for i, item := range t.items { + var final map[string]any + if item["type"] == "message" { + final = messageItemDone(item["id"].(string), t.textSB.String()) + } else { + final = functionCallDone(item) + } + out = append(out, eventLine("response.output_item.done", map[string]any{ + "type": "response.output_item.done", "output_index": i, "item": final, + })) + } + out = append(out, eventLine("response.completed", map[string]any{ + "type": "response.completed", + "response": map[string]any{"id": "resp_stream", "object": "response", "model": t.model, "status": "completed", "usage": t.usage, "output": t.outputArray()}, + })) + return joinLines(out) +} + +// outputArray 组装 response.completed 携带的 output 数组(与 output_item.done 的终态一致)。 +func (t *chatToResponses) outputArray() []any { + o := make([]any, 0, len(t.items)) + for _, item := range t.items { + if item["type"] == "message" { + o = append(o, messageItemDone(item["id"].(string), t.textSB.String())) + continue + } + o = append(o, functionCallDone(item)) + } + return o +} + +// functionCallDone 工具调用条目的终态。 +func functionCallDone(item map[string]any) map[string]any { + return map[string]any{ + "type": "function_call", "id": item["id"], "call_id": item["call_id"], + "name": item["name"], "arguments": item["arguments"], + } +} diff --git a/server/internal/proxy/gateway.go b/server/internal/proxy/gateway.go index 9dff553..60005fd 100644 --- a/server/internal/proxy/gateway.go +++ b/server/internal/proxy/gateway.go @@ -128,6 +128,23 @@ func (g *Gateway) checkModelAllowed(c *gin.Context, u *store.User, model string) return true } +// checkModelExists 拦截不存在或已禁用的模型。/v1/models 只列启用的模型,请求侧口径必须一致: +// 禁用模型不能只是从列表消失——否则照样能调用、照样计费,管理员的下架操作形同虚设。 +func (g *Gateway) checkModelExists(c *gin.Context, model string) bool { + if model == "" { + apiError(c, http.StatusBadRequest, "invalid_request_error", "model is required") + return false + } + var n int64 + g.db.Model(&store.Model{}).Where("name = ? AND enabled = ?", model, true).Count(&n) + if n == 0 { + apiError(c, http.StatusNotFound, "model_not_found", + fmt.Sprintf("The model %q does not exist or is not enabled", model)) + return false + } + return true +} + func contains(list []string, s string) bool { for _, v := range list { if v == s { diff --git a/server/internal/proxy/handlers.go b/server/internal/proxy/handlers.go index b7e5f2f..6372c76 100644 --- a/server/internal/proxy/handlers.go +++ b/server/internal/proxy/handlers.go @@ -38,6 +38,9 @@ func (g *Gateway) chatCompletions(c *gin.Context) { c.Set("protocol", convert.ProtoChat) c.Set("model_name", br.Model) g.recordRawRequest(c, u, body) + if !g.checkModelExists(c, br.Model) { + return + } if !g.checkModelAllowed(c, u, br.Model) { apiError(c, http.StatusForbidden, "model_not_allowed", "模型未对你开放,请联系管理员") return @@ -71,6 +74,9 @@ func (g *Gateway) responses(c *gin.Context) { c.Set("protocol", convert.ProtoResponses) c.Set("model_name", br.Model) g.recordRawRequest(c, u, body) + if !g.checkModelExists(c, br.Model) { + return + } if !g.checkModelAllowed(c, u, br.Model) { apiError(c, http.StatusForbidden, "model_not_allowed", "模型未对你开放,请联系管理员") return @@ -104,6 +110,9 @@ func (g *Gateway) messages(c *gin.Context) { c.Set("protocol", convert.ProtoMessages) c.Set("model_name", br.Model) g.recordRawRequest(c, u, body) + if !g.checkModelExists(c, br.Model) { + return + } if !g.checkModelAllowed(c, u, br.Model) { apiError(c, http.StatusForbidden, "model_not_allowed", "模型未对你开放,请联系管理员") return diff --git a/server/internal/proxy/passthrough.go b/server/internal/proxy/passthrough.go index fc59134..1e7e86e 100644 --- a/server/internal/proxy/passthrough.go +++ b/server/internal/proxy/passthrough.go @@ -10,6 +10,7 @@ import ( "errors" "fmt" "io" + "log" "net/http" "strconv" "strings" @@ -305,7 +306,13 @@ func (g *Gateway) proxyOne(c *gin.Context, ch *store.Channel, plan *upstreamPlan req.Header.Set("User-Agent", ua) } if plan.path == "/messages" { + // Anthropic 官方 API 只认 x-api-key(Bearer 是 OAuth token 的用法), + // 兼容网关两种都收;带上 anthropic-version 与客户端声明的 beta 头。 + req.Header.Set("x-api-key", upKey) req.Header.Set("anthropic-version", "2023-06-01") + if beta := c.GetHeader("anthropic-beta"); beta != "" { + req.Header.Set("anthropic-beta", beta) + } } for _, h := range []string{"OpenAI-Organization", "OpenAI-Project", "OpenAI-Beta"} { if v := c.GetHeader(h); v != "" { @@ -722,7 +729,9 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time float64(cacheRead)/1e6*model.CacheReadPrice + float64(cacheCreate)/1e6*model.CacheWritePriceOr() } else { - cost = float64(in+cacheRead+cacheCreate)/1e6*0.15 + float64(out)/1e6*0.60 // 无定价模型时按示例价 + // 模型在记账时已被删除(请求进行中被下架):无法定价,按 0 记账留痕。 + // 虚构单价扣用户的钱既不可审计也无法对账。 + log.Printf("proxy: trace=%s model %q not found at billing time, cost recorded as 0", trace, mn) } // 落库的 input_tokens 存输入总量(含缓存读/写),与上游 message_start 口径一致,便于对账展示。 in += cacheRead + cacheCreate @@ -793,6 +802,11 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time RawResponse: rawResp, CreatedAt: time.Now().UTC(), }) + + // 请求级一行日志:usage_logs 落库是异步的且可能丢,排障时日志是唯一即时依据 + log.Printf("proxy: trace=%s user=%d key=%d model=%q channel=%d proto=%s in=%d out=%d cache_r=%d cache_w=%d cost=%.6f status=%s http=%d latency=%dms err=%s", + traceStr, uidVal, kidVal, mn, chID, p, in, out, cacheRead, cacheCreate, cost, status, + c.Writer.Status(), latency, errCode) } // recordError 失败请求的记账(不产生扣费,status=error)。 diff --git a/server/internal/proxy/proxy_test.go b/server/internal/proxy/proxy_test.go index 68e9c1c..475831d 100644 --- a/server/internal/proxy/proxy_test.go +++ b/server/internal/proxy/proxy_test.go @@ -605,3 +605,76 @@ func TestAllChannelsFailedAttributesLastChannel(t *testing.T) { t.Fatalf("失败不应产生扣费:%+v", l) } } + +// Anthropic 原生上游必须收到 x-api-key(官方 API 不接受 API key 走 Bearer)。 +func TestAnthropicUpstreamUsesXAPIKey(t *testing.T) { + var gotXAPIKey, gotAuth, gotVersion string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotXAPIKey = r.Header.Get("x-api-key") + gotAuth = r.Header.Get("Authorization") + gotVersion = r.Header.Get("anthropic-version") + _, _ = io.WriteString(w, `{"id":"msg_1","type":"message","role":"assistant","model":"m1","content":[{"type":"text","text":"hi"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}`) + })) + defer srv.Close() + + h := newHarness(t, srv.URL, 3000, true, nil) + // 把渠道改造成 Anthropic 原生(messages 协议) + if err := h.db.Model(&store.Channel{}).Where("id = ?", 1).Updates(map[string]any{ + "provider": store.ChannelProviderAnthropic, + "formats": []string{store.FormatMessages}, + }).Error; err != nil { + t.Fatal(err) + } + + // Anthropic 客户端走 x-api-key 头 + /v1/messages + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + req := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(`{"model":"m1","messages":[{"role":"user","content":"hi"}]}`)) + req.Header.Set("x-api-key", h.key) + req.Header.Set("anthropic-version", "2023-06-01") + req.Header.Set("anthropic-beta", "prompt-caching-2024-07-31") + c.Request = req + h.gw.Auth(c) + if !c.IsAborted() { + h.gw.Handle(c) + } + if w.Code != http.StatusOK { + t.Fatalf("messages 请求应成功,实际 %d: %s", w.Code, w.Body.String()) + } + if gotXAPIKey != "upstream-key" { + t.Fatalf("上游应收到 x-api-key=upstream-key,实际 %q", gotXAPIKey) + } + if gotAuth != "Bearer upstream-key" { + t.Fatalf("兼容网关仍需 Bearer 头,实际 %q", gotAuth) + } + if gotVersion == "" { + t.Fatal("上游应收到 anthropic-version") + } +} + +// 禁用/不存在的模型必须 404:不能只是从 /v1/models 列表消失、请求照样计费。 +func TestDisabledOrUnknownModelIsRejected(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Fatal("禁用模型不应触达上游") + })) + defer srv.Close() + + h := newHarness(t, srv.URL, 3000, true, nil) + if err := h.db.Model(&store.Model{}).Where("id = ?", 1).Update("enabled", false).Error; err != nil { + t.Fatal(err) + } + + w := h.run(t, nil, `{"model":"m1","messages":[{"role":"user","content":"hi"}]}`) + if w.Code != http.StatusNotFound { + t.Fatalf("禁用模型应 404,实际 %d: %s", w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "model_not_found") { + t.Fatalf("错误码应为 model_not_found: %s", w.Body.String()) + } + + // 空模型名直接 400 + w = h.run(t, nil, `{"messages":[{"role":"user","content":"hi"}]}`) + if w.Code != http.StatusBadRequest { + t.Fatalf("空模型应 400,实际 %d: %s", w.Code, w.Body.String()) + } +}