Merge pull request #1342 from trheyi/main

Refine interrupt handling and streamline completion request validation
This commit is contained in:
Max 2025-11-24 12:16:01 +08:00 committed by GitHub
commit f4054327c2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 11 additions and 41 deletions

View file

@ -73,9 +73,9 @@ func (ic *InterruptController) handleSignal(signal *InterruptSignal) {
ic.pending = append(ic.pending, signal) ic.pending = append(ic.pending, signal)
} }
// For force interrupt, cancel the interrupt context // For force interrupt with no messages (pure cancellation), cancel the interrupt context
// This allows LLM streaming and other operations to check and stop // This allows LLM streaming and other operations to check and stop
if signal.Type == InterruptForce { if signal.Type == InterruptForce && len(signal.Messages) == 0 {
if ic.cancel != nil { if ic.cancel != nil {
ic.cancel() ic.cancel()
// Create a new context for potential future operations // Create a new context for potential future operations

View file

@ -520,10 +520,11 @@ func TestInterruptContext(t *testing.T) {
// Get context before interrupt // Get context before interrupt
interruptCtx := ctx.Interrupt.Context() interruptCtx := ctx.Interrupt.Context()
// Send force interrupt // Send force interrupt with empty messages (pure cancellation)
// This is the pattern for stopping streaming without appending messages
signal := &InterruptSignal{ signal := &InterruptSignal{
Type: InterruptForce, Type: InterruptForce,
Messages: []Message{{Role: RoleUser, Content: "force stop"}}, Messages: []Message{}, // Empty messages = pure cancellation
Timestamp: time.Now().UnixMilli(), Timestamp: time.Now().UnixMilli(),
} }
err := SendInterrupt(ctx.ID, signal) err := SendInterrupt(ctx.ID, signal)
@ -536,9 +537,9 @@ func TestInterruptContext(t *testing.T) {
// The OLD context should be cancelled // The OLD context should be cancelled
select { select {
case <-interruptCtx.Done(): case <-interruptCtx.Done():
t.Log("✓ Force interrupt cancelled the old context") t.Log("✓ Force interrupt with empty messages cancelled the old context")
case <-time.After(200 * time.Millisecond): case <-time.After(200 * time.Millisecond):
t.Error("Old context was not cancelled after force interrupt") t.Error("Old context was not cancelled after force interrupt with empty messages")
} }
// Note: IsInterrupted() checks the NEW context (which was recreated) // Note: IsInterrupted() checks the NEW context (which was recreated)

View file

@ -6,7 +6,6 @@ import (
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"github.com/yaoapp/kun/log" "github.com/yaoapp/kun/log"
"github.com/yaoapp/kun/utils"
"github.com/yaoapp/yao/agent" "github.com/yaoapp/yao/agent"
"github.com/yaoapp/yao/agent/assistant" "github.com/yaoapp/yao/agent/assistant"
"github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/context"
@ -44,24 +43,6 @@ func GinCreateCompletions(c *gin.Context) {
ctx.Release() ctx.Release()
}() }()
// Print request info for debugging
fmt.Println("-----------------------------------------------")
fmt.Println("Chat ID: ", ctx.ChatID)
fmt.Println("Assistant ID: ", ctx.AssistantID)
fmt.Println("Model: ", completionReq.Model)
fmt.Println("Locale: ", ctx.Locale)
fmt.Println("Messages count: ", len(completionReq.Messages))
if completionReq.Temperature != nil {
fmt.Println("Temperature: ", *completionReq.Temperature)
}
if completionReq.Stream != nil {
fmt.Println("Stream: ", *completionReq.Stream)
}
if completionReq.Metadata != nil {
fmt.Println("Metadata: ", completionReq.Metadata)
}
fmt.Println("-----------------------------------------------")
ast, err := assistant.Get(ctx.AssistantID) ast, err := assistant.Get(ctx.AssistantID)
if err != nil { if err != nil {
response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{ response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{
@ -80,13 +61,9 @@ func GinCreateCompletions(c *gin.Context) {
// Stream the completion (uses default handler which sends to ctx.Writer) // Stream the completion (uses default handler which sends to ctx.Writer)
// The Stream method will automatically close the writer and send [DONE] marker // The Stream method will automatically close the writer and send [DONE] marker
log.Trace("[HTTP] Calling ast.Stream()") log.Trace("[HTTP] Calling ast.Stream()")
res, err := ast.Stream(ctx, completionReq.Messages) _, err = ast.Stream(ctx, completionReq.Messages)
log.Trace("[HTTP] ast.Stream() returned, err=%v", err) log.Trace("[HTTP] ast.Stream() returned, err=%v", err)
if err != nil { if err != nil {
fmt.Println("-----------------------------------------------")
fmt.Println("Error: ", err.Error())
fmt.Println("-----------------------------------------------")
response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{ response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{
Code: response.ErrServerError.Code, Code: response.ErrServerError.Code,
ErrorDescription: "Failed to stream: " + err.Error(), ErrorDescription: "Failed to stream: " + err.Error(),
@ -94,12 +71,6 @@ func GinCreateCompletions(c *gin.Context) {
return return
} }
fmt.Println("-----------------------------------------------")
fmt.Println("Stream completed successfully")
fmt.Println("Response: ")
utils.Dump(res)
fmt.Println("-----------------------------------------------")
// c.JSON(response.StatusOK, gin.H{ // c.JSON(response.StatusOK, gin.H{
// "message": "Create Completions", // "message": "Create Completions",
// "chat_id": ctx.ChatID, // "chat_id": ctx.ChatID,
@ -224,10 +195,11 @@ func GinAppendMessages(c *gin.Context) {
} }
// Validate messages // Validate messages
if len(req.Messages) == 0 { // Allow empty messages for force interrupt (pure cancellation without appending)
if len(req.Messages) == 0 && req.Type != context.InterruptForce {
response.RespondWithError(c, response.StatusBadRequest, &response.ErrorResponse{ response.RespondWithError(c, response.StatusBadRequest, &response.ErrorResponse{
Code: response.ErrInvalidRequest.Code, Code: response.ErrInvalidRequest.Code,
ErrorDescription: "At least one message is required", ErrorDescription: "At least one message is required (unless force interrupt for cancellation)",
}) })
return return
} }
@ -250,9 +222,6 @@ func GinAppendMessages(c *gin.Context) {
return return
} }
log.Trace("[INTERRUPT] Interrupt signal sent successfully: context_id=%s, type=%s, messages=%d",
contextID, req.Type, len(req.Messages))
// Return success response // Return success response
response.RespondWithSuccess(c, response.StatusOK, gin.H{ response.RespondWithSuccess(c, response.StatusOK, gin.H{
"message": "Messages appended successfully", "message": "Messages appended successfully",