231 lines
8.3 KiB
Go
231 lines
8.3 KiB
Go
package openai
|
|
|
|
import (
|
|
"context"
|
|
"go.uber.org/zap"
|
|
edgeservice "iop/apps/edge/internal/service"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
func shouldSynthesizeTextToolCalls(dispatch edgeservice.RunDispatch, textToolFallback bool) bool {
|
|
return textToolFallback || strings.TrimSpace(dispatch.Adapter) == "cli"
|
|
}
|
|
|
|
func chatSubmitRunRequest(dispatch routeDispatch, req chatCompletionRequest, workspace, prompt string, input map[string]any, metadata map[string]string) edgeservice.SubmitRunRequest {
|
|
return edgeservice.SubmitRunRequest{
|
|
NodeRef: dispatch.NodeRef,
|
|
ModelGroupKey: strings.TrimSpace(req.Model),
|
|
Adapter: dispatch.Adapter,
|
|
Target: dispatch.Target,
|
|
SessionID: dispatch.SessionID,
|
|
Workspace: workspace,
|
|
Prompt: prompt,
|
|
Input: input,
|
|
TimeoutSec: dispatch.TimeoutSec,
|
|
MaxQueue: dispatch.MaxQueue,
|
|
QueueTimeoutMS: dispatch.QueueTimeoutMS,
|
|
Metadata: metadata,
|
|
ProviderPool: dispatch.ProviderPool,
|
|
}
|
|
}
|
|
|
|
func (s *Server) completeChatCompletion(w http.ResponseWriter, dc *chatDispatchContext, handle edgeservice.RunResult) {
|
|
r := dc.r
|
|
req := dc.req
|
|
outputPolicy := dc.outputPolicy
|
|
attempt := 1
|
|
for {
|
|
result, err := collectChatCompletionOutput(r.Context(), req, handle, outputPolicy)
|
|
if err != nil {
|
|
s.cancelRunOnHTTPGiveUp(handle.Dispatch(), err)
|
|
handle.Close()
|
|
emitUsageMetrics(dc.usageLabels(s, responseModeNormalized), usageStatusForError(err), usageObservation{})
|
|
writeError(w, httpStatusForRunError(err), "run_error", err.Error())
|
|
return
|
|
}
|
|
valErr := result.toolValidationErr
|
|
if valErr == nil {
|
|
valErr = validateToolCallResponse(dc.validation, result.toolCalls, result.toolCallOrigin)
|
|
}
|
|
if valErr != nil {
|
|
failedRunID := handle.Dispatch().RunID
|
|
if attempt < maxToolValidationAttempts {
|
|
handle.Close()
|
|
attempt++
|
|
retryReq := dc.submitReq
|
|
retryReq.Metadata = toolValidationAttemptMetadata(dc.submitReq.Metadata, attempt, failedRunID, valErr.Error())
|
|
s.logger.Warn("openai chat completion tool validation retry",
|
|
zap.String("run_id", failedRunID),
|
|
zap.Int("attempt", attempt),
|
|
zap.String("reason", valErr.Error()),
|
|
)
|
|
next, submitErr := dc.retrySubmit(r.Context(), retryReq)
|
|
if submitErr != nil {
|
|
emitUsageMetrics(dc.usageLabels(s, responseModeNormalized), usageStatusError, usageObservation{})
|
|
writeError(w, http.StatusBadGateway, "tool_validation_retry_error", submitErr.Error())
|
|
return
|
|
}
|
|
// Extract RunResult from either a direct RunResult or a
|
|
// ProviderPoolDispatchResult (pool retry). For provider-pool
|
|
// retries, only normalized results with a non-nil Run are
|
|
// accepted; tunnel or nil Run results are rejected to prevent
|
|
// nil handle panic downstream.
|
|
if rr, ok := next.(edgeservice.RunResult); ok {
|
|
handle = rr
|
|
} else if ppd, ok := next.(*edgeservice.ProviderPoolDispatchResult); ok {
|
|
if ppd.Path != edgeservice.ProviderPoolPathNormalized || ppd.Run == nil {
|
|
if ppd.Tunnel != nil {
|
|
ppd.Tunnel.Close()
|
|
}
|
|
writeError(w, http.StatusBadGateway, "tool_validation_retry_error", "provider-pool retry selected tunnel path")
|
|
return
|
|
}
|
|
handle = ppd.Run
|
|
} else {
|
|
handle.Close()
|
|
writeError(w, http.StatusInternalServerError, "run_error", "unexpected retry result type")
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
handle.Close()
|
|
s.logger.Warn("openai chat completion tool validation failed",
|
|
zap.String("run_id", failedRunID),
|
|
zap.Int("attempt", attempt),
|
|
zap.String("reason", valErr.Error()),
|
|
)
|
|
emitUsageMetrics(dc.usageLabels(s, responseModeNormalized), usageStatusError, usageObservation{})
|
|
writeError(w, http.StatusBadGateway, "tool_validation_error", valErr.Error())
|
|
return
|
|
}
|
|
handle.Close()
|
|
emitUsageMetrics(
|
|
dc.usageLabels(s, result.responseMode),
|
|
usageStatusSuccess, usageObservationFromOpenAIUsage(result.usage, result.reasoningLen),
|
|
)
|
|
toolCallNames := extractToolCallNames(result.toolCalls)
|
|
s.logger.Info("openai chat completion output",
|
|
zap.String("run_id", handle.Dispatch().RunID),
|
|
zap.Bool("strict_output", outputPolicy.Strict),
|
|
zap.Bool("strict_stream_buffer", outputPolicy.StreamBuffer),
|
|
zap.String("xml_completion_tool", outputPolicy.XMLCompletionTool),
|
|
zap.Bool("normalized", result.normalized),
|
|
zap.String("response_mode", result.responseMode),
|
|
zap.Int("content_len", result.contentLen),
|
|
zap.Int("reasoning_len", result.reasoningLen),
|
|
zap.String("assembled_content", result.message.Content),
|
|
zap.String("assembled_reasoning", result.message.ReasoningContent),
|
|
zap.Strings("assembled_tool_calls", toolCallNames),
|
|
zap.Int("assembled_tool_call_count", len(toolCallNames)),
|
|
)
|
|
created := time.Now().Unix()
|
|
writeJSON(w, http.StatusOK, chatCompletionResponse{
|
|
ID: "chatcmpl-" + handle.Dispatch().RunID,
|
|
Object: "chat.completion",
|
|
Created: created,
|
|
Model: responseModel(req.Model, handle.Dispatch().Target),
|
|
Choices: []chatCompletionChoice{{
|
|
Index: 0,
|
|
Message: result.message,
|
|
FinishReason: result.finishReason,
|
|
}},
|
|
Usage: result.usage,
|
|
})
|
|
return
|
|
}
|
|
}
|
|
|
|
type chatCompletionOutput struct {
|
|
message chatMessage
|
|
finishReason string
|
|
usage *openAIUsage
|
|
normalized bool
|
|
responseMode string
|
|
contentLen int
|
|
reasoningLen int
|
|
toolCalls []any
|
|
toolCallOrigin string
|
|
toolValidationErr error
|
|
}
|
|
|
|
func collectChatCompletionOutput(ctx context.Context, req chatCompletionRequest, handle edgeservice.RunResult, outputPolicy strictOutputPolicy) (chatCompletionOutput, error) {
|
|
text, reasoning, finishReason, nativeToolCalls, usage, _, err := collectRunResult(ctx, handle.Stream(), handle.WaitTimeout())
|
|
if err != nil {
|
|
return chatCompletionOutput{}, err
|
|
}
|
|
text, reasoning, normalized := normalizeCompletionOutput(outputPolicy, text, reasoning, req.explicitlyIncludesReasoning())
|
|
message := chatMessage{Role: "assistant", Content: text, ReasoningContent: reasoning}
|
|
includeReasoning := req.includeReasoning()
|
|
if !includeReasoning {
|
|
message.ReasoningContent = ""
|
|
}
|
|
var toolCalls []any
|
|
toolCallOrigin := ""
|
|
var toolValidationErr error
|
|
|
|
if len(nativeToolCalls) > 0 {
|
|
if len(req.Tools) > 0 && strings.TrimSpace(message.Content) != "" {
|
|
filter := &streamToolTextFilter{}
|
|
message.Content = filter.Append(message.Content) + filter.FlushNonCandidate()
|
|
}
|
|
nativeToolCalls = normalizeNativeToolCallsForResponse(nativeToolCalls, req.Tools)
|
|
message.ToolCalls = nativeToolCalls
|
|
toolCalls = nativeToolCalls
|
|
toolCallOrigin = toolCallOriginNative
|
|
finishReason = "tool_calls"
|
|
} else if len(req.Tools) > 0 {
|
|
res := synthesizeToolCallsFromTextResult(text, req.Tools, handle.Dispatch().RunID)
|
|
if res.validationErr != nil {
|
|
toolValidationErr = res.validationErr
|
|
} else if len(res.toolCalls) > 0 {
|
|
message.Content = res.cleaned
|
|
message.ReasoningContent = ""
|
|
message.ToolCalls = res.toolCalls
|
|
toolCalls = res.toolCalls
|
|
toolCallOrigin = toolCallOriginText
|
|
finishReason = "tool_calls"
|
|
}
|
|
}
|
|
if len(message.ToolCalls) == 0 && strings.TrimSpace(message.Content) == "" && strings.TrimSpace(reasoning) != "" {
|
|
if includeReasoning {
|
|
message.Content = reasoningOnlyFallbackContent(reasoning, finishReason)
|
|
} else {
|
|
message.Content = hiddenReasoningFallbackContent(finishReason)
|
|
}
|
|
}
|
|
|
|
return chatCompletionOutput{
|
|
message: message,
|
|
finishReason: finishReason,
|
|
usage: usage,
|
|
normalized: normalized,
|
|
responseMode: responseModeNormalized,
|
|
contentLen: len(message.Content),
|
|
reasoningLen: len(reasoning),
|
|
toolCalls: toolCalls,
|
|
toolCallOrigin: toolCallOrigin,
|
|
toolValidationErr: toolValidationErr,
|
|
}, nil
|
|
}
|
|
|
|
func reasoningOnlyFallbackContent(reasoning, finishReason string) string {
|
|
content := strings.TrimSpace(reasoning)
|
|
if content == "" {
|
|
return ""
|
|
}
|
|
reason := strings.TrimSpace(finishReason)
|
|
if reason == "" || reason == "stop" {
|
|
return content
|
|
}
|
|
return content + "\n\n[IOP notice: model finished before producing final assistant content; finish_reason=" + reason + "]"
|
|
}
|
|
|
|
func hiddenReasoningFallbackContent(finishReason string) string {
|
|
reason := strings.TrimSpace(finishReason)
|
|
if reason == "" {
|
|
reason = "unknown"
|
|
}
|
|
return "[IOP notice: model produced reasoning but no final assistant content; reasoning was hidden by include_reasoning=false; finish_reason=" + reason + "]"
|
|
}
|