refactor(model): 拆 openai.go 652→216行(纯搬移零逻辑改动) — encode/response/stream 三件套对齐 anthropic 布局
- 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, 拆分前既有,零逻辑改动未动), 另开步骤重构
This commit is contained in:
AskaEth committed
1 parent
ca2faa0f95
commit
610b330025
4 files changed
+455
-436
No files matched your search
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in new issue
Block a user