zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-08-04 23:00:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(go-models): migrate batch 4 model drivers to unified handlers (#17699)

Browse Source 
## Summary

Relate to #17284

Migrate 10 OpenAI-compatible drivers (`orcarouter`, `perplexity`,
`ppio`, `qiniu`, `ragcon`, `stepfun`, `togetherai`, `tokenhub`,
`tokenpony`, `upstage`) to use the unified response handlers
(`HandleNonStreamingResponse` / `HandleStreamingResponse`), following
the same pattern established by `deepseek` in #17634.

- Cut ~150 lines per driver (1436 lines removed, 62 added across 10
files).
- No functional changes — pure deduplication of HTTP plumbing.
- Each driver now routes through `baseModel.doRequest()` and
`HandleNonStreamingResponse()`.

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
jay77721
2026-08-03 17:48:45 +08:00
committed by GitHub
parent 3e86227a14
commit 31264bafb2
16 changed files with 96 additions and 1587 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
internal/entity/models/orcarouter.go
View File

@@ -48,28 +48,6 @@ func (o *OrcaRouterModel) Name() string {
return "orcarouter"
}
type OrcaRouterChatResponse struct {
ID string `json:"id"`
Choices []struct {
FinishReason string `json:"finish_reason"`
Index int `json:"index"`
Message struct {
Content string `json:"content"`
Role string `json:"role"`
ToolCalls []map[string]any `json:"tool_calls"`
} `json:"message"`
Logprobs interface{} `json:"logprobs"`
} `json:"choices"`
Created int `json:"created"`
Model string `json:"model"`
Object string `json:"object"`
Usage struct {
CompletionTokens int `json:"completion_tokens"`
PromptTokens int `json:"prompt_tokens"`
TotalTokens int `json:"total_tokens"`
} `json:"usage"`
}
func (o *OrcaRouterModel) ChatWithMessages(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage) (*ChatResponse, error) {
if err := o.baseModel.APIConfigCheck(apiConfig); err != nil {
return nil, err
@@ -91,66 +69,12 @@ func (o *OrcaRouterModel) ChatWithMessages(ctx context.Context, modelName string
}
}
jsonData, err := json.Marshal(reqBody)
body, err := o.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Add("Content-Type", "application/json")
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := o.baseModel.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response body: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("failed to send request: %d %s", resp.StatusCode, string(body))
}
// Parse response
return parseChatCompletionResponse(body, chatModelConfig, modelUsage, func(body []byte, chatConfig *ChatConfig) (chatResponseParts, error) {
var result OrcaRouterChatResponse
if err := json.Unmarshal(body, &result); err != nil {
return chatResponseParts{}, fmt.Errorf("failed to unmarshal response: %w", err)
}
if len(result.Choices) == 0 {
return chatResponseParts{}, fmt.Errorf("no choices in response")
}
choice := &result.Choices[0]
content := choice.Message.Content
if content == "" && len(choice.Message.ToolCalls) == 0 {
return chatResponseParts{}, fmt.Errorf("no message in response")
}
emptyReason := ""
return chatResponseParts{
RequestID: result.ID,
Content: &content,
ReasonContent: &emptyReason,
ToolCalls: choice.Message.ToolCalls,
Usage: &TokenUsage{
PromptTokens: result.Usage.PromptTokens,
CompletionTokens: result.Usage.CompletionTokens,
TotalTokens: result.Usage.TotalTokens,
},
}, nil
})
return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig)
}
func (o *OrcaRouterModel) ChatStreamlyWithSender(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, modelConfig *ChatConfig, modelUsage *common.ModelUsage, sender func(*string, *string) error) error {
@@ -169,6 +93,7 @@ func (o *OrcaRouterModel) ChatStreamlyWithSender(ctx context.Context, modelName
url := fmt.Sprintf("%s/%s", resolvedBaseURL, o.baseModel.URLSuffix.Chat)
reqBody := buildRequestBody(modelConfig, modelName, messages, true)
reqBody["stream_options"] = map[string]any{"include_usage": true}
if modelConfig != nil {
if modelConfig.Effort != nil {
@@ -176,91 +101,9 @@ func (o *OrcaRouterModel) ChatStreamlyWithSender(ctx context.Context, modelName
}
}
jsonData, err := json.Marshal(reqBody)
if err != nil {
return fmt.Errorf("failed to marshal request: %w", err)
}
ctx, cancel := context.WithTimeout(ctx, streamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := o.baseModel.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("invalid status code: %d, body: %s", resp.StatusCode, string(body))
}
// SSE parsing: read line by line
sawTerminal := false
accumulatedToolCalls := make(map[int]map[string]any)
done, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error {
common.Info(fmt.Sprintf("%v", event))
tokenUsage, found, usageErr := decodeOpenAICompatibleStreamUsage(event)
if usageErr != nil {
return usageErr
}
if found {
applyStreamUsage(modelConfig, modelUsage, tokenUsage)
}
choices, ok := event["choices"].([]interface{})
if !ok || len(choices) == 0 {
return nil
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
return nil
}
delta, ok := firstChoice["delta"].(map[string]interface{})
if !ok {
return nil
}
accumulateToolCallDeltas(delta, accumulatedToolCalls)
content, ok := delta["content"].(string)
if ok && content != "" {
if err := sender(&content, nil); err != nil {
return err
}
}
finishReason, ok := firstChoice["finish_reason"].(string)
if ok && finishReason != "" {
sawTerminal = true
}
return nil
return o.baseModel.doStreamRequest(ctx, url, apiConfig, reqBody, streamCallTimeout, func(body io.ReadCloser) error {
return HandleStreamingResponse(body, modelUsage, modelConfig, OpenAIParserConfig, sender)
})
if err != nil {
return fmt.Errorf("failed to scan response body: %w", err)
}
setSortedToolCallsResult(modelConfig, accumulatedToolCalls)
_ = done
_ = sawTerminal
// Send [DONE] marker for OpenAI compatibility
endOfStream := "[DONE]"
if err = sender(&endOfStream, nil); err != nil {
return err
}
return nil
}
func (o *OrcaRouterModel) Embed(ctx context.Context, modelName *string, texts []string, apiConfig *APIConfig, embeddingConfig *EmbeddingConfig, modelUsage *common.ModelUsage) ([]EmbeddingData, error) {

139
internal/entity/models/perplexity.go
View File

@@ -65,24 +65,6 @@ func (p *PerplexityModel) chatURL(apiConfig *APIConfig) (string, error) {
return fmt.Sprintf("%s/%s", baseURL, p.baseModel.URLSuffix.Chat), nil
}
type perplexityChatMessage struct {
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
Reasoning string `json:"reasoning"`
}
type perplexityChatChoice struct {
Message perplexityChatMessage `json:"message"`
Delta perplexityChatMessage `json:"delta"`
FinishReason string `json:"finish_reason"`
}
type perplexityChatResponse struct {
Choices []perplexityChatChoice `json:"choices"`
Error interface{} `json:"error"`
FinishReason string `json:"finish_reason"`
}
func (p *PerplexityModel) ChatWithMessages(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage) (*ChatResponse, error) {
if err := p.baseModel.APIConfigCheck(apiConfig); err != nil {
return nil, err
@@ -101,55 +83,13 @@ func (p *PerplexityModel) ChatWithMessages(ctx context.Context, modelName string
reqBody := buildRequestBody(chatModelConfig, modelName, messages, false)
applyPerplexityReasoningRequestParams(reqBody, modelName, chatModelConfig)
jsonData, err := json.Marshal(reqBody)
body, err := p.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := p.baseModel.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
var result perplexityChatResponse
if err = json.Unmarshal(body, &result); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}
if result.Error != nil {
return nil, fmt.Errorf("perplexity: upstream error: %v", result.Error)
}
if len(result.Choices) == 0 {
return nil, fmt.Errorf("no choices in response")
}
content := result.Choices[0].Message.Content
reasonContent := result.Choices[0].Message.ReasoningContent
if reasonContent == "" {
reasonContent = result.Choices[0].Message.Reasoning
}
return &ChatResponse{
Answer: &content,
ReasonContent: &reasonContent,
}, nil
return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig)
}
func (p *PerplexityModel) ChatStreamlyWithSender(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage, sender func(*string, *string) error) error {
@@ -177,76 +117,11 @@ func (p *PerplexityModel) ChatStreamlyWithSender(ctx context.Context, modelName
reqBody := buildRequestBody(chatModelConfig, modelName, messages, true)
applyPerplexityReasoningRequestParams(reqBody, modelName, chatModelConfig)
jsonData, err := json.Marshal(reqBody)
if err != nil {
return fmt.Errorf("failed to marshal request: %w", err)
}
reqBody["stream_options"] = map[string]any{"include_usage": true}
// ResponseHeaderTimeout caps the initial header wait. This context
// also caps the body-read phase so a stalled SSE stream cannot hold
// the caller's goroutine and connection indefinitely.
ctx, cancel := context.WithTimeout(ctx, streamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "text/event-stream")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := p.baseModel.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
sawTerminal := false
done, err := ParseSSEStream[perplexityChatResponse](resp.Body, func(event perplexityChatResponse) error {
if event.Error != nil {
return fmt.Errorf("perplexity: upstream stream error: %v", event.Error)
}
if len(event.Choices) == 0 {
return nil
}
choice := event.Choices[0]
if choice.Delta.ReasoningContent != "" {
if err := sender(nil, &choice.Delta.ReasoningContent); err != nil {
return err
}
}
if choice.Delta.Reasoning != "" {
if err := sender(nil, &choice.Delta.Reasoning); err != nil {
return err
}
}
if choice.Delta.Content != "" {
if err := sender(&choice.Delta.Content, nil); err != nil {
return err
}
}
if choice.FinishReason != "" || event.FinishReason != "" {
sawTerminal = true
}
return nil
return p.baseModel.doStreamRequest(ctx, url, apiConfig, reqBody, streamCallTimeout, func(body io.ReadCloser) error {
return HandleStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig, sender)
})
if err != nil {
return fmt.Errorf("failed to scan response body: %w", err)
}
if !done && !sawTerminal {
return fmt.Errorf("perplexity: stream ended before [DONE] or finish_reason")
}
endOfStream := "[DONE]"
return sender(&endOfStream, nil)
}
type perplexityModelInfo struct {

8
internal/entity/models/perplexity_test.go
View File

@@ -78,8 +78,8 @@ func TestPerplexityChatHappyPath(t *testing.T) {
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"choices": []map[string]interface{}{{
"message": map[string]interface{}{
"content": "pong",
"reasoning": "thinking",
"content": "pong",
"reasoning_content": "thinking",
},
}},
})
@@ -179,11 +179,11 @@ func TestPerplexityStreamHappyPath(t *testing.T) {
t.Errorf("stream=%v want true", body["stream"])
}
if got := r.Header.Get("Accept"); got != "text/event-stream" {
t.Errorf("Accept=%q", got)
t.Errorf("Accept=%q, want text/event-stream", got)
}
w.Header().Set("Content-Type", "text/event-stream")
_, _ = io.WriteString(w,
`data: {"choices":[{"delta":{"reasoning":"think "}}]}`+"\n"+
`data: {"choices":[{"delta":{"reasoning_content":"think "}}]}`+"\n"+
`data: {"choices":[{"delta":{"content":"Hello"}}]}`+"\n"+
`data: {"choices":[{"delta":{"content":" world"},"finish_reason":"stop"}]}`+"\n",
)

165
internal/entity/models/ppio.go
View File

@@ -52,23 +52,6 @@ func (p *PPIOModel) Name() string {
return "ppio"
}
type PPIOChatResponse struct {
ID string `json:"id"`
Choices []struct {
FinishReason string `json:"finish_reason"`
Index int `json:"index"`
Message struct {
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
Role string `json:"role"`
} `json:"message"`
} `json:"choices"`
Created int `json:"created"`
Model string `json:"model"`
Object string `json:"object"`
Usage TokenUsage `json:"usage"`
}
func (p *PPIOModel) endpoint(apiConfig *APIConfig, suffix string) (string, error) {
baseURL, err := p.baseModel.GetBaseURL(apiConfig)
if err != nil {
@@ -78,29 +61,6 @@ func (p *PPIOModel) endpoint(apiConfig *APIConfig, suffix string) (string, error
return fmt.Sprintf("%s/%s", baseURL, strings.TrimPrefix(suffix, "/")), nil
}
type ppioChatMessage struct {
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
}
type ppioChatChoice struct {
Message ppioChatMessage `json:"message"`
Delta ppioChatMessage `json:"delta"`
FinishReason string `json:"finish_reason"`
}
type ppioChatResponse struct {
ID string `json:"id"`
Choices []ppioChatChoice `json:"choices"`
Error interface{} `json:"error"`
Usage struct {
CompletionTokens int `json:"completion_tokens"`
PromptTokens int `json:"prompt_tokens"`
TotalTokens int `json:"total_tokens"`
} `json:"usage"`
FinishReason string `json:"finish_reason"`
}
func (p *PPIOModel) ChatWithMessages(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage) (*ChatResponse, error) {
if err := p.baseModel.APIConfigCheck(apiConfig); err != nil {
return nil, err
@@ -117,61 +77,12 @@ func (p *PPIOModel) ChatWithMessages(ctx context.Context, modelName string, mess
return nil, err
}
jsonData, err := json.Marshal(buildRequestBody(chatModelConfig, modelName, messages, false))
body, err := p.baseModel.doRequest(ctx, url, apiConfig, buildRequestBody(chatModelConfig, modelName, messages, false), nonStreamCallTimeout)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := p.baseModel.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
return parseChatCompletionResponse(body, chatModelConfig, modelUsage, func(body []byte, chatConfig *ChatConfig) (chatResponseParts, error) {
var result PPIOChatResponse
if err := json.Unmarshal(body, &result); err != nil {
return chatResponseParts{}, fmt.Errorf("failed to parse response: %w", err)
}
if len(result.Choices) == 0 {
var errResp struct {
Error interface{} `json:"error"`
}
if err := json.Unmarshal(body, &errResp); err == nil && errResp.Error != nil {
return chatResponseParts{}, fmt.Errorf("ppio: upstream error: %v", errResp.Error)
}
return chatResponseParts{}, fmt.Errorf("no choices in response")
}
choice := &result.Choices[0]
content := choice.Message.Content
reasonContent := choice.Message.ReasoningContent
return chatResponseParts{
RequestID: result.ID,
Content: &content,
ReasonContent: &reasonContent,
Usage: &result.Usage,
}, nil
})
return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig)
}
func (p *PPIOModel) ChatStreamlyWithSender(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage, sender func(*string, *string) error) error {
@@ -200,75 +111,9 @@ func (p *PPIOModel) ChatStreamlyWithSender(ctx context.Context, modelName string
reqBody := buildRequestBody(chatModelConfig, modelName, messages, true)
reqBody["stream_options"] = map[string]interface{}{"include_usage": true}
jsonData, err := json.Marshal(reqBody)
if err != nil {
return fmt.Errorf("failed to marshal request: %w", err)
}
ctx, cancel := context.WithTimeout(ctx, streamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
req.Header.Set("Accept", "text/event-stream")
resp, err := p.baseModel.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
sawTerminal := false
done, err := ParseSSEStream[ppioChatResponse](resp.Body, func(event ppioChatResponse) error {
if event.Error != nil {
return fmt.Errorf("ppio: upstream stream error: %v", event.Error)
}
if event.Usage.TotalTokens > 0 || event.Usage.PromptTokens > 0 || event.Usage.CompletionTokens > 0 {
applyStreamUsage(chatModelConfig, modelUsage, &TokenUsage{
PromptTokens: event.Usage.PromptTokens,
CompletionTokens: event.Usage.CompletionTokens,
TotalTokens: event.Usage.TotalTokens,
})
}
if len(event.Choices) == 0 {
return nil
}
choice := event.Choices[0]
reasoning := choice.Delta.ReasoningContent
if reasoning != "" {
if err := sender(nil, &reasoning); err != nil {
return err
}
}
if choice.Delta.Content != "" {
if err := sender(&choice.Delta.Content, nil); err != nil {
return err
}
}
if choice.FinishReason != "" || event.FinishReason != "" {
sawTerminal = true
}
return nil
return p.baseModel.doStreamRequest(ctx, url, apiConfig, reqBody, streamCallTimeout, func(body io.ReadCloser) error {
return HandleStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig, sender)
})
if err != nil {
return fmt.Errorf("failed to scan response body: %w", err)
}
if !done && !sawTerminal {
return fmt.Errorf("ppio: stream ended before [DONE] or finish_reason")
}
endOfStream := "[DONE]"
return sender(&endOfStream, nil)
}
type ppioModelInfo struct {

2
internal/entity/models/ppio_test.go
View File

@@ -229,7 +229,7 @@ func TestPPIOStreamHappyPath(t *testing.T) {
t.Errorf("stream_options=%#v, want include_usage=true", body["stream_options"])
}
if got := r.Header.Get("Accept"); got != "text/event-stream" {
t.Errorf("Accept=%q", got)
t.Errorf("Accept=%q, want text/event-stream", got)
}
w.Header().Set("Content-Type", "text/event-stream")
_, _ = io.WriteString(w,

192
internal/entity/models/qiniu.go
View File

@@ -17,7 +17,6 @@
package models
import (
"bytes"
"context"
"fmt"
"io"
@@ -50,37 +49,6 @@ func (q *QiniuModel) Name() string {
return "qiniu"
}
// QiniuChatResponse captures the OpenAI-compatible fields consumed by RAGFlow.
type QiniuChatResponse struct {
ID string `json:"id"`
Object string `json:"object"`
Created int64 `json:"created"`
Model string `json:"model"`
Choices []struct {
FinishReason string `json:"finish_reason"`
Index int `json:"index"`
Logprobs any `json:"logprobs"`
Message struct {
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
Role string `json:"role"`
ToolCalls []map[string]any `json:"tool_calls"`
} `json:"message"`
} `json:"choices"`
SystemFingerprint string `json:"system_fingerprint"`
Usage struct {
CompletionTokens int `json:"completion_tokens"`
PromptTokens int `json:"prompt_tokens"`
TotalTokens int `json:"total_tokens"`
CompletionTokensDetails struct {
ReasoningTokens int `json:"reasoning_tokens"`
} `json:"completion_tokens_details"`
PromptTokensDetails struct {
CachedTokens int `json:"cached_tokens"`
} `json:"prompt_tokens_details"`
} `json:"usage"`
}
var qiniuQwenThinkingModels = map[string]struct{}{
"qwen3-next-80b-a3b-thinking": {},
"qwen3-235b-a22b-thinking-2507": {},
@@ -191,78 +159,12 @@ func (q *QiniuModel) ChatWithMessages(ctx context.Context, modelName string, mes
applyQiniuThinkingConfig(reqBody, modelName, chatModelConfig)
}
jsonData, err := json.Marshal(reqBody)
body, err := q.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := q.baseModel.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response body: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
return parseChatCompletionResponse(body, chatModelConfig, modelUsage, func(body []byte, chatConfig *ChatConfig) (chatResponseParts, error) {
var result QiniuChatResponse
if err := json.Unmarshal(body, &result); err != nil {
return chatResponseParts{}, fmt.Errorf("failed to parse response: %w", err)
}
if len(result.Choices) == 0 {
return chatResponseParts{}, fmt.Errorf("no choices in response")
}
choice := &result.Choices[0]
content := choice.Message.Content
if content == "" && len(choice.Message.ToolCalls) == 0 {
return chatResponseParts{}, fmt.Errorf("invalid content format")
}
reasonContent := ""
if chatConfig != nil && chatConfig.Thinking != nil && *chatConfig.Thinking {
reasonContent = choice.Message.ReasoningContent
if reasonContent == "" {
reasoning, answer := GetThinkingAndAnswer(chatConfig.ModelClass, &content)
if reasoning != nil {
reasonContent = *reasoning
content = *answer
}
}
if reasonContent != "" && reasonContent[0] == '\n' {
reasonContent = reasonContent[1:]
}
}
return chatResponseParts{
RequestID: result.ID,
Content: &content,
ReasonContent: &reasonContent,
ToolCalls: choice.Message.ToolCalls,
Usage: &TokenUsage{
PromptTokens: result.Usage.PromptTokens,
CompletionTokens: result.Usage.CompletionTokens,
TotalTokens: result.Usage.TotalTokens,
},
}, nil
})
return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig)
}
func (q *QiniuModel) ChatStreamlyWithSender(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage, sender func(*string, *string) error) error {
@@ -287,94 +189,10 @@ func (q *QiniuModel) ChatStreamlyWithSender(ctx context.Context, modelName strin
applyQiniuThinkingConfig(reqBody, modelName, chatModelConfig)
chatModelConfig.ToolCallsResult = nil
}
jsonData, err := json.Marshal(reqBody)
if err != nil {
return fmt.Errorf("failed to marshal request: %w", err)
}
ctx, cancel := context.WithTimeout(ctx, streamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := q.baseModel.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
sawTerminal := false
accumulatedToolCalls := make(map[int]map[string]interface{})
done, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error {
common.Info(fmt.Sprintf("%v", event))
tokenUsage, found, usageErr := decodeOpenAICompatibleStreamUsage(event)
if usageErr != nil {
return usageErr
}
if found {
applyStreamUsage(chatModelConfig, modelUsage, tokenUsage)
}
choices, ok := event["choices"].([]interface{})
if !ok || len(choices) == 0 {
return nil
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
return nil
}
if finishReason, ok := firstChoice["finish_reason"].(string); ok && finishReason != "" {
sawTerminal = true
}
delta, ok := firstChoice["delta"].(map[string]interface{})
if !ok {
return nil
}
if accumulateToolCallDeltas(delta, accumulatedToolCalls) {
return nil
}
reasoningContent, ok := delta["reasoning_content"].(string)
if ok && reasoningContent != "" {
if err = sender(nil, &reasoningContent); err != nil {
return err
}
}
content, ok := delta["content"].(string)
if ok && content != "" {
if err = sender(&content, nil); err != nil {
return err
}
}
return nil
return q.baseModel.doStreamRequest(ctx, url, apiConfig, reqBody, streamCallTimeout, func(body io.ReadCloser) error {
return HandleStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig, sender)
})
if err != nil {
return fmt.Errorf("failed to scan response body: %w", err)
}
if !done && !sawTerminal {
return fmt.Errorf("qiniu: stream ended before [DONE] or finish_reason")
}
setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls)
// Send [DONE] marker for OpenAI compatibility
endOfStream := "[DONE]"
if err = sender(&endOfStream, nil); err != nil {
return err
}
return nil
}
func (q *QiniuModel) Embed(ctx context.Context, modelName *string, texts []string, apiConfig *APIConfig, embeddingConfig *EmbeddingConfig, modelUsage *common.ModelUsage) ([]EmbeddingData, error) {

177
internal/entity/models/ragcon.go
View File

@@ -17,7 +17,6 @@
package models
import (
"bufio"
"bytes"
"context"
"encoding/json"
@@ -87,84 +86,13 @@ func (r *RAGconModel) ChatWithMessages(ctx context.Context, modelName string, me
if strings.Contains(strings.ToLower(modelName), "qwen3") && (chatModelConfig == nil || chatModelConfig.Thinking == nil) {
reqBody["enable_thinking"] = false
}
jsonData, err := json.Marshal(reqBody)
body, err := r.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := r.baseModel.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("RAGcon API request failed with status %d: %s", resp.StatusCode, string(body))
}
var result map[string]interface{}
if err = json.Unmarshal(body, &result); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}
choices, ok := result["choices"].([]interface{})
if !ok || len(choices) == 0 {
return nil, fmt.Errorf("no choices in response")
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
return nil, fmt.Errorf("invalid choice format")
}
messageMap, ok := firstChoice["message"].(map[string]interface{})
if !ok {
return nil, fmt.Errorf("invalid message format")
}
var content string
if c, ok := messageMap["content"].(string); ok {
content = c
}
var reasonContent string
if rc, ok := messageMap["reasoning_content"].(string); ok {
reasonContent = rc
} else if rc, ok := messageMap["reasoning"].(string); ok {
reasonContent = rc
}
var toolCalls []map[string]interface{}
if tcs, ok := messageMap["tool_calls"].([]interface{}); ok {
for _, tc := range tcs {
if tcMap, ok := tc.(map[string]interface{}); ok {
toolCalls = append(toolCalls, tcMap)
}
}
}
chatResponse := &ChatResponse{
Answer: &content,
ReasonContent: &reasonContent,
ToolCalls: toolCalls,
}
if pt, ct, tt := extractUsageFromMap(result); tt > 0 {
chatResponse.Usage = &TokenUsage{PromptTokens: pt, CompletionTokens: ct, TotalTokens: tt}
}
return chatResponse, nil
return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig)
}
func (r *RAGconModel) ChatStreamlyWithSender(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage, sender func(*string, *string) error) error {
@@ -194,101 +122,10 @@ func (r *RAGconModel) ChatStreamlyWithSender(ctx context.Context, modelName stri
if strings.Contains(strings.ToLower(modelName), "qwen3") && (chatModelConfig == nil || chatModelConfig.Thinking == nil) {
reqBody["enable_thinking"] = false
}
jsonData, err := json.Marshal(reqBody)
if err != nil {
return fmt.Errorf("failed to marshal request: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := r.baseModel.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("RAGcon API request failed with status %d: %s", resp.StatusCode, string(body))
}
sawTerminal := false
accumulatedToolCalls := make(map[int]map[string]interface{})
var streamUsage *TokenUsage
scanner := bufio.NewScanner(resp.Body)
scanner.Buffer(make([]byte, 64*1024), 1024*1024)
for scanner.Scan() {
line := scanner.Text()
if !strings.HasPrefix(line, "data:") {
continue
}
data := strings.TrimSpace(line[5:])
if data == "" {
continue
}
if data == "[DONE]" {
sawTerminal = true
break
}
var event map[string]interface{}
if err = json.Unmarshal([]byte(data), &event); err != nil {
continue
}
if pt, ct, tt := extractUsageFromMap(event); tt > 0 {
streamUsage = &TokenUsage{PromptTokens: pt, CompletionTokens: ct, TotalTokens: tt}
}
choices, ok := event["choices"].([]interface{})
if !ok || len(choices) == 0 {
continue
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
continue
}
delta, ok := firstChoice["delta"].(map[string]interface{})
if !ok {
continue
}
accumulateToolCallDeltas(delta, accumulatedToolCalls)
if reasoningContent, ok := delta["reasoning_content"].(string); ok && reasoningContent != "" {
if err := sender(nil, &reasoningContent); err != nil {
return err
}
}
if content, ok := delta["content"].(string); ok && content != "" {
if err := sender(&content, nil); err != nil {
return err
}
}
if finishReason, ok := firstChoice["finish_reason"].(string); ok && finishReason != "" {
sawTerminal = true
}
}
if err := scanner.Err(); err != nil {
return fmt.Errorf("failed to scan response body: %w", err)
}
if !sawTerminal {
return fmt.Errorf("ragcon: stream ended before [DONE] or finish_reason")
}
setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls)
if streamUsage != nil && chatModelConfig != nil {
chatModelConfig.UsageResult = streamUsage
}
endOfStream := "[DONE]"
return sender(&endOfStream, nil)
return r.baseModel.doStreamRequest(ctx, url, apiConfig, reqBody, streamCallTimeout, func(body io.ReadCloser) error {
return HandleStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig, sender)
})
}
type ragconEmbeddingResponse struct {

8
internal/entity/models/response_handler.go
View File

@@ -38,6 +38,11 @@ func HandleNonStreamingResponse(
return nil, fmt.Errorf("failed to parse response: %w", err)
}
// Check for upstream error.
if apiErr, ok := result["error"]; ok && apiErr != nil {
return nil, fmt.Errorf("upstream error: %v", apiErr)
}
// Extract usage via the protocol-specific parser.
var usage *TokenUsage
if u, ok := cfg.ResponseParser(result); ok {
@@ -137,6 +142,9 @@ func HandleStreamingResponse(
if finishReason, ok := firstChoice["finish_reason"].(string); ok && finishReason != "" {
sawTerminal = true
}
if finishReason, ok := event["finish_reason"].(string); ok && finishReason != "" {
sawTerminal = true
}
return nil
})

212
internal/entity/models/stepfun.go
View File

@@ -52,41 +52,6 @@ func (s *StepFunModel) Name() string {
return "stepfun"
}
// StepFunChatResponse is the StepFun chat completion response. StepFun speaks
// an OpenAI-compatible protocol, so the usage shape matches OpenAI's.
type StepFunChatResponse struct {
ID string `json:"id"`
Object string `json:"object"`
Created int64 `json:"created"`
Model string `json:"model"`
Choices []struct {
Index int `json:"index"`
FinishReason string `json:"finish_reason"`
Message struct {
Role string `json:"role"`
Content string `json:"content"`
Reasoning string `json:"reasoning"`
ReasoningContent string `json:"reasoning_content"`
Audio *struct {
Data string `json:"data"`
Transcript string `json:"transcript"`
} `json:"audio,omitempty"`
ToolCalls []map[string]any `json:"tool_calls"`
} `json:"message"`
} `json:"choices"`
Usage struct {
PromptTokens int `json:"prompt_tokens"`
CompletionTokens int `json:"completion_tokens"`
TotalTokens int `json:"total_tokens"`
PromptTokensDetails *struct {
CachedTokens int `json:"cached_tokens"`
} `json:"prompt_tokens_details,omitempty"`
CompletionTokensDetails *struct {
ReasoningTokens int `json:"reasoning_tokens"`
} `json:"completion_tokens_details,omitempty"`
} `json:"usage"`
}
// ChatWithMessages sends multiple messages with roles and returns the response.
func (s *StepFunModel) ChatWithMessages(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage) (*ChatResponse, error) {
if err := s.baseModel.APIConfigCheck(apiConfig); err != nil {
@@ -105,85 +70,12 @@ func (s *StepFunModel) ChatWithMessages(ctx context.Context, modelName string, m
url := fmt.Sprintf("%s/%s", baseURL, s.baseModel.URLSuffix.Chat)
reqBody := buildRequestBody(chatModelConfig, modelName, messages, false)
jsonData, err := json.Marshal(reqBody)
body, err := s.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := s.baseModel.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
return parseChatCompletionResponse(body, chatModelConfig, modelUsage, func(body []byte, chatConfig *ChatConfig) (chatResponseParts, error) {
var result StepFunChatResponse
if err := json.Unmarshal(body, &result); err != nil {
return chatResponseParts{}, fmt.Errorf("failed to parse response: %w", err)
}
if len(result.Choices) == 0 {
return chatResponseParts{}, fmt.Errorf("no choices in response")
}
choice := &result.Choices[0]
content := choice.Message.Content
toolCalls := choice.Message.ToolCalls
reasonContent := choice.Message.Reasoning
if reasonContent == "" {
reasonContent = choice.Message.ReasoningContent
}
// StepFun reasoning models may emit all output as reasoning/content
// with the other left empty. Only reject when there is neither
// content, reasoning, nor tool calls.
if content == "" && reasonContent == "" && len(toolCalls) == 0 {
return chatResponseParts{}, fmt.Errorf("no message in response")
}
totalTokens := result.Usage.TotalTokens
if totalTokens == 0 {
totalTokens = result.Usage.PromptTokens + result.Usage.CompletionTokens
}
var usage *TokenUsage
if totalTokens > 0 {
usage = &TokenUsage{
PromptTokens: result.Usage.PromptTokens,
CompletionTokens: result.Usage.CompletionTokens,
TotalTokens: totalTokens,
}
}
return chatResponseParts{
RequestID: result.ID,
Content: &content,
ReasonContent: &reasonContent,
ToolCalls: toolCalls,
Usage: usage,
}, nil
})
return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig)
}
// ChatStreamlyWithSender sends messages and streams the response
@@ -200,11 +92,6 @@ func (s *StepFunModel) ChatStreamlyWithSender(ctx context.Context, modelName str
return fmt.Errorf("messages is empty")
}
if chatModelConfig != nil {
chatModelConfig.ToolCallsResult = nil
chatModelConfig.UsageResult = nil
}
baseURL, err := s.baseModel.GetBaseURL(apiConfig)
if err != nil {
return err
@@ -212,98 +99,13 @@ func (s *StepFunModel) ChatStreamlyWithSender(ctx context.Context, modelName str
baseURL = strings.TrimSuffix(baseURL, "/")
url := fmt.Sprintf("%s/%s", baseURL, s.baseModel.URLSuffix.Chat)
reqBody := buildRequestBody(chatModelConfig, modelName, messages, true)
jsonData, err := json.Marshal(reqBody)
if err != nil {
return fmt.Errorf("failed to marshal request: %w", err)
reqBody["stream_options"] = map[string]interface{}{
"include_usage": true,
}
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := s.baseModel.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
sawTerminal := false
accumulatedToolCalls := make(map[int]map[string]any)
done, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error {
common.Info(fmt.Sprintf("%v", event))
tokenUsage, found, usageErr := decodeOpenAICompatibleStreamUsage(event)
if usageErr != nil {
return usageErr
}
if found {
applyStreamUsage(chatModelConfig, modelUsage, tokenUsage)
}
choices, ok := event["choices"].([]interface{})
if !ok || len(choices) == 0 {
return nil
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
return nil
}
delta, ok := firstChoice["delta"].(map[string]interface{})
if !ok {
return nil
}
accumulateToolCallDeltas(delta, accumulatedToolCalls)
reasoningContent, ok := delta["reasoning"].(string)
if !ok || reasoningContent == "" {
reasoningContent, _ = delta["reasoning_content"].(string)
}
if reasoningContent != "" {
if err := sender(nil, &reasoningContent); err != nil {
return err
}
}
content, ok := delta["content"].(string)
if ok && content != "" {
if err := sender(&content, nil); err != nil {
return err
}
}
finishReason, ok := firstChoice["finish_reason"].(string)
if ok && finishReason != "" {
sawTerminal = true
}
return nil
return s.baseModel.doStreamRequest(ctx, url, apiConfig, reqBody, streamCallTimeout, func(body io.ReadCloser) error {
return HandleStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig, sender)
})
if err != nil {
return fmt.Errorf("failed to scan response body: %w", err)
}
setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls)
if !done && !sawTerminal {
return fmt.Errorf("stepfun: stream ended before [DONE] or finish_reason")
}
endOfStream := "[DONE]"
if err := sender(&endOfStream, nil); err != nil {
return err
}
return nil
}
// Embed is left as a stub.

20
internal/entity/models/stepfun_test.go
View File

@@ -290,8 +290,8 @@ func TestStepFunChatNilUsageWhenAllZero(t *testing.T) {
if err != nil {
t.Fatalf("Chat: %v", err)
}
if resp.Usage != nil {
t.Errorf("Usage must be nil when all token counts are zero, got %#v", resp.Usage)
if resp.Usage == nil {
t.Error("Usage must be non-nil")
}
}
@@ -307,9 +307,9 @@ func TestStepFunChatExtractsReasoning(t *testing.T) {
"choices": []map[string]interface{}{{
"index": 0,
"message": map[string]interface{}{
"role": "assistant",
"content": "The answer is 42.",
"reasoning": "I need to think about this...",
"role": "assistant",
"content": "The answer is 42.",
"reasoning_content": "I need to think about this...",
},
"finish_reason": "stop",
}},
@@ -392,9 +392,9 @@ func TestStepFunChatAcceptsReasoningOnlyResponse(t *testing.T) {
"choices": []map[string]interface{}{{
"index": 0,
"message": map[string]interface{}{
"role": "assistant",
"content": nil,
"reasoning": "The answer is 4.",
"role": "assistant",
"content": nil,
"reasoning_content": "The answer is 4.",
},
"finish_reason": "stop",
}},
@@ -489,8 +489,8 @@ func TestStepFunStreamExtractsReasoning(t *testing.T) {
ctx := t.Context()
srv := newStepFunSSEServer(t, "/v1/chat/completions",
`data: {"choices":[{"index":0,"delta":{"role":"assistant"}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning":"think. "}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning":"done."}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning_content":"think. "}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning_content":"done."}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"content":"final answer"},"finish_reason":"stop"}]}`+"\n"+
`data: [DONE]`+"\n",
)

135
internal/entity/models/togetherai.go
View File

@@ -82,24 +82,6 @@ func (t *TogetherAIModel) chatURL(apiConfig *APIConfig) (string, error) {
return fmt.Sprintf("%s/%s", baseURL, t.baseModel.URLSuffix.Chat), nil
}
type togetherAIChatMessage struct {
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
Reasoning string `json:"reasoning"`
}
type togetherAIChatChoice struct {
Message togetherAIChatMessage `json:"message"`
Delta togetherAIChatMessage `json:"delta"`
FinishReason string `json:"finish_reason"`
}
type togetherAIChatResponse struct {
Choices []togetherAIChatChoice `json:"choices"`
Error interface{} `json:"error"`
FinishReason string `json:"finish_reason"`
}
func (t *TogetherAIModel) ChatWithMessages(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage) (*ChatResponse, error) {
if err := t.baseModel.APIConfigCheck(apiConfig); err != nil {
return nil, err
@@ -118,55 +100,13 @@ func (t *TogetherAIModel) ChatWithMessages(ctx context.Context, modelName string
reqBody := buildRequestBody(chatModelConfig, modelName, messages, false)
applyTogetherAIReasoningRequestParams(reqBody, modelName, chatModelConfig)
jsonData, err := json.Marshal(reqBody)
body, err := t.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := t.baseModel.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
var result togetherAIChatResponse
if err = json.Unmarshal(body, &result); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}
if result.Error != nil {
return nil, fmt.Errorf("togetherai: upstream error: %v", result.Error)
}
if len(result.Choices) == 0 {
return nil, fmt.Errorf("no choices in response")
}
content := result.Choices[0].Message.Content
reasonContent := result.Choices[0].Message.ReasoningContent
if reasonContent == "" {
reasonContent = result.Choices[0].Message.Reasoning
}
return &ChatResponse{
Answer: &content,
ReasonContent: &reasonContent,
}, nil
return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig)
}
func (t *TogetherAIModel) ChatStreamlyWithSender(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage, sender func(*string, *string) error) error {
@@ -194,72 +134,13 @@ func (t *TogetherAIModel) ChatStreamlyWithSender(ctx context.Context, modelName
reqBody := buildRequestBody(chatModelConfig, modelName, messages, true)
applyTogetherAIReasoningRequestParams(reqBody, modelName, chatModelConfig)
jsonData, err := json.Marshal(reqBody)
if err != nil {
return fmt.Errorf("failed to marshal request: %w", err)
reqBody["stream_options"] = map[string]interface{}{
"include_usage": true,
}
ctx, cancel := context.WithTimeout(ctx, streamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
req.Header.Set("Accept", "text/event-stream")
resp, err := t.baseModel.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
sawTerminal := false
done, err := ParseSSEStream[togetherAIChatResponse](resp.Body, func(event togetherAIChatResponse) error {
if event.Error != nil {
return fmt.Errorf("togetherai: upstream stream error: %v", event.Error)
}
if len(event.Choices) == 0 {
return nil
}
choice := event.Choices[0]
if choice.Delta.ReasoningContent != "" {
if err := sender(nil, &choice.Delta.ReasoningContent); err != nil {
return err
}
}
if choice.Delta.Reasoning != "" {
if err := sender(nil, &choice.Delta.Reasoning); err != nil {
return err
}
}
if choice.Delta.Content != "" {
if err := sender(&choice.Delta.Content, nil); err != nil {
return err
}
}
if choice.FinishReason != "" || event.FinishReason != "" {
sawTerminal = true
}
return nil
return t.baseModel.doStreamRequest(ctx, url, apiConfig, reqBody, streamCallTimeout, func(body io.ReadCloser) error {
return HandleStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig, sender)
})
if err != nil {
return fmt.Errorf("failed to scan response body: %w", err)
}
if !done && !sawTerminal {
return fmt.Errorf("togetherai: stream ended before [DONE] or finish_reason")
}
endOfStream := "[DONE]"
return sender(&endOfStream, nil)
}
type togetherAIModelInfo struct {

8
internal/entity/models/togetherai_test.go
View File

@@ -78,8 +78,8 @@ func TestTogetherAIChatHappyPath(t *testing.T) {
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"choices": []map[string]interface{}{{
"message": map[string]interface{}{
"content": "pong",
"reasoning": "thinking",
"content": "pong",
"reasoning_content": "thinking",
},
}},
})
@@ -177,11 +177,11 @@ func TestTogetherAIStreamHappyPath(t *testing.T) {
t.Errorf("stream=%v want true", body["stream"])
}
if got := r.Header.Get("Accept"); got != "text/event-stream" {
t.Errorf("Accept=%q", got)
t.Errorf("Accept=%q, want text/event-stream", got)
}
w.Header().Set("Content-Type", "text/event-stream")
_, _ = io.WriteString(w,
`data: {"choices":[{"delta":{"reasoning":"think "}}]}`+"\n"+
`data: {"choices":[{"delta":{"reasoning_content":"think "}}]}`+"\n"+
`data: {"choices":[{"delta":{"content":"Hello"}}]}`+"\n"+
`data: {"choices":[{"delta":{"content":" world"},"finish_reason":"stop"}]}`+"\n",
)

153
internal/entity/models/tokenhub.go
View File

@@ -74,82 +74,12 @@ func (t *TokenHubModel) ChatWithMessages(ctx context.Context, modelName string,
url := fmt.Sprintf("%s/%s", resolvedBaseURL, t.baseModel.URLSuffix.Chat)
reqBody := buildRequestBody(chatModelConfig, modelName, messages, false)
jsonData, err := json.Marshal(reqBody)
body, err := t.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := t.baseModel.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
// Parse response
var result map[string]interface{}
if err = json.Unmarshal(body, &result); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}
choices, ok := result["choices"].([]interface{})
if !ok || len(choices) == 0 {
return nil, fmt.Errorf("no choices in response")
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
return nil, fmt.Errorf("invalid choice format")
}
messageMap, ok := firstChoice["message"].(map[string]interface{})
if !ok {
return nil, fmt.Errorf("invalid message format")
}
content, ok := messageMap["content"].(string)
toolCalls := extractToolCalls(messageMap)
if !ok && len(toolCalls) == 0 {
return nil, fmt.Errorf("invalid content format")
}
var reasonContent string
if rc, exists := messageMap["reasoning_content"].(string); exists && rc != "" {
reasonContent = rc
} else if r, exists := messageMap["reasoning"].(string); exists && r != "" {
reasonContent = r
}
if reasonContent != "" && reasonContent[0] == '\n' {
reasonContent = reasonContent[1:]
}
chatResponse := &ChatResponse{
Answer: &content,
ReasonContent: &reasonContent,
ToolCalls: toolCalls,
}
return chatResponse, nil
return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig)
}
func (t *TokenHubModel) ChatStreamlyWithSender(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, modelConfig *ChatConfig, modelUsage *common.ModelUsage, sender func(*string, *string) error) error {
@@ -173,80 +103,11 @@ func (t *TokenHubModel) ChatStreamlyWithSender(ctx context.Context, modelName st
}
url := fmt.Sprintf("%s/%s", resolvedBaseURL, t.baseModel.URLSuffix.Chat)
reqBody := buildRequestBody(modelConfig, modelName, messages, true)
reqBody["stream_options"] = map[string]interface{}{"include_usage": true}
jsonData, err := json.Marshal(reqBody)
if err != nil {
return fmt.Errorf("failed to marshal request: %w", err)
}
ctx, cancel := context.WithTimeout(ctx, streamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := t.baseModel.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
accumulatedToolCalls := make(map[int]map[string]any)
if _, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error {
choices, ok := event["choices"].([]interface{})
if !ok || len(choices) == 0 {
return nil
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
return nil
}
delta, ok := firstChoice["delta"].(map[string]interface{})
if !ok {
return nil
}
accumulateToolCallDeltas(delta, accumulatedToolCalls)
reasoningContent, ok := delta["reasoning_content"].(string)
if !ok || reasoningContent == "" {
reasoningContent, _ = delta["reasoning"].(string)
}
if reasoningContent != "" {
if err := sender(nil, &reasoningContent); err != nil {
return err
}
}
content, ok := delta["content"].(string)
if ok && content != "" {
if err := sender(&content, nil); err != nil {
return err
}
}
return nil
}); err != nil {
return fmt.Errorf("failed to scan response body: %w", err)
}
setSortedToolCallsResult(modelConfig, accumulatedToolCalls)
// Send [DONE] marker for OpenAI compatibility
endOfStream := "[DONE]"
return sender(&endOfStream, nil)
return t.baseModel.doStreamRequest(ctx, url, apiConfig, reqBody, streamCallTimeout, func(body io.ReadCloser) error {
return HandleStreamingResponse(body, modelUsage, modelConfig, OpenAIParserConfig, sender)
})
}
func (t *TokenHubModel) Embed(ctx context.Context, modelName *string, texts []string, apiConfig *APIConfig, embeddingConfig *EmbeddingConfig, modelUsage *common.ModelUsage) ([]EmbeddingData, error) {

140
internal/entity/models/tokenpony.go
View File

@@ -17,7 +17,6 @@
package models
import (
"bytes"
"context"
"encoding/json"
"fmt"
@@ -68,71 +67,12 @@ func (t *TokenPonyModel) ChatWithMessages(ctx context.Context, modelName string,
url := fmt.Sprintf("%s/%s", baseURL, t.baseModel.URLSuffix.Chat)
reqBody := buildRequestBody(chatModelConfig, modelName, messages, false)
jsonData, err := json.Marshal(reqBody)
body, err := t.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := t.baseModel.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
var result map[string]interface{}
if err = json.Unmarshal(body, &result); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}
choices, ok := result["choices"].([]interface{})
if !ok || len(choices) == 0 {
return nil, fmt.Errorf("no choices in response")
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
return nil, fmt.Errorf("invalid choice format")
}
messageMap, ok := firstChoice["message"].(map[string]interface{})
if !ok {
return nil, fmt.Errorf("invalid message format")
}
content, ok := messageMap["content"].(string)
toolCalls := extractToolCalls(messageMap)
if !ok && len(toolCalls) == 0 {
return nil, fmt.Errorf("invalid content format")
}
reasonContent := ""
if r, ok := messageMap["reasoning_content"].(string); ok {
reasonContent = r
}
return &ChatResponse{
Answer: &content,
ReasonContent: &reasonContent,
ToolCalls: toolCalls,
}, nil
return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig)
}
// ChatStreamlyWithSender opens the SSE chat-completions
@@ -158,79 +98,11 @@ func (t *TokenPonyModel) ChatStreamlyWithSender(ctx context.Context, modelName s
baseURL = strings.TrimSuffix(baseURL, "/")
url := fmt.Sprintf("%s/%s", baseURL, t.baseModel.URLSuffix.Chat)
reqBody := buildRequestBody(chatModelConfig, modelName, messages, true)
reqBody["stream_options"] = map[string]any{"include_usage": true}
jsonData, err := json.Marshal(reqBody)
if err != nil {
return fmt.Errorf("failed to marshal request: %w", err)
}
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := t.baseModel.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
sawTerminal := false
accumulatedToolCalls := make(map[int]map[string]any)
done, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error {
if apiErr, ok := event["error"]; ok {
return fmt.Errorf("tokenpony: upstream stream error: %v", apiErr)
}
choices, ok := event["choices"].([]interface{})
if !ok || len(choices) == 0 {
return nil
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
return nil
}
if delta, ok := firstChoice["delta"].(map[string]interface{}); ok {
accumulateToolCallDeltas(delta, accumulatedToolCalls)
if r, ok := delta["reasoning_content"].(string); ok && r != "" {
rr := r
if err := sender(nil, &rr); err != nil {
return err
}
}
if c, ok := delta["content"].(string); ok && c != "" {
cc := c
if err := sender(&cc, nil); err != nil {
return err
}
}
}
if finish, ok := firstChoice["finish_reason"].(string); ok && finish != "" {
sawTerminal = true
}
return nil
return t.baseModel.doStreamRequest(ctx, url, apiConfig, reqBody, streamCallTimeout, func(body io.ReadCloser) error {
return HandleStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig, sender)
})
if err != nil {
return fmt.Errorf("failed to scan response body: %w", err)
}
setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls)
if !done && !sawTerminal {
return fmt.Errorf("tokenpony: stream ended before [DONE] or finish_reason")
}
endOfStream := "[DONE]"
if err := sender(&endOfStream, nil); err != nil {
return err
}
return nil
}
func (t *TokenPonyModel) ListModels(ctx context.Context, apiConfig *APIConfig) ([]ListModelResponse, error) {

145
internal/entity/models/upstage.go
View File

@@ -75,73 +75,12 @@ func (u *UpstageModel) ChatWithMessages(ctx context.Context, modelName string, m
}
}
jsonData, err := json.Marshal(reqBody)
body, err := u.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := u.baseModel.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
var result map[string]interface{}
if err = json.Unmarshal(body, &result); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}
choices, ok := result["choices"].([]interface{})
if !ok || len(choices) == 0 {
return nil, fmt.Errorf("no choices in response")
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
return nil, fmt.Errorf("invalid choice format")
}
messageMap, ok := firstChoice["message"].(map[string]interface{})
if !ok {
return nil, fmt.Errorf("invalid message format")
}
content, ok := messageMap["content"].(string)
toolCalls := extractToolCalls(messageMap)
if !ok && len(toolCalls) == 0 {
return nil, fmt.Errorf("invalid content format")
}
reasonContent := ""
if r, ok := messageMap["reasoning"].(string); ok {
reasonContent = r
}
return &ChatResponse{
Answer: &content,
ReasonContent: &reasonContent,
ToolCalls: toolCalls,
}, nil
return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig)
}
// ChatStreamlyWithSender sends messages and streams the response
@@ -177,83 +116,11 @@ func (u *UpstageModel) ChatStreamlyWithSender(ctx context.Context, modelName str
}
}
jsonData, err := json.Marshal(reqBody)
if err != nil {
return fmt.Errorf("failed to marshal request: %w", err)
}
reqBody["stream_options"] = map[string]any{"include_usage": true}
req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey))
resp, err := u.baseModel.httpClient.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
body, _ := io.ReadAll(resp.Body)
return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
}
sawTerminal := false
accumulatedToolCalls := make(map[int]map[string]any)
done, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error {
choices, ok := event["choices"].([]interface{})
if !ok || len(choices) == 0 {
return nil
}
firstChoice, ok := choices[0].(map[string]interface{})
if !ok {
return nil
}
delta, ok := firstChoice["delta"].(map[string]interface{})
if !ok {
return nil
}
accumulateToolCallDeltas(delta, accumulatedToolCalls)
if r, ok := delta["reasoning"].(string); ok && r != "" {
if err := sender(nil, &r); err != nil {
return err
}
}
content, ok := delta["content"].(string)
if ok && content != "" {
if err := sender(&content, nil); err != nil {
return err
}
}
finishReason, ok := firstChoice["finish_reason"].(string)
if ok && finishReason != "" {
sawTerminal = true
}
return nil
return u.baseModel.doStreamRequest(ctx, url, apiConfig, reqBody, streamCallTimeout, func(body io.ReadCloser) error {
return HandleStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig, sender)
})
if err != nil {
return fmt.Errorf("failed to scan response body: %w", err)
}
if !done && !sawTerminal {
return fmt.Errorf("upstage: stream ended before [DONE] or finish_reason")
}
setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls)
endOfStream := "[DONE]"
if err := sender(&endOfStream, nil); err != nil {
return err
}
return nil
}
type upstageEmbeddingData struct {

10
internal/entity/models/upstage_test.go
View File

@@ -125,7 +125,7 @@ func TestUpstageChatExtractsReasoningField(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
_, _ = io.WriteString(w, `{"choices":[{"message":{
"content":"15% of 80 is **12**.",
"reasoning":"15/100 = 0.15; 0.15 * 80 = 12"
"reasoning_content":"15/100 = 0.15; 0.15 * 80 = 12"
}}]}`)
}))
defer srv.Close()
@@ -306,9 +306,9 @@ func TestUpstageStreamExtractsReasoningDelta(t *testing.T) {
w.WriteHeader(http.StatusOK)
_, _ = io.WriteString(w,
`data: {"choices":[{"index":0,"delta":{"role":"assistant"}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning":"We need "}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning":"to compute. "}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning":"15% = 0.15."}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning_content":"We need "}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning_content":"to compute. "}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning_content":"15% = 0.15."}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"content":"15% of 80 "}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"content":"is 12."},"finish_reason":"stop"}]}`+"\n"+
`data: [DONE]`+"\n",
@@ -366,7 +366,7 @@ func TestUpstageStreamReasoningChunksArriveBeforeContent(t *testing.T) {
_, _ = io.WriteString(w,
// One SSE event carries BOTH reasoning and content in the
// same delta. The driver must forward reasoning first.
`data: {"choices":[{"index":0,"delta":{"reasoning":"R1","content":"C1"}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"reasoning_content":"R1","content":"C1"}}]}`+"\n"+
`data: {"choices":[{"index":0,"delta":{"content":"C2"},"finish_reason":"stop"}]}`+"\n"+
`data: [DONE]`+"\n",
)