diff --git a/agent/context/context.go b/agent/context/context.go index dd071a2c..eda88171 100644 --- a/agent/context/context.go +++ b/agent/context/context.go @@ -87,6 +87,17 @@ func (ctx *Context) Release() { ctx = nil } +// Send sends data to the context's writer +// This is used by the output module to send messages to the client +func (ctx *Context) Send(data []byte) error { + if ctx.Writer == nil { + return nil // No writer, silently ignore + } + + _, err := ctx.Writer.Write(data) + return err +} + // Map the context to a map func (ctx *Context) Map() map[string]interface{} { data := map[string]interface{}{} diff --git a/agent/llm/handlers/stream.go b/agent/llm/handlers/stream.go index 5e1be574..39189fb6 100644 --- a/agent/llm/handlers/stream.go +++ b/agent/llm/handlers/stream.go @@ -2,62 +2,215 @@ package handlers import ( "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output" + "github.com/yaoapp/yao/agent/output/message" ) // 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, + inGroup: false, + currentID: "", + } + return func(chunkType context.StreamChunkType, data []byte) int { - // TODO: Implement default stream handling - // - Parse streaming chunk data based on chunkType - // - Extract content from chunk - // - Send message via ctx (SSE, WebSocket, etc.) - // - Handle different chunk types (text, thinking, tool_calls, etc.) - // - Return 0 to continue streaming, non-zero to stop - + // Handle different chunk types switch chunkType { - case context.ChunkText: - // Handle text content - case context.ChunkThinking: - // Handle reasoning/thinking content - case context.ChunkToolCall: - // Handle tool calls - case context.ChunkMetadata: - // Handle metadata (usage, finish_reason) - case context.ChunkError: - // Handle error - return 1 // Stop on error - } + case context.ChunkStreamStart: + return state.handleStreamStart(data) - return 0 // Continue streaming + case context.ChunkGroupStart: + return state.handleGroupStart(data) + + case context.ChunkText: + return state.handleText(data) + + case context.ChunkThinking: + return state.handleThinking(data) + + case context.ChunkToolCall: + return state.handleToolCall(data) + + case context.ChunkMetadata: + return state.handleMetadata(data) + + case context.ChunkError: + return state.handleError(data) + + case context.ChunkGroupEnd: + return state.handleGroupEnd(data) + + case context.ChunkStreamEnd: + return state.handleStreamEnd(data) + + default: + // Unknown chunk type, continue + return 0 + } } } -// SendStreamChunk sends a stream chunk via context -// Used internally by DefaultStreamHandler -func SendStreamChunk(ctx *context.Context, chunkType context.StreamChunkType, data []byte) error { - // TODO: Implement sending stream chunk - // - Format chunk for transport (SSE, WebSocket) - // - Send via ctx's connection - // - Handle errors and retries - return nil +// streamState manages the state of the streaming process +type streamState struct { + ctx *context.Context + inGroup bool + currentID string + buffer []byte } -// FormatSSE formats streaming data as Server-Sent Events format -func FormatSSE(chunkType context.StreamChunkType, data []byte) string { - // TODO: Implement SSE formatting - // - Format as "data: {...}\n\n" - // - Include chunk type in the message - // - Handle special cases (done, error) - // - Ensure proper JSON encoding - return "" +// handleStreamStart handles stream start event +func (s *streamState) handleStreamStart(data []byte) int { + // Send loading message to indicate stream has started + msg := output.NewLoadingMessage("Connecting...") + output.Send(s.ctx, msg) + return 0 // Continue } -// FormatWebSocket formats streaming data as WebSocket message -func FormatWebSocket(chunkType context.StreamChunkType, data []byte) []byte { - // TODO: Implement WebSocket formatting - // - Format as JSON message with chunk type - // - Add message type/metadata - // - Handle binary vs text frames - return nil +// handleGroupStart handles group start event +func (s *streamState) handleGroupStart(data []byte) int { + s.inGroup = true + s.currentID = generateMessageID() + s.buffer = []byte{} + return 0 // Continue +} + +// handleText handles text content chunks +func (s *streamState) handleText(data []byte) int { + if len(data) == 0 { + return 0 + } + + // Ensure we have a message ID + if s.currentID == "" { + s.currentID = generateMessageID() + } + + // Append to buffer + s.buffer = append(s.buffer, data...) + + // Send delta message + msg := &message.Message{ + ID: s.currentID, + Type: message.TypeText, + Delta: true, + Props: map[string]interface{}{ + "content": string(data), + }, + } + + if err := output.Send(s.ctx, msg); err != nil { + // Log error but continue streaming + return 0 + } + + return 0 // Continue +} + +// handleThinking handles thinking/reasoning chunks +func (s *streamState) handleThinking(data []byte) int { + if len(data) == 0 { + return 0 + } + + // Ensure we have a message ID + if s.currentID == "" { + s.currentID = generateMessageID() + } + + // Append to buffer + s.buffer = append(s.buffer, data...) + + // Send delta message + msg := &message.Message{ + ID: s.currentID, + Type: message.TypeThinking, + Delta: true, + Props: map[string]interface{}{ + "content": string(data), + }, + } + + if err := output.Send(s.ctx, msg); err != nil { + return 0 + } + + return 0 // Continue +} + +// handleToolCall handles tool call chunks +func (s *streamState) handleToolCall(data []byte) int { + // Tool calls are usually complete JSON objects + // Parse and send as tool_call message + msg := &message.Message{ + ID: generateMessageID(), + Type: message.TypeToolCall, + Delta: true, + Props: map[string]interface{}{ + // TODO: Parse tool call data + "raw": string(data), + }, + } + + output.Send(s.ctx, msg) + return 0 // Continue +} + +// handleMetadata handles metadata chunks (usage, finish_reason, etc.) +func (s *streamState) handleMetadata(data []byte) int { + // Metadata is usually not displayed to users + // Could be logged or stored for analytics + return 0 // Continue +} + +// handleError handles error chunks +func (s *streamState) handleError(data []byte) int { + // Send error message + msg := output.NewErrorMessage(string(data), "stream_error") + output.Send(s.ctx, msg) + + return 1 // Stop streaming on error +} + +// handleGroupEnd handles group end event +func (s *streamState) handleGroupEnd(data []byte) int { + if !s.inGroup { + return 0 + } + + // Send done message with complete content + if s.currentID != "" && len(s.buffer) > 0 { + msg := &message.Message{ + ID: s.currentID, + Type: message.TypeText, // Default to text + Done: true, + Props: map[string]interface{}{ + "content": string(s.buffer), + }, + } + output.Send(s.ctx, msg) + } + + // Reset state + s.inGroup = false + s.currentID = "" + s.buffer = []byte{} + + return 0 // Continue +} + +// handleStreamEnd handles stream end event +func (s *streamState) handleStreamEnd(data []byte) int { + // Flush any remaining data + output.Flush(s.ctx) + return 0 // Continue (stream will end naturally) +} + +// generateMessageID generates a unique message ID +func generateMessageID() string { + // TODO: Implement proper ID generation + // For now, use a simple approach + return output.GenerateID() } diff --git a/agent/output/BUILTIN_TYPES.md b/agent/output/BUILTIN_TYPES.md new file mode 100644 index 00000000..c0d04ce1 --- /dev/null +++ b/agent/output/BUILTIN_TYPES.md @@ -0,0 +1,526 @@ +# Built-in Message Types + +Built-in message types are standardized types that all adapters must support. These types have predefined Props structures to ensure consistency across different output formats. + +## Type Constants + +Defined in `types.go`: + +```go +const ( + TypeText = "text" // Plain text or Markdown content + TypeThinking = "thinking" // Reasoning/thinking process + TypeLoading = "loading" // Loading/processing indicator + TypeToolCall = "tool_call" // LLM tool/function call + TypeError = "error" // Error message + TypeImage = "image" // Image content + TypeAudio = "audio" // Audio content + TypeVideo = "video" // Video content + TypeAction = "action" // System action (silent in standard clients) +) +``` + +## Standard Props Structures + +### 1. Text (`text`) + +**Purpose:** Plain text or Markdown content + +**Props Structure:** + +```go +type TextProps struct { + Content string `json:"content"` // Text content (supports Markdown) +} +``` + +**Example:** + +```json +{ + "type": "text", + "props": { + "content": "Hello **world**!" + } +} +``` + +**Helper:** + +```go +msg := output.NewTextMessage("Hello **world**!") +``` + +--- + +### 2. Thinking (`thinking`) + +**Purpose:** Reasoning or thinking process (used by o1 models, DeepSeek R1, etc.) + +**Props Structure:** + +```go +type ThinkingProps struct { + Content string `json:"content"` // Reasoning/thinking content +} +``` + +**Example:** + +```json +{ + "type": "thinking", + "props": { + "content": "Let me analyze this step by step..." + } +} +``` + +**Helper:** + +```go +msg := output.NewThinkingMessage("Let me analyze this step by step...") +``` + +--- + +### 3. Loading (`loading`) + +**Purpose:** Loading or processing indicator (preprocessing, knowledge base search, data fetching, etc.) + +**Props Structure:** + +```go +type LoadingProps struct { + Message string `json:"message"` // Loading message +} +``` + +**Example:** + +```json +{ + "type": "loading", + "props": { + "message": "Searching knowledge base..." + } +} +``` + +**Helper:** + +```go +msg := output.NewLoadingMessage("Searching knowledge base...") +``` + +**Use Cases:** + +- Knowledge base search: `"Searching knowledge base..."` +- Data preprocessing: `"Processing uploaded file..."` +- External API calls: `"Fetching data from API..."` +- Database queries: `"Querying database..."` + +**Example in Hook:** + +```go +// In Create hook, show preprocessing steps +func Create(ctx *context.Context, messages []context.Message) (*context.HookCreateResponse, error) { + // Send loading message for knowledge base search + output.Send(ctx, output.NewLoadingMessage("Searching knowledge base...")) + + // Do the actual search + results := searchKnowledgeBase(messages) + + // Send another loading message for processing + output.Send(ctx, output.NewLoadingMessage("Processing results...")) + + // Process and return + return &context.HookCreateResponse{ + Messages: buildMessages(results), + }, nil +} +``` + +**Result in OpenAI Client:** + +- Shows as thinking/reasoning process +- User sees "Searching knowledge base..." and "Processing results..." +- Provides transparency into what's happening + +--- + +### 4. Tool Call (`tool_call`) + +**Purpose:** LLM tool or function call + +**Props Structure:** + +```go +type ToolCallProps struct { + ID string `json:"id"` // Tool call ID + Name string `json:"name"` // Function/tool name + Arguments string `json:"arguments,omitempty"` // JSON string of arguments +} +``` + +**Example:** + +```json +{ + "type": "tool_call", + "props": { + "id": "call_abc123", + "name": "get_weather", + "arguments": "{\"location\": \"San Francisco\"}" + } +} +``` + +**Helper:** + +```go +msg := output.NewToolCallMessage( + "call_abc123", + "get_weather", + "{\"location\": \"San Francisco\"}", +) +``` + +--- + +### 5. Error (`error`) + +**Purpose:** Error message + +**Props Structure:** + +```go +type ErrorProps struct { + Message string `json:"message"` // Error message + Code string `json:"code,omitempty"` // Error code + Details string `json:"details,omitempty"` // Additional error details +} +``` + +**Example:** + +```json +{ + "type": "error", + "props": { + "message": "Connection timeout", + "code": "TIMEOUT", + "details": "Failed to connect to database after 30s" + } +} +``` + +**Helper:** + +```go +msg := output.NewErrorMessage("Connection timeout", "TIMEOUT") +``` + +--- + +### 6. Action (`action`) + +**Purpose:** System-level action/command (not displayed to user, only processed by client) + +**Props Structure:** + +```go +type ActionProps struct { + Name string `json:"name"` // Action name + Payload map[string]interface{} `json:"payload,omitempty"` // Action parameters +} +``` + +**Example:** + +```json +{ + "type": "action", + "props": { + "name": "open_panel", + "payload": { + "panel_id": "user_profile", + "user_id": "123" + } + } +} +``` + +**Helper:** + +```go +msg := output.NewActionMessage("open_panel", map[string]interface{}{ + "panel_id": "user_profile", + "user_id": "123", +}) +``` + +**Use Cases:** + +- Open sidebar/panel: `"open_panel"` +- Navigate to page: `"navigate"` +- Trigger UI update: `"refresh_view"` +- Close modal: `"close_modal"` +- Scroll to element: `"scroll_to"` + +**Important Notes:** + +- **Silent in OpenAI clients**: Action messages are NOT sent to standard chat clients +- **CUI clients only**: Only CUI clients process action messages +- **System-level**: Used for controlling the UI/application, not chat content + +**Example in Hook:** + +```go +// Send action to open a panel with user details +output.Send(ctx, output.NewActionMessage("open_panel", map[string]interface{}{ + "panel_id": "user_details", + "user_id": user.ID, +})) + +// Send text message (visible to user) +output.Send(ctx, output.NewTextMessage("I've opened the user details panel for you.")) +``` + +**Result:** + +- **CUI client**: Panel opens, text message displays +- **OpenAI client**: Only text message displays (action is silent) + +--- + +### 7. Image (`image`) + +**Purpose:** Image content + +**Props Structure:** + +```go +type ImageProps struct { + URL string // Required: Image URL or base64 data + Alt string // Alternative text + Width int // Image width in pixels + Height int // Image height in pixels + Detail string // OpenAI detail level: "auto", "low", "high" +} +``` + +**Example:** + +```json +{ + "type": "image", + "props": { + "url": "https://example.com/avatar.jpg", + "alt": "User avatar", + "width": 200, + "height": 200 + } +} +``` + +**Helper:** + +```go +msg := output.NewImageMessage("https://example.com/avatar.jpg", "User avatar") +``` + +**Adapter Behavior:** + +- **CUI**: Renders image directly with `` tag +- **OpenAI**: Converts to Markdown `![alt](url)` - **displays inline** in Markdown-supporting clients + +--- + +### 8. Audio (`audio`) + +**Purpose:** Audio content + +**Props Structure:** + +```go +type AudioProps struct { + URL string // Required: Audio URL or base64 data + Format string // Audio format: "mp3", "wav", "ogg" + Duration float64 // Duration in seconds + Transcript string // Audio transcript text + Autoplay bool // Whether to autoplay + Controls bool // Whether to show controls +} +``` + +**Example:** + +```json +{ + "type": "audio", + "props": { + "url": "https://example.com/audio.mp3", + "format": "mp3", + "duration": 120.5, + "transcript": "This is the audio content...", + "controls": true + } +} +``` + +**Helper:** + +```go +msg := output.NewAudioMessage("https://example.com/audio.mp3", "mp3") +``` + +**Adapter Behavior:** + +- **CUI**: Renders audio player with controls +- **OpenAI**: Converts to link `🔊 [Play Audio](url)` - can't display inline + +--- + +### 9. Video (`video`) + +**Purpose:** Video content + +**Props Structure:** + +```go +type VideoProps struct { + URL string // Required: Video URL + Format string // Video format: "mp4", "webm" + Duration float64 // Duration in seconds + Thumbnail string // Thumbnail/poster image URL + Width int // Video width in pixels + Height int // Video height in pixels + Autoplay bool // Whether to autoplay + Controls bool // Whether to show controls + Loop bool // Whether to loop +} +``` + +**Example:** + +```json +{ + "type": "video", + "props": { + "url": "https://example.com/video.mp4", + "format": "mp4", + "thumbnail": "https://example.com/poster.jpg", + "width": 640, + "height": 360, + "controls": true + } +} +``` + +**Helper:** + +```go +msg := output.NewVideoMessage("https://example.com/video.mp4") +``` + +**Adapter Behavior:** + +- **CUI**: Renders video player with controls +- **OpenAI**: Converts to link `🎬 [Watch Video](url)` - can't display inline + +--- + +## Adapter Requirements + +All adapters (CUI, OpenAI, etc.) **must** support these built-in types with their standard Props structures. + +### CUI Adapter + +CUI adapter passes built-in types through without transformation: + +```json +{ + "type": "text", + "props": { + "content": "Hello world" + } +} +``` + +### OpenAI Adapter + +OpenAI adapter converts built-in types to OpenAI format: + +| Type | OpenAI Format | Field | Note | +| ----------- | ------------------------- | ----------------------------- | ----------------------------------------- | +| `text` | `delta.content` | `props.content` | | +| `thinking` | `delta.reasoning_content` | `props.content` | Reasoning content (o1 models) | +| `loading` | `delta.reasoning_content` | `props.message` | Shows as thinking in OpenAI clients | +| `tool_call` | `delta.tool_calls` | `props.{id, name, arguments}` | | +| `error` | `error` | `props.{message, code}` | | +| `image` | `delta.content` | `props.{url, alt}` | Markdown: `![alt](url)` - displays inline | +| `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 | + +--- + +## Custom Types + +Any type **not** in the built-in list is considered a custom type. Adapters may handle custom types differently: + +- **CUI:** Pass through as-is +- **OpenAI:** Convert to Markdown link + +Example custom type: + +```json +{ + "type": "image", + "props": { + "url": "https://example.com/image.jpg", + "alt": "Description" + } +} +``` + +--- + +## Checking Built-in Types + +```go +// Check if a type is built-in +if output.IsBuiltinType(msg.Type) { + // Handle as standard type +} else { + // Handle as custom type +} +``` + +--- + +## Guidelines for New Built-in Types + +When adding new built-in types: + +1. ✅ Add constant to `types.go` +2. ✅ Define Props structure +3. ✅ Add helper function in `builtin.go` +4. ✅ Update all adapters to support it +5. ✅ Document in this file +6. ✅ Add tests + +**Only add built-in types for:** + +- Universal concepts (text, errors, etc.) +- LLM-specific features (thinking, tool_calls) +- Types that need cross-adapter consistency + +**Do NOT add built-in types for:** + +- UI components (buttons, forms, etc.) +- Rich media (images, videos, etc.) +- Application-specific widgets + +These should remain custom types. diff --git a/agent/output/README.md b/agent/output/README.md new file mode 100644 index 00000000..95c976d0 --- /dev/null +++ b/agent/output/README.md @@ -0,0 +1,445 @@ +# Output Module + +The output module provides a unified API for sending messages to different client types (CUI, OpenAI-compatible, etc.) with support for streaming and rich media content. + +## Architecture + +``` +agent/output/ +├── message/ # Core types and interfaces (no dependencies) +│ ├── types.go # Message, MessageGroup, Props structures +│ └── interfaces.go # Writer, Adapter, Factory interfaces +├── adapters/ # Client-specific adapters +│ ├── cui/ # CUI adapter (native DSL) +│ │ ├── adapter.go +│ │ └── writer.go +│ └── openai/ # OpenAI adapter (converts to OpenAI format) +│ ├── adapter.go +│ ├── converter.go +│ ├── writer.go +│ ├── types.go +│ └── factory.go +├── output.go # Main API (Send, GetWriter, etc.) +├── builtin.go # Helper functions for built-in types +└── BUILTIN_TYPES.md # Documentation for built-in types +``` + +## DSL Structure + +### Message Structure + +The universal message DSL is a JSON structure that supports streaming, rich media, and incremental updates: + +```go +type Message struct { + // Core fields + Type string `json:"type"` // Message type (e.g., "text", "image", "action") + Props map[string]interface{} `json:"props,omitempty"` // Type-specific properties + + // Streaming control + ID string `json:"id,omitempty"` // Unique message ID (for merging in streaming) + Delta bool `json:"delta,omitempty"` // Whether this is an incremental update + Done bool `json:"done,omitempty"` // Whether the message is complete + + // Delta update control (for incremental props updates) + DeltaPath string `json:"delta_path,omitempty"` // Which field to update (e.g., "content", "items.0.name") + DeltaAction string `json:"delta_action,omitempty"` // How to update ("append", "replace", "merge", "set") + + // Type correction (for streaming type inference) + TypeChange bool `json:"type_change,omitempty"` // Marks this as a type correction message + + // Message grouping (for semantically related messages) + GroupID string `json:"group_id,omitempty"` // Parent message group ID + GroupStart bool `json:"group_start,omitempty"` // Marks the start of a group + GroupEnd bool `json:"group_end,omitempty"` // Marks the end of a group + + // Metadata + Metadata *Metadata `json:"metadata,omitempty"` // Timestamp, sequence, trace ID +} +``` + +### Field Descriptions + +#### Core Fields + +- **`Type`** (required): Determines how the message should be rendered + + - Built-in types: `text`, `thinking`, `loading`, `tool_call`, `error`, `image`, `audio`, `video`, `action` + - Custom types: Any string (frontend must have corresponding component) + +- **`Props`** (optional): Type-specific properties passed to the rendering component + - For `text`: `{"content": "Hello"}` + - For `image`: `{"url": "...", "alt": "..."}` + - For custom types: Any JSON-serializable data + +#### Streaming Control + +- **`ID`** (optional): Unique identifier for message tracking + + - Used to merge multiple delta updates into a single message + - Auto-generated if not provided + - Example: `"msg_1234567890_9876543210"` + +- **`Delta`** (optional): Marks this as an incremental update + + - `true`: Append/update to existing message with same ID + - `false`: Complete message (default) + - Used for streaming LLM responses + +- **`Done`** (optional): Marks message as complete + - `true`: No more updates will come for this message ID + - `false`: More updates may follow + - Typically sent as final message in a delta sequence + +#### Delta Update Control + +For complex, structured messages that need field-level updates: + +- **`DeltaPath`** (optional): JSON path to the field being updated + + - Simple: `"content"` (updates `props.content`) + - Nested: `"user.name"` (updates `props.user.name`) + - Array: `"items.0.title"` (updates `props.items[0].title`) + +- **`DeltaAction`** (optional): How to apply the delta update + - `"append"`: Concatenate to existing string/array + - `"replace"`: Replace entire value + - `"merge"`: Merge objects (shallow merge) + - `"set"`: Set new field (if doesn't exist) + +#### Type Correction + +- **`TypeChange`** (optional): Indicates message type was corrected + - Used when initial type inference was wrong + - Frontend should re-render with new type + - Example: Initially sent as `text`, corrected to `thinking` + +#### Message Grouping + +For grouping semantically related messages (e.g., image + caption): + +- **`GroupID`** (optional): Identifier for the message group +- **`GroupStart`** (optional): Marks the beginning of a group +- **`GroupEnd`** (optional): Marks the end of a group + +#### Metadata + +- **`Metadata`** (optional): Additional message metadata + ```go + type Metadata struct { + Timestamp int64 // Unix nanoseconds + Sequence int // Message sequence number + TraceID string // For debugging/logging + } + ``` + +### Message Examples + +#### Simple Text Message + +```json +{ + "type": "text", + "props": { + "content": "Hello, world!" + } +} +``` + +#### Streaming Text (Delta Updates) + +```json +// First chunk +{ + "id": "msg_123", + "type": "text", + "delta": true, + "props": { + "content": "Hello" + } +} + +// Second chunk (appends) +{ + "id": "msg_123", + "type": "text", + "delta": true, + "props": { + "content": ", world" + } +} + +// Final chunk (marks done) +{ + "id": "msg_123", + "type": "text", + "delta": true, + "done": true, + "props": { + "content": "!" + } +} +``` + +#### Complex Type with Nested Updates + +```json +// Initial message +{ + "id": "msg_456", + "type": "table", + "props": { + "columns": ["Name", "Age"], + "rows": [] + } +} + +// Add first row +{ + "id": "msg_456", + "type": "table", + "delta": true, + "delta_path": "rows", + "delta_action": "append", + "props": { + "rows": [{"name": "Alice", "age": 30}] + } +} + +// Add second row +{ + "id": "msg_456", + "type": "table", + "delta": true, + "delta_path": "rows", + "delta_action": "append", + "props": { + "rows": [{"name": "Bob", "age": 25}] + } +} +``` + +#### Type Correction + +```json +// Initial guess (text) +{ + "id": "msg_789", + "type": "text", + "delta": true, + "props": { + "content": "Let me think..." + } +} + +// Correction (actually thinking) +{ + "id": "msg_789", + "type": "thinking", + "type_change": true, + "props": { + "content": "Let me think..." + } +} +``` + +#### Message Group + +```json +// Group start +{ + "group_id": "grp_001", + "group_start": true +} + +// Image in group +{ + "type": "image", + "group_id": "grp_001", + "props": { + "url": "https://example.com/photo.jpg", + "alt": "Beautiful sunset" + } +} + +// Caption in group +{ + "type": "text", + "group_id": "grp_001", + "props": { + "content": "Captured at Golden Gate Bridge" + } +} + +// Group end +{ + "group_id": "grp_001", + "group_end": true +} +``` + +## Key Design Decisions + +### 1. Separate `message` Package + +To avoid circular dependencies, all core types and interfaces are defined in the `message` sub-package: + +- `message.Message` - Universal message DSL +- `message.Writer` - Interface for writing messages +- `message.Adapter` - Interface for format conversion + +This allows: + +- `handlers` → `output` → `message` ✅ +- `output/adapters` → `message` ✅ +- No circular dependencies! + +### 2. Adapter Pattern + +Different clients require different formats: + +**CUI Clients:** + +```json +{ + "type": "text", + "props": { "content": "Hello" } +} +``` + +**OpenAI Clients:** + +```json +{ + "choices": [ + { + "delta": { "content": "Hello" } + } + ] +} +``` + +Adapters handle the transformation automatically based on `ctx.Accept`. + +### 3. Built-in Types + +9 standardized message types with defined Props structures: + +| Type | Purpose | CUI | OpenAI | +| ----------- | ------------------ | ------- | ------------------------- | +| `text` | Text content | Direct | `delta.content` | +| `thinking` | LLM reasoning | Direct | `delta.reasoning_content` | +| `loading` | Progress indicator | Direct | `delta.reasoning_content` | +| `tool_call` | Function calls | Direct | `delta.tool_calls` | +| `error` | Error messages | Direct | `error` | +| `image` | Images | Render | `![](url)` markdown | +| `audio` | Audio | Player | Link | +| `video` | Video | Player | Link | +| `action` | System commands | Execute | Silent | + +## Usage + +### Basic Usage + +```go +import ( + "github.com/yaoapp/yao/agent/output" + "github.com/yaoapp/yao/agent/output/message" +) + +// Send a text message +msg := output.NewTextMessage("Hello world") +output.Send(ctx, msg) + +// Send a loading indicator +loading := output.NewLoadingMessage("Searching knowledge base...") +output.Send(ctx, loading) + +// Send an image +img := output.NewImageMessage("https://example.com/image.jpg", "Description") +output.Send(ctx, img) + +// Send an error +err := output.NewErrorMessage("Connection failed", "TIMEOUT") +output.Send(ctx, err) +``` + +### Streaming Messages + +```go +// Send delta (incremental) updates +msg := &message.Message{ + ID: "msg_123", + Type: message.TypeText, + Delta: true, // Incremental update + Props: map[string]interface{}{ + "content": "Hello", + }, +} +output.Send(ctx, msg) + +// Mark as complete +msg.Delta = false +msg.Done = true +msg.Props["content"] = "Hello world!" // Full content +output.Send(ctx, msg) +``` + +### Custom Writers + +```go +// Register a custom writer factory +factory := &MyCustomFactory{} +output.SetWriterFactory(factory) + +// Now all calls to output.Send will use your custom writer +``` + +## Integration with Handlers + +The `handlers` package uses the output module for streaming: + +```go +func DefaultStreamHandler(ctx *context.Context) context.StreamFunc { + return func(chunkType context.StreamChunkType, data []byte) int { + switch chunkType { + case context.ChunkText: + msg := output.NewTextMessage(string(data)) + output.Send(ctx, msg) + case context.ChunkThinking: + msg := output.NewThinkingMessage(string(data)) + output.Send(ctx, msg) + // ... handle other types + } + return 0 // Continue + } +} +``` + +## Context-based Routing + +The output module automatically selects the right writer based on `ctx.Accept`: + +| `ctx.Accept` | Writer | Format | +| ------------- | ------ | --------------------- | +| `standard` | OpenAI | OpenAI-compatible SSE | +| `cui-web` | CUI | Universal DSL JSON | +| `cui-native` | CUI | Universal DSL JSON | +| `cui-desktop` | CUI | Universal DSL JSON | + +## Writer Caching + +Writers are cached per context to avoid recreating them: + +```go +// Get or create writer (cached) +writer := output.GetWriter(ctx) + +// Clear cache when done +output.Close(ctx) // Also closes the writer +``` + +## See Also + +- [BUILTIN_TYPES.md](./BUILTIN_TYPES.md) - Complete documentation of built-in message types +- [adapters/openai/README.md](./adapters/openai/README.md) - OpenAI adapter documentation diff --git a/agent/output/adapters/cui/adapter.go b/agent/output/adapters/cui/adapter.go new file mode 100644 index 00000000..fbcb16b5 --- /dev/null +++ b/agent/output/adapters/cui/adapter.go @@ -0,0 +1,26 @@ +package cui + +import "github.com/yaoapp/yao/agent/output/message" + +// Adapter implements the message.Adapter interface for CUI clients. +// It performs no conversion and outputs messages as-is, as CUI clients +// are designed to directly consume the universal DSL. +type Adapter struct{} + +// NewAdapter creates a new CUI adapter. +func NewAdapter() *Adapter { + return &Adapter{} +} + +// Adapt converts a universal Message to one or more client-specific chunks. +// 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. + return []interface{}{msg}, nil +} + +// SupportsType checks if the adapter explicitly supports a given message type. +// CUI adapter supports all types as it renders them directly. +func (a *Adapter) SupportsType(msgType string) bool { + return true +} diff --git a/agent/output/adapters/cui/factory.go b/agent/output/adapters/cui/factory.go new file mode 100644 index 00000000..1bac40fc --- /dev/null +++ b/agent/output/adapters/cui/factory.go @@ -0,0 +1,25 @@ +package cui + +import ( + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/message" +) + +// Factory is the factory for creating CUI writers and adapters +type Factory struct{} + +// NewFactory creates a new CUI factory +func NewFactory() *Factory { + return &Factory{} +} + +// CreateWriter creates a CUI writer +func (f *Factory) CreateWriter(ctx *context.Context) (message.Writer, error) { + return NewWriter(ctx) +} + +// CreateAdapter creates a CUI adapter +func (f *Factory) CreateAdapter(ctx *context.Context) (message.Adapter, error) { + return NewAdapter(), nil +} + diff --git a/agent/output/adapters/cui/writer.go b/agent/output/adapters/cui/writer.go new file mode 100644 index 00000000..b14fb742 --- /dev/null +++ b/agent/output/adapters/cui/writer.go @@ -0,0 +1,79 @@ +package cui + +import ( + "encoding/json" + + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/message" +) + +// Writer implements the message.Writer interface for CUI clients +type Writer struct { + ctx *context.Context + adapter *Adapter +} + +// NewWriter creates a new CUI writer +func NewWriter(ctx *context.Context) (*Writer, error) { + return &Writer{ + ctx: ctx, + adapter: NewAdapter(), + }, nil +} + +// Write writes a single message to the output stream +func (w *Writer) Write(msg *message.Message) error { + // CUI adapter passes messages through as-is + chunks, err := w.adapter.Adapt(msg) + if err != nil { + return err + } + + // Send each chunk + for _, chunk := range chunks { + if err := w.sendChunk(chunk); err != nil { + return err + } + } + + return nil +} + +// WriteGroup writes a message group to the output stream +func (w *Writer) WriteGroup(group *message.MessageGroup) error { + // For CUI, we send a group start message, all messages, then a group end message + // The group structure itself is also sent for clients that want it + + // Send the group + if err := w.sendChunk(group); err != nil { + return err + } + + return nil +} + +// Flush flushes any buffered data to the output stream +func (w *Writer) Flush() error { + // For SSE, we don't need explicit flushing + // The underlying connection handles it + return nil +} + +// Close closes the writer and cleans up resources +func (w *Writer) Close() error { + // Nothing to clean up for CUI writer + return nil +} + +// sendChunk sends a chunk to the output stream +func (w *Writer) sendChunk(chunk interface{}) error { + // Convert chunk to JSON + data, err := json.Marshal(chunk) + if err != nil { + return err + } + + // Send via context's writer + // The context knows how to send data based on the connection type (SSE, WebSocket, etc.) + return w.ctx.Send(data) +} diff --git a/agent/output/adapters/openai/README.md b/agent/output/adapters/openai/README.md new file mode 100644 index 00000000..c90b0bfa --- /dev/null +++ b/agent/output/adapters/openai/README.md @@ -0,0 +1,240 @@ +# OpenAI Adapter + +OpenAI adapter converts universal DSL messages to OpenAI-compatible format. + +## Conversion Rules + +### Built-in Types (Standard) + +These types are defined in `output.types.go` and have standardized Props structures that all adapters must support: + +| Message Type | Constant | Props Structure | OpenAI Format | Description | +| ------------ | --------------------- | --------------- | ------------------------- | ------------------------------------- | +| `text` | `output.TypeText` | `TextProps` | `delta.content` | Plain text or Markdown | +| `thinking` | `output.TypeThinking` | `ThinkingProps` | `delta.reasoning_content` | Reasoning process (o1 models) | +| `loading` | `output.TypeLoading` | `LoadingProps` | `delta.reasoning_content` | Loading indicator (shows as thinking) | +| `tool_call` | `output.TypeToolCall` | `ToolCallProps` | `delta.tool_calls` | Tool/function calls | +| `error` | `output.TypeError` | `ErrorProps` | `error` | Error messages | +| `action` | `output.TypeAction` | `ActionProps` | (not sent) | System actions (silent) | + +### Custom Types + +All other message types (not in the built-in list) are converted to Markdown links: + +| Format | Example | +| ---------------------- | ------------------------------- | +| `delta.content` (link) | `"🖼️ [View Image](https://...)` | + +## Usage + +### Basic Usage + +```go +import ( + "github.com/yaoapp/yao/agent/output/adapters/openai" +) + +// Create adapter with default config +adapter := openai.NewAdapter() + +// Convert message +chunks, err := adapter.Adapt(msg) +``` + +### With Custom Configuration + +```go +// Create adapter with options +adapter := openai.NewAdapter( + openai.WithBaseURL("https://api.example.com"), + openai.WithModel("gpt-4"), + openai.WithLinkTemplate("image", "🖼️ [View Image](%s)"), + openai.WithLinkTransformer(myOTPTransformer), +) +``` + +### With Link Transformer (OTP) + +```go +// Define OTP transformer +func otpTransformer(url string, msgType string, msgID string) (string, error) { + // Generate OTP token + otp := generateOTP(msgID, 3600) // 1 hour expiry + + // Create short link with OTP + shortURL := fmt.Sprintf("https://api.example.com/s/%s?t=%s", msgID, otp) + + return shortURL, nil +} + +// Use transformer +adapter := openai.NewAdapter( + openai.WithLinkTransformer(otpTransformer), +) +``` + +### Custom Converter + +```go +// Register custom converter for a specific type +adapter := openai.NewAdapter( + openai.WithConverter("my_widget", func(msg *output.Message, config *openai.AdapterConfig) ([]interface{}, error) { + // Custom conversion logic + return []interface{}{ + // OpenAI format chunk + }, nil + }), +) +``` + +## Examples + +### Text Message (Built-in Type) + +**Input (DSL):** + +```json +{ + "type": "text", + "props": { + "content": "Hello world" + } +} +``` + +Or using helper: + +```go +msg := output.NewTextMessage("Hello world") +``` + +**Output (OpenAI):** + +```json +{ + "id": "M1", + "object": "chat.completion.chunk", + "model": "yao-agent", + "choices": [ + { + "delta": { + "content": "Hello world" + } + } + ] +} +``` + +### Image Message + +**Input (DSL):** + +```json +{ + "id": "M2", + "type": "image", + "props": { + "url": "https://example.com/avatar.jpg" + } +} +``` + +**Output (OpenAI):** + +```json +{ + "id": "M2", + "object": "chat.completion.chunk", + "model": "yao-agent", + "choices": [ + { + "delta": { + "content": "🖼️ [View Image](https://api.example.com/s/M2?t=abc123)" + } + } + ] +} +``` + +### Button Message + +**Input (DSL):** + +```json +{ + "id": "M3", + "type": "button", + "props": { + "text": "Approve", + "action": "workflow.approve" + } +} +``` + +**Output (OpenAI):** + +```json +{ + "id": "M3", + "object": "chat.completion.chunk", + "model": "yao-agent", + "choices": [ + { + "delta": { + "content": "🔘 [Approve](https://api.example.com/s/M3?t=abc123)" + } + } + ] +} +``` + +## Link Templates + +Default templates: + +```go +"image": "🖼️ [View Image](%s)" +"audio": "🔊 [Play Audio](%s)" +"video": "🎬 [Watch Video](%s)" +"file": "📎 [Download File](%s)" +"page": "📄 [Open Page](%s)" +"table": "📊 [View Table](%s)" +"chart": "📈 [View Chart](%s)" +"list": "📋 [View List](%s)" +"form": "📝 [Fill Form](%s)" +"button": "🔘 [%s](%s)" // Special: button text + link +``` + +Customize templates: + +```go +adapter := openai.NewAdapter( + openai.WithLinkTemplate("image", "📷 Image: %s"), + openai.WithLinkTemplate("video", "🎥 Watch: %s"), +) +``` + +## Link Transformer (TODO) + +The link transformer is currently left empty for future implementation of OTP/short link functionality. + +**Planned features:** + +- Generate one-time password (OTP) for secure access +- Create short URLs for better readability +- Set expiration time for links +- Track link access for analytics + +**Example implementation:** + +```go +func otpTransformer(url string, msgType string, msgID string) (string, error) { + // TODO: Implement OTP generation + // 1. Generate OTP token with expiry + // 2. Store mapping: token -> (url, msgType, msgID, expiry) + // 3. Create short URL with token + // 4. Return short URL + + return url, nil // Currently pass-through +} +``` diff --git a/agent/output/adapters/openai/adapter.go b/agent/output/adapters/openai/adapter.go new file mode 100644 index 00000000..3e434f34 --- /dev/null +++ b/agent/output/adapters/openai/adapter.go @@ -0,0 +1,91 @@ +package openai + +import "github.com/yaoapp/yao/agent/output/message" + +// Adapter is the OpenAI adapter that converts messages to OpenAI format +type Adapter struct { + config *AdapterConfig + registry *ConverterRegistry +} + +// NewAdapter creates a new OpenAI adapter with default configuration +func NewAdapter(options ...Option) *Adapter { + adapter := &Adapter{ + config: DefaultAdapterConfig(), + registry: NewConverterRegistry(), + } + + // Apply options + for _, opt := range options { + opt(adapter) + } + + return adapter +} + +// Option is a function that configures the adapter +type Option func(*Adapter) + +// WithBaseURL sets the base URL for generating view links +func WithBaseURL(baseURL string) Option { + return func(a *Adapter) { + a.config.BaseURL = baseURL + } +} + +// WithLinkTemplate sets a custom link template for a message type +func WithLinkTemplate(msgType string, template string) Option { + return func(a *Adapter) { + a.config.LinkTemplates[msgType] = template + } +} + +// WithLinkTransformer sets the link transformer function +func WithLinkTransformer(transformer LinkTransformer) Option { + return func(a *Adapter) { + a.config.LinkTransformer = transformer + } +} + +// WithModel sets the model name for OpenAI responses +func WithModel(model string) Option { + return func(a *Adapter) { + a.config.Model = model + } +} + +// WithConverter registers a custom converter for a message type +func WithConverter(msgType string, converter ConverterFunc) Option { + return func(a *Adapter) { + a.registry.Register(msgType, converter) + } +} + +// Adapt converts a universal Message to OpenAI-compatible format +func (a *Adapter) Adapt(msg *message.Message) ([]interface{}, error) { + // Get converter for this message type + converter, exists := a.registry.GetConverter(msg.Type) + if !exists { + // Use default converter for unknown types (convert to link) + converter = convertToLink + } + + // Convert the message + return converter(msg, a.config) +} + +// SupportsType checks if the adapter explicitly supports a given message type +func (a *Adapter) SupportsType(msgType string) bool { + _, exists := a.registry.GetConverter(msgType) + return exists +} + +// GetConfig returns the adapter configuration +func (a *Adapter) GetConfig() *AdapterConfig { + return a.config +} + +// GetRegistry returns the converter registry +func (a *Adapter) GetRegistry() *ConverterRegistry { + return a.registry +} diff --git a/agent/output/adapters/openai/converter.go b/agent/output/adapters/openai/converter.go new file mode 100644 index 00000000..16d4cd5d --- /dev/null +++ b/agent/output/adapters/openai/converter.go @@ -0,0 +1,260 @@ +package openai + +import ( + "fmt" + "time" + + "github.com/yaoapp/yao/agent/output/message" +) + +// ConverterRegistry manages message type converters +type ConverterRegistry struct { + converters map[string]ConverterFunc +} + +// NewConverterRegistry creates a new converter registry with default converters +func NewConverterRegistry() *ConverterRegistry { + return &ConverterRegistry{ + converters: map[string]ConverterFunc{ + message.TypeText: convertText, + message.TypeThinking: convertThinking, + message.TypeLoading: convertLoading, + message.TypeToolCall: convertToolCall, + message.TypeError: convertError, + message.TypeImage: convertImage, + message.TypeAudio: convertToLink, + message.TypeVideo: convertToLink, + message.TypeAction: convertAction, + }, + } +} + +// Register registers a custom converter for a message type +func (r *ConverterRegistry) Register(msgType string, converter ConverterFunc) { + r.converters[msgType] = converter +} + +// GetConverter retrieves a converter for a given message type. +func (r *ConverterRegistry) GetConverter(msgType string) (ConverterFunc, bool) { + converter, exists := r.converters[msgType] + return converter, exists +} + +// Convert converts a message using registered converters +// If no converter is found, converts to link format +func (r *ConverterRegistry) Convert(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { + // Check for registered converter + if converter, exists := r.converters[msg.Type]; exists { + return converter(msg, config) + } + + // Fallback: convert to link format + return convertToLink(msg, config) +} + +// convertText converts text messages to OpenAI format +func convertText(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { + content := getStringProp(msg.Props, "content", "") + + return []interface{}{ + createOpenAIChunk(msg.ID, config.Model, map[string]interface{}{ + "content": content, + }), + }, nil +} + +// convertThinking converts thinking messages to OpenAI reasoning format (o1 series) +func convertThinking(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { + content := getStringProp(msg.Props, "content", "") + + return []interface{}{ + createOpenAIChunk(msg.ID, config.Model, map[string]interface{}{ + "reasoning_content": content, + }), + }, nil +} + +// convertLoading converts loading messages to OpenAI reasoning format +// This makes loading messages visible in standard OpenAI clients as thinking process +func convertLoading(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { + message := getStringProp(msg.Props, "message", "Processing...") + + // Convert loading to reasoning_content so it shows in OpenAI clients + return []interface{}{ + createOpenAIChunk(msg.ID, config.Model, map[string]interface{}{ + "reasoning_content": message, + }), + }, nil +} + +// convertToolCall converts tool_call messages to OpenAI format +func convertToolCall(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { + // Tool call format varies, pass through the props + toolCalls := []map[string]interface{}{} + + // If props contain tool call data, use it + if id, ok := msg.Props["id"].(string); ok { + toolCall := map[string]interface{}{ + "id": id, + "type": "function", + } + + if function, ok := msg.Props["function"].(map[string]interface{}); ok { + toolCall["function"] = function + } + + toolCalls = append(toolCalls, toolCall) + } + + return []interface{}{ + createOpenAIChunk(msg.ID, config.Model, map[string]interface{}{ + "tool_calls": toolCalls, + }), + }, nil +} + +// convertError converts error messages to OpenAI error format +func convertError(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { + message := getStringProp(msg.Props, "message", "An error occurred") + code := getStringProp(msg.Props, "code", "server_error") + + return []interface{}{ + map[string]interface{}{ + "error": map[string]interface{}{ + "message": message, + "type": code, + "code": code, + }, + }, + }, nil +} + +// convertAction converts action messages to nothing (silent in OpenAI clients) +// Action messages are system-level commands (open panel, navigate, etc.) +// and should not be sent to standard chat clients +func convertAction(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { + // Return empty slice - no output for action messages in OpenAI format + return []interface{}{}, nil +} + +// convertImage converts image messages to Markdown image format +// Uses ![alt](url) which displays inline in Markdown-supporting clients +func convertImage(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { + // Get URL + url, ok := msg.Props["url"].(string) + if !ok || url == "" { + return nil, fmt.Errorf("image message missing url") + } + + // Transform URL if transformer is provided + if config.LinkTransformer != nil { + transformedURL, err := config.LinkTransformer(url, msg.Type, msg.ID) + if err != nil { + return nil, err + } + url = transformedURL + } + + // Get alt text (default to "Image") + alt := getStringProp(msg.Props, "alt", "Image") + + // Format as Markdown image: ![alt](url) + template := getLinkTemplate(msg.Type, config) + text := fmt.Sprintf(template, alt, url) + + return []interface{}{ + createOpenAIChunk(msg.ID, config.Model, map[string]interface{}{ + "content": text, + }), + }, nil +} + +// convertToLink converts any message type to a Markdown link format +func convertToLink(msg *message.Message, config *AdapterConfig) ([]interface{}, error) { + // Generate link + link, err := generateViewLink(msg, config) + if err != nil { + return nil, err + } + + // Get template + template := getLinkTemplate(msg.Type, config) + + // Format text + var text string + if msg.Type == "button" { + // Button is special: needs button text + buttonText := getStringProp(msg.Props, "text", "Button") + text = fmt.Sprintf(template, buttonText, link) + } else { + text = fmt.Sprintf(template, link) + } + + return []interface{}{ + createOpenAIChunk(msg.ID, config.Model, map[string]interface{}{ + "content": text, + }), + }, nil +} + +// generateViewLink generates a view link for a message +func generateViewLink(msg *message.Message, config *AdapterConfig) (string, error) { + // If Props contains a URL, use it + if url, ok := msg.Props["url"].(string); ok { + // Transform URL if transformer is provided + if config.LinkTransformer != nil { + return config.LinkTransformer(url, msg.Type, msg.ID) + } + return url, nil + } + + // Generate view link: {baseURL}/agent/view/{type}/{id} + baseURL := config.BaseURL + if baseURL == "" { + baseURL = "" // TODO: Get from environment or context + } + + viewURL := fmt.Sprintf("%s/agent/view/%s/%s", baseURL, msg.Type, msg.ID) + + // Transform URL if transformer is provided + if config.LinkTransformer != nil { + return config.LinkTransformer(viewURL, msg.Type, msg.ID) + } + + return viewURL, nil +} + +// getLinkTemplate gets the link template for a message type +func getLinkTemplate(msgType string, config *AdapterConfig) string { + if template, exists := config.LinkTemplates[msgType]; exists { + return template + } + + // Default fallback template + return "📎 [View %s](" + msgType + ")" +} + +// createOpenAIChunk creates an OpenAI chat completion chunk +func createOpenAIChunk(id string, model string, delta map[string]interface{}) map[string]interface{} { + return map[string]interface{}{ + "id": id, + "object": "chat.completion.chunk", + "created": time.Now().Unix(), + "model": model, + "choices": []map[string]interface{}{ + { + "index": 0, + "delta": delta, + "finish_reason": nil, + }, + }, + } +} + +// getStringProp safely gets a string property from props +func getStringProp(props map[string]interface{}, key string, defaultValue string) string { + if val, ok := props[key].(string); ok { + return val + } + return defaultValue +} diff --git a/agent/output/adapters/openai/factory.go b/agent/output/adapters/openai/factory.go new file mode 100644 index 00000000..eecb720e --- /dev/null +++ b/agent/output/adapters/openai/factory.go @@ -0,0 +1,28 @@ +package openai + +import ( + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/message" +) + +// Factory is the factory for creating OpenAI writers and adapters +type Factory struct { + options []Option +} + +// NewFactory creates a new OpenAI factory with options +func NewFactory(options ...Option) *Factory { + return &Factory{ + options: options, + } +} + +// CreateWriter creates an OpenAI writer +func (f *Factory) CreateWriter(ctx *context.Context) (message.Writer, error) { + return NewWriter(ctx) +} + +// CreateAdapter creates an OpenAI adapter +func (f *Factory) CreateAdapter(ctx *context.Context) (message.Adapter, error) { + return NewAdapter(f.options...), nil +} diff --git a/agent/output/adapters/openai/types.go b/agent/output/adapters/openai/types.go new file mode 100644 index 00000000..68df9429 --- /dev/null +++ b/agent/output/adapters/openai/types.go @@ -0,0 +1,62 @@ +package openai + +import "github.com/yaoapp/yao/agent/output/message" + +// ConverterFunc converts a message to OpenAI format chunks +type ConverterFunc func(msg *message.Message, config *AdapterConfig) ([]interface{}, error) + +// LinkTransformer transforms a URL to a secure link (with OTP, short URL, etc.) +// Returns the transformed link or error +type LinkTransformer func(url string, msgType string, msgID string) (string, error) + +// AdapterConfig holds the configuration for OpenAI adapter +type AdapterConfig struct { + // BaseURL is the base URL for generating view links + // Example: "https://api.example.com" + BaseURL string + + // LinkTemplates defines the Markdown template for each message type + // %s will be replaced with the link + // Example: "🖼️ [View Image](%s)" + LinkTemplates map[string]string + + // LinkTransformer transforms URLs to secure links with OTP + // If nil, URLs are used as-is + LinkTransformer LinkTransformer + + // Model name to include in OpenAI responses + Model string +} + +// DefaultLinkTemplates provides default Markdown templates for non-text message types +var DefaultLinkTemplates = map[string]string{ + "image": "![%s](%s)", // Markdown image: ![alt](url) - displays inline + "audio": "🔊 [Play Audio](%s)", // Link (audio can't display inline in Markdown) + "video": "🎬 [Watch Video](%s)", // Link (video can't display inline in Markdown) + "file": "📎 [Download File](%s)", + "page": "📄 [Open Page](%s)", + "table": "📊 [View Table](%s)", + "chart": "📈 [View Chart](%s)", + "list": "📋 [View List](%s)", + "form": "📝 [Fill Form](%s)", + "button": "🔘 [%s](%s)", // Special: button needs two params (text, link) +} + +// DefaultAdapterConfig returns a default adapter configuration +func DefaultAdapterConfig() *AdapterConfig { + return &AdapterConfig{ + BaseURL: "", // Will be set from environment or context + LinkTemplates: copyLinkTemplates(DefaultLinkTemplates), + LinkTransformer: nil, // No transformation by default + Model: "yao-agent", + } +} + +// copyLinkTemplates creates a copy of link templates +func copyLinkTemplates(templates map[string]string) map[string]string { + copy := make(map[string]string, len(templates)) + for k, v := range templates { + copy[k] = v + } + return copy +} diff --git a/agent/output/adapters/openai/writer.go b/agent/output/adapters/openai/writer.go new file mode 100644 index 00000000..c570c027 --- /dev/null +++ b/agent/output/adapters/openai/writer.go @@ -0,0 +1,92 @@ +package openai + +import ( + "encoding/json" + + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/message" +) + +// Writer implements the message.Writer interface for OpenAI-compatible clients +type Writer struct { + ctx *context.Context + adapter *Adapter +} + +// NewWriter creates a new OpenAI writer +func NewWriter(ctx *context.Context) (*Writer, error) { + // Create adapter with default config + adapter := NewAdapter() + + return &Writer{ + ctx: ctx, + adapter: adapter, + }, nil +} + +// Write writes a single message to the output stream +func (w *Writer) Write(msg *message.Message) error { + // Convert message to OpenAI format using adapter + chunks, err := w.adapter.Adapt(msg) + if err != nil { + return err + } + + // Send each chunk + for _, chunk := range chunks { + if err := w.sendChunk(chunk); err != nil { + return err + } + } + + return nil +} + +// WriteGroup writes a message group to the output stream +func (w *Writer) WriteGroup(group *message.MessageGroup) error { + // For OpenAI, we don't send group markers + // Just send each message individually + for _, msg := range group.Messages { + if err := w.Write(msg); err != nil { + return err + } + } + + return nil +} + +// Flush flushes any buffered data to the output stream +func (w *Writer) Flush() error { + // For SSE, we don't need explicit flushing + // The underlying connection handles it + return nil +} + +// Close closes the writer and cleans up resources +func (w *Writer) Close() error { + // Send final [DONE] message for OpenAI compatibility + return w.sendDone() +} + +// sendChunk sends a chunk to the output stream in SSE format +func (w *Writer) sendChunk(chunk interface{}) error { + // Convert chunk to JSON + data, err := json.Marshal(chunk) + if err != nil { + return err + } + + // Format as SSE: "data: {json}\n\n" + sseData := append([]byte("data: "), data...) + sseData = append(sseData, []byte("\n\n")...) + + // Send via context's writer + return w.ctx.Send(sseData) +} + +// sendDone sends the final [DONE] message +func (w *Writer) sendDone() error { + // OpenAI SSE format uses "data: [DONE]\n\n" to signal completion + doneData := []byte("data: [DONE]\n\n") + return w.ctx.Send(doneData) +} diff --git a/agent/output/builtin.go b/agent/output/builtin.go new file mode 100644 index 00000000..0932e0b1 --- /dev/null +++ b/agent/output/builtin.go @@ -0,0 +1,129 @@ +package output + +import ( + "fmt" + "math/rand" + "time" + + "github.com/yaoapp/yao/agent/output/message" +) + +// Helper functions for creating built-in message types + +func init() { + rand.Seed(time.Now().UnixNano()) +} + +// NewTextMessage creates a text message +func NewTextMessage(content string) *message.Message { + return &message.Message{ + Type: message.TypeText, + Props: map[string]interface{}{ + "content": content, + }, + } +} + +// NewThinkingMessage creates a thinking message +func NewThinkingMessage(content string) *message.Message { + return &message.Message{ + Type: message.TypeThinking, + Props: map[string]interface{}{ + "content": content, + }, + } +} + +// NewLoadingMessage creates a loading message +func NewLoadingMessage(msg string) *message.Message { + return &message.Message{ + Type: message.TypeLoading, + Props: map[string]interface{}{ + "message": msg, + }, + } +} + +// NewToolCallMessage creates a tool call message +func NewToolCallMessage(id, name, arguments string) *message.Message { + return &message.Message{ + Type: message.TypeToolCall, + Props: map[string]interface{}{ + "id": id, + "name": name, + "arguments": arguments, + }, + } +} + +// NewErrorMessage creates an error message +func NewErrorMessage(msg, code string) *message.Message { + return &message.Message{ + Type: message.TypeError, + Props: map[string]interface{}{ + "message": msg, + "code": code, + }, + } +} + +// NewActionMessage creates an action message +func NewActionMessage(name string, payload map[string]interface{}) *message.Message { + return &message.Message{ + Type: message.TypeAction, + Props: map[string]interface{}{ + "name": name, + "payload": payload, + }, + } +} + +// NewImageMessage creates an image message +func NewImageMessage(url string, alt string) *message.Message { + return &message.Message{ + Type: message.TypeImage, + Props: map[string]interface{}{ + "url": url, + "alt": alt, + }, + } +} + +// NewAudioMessage creates an audio message +func NewAudioMessage(url string, format string) *message.Message { + return &message.Message{ + Type: message.TypeAudio, + Props: map[string]interface{}{ + "url": url, + "format": format, + }, + } +} + +// NewVideoMessage creates a video message +func NewVideoMessage(url string) *message.Message { + return &message.Message{ + Type: message.TypeVideo, + Props: map[string]interface{}{ + "url": url, + }, + } +} + +// 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: + return true + default: + return false + } +} + +// GenerateID generates a unique message ID +func GenerateID() string { + // Generate a random ID with timestamp prefix for uniqueness + timestamp := time.Now().UnixNano() + random := rand.Int63() + return fmt.Sprintf("msg_%d_%d", timestamp, random) +} diff --git a/agent/output/message/interfaces.go b/agent/output/message/interfaces.go new file mode 100644 index 00000000..9ea46745 --- /dev/null +++ b/agent/output/message/interfaces.go @@ -0,0 +1,56 @@ +package message + +import "github.com/yaoapp/yao/agent/context" + +// Writer is the interface for writing output messages +// Different writers handle different output formats (SSE, WebSocket, Standard, etc.) +type Writer interface { + // Write writes a single message + Write(msg *Message) error + + // WriteGroup writes a group of messages + WriteGroup(group *MessageGroup) error + + // Flush flushes any buffered data + Flush() error + + // Close closes the writer and releases resources + Close() error +} + +// Adapter is the interface for adapting messages to different formats +// Adapters transform messages from the universal DSL to specific client formats +type Adapter interface { + // Adapt transforms a message to the target format + // Returns a slice of output chunks (some messages may be split into multiple chunks) + Adapt(msg *Message) ([]interface{}, error) + + // SupportsType checks if this adapter supports a specific message type + SupportsType(msgType string) bool +} + +// WriterFactory creates writers based on context +type WriterFactory interface { + // NewWriter creates a writer for the given context + NewWriter(ctx *context.Context, adapter Adapter) (Writer, error) +} + +// AdapterFactory creates adapters based on context +type AdapterFactory interface { + // NewAdapter creates an adapter for the given context + NewAdapter(ctx *context.Context) (Adapter, error) +} + +// StreamHandler handles streaming message processing +// It bridges between LLM streaming chunks and output messages +type StreamHandler interface { + // Handle processes a streaming chunk from LLM + Handle(chunkType context.StreamChunkType, data []byte) error + + // Flush flushes any pending messages + Flush() error + + // Close closes the handler + Close() error +} + diff --git a/agent/output/message/types.go b/agent/output/message/types.go new file mode 100644 index 00000000..6593b607 --- /dev/null +++ b/agent/output/message/types.go @@ -0,0 +1,158 @@ +package message + +// Message represents a universal message structure (DSL) +// All messages are expressed through Type + Props, without predefining specific types +type Message struct { + // Core fields + Type string `json:"type"` // Message type (frontend decides how to render) + Props map[string]interface{} `json:"props,omitempty"` // Message properties (passed to frontend component) + + // Streaming control + ID string `json:"id,omitempty"` // Message ID (used for merging messages in streaming scenarios) + Delta bool `json:"delta,omitempty"` // Whether this is an incremental update + Done bool `json:"done,omitempty"` // Whether the message is complete + + // Delta update control + DeltaPath string `json:"delta_path,omitempty"` // Update path (e.g., "content", "data", "items.0.name") + DeltaAction string `json:"delta_action,omitempty"` // Update action (append, replace, merge, set) + + // Type correction (for streaming scenarios) + TypeChange bool `json:"type_change,omitempty"` // Marks this as a type correction message + + // Message group + GroupID string `json:"group_id,omitempty"` // Parent message group ID + GroupStart bool `json:"group_start,omitempty"` // Marks the start of a message group + GroupEnd bool `json:"group_end,omitempty"` // Marks the end of a message group + + // Metadata + Metadata *Metadata `json:"metadata,omitempty"` // Additional metadata +} + +// Metadata represents message metadata +type Metadata struct { + Timestamp int64 `json:"timestamp,omitempty"` // Timestamp in nanoseconds + Sequence int `json:"sequence,omitempty"` // Sequence number (for ordering) + TraceID string `json:"trace_id,omitempty"` // Trace ID (for debugging) +} + +// MessageGroup represents a semantically complete group of messages +type MessageGroup struct { + ID string `json:"id"` // Message group ID + Messages []*Message `json:"messages"` // List of messages + Metadata *Metadata `json:"metadata,omitempty"` // Metadata +} + +// Built-in message types that all adapters must support +// These types have standardized Props structures +const ( + // Content types + TypeText = "text" // Plain text or Markdown content + TypeThinking = "thinking" // Reasoning/thinking process (e.g., o1 models) + TypeLoading = "loading" // Loading/processing indicator (preprocessing, knowledge base search, etc.) + TypeToolCall = "tool_call" // LLM tool/function call + TypeError = "error" // Error message + + // Media types (with OpenAI support) + TypeImage = "image" // Image content + TypeAudio = "audio" // Audio content + TypeVideo = "video" // Video content + + // System types (not visible in standard chat clients) + TypeAction = "action" // System action (open panel, navigate, etc.) - silent in OpenAI clients +) + +// Standard Props structures for built-in types + +// TextProps defines the standard structure for text messages +// Type: "text" +// Props: {"content": string} +type TextProps struct { + Content string `json:"content"` // Text content (supports Markdown) +} + +// ThinkingProps defines the standard structure for thinking messages +// Type: "thinking" +// Props: {"content": string} +type ThinkingProps struct { + Content string `json:"content"` // Reasoning/thinking content +} + +// LoadingProps defines the standard structure for loading messages +// Type: "loading" +// Props: {"message": string} +type LoadingProps struct { + Message string `json:"message"` // Loading message (e.g., "Searching knowledge base...") +} + +// ToolCallProps defines the standard structure for tool_call messages +// Type: "tool_call" +// Props: {"id": string, "name": string, "arguments": string} +type ToolCallProps struct { + ID string `json:"id"` // Tool call ID + Name string `json:"name"` // Function/tool name + Arguments string `json:"arguments,omitempty"` // JSON string of arguments +} + +// ErrorProps defines the standard structure for error messages +// Type: "error" +// Props: {"message": string, "code": string} +type ErrorProps struct { + Message string `json:"message"` // Error message + Code string `json:"code,omitempty"` // Error code + Details string `json:"details,omitempty"` // Additional error details +} + +// ActionProps defines the standard structure for action messages +// Type: "action" +// Props: {"name": string, "payload": map} +type ActionProps struct { + Name string `json:"name"` // Action name (e.g., "open_panel", "navigate") + Payload map[string]interface{} `json:"payload,omitempty"` // Action payload/parameters +} + +// ImageProps defines the standard structure for image messages +// Type: "image" +// Props: {"url": string, "alt": string, "width": int, "height": int, "detail": string} +type ImageProps struct { + URL string `json:"url"` // Required: Image URL or base64 encoded data + Alt string `json:"alt,omitempty"` // Alternative text + Width int `json:"width,omitempty"` // Image width in pixels + Height int `json:"height,omitempty"` // Image height in pixels + Detail string `json:"detail,omitempty"` // OpenAI detail level: "auto", "low", "high" +} + +// AudioProps defines the standard structure for audio messages +// Type: "audio" +// Props: {"url": string, "format": string, "duration": float64, "transcript": string, "autoplay": bool} +type AudioProps struct { + URL string `json:"url"` // Required: Audio URL or base64 encoded data + Format string `json:"format,omitempty"` // Audio format: "mp3", "wav", "ogg", etc. + Duration float64 `json:"duration,omitempty"` // Duration in seconds + Transcript string `json:"transcript,omitempty"` // Audio transcript text + Autoplay bool `json:"autoplay,omitempty"` // Whether to autoplay + Controls bool `json:"controls,omitempty"` // Whether to show controls (default: true) +} + +// VideoProps defines the standard structure for video messages +// Type: "video" +// Props: {"url": string, "format": string, "duration": float64, "thumbnail": string, "width": int, "height": int, "autoplay": bool} +type VideoProps struct { + URL string `json:"url"` // Required: Video URL + Format string `json:"format,omitempty"` // Video format: "mp4", "webm", etc. + Duration float64 `json:"duration,omitempty"` // Duration in seconds + Thumbnail string `json:"thumbnail,omitempty"` // Thumbnail/poster image URL + Width int `json:"width,omitempty"` // Video width in pixels + Height int `json:"height,omitempty"` // Video height in pixels + Autoplay bool `json:"autoplay,omitempty"` // Whether to autoplay + Controls bool `json:"controls,omitempty"` // Whether to show controls (default: true) + Loop bool `json:"loop,omitempty"` // Whether to loop +} + +// Delta action constants for incremental updates +const ( + DeltaAppend = "append" // Append (for arrays, strings) + DeltaReplace = "replace" // Replace (for any value) + DeltaMerge = "merge" // Merge (for objects) + DeltaSet = "set" // Set (for new fields) +) + diff --git a/agent/output/output.go b/agent/output/output.go index ad89311b..65aa99f0 100644 --- a/agent/output/output.go +++ b/agent/output/output.go @@ -1 +1,149 @@ package output + +import ( + "fmt" + "sync" + + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/adapters/cui" + "github.com/yaoapp/yao/agent/output/adapters/openai" + "github.com/yaoapp/yao/agent/output/message" +) + +var ( + writerCache = make(map[*context.Context]message.Writer) + writerMutex sync.RWMutex + globalFactory message.WriterFactory +) + +// Send sends a single message using the appropriate writer for the context +func Send(ctx *context.Context, msg *message.Message) error { + writer, err := GetWriter(ctx) + if err != nil { + return err + } + return writer.Write(msg) +} + +// SendGroup sends a message group using the appropriate writer for the context +func SendGroup(ctx *context.Context, group *message.MessageGroup) error { + writer, err := GetWriter(ctx) + if err != nil { + return err + } + return writer.WriteGroup(group) +} + +// GetWriter gets or creates a writer for the given context +// Writers are cached per context to avoid recreating them +func GetWriter(ctx *context.Context) (message.Writer, error) { + // Try to get cached writer + writerMutex.RLock() + writer, exists := writerCache[ctx] + writerMutex.RUnlock() + + if exists { + return writer, nil + } + + // Create new writer + writerMutex.Lock() + defer writerMutex.Unlock() + + // Double-check after acquiring write lock + if writer, exists := writerCache[ctx]; exists { + return writer, nil + } + + // Create writer based on context.Accept + writer, err := createWriter(ctx) + if err != nil { + return nil, err + } + + // Cache the writer + writerCache[ctx] = writer + + return writer, nil +} + +// createWriter creates a writer based on context.Accept +func createWriter(ctx *context.Context) (message.Writer, error) { + // If global factory is set, use it + if globalFactory != nil { + return globalFactory.NewWriter(ctx, nil) + } + + // Default: create based on Accept type + switch ctx.Accept { + case context.AcceptStandard: + // OpenAI-compatible format + return openai.NewWriter(ctx) + + case context.AcceptWebCUI, context.AccepNativeCUI, context.AcceptDesktopCUI: + // CUI format + return cui.NewWriter(ctx) + + default: + // Default to CUI + return cui.NewWriter(ctx) + } +} + +// SetWriterFactory sets a custom writer factory +// This allows applications to provide their own writer implementations +func SetWriterFactory(factory message.WriterFactory) { + globalFactory = factory +} + +// ClearWriterCache clears the writer cache +// Should be called when contexts are cleaned up +func ClearWriterCache(ctx *context.Context) { + writerMutex.Lock() + defer writerMutex.Unlock() + delete(writerCache, ctx) +} + +// ClearAllWriterCache clears all cached writers +func ClearAllWriterCache() { + writerMutex.Lock() + defer writerMutex.Unlock() + writerCache = make(map[*context.Context]message.Writer) +} + +// Flush flushes the writer for the given context +func Flush(ctx *context.Context) error { + writer, err := GetWriter(ctx) + if err != nil { + return err + } + return writer.Flush() +} + +// Close closes the writer for the given context and removes it from cache +func Close(ctx *context.Context) error { + writer, err := GetWriter(ctx) + if err != nil { + return err + } + + err = writer.Close() + ClearWriterCache(ctx) + return err +} + +// SendMulti is a convenience function to send multiple messages +func SendMulti(ctx *context.Context, messages ...*message.Message) error { + writer, err := GetWriter(ctx) + if err != nil { + return err + } + + for _, msg := range messages { + if err := writer.Write(msg); err != nil { + return fmt.Errorf("failed to send message: %w", err) + } + } + + return nil +}