SakurasanandClaude
e96f194a59
- 上游 chat 流 finish_reason 块先于 usage 块, 原实现在 finish 时即发 response.completed 导致 usage 为空; 改为等 usage 块到达(或 [DONE] 兜底)再发, completed 携带 input/output tokens Co-Authored-By: Claude <noreply@anthropic.com>
485 lines
14 KiB
Go
485 lines
14 KiB
Go
package convert
|
|
|
|
import (
|
|
"encoding/json"
|
|
"strings"
|
|
)
|
|
|
|
// sseState 记录上一行 event 名与通用状态。
|
|
type sseState struct {
|
|
event string
|
|
}
|
|
|
|
// parseLine 解析一行 SSE；返回是否 data 行及其内容、是否 [DONE]。
|
|
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]":
|
|
return true, "[DONE]", true
|
|
case strings.HasPrefix(str, "data: "):
|
|
return true, 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
|
|
}
|
|
|
|
func newMessagesToChat() *messagesToChat { return &messagesToChat{} }
|
|
|
|
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_delta":
|
|
delta, _ := m["delta"].(map[string]any)
|
|
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
|
|
blockStarted bool
|
|
model string
|
|
stopReason string
|
|
usage any
|
|
}
|
|
|
|
func newChatToMessages() *chatToMessages { return &chatToMessages{} }
|
|
|
|
func (t *chatToMessages) line(line []byte) []byte {
|
|
isData, data, done := t.parseLine(line)
|
|
if !isData {
|
|
return nil
|
|
}
|
|
if done {
|
|
// 汇聚最终 message_delta + content_block_stop + message_stop
|
|
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
|
|
}
|
|
var out [][]byte
|
|
out = append(out, eventLine("message_delta", md))
|
|
if t.blockStarted {
|
|
out = append(out, eventLine("content_block_stop", map[string]any{"type": "content_block_stop", "index": 0}))
|
|
}
|
|
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 + content_block_start
|
|
if !t.started {
|
|
role, _ := delta["role"].(string)
|
|
content, _ := delta["content"].(string)
|
|
if role == "assistant" || content != "" {
|
|
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},
|
|
},
|
|
}))
|
|
out = append(out, eventLine("content_block_start", map[string]any{
|
|
"type": "content_block_start", "index": 0, "content_block": map[string]any{"type": "text", "text": ""},
|
|
}))
|
|
t.blockStarted = true
|
|
}
|
|
}
|
|
if content, _ := delta["content"].(string); content != "" {
|
|
if !t.started {
|
|
t.started = true
|
|
t.blockStarted = 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{}},
|
|
}))
|
|
out = append(out, eventLine("content_block_start", map[string]any{
|
|
"type": "content_block_start", "index": 0, "content_block": map[string]any{"type": "text", "text": ""},
|
|
}))
|
|
}
|
|
out = append(out, eventLine("content_block_delta", map[string]any{
|
|
"type": "content_block_delta", "index": 0, "delta": map[string]any{"type": "text_delta", "text": content},
|
|
}))
|
|
}
|
|
if u, ok := m["usage"]; ok {
|
|
t.usage = u
|
|
}
|
|
return joinLines(out)
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
func newResponsesToMessages() *responsesToMessages { return &responsesToMessages{} }
|
|
|
|
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
|
|
switch evt {
|
|
case "response.created":
|
|
if !t.started {
|
|
t.started = true
|
|
id, _ := m["response"].(map[string]any)
|
|
rid := ""
|
|
if id != nil {
|
|
rid, _ = id["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{},
|
|
},
|
|
}))
|
|
out = append(out, eventLine("content_block_start", map[string]any{
|
|
"type": "content_block_start", "index": 0, "content_block": map[string]any{"type": "text", "text": ""},
|
|
}))
|
|
}
|
|
case "response.output_text.delta":
|
|
delta, _ := m["delta"].(string)
|
|
if delta != "" {
|
|
out = append(out, eventLine("content_block_delta", map[string]any{
|
|
"type": "content_block_delta", "index": 0, "delta": map[string]any{"type": "text_delta", "text": delta},
|
|
}))
|
|
}
|
|
case "response.completed":
|
|
out = append(out, eventLine("message_delta", map[string]any{
|
|
"type": "message_delta", "delta": map[string]any{"stop_reason": "end_turn", "stop_sequence": nil},
|
|
}))
|
|
out = append(out, eventLine("content_block_stop", map[string]any{"type": "content_block_stop", "index": 0}))
|
|
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)
|
|
}
|