package convert import ( "encoding/json" "strings" ) // sseState 记录上一行 event 名与通用状态。 type sseState struct { event string } // parseLine 解析一行 SSE;返回是否 data 行及其内容、是否 [DONE]。 // data: 后可跟空格(标准)或紧贴 JSON(上游如火山方舟会省略空格)。 func (s *sseState) parseLine(line []byte) (isData bool, data string, done bool) { str := strings.TrimRight(string(line), "\r\n") switch { case strings.HasPrefix(str, "event: "): s.event = strings.TrimSpace(strings.TrimPrefix(str, "event: ")) return false, "", false case str == "data: [DONE]" || str == "data:[DONE]": return true, "[DONE]", true case strings.HasPrefix(str, "data:"): return true, strings.TrimLeft(strings.TrimPrefix(str, "data:"), " "), false default: return false, "", false } } func eventData(line string) map[string]any { var m map[string]any _ = json.Unmarshal([]byte(line), &m) return m } func dataLine(obj any) []byte { b, _ := json.Marshal(obj) return append(append([]byte("data: "), b...), '\n', '\n') } func eventLine(name string, obj any) []byte { b, _ := json.Marshal(obj) out := append([]byte("event: "+name+"\ndata: "), b...) return append(out, '\n', '\n') } // --------------------------------------------------------------------------- // Messages → Chat type messagesToChat struct { sseState id, model string toolIdx map[int]int // messages content block index → chat tool_calls index(顺序编号,避开文本块) nextTool int } func newMessagesToChat() *messagesToChat { return &messagesToChat{toolIdx: map[int]int{}} } func (t *messagesToChat) line(line []byte) []byte { isData, data, done := t.parseLine(line) if !isData { return nil } if done { return []byte("data: [DONE]\n\n") } m := eventData(data) evt, _ := m["type"].(string) switch evt { case "message_start": msg, _ := m["message"].(map[string]any) t.id, _ = msg["id"].(string) t.model, _ = msg["model"].(string) return dataLine(map[string]any{ "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "object": "chat.completion.chunk", "model": t.model, "choices": []any{map[string]any{"index": 0, "delta": map[string]any{"role": "assistant"}, "finish_reason": nil}}, }) case "content_block_start": cb, _ := m["content_block"].(map[string]any) if cb == nil || cb["type"] != "tool_use" { return nil } blockIdx, _ := m["index"].(float64) tool := t.nextTool t.nextTool++ t.toolIdx[int(blockIdx)] = tool toolID, _ := cb["id"].(string) name, _ := cb["name"].(string) return dataLine(map[string]any{ "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "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": tool, "id": toolID, "type": "function", "function": map[string]any{"name": name, "arguments": ""}}}, }, "finish_reason": nil}}, }) case "content_block_delta": delta, _ := m["delta"].(map[string]any) deltaType, _ := delta["type"].(string) if deltaType == "input_json_delta" { blockIdx, _ := m["index"].(float64) tool, ok := t.toolIdx[int(blockIdx)] if !ok { return nil } partial, _ := delta["partial_json"].(string) if partial == "" { return nil } return dataLine(map[string]any{ "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "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": tool, "function": map[string]any{"arguments": partial}}}, }, "finish_reason": nil}}, }) } text, _ := delta["text"].(string) if text == "" { return nil } return dataLine(map[string]any{ "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "object": "chat.completion.chunk", "model": t.model, "choices": []any{map[string]any{"index": 0, "delta": map[string]any{"content": text}, "finish_reason": nil}}, }) case "message_delta": delta, _ := m["delta"].(map[string]any) stop, _ := delta["stop_reason"].(string) var out [][]byte out = append(out, dataLine(map[string]any{ "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "object": "chat.completion.chunk", "model": t.model, "choices": []any{map[string]any{"index": 0, "delta": map[string]any{}, "finish_reason": messagesStopToChat(stop)}}, })) if u, ok := m["usage"]; ok { out = append(out, dataLine(map[string]any{ "id": "chatcmpl-" + strings.TrimPrefix(t.id, "msg_"), "object": "chat.completion.chunk", "model": t.model, "choices": []any{}, "usage": u, })) } return joinLines(out) case "message_stop": return []byte("data: [DONE]\n\n") } return nil } func joinLines(lines [][]byte) []byte { return []byte(strings.Join(func() []string { var s []string for _, l := range lines { s = append(s, string(l)) } return s }(), "")) } // --------------------------------------------------------------------------- // Chat → Messages type chatToMessages struct { sseState started bool // message_start 已发出 nextIndex int // 下一个 content block index(顺序分配) textIndex int // 文本块 index;-1 = 未开始 toolIdx map[int]int // chat delta.tool_calls[].index → messages block index openBlocks []int // 已开始未停止的 block index,按开始顺序 model string stopReason string usage any } func newChatToMessages() *chatToMessages { return &chatToMessages{textIndex: -1, toolIdx: map[int]int{}} } func (t *chatToMessages) line(line []byte) []byte { isData, data, done := t.parseLine(line) if !isData { return nil } if done { // 汇聚最终:先对每个已开始未停止的块发 content_block_stop,再 message_delta + message_stop var out [][]byte for _, idx := range t.openBlocks { out = append(out, eventLine("content_block_stop", map[string]any{"type": "content_block_stop", "index": idx})) } md := map[string]any{"type": "message_delta", "delta": map[string]any{ "stop_reason": stopReasonOrEnd(t.stopReason), "stop_sequence": nil, }} if t.usage != nil { md["usage"] = t.usage } out = append(out, eventLine("message_delta", md)) out = append(out, eventLine("message_stop", map[string]any{"type": "message_stop"})) return joinLines(out) } m := eventData(data) // chat 块:delta / finish_reason 在 choices[0] 内 delta := map[string]any{} if choices, ok := m["choices"].([]any); ok && len(choices) > 0 { if c0, ok := choices[0].(map[string]any); ok { if d, ok := c0["delta"].(map[string]any); ok { delta = d } if fr, _ := c0["finish_reason"].(string); fr != "" { t.stopReason = fr } } } if t.model == "" { t.model, _ = m["model"].(string) } id, _ := m["id"].(string) var out [][]byte // message_start 只在实际有内容(文本或工具)时发出,避免 GLM 等模型的 // reasoning_content 块(带 role 无 content)提前开出一个空文本块。 ensureStarted := func() { if t.started { return } t.started = true out = append(out, eventLine("message_start", map[string]any{ "type": "message_start", "message": map[string]any{ "id": "msg_" + strings.TrimPrefix(id, "chatcmpl-"), "type": "message", "role": "assistant", "model": t.model, "content": []any{}, "usage": map[string]any{"input_tokens": 0, "output_tokens": 0}, }, })) } // 文本:delta.content(string;兼容 {type:text,text} 数组) if content := deltaText(delta); content != "" { if t.textIndex < 0 { t.textIndex = t.nextIndex t.nextIndex++ ensureStarted() out = append(out, eventLine("content_block_start", map[string]any{ "type": "content_block_start", "index": t.textIndex, "content_block": map[string]any{"type": "text", "text": ""}, })) t.openBlocks = append(t.openBlocks, t.textIndex) } out = append(out, eventLine("content_block_delta", map[string]any{ "type": "content_block_delta", "index": t.textIndex, "delta": map[string]any{"type": "text_delta", "text": content}, })) } // 工具调用:delta.tool_calls(并行调用各 index 独立成块;arguments 支持整段/分段两种流式) if tcs, ok := delta["tool_calls"].([]any); ok { for _, tc := range tcs { call, ok := tc.(map[string]any) if !ok { continue } idx, _ := call["index"].(float64) tcIdx := int(idx) fn, _ := call["function"].(map[string]any) name, _ := fn["name"].(string) args, _ := fn["arguments"].(string) blockIdx, seen := t.toolIdx[tcIdx] if !seen { blockIdx = t.nextIndex t.nextIndex++ t.toolIdx[tcIdx] = blockIdx toolID, _ := call["id"].(string) ensureStarted() out = append(out, eventLine("content_block_start", map[string]any{ "type": "content_block_start", "index": blockIdx, "content_block": map[string]any{ "type": "tool_use", "id": toolID, "name": name, "input": map[string]any{}, }, })) t.openBlocks = append(t.openBlocks, blockIdx) } if args != "" { out = append(out, eventLine("content_block_delta", map[string]any{ "type": "content_block_delta", "index": blockIdx, "delta": map[string]any{"type": "input_json_delta", "partial_json": args}, })) } } } if u, ok := m["usage"]; ok { t.usage = u } return joinLines(out) } // deltaText 取 chat delta.content 文本(string 或 [{type:text,text}] 数组拼接)。 func deltaText(delta map[string]any) string { if s, ok := delta["content"].(string); ok { return s } if arr, ok := delta["content"].([]any); ok { var parts []string for _, b := range arr { if bm, ok := b.(map[string]any); ok { if t, _ := bm["text"].(string); t != "" { parts = append(parts, t) } } } return strings.Join(parts, "") } return "" } func stopReasonOrEnd(s string) string { if s == "" { return "end_turn" } return chatStopToMessages(s) } // --------------------------------------------------------------------------- // Responses → Messages type responsesToMessages struct { sseState started bool model string usage any nextIndex int // 下一个 content block index(顺序分配) textIndex int // 文本块 index;-1 = 未开始 toolIdx map[string]int // function_call item_id → messages block index openBlocks []int // 已开始未停止的 block index,按开始顺序 anyTool bool } func newResponsesToMessages() *responsesToMessages { return &responsesToMessages{textIndex: -1, toolIdx: map[string]int{}} } func (t *responsesToMessages) line(line []byte) []byte { isData, data, done := t.parseLine(line) if !isData || done { return nil } m := eventData(data) evt, _ := m["type"].(string) if resp, ok := m["response"].(map[string]any); ok { if t.model == "" { t.model, _ = resp["model"].(string) } if u, ok := resp["usage"]; ok { t.usage = u } } var out [][]byte // message_start 只在 response.created 时发出;文本/工具块在对应事件到达时再开, // 避免纯函数调用响应提前开出一个空文本块。 ensureStarted := func() { if t.started { return } t.started = true rid := "" if resp, ok := m["response"].(map[string]any); ok { rid, _ = resp["id"].(string) } out = append(out, eventLine("message_start", map[string]any{ "type": "message_start", "message": map[string]any{ "id": "msg_" + strings.TrimPrefix(rid, "resp_"), "type": "message", "role": "assistant", "model": t.model, "content": []any{}, }, })) } switch evt { case "response.created": ensureStarted() case "response.output_text.delta": delta, _ := m["delta"].(string) if delta == "" { return nil } if t.textIndex < 0 { t.textIndex = t.nextIndex t.nextIndex++ ensureStarted() out = append(out, eventLine("content_block_start", map[string]any{ "type": "content_block_start", "index": t.textIndex, "content_block": map[string]any{"type": "text", "text": ""}, })) t.openBlocks = append(t.openBlocks, t.textIndex) } out = append(out, eventLine("content_block_delta", map[string]any{ "type": "content_block_delta", "index": t.textIndex, "delta": map[string]any{"type": "text_delta", "text": delta}, })) case "response.output_item.added": item, _ := m["item"].(map[string]any) if item == nil || item["type"] != "function_call" { return nil } blockIdx := t.nextIndex t.nextIndex++ t.anyTool = true itemID, _ := item["id"].(string) t.toolIdx[itemID] = blockIdx toolUseID, _ := item["call_id"].(string) if toolUseID == "" { toolUseID = itemID } name, _ := item["name"].(string) ensureStarted() out = append(out, eventLine("content_block_start", map[string]any{ "type": "content_block_start", "index": blockIdx, "content_block": map[string]any{ "type": "tool_use", "id": toolUseID, "name": name, "input": map[string]any{}, }, })) t.openBlocks = append(t.openBlocks, blockIdx) case "response.function_call_arguments.delta": itemID, _ := m["item_id"].(string) blockIdx, ok := t.toolIdx[itemID] if !ok { return nil } delta, _ := m["delta"].(string) if delta == "" { return nil } out = append(out, eventLine("content_block_delta", map[string]any{ "type": "content_block_delta", "index": blockIdx, "delta": map[string]any{"type": "input_json_delta", "partial_json": delta}, })) case "response.completed": for _, idx := range t.openBlocks { out = append(out, eventLine("content_block_stop", map[string]any{"type": "content_block_stop", "index": idx})) } stop := "end_turn" if t.anyTool { stop = "tool_use" } md := map[string]any{"type": "message_delta", "delta": map[string]any{"stop_reason": stop, "stop_sequence": nil}} if t.usage != nil { md["usage"] = t.usage } out = append(out, eventLine("message_delta", md)) out = append(out, eventLine("message_stop", map[string]any{"type": "message_stop"})) } return joinLines(out) } // --------------------------------------------------------------------------- // Messages → Responses type messagesToResponses struct { sseState model string usage any done bool } func newMessagesToResponses() *messagesToResponses { return &messagesToResponses{} } func (t *messagesToResponses) line(line []byte) []byte { isData, data, done := t.parseLine(line) if !isData || done { return nil } m := eventData(data) evt, _ := m["type"].(string) if msg, ok := m["message"].(map[string]any); ok { if t.model == "" { t.model, _ = msg["model"].(string) } if u, ok := msg["usage"]; ok { t.usage = u } } if u, ok := m["usage"]; ok { t.usage = u } var out [][]byte switch evt { case "message_start": id, _ := m["message"].(map[string]any) rid := "" if id != nil { rid, _ = id["id"].(string) } out = append(out, eventLine("response.created", map[string]any{ "type": "response.created", "response": map[string]any{ "id": "resp_" + strings.TrimPrefix(rid, "msg_"), "object": "response", "model": t.model, "status": "in_progress", }, })) 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, })) } case "message_stop": if !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, }, })) } } return joinLines(out) } // --------------------------------------------------------------------------- // Responses → Chat type responsesToChat struct { sseState id, model string } func newResponsesToChat() *responsesToChat { return &responsesToChat{} } func (t *responsesToChat) line(line []byte) []byte { isData, data, done := t.parseLine(line) if !isData { return nil } if done { return nil } m := eventData(data) evt, _ := m["type"].(string) if resp, ok := m["response"].(map[string]any); ok { if t.model == "" { t.model, _ = resp["model"].(string) } if t.id == "" { t.id, _ = resp["id"].(string) } } var out [][]byte switch evt { case "response.created": 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{"role": "assistant"}, "finish_reason": nil}}, })) 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}}, })) } 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"}}, })) if u, ok := m["response"].(map[string]any); ok { if usage, ok := u["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, })) } } out = append(out, []byte("data: [DONE]\n\n")) } return joinLines(out) } // --------------------------------------------------------------------------- // Chat → Responses type chatToResponses struct { sseState model string usage any finishSeen bool done bool } func newChatToResponses() *chatToResponses { return &chatToResponses{} } func (t *chatToResponses) line(line []byte) []byte { isData, data, done := t.parseLine(line) if !isData { return nil } if done { // 流结束兜底: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 nil } m := eventData(data) if t.model == "" { t.model, _ = m["model"].(string) } if u, ok := m["usage"]; ok { t.usage = u } delta := map[string]any{} var finish string if choices, ok := m["choices"].([]any); ok && len(choices) > 0 { if c0, ok := choices[0].(map[string]any); ok { if d, ok := c0["delta"].(map[string]any); ok { delta = d } finish, _ = c0["finish_reason"].(string) } } if finish != "" { t.finishSeen = true } var out [][]byte if role, _ := delta["role"].(string); role == "assistant" { out = append(out, eventLine("response.created", map[string]any{ "type": "response.created", "response": map[string]any{"id": "resp_stream", "object": "response", "model": t.model, "status": "in_progress"}, })) } if content, _ := delta["content"].(string); content != "" { 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, })) } // 上游 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, }, })) } return joinLines(out) }