Merge pull request #1314 from trheyi/main

Implement event message handling and enhance output capabilities

agent/assistant/agent.go

@@ -6,6 +6,7 @@ import (
 	"github.com/yaoapp/gou/connector"
 	"github.com/yaoapp/yao/agent/context"
 	"github.com/yaoapp/yao/agent/llm"
+	"github.com/yaoapp/yao/agent/output"
 	"github.com/yaoapp/yao/utils/jsonschema"
 )
 
@@ -100,6 +101,17 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa
 
 	_ = doneResponse // doneResponse is available for further processing
 
+	// Close the output writer to send [DONE] marker and flush data
+	if err := output.Close(ctx); err != nil {
+		// Log error but don't fail the request
+		fmt.Printf("Warning: Failed to close output writer: %v\n", err)
+	}
+
+	// Flush any remaining data to the client
+	if err := output.Flush(ctx); err != nil {
+		fmt.Printf("Warning: Failed to flush output: %v\n", err)
+	}
+
 	return &context.Response{Create: createResponse, Done: doneResponse, Completion: completionResponse}, nil
 }
 
@@ -132,15 +144,24 @@ func (ast *Assistant) GetConnector(ctx *context.Context) (connector.Connector, *
 
 // getConnectorCapabilities get the capabilities of a connector from settings
 func (ast *Assistant) getConnectorCapabilities(connectorID string) *context.ModelCapabilities {
+	// Initialize with default capabilities (all disabled)
+	falseVal := false
+	capabilities := &context.ModelCapabilities{
+		Vision:    &falseVal,
+		ToolCalls: &falseVal,
+		Audio:     &falseVal,
+		Reasoning: &falseVal,
+		Streaming: &falseVal,
+	}
+
 	// Get connector setting from global settings
 	setting, exists := connectorSettings[connectorID]
 	if !exists {
-		return nil
+		// Return default capabilities if connector not found in settings
+		return capabilities
 	}
 
-	// Convert ConnectorSetting to ModelCapabilities
+	// Update capabilities based on connector settings
-	capabilities := &context.ModelCapabilities{}
-
 	if setting.Vision {
 		v := true
 		capabilities.Vision = &v

agent/context/context_test.go

@@ -90,7 +90,7 @@ func TestGetCompletionRequest(t *testing.T) {
 			expectedLocale:      "fr-fr",
 			expectedTheme:       "auto",
 			expectedReferer:     RefererAPI,
-			expectedAccept:      AcceptWebCUI,
+			expectedAccept:      AcceptStandard,
 			expectedAssistantID: "test456",
 			expectError:         false,
 		},
@@ -132,7 +132,7 @@ func TestGetCompletionRequest(t *testing.T) {
 			expectedLocale:      "",
 			expectedTheme:       "",
 			expectedReferer:     RefererAPI,
-			expectedAccept:      AcceptWebCUI,
+			expectedAccept:      AcceptStandard,
 			expectedAssistantID: "minimal",
 			expectError:         false,
 		},

agent/context/openapi.go

@@ -254,7 +254,7 @@ func GetReferer(c *gin.Context, req *CompletionRequest) string {
 // 1. Query parameter "accept"
 // 2. Header "X-Yao-Accept"
 // 3. CompletionRequest metadata "accept" (from payload)
-// 4. Parse from client type (User-Agent)
+// 4. Default to "standard" (OpenAI-compatible format)
 func GetAccept(c *gin.Context, req *CompletionRequest) Accept {
 	// Priority 1: Query parameter
 	if accept := c.Query("accept"); accept != "" {
@@ -275,10 +275,13 @@ func GetAccept(c *gin.Context, req *CompletionRequest) Accept {
 		}
 	}
 
-	// Priority 4: Parse from User-Agent
+	// Priority 4: Default to "standard" (OpenAI-compatible format)
-	userAgent := c.GetHeader("User-Agent")
+	return AcceptStandard
-	clientType := getClientType(userAgent)
+
-	return parseAccept(clientType)
+	// // Future: Parse from User-Agent if needed
+	// userAgent := c.GetHeader("User-Agent")
+	// clientType := getClientType(userAgent)
+	// return parseAccept(clientType)
 }
 
 // GetChatID get the chat ID from the request

agent/context/openapi_test.go

@@ -493,6 +493,35 @@ func TestGetReferer_FromMetadata(t *testing.T) {
 	}
 }
 
+func TestGetAccept_FromQuery(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	req := httptest.NewRequest("GET", "/chat/completions?accept=cui-web", nil)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = req
+
+	accept := GetAccept(c, nil)
+	if accept != AcceptWebCUI {
+		t.Errorf("Expected accept 'cui-web' from query, got '%s'", accept)
+	}
+}
+
+func TestGetAccept_FromHeader(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	req := httptest.NewRequest("GET", "/chat/completions", nil)
+	req.Header.Set("X-Yao-Accept", "cui-desktop")
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = req
+
+	accept := GetAccept(c, nil)
+	if accept != AcceptDesktopCUI {
+		t.Errorf("Expected accept 'cui-desktop' from header, got '%s'", accept)
+	}
+}
+
 func TestGetAccept_FromMetadata(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 
@@ -513,6 +542,41 @@ func TestGetAccept_FromMetadata(t *testing.T) {
 	}
 }
 
+func TestGetAccept_Default(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	req := httptest.NewRequest("GET", "/chat/completions", nil)
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = req
+
+	accept := GetAccept(c, nil)
+	if accept != AcceptStandard {
+		t.Errorf("Expected default accept 'standard', got '%s'", accept)
+	}
+}
+
+func TestGetAccept_Priority(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	req := httptest.NewRequest("GET", "/chat/completions?accept=cui-web", nil)
+	req.Header.Set("X-Yao-Accept", "cui-desktop")
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = req
+
+	completionReq := &CompletionRequest{
+		Metadata: map[string]interface{}{
+			"accept": "cui-native",
+		},
+	}
+
+	accept := GetAccept(c, completionReq)
+	if accept != AcceptWebCUI {
+		t.Errorf("Expected query parameter to take priority, got '%s'", accept)
+	}
+}
+
 func TestGetAssistantID_FromModel(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 

agent/llm/handlers/stream.go

@@ -1,6 +1,8 @@
 package handlers
 
 import (
+	"fmt"
+
 	"github.com/yaoapp/yao/agent/context"
 	"github.com/yaoapp/yao/agent/output"
 	"github.com/yaoapp/yao/agent/output/message"
@@ -9,6 +11,7 @@ import (
 // DefaultStreamHandler creates a default stream handler that sends messages via context
 // This handler is used when no custom handler is provided
 func DefaultStreamHandler(ctx *context.Context) context.StreamFunc {
+
 	// Create stream state manager
 	state := &streamState{
 		ctx:       ctx,
@@ -17,6 +20,10 @@ func DefaultStreamHandler(ctx *context.Context) context.StreamFunc {
 	}
 
 	return func(chunkType context.StreamChunkType, data []byte) int {
+		fmt.Println("-----------------------------------------------")
+		fmt.Println("Chunk Type: ", string(chunkType))
+		fmt.Println("Data: ", string(data))
+		fmt.Println("-----------------------------------------------")
 		// Handle different chunk types
 		switch chunkType {
 		case context.ChunkStreamStart:
@@ -63,8 +70,9 @@ type streamState struct {
 
 // handleStreamStart handles stream start event
 func (s *streamState) handleStreamStart(data []byte) int {
-	// Send loading message to indicate stream has started
+	// Send event message to indicate stream has started
-	msg := output.NewLoadingMessage("Connecting...")
+	// This is a lifecycle event, CUI clients can show it, OpenAI clients will ignore it
+	msg := output.NewEventMessage("stream_start", "Connecting...", nil)
 	output.Send(s.ctx, msg)
 	return 0 // Continue
 }

agent/llm/providers/factory.go

@@ -18,12 +18,19 @@ type LLM interface {
 
 // SelectProvider select the appropriate provider based on connector and capabilities
 func SelectProvider(conn connector.Connector, options *context.CompletionOptions) (LLM, error) {
-	if options == nil || options.Capabilities == nil {
+
-		return nil, fmt.Errorf("options and capabilities are required")
+	if options == nil {
+		return nil, fmt.Errorf("options are required")
+	}
+
+	if options.Capabilities == nil {
+		return nil, fmt.Errorf("capabilities are required")
 	}
 
 	capabilities := options.Capabilities
 
+	// return openai.New(conn, capabilities), nil
+
 	// Priority 1: Reasoning models (special response format)
 	if capabilities.Reasoning != nil && *capabilities.Reasoning {
 		return reasoning.New(conn, capabilities), nil

agent/llm/providers/openai/openai.go

@@ -303,9 +303,13 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess
 			return http.HandlerReturnOk
 		}
 
+		// Log raw stream data for debugging
+		log.Trace("OpenAI Stream Raw Data: %s", string(data))
+
 		// Parse SSE data
 		dataStr := string(data)
 		if !strings.HasPrefix(dataStr, "data: ") {
+			log.Trace("Skipping non-SSE line: %s", dataStr)
 			return http.HandlerReturnOk
 		}
 
@@ -444,8 +448,65 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess
 		return http.HandlerReturnOk
 	}
 
+	// Log request for debugging
+	if requestBodyJSON, marshalErr := jsoniter.Marshal(requestBody); marshalErr == nil {
+		log.Debug("OpenAI Stream Request - URL: %s, Body: %s", url, string(requestBodyJSON))
+	}
+
+	// Buffer to capture non-SSE error responses
+	var errorBuffer strings.Builder
+	errorDetected := false
+
+	// Wrap streamHandler to detect JSON error responses
+	wrappedHandler := func(data []byte) int {
+		dataStr := string(data)
+
+		// Detect if this looks like a JSON error response (starts with "{" or contains "error")
+		if strings.Contains(dataStr, `"error"`) || (strings.TrimSpace(dataStr) == "{" && !errorDetected) {
+			errorDetected = true
+		}
+
+		// If error detected, accumulate all data for parsing
+		if errorDetected {
+			errorBuffer.Write(data)
+			errorBuffer.WriteString("\n")
+			return http.HandlerReturnOk
+		}
+
+		// Otherwise, use normal handler
+		return streamHandler(data)
+	}
+
 	// Make streaming request (goCtx already set at function start)
-	err = req.Stream(goCtx, "POST", requestBody, streamHandler)
+	err = req.Stream(goCtx, "POST", requestBody, wrappedHandler)
+
+	// Check if we captured an error response
+	if errorDetected && errorBuffer.Len() > 0 {
+		errorJSON := errorBuffer.String()
+		log.Error("OpenAI API returned error response: %s", errorJSON)
+
+		// Try to parse error
+		var apiError struct {
+			Error struct {
+				Message string `json:"message"`
+				Type    string `json:"type"`
+				Param   string `json:"param"`
+				Code    string `json:"code"`
+			} `json:"error"`
+		}
+
+		if parseErr := jsoniter.UnmarshalFromString(errorJSON, &apiError); parseErr == nil && apiError.Error.Message != "" {
+			err = fmt.Errorf("OpenAI API error: %s (type: %s, param: %s, code: %s)",
+				apiError.Error.Message, apiError.Error.Type, apiError.Error.Param, apiError.Error.Code)
+		} else {
+			err = fmt.Errorf("OpenAI API error: %s", strings.TrimSpace(errorJSON))
+		}
+	}
+
+	// Log any error from streaming
+	if err != nil {
+		log.Error("OpenAI Stream Error: %v", err)
+	}
 
 	// Check if error is due to context cancellation
 	if err != nil && goCtx.Err() != nil {
@@ -495,6 +556,14 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess
 	// Check if we received any data
 	if accumulator.id == "" {
 		log.Warn("OpenAI stream completed but no data was received (accumulator.id is empty)")
+
+		// Log request details for debugging
+		if requestBodyJSON, err := jsoniter.Marshal(requestBody); err == nil {
+			log.Error("Request body that caused empty response: %s", string(requestBodyJSON))
+		}
+		log.Error("Request URL: %s", url)
+		log.Error("Model in accumulator: %s, Created: %d", accumulator.model, accumulator.created)
+
 		err := fmt.Errorf("no data received from OpenAI API")
 
 		// End current group if active
@@ -821,10 +890,13 @@ func (p *Provider) buildRequestBody(messages []context.Message, options *context
 		body["temperature"] = *options.Temperature
 	}
 
+	// Use max_completion_tokens (modern API parameter for GPT-5+)
+	// GPT-5 models only support max_completion_tokens (not max_tokens)
 	if options.MaxCompletionTokens != nil {
 		body["max_completion_tokens"] = *options.MaxCompletionTokens
 	} else if options.MaxTokens != nil {
-		body["max_tokens"] = *options.MaxTokens
+		// Fallback: convert MaxTokens to max_completion_tokens for compatibility
+		body["max_completion_tokens"] = *options.MaxTokens
 	}
 
 	if options.TopP != nil {

agent/output/BUILTIN_TYPES.md

@@ -17,6 +17,7 @@ const (
     TypeAudio    = "audio"     // Audio content
     TypeVideo    = "video"     // Video content
     TypeAction   = "action"    // System action (silent in standard clients)
+    TypeEvent    = "event"     // Lifecycle event (silent in standard clients)
 )
 ```
 
@@ -294,7 +295,84 @@ output.Send(ctx, output.NewTextMessage("I've opened the user details panel for y
 
 ---
 
-### 7. Image (`image`)
+### 7. Event (`event`)
+
+**Purpose:** Lifecycle event messages (stream_start, stream_end, connecting, etc.)
+
+**Props Structure:**
+
+```go
+type EventProps struct {
+    Event   string                 `json:"event"`             // Event type
+    Message string                 `json:"message,omitempty"` // Human-readable message
+    Data    map[string]interface{} `json:"data,omitempty"`    // Additional event data
+}
+```
+
+**Example:**
+
+```json
+{
+  "type": "event",
+  "props": {
+    "event": "stream_start",
+    "message": "Starting stream...",
+    "data": {
+      "model": "gpt-4",
+      "session_id": "sess_123"
+    }
+  }
+}
+```
+
+**Helper:**
+
+```go
+msg := output.NewEventMessage("stream_start", "Starting stream...", map[string]interface{}{
+    "model": "gpt-4",
+    "session_id": "sess_123",
+})
+```
+
+**Use Cases:**
+
+- Stream lifecycle: `"stream_start"`, `"stream_end"`
+- Connection status: `"connecting"`, `"connected"`, `"disconnected"`
+- Processing stages: `"preprocessing"`, `"postprocessing"`
+- Agent state: `"thinking"`, `"executing"`, `"completed"`
+
+**Important Notes:**
+
+- **Silent in OpenAI clients**: Event messages are NOT sent to standard chat clients
+- **CUI clients only**: Only CUI clients process event messages
+- **Lifecycle tracking**: Used for tracking agent/stream lifecycle, not chat content
+- **Non-blocking**: Events don't interrupt the main message flow
+
+**Example in Hook:**
+
+```go
+// Send stream start event
+output.Send(ctx, output.NewEventMessage("stream_start", "Initializing...", map[string]interface{}{
+    "timestamp": time.Now().Unix(),
+}))
+
+// Do processing
+processData()
+
+// Send stream end event
+output.Send(ctx, output.NewEventMessage("stream_end", "Stream completed", map[string]interface{}{
+    "duration_ms": 1500,
+}))
+```
+
+**Result:**
+
+- **CUI client**: Tracks lifecycle, may show status indicators
+- **OpenAI client**: Events are silent (not sent to client)
+
+---
+
+### 8. Image (`image`)
 
 **Purpose:** Image content
 
@@ -337,7 +415,7 @@ msg := output.NewImageMessage("https://example.com/avatar.jpg", "User avatar")
 
 ---
 
-### 8. Audio (`audio`)
+### 9. Audio (`audio`)
 
 **Purpose:** Audio content
 
@@ -382,7 +460,7 @@ msg := output.NewAudioMessage("https://example.com/audio.mp3", "mp3")
 
 ---
 
-### 9. Video (`video`)
+### 10. Video (`video`)
 
 **Purpose:** Video content
 
@@ -463,6 +541,7 @@ OpenAI adapter converts built-in types to OpenAI format:
 | `audio`     | `delta.content`           | `props.url`                   | Markdown link (can't display inline)      |
 | `video`     | `delta.content`           | `props.url`                   | Markdown link (can't display inline)      |
 | `action`    | (not sent)                | -                             | Silent - system actions only              |
+| `event`     | (not sent)                | -                             | Silent - lifecycle events only            |
 
 ---
 
@@ -520,7 +599,7 @@ When adding new built-in types:
 **Do NOT add built-in types for:**
 
 - UI components (buttons, forms, etc.)
-- Rich media (images, videos, etc.)
 - Application-specific widgets
+- Domain-specific data types
 
 These should remain custom types.

agent/output/README.md

@@ -64,7 +64,7 @@ type Message struct {
 
 - **`Type`** (required): Determines how the message should be rendered
 
-  - Built-in types: `text`, `thinking`, `loading`, `tool_call`, `error`, `image`, `audio`, `video`, `action`
+  - Built-in types: `text`, `thinking`, `loading`, `tool_call`, `error`, `image`, `audio`, `video`, `action`, `event`
   - Custom types: Any string (frontend must have corresponding component)
 
 - **`Props`** (optional): Type-specific properties passed to the rendering component
@@ -323,7 +323,7 @@ Adapters handle the transformation automatically based on `ctx.Accept`.
 
 ### 3. Built-in Types
 
-9 standardized message types with defined Props structures:
+10 standardized message types with defined Props structures:
 
 | Type        | Purpose            | CUI     | OpenAI                    |
 | ----------- | ------------------ | ------- | ------------------------- |
@@ -336,6 +336,7 @@ Adapters handle the transformation automatically based on `ctx.Accept`.
 | `audio`     | Audio              | Player  | Link                      |
 | `video`     | Video              | Player  | Link                      |
 | `action`    | System commands    | Execute | Silent                    |
+| `event`     | Lifecycle events   | Track   | Silent                    |
 
 ## Usage
 

agent/output/adapters/cui/adapter.go

@@ -16,6 +16,8 @@ func NewAdapter() *Adapter {
 // For CUI, it simply returns the original message as a single chunk.
 func (a *Adapter) Adapt(msg *message.Message) ([]interface{}, error) {
 	// CUI clients consume the universal DSL directly, so no conversion is needed.
+	// This includes all message types like text, thinking, loading, events, etc.
+	// CUI clients can choose to display or ignore event messages.
 	return []interface{}{msg}, nil
 }
 

agent/output/adapters/openai/adapter.go

@@ -63,6 +63,11 @@ func WithConverter(msgType string, converter ConverterFunc) Option {
 
 // Adapt converts a universal Message to OpenAI-compatible format
 func (a *Adapter) Adapt(msg *message.Message) ([]interface{}, error) {
+	// Skip event messages - they are CUI-only lifecycle events
+	if msg.Type == message.TypeEvent {
+		return []interface{}{}, nil // Return empty array, nothing to send
+	}
+
 	// Get converter for this message type
 	converter, exists := a.registry.GetConverter(msg.Type)
 	if !exists {

agent/output/adapters/openai/writer.go

@@ -2,6 +2,7 @@ package openai
 
 import (
 	"encoding/json"
+	"fmt"
 
 	"github.com/yaoapp/yao/agent/context"
 	"github.com/yaoapp/yao/agent/output/message"
@@ -11,6 +12,7 @@ import (
 type Writer struct {
 	ctx        *context.Context
 	adapter    *Adapter
+	firstChunk bool // Track if this is the first chunk to add role
 }
 
 // NewWriter creates a new OpenAI writer
@@ -21,6 +23,7 @@ func NewWriter(ctx *context.Context) (*Writer, error) {
 	return &Writer{
 		ctx:        ctx,
 		adapter:    adapter,
+		firstChunk: true, // First chunk should include role
 	}, nil
 }
 
@@ -34,6 +37,18 @@ func (w *Writer) Write(msg *message.Message) error {
 
 	// Send each chunk
 	for _, chunk := range chunks {
+		// Add role to first text chunk
+		if w.firstChunk && (msg.Type == message.TypeText || msg.Type == message.TypeThinking) {
+			if chunkMap, ok := chunk.(map[string]interface{}); ok {
+				if choices, ok := chunkMap["choices"].([]map[string]interface{}); ok && len(choices) > 0 {
+					if delta, ok := choices[0]["delta"].(map[string]interface{}); ok {
+						delta["role"] = "assistant"
+						w.firstChunk = false
+					}
+				}
+			}
+		}
+
 		if err := w.sendChunk(chunk); err != nil {
 			return err
 		}
@@ -76,6 +91,11 @@ func (w *Writer) sendChunk(chunk interface{}) error {
 		return err
 	}
 
+	// Debug: print the chunk being sent
+	fmt.Println("-----------------------------------------------")
+	fmt.Println("Sending SSE chunk: ", string(data))
+	fmt.Println("-----------------------------------------------")
+
 	// Format as SSE: "data: {json}\n\n"
 	sseData := append([]byte("data: "), data...)
 	sseData = append(sseData, []byte("\n\n")...)

agent/output/builtin.go

@@ -78,6 +78,18 @@ func NewActionMessage(name string, payload map[string]interface{}) *message.Mess
 	}
 }
 
+// NewEventMessage creates an event message
+func NewEventMessage(event string, msg string, data map[string]interface{}) *message.Message {
+	return &message.Message{
+		Type: message.TypeEvent,
+		Props: map[string]interface{}{
+			"event":   event,
+			"message": msg,
+			"data":    data,
+		},
+	}
+}
+
 // NewImageMessage creates an image message
 func NewImageMessage(url string, alt string) *message.Message {
 	return &message.Message{
@@ -113,7 +125,7 @@ func NewVideoMessage(url string) *message.Message {
 // IsBuiltinType checks if a message type is a built-in type
 func IsBuiltinType(msgType string) bool {
 	switch msgType {
-	case message.TypeText, message.TypeThinking, message.TypeLoading, message.TypeToolCall, message.TypeError, message.TypeImage, message.TypeAudio, message.TypeVideo, message.TypeAction:
+	case message.TypeText, message.TypeThinking, message.TypeLoading, message.TypeToolCall, message.TypeError, message.TypeImage, message.TypeAudio, message.TypeVideo, message.TypeAction, message.TypeEvent:
 		return true
 	default:
 		return false

agent/output/message/types.go

@@ -59,6 +59,7 @@ const (
 
 	// System types (not visible in standard chat clients)
 	TypeAction = "action" // System action (open panel, navigate, etc.) - silent in OpenAI clients
+	TypeEvent  = "event"  // Lifecycle event (stream_start, stream_end, etc.) - CUI only, silent in OpenAI clients
 )
 
 // Standard Props structures for built-in types
@@ -110,6 +111,15 @@ type ActionProps struct {
 	Payload map[string]interface{} `json:"payload,omitempty"` // Action payload/parameters
 }
 
+// EventProps defines the standard structure for event messages
+// Type: "event"
+// Props: {"event": string, "message": string, "data": map}
+type EventProps struct {
+	Event   string                 `json:"event"`             // Event type (e.g., "stream_start", "stream_end", "connecting")
+	Message string                 `json:"message,omitempty"` // Human-readable message (e.g., "Connecting...")
+	Data    map[string]interface{} `json:"data,omitempty"`    // Additional event data
+}
+
 // ImageProps defines the standard structure for image messages
 // Type: "image"
 // Props: {"url": string, "alt": string, "width": int, "height": int, "detail": string}

agent/output/output.go

@@ -85,8 +85,8 @@ func createWriter(ctx *context.Context) (message.Writer, error) {
 		return cui.NewWriter(ctx)
 
 	default:
-		// Default to CUI
+		// Default to Standard
-		return cui.NewWriter(ctx)
+		return openai.NewWriter(ctx)
 	}
 }
 

openapi/chat/completions.go

@@ -4,7 +4,9 @@ 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"
 )
@@ -24,7 +26,6 @@ func GinCreateCompletions(c *gin.Context) {
 
 	completionReq, ctx, err := context.GetCompletionRequest(c, cache)
 	if err != nil {
-
 		fmt.Println("-----------------------------------------------")
 		fmt.Println("Error: ", err.Error())
 		fmt.Println("-----------------------------------------------")
@@ -38,6 +39,7 @@ func GinCreateCompletions(c *gin.Context) {
 
 	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)
@@ -54,13 +56,49 @@ func GinCreateCompletions(c *gin.Context) {
 	}
 	fmt.Println("-----------------------------------------------")
 
-	c.JSON(response.StatusOK, gin.H{
+	ast, err := assistant.Get(ctx.AssistantID)
-		"message":        "Create Completions",
+	if err != nil {
-		"chat_id":        ctx.ChatID,
+		response.RespondWithError(c, response.StatusInternalServerError, &response.ErrorResponse{
-		"assistant_id":   ctx.AssistantID,
+			Code:             response.ErrServerError.Code,
-		"model":          completionReq.Model,
+			ErrorDescription: "Failed to get assistant: " + err.Error(),
-		"messages_count": len(completionReq.Messages),
 		})
+		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 ---")