From d973686760af05e0aeac6ee9fb79a11c542c142d Mon Sep 17 00:00:00 2001 From: Max Date: Sun, 23 Nov 2025 09:55:04 +0800 Subject: [PATCH 1/2] Implement temporary mock SID for chat handling and comment out userID retrieval in chat queries - Added a temporary mock SID value in the handleChat, handleChatList, and handleChatLatest functions to facilitate testing. - Commented out userID retrieval in GetChat, GetChatWithFilter, and getChatsWithFilter methods to simplify query logic during development. - Updated GetHistory method to comment out userID usage, maintaining consistency across chat-related functionalities. --- agent/api/api.go | 3 +++ agent/store/xun/chat.go | 36 ++++++++++++++++++------------------ agent/store/xun/history.go | 10 +++++----- 3 files changed, 26 insertions(+), 23 deletions(-) diff --git a/agent/api/api.go b/agent/api/api.go index 3fe8ac21..cd1b302d 100644 --- a/agent/api/api.go +++ b/agent/api/api.go @@ -218,6 +218,7 @@ func (agent *API) handleChat(c *gin.Context) { // handleChatList handles the chat list request func (agent *API) handleChatList(c *gin.Context) { sid := c.GetString("__sid") + sid = "temporary-mock-sid-123" if sid == "" { c.JSON(400, gin.H{"message": "sid is required", "code": 400}) c.Done() @@ -262,6 +263,7 @@ func (agent *API) handleChatList(c *gin.Context) { // handleChatHistory handles the chat history request func (agent *API) handleChatHistory(c *gin.Context) { sid := c.GetString("__sid") + sid = "temporary-mock-sid-123" if sid == "" { c.JSON(400, gin.H{"message": "sid is required", "code": 400}) c.Done() @@ -372,6 +374,7 @@ func (agent *API) getSessionID(c *gin.Context) string { // handleChatLatest handles getting the latest chat func (agent *API) handleChatLatest(c *gin.Context) { sid := c.GetString("__sid") + sid = "temporary-mock-sid-123" if sid == "" { c.JSON(400, gin.H{"message": "sid is required", "code": 400}) c.Done() diff --git a/agent/store/xun/chat.go b/agent/store/xun/chat.go index d33811c7..fe8ef3d3 100644 --- a/agent/store/xun/chat.go +++ b/agent/store/xun/chat.go @@ -29,15 +29,15 @@ func (conv *Xun) UpdateChatTitle(sid string, cid string, title string) error { // GetChat get the chat info and its history func (conv *Xun) GetChat(sid string, cid string, locale ...string) (*types.ChatInfo, error) { - userID, err := conv.getUserID(sid) - if err != nil { - return nil, err - } + // userID, err := conv.getUserID(sid) + // if err != nil { + // return nil, err + // } // Get chat info qb := conv.newQueryChat(). Select("chat_id", "title", "assistant_id"). - Where("sid", userID). + // Where("sid", userID). Where("chat_id", cid) row, err := qb.First() @@ -93,15 +93,15 @@ func (conv *Xun) GetChat(sid string, cid string, locale ...string) (*types.ChatI // GetChatWithFilter get the chat info and its history with filter options func (conv *Xun) GetChatWithFilter(sid string, cid string, filter types.ChatFilter, locale ...string) (*types.ChatInfo, error) { - userID, err := conv.getUserID(sid) - if err != nil { - return nil, err - } + // userID, err := conv.getUserID(sid) + // if err != nil { + // return nil, err + // } // Get chat info qb := conv.newQueryChat(). Select("chat_id", "title", "assistant_id"). - Where("sid", userID). + // Where("sid", userID). Where("chat_id", cid) row, err := qb.First() @@ -209,10 +209,10 @@ func (conv *Xun) GetChats(sid string, filter types.ChatFilter, locale ...string) // getChatsWithFilter get the chats with filter options func (conv *Xun) getChatsWithFilter(sid string, filter types.ChatFilter, locale ...string) (*types.ChatGroupResponse, error) { - userID, err := conv.getUserID(sid) - if err != nil { - return nil, err - } + // userID, err := conv.getUserID(sid) + // if err != nil { + // return nil, err + // } // Set default values if filter.Page <= 0 { @@ -226,8 +226,8 @@ func (conv *Xun) getChatsWithFilter(sid string, filter types.ChatFilter, locale } // Get total count - qbCount := conv.newQueryChat(). - Where("sid", userID) + qbCount := conv.newQueryChat() + // Where("sid", userID) // Apply silent filter if provided if filter.Silent != nil { @@ -257,8 +257,8 @@ func (conv *Xun) getChatsWithFilter(sid string, filter types.ChatFilter, locale // Get chats with pagination qb := conv.newQueryChat(). - Select("chat_id", "title", "assistant_id", "silent", "created_at", "updated_at"). - Where("sid", userID) + Select("chat_id", "title", "assistant_id", "silent", "created_at", "updated_at") + // Where("sid", userID) // Apply silent filter if provided if filter.Silent != nil { diff --git a/agent/store/xun/history.go b/agent/store/xun/history.go index 6d1a68d8..23b589da 100644 --- a/agent/store/xun/history.go +++ b/agent/store/xun/history.go @@ -13,14 +13,14 @@ import ( // GetHistory get the history func (conv *Xun) GetHistory(sid string, cid string, locale ...string) ([]map[string]interface{}, error) { - userID, err := conv.getUserID(sid) - if err != nil { - return nil, err - } + // userID, err := conv.getUserID(sid) + // if err != nil { + // return nil, err + // } qb := conv.newQuery(). Select("role", "name", "content", "context", "assistant_id", "assistant_name", "assistant_avatar", "mentions", "uid", "silent", "created_at", "updated_at"). - Where("sid", userID). + // Where("sid", userID). Where("cid", cid). OrderBy("id", "desc") From ab6fd8db09a356b1b6b2e62d9292d62914c99d09 Mon Sep 17 00:00:00 2001 From: Max Date: Sun, 23 Nov 2025 17:09:30 +0800 Subject: [PATCH 2/2] Refactor completion request validation and enhance stream handling for real-time output - Updated the completion request parsing logic to make the model field optional, allowing for better flexibility in handling requests. - Enhanced the stream end handling by adding JSON parsing for stream end data and sending a structured message to the frontend. - Modified the output writer to format data as Server-Sent Events (SSE) and ensured immediate flushing for real-time streaming in both CUI and OpenAI adapters. --- agent/context/openapi.go | 8 +++----- agent/llm/handlers/stream.go | 12 ++++++++++++ agent/output/adapters/cui/writer.go | 13 ++++++++++++- agent/output/adapters/openai/writer.go | 11 +++++++++++ 4 files changed, 38 insertions(+), 6 deletions(-) diff --git a/agent/context/openapi.go b/agent/context/openapi.go index 76bfb092..6e70e238 100644 --- a/agent/context/openapi.go +++ b/agent/context/openapi.go @@ -421,18 +421,16 @@ func parseCompletionRequestData(c *gin.Context) (*CompletionRequest, error) { } // If we got valid data from body, validate and return - if req.Model != "" && len(req.Messages) > 0 { + // Model is optional if assistant_id can be extracted later + if len(req.Messages) > 0 { return &req, nil } } } // Fallback: Try to parse from query parameters - // Required fields + // Model is optional (can be extracted from assistant_id) model := c.Query("model") - if model == "" { - return nil, fmt.Errorf("model field is required") - } req.Model = model // Messages (required, must be JSON string in query) diff --git a/agent/llm/handlers/stream.go b/agent/llm/handlers/stream.go index 58867659..5eb1b725 100644 --- a/agent/llm/handlers/stream.go +++ b/agent/llm/handlers/stream.go @@ -232,6 +232,18 @@ func (s *streamState) handleGroupEnd(data []byte) int { // handleStreamEnd handles stream end event func (s *streamState) handleStreamEnd(data []byte) int { + // Parse the stream end data + var endData context.StreamEndData + if err := jsoniter.Unmarshal(data, &endData); err != nil { + log.Error("Failed to parse stream_end data: %v", err) + output.Flush(s.ctx) + return 0 + } + + // Send stream_end event as a message to frontend + msg := output.NewEventMessage("stream_end", "Stream completed", endData) + output.Send(s.ctx, msg) + // Flush any remaining data output.Flush(s.ctx) return 0 // Continue (stream will end naturally) diff --git a/agent/output/adapters/cui/writer.go b/agent/output/adapters/cui/writer.go index a16e7723..cc173148 100644 --- a/agent/output/adapters/cui/writer.go +++ b/agent/output/adapters/cui/writer.go @@ -99,14 +99,25 @@ func (w *Writer) sendChunk(chunk interface{}) error { }) } + // Format as SSE (Server-Sent Events) format: "data: {json}\n\n" + sseData := []byte("data: ") + sseData = append(sseData, data...) + sseData = append(sseData, '\n', '\n') + // Send via context's writer // The context knows how to send data based on the connection type (SSE, WebSocket, etc.) - if err := w.ctx.Send(data); err != nil { + if err := w.ctx.Send(sseData); err != nil { if trace, _ := w.ctx.Trace(); trace != nil { trace.Error(i18n.T(w.ctx.Locale, "output.cui.writer.send_error"), map[string]any{"error": err.Error()}) // "CUI Writer: Failed to send data to client" } return err } + // Flush immediately to ensure real-time streaming + // Cast to http.ResponseWriter and call Flush if available + if flusher, ok := w.ctx.Writer.(interface{ Flush() }); ok { + flusher.Flush() + } + return nil } diff --git a/agent/output/adapters/openai/writer.go b/agent/output/adapters/openai/writer.go index 92e29a43..cbac69ec 100644 --- a/agent/output/adapters/openai/writer.go +++ b/agent/output/adapters/openai/writer.go @@ -157,6 +157,12 @@ func (w *Writer) sendChunk(chunk interface{}) error { return err } + // Flush immediately to ensure real-time streaming + // Cast to http.ResponseWriter and call Flush if available + if flusher, ok := w.ctx.Writer.(interface{ Flush() }); ok { + flusher.Flush() + } + return nil } @@ -176,5 +182,10 @@ func (w *Writer) sendDone() error { return err } + // Flush the final [DONE] message + if flusher, ok := w.ctx.Writer.(interface{ Flush() }); ok { + flusher.Flush() + } + return nil }