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.
This commit is contained in:
Max
parent
7ddea66106
commit
b43929210e
16 changed files with 380 additions and 38 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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")...)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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 ---")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue