Skip to content

Commit 4c2c661

Browse files
chenymechenyme
committed
feat: implement OpenAI responses background mode support and related error handling
1 parent 9c36769 commit 4c2c661

8 files changed

Lines changed: 368 additions & 0 deletions

File tree

‎backend/internal/application/conversation/service_message_send.go‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -215,6 +215,8 @@ func (s *Service) sendMessageInternal(
215215
var resolvedRoute *channel.ResolvedRoute
216216
var filteredOptions map[string]interface{}
217217
var totalServerSideToolUsage map[string]int64
218+
var responsesBackgroundRouteConfig llm.RouteConfig
219+
var responsesBackgroundRecovery openAIResponsesBackgroundRecoveryState
218220
userContentEstimatedInputTokens := int64(0)
219221
usageAccumulator := &messageUsageAccumulator{}
220222
upstreamCallStarted := false
@@ -224,6 +226,13 @@ func (s *Service) sendMessageInternal(
224226
runState.bind(&userMessage, &assistantMessage, &traceRecorder, &result, ctx)
225227
defer func() {
226228
if retErr != nil {
229+
if errors.Is(retErr, ErrMessageGenerationCanceled) {
230+
if usage, ok := s.recoverOpenAIResponsesBackgroundUsage(responsesBackgroundRouteConfig, responsesBackgroundRecovery); ok {
231+
if delta := diffLLMUsage(usage, responsesBackgroundRecovery.ObservedUsage); delta != (llm.Usage{}) {
232+
usageAccumulator.addObservedUsage(delta)
233+
}
234+
}
235+
}
227236
if retained := s.persistInterruptedMessageGeneration(ctx, persistInterruptedMessageGenerationInput{
228237
SendInput: input,
229238
UserMessage: userMessage,
@@ -728,6 +737,7 @@ func (s *Service) sendMessageInternal(
728737
AttributionReferer: attributionReferer,
729738
AttributionTitle: attributionTitle,
730739
}
740+
responsesBackgroundRouteConfig = routeConfig
731741
filteredOptions = filterModelOptions(input.Options, route.Protocol, modelOptionPolicyConfig{
732742
Mode: cfg.ModelOptionPolicyMode,
733743
AllowedPathsJSON: cfg.ModelOptionAllowedPaths,
@@ -741,6 +751,10 @@ func (s *Service) sendMessageInternal(
741751
Tools: toolRuntime.definitions,
742752
Options: filteredOptions,
743753
}
754+
if supportsOpenAIResponsesBackgroundMode(route) {
755+
generateInput.ResponsesBackground = true
756+
sendSpan.SetAttributes(attribute.Bool("conversation.responses_background", true))
757+
}
744758
fullLLMMessages := llmMessages
745759
applyOpenAIResponsesInstructions(route, routeConfig.Endpoint, &generateInput)
746760
estimatedPromptTokens = estimateGenerateInputTokens(generateInput)
@@ -838,6 +852,11 @@ func (s *Service) sendMessageInternal(
838852
}
839853
callPromptShape := summarizePromptShape(callPromptMode, currentInput.Messages, currentInput.Messages, currentInput.PreviousResponseID)
840854
usageAccumulator.beginCall(currentInput)
855+
if currentInput.ResponsesBackground {
856+
responsesBackgroundRecovery = openAIResponsesBackgroundRecoveryState{Enabled: true}
857+
} else {
858+
responsesBackgroundRecovery = openAIResponsesBackgroundRecoveryState{}
859+
}
841860
generationCtx, generationSpan := platformtracing.Start(ctx, "conversation.llm.generate",
842861
trace.WithAttributes(append([]attribute.KeyValue{
843862
attribute.Int64("conversation.id", int64(input.ConversationID)),
@@ -846,6 +865,7 @@ func (s *Service) sendMessageInternal(
846865
attribute.String("llm.endpoint", routeConfig.Endpoint),
847866
attribute.Bool("llm.stream", streamRequested && streamSupported),
848867
attribute.Bool("llm.tools_disabled", currentInput.DisableTools),
868+
attribute.Bool("llm.responses_background", currentInput.ResponsesBackground),
849869
attribute.Int("llm.message_count", len(currentInput.Messages)),
850870
attribute.Int("llm.tool_count", len(currentInput.Tools)),
851871
}, promptShapeTraceAttributes("llm.prompt", callPromptShape)...)...),
@@ -908,6 +928,11 @@ func (s *Service) sendMessageInternal(
908928
callStreamUsage := llm.Usage{}
909929
upstreamCallStarted = true
910930
output, streamErr := s.llmClient.GenerateStream(generationCtx, routeConfig, currentInput, func(event llm.GenerateStreamEvent) error {
931+
if currentInput.ResponsesBackground {
932+
if responseID := strings.TrimSpace(event.ResponseID); responseID != "" {
933+
responsesBackgroundRecovery.ResponseID = responseID
934+
}
935+
}
911936
if s.isMessageGenerationCanceled(generationCtx, runID) {
912937
return ErrMessageGenerationCanceled
913938
}
@@ -916,6 +941,9 @@ func (s *Service) sendMessageInternal(
916941
// 这里先换算成本次调用内增量，再累加成本轮消息总量，保证实时展示和最终账单口径一致。
917942
usageDelta := diffLLMUsage(event.Usage, callStreamUsage)
918943
callStreamUsage = event.Usage
944+
if currentInput.ResponsesBackground {
945+
responsesBackgroundRecovery.ObservedUsage = callStreamUsage
946+
}
919947
currentUsage := usageAccumulator.addObservedUsage(usageDelta)
920948
if input.OnEvent != nil {
921949
if err := emitLLMUsageEvent(input.OnEvent, currentUsage); err != nil {
@@ -1011,6 +1039,25 @@ func (s *Service) sendMessageInternal(
10111039
if handleCanceledGeneration(err) {
10121040
return nil, retErr
10131041
}
1042+
if err != nil && generateInput.ResponsesBackground &&
1043+
strings.TrimSpace(streamedText.String()) == "" &&
1044+
shouldRetryWithoutResponsesBackground(err) {
1045+
if s.logger != nil {
1046+
s.logger.Warn("openai_responses_background_rejected_retry_standard",
1047+
zap.String("trace_id", traceid.FromContext(ctx)),
1048+
zap.Uint("conversation_id", input.ConversationID),
1049+
zap.String("protocol", route.Protocol),
1050+
zap.String("upstream_name", route.UpstreamName),
1051+
zap.Error(err),
1052+
)
1053+
}
1054+
generateInput.ResponsesBackground = false
1055+
responsesBackgroundRecovery = openAIResponsesBackgroundRecoveryState{}
1056+
upstreamOutput, err = runGenerate(generateInput)
1057+
if handleCanceledGeneration(err) {
1058+
return nil, retErr
1059+
}
1060+
}
10141061
if err != nil && strings.TrimSpace(generateInput.PreviousResponseID) != "" &&
10151062
strings.TrimSpace(streamedText.String()) == "" &&
10161063
shouldRetryWithoutPreviousResponseID(err) {

‎backend/internal/application/conversation/service_responses_background.go‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
package conversation
2+
3+
import (
4+
"context"
5+
"strings"
6+
"time"
7+
8+
"github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/llm"
9+
"go.uber.org/zap"
10+
)
11+
12+
type openAIResponsesBackgroundRecoveryState struct {
13+
Enabled bool
14+
ResponseID string
15+
ObservedUsage llm.Usage
16+
}
17+
18+
func (s *Service) recoverOpenAIResponsesBackgroundUsage(route llm.RouteConfig, state openAIResponsesBackgroundRecoveryState) (llm.Usage, bool) {
19+
responseID := strings.TrimSpace(state.ResponseID)
20+
if s == nil || s.llmClient == nil || !state.Enabled || responseID == "" {
21+
return llm.Usage{}, false
22+
}
23+
24+
ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
25+
defer cancel()
26+
27+
var cancelErr error
28+
if output, err := s.llmClient.CancelOpenAIResponse(ctx, route, responseID); err == nil {
29+
if output != nil && output.Usage != (llm.Usage{}) {
30+
return output.Usage, true
31+
}
32+
} else {
33+
cancelErr = err
34+
}
35+
36+
var retrieveErr error
37+
for attempt := 0; attempt < 4; attempt++ {
38+
select {
39+
case <-ctx.Done():
40+
return llm.Usage{}, false
41+
case <-time.After(600 * time.Millisecond):
42+
}
43+
output, err := s.llmClient.RetrieveOpenAIResponse(ctx, route, responseID)
44+
if err != nil {
45+
retrieveErr = err
46+
continue
47+
}
48+
if output != nil && output.Usage != (llm.Usage{}) {
49+
return output.Usage, true
50+
}
51+
}
52+
if s.logger != nil {
53+
s.logger.Warn("openai_responses_background_usage_recovery_failed",
54+
zap.String("response_id", responseID),
55+
zap.String("protocol", route.Protocol),
56+
zap.NamedError("cancel_error", cancelErr),
57+
zap.NamedError("retrieve_error", retrieveErr),
58+
)
59+
}
60+
61+
return llm.Usage{}, false
62+
}

‎backend/internal/application/conversation/service_stateful_responses.go‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package conversation
22

33
import (
4+
"encoding/json"
45
"errors"
56
"net/url"
67
"strings"
@@ -56,6 +57,43 @@ func supportsPreviousResponseIDRoute(route *channel.ResolvedRoute) bool {
5657
isOfficialOpenAIBaseURL(route.BaseURL)
5758
}
5859

60+
func supportsOpenAIResponsesBackgroundMode(route *channel.ResolvedRoute) bool {
61+
if route == nil ||
62+
!strings.EqualFold(strings.TrimSpace(route.Protocol), llm.AdapterOpenAIResponses) ||
63+
!isOfficialOpenAIBaseURL(route.BaseURL) {
64+
return false
65+
}
66+
capabilities := decodeModelCapabilities(route.ModelCapabilitiesJSON)
67+
return boolCapability(capabilities, "responsesBackgroundMode") ||
68+
nestedBoolCapability(capabilities, "responses", "backgroundMode")
69+
}
70+
71+
func decodeModelCapabilities(raw string) map[string]interface{} {
72+
parsed := make(map[string]interface{})
73+
if err := json.Unmarshal([]byte(strings.TrimSpace(raw)), &parsed); err != nil {
74+
return nil
75+
}
76+
return parsed
77+
}
78+
79+
func boolCapability(capabilities map[string]interface{}, key string) bool {
80+
value, ok := capabilities[key]
81+
if !ok {
82+
return false
83+
}
84+
enabled, ok := value.(bool)
85+
return ok && enabled
86+
}
87+
88+
func nestedBoolCapability(capabilities map[string]interface{}, parentKey string, childKey string) bool {
89+
parent, ok := capabilities[parentKey].(map[string]interface{})
90+
if !ok {
91+
return false
92+
}
93+
enabled, ok := parent[childKey].(bool)
94+
return ok && enabled
95+
}
96+
5997
func isOfficialOpenAIBaseURL(raw string) bool {
6098
value := strings.TrimSpace(raw)
6199
if value == "" {
@@ -149,3 +187,28 @@ func shouldRetryWithoutPreviousResponseID(err error) bool {
149187
strings.Contains(text, "response_id") ||
150188
strings.Contains(text, "unknown parameter")
151189
}
190+
191+
func shouldRetryWithoutResponsesBackground(err error) bool {
192+
if err == nil {
193+
return false
194+
}
195+
var upstreamErr *llm.UpstreamError
196+
if !errors.As(err, &upstreamErr) {
197+
return false
198+
}
199+
if upstreamErr.StatusCode != 400 && upstreamErr.StatusCode != 409 && upstreamErr.StatusCode != 422 {
200+
return false
201+
}
202+
text := strings.ToLower(upstreamErr.Message + "\n" + upstreamErr.Body)
203+
if !strings.Contains(text, "background") && !strings.Contains(text, "store") &&
204+
!strings.Contains(text, "zero data retention") && !strings.Contains(text, "zdr") {
205+
return false
206+
}
207+
return strings.Contains(text, "unsupported") ||
208+
strings.Contains(text, "not supported") ||
209+
strings.Contains(text, "unknown parameter") ||
210+
strings.Contains(text, "unrecognized") ||
211+
strings.Contains(text, "invalid") ||
212+
strings.Contains(text, "zero data retention") ||
213+
strings.Contains(text, "zdr")
214+
}

‎backend/internal/application/conversation/service_stateful_responses_test.go‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,66 @@ func TestSupportsPreviousResponseIDRouteOnlyAllowsOfficialOpenAIResponses(t *tes
7171
}
7272
}
7373

74+
func TestSupportsOpenAIResponsesBackgroundModeRequiresOfficialRouteAndCapability(t *testing.T) {
75+
if !supportsOpenAIResponsesBackgroundMode(&channel.ResolvedRoute{
76+
Protocol: llm.AdapterOpenAIResponses,
77+
BaseURL: "https://api.openai.com/v1",
78+
ModelCapabilitiesJSON: `{"responsesBackgroundMode":true}`,
79+
}) {
80+
t.Fatalf("expected explicit official OpenAI Responses capability to enable background")
81+
}
82+
if !supportsOpenAIResponsesBackgroundMode(&channel.ResolvedRoute{
83+
Protocol: llm.AdapterOpenAIResponses,
84+
BaseURL: "https://api.openai.com/v1",
85+
ModelCapabilitiesJSON: `{"responses":{"backgroundMode":true}}`,
86+
}) {
87+
t.Fatalf("expected nested official OpenAI Responses capability to enable background")
88+
}
89+
if supportsOpenAIResponsesBackgroundMode(&channel.ResolvedRoute{
90+
Protocol: llm.AdapterOpenAIResponses,
91+
BaseURL: "https://reverse.example.com/v1",
92+
ModelCapabilitiesJSON: `{"responsesBackgroundMode":true}`,
93+
}) {
94+
t.Fatalf("expected custom Responses-compatible route to disable background")
95+
}
96+
if supportsOpenAIResponsesBackgroundMode(&channel.ResolvedRoute{
97+
Protocol: llm.AdapterOpenRouterResponses,
98+
BaseURL: "https://openrouter.ai/api/v1",
99+
ModelCapabilitiesJSON: `{"responsesBackgroundMode":true}`,
100+
}) {
101+
t.Fatalf("expected non-official protocol to disable background")
102+
}
103+
if supportsOpenAIResponsesBackgroundMode(&channel.ResolvedRoute{
104+
Protocol: llm.AdapterOpenAIResponses,
105+
BaseURL: "https://api.openai.com/v1",
106+
ModelCapabilitiesJSON: `{"responsesBackgroundMode":"true"}`,
107+
}) {
108+
t.Fatalf("expected non-boolean capability to disable background")
109+
}
110+
}
111+
112+
func TestShouldRetryWithoutResponsesBackground(t *testing.T) {
113+
err := &llm.UpstreamError{
114+
StatusCode: 400,
115+
Message: "Unknown parameter: background",
116+
}
117+
if !shouldRetryWithoutResponsesBackground(err) {
118+
t.Fatalf("expected unsupported background error to be retryable")
119+
}
120+
if shouldRetryWithoutResponsesBackground(&llm.UpstreamError{
121+
StatusCode: 429,
122+
Message: "rate limit",
123+
}) {
124+
t.Fatalf("expected rate limit to stay non-retryable")
125+
}
126+
if shouldRetryWithoutResponsesBackground(&llm.UpstreamError{
127+
StatusCode: 400,
128+
Message: "invalid temperature",
129+
}) {
130+
t.Fatalf("expected unrelated validation error to stay non-retryable")
131+
}
132+
}
133+
74134
func TestBuildStatefulResponseMessagesKeepsLatestUserOnly(t *testing.T) {
75135
messages := []llm.Message{
76136
{Role: "system", Content: "behavior"},

‎backend/internal/infra/llm/client.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,9 @@ type GenerateInput struct {
141141
// 非空时：仅在 input 中发送本轮新消息，服务端从存储状态续接历史。
142142
// 空串时：退回全量发送模式，适用于所有 adapter。
143143
PreviousResponseID string
144+
// ResponsesBackground 表示官方 OpenAI Responses 请求使用 background mode。
145+
// 这是服务端能力开关，不从用户 Options 透传，避免改变未显式启用模型的数据保留语义。
146+
ResponsesBackground bool
144147
// ImageEditMask 仅供图片编辑 adapter 使用，表示透明区域掩码。
145148
ImageEditMask *ContentPart
146149
}

0 commit comments

Comments
 (0)