From 1ee094ffa176494201a4091b143458ed8097dbb8 Mon Sep 17 00:00:00 2001 From: CangShui Date: Sun, 13 Sep 2026 17:43:01 +0800 Subject: [PATCH] feat: OpenAI Responses API + session normalization (v1.8.1) - add /v1/responses (and /responses): OpenAI Responses API support - streaming semantic events, non-streaming aggregation, function tools - input item / content / tool / tool_choice conversion to chat format - ensure first message is system: fix upstream 11128 "first message is not system prompt" for non-harness clients (assistant/tool-leading or missing system); normalize developer role - fix tool call delta merging (applyToolCallDelta) across streamed chunks - refactor handleChatCompletions: extract upstreamChat for 429 cooldown fallback across the account pool - bump version to 1.8.1 - tests: 37 passing (session normalization, responses conversion, tool call merge, responses<->chat mapping) --- README.md | 3 +- main.go | 433 ++++++++++++++++++++----------- main_test.go | 309 +++++++++++++++++++++++ responses.go | 699 +++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 1299 insertions(+), 145 deletions(-) create mode 100644 responses.go diff --git a/README.md b/README.md index 6678c77..691632b 100644 --- a/README.md +++ b/README.md @@ -21,9 +21,10 @@ - **429 频率限制自动冷却**:任一账号触发上游频率限制(HTTP 429 / code 6004,中英文消息均可识别)时,自动解析消息中的重置时间(中文如 `将在 2026-09-04 07:48:15 UTC+8 重置`,英文如 `will reset at 2026-09-05 01:57:00 UTC+8`),将该账号屏蔽至重置时间;冷却期间自动改用其他账号代偿,冷却到期自动恢复。 - **授权失效自动禁用**:账号授权过期、被撤销或令牌刷新失败(HTTP 401/403 / invalid token / 登录已过期)时,自动将该账号**禁止调度并删除凭据文件**,同时写入持久化失效标记;`status` / 启动日志会明确提示该账号失效原因与重新登录命令,重新 `login` 后自动恢复调度。 - **后台自动续期**:运行期间每 5 分钟检查所有账号 Token,距过期不足 15 分钟自动刷新并持久化回各自凭据文件。 -- **OpenAI 兼容协议**:`POST /v1/chat/completions`(SSE 流式 + 非流式聚合)、`GET /v1/models`、`GET /health`。 +- **OpenAI 兼容协议**:`POST /v1/chat/completions`(SSE 流式 + 非流式聚合)、`POST /v1/responses`(OpenAI Responses API,支持流式语义事件、非流式与 function tools)、`GET /v1/models`、`GET /health`。 - **深度思考透传规则**:仅当客户端显式请求 `reasoning_effort` 时转发;绝不强制注入,避免触发上游内容安全策略。 - **反审查净化**:自动改写 Claude Code 等框架被上游逐字拉黑的固定 Prompt 语句。 +- **会话结构自动归一化**:自动保证首条消息为 `system`,修复部分非 harness 客户端(以 `assistant` 续写或以 `tool` 回传工具结果开头)触发的上游 `first message is not system prompt` (code 11128) 报错;同时兼容 OpenAI 新版 `developer` 角色。 - **单账号串行化**:同一账号请求自动排队,避免并发双发触发上游风控;不同账号之间可并行。 ## 快速上手 diff --git a/main.go b/main.go index 4bd7c1a..f6ccd73 100644 --- a/main.go +++ b/main.go @@ -31,10 +31,14 @@ import ( ) const ( - version = "1.7.0" + version = "1.8.1" // 状态快照文件名:serve 后台周期写入,monitor 前台命令实时读取展示 statusSnapshotFile = "workbuddy-status.json" + + // defaultSystemPrompt 是当客户端首条消息不是 system 时注入的保底系统提示, + // 用于满足腾讯上游「首条消息必须是 system prompt」的硬性要求 (code 11128)。 + defaultSystemPrompt = "You are a helpful assistant." ) // ----------------------------------------------------------------------------- @@ -1540,6 +1544,8 @@ func runServe() { mux := http.NewServeMux() mux.HandleFunc("/v1/chat/completions", handleChatCompletions) mux.HandleFunc("/chat/completions", handleChatCompletions) + mux.HandleFunc("/v1/responses", handleResponses) + mux.HandleFunc("/responses", handleResponses) mux.HandleFunc("/v1/models", handleModels) mux.HandleFunc("/models", handleModels) mux.HandleFunc("/health", handleHealth) @@ -1678,6 +1684,104 @@ func authMiddleware(next http.Handler) http.Handler { // 路由处理: /v1/chat/completions (支持任意 model 透传) // ----------------------------------------------------------------------------- +// upstreamChat 完成「多账号轮询 + 429 冷却代偿 + 授权失效禁用 + 单账号串行」的上游调度。 +// 成功时返回 200 响应(调用方负责关闭 Body)与命中的账号/站点;失败时函数内部已写回 +// 错误响应并返回 ok=false。Chat Completions 与 Responses 两个入口共用此逻辑。 +func upstreamChat(w http.ResponseWriter, r *http.Request, reqID uint64, upstreamBytes []byte, startTime time.Time) (*http.Response, *Account, *upstreamProfile, bool) { + accountMu.Lock() + poolSize := len(accounts) + accountMu.Unlock() + if poolSize == 0 { + writeOpenAIError(w, http.StatusUnauthorized, "no_auth", "未找到有效登录凭据,请先执行 login 命令扫码登录") + return nil, nil, nil, false + } + + var lastRateErr string + var lastAuthErr string + for attempt := 0; attempt < poolSize; attempt++ { + acc, err := nextAccount() + if err != nil { + // 所有账号均不可用(冷却或失效) + msg := fmt.Sprintf("无可用账号: %v", err) + if lastAuthErr != "" { + msg += " | 最近一次授权失效: " + truncate(lastAuthErr, 200) + } + if lastRateErr != "" { + msg += " | 最近一次频率限制: " + truncate(lastRateErr, 200) + } + log.Printf("[#%d] %s", reqID, msg) + writeOpenAIError(w, http.StatusServiceUnavailable, "no_available_account", msg) + return nil, nil, nil, false + } + _ = ensureValidTokenFor(acc) + // 令牌刷新时若发现授权失效会禁用账号;若被禁用则跳过换下一个 + accountMu.Lock() + disabled := acc.Disabled + accountMu.Unlock() + if disabled { + lastAuthErr = acc.DisabledReason + continue + } + + // 按账号所属站点(国内站/国际站)路由上游与指纹 Header + prof := acc.Profile() + + upstreamReq, err := http.NewRequestWithContext(r.Context(), http.MethodPost, prof.chatURL(), bytes.NewReader(upstreamBytes)) + if err != nil { + writeOpenAIError(w, http.StatusInternalServerError, "req_create_error", err.Error()) + return nil, nil, nil, false + } + // 注入 CodeBuddy 凭据与指纹 Header + backendHeaders(upstreamReq, acc.Auth, prof) + + // 限制同一账号向腾讯上游的请求严格单并发串行排队,防止并发双发触发腾讯风控 + acc.lock.Lock() + resp, err := cfg.HttpClient.Do(upstreamReq) + acc.lock.Unlock() + if err != nil { + log.Printf("[#%d] 账号 %s [%s] 上游请求失败: %v", reqID, acc.Path, prof.Label, err) + writeOpenAIError(w, http.StatusBadGateway, "upstream_network_error", fmt.Sprintf("网络转发失败: %v", err)) + return nil, nil, nil, false + } + + // 上游非 200 响应处理 + if resp.StatusCode >= 400 { + errBody, _ := io.ReadAll(resp.Body) + resp.Body.Close() + errStr := string(errBody) + log.Printf("[#%d] 账号 %s [%s] 上游返回 HTTP %d: %s (耗时 %v)", reqID, acc.Path, prof.Label, resp.StatusCode, errStr, time.Since(startTime)) + + if isRateLimited(resp.StatusCode, errStr) { + // 429 频率限制:解析重置时间并屏蔽该账号,交由其他账号代偿 + until, ok := parseResetTime(errStr) + if !ok { + until = time.Now().Add(60 * time.Second) // 无法解析时默认冷却 60 秒 + } + markCooldown(acc, until, errStr) + lastRateErr = errStr + continue // 尝试下一个账号 + } + + if isAuthFailure(resp.StatusCode, errStr) { + // 授权失效(401/403 / token 无效 / 登录过期):禁用该账号并删除凭据文件, + // 自动改用下一个可用账号,控制台提示用户重新登录 + disableAccount(acc, fmt.Sprintf("上游鉴权失败 (HTTP %d): %s", resp.StatusCode, truncate(errStr, 200))) + lastAuthErr = errStr + continue // 尝试下一个账号 + } + + writeOpenAIError(w, resp.StatusCode, "upstream_error", fmt.Sprintf("upstream %d: %s", resp.StatusCode, errStr)) + return nil, nil, nil, false + } + + return resp, acc, prof, true + } + + // 理论上不可达(poolSize 次尝试后未成功即已在循环内返回) + writeOpenAIError(w, http.StatusTooManyRequests, "all_accounts_cooldown", "所有账号均处于冷却状态") + return nil, nil, nil, false +} + func handleChatCompletions(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { writeOpenAIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "仅支持 POST 请求") @@ -1719,6 +1823,10 @@ func handleChatCompletions(w http.ResponseWriter, r *http.Request) { // 模板净化:改写 Claude Code 等框架被腾讯官方逐字拉黑的固定 prompt 语句 sanitizeMessages(reqObj) + // 会话结构归一化:保证首条消息为 system,修复部分非 harness 客户端 + //(以 assistant / tool 续写或回传工具结果)触发的上游 11128 错误 + ensureLeadingSystemMessage(reqObj) + upstreamBytes, err := json.Marshal(reqObj) if err != nil { writeOpenAIError(w, http.StatusInternalServerError, "encode_error", "序列化请求失败") @@ -1731,150 +1839,64 @@ func handleChatCompletions(w http.ResponseWriter, r *http.Request) { log.Printf("[#%d] POST /v1/chat/completions -> Upstream [Model: %s, Stream: %v]", reqID, modelName, isStream) } - // 多账号轮询 + 429 自动冷却代偿: - // 按轮询顺序尝试账号;命中 429 频率限制时立即将当前账号屏蔽到重置时间, - // 并自动改用下一个可用账号重试(代偿),直至成功或所有账号均不可用。 - accountMu.Lock() - poolSize := len(accounts) - accountMu.Unlock() - if poolSize == 0 { - writeOpenAIError(w, http.StatusUnauthorized, "no_auth", "未找到有效登录凭据,请先执行 login 命令扫码登录") + resp, acc, prof, ok := upstreamChat(w, r, reqID, upstreamBytes, startTime) + if !ok { + return + } + if isStream { + streamChatResponse(w, resp, reqID, acc, prof, startTime) + } else { + writeChatAggregate(w, resp, modelName, reqID, acc, prof, startTime) + } +} + +// streamChatResponse 将上游 SSE 逐行透传为 OpenAI Chat Completions 流式响应。 +func streamChatResponse(w http.ResponseWriter, resp *http.Response, reqID uint64, acc *Account, prof *upstreamProfile, startTime time.Time) { + defer resp.Body.Close() + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + + flusher, ok := w.(http.Flusher) + if !ok { + writeOpenAIError(w, http.StatusInternalServerError, "streaming_unsupported", "服务器不支持流式响应 Flush") return } - var lastRateErr string - var lastAuthErr string - for attempt := 0; attempt < poolSize; attempt++ { - acc, err := nextAccount() - if err != nil { - // 所有账号均不可用(冷却或失效) - msg := fmt.Sprintf("无可用账号: %v", err) - if lastAuthErr != "" { - msg += " | 最近一次授权失效: " + truncate(lastAuthErr, 200) - } - if lastRateErr != "" { - msg += " | 最近一次频率限制: " + truncate(lastRateErr, 200) - } - log.Printf("[#%d] %s", reqID, msg) - writeOpenAIError(w, http.StatusServiceUnavailable, "no_available_account", msg) - return - } - _ = ensureValidTokenFor(acc) - // 令牌刷新时若发现授权失效会禁用账号;若被禁用则跳过换下一个 - accountMu.Lock() - disabled := acc.Disabled - accountMu.Unlock() - if disabled { - lastAuthErr = acc.DisabledReason + scanner := bufio.NewScanner(resp.Body) + scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) + for scanner.Scan() { + cleanData := stripDataPrefix(scanner.Text()) + if cleanData == "" { continue } - - // 按账号所属站点(国内站/国际站)路由上游与指纹 Header - prof := acc.Profile() - - upstreamReq, err := http.NewRequestWithContext(r.Context(), http.MethodPost, prof.chatURL(), bytes.NewReader(upstreamBytes)) - if err != nil { - writeOpenAIError(w, http.StatusInternalServerError, "req_create_error", err.Error()) - return + if cleanData == "[DONE]" { + _, _ = fmt.Fprintf(w, "data: [DONE]\n\n") + flusher.Flush() + break } - // 注入 CodeBuddy 凭据与指纹 Header - backendHeaders(upstreamReq, acc.Auth, prof) - - // 限制同一账号向腾讯上游的请求严格单并发串行排队,防止 DSH 同时发起标题生成+主对话触发腾讯 11128 风控 - acc.lock.Lock() - resp, err := cfg.HttpClient.Do(upstreamReq) - acc.lock.Unlock() - if err != nil { - log.Printf("[#%d] 账号 %s [%s] 上游请求失败: %v", reqID, acc.Path, prof.Label, err) - writeOpenAIError(w, http.StatusBadGateway, "upstream_network_error", fmt.Sprintf("网络转发失败: %v", err)) - return + if cleanedChunk := cleanChunkJSON(cleanData); cleanedChunk != "" { + _, _ = fmt.Fprintf(w, "data: %s\n\n", cleanedChunk) + flusher.Flush() } + } + log.Printf("[#%d] 流式输出完成 (账号 %s [%s], 耗时 %v)", reqID, acc.Path, prof.Label, time.Since(startTime)) +} - // 上游非 200 响应处理 - if resp.StatusCode >= 400 { - errBody, _ := io.ReadAll(resp.Body) - resp.Body.Close() - errStr := string(errBody) - log.Printf("[#%d] 账号 %s [%s] 上游返回 HTTP %d: %s (耗时 %v)", reqID, acc.Path, prof.Label, resp.StatusCode, errStr, time.Since(startTime)) - - if isRateLimited(resp.StatusCode, errStr) { - // 429 频率限制:解析重置时间并屏蔽该账号,交由其他账号代偿 - until, ok := parseResetTime(errStr) - if !ok { - until = time.Now().Add(60 * time.Second) // 无法解析时默认冷却 60 秒 - } - markCooldown(acc, until, errStr) - lastRateErr = errStr - continue // 尝试下一个账号 - } - - if isAuthFailure(resp.StatusCode, errStr) { - // 授权失效(401/403 / token 无效 / 登录过期):禁用该账号并删除凭据文件, - // 自动改用下一个可用账号,控制台提示用户重新登录 - disableAccount(acc, fmt.Sprintf("上游鉴权失败 (HTTP %d): %s", resp.StatusCode, truncate(errStr, 200))) - lastAuthErr = errStr - continue // 尝试下一个账号 - } - - writeOpenAIError(w, resp.StatusCode, "upstream_error", fmt.Sprintf("upstream %d: %s", resp.StatusCode, errStr)) - return - } - - if isStream { - // 客户端需要流式响应 - w.Header().Set("Content-Type", "text/event-stream") - w.Header().Set("Cache-Control", "no-cache") - w.Header().Set("Connection", "keep-alive") - w.Header().Set("X-Accel-Buffering", "no") - - flusher, ok := w.(http.Flusher) - if !ok { - resp.Body.Close() - writeOpenAIError(w, http.StatusInternalServerError, "streaming_unsupported", "服务器不支持流式响应 Flush") - return - } - - scanner := bufio.NewScanner(resp.Body) - scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) - - for scanner.Scan() { - line := scanner.Text() - cleanData := stripDataPrefix(line) - if cleanData == "" { - continue - } - if cleanData == "[DONE]" { - _, _ = fmt.Fprintf(w, "data: [DONE]\n\n") - flusher.Flush() - break - } - cleanedChunk := cleanChunkJSON(cleanData) - if cleanedChunk != "" { - _, _ = fmt.Fprintf(w, "data: %s\n\n", cleanedChunk) - flusher.Flush() - } - } - resp.Body.Close() - log.Printf("[#%d] 流式输出完成 (账号 %s [%s], 耗时 %v)", reqID, acc.Path, prof.Label, time.Since(startTime)) - } else { - // 客户端需要完整 JSON 响应,聚合 SSE 数据流 - completionJSON, err := aggregateCompletion(resp.Body, modelName) - resp.Body.Close() - if err != nil { - log.Printf("[#%d] 聚合响应失败: %v", reqID, err) - writeOpenAIError(w, http.StatusInternalServerError, "aggregate_error", "聚合上游流式响应失败: "+err.Error()) - return - } - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusOK) - _, _ = w.Write(completionJSON) - log.Printf("[#%d] 非流式响应完成 (账号 %s [%s], 耗时 %v)", reqID, acc.Path, prof.Label, time.Since(startTime)) - } +// writeChatAggregate 聚合上游 SSE 为完整 Chat Completions JSON 响应。 +func writeChatAggregate(w http.ResponseWriter, resp *http.Response, modelName string, reqID uint64, acc *Account, prof *upstreamProfile, startTime time.Time) { + defer resp.Body.Close() + completionJSON, err := aggregateCompletion(resp.Body, modelName) + if err != nil { + log.Printf("[#%d] 聚合响应失败: %v", reqID, err) + writeOpenAIError(w, http.StatusInternalServerError, "aggregate_error", "聚合上游流式响应失败: "+err.Error()) return } - - // 理论上不可达(poolSize 次尝试后未成功即已在循环内返回) - writeOpenAIError(w, http.StatusTooManyRequests, "all_accounts_cooldown", "所有账号均处于冷却状态") + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(completionJSON) + log.Printf("[#%d] 非流式响应完成 (账号 %s [%s], 耗时 %v)", reqID, acc.Path, prof.Label, time.Since(startTime)) } // ----------------------------------------------------------------------------- @@ -1918,7 +1940,7 @@ func handleIndex(w http.ResponseWriter, r *http.Request) { return } w.Header().Set("Content-Type", "text/plain; charset=utf-8") - _, _ = fmt.Fprintf(w, "WorkBuddy Local Gateway v%s is running.\n\nEndpoints:\n- POST /v1/chat/completions\n- GET /v1/models\n- GET /health\n", version) + _, _ = fmt.Fprintf(w, "WorkBuddy Local Gateway v%s is running.\n\nEndpoints:\n- POST /v1/chat/completions\n- POST /v1/responses\n- GET /v1/models\n- GET /health\n", version) } // ----------------------------------------------------------------------------- @@ -1948,6 +1970,12 @@ func sanitizeMessages(obj map[string]any) { if !ok { continue } + // OpenAI 新版 developer 角色(GPT-5 系客户端/Codex)不被腾讯上游接受, + // 会返回 11128 "Illegal API invocation from an unapproved channel", + // 统一归一化为 system(语义等价)。 + if roleOfMessage(msg) == "developer" { + msg["role"] = "system" + } switch c := msg["content"].(type) { case string: msg["content"] = sanitizeBlockedTemplates(c) @@ -1965,6 +1993,66 @@ func sanitizeMessages(obj map[string]any) { } } +// ensureLeadingSystemMessage 保证 messages 的首条消息符合腾讯上游的会话结构校验。 +// +// 上游要求首条消息为 system prompt,否则返回 HTTP 400 +// {"code":11128,"msg":"first message is not system prompt"}。实测国内站对 user 开头较宽容, +// 但国际站(workbuddy.ai)严格校验;账号池混挂时表现为约 50% 请求随机失败。部分非 harness +// 客户端在续写或仅回传工具结果时还会以 assistant / tool 作为首条消息。此处统一归一化为: +// 1. 首条已是 system:原样透传; +// 2. 首条是 developer(OpenAI 新版 system 别名):重命名为 system; +// 3. 后续存在 system/developer:提升到首位(developer 归一化为 system),其余保持原序; +// 4. 其余情况(user / assistant / tool 开头且无 system):在最前注入一条保底 system。 +func ensureLeadingSystemMessage(obj map[string]any) { + messages, ok := obj["messages"].([]any) + if !ok || len(messages) == 0 { + obj["messages"] = []any{map[string]any{"role": "system", "content": defaultSystemPrompt}} + return + } + + switch roleOfMessage(messages[0]) { + case "system": + return + case "developer": + if msg, ok := messages[0].(map[string]any); ok { + msg["role"] = "system" + } + return + } + + // 后续存在 system/developer:提升到首位,其余保持原序 + for i := 1; i < len(messages); i++ { + switch roleOfMessage(messages[i]) { + case "system", "developer": + if msg, ok := messages[i].(map[string]any); ok { + msg["role"] = "system" + } + reordered := make([]any, 0, len(messages)) + reordered = append(reordered, messages[i]) + reordered = append(reordered, messages[:i]...) + reordered = append(reordered, messages[i+1:]...) + obj["messages"] = reordered + return + } + } + + // 无任何 system:在最前注入保底 system(兼容国内站/国际站) + injected := make([]any, 0, len(messages)+1) + injected = append(injected, map[string]any{"role": "system", "content": defaultSystemPrompt}) + injected = append(injected, messages...) + obj["messages"] = injected +} + +// roleOfMessage 读取消息的 role 字段并归一化为小写去空格;非法结构返回空串。 +func roleOfMessage(m any) string { + msg, ok := m.(map[string]any) + if !ok { + return "" + } + role, _ := msg["role"].(string) + return strings.ToLower(strings.TrimSpace(role)) +} + func sanitizeBlockedTemplates(s string) string { s = strings.ReplaceAll(s, "You are Claude Code, Anthropic's official CLI for Claude.", @@ -2053,11 +2141,55 @@ func doJSON(client *http.Client, method, fullURL string, headers func(*http.Requ // 流式聚合与格式清理 // ----------------------------------------------------------------------------- +// mergedToolCall 累积合并上游按 index 分片下发的工具调用增量。 +type mergedToolCall struct { + ID string + Type string + Name string + Args strings.Builder +} + +// applyToolCallDelta 将上游 tool_calls 增量按 index 归并进 map;order 记录首次出现的 +// index 顺序,保证最终输出顺序稳定。Chat Completions 聚合与 Responses 流式共用。 +func applyToolCallDelta(toolCalls map[int]*mergedToolCall, order *[]int, tcs []any) { + for _, tcAny := range tcs { + tc, ok := tcAny.(map[string]any) + if !ok { + continue + } + idx := 0 + if v, ok := tc["index"].(float64); ok { + idx = int(v) + } + st, exists := toolCalls[idx] + if !exists { + st = &mergedToolCall{Type: "function"} + toolCalls[idx] = st + *order = append(*order, idx) + } + if id, ok := tc["id"].(string); ok && id != "" { + st.ID = id + } + if t, ok := tc["type"].(string); ok && t != "" { + st.Type = t + } + if fn, ok := tc["function"].(map[string]any); ok { + if n, ok := fn["name"].(string); ok && n != "" { + st.Name = n + } + if a, ok := fn["arguments"].(string); ok && a != "" { + st.Args.WriteString(a) + } + } + } +} + func aggregateCompletion(r io.Reader, model string) ([]byte, error) { var content, reasoning, role, respModel, respID, finish string var created int64 var usage map[string]any - var toolCalls []map[string]any + toolCalls := map[int]*mergedToolCall{} + var toolOrder []int scanner := bufio.NewScanner(r) scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) @@ -2096,11 +2228,7 @@ func aggregateCompletion(r io.Reader, model string) ([]byte, error) { reasoning += v } if tcs, ok := delta["tool_calls"].([]any); ok { - for _, tc := range tcs { - if call, ok := tc.(map[string]any); ok { - toolCalls = append(toolCalls, call) - } - } + applyToolCallDelta(toolCalls, &toolOrder, tcs) } } if v, ok := choice["finish_reason"].(string); ok && v != "" { @@ -2113,8 +2241,25 @@ func aggregateCompletion(r io.Reader, model string) ([]byte, error) { if reasoning != "" { message["reasoning_content"] = reasoning } - if len(toolCalls) > 0 { - message["tool_calls"] = toolCalls + if len(toolOrder) > 0 { + calls := make([]map[string]any, 0, len(toolOrder)) + for _, idx := range toolOrder { + st := toolCalls[idx] + id := st.ID + if id == "" { + id = fmt.Sprintf("call_%d_%d", time.Now().UnixNano(), idx) + } + calls = append(calls, map[string]any{ + "id": id, + "type": ifEmpty(st.Type, "function"), + "index": idx, + "function": map[string]any{ + "name": st.Name, + "arguments": st.Args.String(), + }, + }) + } + message["tool_calls"] = calls } if created == 0 { created = time.Now().Unix() diff --git a/main_test.go b/main_test.go index d30ac49..c24619a 100644 --- a/main_test.go +++ b/main_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "os" "path/filepath" + "strings" "testing" "time" ) @@ -601,3 +602,311 @@ func TestWriteStatusSnapshot(t *testing.T) { } t.Logf("snapshot states: %v", states) } + +// 构造 messages 便于表驱动测试 +func msg(role, content string) map[string]any { + return map[string]any{"role": role, "content": content} +} + +// 提取 messages 各条 role,便于断言 +func rolesOf(obj map[string]any) []string { + messages, _ := obj["messages"].([]any) + roles := make([]string, 0, len(messages)) + for _, m := range messages { + roles = append(roles, roleOfMessage(m)) + } + return roles +} + +// 验证会话结构归一化:修复上游 11128「first message is not system prompt」 +func TestEnsureLeadingSystemMessage(t *testing.T) { + cases := []struct { + name string + messages []any + wantRoles []string + wantInject bool // 首条是否应为注入的保底 system + }{ + { + name: "首条已是 system 保持原样", + messages: []any{msg("system", "You are helpful"), msg("user", "hi")}, + wantRoles: []string{"system", "user"}, + }, + { + name: "首条 user 且无 system(国内站宽容/国际站必须,统一注入 system)", + messages: []any{msg("user", "hi")}, + wantRoles: []string{"system", "user"}, + wantInject: true, + }, + { + name: "首条 assistant(续写)注入 system", + messages: []any{msg("assistant", "Sure")}, + wantRoles: []string{"system", "assistant"}, + wantInject: true, + }, + { + name: "首条 tool(仅回传工具结果)注入 system", + messages: []any{msg("tool", "result")}, + wantRoles: []string{"system", "tool"}, + wantInject: true, + }, + { + name: "后续 system 提升到首位", + messages: []any{msg("user", "hi"), msg("system", "You are helpful"), msg("user", "bye")}, + wantRoles: []string{"system", "user", "user"}, + }, + { + name: "首条 developer 归一化为 system", + messages: []any{msg("developer", "You are helpful"), msg("user", "hi")}, + wantRoles: []string{"system", "user"}, + }, + { + name: "空 messages 注入 system", + messages: []any{}, + wantRoles: []string{"system"}, + wantInject: true, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + obj := map[string]any{"messages": c.messages} + ensureLeadingSystemMessage(obj) + got := rolesOf(obj) + if len(got) != len(c.wantRoles) { + t.Fatalf("roles = %v, want %v", got, c.wantRoles) + } + for i := range got { + if got[i] != c.wantRoles[i] { + t.Fatalf("roles = %v, want %v", got, c.wantRoles) + } + } + messages, _ := obj["messages"].([]any) + first, _ := messages[0].(map[string]any) + if c.wantInject { + if content, _ := first["content"].(string); content != defaultSystemPrompt { + t.Fatalf("injected system content = %q, want %q", content, defaultSystemPrompt) + } + } + t.Logf("roles -> %v", got) + }) + } +} + +// 验证缺失 / 非法 messages 字段时也能安全注入(不得 panic) +func TestEnsureLeadingSystemMessageMissingField(t *testing.T) { + obj := map[string]any{} + ensureLeadingSystemMessage(obj) + messages, ok := obj["messages"].([]any) + if !ok || len(messages) != 1 { + t.Fatalf("expected 1 injected message, got %#v", obj["messages"]) + } + if roleOfMessage(messages[0]) != "system" { + t.Fatalf("expected system first, got %v", roleOfMessage(messages[0])) + } + t.Log("missing messages field handled") +} + +// 验证 developer 角色(GPT-5/Codex)被归一化为 system,避免上游 11128 +// "Illegal API invocation from an unapproved channel" +func TestSanitizeMessagesNormalizesDeveloperRole(t *testing.T) { + obj := map[string]any{"messages": []any{ + msg("system", "You are helpful."), + msg("developer", "Be terse."), + msg("user", "hi"), + }} + sanitizeMessages(obj) + roles := rolesOf(obj) + for _, r := range roles { + if r == "developer" { + t.Fatalf("developer role should be normalized, got %v", roles) + } + } + if roles[1] != "system" { + t.Fatalf("roles = %v, want second = system", roles) + } + t.Logf("roles normalized: %v", roles) +} + +// 验证上游 tool_calls 增量按 index 正确归并为一个完整工具调用 +func TestApplyToolCallDeltaMerge(t *testing.T) { + merged := map[int]*mergedToolCall{} + var order []int + applyToolCallDelta(merged, &order, []any{ + map[string]any{"index": float64(0), "id": "call_1", "type": "function", + "function": map[string]any{"name": "get_weather", "arguments": ""}}, + }) + applyToolCallDelta(merged, &order, []any{ + map[string]any{"index": float64(0), "function": map[string]any{"arguments": `{"city":`}}, + }) + applyToolCallDelta(merged, &order, []any{ + map[string]any{"index": float64(0), "function": map[string]any{"arguments": `"Beijing"}`}}, + }) + if len(order) != 1 || order[0] != 0 { + t.Fatalf("order = %v, want [0]", order) + } + st := merged[0] + if st.ID != "call_1" || st.Name != "get_weather" { + t.Fatalf("id/name = %q/%q", st.ID, st.Name) + } + if got := st.Args.String(); got != `{"city":"Beijing"}` { + t.Fatalf("args = %q", got) + } + t.Logf("merged tool call: %s(%s) id=%s", st.Name, st.Args.String(), st.ID) +} + +// 验证 aggregateCompletion 正确合并流式 tool_calls(修复旧的按片追加问题) +func TestAggregateCompletionToolCalls(t *testing.T) { + sse := strings.Join([]string{ + `data: {"id":"cmpl-1","model":"hy3-preview","created":1700000000,"choices":[{"index":0,"delta":{"role":"assistant","tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"get_weather","arguments":""}}]}}]}`, + `data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"city\":"}}]}}]}`, + `data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"Beijing\"}"}}]}}]}`, + `data: [DONE]`, + }, "\n") + + out, err := aggregateCompletion(strings.NewReader(sse), "hy3-preview") + if err != nil { + t.Fatal(err) + } + var chat map[string]any + if err := json.Unmarshal(out, &chat); err != nil { + t.Fatal(err) + } + choices := chat["choices"].([]any) + msg := choices[0].(map[string]any)["message"].(map[string]any) + calls, ok := msg["tool_calls"].([]any) + if !ok || len(calls) != 1 { + t.Fatalf("expected 1 merged tool_call, got %#v", msg["tool_calls"]) + } + call := calls[0].(map[string]any) + fn := call["function"].(map[string]any) + if fn["name"] != "get_weather" || fn["arguments"] != `{"city":"Beijing"}` || call["id"] != "call_1" { + t.Fatalf("bad merged call: %#v", call) + } + t.Logf("aggregateCompletion merged: %v", call) +} + +// 验证 Responses -> Chat Completions 请求转换(instructions/input/tools/tool_choice/参数) +func TestResponsesToChatRequest(t *testing.T) { + respReq := map[string]any{ + "model": "hy3-preview", + "instructions": "You are helpful.", + "max_output_tokens": float64(128), + "temperature": float64(0.3), + "input": []any{ + map[string]any{"role": "user", "content": []any{ + map[string]any{"type": "input_text", "text": "weather?"}, + }}, + map[string]any{"type": "function_call", "call_id": "call_1", "name": "get_weather", "arguments": `{"city":"BJ"}`}, + map[string]any{"type": "function_call_output", "call_id": "call_1", "output": "sunny"}, + }, + "tools": []any{map[string]any{ + "type": "function", "name": "get_weather", "description": "Get weather", + "parameters": map[string]any{"type": "object"}, + }}, + "tool_choice": "auto", + "reasoning": map[string]any{"effort": "high"}, + } + + chat, err := responsesToChatRequest(respReq, "hy3-preview") + if err != nil { + t.Fatal(err) + } + if chat["model"] != "hy3-preview" || chat["max_tokens"] != float64(128) || chat["temperature"] != float64(0.3) { + t.Fatalf("bad top-level fields: %#v", chat) + } + if chat["reasoning_effort"] != "high" { + t.Fatalf("reasoning_effort = %v", chat["reasoning_effort"]) + } + messages := chat["messages"].([]any) + if len(messages) != 4 { + t.Fatalf("expected 4 messages (system+user+assistant+tool), got %d: %#v", len(messages), messages) + } + if rolesOf(map[string]any{"messages": messages})[0] != "system" { + t.Fatalf("first message should be system") + } + assistant := messages[2].(map[string]any) + if assistant["role"] != "assistant" { + t.Fatalf("3rd message role = %v", assistant["role"]) + } + toolMsg := messages[3].(map[string]any) + if toolMsg["role"] != "tool" || toolMsg["tool_call_id"] != "call_1" || toolMsg["content"] != "sunny" { + t.Fatalf("tool message = %#v", toolMsg) + } + tools := chat["tools"].([]any) + fn := tools[0].(map[string]any)["function"].(map[string]any) + if fn["name"] != "get_weather" { + t.Fatalf("tools not flattened: %#v", tools) + } + if chat["tool_choice"] != "auto" { + t.Fatalf("tool_choice = %v", chat["tool_choice"]) + } + t.Log("responses request converted OK") +} + +// 验证 responsesToChatRequest 对纯字符串 input 的处理与空 input 报错 +func TestResponsesToChatRequestStringInput(t *testing.T) { + chat, err := responsesToChatRequest(map[string]any{"model": "x", "input": "hello"}, "x") + if err != nil { + t.Fatal(err) + } + messages := chat["messages"].([]any) + if len(messages) != 1 || messages[0].(map[string]any)["content"] != "hello" { + t.Fatalf("bad messages: %#v", messages) + } + if _, err := responsesToChatRequest(map[string]any{"model": "x"}, "x"); err == nil { + t.Fatal("empty input should error") + } +} + +// 验证 chat.completion -> Responses 非流式响应对象转换 +func TestChatCompletionToResponses(t *testing.T) { + chat := map[string]any{ + "id": "cmpl-123", "created": float64(1700000000), "model": "hy3-preview", + "choices": []any{map[string]any{"index": 0, "message": map[string]any{ + "role": "assistant", "content": "hello", "reasoning_content": "thinking", + "tool_calls": []any{map[string]any{ + "id": "call_1", "type": "function", + "function": map[string]any{"name": "get_weather", "arguments": `{"city":"BJ"}`}, + }}, + }}}, + "usage": map[string]any{ + "prompt_tokens": float64(10), "completion_tokens": float64(5), "total_tokens": float64(15), + "prompt_tokens_details": map[string]any{"cached_tokens": float64(2)}, + "completion_tokens_details": map[string]any{"reasoning_tokens": float64(3)}, + }, + } + raw, _ := json.Marshal(chat) + out, err := chatCompletionToResponses(raw, "hy3-preview") + if err != nil { + t.Fatal(err) + } + var result map[string]any + if err := json.Unmarshal(out, &result); err != nil { + t.Fatal(err) + } + if result["object"] != "response" || result["status"] != "completed" { + t.Fatalf("bad envelope: %#v", result) + } + output := result["output"].([]any) + if len(output) != 3 { + t.Fatalf("expected 3 output items (reasoning+message+function_call), got %d", len(output)) + } + if output[0].(map[string]any)["type"] != "reasoning" { + t.Fatalf("output[0] = %#v", output[0]) + } + if output[1].(map[string]any)["type"] != "message" { + t.Fatalf("output[1] = %#v", output[1]) + } + fc := output[2].(map[string]any) + if fc["type"] != "function_call" || fc["call_id"] != "call_1" || fc["name"] != "get_weather" { + t.Fatalf("function_call item bad: %#v", fc) + } + usage := result["usage"].(map[string]any) + if usage["input_tokens"] != float64(10) && usage["input_tokens"] != int64(10) { + t.Fatalf("usage input = %#v", usage["input_tokens"]) + } + if usage["total_tokens"] != float64(15) && usage["total_tokens"] != int64(15) { + t.Fatalf("usage total = %#v", usage["total_tokens"]) + } + t.Logf("responses object output items: %d", len(output)) +} diff --git a/responses.go b/responses.go new file mode 100644 index 0000000..6698d51 --- /dev/null +++ b/responses.go @@ -0,0 +1,699 @@ +package main + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "strings" + "sync/atomic" + "time" + + "github.com/google/uuid" +) + +// ----------------------------------------------------------------------------- +// OpenAI Responses API (/v1/responses) -> 上游 Chat Completions 转译 +// +// 网关只与上游 /v2/chat/completions 通信,此处负责双向协议转换: +// - 请求:input / instructions / tools(扁平) / tool_choice / max_output_tokens / +// reasoning.effort -> chat messages / tools(嵌套 function) / max_tokens / reasoning_effort +// - 非流式响应:chat.completion -> response{object:"response", output:[...]} +// - 流式响应:上游 SSE 增量 -> Responses 语义事件(response.output_text.delta、 +// response.function_call_arguments.delta、response.completed ...) +// ----------------------------------------------------------------------------- + +// handleResponses 处理 POST /v1/responses。 +func handleResponses(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeOpenAIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "仅支持 POST 请求") + return + } + + reqID := atomic.AddUint64(&reqCounter, 1) + startTime := time.Now() + + bodyBytes, err := io.ReadAll(r.Body) + if err != nil { + writeOpenAIError(w, http.StatusBadRequest, "read_error", "读取请求体失败") + return + } + defer r.Body.Close() + + var respReq map[string]any + if err := json.Unmarshal(bodyBytes, &respReq); err != nil { + writeOpenAIError(w, http.StatusBadRequest, "invalid_json", "无效的 JSON 请求体") + return + } + + modelName, _ := respReq["model"].(string) + if modelName == "" { + modelName = "hy4-preview" + } + isStream, _ := respReq["stream"].(bool) + + chatReq, err := responsesToChatRequest(respReq, modelName) + if err != nil { + writeOpenAIError(w, http.StatusBadRequest, "invalid_request", err.Error()) + return + } + chatReq["stream"] = true // 上游强制流式,非流式由网关本地聚合 + + applyThinkingRules(chatReq, modelName) + sanitizeMessages(chatReq) + ensureLeadingSystemMessage(chatReq) + + upstreamBytes, err := json.Marshal(chatReq) + if err != nil { + writeOpenAIError(w, http.StatusInternalServerError, "encode_error", "序列化请求失败") + return + } + + log.Printf("[#%d] POST /v1/responses -> Upstream [Model: %s, Stream: %v]", reqID, modelName, isStream) + + resp, acc, prof, ok := upstreamChat(w, r, reqID, upstreamBytes, startTime) + if !ok { + return + } + if isStream { + streamResponsesResponse(w, resp, modelName, reqID, acc, prof, startTime) + } else { + writeResponsesAggregate(w, resp, modelName, reqID, acc, prof, startTime) + } +} + +// responsesToChatRequest 将 Responses 请求体转换为上游 Chat Completions 请求体。 +func responsesToChatRequest(respReq map[string]any, modelName string) (map[string]any, error) { + chat := map[string]any{"model": modelName} + + messages := []any{} + if instructions, ok := respReq["instructions"].(string); ok && strings.TrimSpace(instructions) != "" { + messages = append(messages, map[string]any{"role": "system", "content": instructions}) + } + + switch input := respReq["input"].(type) { + case string: + if strings.TrimSpace(input) != "" { + messages = append(messages, map[string]any{"role": "user", "content": input}) + } + case []any: + for _, itemAny := range input { + switch item := itemAny.(type) { + case string: + messages = append(messages, map[string]any{"role": "user", "content": item}) + case map[string]any: + messages = append(messages, convertResponsesInputItem(item)...) + } + } + } + + if len(messages) == 0 { + return nil, fmt.Errorf("input 为空:Responses 请求必须提供 input 或 instructions") + } + chat["messages"] = messages + + if v, ok := respReq["temperature"]; ok && v != nil { + chat["temperature"] = v + } + if v, ok := respReq["top_p"]; ok && v != nil { + chat["top_p"] = v + } + if v, ok := respReq["max_output_tokens"]; ok && v != nil { + chat["max_tokens"] = v + } + if tools := convertResponsesTools(respReq["tools"]); len(tools) > 0 { + chat["tools"] = tools + } + if tc := convertResponsesToolChoice(respReq["tool_choice"]); tc != nil { + chat["tool_choice"] = tc + } + if reasoning, ok := respReq["reasoning"].(map[string]any); ok { + if effort, ok := reasoning["effort"].(string); ok && effort != "" && effort != "none" { + chat["reasoning_effort"] = effort + } + } + return chat, nil +} + +// convertResponsesInputItem 将单个 Responses input item 转换为 0..1 条 chat 消息。 +func convertResponsesInputItem(item map[string]any) []any { + switch typ, _ := item["type"].(string); typ { + case "function_call": + callID, _ := item["call_id"].(string) + if callID == "" { + callID, _ = item["id"].(string) + } + name, _ := item["name"].(string) + args, _ := item["arguments"].(string) + return []any{map[string]any{ + "role": "assistant", + "content": nil, + "tool_calls": []any{map[string]any{ + "id": ifEmpty(callID, "call_"+compactUUID()), + "type": "function", + "function": map[string]any{"name": name, "arguments": args}, + }}, + }} + case "function_call_output": + callID, _ := item["call_id"].(string) + return []any{map[string]any{ + "role": "tool", + "tool_call_id": callID, + "content": stringifyToolOutput(item["output"]), + }} + case "reasoning": + // 上游无法接收 reasoning item,忽略(历史上下文不影响后续对话) + return nil + } + + role, _ := item["role"].(string) + if role == "" { + role = "user" + } + return []any{map[string]any{"role": role, "content": convertResponsesContent(item["content"])}} +} + +// convertResponsesContent 将 Responses content(string 或 parts 数组)转换为 chat content。 +func convertResponsesContent(content any) any { + switch c := content.(type) { + case nil: + return "" + case string: + return c + case []any: + parts := make([]any, 0, len(c)) + for _, pAny := range c { + p, ok := pAny.(map[string]any) + if !ok { + if s, ok := pAny.(string); ok { + parts = append(parts, map[string]any{"type": "text", "text": s}) + } + continue + } + switch typ, _ := p["type"].(string); typ { + case "input_text", "output_text", "text": + if t, ok := p["text"].(string); ok { + parts = append(parts, map[string]any{"type": "text", "text": t}) + } + case "refusal": + if t, ok := p["refusal"].(string); ok { + parts = append(parts, map[string]any{"type": "text", "text": t}) + } + case "input_image": + url, _ := p["image_url"].(string) + if url != "" { + parts = append(parts, map[string]any{"type": "image_url", "image_url": map[string]any{"url": url}}) + } + } + } + if len(parts) == 0 { + return "" + } + return parts + default: + return fmt.Sprintf("%v", c) + } +} + +// stringifyToolOutput 将 function_call_output 的 output 统一转为字符串。 +func stringifyToolOutput(output any) string { + switch o := output.(type) { + case nil: + return "" + case string: + return o + default: + if b, err := json.Marshal(o); err == nil { + return string(b) + } + return fmt.Sprintf("%v", o) + } +} + +// convertResponsesTools 将 Responses 扁平 function 工具转换为 chat 嵌套 function 工具。 +func convertResponsesTools(toolsAny any) []any { + arr, ok := toolsAny.([]any) + if !ok { + return nil + } + out := make([]any, 0, len(arr)) + for _, tAny := range arr { + t, ok := tAny.(map[string]any) + if !ok { + continue + } + typ, _ := t["type"].(string) + if typ != "" && typ != "function" { + continue // 仅支持 function 工具 + } + name, _ := t["name"].(string) + desc, _ := t["description"].(string) + params := t["parameters"] + if name == "" { + // 兼容旧式嵌套 {type:"function", function:{...}} + if fn, ok := t["function"].(map[string]any); ok { + name, _ = fn["name"].(string) + desc, _ = fn["description"].(string) + params = fn["parameters"] + } + } + if name == "" { + continue + } + fn := map[string]any{"name": name} + if desc != "" { + fn["description"] = desc + } + if params != nil { + fn["parameters"] = params + } + out = append(out, map[string]any{"type": "function", "function": fn}) + } + return out +} + +// convertResponsesToolChoice 将 Responses tool_choice 转换为 chat tool_choice。 +func convertResponsesToolChoice(tc any) any { + switch v := tc.(type) { + case string: + if v == "auto" || v == "none" || v == "required" { + return v + } + case map[string]any: + if typ, _ := v["type"].(string); typ == "function" { + name, _ := v["name"].(string) + if name == "" { + if fn, ok := v["function"].(map[string]any); ok { + name, _ = fn["name"].(string) + } + } + if name != "" { + return map[string]any{"type": "function", "function": map[string]any{"name": name}} + } + } + } + return nil +} + +// writeResponsesAggregate 聚合上游 SSE 后转换为 Responses 非流式响应。 +func writeResponsesAggregate(w http.ResponseWriter, resp *http.Response, modelName string, reqID uint64, acc *Account, prof *upstreamProfile, startTime time.Time) { + defer resp.Body.Close() + completionJSON, err := aggregateCompletion(resp.Body, modelName) + if err != nil { + log.Printf("[#%d] 聚合响应失败: %v", reqID, err) + writeOpenAIError(w, http.StatusInternalServerError, "aggregate_error", "聚合上游流式响应失败: "+err.Error()) + return + } + out, err := chatCompletionToResponses(completionJSON, modelName) + if err != nil { + log.Printf("[#%d] Responses 转换失败: %v", reqID, err) + writeOpenAIError(w, http.StatusInternalServerError, "translate_error", "转换 Responses 响应失败: "+err.Error()) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(out) + log.Printf("[#%d] Responses 非流式响应完成 (账号 %s [%s], 耗时 %v)", reqID, acc.Path, prof.Label, time.Since(startTime)) +} + +// chatCompletionToResponses 将 chat.completion JSON 转换为 Responses 响应对象。 +func chatCompletionToResponses(chatJSON []byte, modelName string) ([]byte, error) { + var chat map[string]any + if err := json.Unmarshal(chatJSON, &chat); err != nil { + return nil, err + } + + var content, reasoning string + var toolCalls []any + if choices, ok := chat["choices"].([]any); ok && len(choices) > 0 { + if choice, ok := choices[0].(map[string]any); ok { + if msg, ok := choice["message"].(map[string]any); ok { + content, _ = msg["content"].(string) + reasoning, _ = msg["reasoning_content"].(string) + toolCalls, _ = msg["tool_calls"].([]any) + } + } + } + + output := []any{} + if reasoning != "" { + output = append(output, map[string]any{ + "id": "rs_" + compactUUID(), + "type": "reasoning", + "status": "completed", + "summary": []any{map[string]any{"type": "summary_text", "text": reasoning}}, + }) + } + if content != "" || len(toolCalls) == 0 { + output = append(output, map[string]any{ + "id": "msg_" + compactUUID(), + "type": "message", + "status": "completed", + "role": "assistant", + "content": []any{ + map[string]any{"type": "output_text", "text": content, "annotations": []any{}}, + }, + }) + } + for _, tcAny := range toolCalls { + tc, ok := tcAny.(map[string]any) + if !ok { + continue + } + callID, _ := tc["id"].(string) + name, args := "", "" + if fn, ok := tc["function"].(map[string]any); ok { + name, _ = fn["name"].(string) + args, _ = fn["arguments"].(string) + } + output = append(output, map[string]any{ + "id": "fc_" + compactUUID(), + "type": "function_call", + "status": "completed", + "call_id": ifEmpty(callID, "call_"+compactUUID()), + "name": name, + "arguments": args, + }) + } + + now := time.Now().Unix() + created := now + if v, ok := chat["created"].(float64); ok && v > 0 && int64(v) <= now { + created = int64(v) + } + respID, _ := chat["id"].(string) + if respID == "" { + respID = "resp_" + compactUUID() + } else { + respID = "resp_" + strings.TrimPrefix(respID, "chatcmpl-") + } + + result := buildResponsesEnvelope(respID, modelName, created) + result["status"] = "completed" + result["output"] = output + result["completed_at"] = now + if usage, ok := chat["usage"].(map[string]any); ok { + result["usage"] = toResponsesUsage(usage) + } + return json.Marshal(result) +} + +// buildResponsesEnvelope 构造 Responses 响应骨架(含规范要求的常见字段)。 +func buildResponsesEnvelope(id, model string, createdAt int64) map[string]any { + return map[string]any{ + "id": id, + "object": "response", + "created_at": createdAt, + "completed_at": nil, + "status": "in_progress", + "error": nil, + "incomplete_details": nil, + "instructions": nil, + "max_output_tokens": nil, + "model": model, + "output": []any{}, + "parallel_tool_calls": true, + "previous_response_id": nil, + "reasoning": map[string]any{"effort": nil, "summary": nil}, + "store": true, + "temperature": 1.0, + "text": map[string]any{"format": map[string]any{"type": "text"}}, + "tool_choice": "auto", + "tools": []any{}, + "top_p": 1.0, + "truncation": "disabled", + "usage": nil, + "metadata": map[string]any{}, + } +} + +// toResponsesUsage 将 chat usage 转换为 Responses usage 结构。 +func toResponsesUsage(u map[string]any) map[string]any { + in := numOr0(u["prompt_tokens"]) + out := numOr0(u["completion_tokens"]) + total := numOr0(u["total_tokens"]) + if total == 0 { + total = in + out + } + cached, reasoningTokens := 0.0, 0.0 + if d, ok := u["prompt_tokens_details"].(map[string]any); ok { + cached = numOr0(d["cached_tokens"]) + } + if d, ok := u["completion_tokens_details"].(map[string]any); ok { + reasoningTokens = numOr0(d["reasoning_tokens"]) + } + return map[string]any{ + "input_tokens": int64(in), + "input_tokens_details": map[string]any{"cached_tokens": int64(cached)}, + "output_tokens": int64(out), + "output_tokens_details": map[string]any{"reasoning_tokens": int64(reasoningTokens)}, + "total_tokens": int64(total), + } +} + +func numOr0(v any) float64 { + if f, ok := v.(float64); ok { + return f + } + return 0 +} + +// streamResponsesResponse 将上游 Chat Completions SSE 实时转译为 Responses 语义事件流。 +func streamResponsesResponse(w http.ResponseWriter, resp *http.Response, modelName string, reqID uint64, acc *Account, prof *upstreamProfile, startTime time.Time) { + defer resp.Body.Close() + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + + flusher, ok := w.(http.Flusher) + if !ok { + writeOpenAIError(w, http.StatusInternalServerError, "streaming_unsupported", "服务器不支持流式响应 Flush") + return + } + + seq := 0 + emit := func(eventType string, data map[string]any) { + data["type"] = eventType + data["sequence_number"] = seq + seq++ + b, err := json.Marshal(data) + if err != nil { + return + } + _, _ = fmt.Fprintf(w, "event: %s\ndata: %s\n\n", eventType, b) + flusher.Flush() + } + + created := time.Now().Unix() + respID := "resp_" + compactUUID() + envelope := buildResponsesEnvelope(respID, modelName, created) + emit("response.created", map[string]any{"response": envelope}) + emit("response.in_progress", map[string]any{"response": envelope}) + + nextIndex := 0 + idxToItem := map[int]map[string]any{} + + // reasoning item 状态 + reasoningOpen := false + reasoningIndex := -1 + reasoningItemID := "" + var reasoningSB strings.Builder + + // message item 状态 + msgOpen := false + msgIndex := -1 + msgItemID := "" + var msgSB strings.Builder + + // function_call item 状态 + toolCalls := map[int]*mergedToolCall{} + var toolOrder []int + tcItemID := map[int]string{} + tcIndex := map[int]int{} + + openReasoning := func() { + if reasoningOpen { + return + } + reasoningOpen = true + reasoningIndex = nextIndex + nextIndex++ + reasoningItemID = "rs_" + compactUUID() + emit("response.output_item.added", map[string]any{ + "output_index": reasoningIndex, + "item": map[string]any{"id": reasoningItemID, "type": "reasoning", "status": "in_progress", "summary": []any{}}, + }) + emit("response.reasoning_summary_part.added", map[string]any{ + "item_id": reasoningItemID, "output_index": reasoningIndex, "summary_index": 0, + "part": map[string]any{"type": "summary_text", "text": ""}, + }) + } + + openMessage := func() { + if msgOpen { + return + } + msgOpen = true + msgIndex = nextIndex + nextIndex++ + msgItemID = "msg_" + compactUUID() + emit("response.output_item.added", map[string]any{ + "output_index": msgIndex, + "item": map[string]any{"id": msgItemID, "type": "message", "status": "in_progress", "role": "assistant", "content": []any{}}, + }) + emit("response.content_part.added", map[string]any{ + "item_id": msgItemID, "output_index": msgIndex, "content_index": 0, + "part": map[string]any{"type": "output_text", "annotations": []any{}, "text": ""}, + }) + } + + var usage map[string]any + + scanner := bufio.NewScanner(resp.Body) + scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) + for scanner.Scan() { + data := stripDataPrefix(scanner.Text()) + if data == "" || data == "[DONE]" { + continue + } + var chunk map[string]any + if json.Unmarshal([]byte(data), &chunk) != nil { + continue + } + if u, ok := chunk["usage"].(map[string]any); ok { + usage = u + } + choices, _ := chunk["choices"].([]any) + for _, c := range choices { + choice, _ := c.(map[string]any) + delta, _ := choice["delta"].(map[string]any) + if delta == nil { + continue + } + if rc, ok := delta["reasoning_content"].(string); ok && rc != "" { + openReasoning() + reasoningSB.WriteString(rc) + emit("response.reasoning_summary_text.delta", map[string]any{ + "item_id": reasoningItemID, "output_index": reasoningIndex, "summary_index": 0, "delta": rc, + }) + } + if ct, ok := delta["content"].(string); ok && ct != "" { + openMessage() + msgSB.WriteString(ct) + emit("response.output_text.delta", map[string]any{ + "item_id": msgItemID, "output_index": msgIndex, "content_index": 0, "delta": ct, + }) + } + if tcs, ok := delta["tool_calls"].([]any); ok && len(tcs) > 0 { + // 先提取本次参数增量,供合并后补发 delta 事件 + type argDelta struct { + idx int + arg string + } + var argDeltas []argDelta + for _, tcAny := range tcs { + tc, ok := tcAny.(map[string]any) + if !ok { + continue + } + idx := 0 + if v, ok := tc["index"].(float64); ok { + idx = int(v) + } + if fn, ok := tc["function"].(map[string]any); ok { + if a, ok := fn["arguments"].(string); ok && a != "" { + argDeltas = append(argDeltas, argDelta{idx: idx, arg: a}) + } + } + } + before := len(toolOrder) + applyToolCallDelta(toolCalls, &toolOrder, tcs) + for i := before; i < len(toolOrder); i++ { + idx := toolOrder[i] + st := toolCalls[idx] + if st.ID == "" { + st.ID = "call_" + compactUUID() + } + tcItemID[idx] = "fc_" + compactUUID() + tcIndex[idx] = nextIndex + nextIndex++ + emit("response.output_item.added", map[string]any{ + "output_index": tcIndex[idx], + "item": map[string]any{ + "id": tcItemID[idx], "type": "function_call", "status": "in_progress", + "call_id": st.ID, "name": st.Name, "arguments": "", + }, + }) + } + for _, ad := range argDeltas { + emit("response.function_call_arguments.delta", map[string]any{ + "item_id": tcItemID[ad.idx], "output_index": tcIndex[ad.idx], "delta": ad.arg, + }) + } + } + } + } + + // 若无任何输出,补一个空 message,保证 output 非空且事件序列完整 + if !reasoningOpen && !msgOpen && len(toolOrder) == 0 { + openMessage() + } + + // 收尾:reasoning + if reasoningOpen { + text := reasoningSB.String() + emit("response.reasoning_summary_text.done", map[string]any{"item_id": reasoningItemID, "output_index": reasoningIndex, "summary_index": 0, "text": text}) + emit("response.reasoning_summary_part.done", map[string]any{"item_id": reasoningItemID, "output_index": reasoningIndex, "summary_index": 0, "part": map[string]any{"type": "summary_text", "text": text}}) + item := map[string]any{"id": reasoningItemID, "type": "reasoning", "status": "completed", "summary": []any{map[string]any{"type": "summary_text", "text": text}}} + emit("response.output_item.done", map[string]any{"output_index": reasoningIndex, "item": item}) + idxToItem[reasoningIndex] = item + } + + // 收尾:message + if msgOpen { + text := msgSB.String() + emit("response.output_text.done", map[string]any{"item_id": msgItemID, "output_index": msgIndex, "content_index": 0, "text": text}) + emit("response.content_part.done", map[string]any{"item_id": msgItemID, "output_index": msgIndex, "content_index": 0, "part": map[string]any{"type": "output_text", "annotations": []any{}, "text": text}}) + item := map[string]any{"id": msgItemID, "type": "message", "status": "completed", "role": "assistant", "content": []any{map[string]any{"type": "output_text", "annotations": []any{}, "text": text}}} + emit("response.output_item.done", map[string]any{"output_index": msgIndex, "item": item}) + idxToItem[msgIndex] = item + } + + // 收尾:function calls + for _, idx := range toolOrder { + st := toolCalls[idx] + args := st.Args.String() + emit("response.function_call_arguments.done", map[string]any{"item_id": tcItemID[idx], "output_index": tcIndex[idx], "name": st.Name, "arguments": args}) + item := map[string]any{"id": tcItemID[idx], "type": "function_call", "status": "completed", "call_id": st.ID, "name": st.Name, "arguments": args} + emit("response.output_item.done", map[string]any{"output_index": tcIndex[idx], "item": item}) + idxToItem[tcIndex[idx]] = item + } + + output := make([]any, 0, len(idxToItem)) + for i := 0; i < nextIndex; i++ { + if item, ok := idxToItem[i]; ok { + output = append(output, item) + } + } + + final := buildResponsesEnvelope(respID, modelName, created) + final["status"] = "completed" + final["output"] = output + final["completed_at"] = time.Now().Unix() + if usage != nil { + final["usage"] = toResponsesUsage(usage) + } + emit("response.completed", map[string]any{"response": final}) + + _, _ = fmt.Fprintf(w, "data: [DONE]\n\n") + flusher.Flush() + log.Printf("[#%d] Responses 流式输出完成 (账号 %s [%s], 耗时 %v)", reqID, acc.Path, prof.Label, time.Since(startTime)) +} + +// compactUUID 返回去掉连字符的随机 UUID。 +func compactUUID() string { + return strings.ReplaceAll(uuid.NewString(), "-", "") +}