diff --git a/im/backend/internal/app/ai_agent.go b/im/backend/internal/app/ai_agent.go index 0e604d3..32235f0 100644 --- a/im/backend/internal/app/ai_agent.go +++ b/im/backend/internal/app/ai_agent.go @@ -8,6 +8,7 @@ import ( "fmt" "log" "math/rand" + "regexp" "strings" "sync" "time" @@ -258,14 +259,31 @@ func (a *App) processAIReplyJob(ctx context.Context, job aiReplyJob) error { result, callErr := provider.Chat(callCtx, request) latency := int(time.Since(started).Milliseconds()) a.logAICall(ctx, model.ID, "chat", job.AgentUserID, job.ConversationID, latency, result, callErr) + // Reasoning models spend the budget thinking before they answer. With a + // small ceiling the call "succeeds" with nothing to say, so give it room and + // ask once more rather than logging a silent empty reply. + if callErr == nil && result.truncated() { + retry := request + retry.MaxTokens = aiRetryMaxTokens(aiRequestMaxTokens(request, model)) + retryCtx, cancelRetry := context.WithTimeout(ctx, time.Duration(model.TimeoutMS+2000)*time.Millisecond) + started = time.Now() + result, callErr = provider.Chat(retryCtx, retry) + cancelRetry() + a.logAICall(ctx, model.ID, "chat", job.AgentUserID, job.ConversationID, int(time.Since(started).Milliseconds()), result, callErr) + } if callErr != nil { if fallback := strings.TrimSpace(a.configPlain(ctx, "ai.fallback_text", "")); fallback != "" { _ = a.sendAIReply(ctx, job, fallback) } return callErr } + // A short answer that survived the retry is still worth sending; only an + // empty one is a failure. text := sanitizeAIReply(result.Text) if text == "" { + if result.truncated() { + return fmt.Errorf("模型把 %d tokens 全部用于推理,未产出正文;请在模型配置中调高最大 tokens,或改用非推理模型", result.CompletionTokens) + } return fmt.Errorf("模型返回内容为空") } if first && a.configBool(ctx, "ai.disclose_in_chat", true) { @@ -433,6 +451,11 @@ func aiSystemPrompt(binding aiAgentBinding, disclose bool) string { } // Models drift into assistant formatting; the chat bubble wants one plain line. +// Only an actual list marker is removed. Trimming the digit characters +// themselves ate the number out of every reply that opened with one — "30岁啦" +// arrived as "岁啦". +var aiListMarker = regexp.MustCompile(`^(?:[-*•]+\s*|\d{1,2}(?:[.)]\s+|、\s*))`) + func sanitizeAIReply(text string) string { text = strings.TrimSpace(text) if text == "" { @@ -445,7 +468,7 @@ func sanitizeAIReply(text string) string { text = replacer.Replace(text) lines := []string{} for _, line := range strings.Split(text, "\n") { - line = strings.TrimSpace(strings.TrimLeft(strings.TrimSpace(line), "-*•0123456789. ")) + line = strings.TrimSpace(aiListMarker.ReplaceAllString(strings.TrimSpace(line), "")) if line != "" { lines = append(lines, line) } diff --git a/im/backend/internal/app/ai_enqueue_integration_test.go b/im/backend/internal/app/ai_enqueue_integration_test.go index e8b424c..ede95a8 100644 --- a/im/backend/internal/app/ai_enqueue_integration_test.go +++ b/im/backend/internal/app/ai_enqueue_integration_test.go @@ -4,6 +4,10 @@ import ( "context" "encoding/json" "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" "testing" ) @@ -168,3 +172,77 @@ func TestDefaultAIModelAcceptsNameOrIDMySQL(t *testing.T) { t.Fatalf("留空时使用标记为默认的一条: id = %d", model.ID) } } + +// End to end over a real database and a real HTTP model: the first call comes +// back blank because the reasoning ate the budget, and the reply still has to +// arrive. +func TestAIReplyRetriesAfterReasoningExhaustsBudgetMySQL(t *testing.T) { + db := isolatedIMDatabase(t) + a := &App{db: db, hub: NewHub(), config: Config{Environment: "development", JWTSecret: "isolated-im-regression-secret-only"}} + budgets := []int{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + raw, _ := io.ReadAll(io.LimitReader(r.Body, 1<<20)) + _ = json.Unmarshal(raw, &body) + budgets = append(budgets, int(body["max_tokens"].(float64))) + w.Header().Set("Content-Type", "application/json") + if len(budgets) == 1 { + _, _ = w.Write([]byte(`{"id":"a","choices":[{"message":{"content":"","reasoning_content":"想一想"},"finish_reason":"length"}],"usage":{"prompt_tokens":240,"completion_tokens":256}}`)) + return + } + _, _ = w.Write([]byte(`{"id":"b","choices":[{"message":{"content":"在的,刚忙完","reasoning_content":"想好了"},"finish_reason":"stop"}],"usage":{"prompt_tokens":240,"completion_tokens":40}}`)) + })) + defer server.Close() + for _, statement := range []string{ + `INSERT INTO system_configs(config_key,config_value,value_type,description) VALUES('ai.enabled','true','boolean',''),('ai.disclose_in_chat','false','boolean','') ON DUPLICATE KEY UPDATE config_value=VALUES(config_value)`, + `UPDATE users SET is_test=1,test_batch='regression' WHERE id=2`, + `INSERT INTO ai_agents(user_id,model_id,enabled) VALUES(2,1,1)`, + } { + if _, err := db.Exec(statement); err != nil { + t.Fatal(err) + } + } + cipher, err := a.encryptSecret("test-key") + if err != nil { + t.Fatal(err) + } + // max_tokens 256 is the setting that produced blank replies in production. + if _, err := db.Exec(`INSERT INTO ai_models(id,name,protocol,base_url,model_name,api_key_cipher,temperature,max_tokens,timeout_ms,status,is_default) VALUES(1,'reasoning','openai',?,'reasoning-model',?,0.80,256,8000,1,1)`, server.URL, cipher); err != nil { + t.Fatal(err) + } + var created struct { + ID int64 `json:"id"` + } + if err := json.Unmarshal(imTestCall(t, a.directConversation, 1, "POST", "/api/v1/im/conversations", `{"userId":2}`, 200), &created); err != nil { + t.Fatal(err) + } + if _, _, err := a.persistMessageContext(context.Background(), created.ID, 1, "trigger", 1, map[string]any{"text": "在吗"}); err != nil { + t.Fatal(err) + } + var job aiReplyJob + if err := db.QueryRow(`SELECT id,conversation_id,agent_user_id,attempts FROM ai_reply_jobs WHERE conversation_id=?`, created.ID). + Scan(&job.ID, &job.ConversationID, &job.AgentUserID, &job.Attempts); err != nil { + t.Fatal(err) + } + if err := a.processAIReplyJob(context.Background(), job); err != nil { + t.Fatalf("推理耗尽预算后应当重试成功: %v", err) + } + if len(budgets) != 2 || budgets[0] != 256 || budgets[1] != 1024 { + t.Fatalf("budgets = %v, want [256 1024]", budgets) + } + var text string + if err := db.QueryRow(`SELECT CAST(body AS CHAR) FROM im_messages WHERE conversation_id=? AND sender_id=2 ORDER BY id DESC LIMIT 1`, created.ID).Scan(&text); err != nil { + t.Fatal(err) + } + if !strings.Contains(text, "刚忙完") { + t.Fatalf("回复没有落库: %s", text) + } + // Both attempts are on the record, so the cost of the retry is visible. + var calls int + if err := db.QueryRow(`SELECT COUNT(*) FROM ai_call_logs WHERE conversation_id=?`, created.ID).Scan(&calls); err != nil { + t.Fatal(err) + } + if calls != 2 { + t.Fatalf("calls = %d, want 2", calls) + } +} diff --git a/im/backend/internal/app/ai_providers.go b/im/backend/internal/app/ai_providers.go index df05b1a..cdae894 100644 --- a/im/backend/internal/app/ai_providers.go +++ b/im/backend/internal/app/ai_providers.go @@ -36,12 +36,21 @@ type aiChatRequest struct { type aiChatResult struct { Text string + Reasoning string PromptTokens int CompletionTokens int FinishReason string RequestID string } +// The model ran out of budget mid-thought. For a reasoning model that usually +// means the whole ceiling went on the chain of thought and Text is empty; for +// any model it can also mean the answer itself was cut off. Both deserve one +// retry with room to finish rather than a blank or half-finished message. +func (r aiChatResult) truncated() bool { + return r.FinishReason == "length" || (r.Text == "" && r.Reasoning != "") +} + type aiProvider interface { Chat(ctx context.Context, req aiChatRequest) (aiChatResult, error) } @@ -160,6 +169,20 @@ func aiRequestMaxTokens(req aiChatRequest, model aiModel) int { return 256 } +// One retry, with enough room for the thinking plus a short answer, and a cap +// so a misconfigured model cannot bill an unbounded completion. +func aiRetryMaxTokens(current int) int { + const floor, ceiling = 1024, 4096 + raised := current * 4 + if raised < floor { + raised = floor + } + if raised > ceiling { + raised = ceiling + } + return raised +} + func aiPostJSON(ctx context.Context, client *http.Client, endpoint string, headers map[string]string, payload any) ([]byte, error) { body, err := json.Marshal(payload) if err != nil { @@ -234,6 +257,11 @@ func (p openAIProvider) Chat(ctx context.Context, req aiChatRequest) (aiChatResu Choices []struct { Message struct { Content string `json:"content"` + // Reasoning models put their chain of thought here and the answer + // in Content. Spending the whole budget on the former leaves the + // latter empty, which is indistinguishable from a broken model + // unless this field is read. + ReasoningContent string `json:"reasoning_content"` } `json:"message"` FinishReason string `json:"finish_reason"` } `json:"choices"` @@ -256,6 +284,7 @@ func (p openAIProvider) Chat(ctx context.Context, req aiChatRequest) (aiChatResu } return aiChatResult{ Text: strings.TrimSpace(payload.Choices[0].Message.Content), + Reasoning: strings.TrimSpace(payload.Choices[0].Message.ReasoningContent), PromptTokens: payload.Usage.PromptTokens, CompletionTokens: payload.Usage.CompletionTokens, FinishReason: payload.Choices[0].FinishReason, diff --git a/im/backend/internal/app/ai_providers_test.go b/im/backend/internal/app/ai_providers_test.go index b8215d3..6d1bf62 100644 --- a/im/backend/internal/app/ai_providers_test.go +++ b/im/backend/internal/app/ai_providers_test.go @@ -216,3 +216,107 @@ func TestDebugProviderIsDeterministicAndOffline(t *testing.T) { t.Fatal("debug provider must answer even without history") } } + +// A reasoning model answers in two parts. When the token ceiling is reached +// before it stops thinking, the reply field is empty and the whole budget shows +// up as completion tokens — exactly what the admin log showed as "成功" with a +// blank reply. +func TestReasoningModelExhaustingItsBudgetIsRecognised(t *testing.T) { + server, _, _ := captureAIRequest(t, `{"id":"call-1","choices":[{"message":{"content":"","reasoning_content":"先想想怎么回答"},"finish_reason":"length"}],"usage":{"prompt_tokens":240,"completion_tokens":256}}`, 200) + result, err := chatOnce(t, "openai", server.URL) + if err != nil { + t.Fatalf("空正文本身不是传输错误: %v", err) + } + if result.Text != "" || result.Reasoning != "先想想怎么回答" { + t.Fatalf("result = %+v", result) + } + if !result.truncated() { + t.Fatal("必须能识别出被截断") + } + + // The ordinary case must not be mistaken for it. + plain, _, _ := captureAIRequest(t, `{"id":"call-2","choices":[{"message":{"content":"在的,刚吃过"},"finish_reason":"stop"}],"usage":{"prompt_tokens":240,"completion_tokens":62}}`, 200) + answered, err := chatOnce(t, "openai", plain.URL) + if err != nil { + t.Fatal(err) + } + if answered.Text != "在的,刚吃过" || answered.truncated() { + t.Fatalf("result = %+v", answered) + } + + // An empty reply with no reasoning and a normal stop is still just empty. + blank, _, _ := captureAIRequest(t, `{"id":"call-3","choices":[{"message":{"content":""},"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":0}}`, 200) + empty, err := chatOnce(t, "openai", blank.URL) + if err != nil { + t.Fatal(err) + } + if empty.truncated() { + t.Fatal("普通空回复不该被当成截断") + } +} + +func TestAIRetryMaxTokensStaysWithinBounds(t *testing.T) { + for _, test := range []struct{ current, want int }{ + {256, 1024}, // the configured value that produced blank replies + {64, 1024}, // never retry with less than a usable budget + {512, 2048}, // four times when that is already sensible + {2048, 4096}, // capped, so a misconfiguration cannot bill without limit + {8192, 4096}, + } { + if got := aiRetryMaxTokens(test.current); got != test.want { + t.Fatalf("aiRetryMaxTokens(%d) = %d, want %d", test.current, got, test.want) + } + } +} + +// The request the retry sends has to actually carry the raised ceiling. +func TestChatRequestMaxTokensOverridesTheModelConfiguration(t *testing.T) { + server, body, _ := captureAIRequest(t, `{"id":"c","choices":[{"message":{"content":"好"},"finish_reason":"stop"}],"usage":{}}`, 200) + provider, err := newAIProvider(testModel("openai", server.URL), false) + if err != nil { + t.Fatal(err) + } + if _, err = provider.Chat(context.Background(), aiChatRequest{ + Messages: []aiChatMessage{{Role: "user", Text: "在吗"}}, + MaxTokens: aiRetryMaxTokens(128), + }); err != nil { + t.Fatal(err) + } + if got := (*body)["max_tokens"]; got != float64(1024) { + t.Fatalf("max_tokens = %v, want 1024", got) + } +} + +// A reply that opens with a number used to arrive with the number missing: the +// list-marker trimming removed the digits themselves. +func TestSanitizeAIReplyKeepsLeadingNumbers(t *testing.T) { + for _, test := range []struct{ raw, want string }{ + {"30岁啦,你呢", "30岁啦,你呢"}, + {"1. 先自我介绍", "先自我介绍"}, + {"2、我在深圳", "我在深圳"}, + {"- 喜欢跑步", "喜欢跑步"}, + {"• 也喜欢音乐", "也喜欢音乐"}, + {"**在的**", "在的"}, + {"18612725546 是我的号码", "18612725546 是我的号码"}, + {"3 个月前搬来的", "3 个月前搬来的"}, + } { + if got := sanitizeAIReply(test.raw); got != test.want { + t.Fatalf("sanitizeAIReply(%q) = %q, want %q", test.raw, got, test.want) + } + } +} + +// A cut-off answer is worth one more attempt too, not just an empty one. +func TestTruncatedAnswerIsRetriedAndKeptIfItSurvives(t *testing.T) { + cut, _, _ := captureAIRequest(t, `{"id":"c","choices":[{"message":{"content":"30岁啦,"},"finish_reason":"length"}],"usage":{"prompt_tokens":240,"completion_tokens":1024}}`, 200) + result, err := chatOnce(t, "openai", cut.URL) + if err != nil { + t.Fatal(err) + } + if !result.truncated() { + t.Fatal("被截断的正文也要触发重试") + } + if sanitizeAIReply(result.Text) != "30岁啦," { + t.Fatalf("text = %q", result.Text) + } +}