Skip to content

Commit a26b420

Browse files
chenymechenyme
authored
Merge pull request #65 from DEEIX-AI/error_message
feat: implement message continuation and error handling improvements
2 parents 4494432 + cdeeb39 commit a26b420

24 files changed

Lines changed: 737 additions & 155 deletions

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

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -155,7 +155,14 @@ func normalizeDefaultBranchContext(
155155
}
156156

157157
func isContextMessage(item *model.Message) bool {
158-
return item != nil && strings.EqualFold(strings.TrimSpace(item.Status), "success")
158+
if item == nil {
159+
return false
160+
}
161+
status := strings.TrimSpace(item.Status)
162+
if strings.EqualFold(status, "success") {
163+
return true
164+
}
165+
return item.Role == "assistant" && strings.EqualFold(status, "interrupted")
159166
}
160167

161168
func selectLatestDefaultParentCandidate(messages []model.Message) *model.Message {

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

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -286,17 +286,41 @@ func upstreamErrorSummary(err *llm.UpstreamError) string {
286286
return ""
287287
}
288288
lines := make([]string, 0, 3)
289+
if isSuccessfulUpstreamStatus(err.StatusCode) {
290+
lines = append(lines, fmt.Sprintf("模型响应格式不兼容（HTTP %d）", err.StatusCode))
291+
lines = append(lines, "错误：上游返回成功状态码，但响应格式与当前协议不兼容")
292+
return strings.Join(lines, "\n")
293+
}
289294
if err.StatusCode > 0 {
290295
lines = append(lines, fmt.Sprintf("模型请求失败（HTTP %d）", err.StatusCode))
291296
} else {
292297
lines = append(lines, "模型请求失败")
293298
}
294-
if message := strings.TrimSpace(err.Message); message != "" {
299+
if message := normalizeUpstreamErrorMessage(err.Message); message != "" {
295300
lines = append(lines, "错误："+message)
296301
}
297302
return strings.Join(lines, "\n")
298303
}
299304

305+
func isSuccessfulUpstreamStatus(statusCode int) bool {
306+
return statusCode >= 200 && statusCode < 300
307+
}
308+
309+
func normalizeUpstreamErrorMessage(message string) string {
310+
value := strings.TrimSpace(message)
311+
if value == "" || looksLikeRawSSEBody(value) {
312+
return ""
313+
}
314+
return value
315+
}
316+
317+
func looksLikeRawSSEBody(value string) bool {
318+
normalized := strings.TrimSpace(value)
319+
return strings.HasPrefix(normalized, "data:") ||
320+
strings.Contains(normalized, "\ndata:") ||
321+
strings.Contains(normalized, " data:")
322+
}
323+
300324
func wrapUpstreamRequestError(cause error) error {
301325
if cause == nil {
302326
return ErrUpstreamRequestFailed

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

Lines changed: 254 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,15 @@ package conversation
22

33
import (
44
"context"
5+
"errors"
56
"strings"
67
"time"
78

9+
"github.com/DEEIX-AI/DEEIX-Chat/backend/internal/application/channel"
810
model "github.com/DEEIX-AI/DEEIX-Chat/backend/internal/domain/conversation"
11+
"github.com/DEEIX-AI/DEEIX-Chat/backend/internal/infra/llm"
12+
"github.com/DEEIX-AI/DEEIX-Chat/backend/internal/pkg/traceid"
13+
"go.uber.org/zap"
914
)
1015

1116
type persistMessageGenerationInput struct {
@@ -25,6 +30,42 @@ type persistMessageGenerationInput struct {
2530
ToolCallRows []model.ToolCall
2631
}
2732

33+
type persistInterruptedMessageGenerationInput struct {
34+
SendInput SendMessageInput
35+
UserMessage *model.Message
36+
AssistantMessage *model.Message
37+
AssistantText string
38+
EstimatedInputTokens int64
39+
Usage llm.Usage
40+
AssistantLatency int64
41+
Error error
42+
ToolCallRows []model.ToolCall
43+
TraceRecorder *messageTraceRecorder
44+
Route *channel.ResolvedRoute
45+
EffectiveOptions map[string]interface{}
46+
ServerSideToolUsage map[string]int64
47+
StartedAt time.Time
48+
}
49+
50+
type interruptedMessageGenerationMetrics struct {
51+
InputTokens int64
52+
OutputTokens int64
53+
LatencyMS int64
54+
ErrorCode string
55+
ErrorMessage string
56+
CacheReadTokens int64
57+
CacheWriteTokens int64
58+
ReasoningTokens int64
59+
}
60+
61+
type persistMessageToolCallsInput struct {
62+
SendInput SendMessageInput
63+
UserMessageID uint
64+
AssistantMessageID uint
65+
RunID string
66+
Rows []model.ToolCall
67+
}
68+
2869
func (s *Service) persistSuccessfulMessageGeneration(ctx context.Context, input persistMessageGenerationInput) error {
2970
input.UserMessage.InputTokens = input.InputTokens
3071
input.UserMessage.CacheReadTokens = input.CacheReadTokens
@@ -56,31 +97,14 @@ func (s *Service) persistSuccessfulMessageGeneration(ctx context.Context, input
5697
input.AssistantMessage.LatencyMS = input.AssistantLatency
5798
input.AssistantMessage.Status = "success"
5899

59-
if len(input.ToolCallRows) > 0 {
60-
for i := range input.ToolCallRows {
61-
if input.ToolCallRows[i].ConversationID == 0 {
62-
input.ToolCallRows[i].ConversationID = input.SendInput.ConversationID
63-
}
64-
if input.ToolCallRows[i].UserID == 0 {
65-
input.ToolCallRows[i].UserID = input.SendInput.UserID
66-
}
67-
if input.ToolCallRows[i].MessageID == 0 {
68-
input.ToolCallRows[i].MessageID = input.AssistantMessage.ID
69-
}
70-
if strings.TrimSpace(input.ToolCallRows[i].RunID) == "" {
71-
input.ToolCallRows[i].RunID = input.AssistantMessage.RunID
72-
}
73-
}
74-
if err := s.repo.CreateConversationToolCalls(ctx, input.ToolCallRows); err != nil {
75-
return err
76-
}
77-
s.persistToolContextArtifacts(ctx, toolContextArtifactInput{
78-
ConversationID: input.SendInput.ConversationID,
79-
UserID: input.SendInput.UserID,
80-
MessageID: input.UserMessage.ID,
81-
RunID: input.AssistantMessage.RunID,
82-
Rows: input.ToolCallRows,
83-
})
100+
if err := s.persistMessageToolCalls(ctx, persistMessageToolCallsInput{
101+
SendInput: input.SendInput,
102+
UserMessageID: input.UserMessage.ID,
103+
AssistantMessageID: input.AssistantMessage.ID,
104+
RunID: input.AssistantMessage.RunID,
105+
Rows: input.ToolCallRows,
106+
}); err != nil {
107+
return err
84108
}
85109

86110
s.updateStatefulResponseAsync(input.SendInput.ConversationID, input.ResponseID, input.StatefulPromptFingerprint)
@@ -90,6 +114,211 @@ func (s *Service) persistSuccessfulMessageGeneration(ctx context.Context, input
90114
return nil
91115
}
92116

117+
// persistInterruptedMessageGeneration 在模型调用已经产生可见内容或工具轨迹后失败时，保留本轮 assistant 消息。
118+
// 显式取消由取消流程单独处理，避免把用户主动停止误标为异常中断。
119+
func (s *Service) persistInterruptedMessageGeneration(ctx context.Context, input persistInterruptedMessageGenerationInput) *SendMessageResult {
120+
if !shouldPersistInterruptedMessageGeneration(input) {
121+
return nil
122+
}
123+
persistCtx := ctx
124+
var cancel context.CancelFunc
125+
if persistCtx == nil || persistCtx.Err() != nil {
126+
persistCtx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
127+
defer cancel()
128+
}
129+
130+
metrics := resolveInterruptedMessageGenerationMetrics(input)
131+
132+
if err := s.repo.UpdateMessageUsage(
133+
persistCtx,
134+
input.UserMessage.ID,
135+
metrics.InputTokens,
136+
0,
137+
metrics.CacheReadTokens,
138+
metrics.CacheWriteTokens,
139+
0,
140+
); err != nil {
141+
s.logger.Warn("persist_interrupted_user_usage_failed",
142+
zap.String("trace_id", traceid.FromContext(ctx)),
143+
zap.Uint("message_id", input.UserMessage.ID),
144+
zap.Error(err),
145+
)
146+
}
147+
if err := s.repo.UpdateMessageState(persistCtx, input.UserMessage.ID, "success", "", ""); err != nil {
148+
s.logger.Warn("persist_interrupted_user_state_failed",
149+
zap.String("trace_id", traceid.FromContext(ctx)),
150+
zap.Uint("message_id", input.UserMessage.ID),
151+
zap.Error(err),
152+
)
153+
}
154+
if err := s.repo.UpdateAssistantMessageCompletion(
155+
persistCtx,
156+
input.AssistantMessage.ID,
157+
input.AssistantText,
158+
metrics.OutputTokens,
159+
metrics.ReasoningTokens,
160+
metrics.LatencyMS,
161+
"interrupted",
162+
metrics.ErrorCode,
163+
metrics.ErrorMessage,
164+
); err != nil {
165+
s.logger.Error("persist_interrupted_assistant_completion_failed",
166+
zap.String("trace_id", traceid.FromContext(ctx)),
167+
zap.Uint("message_id", input.AssistantMessage.ID),
168+
zap.Error(err),
169+
)
170+
return nil
171+
}
172+
applyInterruptedMessageGenerationState(input, metrics)
173+
174+
if err := s.persistMessageToolCalls(persistCtx, persistMessageToolCallsInput{
175+
SendInput: input.SendInput,
176+
UserMessageID: input.UserMessage.ID,
177+
AssistantMessageID: input.AssistantMessage.ID,
178+
RunID: input.AssistantMessage.RunID,
179+
Rows: input.ToolCallRows,
180+
}); err != nil {
181+
s.logger.Warn("persist_interrupted_tool_calls_failed",
182+
zap.String("trace_id", traceid.FromContext(ctx)),
183+
zap.Uint("message_id", input.AssistantMessage.ID),
184+
zap.Error(err),
185+
)
186+
}
187+
if input.TraceRecorder != nil {
188+
input.TraceRecorder.fail(input.Error)
189+
input.TraceRecorder.attachToMessage(input.AssistantMessage)
190+
}
191+
192+
return buildInterruptedSendMessageResult(input, metrics)
193+
}
194+
195+
// shouldPersistInterruptedMessageGeneration 只在已有可展示内容或可追踪工具结果时保留中断消息。
196+
func shouldPersistInterruptedMessageGeneration(input persistInterruptedMessageGenerationInput) bool {
197+
if input.Error == nil || input.UserMessage == nil || input.AssistantMessage == nil {
198+
return false
199+
}
200+
if errors.Is(input.Error, ErrMessageGenerationCanceled) {
201+
return false
202+
}
203+
hasRetainedToolTrace := len(input.ToolCallRows) > 0 || len(input.ServerSideToolUsage) > 0
204+
return strings.TrimSpace(input.AssistantText) != "" || hasRetainedToolTrace
205+
}
206+
207+
// resolveInterruptedMessageGenerationMetrics 统一处理中断消息的真实 usage 与估算兜底。
208+
func resolveInterruptedMessageGenerationMetrics(input persistInterruptedMessageGenerationInput) interruptedMessageGenerationMetrics {
209+
inputTokens := input.Usage.InputTokens
210+
if inputTokens <= 0 {
211+
inputTokens = input.EstimatedInputTokens
212+
}
213+
outputTokens := input.Usage.OutputTokens
214+
if outputTokens <= 0 && strings.TrimSpace(input.AssistantText) != "" {
215+
outputTokens = estimateTokens(input.AssistantText)
216+
}
217+
latencyMS := input.AssistantLatency
218+
if latencyMS < 0 {
219+
latencyMS = time.Since(input.StartedAt).Milliseconds()
220+
}
221+
if latencyMS < 0 {
222+
latencyMS = 0
223+
}
224+
return interruptedMessageGenerationMetrics{
225+
InputTokens: inputTokens,
226+
OutputTokens: outputTokens,
227+
LatencyMS: latencyMS,
228+
ErrorCode: classifyRunErrorCode(input.Error),
229+
ErrorMessage: truncateError(messageErrorSummary(input.Error), 255),
230+
CacheReadTokens: input.Usage.CacheReadTokens,
231+
CacheWriteTokens: input.Usage.CacheWriteTokens,
232+
ReasoningTokens: input.Usage.ReasoningTokens,
233+
}
234+
}
235+
236+
// applyInterruptedMessageGenerationState 同步内存消息对象，保证后续响应、run 记录和持久化状态一致。
237+
func applyInterruptedMessageGenerationState(input persistInterruptedMessageGenerationInput, metrics interruptedMessageGenerationMetrics) {
238+
input.UserMessage.Status = "success"
239+
input.UserMessage.ErrorCode = ""
240+
input.UserMessage.ErrorMessage = ""
241+
input.UserMessage.InputTokens = metrics.InputTokens
242+
input.UserMessage.CacheReadTokens = metrics.CacheReadTokens
243+
input.UserMessage.CacheWriteTokens = metrics.CacheWriteTokens
244+
input.UserMessage.TokenUsage = metrics.InputTokens + metrics.CacheReadTokens + metrics.CacheWriteTokens
245+
246+
input.AssistantMessage.Content = input.AssistantText
247+
input.AssistantMessage.TokenUsage = metrics.OutputTokens + metrics.ReasoningTokens
248+
input.AssistantMessage.OutputTokens = metrics.OutputTokens
249+
input.AssistantMessage.ReasoningTokens = metrics.ReasoningTokens
250+
input.AssistantMessage.LatencyMS = metrics.LatencyMS
251+
input.AssistantMessage.Status = "interrupted"
252+
input.AssistantMessage.ErrorCode = metrics.ErrorCode
253+
input.AssistantMessage.ErrorMessage = metrics.ErrorMessage
254+
}
255+
256+
// buildInterruptedSendMessageResult 构造中断回复响应，供 handler 继续走计费和前端展示链路。
257+
func buildInterruptedSendMessageResult(input persistInterruptedMessageGenerationInput, metrics interruptedMessageGenerationMetrics) *SendMessageResult {
258+
result := &SendMessageResult{
259+
UserMessage: *input.UserMessage,
260+
AssistantMessage: *input.AssistantMessage,
261+
EffectiveOptions: input.EffectiveOptions,
262+
UsageSpeed: input.Usage.Speed,
263+
UsageServiceTier: input.Usage.ServiceTier,
264+
CacheWrite5mTokens: input.Usage.CacheWrite5mTokens,
265+
CacheWrite1hTokens: input.Usage.CacheWrite1hTokens,
266+
ServerSideToolUsage: input.ServerSideToolUsage,
267+
LatencyMS: metrics.LatencyMS,
268+
}
269+
if input.Route != nil {
270+
result.UpstreamID = input.Route.UpstreamID
271+
result.UpstreamName = input.Route.UpstreamName
272+
result.PlatformModelName = input.Route.PlatformModelName
273+
result.RoutedBindingCode = input.Route.BindingCode
274+
result.UpstreamModelName = input.Route.UpstreamModel
275+
result.UpstreamProtocol = input.Route.Protocol
276+
}
277+
return result
278+
}
279+
280+
// persistMessageToolCalls 持久化工具调用并写入上下文 artifact，成功和中断路径共用同一套归属规则。
281+
func (s *Service) persistMessageToolCalls(ctx context.Context, input persistMessageToolCallsInput) error {
282+
rows := normalizeMessageToolCallRows(input)
283+
if len(rows) == 0 {
284+
return nil
285+
}
286+
if err := s.repo.CreateConversationToolCalls(ctx, rows); err != nil {
287+
return err
288+
}
289+
s.persistToolContextArtifacts(ctx, toolContextArtifactInput{
290+
ConversationID: input.SendInput.ConversationID,
291+
UserID: input.SendInput.UserID,
292+
MessageID: input.UserMessageID,
293+
RunID: input.RunID,
294+
Rows: rows,
295+
})
296+
return nil
297+
}
298+
299+
// normalizeMessageToolCallRows 补齐工具调用归属字段，避免不同路径写入的 trace 缺少 message/run 关联。
300+
func normalizeMessageToolCallRows(input persistMessageToolCallsInput) []model.ToolCall {
301+
if len(input.Rows) == 0 {
302+
return nil
303+
}
304+
rows := append([]model.ToolCall(nil), input.Rows...)
305+
for i := range rows {
306+
if rows[i].ConversationID == 0 {
307+
rows[i].ConversationID = input.SendInput.ConversationID
308+
}
309+
if rows[i].UserID == 0 {
310+
rows[i].UserID = input.SendInput.UserID
311+
}
312+
if rows[i].MessageID == 0 {
313+
rows[i].MessageID = input.AssistantMessageID
314+
}
315+
if strings.TrimSpace(rows[i].RunID) == "" {
316+
rows[i].RunID = input.RunID
317+
}
318+
}
319+
return rows
320+
}
321+
93322
func (s *Service) updateStatefulResponseAsync(conversationID uint, responseID string, promptFingerprint string) {
94323
respID := strings.TrimSpace(responseID)
95324
if respID == "" {

0 commit comments

Comments
 (0)