mirrors2/openteam
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 补齐三个方向的流式工具调用；Anthropic 上游走 x-api-key；禁用模型真正下架

Browse Source 
- 流式转换补齐 chat→responses / responses→chat / messages→responses 的工具调用（此前只转文本，SDK 客户端拿不到 tool_calls）；completed 事件携带完整 output 数组
- Anthropic 协议上游同时发 x-api-key（官方 API 不接受 API key 走 Bearer）并透传 anthropic-beta
- 请求侧校验模型存在且启用：禁用模型 404 model_not_found，不再只是从 /v1/models 消失、实际照发照扣费；空模型名 400
- 模型在请求中途被删时按 0 记账留痕，不再按硬编码示例价扣用户的钱
- finishUsage 新增一行式请求日志（trace/user/model/channel/tokens/cost/status/latency），记账是异步的，日志是唯一即时排障依据
This commit is contained in:
Sakurasan committed 2026-10-29 02:56:37 +08:00
1 parent 984a31cbc5
commit d6db0f8162
6 files changed
+485 -31
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/internal/proxy/convert/convert_test.go
+114
View File
@@ -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)
}
}
server/internal/proxy/convert/stream.go
+257 -30
View File
@@ -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"],
}
}