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 }