From 610b3300250da72c45e2d160e99a61cb6aeac810 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Wed, 30 Sep 2026 08:45:04 +0800 Subject: [PATCH] =?UTF-8?q?refactor(model):=20=E6=8B=86=20openai.go=20652?= =?UTF-8?q?=E2=86=92216=E8=A1=8C(=E7=BA=AF=E6=90=AC=E7=A7=BB=E9=9B=B6?= =?UTF-8?q?=E9=80=BB=E8=BE=91=E6=94=B9=E5=8A=A8)=20=E2=80=94=20encode/resp?= =?UTF-8?q?onse/stream=20=E4=B8=89=E4=BB=B6=E5=A5=97=E5=AF=B9=E9=BD=90=20a?= =?UTF-8?q?nthropic=20=E5=B8=83=E5=B1=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - openai.go 216(适配器+Chat/ChatStream+共享助手); openai_encode.go 202(请求wire+build*); openai_response.go 79(响应wire+responseFromWire); openai_stream.go 174(scanSSE) - 字节级比对 body identical; 导出API不变(go doc diff 空); gofmt 干净 - 纯净基线(HEAD+本改)全仓 go build/vet/test 绿 + win/darwin 交叉编译绿 - 遗留: scanSSE 138行(>80, 拆分前既有,零逻辑改动未动), 另开步骤重构 --- internal/runtime/model/openai.go | 436 ---------------------- internal/runtime/model/openai_encode.go | 202 ++++++++++ internal/runtime/model/openai_response.go | 79 ++++ internal/runtime/model/openai_stream.go | 174 +++++++++ 4 files changed, 455 insertions(+), 436 deletions(-) create mode 100644 internal/runtime/model/openai_encode.go create mode 100644 internal/runtime/model/openai_response.go create mode 100644 internal/runtime/model/openai_stream.go diff --git a/internal/runtime/model/openai.go b/internal/runtime/model/openai.go index 4c12ae2..2e9c2b7 100644 --- a/internal/runtime/model/openai.go +++ b/internal/runtime/model/openai.go @@ -1,7 +1,6 @@ package model import ( - "bufio" "bytes" "context" "encoding/json" @@ -30,132 +29,6 @@ const ( defaultHTTPTimeout = 180 * time.Second ) -// openAIChatRequest is the JSON body sent to /chat/completions. Field names -// match the provider protocol so the struct marshals directly. -type openAIChatRequest struct { - Model string `json:"model"` - Messages []openAIMessage `json:"messages"` - Temperature *float64 `json:"temperature,omitempty"` - MaxTokens int `json:"max_tokens,omitempty"` - Stream bool `json:"stream,omitempty"` - StreamOptions *openAIStreamOpts `json:"stream_options,omitempty"` - Tools []openAITool `json:"tools,omitempty"` -} - -// openAIStreamOpts asks the provider to include usage in the final stream chunk. -type openAIStreamOpts struct { - IncludeUsage bool `json:"include_usage"` -} - -// openAIMessage is one request message in the wire body. Content is either a -// plain string (text-only) or an array of {type,text|image_url} parts. Tool -// messages carry ToolCallID (required by OpenAI); assistant messages that -// requested functions carry ToolCalls so the following tool messages can be -// correlated. -type openAIMessage struct { - Role string `json:"role"` - Content any `json:"content"` - ToolCallID string `json:"tool_call_id,omitempty"` - ToolCalls []openAIToolCall `json:"tool_calls,omitempty"` -} - -// openAIToolCall is the wire form of one assistant tool_call in a request. OpenAI -// requires each subsequent tool message to reference the id back to these calls. -type openAIToolCall struct { - ID string `json:"id"` - Type string `json:"type"` - Function openAIToolCallFn `json:"function"` -} - -// openAITool is the wire form of a function tool schema (M2 placeholder). -type openAITool struct { - Type string `json:"type"` - Function openAIToolFn `json:"function"` -} - -// openAIToolFn carries the function definition. -type openAIToolFn struct { - Name string `json:"name"` - Description string `json:"description,omitempty"` - Parameters any `json:"parameters,omitempty"` -} - -// openAIChatResponse is the JSON body returned by /chat/completions, for both -// the non-streaming payload and each SSE stream chunk. -type openAIChatResponse struct { - ID string `json:"id"` - Model string `json:"model"` - Choices []openAIChoice `json:"choices"` - Usage *openAIUsage `json:"usage,omitempty"` -} - -// openAIChoice is one candidate completion. -type openAIChoice struct { - Index int `json:"index"` - Message openAIResponseMessage `json:"message"` - Delta *openAIDelta `json:"delta,omitempty"` - FinishReason string `json:"finish_reason,omitempty"` -} - -// openAIResponseMessage is the assistant message inside a non-streaming choice. -type openAIResponseMessage struct { - Role string `json:"role"` - Content string `json:"content"` - ToolCalls []openAIToolCall `json:"tool_calls,omitempty"` -} - -// openAIDelta is the incremental payload inside a stream chunk. Reasoning content -// (OpenAI o-series / DeepSeek / DashScope) arrives in ReasoningContent; streaming -// function calls arrive in ToolCalls, accumulated per fragment index. -type openAIDelta struct { - Role string `json:"role"` - Content string `json:"content"` - ReasoningContent string `json:"reasoning_content"` - ToolCalls []openAIToolCallDelta `json:"tool_calls"` -} - -// openAIToolCallDelta is one streaming function-call fragment. OpenAI emits a -// sequence of these sharing the same Index: the ID and Function.Name usually -// arrive once (in the first fragment) while Function.Arguments streams across -// fragments. -type openAIToolCallDelta struct { - Index int `json:"index"` - ID string `json:"id"` - Function openAIToolCallFn `json:"function"` -} - -// openAIToolCallFn is the function part of a streaming tool-call fragment. -type openAIToolCallFn struct { - Name string `json:"name"` - Arguments string `json:"arguments"` -} - -// openAIToolAcc accumulates the partial state of one streaming tool call (keyed -// by the provider's tool_calls[].index) so its arguments re-assemble across the -// delta fragments. started guards a single ToolCallStart emission per block. -type openAIToolAcc struct { - id string - name string - args strings.Builder - started bool -} - -// idOrName resolves the block identifier used to correlate start/delta/end: the -// provider tool-call id when present, else the function name. -func (a *openAIToolAcc) idOrName() string { - if a.id != "" { - return a.id - } - return a.name -} - -// openAIUsage reports token consumption. -type openAIUsage struct { - PromptTokens int `json:"prompt_tokens"` - CompletionTokens int `json:"completion_tokens"` - TotalTokens int `json:"total_tokens"` -} - // OpenAIChatAdapter calls an OpenAI-compatible /chat/completions endpoint. It is // constructed from a config.LLMConfig and implements Adapter for both // non-streaming and token-level SSE streaming. @@ -308,146 +181,6 @@ func (a *OpenAIChatAdapter) endpoint() string { return a.baseURL + "/chat/completions" } -func (a *OpenAIChatAdapter) buildRequest(req Request, stream bool) ([]byte, error) { - msgs, err := buildMessages(req.Messages) - if err != nil { - return nil, err - } - wire := openAIChatRequest{ - Model: req.Model, - Messages: msgs, - Temperature: req.Temperature, - MaxTokens: req.MaxTokens, - Stream: stream, - Tools: buildTools(req.Tools), - } - if stream { - wire.StreamOptions = &openAIStreamOpts{IncludeUsage: true} - } - buf, err := json.Marshal(wire) - if err != nil { - return nil, fmt.Errorf("model: %w: marshal request: %v", ErrBadRequest, err) - } - return buf, nil -} - -func buildMessages(msgs []Message) ([]openAIMessage, error) { - out := make([]openAIMessage, 0, len(msgs)) - for _, m := range msgs { - wire, err := encodeOpenAIMessage(m) - if err != nil { - return nil, err - } - out = append(out, wire) - } - return out, nil -} - -// encodeOpenAIMessage renders one normalized message in OpenAI wire form. A -// RoleTool turn becomes a tool message carrying tool_call_id (required by OpenAI, -// and the audit's gap); an assistant turn that advertised tool calls carries the -// corresponding tool_calls array so the provider can correlate the tool messages. -func encodeOpenAIMessage(m Message) (openAIMessage, error) { - if m.Role == RoleTool { - return openAIMessage{ - Role: string(RoleTool), - ToolCallID: toolMessageCallID(m), - Content: messageText(m), - }, nil - } - content, err := encodeContent(m.Content) - if err != nil { - return openAIMessage{}, err - } - return openAIMessage{Role: string(m.Role), Content: content, ToolCalls: encodeToolCalls(m.Content)}, nil -} - -// toolMessageCallID resolves the tool_call_id for a RoleTool turn: the -// message-level id, else the id carried on the first tool-result part. -func toolMessageCallID(m Message) string { - if m.ToolCallID != "" { - return m.ToolCallID - } - for _, p := range m.Content { - if p.ToolCallID != "" { - return p.ToolCallID - } - } - return "" -} - -// encodeToolCalls builds the assistant tool_calls array from a turn's tool-call -// parts. Arguments are serialized to JSON because OpenAI expects a string. -func encodeToolCalls(parts []Content) []openAIToolCall { - calls := make([]openAIToolCall, 0, len(parts)) - for _, p := range parts { - if p.Type != ContentTypeToolCall { - continue - } - arguments := "{}" - if len(p.ToolArgs) > 0 { - if b, err := json.Marshal(p.ToolArgs); err == nil { - arguments = string(b) - } - } - calls = append(calls, openAIToolCall{ - ID: p.ToolCallID, - Type: "function", - Function: openAIToolCallFn{Name: p.ToolCallName, Arguments: arguments}, - }) - } - return calls -} - -func encodeContent(parts []Content) (any, error) { - if len(parts) == 0 { - return "", nil - } - arr := make([]any, 0, len(parts)) - for _, p := range parts { - switch p.Type { - case ContentTypeText: - arr = append(arr, map[string]any{"type": "text", "text": p.Text}) - case ContentTypeImage: - arr = append(arr, map[string]any{ - "type": "image_url", - "image_url": map[string]any{"url": p.ImageURL}, - }) - case ContentTypeToolCall, ContentTypeToolResult: - // Tool-call and tool-result parts are encoded at the message level - // (tool_calls array / tool_call_id), never as text-image content. - continue - default: - return nil, fmt.Errorf("model: %w: unsupported content type %q", ErrBadRequest, p.Type) - } - } - if len(arr) == 0 { - return "", nil - } - if len(arr) == 1 { - if blk, ok := arr[0].(map[string]any); ok && blk["type"] == "image_url" { - return arr, nil - } - return arr[0].(map[string]any)["text"], nil - } - return arr, nil -} - -func buildTools(tools []ToolSpec) []openAITool { - if len(tools) == 0 { - return nil - } - out := make([]openAITool, 0, len(tools)) - for _, t := range tools { - out = append(out, openAITool{Type: "function", Function: openAIToolFn{ - Name: t.Name, - Description: t.Description, - Parameters: t.Parameters, - }}) - } - return out -} - func (a *OpenAIChatAdapter) doRequest(httpReq *http.Request, stream bool) (*http.Response, error) { if a.apiKey != "" { httpReq.Header.Set("Authorization", "Bearer "+a.apiKey) @@ -481,172 +214,3 @@ func (a *OpenAIChatAdapter) record(req Request, start time.Time, tokens int) { TookMs: time.Since(start).Milliseconds(), }) } - -func responseFromWire(wire *openAIChatResponse) Response { - out := Response{Model: wire.Model} - if len(wire.Choices) > 0 { - text := wire.Choices[0].Message.Content - out.Content = text - out.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: text}}} - out.StopReason = wire.Choices[0].FinishReason - for _, tc := range wire.Choices[0].Message.ToolCalls { - args := map[string]any{} - if tc.Function.Arguments != "" { - _ = json.Unmarshal([]byte(tc.Function.Arguments), &args) - } - out.ToolCalls = append(out.ToolCalls, ToolCall{Name: tc.Function.Name, ID: tc.ID, Args: args}) - } - } - if u := wire.Usage; u != nil { - out.PromptTokens = u.PromptTokens - out.CompletionTokens = u.CompletionTokens - out.TotalTokens = u.TotalTokens - } - return out -} - -// scanSSE incrementally decodes stream chunks, accumulates the full body text -// with a strings.Builder (never reading the whole body at once) and captures -// usage from the final chunk. It splits each delta into its segment and brackets -// them: reasoning_content as thinking_start/delta/end, tool_calls fragments as -// tool_call_start/delta/end, and body text as text_start/delta/end. The -// accumulated tool-call arguments re-assemble across fragments into a structured -// Args map, and are attached to the returned Response alongside the text. -func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) { - scanner := bufio.NewScanner(r) - scanner.Buffer(make([]byte, 0, sseBufInit), maxSSELineBytes) - agg := Response{} - var text strings.Builder - textStarted := false - thinkingStarted := false - toolCalls := map[int]*openAIToolAcc{} - var toolOrder []int - - endThinking := func() { - if onEvent != nil && thinkingStarted { - thinkingStarted = false - onEvent(StreamEvent{Kind: StreamKindThinkingEnd}) - } - } - - for scanner.Scan() { - line := strings.TrimSpace(scanner.Text()) - if line == "" || !strings.HasPrefix(line, sseDataPrefix) { - continue - } - data := strings.TrimSpace(strings.TrimPrefix(line, sseDataPrefix)) - if data == sseDone { - break - } - var chunk openAIChatResponse - if err := json.Unmarshal([]byte(data), &chunk); err != nil { - return Response{}, fmt.Errorf("model: %w: decode stream chunk: %v", ErrUpstream, err) - } - if u := chunk.Usage; u != nil { - agg.PromptTokens = u.PromptTokens - agg.CompletionTokens = u.CompletionTokens - agg.TotalTokens = u.TotalTokens - } - if len(chunk.Choices) > 0 && chunk.Choices[0].FinishReason != "" { - agg.StopReason = chunk.Choices[0].FinishReason - } - if len(chunk.Choices) == 0 || chunk.Choices[0].Delta == nil { - continue - } - delta := chunk.Choices[0].Delta - - // Reasoning / thinking segment (OpenAI o-series, DeepSeek, DashScope). - if delta.ReasoningContent != "" { - if onEvent != nil && !thinkingStarted { - thinkingStarted = true - onEvent(StreamEvent{Kind: StreamKindThinkingStart}) - } - if onEvent != nil { - onEvent(StreamEvent{Kind: StreamKindThinkingDelta, Delta: delta.ReasoningContent}) - } - continue - } - - // Streaming tool-call fragments. - if len(delta.ToolCalls) > 0 { - endThinking() - for _, tc := range delta.ToolCalls { - acc := toolCalls[tc.Index] - if acc == nil { - acc = &openAIToolAcc{} - toolCalls[tc.Index] = acc - toolOrder = append(toolOrder, tc.Index) - } - if acc.id == "" { - acc.id = tc.ID - } - if acc.name == "" { - acc.name = tc.Function.Name - } - if onEvent != nil && !acc.started && acc.idOrName() != "" { - acc.started = true - onEvent(StreamEvent{Kind: StreamKindToolCallStart, Index: tc.Index, ID: acc.idOrName()}) - } - if tc.Function.Arguments != "" { - acc.args.WriteString(tc.Function.Arguments) - if onEvent != nil { - onEvent(StreamEvent{Kind: StreamKindToolCallDelta, Index: tc.Index, ID: acc.idOrName(), Delta: tc.Function.Arguments}) - } - } - } - continue - } - - // Body-text segment. - if delta.Content == "" { - continue - } - endThinking() - text.WriteString(delta.Content) - if onEvent != nil { - if !textStarted { - textStarted = true - onEvent(StreamEvent{Kind: StreamKindTextStart}) - } - onEvent(StreamEvent{Kind: StreamKindTextDelta, Delta: delta.Content}) - } - } - if err := scanner.Err(); err != nil { - return Response{}, fmt.Errorf("model: %w: read stream: %v", ErrUpstream, err) - } - - if onEvent != nil { - if thinkingStarted { - onEvent(StreamEvent{Kind: StreamKindThinkingEnd}) - } - if textStarted { - onEvent(StreamEvent{Kind: StreamKindTextEnd}) - } - // OpenAI has no per-block stop event, so each tool_call block is closed - // when the stream terminates. - for _, idx := range toolOrder { - acc := toolCalls[idx] - if acc == nil || !acc.started { - continue - } - onEvent(StreamEvent{Kind: StreamKindToolCallEnd, Index: idx, ID: acc.idOrName()}) - } - } - - for _, idx := range toolOrder { - acc := toolCalls[idx] - if acc == nil { - continue - } - args := map[string]any{} - if acc.args.Len() > 0 { - if err := json.Unmarshal([]byte(acc.args.String()), &args); err != nil { - args = map[string]any{} - } - } - agg.ToolCalls = append(agg.ToolCalls, ToolCall{Name: acc.name, ID: acc.id, Args: args}) - } - agg.Content = text.String() - agg.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: agg.Content}}} - return agg, nil -} diff --git a/internal/runtime/model/openai_encode.go b/internal/runtime/model/openai_encode.go new file mode 100644 index 0000000..5d5279d --- /dev/null +++ b/internal/runtime/model/openai_encode.go @@ -0,0 +1,202 @@ +package model + +import ( + "encoding/json" + "fmt" +) + +// openAIChatRequest is the JSON body sent to /chat/completions. Field names +// match the provider protocol so the struct marshals directly. +type openAIChatRequest struct { + Model string `json:"model"` + Messages []openAIMessage `json:"messages"` + Temperature *float64 `json:"temperature,omitempty"` + MaxTokens int `json:"max_tokens,omitempty"` + Stream bool `json:"stream,omitempty"` + StreamOptions *openAIStreamOpts `json:"stream_options,omitempty"` + Tools []openAITool `json:"tools,omitempty"` +} + +// openAIStreamOpts asks the provider to include usage in the final stream chunk. +type openAIStreamOpts struct { + IncludeUsage bool `json:"include_usage"` +} + +// openAIMessage is one request message in the wire body. Content is either a +// plain string (text-only) or an array of {type,text|image_url} parts. Tool +// messages carry ToolCallID (required by OpenAI); assistant messages that +// requested functions carry ToolCalls so the following tool messages can be +// correlated. +type openAIMessage struct { + Role string `json:"role"` + Content any `json:"content"` + ToolCallID string `json:"tool_call_id,omitempty"` + ToolCalls []openAIToolCall `json:"tool_calls,omitempty"` +} + +// openAIToolCall is the wire form of one assistant tool_call in a request. OpenAI +// requires each subsequent tool message to reference the id back to these calls. +type openAIToolCall struct { + ID string `json:"id"` + Type string `json:"type"` + Function openAIToolCallFn `json:"function"` +} + +// openAITool is the wire form of a function tool schema (M2 placeholder). +type openAITool struct { + Type string `json:"type"` + Function openAIToolFn `json:"function"` +} + +// openAIToolFn carries the function definition. +type openAIToolFn struct { + Name string `json:"name"` + Description string `json:"description,omitempty"` + Parameters any `json:"parameters,omitempty"` +} + +// openAIToolCallFn is the function part of a streaming tool-call fragment. +type openAIToolCallFn struct { + Name string `json:"name"` + Arguments string `json:"arguments"` +} + +func (a *OpenAIChatAdapter) buildRequest(req Request, stream bool) ([]byte, error) { + msgs, err := buildMessages(req.Messages) + if err != nil { + return nil, err + } + wire := openAIChatRequest{ + Model: req.Model, + Messages: msgs, + Temperature: req.Temperature, + MaxTokens: req.MaxTokens, + Stream: stream, + Tools: buildTools(req.Tools), + } + if stream { + wire.StreamOptions = &openAIStreamOpts{IncludeUsage: true} + } + buf, err := json.Marshal(wire) + if err != nil { + return nil, fmt.Errorf("model: %w: marshal request: %v", ErrBadRequest, err) + } + return buf, nil +} + +func buildMessages(msgs []Message) ([]openAIMessage, error) { + out := make([]openAIMessage, 0, len(msgs)) + for _, m := range msgs { + wire, err := encodeOpenAIMessage(m) + if err != nil { + return nil, err + } + out = append(out, wire) + } + return out, nil +} + +// encodeOpenAIMessage renders one normalized message in OpenAI wire form. A +// RoleTool turn becomes a tool message carrying tool_call_id (required by OpenAI, +// and the audit's gap); an assistant turn that advertised tool calls carries the +// corresponding tool_calls array so the provider can correlate the tool messages. +func encodeOpenAIMessage(m Message) (openAIMessage, error) { + if m.Role == RoleTool { + return openAIMessage{ + Role: string(RoleTool), + ToolCallID: toolMessageCallID(m), + Content: messageText(m), + }, nil + } + content, err := encodeContent(m.Content) + if err != nil { + return openAIMessage{}, err + } + return openAIMessage{Role: string(m.Role), Content: content, ToolCalls: encodeToolCalls(m.Content)}, nil +} + +// toolMessageCallID resolves the tool_call_id for a RoleTool turn: the +// message-level id, else the id carried on the first tool-result part. +func toolMessageCallID(m Message) string { + if m.ToolCallID != "" { + return m.ToolCallID + } + for _, p := range m.Content { + if p.ToolCallID != "" { + return p.ToolCallID + } + } + return "" +} + +// encodeToolCalls builds the assistant tool_calls array from a turn's tool-call +// parts. Arguments are serialized to JSON because OpenAI expects a string. +func encodeToolCalls(parts []Content) []openAIToolCall { + calls := make([]openAIToolCall, 0, len(parts)) + for _, p := range parts { + if p.Type != ContentTypeToolCall { + continue + } + arguments := "{}" + if len(p.ToolArgs) > 0 { + if b, err := json.Marshal(p.ToolArgs); err == nil { + arguments = string(b) + } + } + calls = append(calls, openAIToolCall{ + ID: p.ToolCallID, + Type: "function", + Function: openAIToolCallFn{Name: p.ToolCallName, Arguments: arguments}, + }) + } + return calls +} + +func encodeContent(parts []Content) (any, error) { + if len(parts) == 0 { + return "", nil + } + arr := make([]any, 0, len(parts)) + for _, p := range parts { + switch p.Type { + case ContentTypeText: + arr = append(arr, map[string]any{"type": "text", "text": p.Text}) + case ContentTypeImage: + arr = append(arr, map[string]any{ + "type": "image_url", + "image_url": map[string]any{"url": p.ImageURL}, + }) + case ContentTypeToolCall, ContentTypeToolResult: + // Tool-call and tool-result parts are encoded at the message level + // (tool_calls array / tool_call_id), never as text-image content. + continue + default: + return nil, fmt.Errorf("model: %w: unsupported content type %q", ErrBadRequest, p.Type) + } + } + if len(arr) == 0 { + return "", nil + } + if len(arr) == 1 { + if blk, ok := arr[0].(map[string]any); ok && blk["type"] == "image_url" { + return arr, nil + } + return arr[0].(map[string]any)["text"], nil + } + return arr, nil +} + +func buildTools(tools []ToolSpec) []openAITool { + if len(tools) == 0 { + return nil + } + out := make([]openAITool, 0, len(tools)) + for _, t := range tools { + out = append(out, openAITool{Type: "function", Function: openAIToolFn{ + Name: t.Name, + Description: t.Description, + Parameters: t.Parameters, + }}) + } + return out +} diff --git a/internal/runtime/model/openai_response.go b/internal/runtime/model/openai_response.go new file mode 100644 index 0000000..fed20e2 --- /dev/null +++ b/internal/runtime/model/openai_response.go @@ -0,0 +1,79 @@ +package model + +import ( + "encoding/json" +) + +// openAIChatResponse is the JSON body returned by /chat/completions, for both +// the non-streaming payload and each SSE stream chunk. +type openAIChatResponse struct { + ID string `json:"id"` + Model string `json:"model"` + Choices []openAIChoice `json:"choices"` + Usage *openAIUsage `json:"usage,omitempty"` +} + +// openAIChoice is one candidate completion. +type openAIChoice struct { + Index int `json:"index"` + Message openAIResponseMessage `json:"message"` + Delta *openAIDelta `json:"delta,omitempty"` + FinishReason string `json:"finish_reason,omitempty"` +} + +// openAIResponseMessage is the assistant message inside a non-streaming choice. +type openAIResponseMessage struct { + Role string `json:"role"` + Content string `json:"content"` + ToolCalls []openAIToolCall `json:"tool_calls,omitempty"` +} + +// openAIDelta is the incremental payload inside a stream chunk. Reasoning content +// (OpenAI o-series / DeepSeek / DashScope) arrives in ReasoningContent; streaming +// function calls arrive in ToolCalls, accumulated per fragment index. +type openAIDelta struct { + Role string `json:"role"` + Content string `json:"content"` + ReasoningContent string `json:"reasoning_content"` + ToolCalls []openAIToolCallDelta `json:"tool_calls"` +} + +// openAIToolCallDelta is one streaming function-call fragment. OpenAI emits a +// sequence of these sharing the same Index: the ID and Function.Name usually +// arrive once (in the first fragment) while Function.Arguments streams across +// fragments. +type openAIToolCallDelta struct { + Index int `json:"index"` + ID string `json:"id"` + Function openAIToolCallFn `json:"function"` +} + +// openAIUsage reports token consumption. +type openAIUsage struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` +} + +func responseFromWire(wire *openAIChatResponse) Response { + out := Response{Model: wire.Model} + if len(wire.Choices) > 0 { + text := wire.Choices[0].Message.Content + out.Content = text + out.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: text}}} + out.StopReason = wire.Choices[0].FinishReason + for _, tc := range wire.Choices[0].Message.ToolCalls { + args := map[string]any{} + if tc.Function.Arguments != "" { + _ = json.Unmarshal([]byte(tc.Function.Arguments), &args) + } + out.ToolCalls = append(out.ToolCalls, ToolCall{Name: tc.Function.Name, ID: tc.ID, Args: args}) + } + } + if u := wire.Usage; u != nil { + out.PromptTokens = u.PromptTokens + out.CompletionTokens = u.CompletionTokens + out.TotalTokens = u.TotalTokens + } + return out +} diff --git a/internal/runtime/model/openai_stream.go b/internal/runtime/model/openai_stream.go new file mode 100644 index 0000000..2f7fe55 --- /dev/null +++ b/internal/runtime/model/openai_stream.go @@ -0,0 +1,174 @@ +package model + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "strings" +) + +// openAIToolAcc accumulates the partial state of one streaming tool call (keyed +// by the provider's tool_calls[].index) so its arguments re-assemble across the +// delta fragments. started guards a single ToolCallStart emission per block. +type openAIToolAcc struct { + id string + name string + args strings.Builder + started bool +} + +// idOrName resolves the block identifier used to correlate start/delta/end: the +// provider tool-call id when present, else the function name. +func (a *openAIToolAcc) idOrName() string { + if a.id != "" { + return a.id + } + return a.name +} + +// scanSSE incrementally decodes stream chunks, accumulates the full body text +// with a strings.Builder (never reading the whole body at once) and captures +// usage from the final chunk. It splits each delta into its segment and brackets +// them: reasoning_content as thinking_start/delta/end, tool_calls fragments as +// tool_call_start/delta/end, and body text as text_start/delta/end. The +// accumulated tool-call arguments re-assemble across fragments into a structured +// Args map, and are attached to the returned Response alongside the text. +func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) { + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, sseBufInit), maxSSELineBytes) + agg := Response{} + var text strings.Builder + textStarted := false + thinkingStarted := false + toolCalls := map[int]*openAIToolAcc{} + var toolOrder []int + + endThinking := func() { + if onEvent != nil && thinkingStarted { + thinkingStarted = false + onEvent(StreamEvent{Kind: StreamKindThinkingEnd}) + } + } + + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || !strings.HasPrefix(line, sseDataPrefix) { + continue + } + data := strings.TrimSpace(strings.TrimPrefix(line, sseDataPrefix)) + if data == sseDone { + break + } + var chunk openAIChatResponse + if err := json.Unmarshal([]byte(data), &chunk); err != nil { + return Response{}, fmt.Errorf("model: %w: decode stream chunk: %v", ErrUpstream, err) + } + if u := chunk.Usage; u != nil { + agg.PromptTokens = u.PromptTokens + agg.CompletionTokens = u.CompletionTokens + agg.TotalTokens = u.TotalTokens + } + if len(chunk.Choices) > 0 && chunk.Choices[0].FinishReason != "" { + agg.StopReason = chunk.Choices[0].FinishReason + } + if len(chunk.Choices) == 0 || chunk.Choices[0].Delta == nil { + continue + } + delta := chunk.Choices[0].Delta + + // Reasoning / thinking segment (OpenAI o-series, DeepSeek, DashScope). + if delta.ReasoningContent != "" { + if onEvent != nil && !thinkingStarted { + thinkingStarted = true + onEvent(StreamEvent{Kind: StreamKindThinkingStart}) + } + if onEvent != nil { + onEvent(StreamEvent{Kind: StreamKindThinkingDelta, Delta: delta.ReasoningContent}) + } + continue + } + + // Streaming tool-call fragments. + if len(delta.ToolCalls) > 0 { + endThinking() + for _, tc := range delta.ToolCalls { + acc := toolCalls[tc.Index] + if acc == nil { + acc = &openAIToolAcc{} + toolCalls[tc.Index] = acc + toolOrder = append(toolOrder, tc.Index) + } + if acc.id == "" { + acc.id = tc.ID + } + if acc.name == "" { + acc.name = tc.Function.Name + } + if onEvent != nil && !acc.started && acc.idOrName() != "" { + acc.started = true + onEvent(StreamEvent{Kind: StreamKindToolCallStart, Index: tc.Index, ID: acc.idOrName()}) + } + if tc.Function.Arguments != "" { + acc.args.WriteString(tc.Function.Arguments) + if onEvent != nil { + onEvent(StreamEvent{Kind: StreamKindToolCallDelta, Index: tc.Index, ID: acc.idOrName(), Delta: tc.Function.Arguments}) + } + } + } + continue + } + + // Body-text segment. + if delta.Content == "" { + continue + } + endThinking() + text.WriteString(delta.Content) + if onEvent != nil { + if !textStarted { + textStarted = true + onEvent(StreamEvent{Kind: StreamKindTextStart}) + } + onEvent(StreamEvent{Kind: StreamKindTextDelta, Delta: delta.Content}) + } + } + if err := scanner.Err(); err != nil { + return Response{}, fmt.Errorf("model: %w: read stream: %v", ErrUpstream, err) + } + + if onEvent != nil { + if thinkingStarted { + onEvent(StreamEvent{Kind: StreamKindThinkingEnd}) + } + if textStarted { + onEvent(StreamEvent{Kind: StreamKindTextEnd}) + } + // OpenAI has no per-block stop event, so each tool_call block is closed + // when the stream terminates. + for _, idx := range toolOrder { + acc := toolCalls[idx] + if acc == nil || !acc.started { + continue + } + onEvent(StreamEvent{Kind: StreamKindToolCallEnd, Index: idx, ID: acc.idOrName()}) + } + } + + for _, idx := range toolOrder { + acc := toolCalls[idx] + if acc == nil { + continue + } + args := map[string]any{} + if acc.args.Len() > 0 { + if err := json.Unmarshal([]byte(acc.args.String()), &args); err != nil { + args = map[string]any{} + } + } + agg.ToolCalls = append(agg.ToolCalls, ToolCall{Name: acc.name, ID: acc.id, Args: args}) + } + agg.Content = text.String() + agg.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: agg.Content}}} + return agg, nil +}