From b43929210e7f4cbe167465f660c335871d3b2734 Mon Sep 17 00:00:00 2001 From: Max Date: Sun, 16 Nov 2025 10:30:55 +0800 Subject: [PATCH] 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. --- agent/assistant/agent.go | 29 +++++++-- agent/context/context_test.go | 4 +- agent/context/openapi.go | 13 ++-- agent/context/openapi_test.go | 64 ++++++++++++++++++ agent/llm/handlers/stream.go | 12 +++- agent/llm/providers/factory.go | 11 +++- agent/llm/providers/openai/openai.go | 76 ++++++++++++++++++++- agent/output/BUILTIN_TYPES.md | 87 +++++++++++++++++++++++-- agent/output/README.md | 5 +- agent/output/adapters/cui/adapter.go | 2 + agent/output/adapters/openai/adapter.go | 5 ++ agent/output/adapters/openai/writer.go | 28 ++++++-- agent/output/builtin.go | 14 +++- agent/output/message/types.go | 10 +++ agent/output/output.go | 4 +- openapi/chat/completions.go | 54 ++++++++++++--- 16 files changed, 380 insertions(+), 38 deletions(-) diff --git a/agent/assistant/agent.go b/agent/assistant/agent.go index 85e32ec3..1463ca5f 100644 --- a/agent/assistant/agent.go +++ b/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 - capabilities := &context.ModelCapabilities{} - + // Update capabilities based on connector settings if setting.Vision { v := true capabilities.Vision = &v diff --git a/agent/context/context_test.go b/agent/context/context_test.go index 5d52936e..0fd82779 100644 --- a/agent/context/context_test.go +++ b/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, }, diff --git a/agent/context/openapi.go b/agent/context/openapi.go index d4e54ef0..abba6b70 100644 --- a/agent/context/openapi.go +++ b/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 - userAgent := c.GetHeader("User-Agent") - clientType := getClientType(userAgent) - return parseAccept(clientType) + // Priority 4: Default to "standard" (OpenAI-compatible format) + return AcceptStandard + + // // 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 diff --git a/agent/context/openapi_test.go b/agent/context/openapi_test.go index 1c0b807b..89afa285 100644 --- a/agent/context/openapi_test.go +++ b/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) diff --git a/agent/llm/handlers/stream.go b/agent/llm/handlers/stream.go index 39189fb6..27d0931f 100644 --- a/agent/llm/handlers/stream.go +++ b/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 - msg := output.NewLoadingMessage("Connecting...") + // Send event message to indicate stream has started + // 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 } diff --git a/agent/llm/providers/factory.go b/agent/llm/providers/factory.go index e003b250..32635b58 100644 --- a/agent/llm/providers/factory.go +++ b/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 diff --git a/agent/llm/providers/openai/openai.go b/agent/llm/providers/openai/openai.go index 93917958..3d50aed6 100644 --- a/agent/llm/providers/openai/openai.go +++ b/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 { diff --git a/agent/output/BUILTIN_TYPES.md b/agent/output/BUILTIN_TYPES.md index c0d04ce1..9f5eb2ae 100644 --- a/agent/output/BUILTIN_TYPES.md +++ b/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. diff --git a/agent/output/README.md b/agent/output/README.md index 95c976d0..e10ced23 100644 --- a/agent/output/README.md +++ b/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 diff --git a/agent/output/adapters/cui/adapter.go b/agent/output/adapters/cui/adapter.go index fbcb16b5..3085af51 100644 --- a/agent/output/adapters/cui/adapter.go +++ b/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 } diff --git a/agent/output/adapters/openai/adapter.go b/agent/output/adapters/openai/adapter.go index 3e434f34..0a8e2613 100644 --- a/agent/output/adapters/openai/adapter.go +++ b/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 { diff --git a/agent/output/adapters/openai/writer.go b/agent/output/adapters/openai/writer.go index c570c027..4ef7657e 100644 --- a/agent/output/adapters/openai/writer.go +++ b/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" @@ -9,8 +10,9 @@ import ( // Writer implements the message.Writer interface for OpenAI-compatible clients type Writer struct { - ctx *context.Context - adapter *Adapter + ctx *context.Context + adapter *Adapter + firstChunk bool // Track if this is the first chunk to add role } // NewWriter creates a new OpenAI writer @@ -19,8 +21,9 @@ func NewWriter(ctx *context.Context) (*Writer, error) { adapter := NewAdapter() return &Writer{ - ctx: ctx, - adapter: adapter, + 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")...) diff --git a/agent/output/builtin.go b/agent/output/builtin.go index 0932e0b1..1db502e8 100644 --- a/agent/output/builtin.go +++ b/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 diff --git a/agent/output/message/types.go b/agent/output/message/types.go index 59064b97..28999910 100644 --- a/agent/output/message/types.go +++ b/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} diff --git a/agent/output/output.go b/agent/output/output.go index 65aa99f0..8d410815 100644 --- a/agent/output/output.go +++ b/agent/output/output.go @@ -85,8 +85,8 @@ func createWriter(ctx *context.Context) (message.Writer, error) { return cui.NewWriter(ctx) default: - // Default to CUI - return cui.NewWriter(ctx) + // Default to Standard + return openai.NewWriter(ctx) } } diff --git a/openapi/chat/completions.go b/openapi/chat/completions.go index c6bd23e7..78952261 100644 --- a/openapi/chat/completions.go +++ b/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{ - "message": "Create Completions", - "chat_id": ctx.ChatID, - "assistant_id": ctx.AssistantID, - "model": completionReq.Model, - "messages_count": len(completionReq.Messages), - }) + 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 ---")