yao/openapi/chat/completions.go
Max b43929210e Implement event message handling and enhance output capabilities
- Added support for lifecycle event messages, including a new event type for tracking stream states (e.g., stream_start, stream_end).
- Introduced a NewEventMessage function to create event messages with structured properties.
- Updated the output package to include event messages in the built-in types, ensuring compatibility with CUI clients while remaining silent for OpenAI clients.
- Enhanced the DefaultStreamHandler to utilize event messages for better lifecycle tracking during streaming operations.
- Improved error handling and logging in the OpenAI provider to capture and report streaming errors effectively.
2025-11-16 10:30:55 +08:00

184 lines
6.1 KiB
Go

package chat
import (
"fmt"
"github.com/gin-gonic/gin"
"github.com/yaoapp/kun/utils"
"github.com/yaoapp/yao/agent"
"github.com/yaoapp/yao/agent/assistant"
"github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/openapi/response"
)
// GinCreateCompletions handles POST /chat/:assistant_id/completions - Create a chat completion
func GinCreateCompletions(c *gin.Context) {
agent := agent.GetAgent()
cache, err := agent.GetCacheStore()
if err != nil {
response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{
Code: response.ErrServerError.Code,
ErrorDescription: "Failed to get cache store: " + err.Error(),
})
return
}
completionReq, ctx, err := context.GetCompletionRequest(c, cache)
if err != nil {
fmt.Println("-----------------------------------------------")
fmt.Println("Error: ", err.Error())
fmt.Println("-----------------------------------------------")
response.RespondWithError(c, response.StatusBadRequest, &response.ErrorResponse{
Code: response.ErrInvalidRequest.Code,
ErrorDescription: "Failed to parse request: " + err.Error(),
})
return
}
defer ctx.Release() // Release the context after the request is complete
// 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("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)
if err != nil {
response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{
Code: response.ErrServerError.Code,
ErrorDescription: "Failed to get assistant: " + err.Error(),
})
return
}
// Set SSE headers for streaming response
c.Header("Content-Type", "text/event-stream;charset=utf-8")
c.Header("Cache-Control", "no-cache")
c.Header("Connection", "keep-alive")
c.Header("X-Accel-Buffering", "no") // Disable buffering in nginx
// Stream the completion (uses default handler which sends to ctx.Writer)
// The Stream method will automatically close the writer and send [DONE] marker
res, err := ast.Stream(ctx, completionReq.Messages)
if err != nil {
fmt.Println("-----------------------------------------------")
fmt.Println("Error: ", err.Error())
fmt.Println("-----------------------------------------------")
response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{
Code: response.ErrServerError.Code,
ErrorDescription: "Failed to stream: " + err.Error(),
})
return
}
fmt.Println("-----------------------------------------------")
fmt.Println("Stream completed successfully")
fmt.Println("Response: ")
utils.Dump(res)
fmt.Println("-----------------------------------------------")
// c.JSON(response.StatusOK, gin.H{
// "message": "Create Completions",
// "chat_id": ctx.ChatID,
// "assistant_id": ctx.AssistantID,
// "model": completionReq.Model,
// "messages_count": len(completionReq.Messages),
// })
// // Print headers
// fmt.Println("\n--- Headers ---")
// for key, values := range c.Request.Header {
// for _, value := range values {
// fmt.Printf("%s: %s\n", key, value)
// }
// }
// // Print path parameters
// fmt.Println("\n--- Path Parameters ---")
// for _, param := range c.Params {
// fmt.Printf("%s: %s\n", param.Key, param.Value)
// }
// // Print query parameters
// fmt.Println("\n--- Query Parameters ---")
// for key, values := range c.Request.URL.Query() {
// for _, value := range values {
// fmt.Printf("%s: %s\n", key, value)
// }
// }
// // Print request body
// fmt.Println("\n--- Request Body ---")
// body, err = io.ReadAll(c.Request.Body)
// if err != nil {
// fmt.Printf("Error reading body: %v\n", err)
// } else {
// fmt.Printf("%s\n", string(body))
// // Restore the body for further processing
// c.Request.Body = io.NopCloser(bytes.NewBuffer(body))
// }
// fmt.Println("===============================================")
// // Handle Sid - try multiple methods for maximum compatibility
// var sid string
// // Method 1: Check if client sent X-Session-Id header
// sid = c.GetHeader("X-Session-Id")
// // Method 2: Try to read from cookie
// if sid == "" {
// sid, err = c.Cookie("Sid")
// if err == nil && sid != "" {
// fmt.Printf("Existing Sid from cookie: %s\n", sid)
// }
// } else {
// fmt.Printf("Existing Sid from header: %s\n", sid)
// }
// // Method 3: For clients that can't store cookies/headers (like Electron cross-origin),
// // generate a deterministic session ID based on client fingerprint
// if sid == "" {
// // Use Authorization token if available (most stable identifier)
// authToken := c.GetHeader("Authorization")
// userAgent := c.GetHeader("User-Agent")
// if authToken != "" {
// // Generate stable session ID from auth token
// hash := md5.Sum([]byte(authToken))
// sid = hex.EncodeToString(hash[:])
// fmt.Printf("Generated deterministic Sid from auth token: %s\n", sid)
// } else {
// // Fallback: generate random UUID
// sid = uuid.New().String()
// fmt.Printf("Generated random Sid: %s\n", sid)
// }
// fmt.Printf("Client fingerprint - UserAgent: %s\n", userAgent)
// }
// // Try to set cookie (may not work for cross-origin, but doesn't hurt)
// c.SetCookie("Sid", sid, 86400*30, "/", "", false, false)
// // Return Sid in response header and body for client reference
// c.Header("X-Session-Id", sid)
// response.RespondWithSuccess(c, response.StatusOK, gin.H{"message": "Create Completions", "sid": sid})
}
// GinUpdateCompletions handles PUT /chat/:assistant_id/completions - Update a chat completion metadata
func GinUpdateCompletions(c *gin.Context) {}