From 673e36a7c97ed3e15bf57bf88c070c13fce4de62 Mon Sep 17 00:00:00 2001 From: Max Date: Tue, 25 Nov 2025 22:08:00 +0800 Subject: [PATCH] Refactor: Decouple Output From Context --- agent/assistant/agent.go | 29 +- agent/{llm => assistant}/handlers/stream.go | 44 +- agent/context/JSAPI_OUTPUT.md | 477 ++++++++++++ agent/context/context.go | 14 +- agent/context/interfaces.go | 17 +- agent/context/jsapi.go | 71 ++ agent/context/jsapi_helpers.go | 221 ++++++ agent/context/output.go | 79 ++ agent/context/types.go | 2 + agent/context/types_llm.go | 106 +-- agent/llm/interfaces.go | 7 +- agent/llm/providers/factory.go | 3 +- agent/llm/providers/openai/claude_test.go | 9 +- .../llm/providers/openai/deepseek_r1_test.go | 15 +- .../llm/providers/openai/deepseek_v3_test.go | 5 +- agent/llm/providers/openai/gpt5_test.go | 3 +- agent/llm/providers/openai/openai.go | 61 +- agent/llm/providers/openai/openai_test.go | 43 +- agent/llm/providers/openai/types.go | 13 +- agent/llm/stream.go | 12 - agent/output/adapters/cui/factory.go | 24 - agent/output/adapters/cui/writer.go | 64 +- agent/output/adapters/openai/converter.go | 5 +- agent/output/adapters/openai/factory.go | 28 - agent/output/adapters/openai/writer.go | 85 ++- agent/output/jsapi/output.go | 695 +++++++++--------- agent/output/jsapi/output_test.go | 654 ++++++++-------- agent/output/message/interfaces.go | 16 +- agent/output/message/types.go | 159 ++++ agent/output/output.go | 284 ++++--- main.go | 1 - 31 files changed, 2081 insertions(+), 1165 deletions(-) rename agent/{llm => assistant}/handlers/stream.go (88%) create mode 100644 agent/context/JSAPI_OUTPUT.md create mode 100644 agent/context/jsapi_helpers.go create mode 100644 agent/context/output.go delete mode 100644 agent/llm/stream.go delete mode 100644 agent/output/adapters/cui/factory.go delete mode 100644 agent/output/adapters/openai/factory.go diff --git a/agent/assistant/agent.go b/agent/assistant/agent.go index feb329ed..e85f3179 100644 --- a/agent/assistant/agent.go +++ b/agent/assistant/agent.go @@ -7,17 +7,18 @@ import ( jsoniter "github.com/json-iterator/go" "github.com/yaoapp/gou/connector" "github.com/yaoapp/kun/log" + "github.com/yaoapp/yao/agent/assistant/handlers" "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/llm" - "github.com/yaoapp/yao/agent/output" + "github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/trace/types" "github.com/yaoapp/yao/utils/jsonschema" ) // Stream stream the agent // handler is optional, if not provided, a default handler will be used -func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Message, handler ...context.StreamFunc) (*context.Response, error) { +func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Message, handler ...message.StreamFunc) (*context.Response, error) { log.Trace("[AGENT] Stream started: assistant=%s, contextID=%s", ast.ID, ctx.ID) defer log.Trace("[AGENT] Stream ended: assistant=%s, contextID=%s", ast.ID, ctx.ID) @@ -232,7 +233,7 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa ast.sendAgentStreamEnd(ctx, streamHandler, streamStartTime, "completed", nil, completionResponse) // Close the output writer to send [DONE] marker - if err := output.Close(ctx); err != nil { + if err := ctx.CloseOutput(); err != nil { if trace, _ := ctx.Trace(); trace != nil { trace.Error(i18n.Tr(ast.ID, ctx.Locale, "assistant.agent.stream.close_error"), map[string]any{"error": err.Error()}) // "Failed to close output" } @@ -361,12 +362,12 @@ func (ast *Assistant) BuildRequest(ctx *context.Context, messages []context.Mess } // Info get the assistant information -func (ast *Assistant) Info(locale ...string) *context.AssistantInfo { +func (ast *Assistant) Info(locale ...string) *message.AssistantInfo { lc := "en" if len(locale) > 0 { lc = locale[0] } - return &context.AssistantInfo{ + return &message.AssistantInfo{ ID: ast.ID, Type: ast.Type, Name: i18n.Tr(ast.ID, lc, ast.Name), @@ -688,22 +689,22 @@ func (ast *Assistant) WithHistory(ctx *context.Context, messages []context.Messa } // getStreamHandler returns the stream handler from the provided handlers or a default one -func (ast *Assistant) getStreamHandler(ctx *context.Context, handler ...context.StreamFunc) context.StreamFunc { +func (ast *Assistant) getStreamHandler(ctx *context.Context, handler ...message.StreamFunc) message.StreamFunc { if len(handler) > 0 && handler[0] != nil { return handler[0] } - return llm.DefaultStreamHandler(ctx) + return handlers.DefaultStreamHandler(ctx) } // sendAgentStreamStart sends ChunkStreamStart for root stack only (agent-level stream start) // This ensures only one stream_start per agent execution, even with multiple LLM calls -func (ast *Assistant) sendAgentStreamStart(ctx *context.Context, handler context.StreamFunc, startTime time.Time) { +func (ast *Assistant) sendAgentStreamStart(ctx *context.Context, handler message.StreamFunc, startTime time.Time) { if ctx.Stack == nil || !ctx.Stack.IsRoot() || handler == nil { return } // Build the start data - startData := context.StreamStartData{ + startData := message.StreamStartData{ ContextID: ctx.ID, ChatID: ctx.ChatID, TraceID: ctx.TraceID(), @@ -714,12 +715,12 @@ func (ast *Assistant) sendAgentStreamStart(ctx *context.Context, handler context } if startJSON, err := jsoniter.Marshal(startData); err == nil { - handler(context.ChunkStreamStart, startJSON) + handler(message.ChunkStreamStart, startJSON) } } // sendAgentStreamEnd sends ChunkStreamEnd for root stack only (agent-level stream completion) -func (ast *Assistant) sendAgentStreamEnd(ctx *context.Context, handler context.StreamFunc, startTime time.Time, status string, err error, response *context.CompletionResponse) { +func (ast *Assistant) sendAgentStreamEnd(ctx *context.Context, handler message.StreamFunc, startTime time.Time, status string, err error, response *context.CompletionResponse) { if ctx.Stack == nil || !ctx.Stack.IsRoot() || handler == nil { return } @@ -730,7 +731,7 @@ func (ast *Assistant) sendAgentStreamEnd(ctx *context.Context, handler context.S return } - endData := &context.StreamEndData{ + endData := &message.StreamEndData{ RequestID: ctx.RequestID(), ContextID: ctx.ID, Timestamp: time.Now().UnixMilli(), @@ -749,12 +750,12 @@ func (ast *Assistant) sendAgentStreamEnd(ctx *context.Context, handler context.S } if endJSON, marshalErr := jsoniter.Marshal(endData); marshalErr == nil { - handler(context.ChunkStreamEnd, endJSON) + handler(message.ChunkStreamEnd, endJSON) } } // sendStreamEndOnError sends ChunkStreamEnd with error status for root stack only -func (ast *Assistant) sendStreamEndOnError(ctx *context.Context, handler context.StreamFunc, startTime time.Time, err error) { +func (ast *Assistant) sendStreamEndOnError(ctx *context.Context, handler message.StreamFunc, startTime time.Time, err error) { ast.sendAgentStreamEnd(ctx, handler, startTime, "error", err, nil) } diff --git a/agent/llm/handlers/stream.go b/agent/assistant/handlers/stream.go similarity index 88% rename from agent/llm/handlers/stream.go rename to agent/assistant/handlers/stream.go index 5eb1b725..639a3002 100644 --- a/agent/llm/handlers/stream.go +++ b/agent/assistant/handlers/stream.go @@ -11,7 +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 { +func DefaultStreamHandler(ctx *context.Context) message.StreamFunc { // Create stream state manager state := &streamState{ @@ -20,7 +20,7 @@ func DefaultStreamHandler(ctx *context.Context) context.StreamFunc { currentID: "", } - return func(chunkType context.StreamChunkType, data []byte) int { + return func(chunkType message.StreamChunkType, data []byte) int { trace, _ := ctx.Trace() if trace != nil { trace.Info(i18n.T(ctx.Locale, "llm.handlers.stream.info"), map[string]any{"data": string(data)}) @@ -28,31 +28,31 @@ func DefaultStreamHandler(ctx *context.Context) context.StreamFunc { // Handle different chunk types switch chunkType { - case context.ChunkStreamStart: + case message.ChunkStreamStart: return state.handleStreamStart(data) - case context.ChunkGroupStart: + case message.ChunkGroupStart: return state.handleGroupStart(data) - case context.ChunkText: + case message.ChunkText: return state.handleText(data) - case context.ChunkThinking: + case message.ChunkThinking: return state.handleThinking(data) - case context.ChunkToolCall: + case message.ChunkToolCall: return state.handleToolCall(data) - case context.ChunkMetadata: + case message.ChunkMetadata: return state.handleMetadata(data) - case context.ChunkError: + case message.ChunkError: return state.handleError(data) - case context.ChunkGroupEnd: + case message.ChunkGroupEnd: return state.handleGroupEnd(data) - case context.ChunkStreamEnd: + case message.ChunkStreamEnd: return state.handleStreamEnd(data) default: @@ -75,13 +75,13 @@ type streamState struct { func (s *streamState) handleStreamStart(data []byte) int { // Send event message to indicate stream has started // This is a lifecycle event, CUI clients can show it, OpenAI clients will ignore it - var startData context.StreamStartData + var startData message.StreamStartData err := jsoniter.Unmarshal(data, &startData) if err != nil { log.Error("Failed to unmarshal stream start data: %v", err) } msg := output.NewEventMessage("stream_start", "Stream started", startData) - output.Send(s.ctx, msg) + s.ctx.Send(msg) return 0 } @@ -120,7 +120,7 @@ func (s *streamState) handleText(data []byte) int { }, } - if err := output.Send(s.ctx, msg); err != nil { + if err := s.ctx.Send(msg); err != nil { // Log error but continue streaming return 0 } @@ -155,7 +155,7 @@ func (s *streamState) handleThinking(data []byte) int { }, } - if err := output.Send(s.ctx, msg); err != nil { + if err := s.ctx.Send(msg); err != nil { return 0 } @@ -176,7 +176,7 @@ func (s *streamState) handleToolCall(data []byte) int { }, } - output.Send(s.ctx, msg) + s.ctx.Send(msg) return 0 // Continue } @@ -191,7 +191,7 @@ func (s *streamState) handleMetadata(data []byte) int { func (s *streamState) handleError(data []byte) int { // Send error message msg := output.NewErrorMessage(string(data), "stream_error") - output.Send(s.ctx, msg) + s.ctx.Send(msg) return 1 // Stop streaming on error } @@ -218,7 +218,7 @@ func (s *streamState) handleGroupEnd(data []byte) int { "content": string(s.buffer), }, } - output.Send(s.ctx, msg) + s.ctx.Send(msg) } // Reset state @@ -233,19 +233,19 @@ func (s *streamState) handleGroupEnd(data []byte) int { // handleStreamEnd handles stream end event func (s *streamState) handleStreamEnd(data []byte) int { // Parse the stream end data - var endData context.StreamEndData + var endData message.StreamEndData if err := jsoniter.Unmarshal(data, &endData); err != nil { log.Error("Failed to parse stream_end data: %v", err) - output.Flush(s.ctx) + s.ctx.Flush() return 0 } // Send stream_end event as a message to frontend msg := output.NewEventMessage("stream_end", "Stream completed", endData) - output.Send(s.ctx, msg) + s.ctx.Send(msg) // Flush any remaining data - output.Flush(s.ctx) + s.ctx.Flush() return 0 // Continue (stream will end naturally) } diff --git a/agent/context/JSAPI_OUTPUT.md b/agent/context/JSAPI_OUTPUT.md new file mode 100644 index 00000000..46b67823 --- /dev/null +++ b/agent/context/JSAPI_OUTPUT.md @@ -0,0 +1,477 @@ +# Context Output JS API + +The Context object now provides `Send`, `SendGroup`, and `Flush` methods directly for sending messages to clients from JavaScript. + +## Usage + +### ctx.Send(message) + +Send a single message to the client. + +**Parameters:** + +- `message`: Can be a string (shorthand) or an object + +**String Shorthand:** + +```javascript +// Automatically converts to a text message +ctx.Send("Hello World"); +``` + +**Object Format:** + +```javascript +// Send text message +ctx.Send({ + type: "text", + props: { + content: "Hello from JavaScript", + }, +}); + +// Send loading message +ctx.Send({ + type: "loading", + props: { + message: "Processing...", + }, +}); + +// Send error message +ctx.Send({ + type: "error", + props: { + message: "Something went wrong", + code: "ERR_500", + }, +}); + +// Send custom message +ctx.Send({ + type: "custom_widget", + props: { + data: { foo: "bar" }, + }, +}); +``` + +**Complete Message Object:** + +```javascript +ctx.Send({ + type: "text", + props: { + content: "Hello", + }, + id: "msg_123", // Optional: message ID + delta: true, // Optional: incremental update + done: false, // Optional: whether complete + delta_path: "content", // Optional: update path + delta_action: "append", // Optional: update action (append, replace, merge, set) + group_id: "grp_1", // Optional: message group ID + metadata: { + // Optional: metadata + timestamp: Date.now(), + sequence: 1, + trace_id: "trace_123", + }, +}); +``` + +### ctx.SendGroup(group) + +Send a group of messages to the client. + +**Parameters:** + +- `group`: Message group object + +**Example:** + +```javascript +ctx.SendGroup({ + id: "group_123", + messages: [ + { + type: "text", + props: { content: "First message" }, + }, + { + type: "text", + props: { content: "Second message" }, + }, + ], + metadata: { + timestamp: Date.now(), + }, +}); +``` + +### ctx.Flush() + +Flush the output buffer to ensure all messages are sent to the client. + +**Example:** + +```javascript +ctx.Send("Processing..."); +ctx.Flush(); // Send immediately +``` + +## Complete Examples + +### Using in Hook Functions + +```javascript +/** + * Create hook - called before assistant processes + */ +function Create(input, options) { + const ctx = input.context; + + // Send welcome message + ctx.Send("Welcome to AI Assistant!"); + + // Send loading indicator + ctx.Send({ + type: "loading", + props: { + message: "Thinking...", + }, + }); + + return { messages: input.messages }; +} + +/** + * Done hook - called after assistant completes + */ +function Done(input, output) { + const ctx = input.context; + + // Send completion message + ctx.Send({ + type: "text", + props: { + content: "Processing completed!", + }, + }); + + // Flush output + ctx.Flush(); + + return {}; +} +``` + +### Streaming Response Example + +```javascript +function StreamingResponse(input) { + const ctx = input.context; + + // Send initial message + ctx.Send({ + type: "text", + props: { content: "Starting process" }, + id: "msg_1", + delta: false, + }); + + // Send incremental updates + ctx.Send({ + type: "text", + props: { content: "..." }, + id: "msg_1", + delta: true, + delta_path: "content", + delta_action: "append", + }); + + // Send completion marker + ctx.Send({ + type: "text", + props: { content: "" }, + id: "msg_1", + delta: false, + done: true, + }); + + ctx.Flush(); +} +``` + +### Error Handling Example + +```javascript +function ProcessWithErrorHandling(input) { + const ctx = input.context; + + try { + // Processing logic + ctx.Send("Processing..."); + + // Simulate error + throw new Error("Something went wrong"); + } catch (error) { + // Send error message + ctx.Send({ + type: "error", + props: { + message: error.message, + code: "ERR_PROCESSING", + }, + }); + + ctx.Flush(); + } +} +``` + +### Multi-step Process Example + +```javascript +function MultiStepProcess(input) { + const ctx = input.context; + + // Step 1 + ctx.Send({ + type: "loading", + props: { message: "Step 1: Analyzing input..." }, + }); + ctx.Flush(); + + // ... processing ... + + // Step 2 + ctx.Send({ + type: "loading", + props: { message: "Step 2: Generating response..." }, + }); + ctx.Flush(); + + // ... processing ... + + // Final result + ctx.Send({ + type: "text", + props: { content: "Process completed successfully!" }, + }); + ctx.Flush(); +} +``` + +## Message Types + +Built-in message types supported: + +- `user_input` - User input (display only) +- `text` - Text content (supports Markdown) +- `thinking` - Reasoning/thinking process +- `loading` - Loading indicator +- `tool_call` - Tool/function call +- `error` - Error message +- `image` - Image content +- `audio` - Audio content +- `video` - Video content +- `action` - System action (silent in OpenAI clients) +- `event` - Lifecycle event (CUI only) + +## Message Props by Type + +### Text Message + +```javascript +{ + type: "text", + props: { + content: "Text content (supports Markdown)" + } +} +``` + +### Thinking Message + +```javascript +{ + type: "thinking", + props: { + content: "Reasoning process..." + } +} +``` + +### Loading Message + +```javascript +{ + type: "loading", + props: { + message: "Loading message..." + } +} +``` + +### Tool Call Message + +```javascript +{ + type: "tool_call", + props: { + id: "call_123", + name: "function_name", + arguments: '{"key": "value"}' + } +} +``` + +### Error Message + +```javascript +{ + type: "error", + props: { + message: "Error message", + code: "ERROR_CODE", + details: "Additional details" + } +} +``` + +### Image Message + +```javascript +{ + type: "image", + props: { + url: "https://example.com/image.jpg", + alt: "Image description", + width: 800, + height: 600 + } +} +``` + +### Audio Message + +```javascript +{ + type: "audio", + props: { + url: "https://example.com/audio.mp3", + format: "mp3", + duration: 120.5, + transcript: "Audio transcript...", + autoplay: false, + controls: true + } +} +``` + +### Video Message + +```javascript +{ + type: "video", + props: { + url: "https://example.com/video.mp4", + format: "mp4", + thumbnail: "https://example.com/thumb.jpg", + width: 1920, + height: 1080, + autoplay: false, + controls: true + } +} +``` + +## Delta Updates + +Use delta updates for streaming scenarios: + +```javascript +// Initial message +ctx.Send({ + type: "text", + props: { content: "Hello" }, + id: "msg_1", + delta: false, +}); + +// Append to content +ctx.Send({ + type: "text", + props: { content: " World" }, + id: "msg_1", + delta: true, + delta_path: "content", + delta_action: "append", +}); + +// Mark as complete +ctx.Send({ + type: "text", + props: {}, + id: "msg_1", + done: true, +}); +``` + +**Delta Actions:** + +- `append` - Append to string or array +- `replace` - Replace value +- `merge` - Merge objects +- `set` - Set new field + +## Notes + +1. **No Separate Output API Needed**: The previous `const output = new Output(ctx)` approach is deprecated. Now use `ctx.Send()` methods directly. + +2. **Automatic Client Handling**: Messages are automatically converted to the appropriate format based on `ctx.accept`: + + - `standard` → OpenAI format + - `cui-web`/`cui-native`/`cui-desktop` → CUI native format + +3. **Performance Optimization**: Output objects are automatically cached and managed, no manual management needed. + +4. **Error Handling**: All methods throw JavaScript exceptions on failure, which can be caught with try-catch. + +5. **Streaming Support**: Use delta updates with unique message IDs for real-time streaming scenarios. + +6. **Metadata**: Optional metadata can be attached to messages for tracking, debugging, or custom processing. + +## Migration Guide + +**Before (Deprecated):** + +```javascript +// Old way - no longer needed +const output = new Output(ctx) +output.Send("Hello") +output.SendGroup({ id: "grp1", messages: [...] }) +``` + +**After (Current):** + +```javascript +// New way - simpler and cleaner +ctx.Send("Hello") +ctx.SendGroup({ id: "grp1", messages: [...] }) +ctx.Flush() +``` + +## Best Practices + +1. **Use String Shorthand for Simple Messages**: `ctx.Send("Hello")` instead of `ctx.Send({ type: "text", props: { content: "Hello" } })` + +2. **Always Flush After Important Messages**: Use `ctx.Flush()` to ensure messages are sent immediately + +3. **Use Unique IDs for Delta Updates**: Assign unique IDs to messages that will receive incremental updates + +4. **Handle Errors Gracefully**: Wrap Send operations in try-catch blocks for robust error handling + +5. **Use Loading Indicators**: Show loading messages for long-running operations to improve UX + +6. **Group Related Messages**: Use `SendGroup` for semantically related messages that should be displayed together diff --git a/agent/context/context.go b/agent/context/context.go index 5dd98af5..d559a119 100644 --- a/agent/context/context.go +++ b/agent/context/context.go @@ -157,14 +157,14 @@ func (ctx *Context) Release() { // 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 - } +// func (ctx *Context) Send(data []byte) error { +// if ctx.Writer == nil { +// return nil // No writer, silently ignore +// } - _, err := ctx.Writer.Write(data) - return err -} +// _, err := ctx.Writer.Write(data) +// return err +// } // Trace returns the trace manager for this context, lazily initialized on first call // Uses the TraceID from ctx.Stack if available, or generates a new one diff --git a/agent/context/interfaces.go b/agent/context/interfaces.go index ab6318cb..2eb00e95 100644 --- a/agent/context/interfaces.go +++ b/agent/context/interfaces.go @@ -1,6 +1,10 @@ package context -import "net/http" +import ( + "net/http" + + "github.com/yaoapp/yao/agent/output/message" +) // StreamChunkType represents the type of content in a streaming chunk type StreamChunkType string @@ -23,15 +27,6 @@ const ( ChunkGroupEnd StreamChunkType = "group_end" // Message group ends (text/tool_call/thinking group completes) ) -// StreamFunc the streaming function callback -// Parameters: -// - chunkType: the type of content in this chunk (text, thinking, tool_call, etc.) -// - data: the actual chunk data (could be text, JSON, or other format) -// -// Returns: -// - int: status code (0 = continue, non-zero = stop streaming) -type StreamFunc func(chunkType StreamChunkType, data []byte) int - // Writer is an alias for http.ResponseWriter interface used by an agent to construct a response. // A Writer may not be used after the agent execution has completed. type Writer = http.ResponseWriter @@ -40,7 +35,7 @@ type Writer = http.ResponseWriter type Agent interface { // Stream stream the agent - Stream(ctx *Context, messages []Message, handler StreamFunc) error + Stream(ctx *Context, messages []Message, handler message.StreamFunc) error // Run run the agent Run(ctx *Context, messages []Message) (*Response, error) diff --git a/agent/context/jsapi.go b/agent/context/jsapi.go index fcd4ec53..9c42713b 100644 --- a/agent/context/jsapi.go +++ b/agent/context/jsapi.go @@ -45,6 +45,9 @@ func (ctx *Context) NewObject(v8ctx *v8go.Context) (*v8go.Value, error) { // Set methods jsObject.Set("Trace", ctx.traceMethod(v8ctx.Isolate())) + jsObject.Set("Send", ctx.sendMethod(v8ctx.Isolate())) + jsObject.Set("SendGroup", ctx.sendGroupMethod(v8ctx.Isolate())) + jsObject.Set("Flush", ctx.flushMethod(v8ctx.Isolate())) // Create instance instance, err := jsObject.NewInstance(v8ctx) @@ -157,3 +160,71 @@ func (ctx *Context) traceMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { return traceObj }) } + +// sendMethod implements ctx.Send(message) +// Usage: ctx.Send({ type: "text", props: { content: "Hello" } }) +// Usage: ctx.Send("Hello") // shorthand for text message +func (ctx *Context) sendMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + v8ctx := info.Context() + args := info.Args() + + if len(args) < 1 { + return bridge.JsException(v8ctx, "Send requires a message argument") + } + + // Parse message argument + msg, err := parseMessage(v8ctx, args[0]) + if err != nil { + return bridge.JsException(v8ctx, "invalid message: "+err.Error()) + } + + // Call ctx.Send + if err := ctx.Send(msg); err != nil { + return bridge.JsException(v8ctx, "Send failed: "+err.Error()) + } + + return v8go.Undefined(iso) + }) +} + +// sendGroupMethod implements ctx.SendGroup(group) +// Usage: ctx.SendGroup({ id: "group1", messages: [...] }) +func (ctx *Context) sendGroupMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + v8ctx := info.Context() + args := info.Args() + + if len(args) < 1 { + return bridge.JsException(v8ctx, "SendGroup requires a group argument") + } + + // Parse group argument + group, err := parseGroup(v8ctx, args[0]) + if err != nil { + return bridge.JsException(v8ctx, "invalid group: "+err.Error()) + } + + // Call ctx.SendGroup + if err := ctx.SendGroup(group); err != nil { + return bridge.JsException(v8ctx, "SendGroup failed: "+err.Error()) + } + + return v8go.Undefined(iso) + }) +} + +// flushMethod implements ctx.Flush() +// Usage: ctx.Flush() +func (ctx *Context) flushMethod(iso *v8go.Isolate) *v8go.FunctionTemplate { + return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { + v8ctx := info.Context() + + // Call ctx.Flush + if err := ctx.Flush(); err != nil { + return bridge.JsException(v8ctx, "Flush failed: "+err.Error()) + } + + return v8go.Undefined(iso) + }) +} diff --git a/agent/context/jsapi_helpers.go b/agent/context/jsapi_helpers.go new file mode 100644 index 00000000..6bd5a922 --- /dev/null +++ b/agent/context/jsapi_helpers.go @@ -0,0 +1,221 @@ +package context + +import ( + "fmt" + + "github.com/yaoapp/gou/runtime/v8/bridge" + "github.com/yaoapp/yao/agent/output/message" + "rogchap.com/v8go" +) + +// parseMessage parses a JavaScript value into a message.Message +func parseMessage(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Message, error) { + // Handle string shorthand: convert to text message + if jsValue.IsString() { + return &message.Message{ + Type: message.TypeText, + Props: map[string]interface{}{ + "content": jsValue.String(), + }, + }, nil + } + + // Handle object + if !jsValue.IsObject() { + return nil, fmt.Errorf("message must be a string or object") + } + + // Convert to Go map + goValue, err := bridge.GoValue(jsValue, v8ctx) + if err != nil { + return nil, fmt.Errorf("failed to convert message: %w", err) + } + + msgMap, ok := goValue.(map[string]interface{}) + if !ok { + return nil, fmt.Errorf("message must be an object") + } + + // Build message + msg := &message.Message{} + + // Type field (required) + if msgType, ok := msgMap["type"].(string); ok { + msg.Type = msgType + } else { + return nil, fmt.Errorf("message.type is required and must be a string") + } + + // Props field (optional) + if props, ok := msgMap["props"].(map[string]interface{}); ok { + msg.Props = props + } + + // Optional fields + if id, ok := msgMap["id"].(string); ok { + msg.ID = id + } + if delta, ok := msgMap["delta"].(bool); ok { + msg.Delta = delta + } + if done, ok := msgMap["done"].(bool); ok { + msg.Done = done + } + if deltaPath, ok := msgMap["delta_path"].(string); ok { + msg.DeltaPath = deltaPath + } + if deltaAction, ok := msgMap["delta_action"].(string); ok { + msg.DeltaAction = deltaAction + } + if typeChange, ok := msgMap["type_change"].(bool); ok { + msg.TypeChange = typeChange + } + if groupID, ok := msgMap["group_id"].(string); ok { + msg.GroupID = groupID + } + if groupStart, ok := msgMap["group_start"].(bool); ok { + msg.GroupStart = groupStart + } + if groupEnd, ok := msgMap["group_end"].(bool); ok { + msg.GroupEnd = groupEnd + } + + // Metadata (optional) + if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok { + metadata := &message.Metadata{} + if timestamp, ok := metadataMap["timestamp"].(float64); ok { + metadata.Timestamp = int64(timestamp) + } + if sequence, ok := metadataMap["sequence"].(float64); ok { + metadata.Sequence = int(sequence) + } + if traceID, ok := metadataMap["trace_id"].(string); ok { + metadata.TraceID = traceID + } + msg.Metadata = metadata + } + + return msg, nil +} + +// parseGroup parses a JavaScript value into a message.Group +func parseGroup(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Group, error) { + // Must be an object + if !jsValue.IsObject() { + return nil, fmt.Errorf("group must be an object") + } + + // Convert to Go map + goValue, err := bridge.GoValue(jsValue, v8ctx) + if err != nil { + return nil, fmt.Errorf("failed to convert group: %w", err) + } + + groupMap, ok := goValue.(map[string]interface{}) + if !ok { + return nil, fmt.Errorf("group must be an object") + } + + // Build group + group := &message.Group{} + + // ID field (required) + if id, ok := groupMap["id"].(string); ok { + group.ID = id + } else { + return nil, fmt.Errorf("group.id is required and must be a string") + } + + // Messages field (required) + if messagesArray, ok := groupMap["messages"].([]interface{}); ok { + group.Messages = make([]*message.Message, 0, len(messagesArray)) + for i, msgInterface := range messagesArray { + // Convert to map + msgMap, ok := msgInterface.(map[string]interface{}) + if !ok { + return nil, fmt.Errorf("group.messages[%d] must be an object", i) + } + + // Convert map to Message + msg := &message.Message{} + + // Type field (required) + if msgType, ok := msgMap["type"].(string); ok { + msg.Type = msgType + } else { + return nil, fmt.Errorf("group.messages[%d].type is required", i) + } + + // Props field (optional) + if props, ok := msgMap["props"].(map[string]interface{}); ok { + msg.Props = props + } + + // Optional fields + if id, ok := msgMap["id"].(string); ok { + msg.ID = id + } + if delta, ok := msgMap["delta"].(bool); ok { + msg.Delta = delta + } + if done, ok := msgMap["done"].(bool); ok { + msg.Done = done + } + if deltaPath, ok := msgMap["delta_path"].(string); ok { + msg.DeltaPath = deltaPath + } + if deltaAction, ok := msgMap["delta_action"].(string); ok { + msg.DeltaAction = deltaAction + } + if typeChange, ok := msgMap["type_change"].(bool); ok { + msg.TypeChange = typeChange + } + if groupID, ok := msgMap["group_id"].(string); ok { + msg.GroupID = groupID + } + if groupStart, ok := msgMap["group_start"].(bool); ok { + msg.GroupStart = groupStart + } + if groupEnd, ok := msgMap["group_end"].(bool); ok { + msg.GroupEnd = groupEnd + } + + // Metadata (optional) + if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok { + metadata := &message.Metadata{} + if timestamp, ok := metadataMap["timestamp"].(float64); ok { + metadata.Timestamp = int64(timestamp) + } + if sequence, ok := metadataMap["sequence"].(float64); ok { + metadata.Sequence = int(sequence) + } + if traceID, ok := metadataMap["trace_id"].(string); ok { + metadata.TraceID = traceID + } + msg.Metadata = metadata + } + + group.Messages = append(group.Messages, msg) + } + } else { + return nil, fmt.Errorf("group.messages is required and must be an array") + } + + // Metadata (optional) + if metadataMap, ok := groupMap["metadata"].(map[string]interface{}); ok { + metadata := &message.Metadata{} + if timestamp, ok := metadataMap["timestamp"].(float64); ok { + metadata.Timestamp = int64(timestamp) + } + if sequence, ok := metadataMap["sequence"].(float64); ok { + metadata.Sequence = int(sequence) + } + if traceID, ok := metadataMap["trace_id"].(string); ok { + metadata.TraceID = traceID + } + group.Metadata = metadata + } + + return group, nil +} + diff --git a/agent/context/output.go b/agent/context/output.go new file mode 100644 index 00000000..689744ba --- /dev/null +++ b/agent/context/output.go @@ -0,0 +1,79 @@ +package context + +import ( + "github.com/yaoapp/yao/agent/output" + "github.com/yaoapp/yao/agent/output/message" +) + +// Send sends a message via the output module +func (ctx *Context) Send(msg *message.Message) error { + output, err := ctx.getOutput() + if err != nil { + return err + } + return output.Send(msg) +} + +// SendGroup sends a group of messages via the output module +func (ctx *Context) SendGroup(group *message.Group) error { + output, err := ctx.getOutput() + if err != nil { + return err + } + return output.SendGroup(group) +} + +// Flush flushes the output writer +func (ctx *Context) Flush() error { + output, err := ctx.getOutput() + if err != nil { + return err + } + return output.Flush() +} + +// CloseOutput closes the output writer +func (ctx *Context) CloseOutput() error { + output, err := ctx.getOutput() + if err != nil { + return err + } + return output.Close() +} + +// getOutput gets the output writer for the context +func (ctx *Context) getOutput() (*output.Output, error) { + if ctx.output != nil { + return ctx.output, nil + } + + trace, _ := ctx.Trace() + var options message.Options = message.Options{ + BaseURL: "/", + Writer: ctx.Writer, + Trace: trace, + Locale: ctx.Locale, + Accept: string(ctx.Accept), + } + + // Convert ModelCapabilities to message.ModelCapabilities + if ctx.Capabilities != nil { + options.Capabilities = &message.ModelCapabilities{ + Vision: ctx.Capabilities.Vision, + ToolCalls: ctx.Capabilities.ToolCalls, + Audio: ctx.Capabilities.Audio, + Reasoning: ctx.Capabilities.Reasoning, + Streaming: ctx.Capabilities.Streaming, + JSON: ctx.Capabilities.JSON, + Multimodal: ctx.Capabilities.Multimodal, + TemperatureAdjustable: ctx.Capabilities.TemperatureAdjustable, + } + } + + var err error + ctx.output, err = output.NewOutput(options) + if err != nil { + return nil, err + } + return ctx.output, nil +} diff --git a/agent/context/types.go b/agent/context/types.go index e01e607e..7e1f2d4e 100644 --- a/agent/context/types.go +++ b/agent/context/types.go @@ -6,6 +6,7 @@ import ( "github.com/yaoapp/gou/plan" "github.com/yaoapp/gou/store" + "github.com/yaoapp/yao/agent/output" "github.com/yaoapp/yao/openapi/oauth/types" traceTypes "github.com/yaoapp/yao/trace/types" ) @@ -205,6 +206,7 @@ type Context struct { Writer Writer `json:"-"` // Writer, it will be used to write response data to the client Skip *Skip `json:"skip,omitempty"` // Skip configuration (history, trace, etc.), nil means don't skip anything trace traceTypes.Manager `json:"-"` // Trace manager, lazy initialized on first access + output *output.Output `json:"-"` // Output, it will be used to write response data to the client // Model capabilities (set by assistant, used by output adapters) Capabilities *ModelCapabilities `json:"-"` // Model capabilities for the current connector diff --git a/agent/context/types_llm.go b/agent/context/types_llm.go index 918f5a67..64cc9ea0 100644 --- a/agent/context/types_llm.go +++ b/agent/context/types_llm.go @@ -1,5 +1,7 @@ package context +import "github.com/yaoapp/yao/agent/output/message" + // Uses represents the wrapper configurations for assistant // Used to specify which assistant or MCP server to use for vision, audio, search, and fetch operations type Uses struct { @@ -28,16 +30,7 @@ const ( // ModelCapabilities defines the capabilities of a language model // Used by LLM to select appropriate provider and validate requests -type ModelCapabilities struct { - Vision interface{} `json:"vision,omitempty"` // Supports vision/image input: bool or VisionFormat string ("openai", "claude"/"base64", "default") - ToolCalls *bool `json:"tool_calls,omitempty"` // Supports tool/function calling - Audio *bool `json:"audio,omitempty"` // Supports audio input/output - Reasoning *bool `json:"reasoning,omitempty"` // Supports reasoning/thinking mode (o1, DeepSeek R1) - Streaming *bool `json:"streaming,omitempty"` // Supports streaming responses - JSON *bool `json:"json,omitempty"` // Supports JSON mode - Multimodal *bool `json:"multimodal,omitempty"` // Supports multimodal input (text + images + audio) - TemperatureAdjustable *bool `json:"temperature_adjustable,omitempty"` // Supports temperature adjustment (reasoning models typically don't) -} +type ModelCapabilities message.ModelCapabilities // GetVisionSupport returns whether vision is supported and the format func (m *ModelCapabilities) GetVisionSupport() (bool, VisionFormat) { @@ -140,7 +133,7 @@ type CompletionResponse struct { FinishReason string `json:"finish_reason"` // Why generation stopped (stop, length, tool_calls, content_filter, etc.) // Usage statistics - Usage *UsageInfo `json:"usage,omitempty"` // Token usage statistics + Usage *message.UsageInfo `json:"usage,omitempty"` // Token usage statistics // Additional metadata SystemFingerprint string `json:"system_fingerprint,omitempty"` // System fingerprint for reproducibility @@ -150,32 +143,6 @@ type CompletionResponse struct { Raw interface{} `json:"raw,omitempty"` // Original raw response from the LLM provider } -// UsageInfo represents token usage statistics -// Structure matches OpenAI API: https://platform.openai.com/docs/api-reference/chat/object#chat-object-usage -type UsageInfo struct { - PromptTokens int `json:"prompt_tokens"` // Number of tokens in the prompt - CompletionTokens int `json:"completion_tokens"` // Number of tokens in the generated completion - TotalTokens int `json:"total_tokens"` // Total number of tokens used (prompt + completion) - - // Detailed token breakdown - PromptTokensDetails *PromptTokensDetails `json:"prompt_tokens_details,omitempty"` // Breakdown of tokens used in the prompt - CompletionTokensDetails *CompletionTokensDetails `json:"completion_tokens_details,omitempty"` // Breakdown of tokens used in the completion -} - -// PromptTokensDetails provides detailed breakdown of tokens used in the prompt -type PromptTokensDetails struct { - AudioTokens int `json:"audio_tokens,omitempty"` // Audio input tokens present in the prompt - CachedTokens int `json:"cached_tokens,omitempty"` // Cached tokens present in the prompt -} - -// CompletionTokensDetails provides detailed breakdown of tokens used in the completion -type CompletionTokensDetails struct { - AcceptedPredictionTokens int `json:"accepted_prediction_tokens,omitempty"` // Tokens from predictions that appeared in the completion - AudioTokens int `json:"audio_tokens,omitempty"` // Audio input tokens generated by the model - ReasoningTokens int `json:"reasoning_tokens,omitempty"` // Tokens generated by the model for reasoning (o1, o1-mini, DeepSeek R1) - RejectedPredictionTokens int `json:"rejected_prediction_tokens,omitempty"` // Tokens from predictions that did not appear in the completion -} - // FinishReason constants - why the model stopped generating tokens const ( FinishReasonStop = "stop" // Natural stop point or provided stop sequence reached @@ -210,68 +177,3 @@ type JSONSchema struct { Schema interface{} `json:"schema"` // Required: JSON schema (*jsonschema.Schema or map[string]interface{}) Strict *bool `json:"strict,omitempty"` // Optional: whether to enforce strict schema validation (default: true) } - -// ============================================================================ -// Stream Lifecycle Event Data Structures -// ============================================================================ -// These structures define the data format for stream lifecycle events. -// They provide a standardized way to communicate stream boundaries and metadata -// to the frontend, enabling better UI/UX (progress indicators, timing, etc.). - -// StreamStartData represents the data for stream_start event -// Sent when a streaming request begins -type StreamStartData struct { - ContextID string `json:"context_id"` // Context ID for the response - RequestID string `json:"request_id"` // Unique identifier for this request - Timestamp int64 `json:"timestamp"` // Unix timestamp when stream started - ChatID string `json:"chat_id"` // Chat ID being used (e.g., "chat-123") - TraceID string `json:"trace_id"` // Trace ID being used (e.g., "trace-123") - Assistant *AssistantInfo `json:"assistant,omitempty"` // Assistant information - Metadata map[string]interface{} `json:"metadata,omitempty"` // Metadata to pass to the page for CUI context -} - -// StreamEndData represents the data for stream_end event -// Sent when a streaming request completes (successfully or with error) -type StreamEndData struct { - RequestID string `json:"request_id"` // Corresponding request ID - ContextID string `json:"context_id"` // Context ID for the response - TraceID string `json:"trace_id"` // Trace ID being used (e.g., "trace-123") - Timestamp int64 `json:"timestamp"` // Unix timestamp when stream ended - DurationMs int64 `json:"duration_ms"` // Total duration in milliseconds - Status string `json:"status"` // "completed" | "error" | "cancelled" - Error string `json:"error,omitempty"` // Error message if status is "error" - Usage *UsageInfo `json:"usage,omitempty"` // Token usage statistics - Metadata map[string]interface{} `json:"metadata,omitempty"` // Metadata to pass to the page for CUI context -} - -// GroupStartData represents the data for group_start event -// Sent when a logical message group begins (text, tool_call, thinking, etc.) -type GroupStartData struct { - GroupID string `json:"group_id"` // Unique identifier for this group - Type string `json:"type"` // Group type: "text" | "thinking" | "tool_call" | "refusal" - Timestamp int64 `json:"timestamp"` // Unix timestamp when group started - ToolCall *GroupToolCallInfo `json:"tool_call,omitempty"` // Tool call metadata (if type is "tool_call") - Extra map[string]interface{} `json:"extra,omitempty"` // Additional metadata (for custom providers or future extensions) -} - -// GroupEndData represents the data for group_end event -// Sent when a logical message group completes -type GroupEndData struct { - GroupID string `json:"group_id"` // Corresponding group ID - Type string `json:"type"` // Group type (same as in group_start) - Timestamp int64 `json:"timestamp"` // Unix timestamp when group ended - DurationMs int64 `json:"duration_ms"` // Duration of this group in milliseconds - ChunkCount int `json:"chunk_count"` // Number of data chunks in this group - Status string `json:"status"` // "completed" | "partial" | "error" - ToolCall *GroupToolCallInfo `json:"tool_call,omitempty"` // Complete tool call info (if type is "tool_call") - Extra map[string]interface{} `json:"extra,omitempty"` // Additional metadata -} - -// GroupToolCallInfo contains tool call information for group events -// Used in both group_start (partial info) and group_end (complete info) -type GroupToolCallInfo struct { - ID string `json:"id"` // Tool call ID (e.g., "call_abc123") - Name string `json:"name"` // Function name (may be partial in group_start) - Arguments string `json:"arguments,omitempty"` // Complete arguments (only in group_end) - Index int `json:"index"` // Index in the tool calls array -} diff --git a/agent/llm/interfaces.go b/agent/llm/interfaces.go index 0fd26b58..6f27b6bf 100644 --- a/agent/llm/interfaces.go +++ b/agent/llm/interfaces.go @@ -1,9 +1,12 @@ package llm -import "github.com/yaoapp/yao/agent/context" +import ( + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/message" +) // LLM the LLM interface type LLM interface { - Stream(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler context.StreamFunc) (*context.CompletionResponse, error) + Stream(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler message.StreamFunc) (*context.CompletionResponse, error) Post(ctx *context.Context, messages []context.Message, options *context.CompletionOptions) (*context.CompletionResponse, error) } diff --git a/agent/llm/providers/factory.go b/agent/llm/providers/factory.go index 4927485b..3fa1165c 100644 --- a/agent/llm/providers/factory.go +++ b/agent/llm/providers/factory.go @@ -6,11 +6,12 @@ import ( "github.com/yaoapp/gou/connector" "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/llm/providers/openai" + "github.com/yaoapp/yao/agent/output/message" ) // LLM interface (copied to avoid import cycle) type LLM interface { - Stream(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler context.StreamFunc) (*context.CompletionResponse, error) + Stream(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler message.StreamFunc) (*context.CompletionResponse, error) Post(ctx *context.Context, messages []context.Message, options *context.CompletionOptions) (*context.CompletionResponse, error) } diff --git a/agent/llm/providers/openai/claude_test.go b/agent/llm/providers/openai/claude_test.go index 63ad1819..3b31566f 100644 --- a/agent/llm/providers/openai/claude_test.go +++ b/agent/llm/providers/openai/claude_test.go @@ -8,6 +8,7 @@ import ( "github.com/yaoapp/gou/plan" "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/llm" + "github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/config" "github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/test" @@ -89,7 +90,7 @@ func TestClaudeSonnet4StreamBasic(t *testing.T) { ctx := newClaudeTestContext("test-claude-sonnet4-basic", "claude.sonnet-4_0") var chunks []string - handler := func(chunkType context.StreamChunkType, data []byte) int { + handler := func(chunkType message.StreamChunkType, data []byte) int { chunks = append(chunks, string(data)) t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) return 0 @@ -406,11 +407,11 @@ func TestClaudeSonnet4ThinkingStream(t *testing.T) { var thinkingChunks []string var textChunks []string - handler := func(chunkType context.StreamChunkType, data []byte) int { + handler := func(chunkType message.StreamChunkType, data []byte) int { t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) - if chunkType == context.ChunkThinking { + if chunkType == message.ChunkThinking { thinkingChunks = append(thinkingChunks, string(data)) - } else if chunkType == context.ChunkText { + } else if chunkType == message.ChunkText { textChunks = append(textChunks, string(data)) } return 0 diff --git a/agent/llm/providers/openai/deepseek_r1_test.go b/agent/llm/providers/openai/deepseek_r1_test.go index 93b12d2c..a82f6f84 100644 --- a/agent/llm/providers/openai/deepseek_r1_test.go +++ b/agent/llm/providers/openai/deepseek_r1_test.go @@ -10,6 +10,7 @@ import ( "github.com/yaoapp/gou/plan" "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/llm" + "github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/config" "github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/test" @@ -66,20 +67,20 @@ func TestDeepSeekR1StreamBasic(t *testing.T) { var thinkingGroupEnded bool var textGroupEnded bool - handler := func(chunkType context.StreamChunkType, data []byte) int { + handler := func(chunkType message.StreamChunkType, data []byte) int { dataStr := string(data) t.Logf("Stream chunk [%s]: %s", chunkType, dataStr) // Track different chunk types switch chunkType { - case context.ChunkThinking: + case message.ChunkThinking: reasoningChunks = append(reasoningChunks, dataStr) - case context.ChunkText: + case message.ChunkText: contentChunks = append(contentChunks, dataStr) } // Track group_end events to verify type field - if chunkType == context.ChunkGroupEnd { + if chunkType == message.ChunkGroupEnd { // Parse the group_end data to check the type field var groupEndData struct { GroupID string `json:"group_id"` @@ -327,10 +328,10 @@ func TestDeepSeekR1LogicPuzzle(t *testing.T) { // Track reasoning and content separately var hasReasoning, hasContent bool - handler := func(chunkType context.StreamChunkType, data []byte) int { - if chunkType == context.ChunkThinking && len(data) > 0 { + handler := func(chunkType message.StreamChunkType, data []byte) int { + if chunkType == message.ChunkThinking && len(data) > 0 { hasReasoning = true - } else if chunkType == context.ChunkText && len(data) > 0 { + } else if chunkType == message.ChunkText && len(data) > 0 { hasContent = true } return 0 diff --git a/agent/llm/providers/openai/deepseek_v3_test.go b/agent/llm/providers/openai/deepseek_v3_test.go index 793e9fba..f9194ad6 100644 --- a/agent/llm/providers/openai/deepseek_v3_test.go +++ b/agent/llm/providers/openai/deepseek_v3_test.go @@ -8,6 +8,7 @@ import ( "github.com/yaoapp/gou/plan" "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/llm" + "github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/config" "github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/test" @@ -57,11 +58,11 @@ func TestDeepSeekV3StreamBasic(t *testing.T) { // Track streaming chunks var contentChunks []string - handler := func(chunkType context.StreamChunkType, data []byte) int { + handler := func(chunkType message.StreamChunkType, data []byte) int { dataStr := string(data) t.Logf("Stream chunk [%s]: %s", chunkType, dataStr) - if chunkType == context.ChunkText { + if chunkType == message.ChunkText { contentChunks = append(contentChunks, dataStr) } diff --git a/agent/llm/providers/openai/gpt5_test.go b/agent/llm/providers/openai/gpt5_test.go index a4bc4d90..5e2d709c 100644 --- a/agent/llm/providers/openai/gpt5_test.go +++ b/agent/llm/providers/openai/gpt5_test.go @@ -8,6 +8,7 @@ import ( "github.com/yaoapp/gou/plan" "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/llm" + "github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/config" "github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/test" @@ -52,7 +53,7 @@ func TestGPT5StreamBasic(t *testing.T) { ctx := newGPT5TestContext("test-gpt5-basic", "openai.gpt-5") var chunks []string - handler := func(chunkType context.StreamChunkType, data []byte) int { + handler := func(chunkType message.StreamChunkType, data []byte) int { chunks = append(chunks, string(data)) t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) return 0 diff --git a/agent/llm/providers/openai/openai.go b/agent/llm/providers/openai/openai.go index c3a0dae5..4aaffade 100644 --- a/agent/llm/providers/openai/openai.go +++ b/agent/llm/providers/openai/openai.go @@ -14,11 +14,12 @@ import ( "github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/llm/adapters" "github.com/yaoapp/yao/agent/llm/providers/base" + "github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/utils/jsonschema" ) // startGroup starts a new group and sends group_start event -func (gt *groupTracker) startGroup(groupType context.StreamChunkType, handler context.StreamFunc) { +func (gt *groupTracker) startGroup(groupType message.StreamChunkType, handler message.StreamFunc) { if gt.active { // End previous group first gt.endGroup(handler) @@ -32,39 +33,39 @@ func (gt *groupTracker) startGroup(groupType context.StreamChunkType, handler co gt.toolCallInfo = nil if handler != nil { - startData := &context.GroupStartData{ + startData := &message.GroupStartData{ GroupID: gt.groupID, Type: string(groupType), Timestamp: gt.startTime, } if startJSON, err := jsoniter.Marshal(startData); err == nil { - handler(context.ChunkGroupStart, startJSON) + handler(message.ChunkGroupStart, startJSON) } } } // startToolCallGroup starts a new tool call group with tool call info -func (gt *groupTracker) startToolCallGroup(toolCallInfo *context.GroupToolCallInfo, handler context.StreamFunc) { +func (gt *groupTracker) startToolCallGroup(toolCallInfo *message.GroupToolCallInfo, handler message.StreamFunc) { if gt.active { gt.endGroup(handler) } gt.active = true gt.groupID = fmt.Sprintf("grp_tool_%d", time.Now().UnixNano()) - gt.groupType = context.ChunkToolCall + gt.groupType = message.ChunkToolCall gt.startTime = time.Now().UnixMilli() gt.chunkCount = 0 gt.toolCallInfo = toolCallInfo if handler != nil { - startData := &context.GroupStartData{ + startData := &message.GroupStartData{ GroupID: gt.groupID, - Type: string(context.ChunkToolCall), + Type: string(message.ChunkToolCall), Timestamp: gt.startTime, ToolCall: toolCallInfo, } if startJSON, err := jsoniter.Marshal(startData); err == nil { - handler(context.ChunkGroupStart, startJSON) + handler(message.ChunkGroupStart, startJSON) } } } @@ -77,13 +78,13 @@ func (gt *groupTracker) incrementChunk() { } // endGroup ends the current group and sends group_end event -func (gt *groupTracker) endGroup(handler context.StreamFunc) { +func (gt *groupTracker) endGroup(handler message.StreamFunc) { if !gt.active { return } if handler != nil { - endData := &context.GroupEndData{ + endData := &message.GroupEndData{ GroupID: gt.groupID, Type: string(gt.groupType), Timestamp: time.Now().UnixMilli(), @@ -95,7 +96,7 @@ func (gt *groupTracker) endGroup(handler context.StreamFunc) { endData.ToolCall = gt.toolCallInfo } if endJSON, err := jsoniter.Marshal(endData); err == nil { - handler(context.ChunkGroupEnd, endJSON) + handler(message.ChunkGroupEnd, endJSON) } } @@ -186,7 +187,7 @@ func detectReasoningFormat(cap *context.ModelCapabilities) adapters.ReasoningFor } // Stream stream completion from OpenAI API -func (p *Provider) Stream(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler context.StreamFunc) (*context.CompletionResponse, error) { +func (p *Provider) Stream(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler message.StreamFunc) (*context.CompletionResponse, error) { // Add debug log trace, _ := ctx.Trace() if trace != nil { @@ -338,7 +339,7 @@ func (p *Provider) Stream(ctx *context.Context, messages []context.Message, opti } // streamWithRetry performs a single streaming request attempt -func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler context.StreamFunc) (*context.CompletionResponse, error) { +func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler message.StreamFunc) (*context.CompletionResponse, error) { streamStartTime := time.Now() requestID := fmt.Sprintf("req_%d", streamStartTime.UnixNano()) @@ -381,7 +382,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess newMessages, err := adapter.PreprocessMessages(processedMessages) if err != nil { if handler != nil { - handler(context.ChunkError, []byte(fmt.Sprintf("adapter %s message preprocessing failed: %v", adapter.Name(), err))) + handler(message.ChunkError, []byte(fmt.Sprintf("adapter %s message preprocessing failed: %v", adapter.Name(), err))) } return nil, fmt.Errorf("adapter %s message preprocessing failed: %w", adapter.Name(), err) } @@ -392,7 +393,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess if err != nil { // Send error to handler if handler != nil { - handler(context.ChunkError, []byte(fmt.Sprintf("adapter %s option preprocessing failed: %v", adapter.Name(), err))) + handler(message.ChunkError, []byte(fmt.Sprintf("adapter %s option preprocessing failed: %v", adapter.Name(), err))) } return nil, fmt.Errorf("adapter %s option preprocessing failed: %w", adapter.Name(), err) } @@ -518,13 +519,13 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess // Handle reasoning content (DeepSeek R1) if delta.ReasoningContent != "" { // Start thinking group if not active - if !groupTracker.active || groupTracker.groupType != context.ChunkThinking { - groupTracker.startGroup(context.ChunkThinking, handler) + if !groupTracker.active || groupTracker.groupType != message.ChunkThinking { + groupTracker.startGroup(message.ChunkThinking, handler) } accumulator.reasoningContent += delta.ReasoningContent if handler != nil { - handler(context.ChunkThinking, []byte(delta.ReasoningContent)) + handler(message.ChunkThinking, []byte(delta.ReasoningContent)) groupTracker.incrementChunk() } } @@ -532,13 +533,13 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess // Handle content if delta.Content != "" { // Start text group if not active - if !groupTracker.active || groupTracker.groupType != context.ChunkText { - groupTracker.startGroup(context.ChunkText, handler) + if !groupTracker.active || groupTracker.groupType != message.ChunkText { + groupTracker.startGroup(message.ChunkText, handler) } accumulator.content += delta.Content if handler != nil { - handler(context.ChunkText, []byte(delta.Content)) + handler(message.ChunkText, []byte(delta.Content)) groupTracker.incrementChunk() } } @@ -546,13 +547,13 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess // Handle refusal if delta.Refusal != "" { // Start refusal group if not active - if !groupTracker.active || groupTracker.groupType != context.ChunkRefusal { - groupTracker.startGroup(context.ChunkRefusal, handler) + if !groupTracker.active || groupTracker.groupType != message.ChunkRefusal { + groupTracker.startGroup(message.ChunkRefusal, handler) } accumulator.refusal += delta.Refusal if handler != nil { - handler(context.ChunkRefusal, []byte(delta.Refusal)) + handler(message.ChunkRefusal, []byte(delta.Refusal)) groupTracker.incrementChunk() } } @@ -565,7 +566,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess // Start new tool call group when we first see this tool call if tc.ID != "" { - toolCallInfo := &context.GroupToolCallInfo{ + toolCallInfo := &message.GroupToolCallInfo{ ID: tc.ID, Name: tc.Function.Name, // May be partial or empty initially Index: tc.Index, @@ -600,7 +601,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess // Notify handler of tool call progress if handler != nil { toolCallData, _ := jsoniter.Marshal(delta.ToolCalls) - handler(context.ChunkToolCall, toolCallData) + handler(message.ChunkToolCall, toolCallData) groupTracker.incrementChunk() } } @@ -612,7 +613,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess // Handle usage (in choices, for older API versions) if chunk.Usage != nil { - accumulator.usage = &context.UsageInfo{ + accumulator.usage = &message.UsageInfo{ PromptTokens: chunk.Usage.PromptTokens, CompletionTokens: chunk.Usage.CompletionTokens, TotalTokens: chunk.Usage.TotalTokens, @@ -622,7 +623,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess // Check for usage at the top level (newer API versions with stream_options) if chunk.Usage != nil && accumulator.usage == nil { - accumulator.usage = &context.UsageInfo{ + accumulator.usage = &message.UsageInfo{ PromptTokens: chunk.Usage.PromptTokens, CompletionTokens: chunk.Usage.CompletionTokens, TotalTokens: chunk.Usage.TotalTokens, @@ -718,7 +719,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess // Notify handler of error if provided if handler != nil { errData := []byte(err.Error()) - handler(context.ChunkError, errData) + handler(message.ChunkError, errData) } return nil, fmt.Errorf("streaming request failed: %w", err) } @@ -747,7 +748,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess // Notify handler of error if provided if handler != nil { errData := []byte(err.Error()) - handler(context.ChunkError, errData) + handler(message.ChunkError, errData) } return nil, err } diff --git a/agent/llm/providers/openai/openai_test.go b/agent/llm/providers/openai/openai_test.go index f437cac9..af0aeaca 100644 --- a/agent/llm/providers/openai/openai_test.go +++ b/agent/llm/providers/openai/openai_test.go @@ -11,6 +11,7 @@ import ( "github.com/yaoapp/gou/plan" "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/llm" + "github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/config" "github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/test" @@ -58,7 +59,7 @@ func TestOpenAIStreamBasic(t *testing.T) { // Track streaming chunks var chunks []string - handler := func(chunkType context.StreamChunkType, data []byte) int { + handler := func(chunkType message.StreamChunkType, data []byte) int { chunks = append(chunks, string(data)) t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) return 0 // Continue @@ -243,8 +244,8 @@ func TestOpenAIStreamWithToolCalls(t *testing.T) { // Track streaming chunks var toolCallChunks int - handler := func(chunkType context.StreamChunkType, data []byte) int { - if chunkType == context.ChunkToolCall { + handler := func(chunkType message.StreamChunkType, data []byte) int { + if chunkType == message.ChunkToolCall { toolCallChunks++ } t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) @@ -482,7 +483,7 @@ func TestOpenAIStreamWithInvalidToolCall(t *testing.T) { // Create context ctx := newTestContext("test-stream-basic", "openai.gpt-4o") - handler := func(chunkType context.StreamChunkType, data []byte) int { + handler := func(chunkType message.StreamChunkType, data []byte) int { return 0 // Continue } @@ -604,8 +605,8 @@ func TestOpenAIStreamChunkTypes(t *testing.T) { ctx := newTestContext("test-chunk-types", "openai.gpt-4o") // Track chunk types - chunkTypes := make(map[context.StreamChunkType]int) - handler := func(chunkType context.StreamChunkType, data []byte) int { + chunkTypes := make(map[message.StreamChunkType]int) + handler := func(chunkType message.StreamChunkType, data []byte) int { chunkTypes[chunkType]++ t.Logf("Received chunk type: %s, data length: %d", chunkType, len(data)) return 1 // Continue @@ -621,7 +622,7 @@ func TestOpenAIStreamChunkTypes(t *testing.T) { } // Validate chunk types received - if chunkTypes[context.ChunkText] == 0 { + if chunkTypes[message.ChunkText] == 0 { t.Error("Expected to receive ChunkText, but got 0") } @@ -673,8 +674,8 @@ func TestOpenAIStreamErrorCallback(t *testing.T) { // Track if error chunk was received receivedError := false var errorMessage string - handler := func(chunkType context.StreamChunkType, data []byte) int { - if chunkType == context.ChunkError { + handler := func(chunkType message.StreamChunkType, data []byte) int { + if chunkType == message.ChunkError { receivedError = true errorMessage = string(data) t.Logf("Received error chunk: %s", errorMessage) @@ -1287,19 +1288,19 @@ func TestOpenAIStreamLifecycleEvents(t *testing.T) { var events []string var groupStartReceived, groupEndReceived bool - handler := func(chunkType context.StreamChunkType, data []byte) int { + handler := func(chunkType message.StreamChunkType, data []byte) int { events = append(events, string(chunkType)) switch chunkType { - case context.ChunkStreamStart: + case message.ChunkStreamStart: t.Error("❌ LLM layer should NOT send stream_start (now sent at Agent level)") - case context.ChunkStreamEnd: + case message.ChunkStreamEnd: t.Error("❌ LLM layer should NOT send stream_end (now sent at Agent level)") - case context.ChunkGroupStart: + case message.ChunkGroupStart: groupStartReceived = true - var startData context.GroupStartData + var startData message.GroupStartData if err := json.Unmarshal(data, &startData); err == nil { t.Logf("✓ group_start: type=%s, group_id=%s", startData.Type, startData.GroupID) if startData.GroupID == "" { @@ -1309,9 +1310,9 @@ func TestOpenAIStreamLifecycleEvents(t *testing.T) { t.Errorf("Failed to parse group_start data: %v", err) } - case context.ChunkGroupEnd: + case message.ChunkGroupEnd: groupEndReceived = true - var endData context.GroupEndData + var endData message.GroupEndData if err := json.Unmarshal(data, &endData); err == nil { t.Logf("✓ group_end: type=%s, chunks=%d, duration=%dms", endData.Type, endData.ChunkCount, endData.DurationMs) @@ -1322,7 +1323,7 @@ func TestOpenAIStreamLifecycleEvents(t *testing.T) { t.Errorf("Failed to parse group_end data: %v", err) } - case context.ChunkText: + case message.ChunkText: t.Logf(" text chunk: %s", string(data)) } @@ -1394,12 +1395,12 @@ func TestOpenAIStreamContextCancellation(t *testing.T) { var receivedChunks int - handler := func(chunkType context.StreamChunkType, data []byte) int { - if chunkType == context.ChunkText || chunkType == context.ChunkToolCall { + handler := func(chunkType message.StreamChunkType, data []byte) int { + if chunkType == message.ChunkText || chunkType == message.ChunkToolCall { receivedChunks++ } // Note: stream_end is now sent at Agent level, not LLM level - if chunkType == context.ChunkStreamEnd { + if chunkType == message.ChunkStreamEnd { t.Error("❌ LLM layer should NOT send stream_end (now sent at Agent level)") } return 0 @@ -1467,7 +1468,7 @@ func TestOpenAIStreamWithTemperature(t *testing.T) { // Use callback to collect chunks chunkCount := 0 - var callback context.StreamFunc = func(chunkType context.StreamChunkType, data []byte) int { + var callback message.StreamFunc = func(chunkType message.StreamChunkType, data []byte) int { chunkCount++ return 1 // Continue } diff --git a/agent/llm/providers/openai/types.go b/agent/llm/providers/openai/types.go index 03a871a2..0efa811f 100644 --- a/agent/llm/providers/openai/types.go +++ b/agent/llm/providers/openai/types.go @@ -1,6 +1,9 @@ package openai -import "github.com/yaoapp/yao/agent/context" +import ( + "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/message" +) // StreamChunk represents a chunk from OpenAI's streaming response type StreamChunk struct { @@ -63,7 +66,7 @@ type CompletionResponseFull struct { } `json:"message"` FinishReason string `json:"finish_reason"` } `json:"choices"` - Usage *context.UsageInfo `json:"usage,omitempty"` + Usage *message.UsageInfo `json:"usage,omitempty"` SystemFingerprint string `json:"system_fingerprint,omitempty"` } @@ -78,7 +81,7 @@ type streamAccumulator struct { refusal string toolCalls map[int]*accumulatedToolCall finishReason string - usage *context.UsageInfo + usage *message.UsageInfo } // accumulatedToolCall accumulates a single tool call @@ -93,8 +96,8 @@ type accumulatedToolCall struct { type groupTracker struct { active bool // Whether a group is currently active groupID string // Current group ID - groupType context.StreamChunkType // Current group type + groupType message.StreamChunkType // Current group type startTime int64 // Group start timestamp chunkCount int // Number of chunks in this group - toolCallInfo *context.GroupToolCallInfo // Tool call info if group is tool_call type + toolCallInfo *message.GroupToolCallInfo // Tool call info if group is tool_call type } diff --git a/agent/llm/stream.go b/agent/llm/stream.go deleted file mode 100644 index 3ac5a185..00000000 --- a/agent/llm/stream.go +++ /dev/null @@ -1,12 +0,0 @@ -package llm - -import ( - "github.com/yaoapp/yao/agent/context" - "github.com/yaoapp/yao/agent/llm/handlers" -) - -// DefaultStreamHandler creates a default stream handler -// This is a convenience function that wraps handlers.DefaultStreamHandler -func DefaultStreamHandler(ctx *context.Context) context.StreamFunc { - return handlers.DefaultStreamHandler(ctx) -} diff --git a/agent/output/adapters/cui/factory.go b/agent/output/adapters/cui/factory.go deleted file mode 100644 index b879aef6..00000000 --- a/agent/output/adapters/cui/factory.go +++ /dev/null @@ -1,24 +0,0 @@ -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 index cc173148..07f2088f 100644 --- a/agent/output/adapters/cui/writer.go +++ b/agent/output/adapters/cui/writer.go @@ -2,22 +2,27 @@ package cui import ( "encoding/json" + "net/http" - "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/output/message" + traceTypes "github.com/yaoapp/yao/trace/types" ) // Writer implements the message.Writer interface for CUI clients type Writer struct { - ctx *context.Context + Writer http.ResponseWriter + Trace traceTypes.Manager + Locale string adapter *Adapter } // NewWriter creates a new CUI writer -func NewWriter(ctx *context.Context) (*Writer, error) { +func NewWriter(options message.Options) (*Writer, error) { return &Writer{ - ctx: ctx, + Writer: options.Writer, + Trace: options.Trace, + Locale: options.Locale, adapter: NewAdapter(), }, nil } @@ -27,8 +32,8 @@ func (w *Writer) Write(msg *message.Message) error { // CUI adapter passes messages through as-is chunks, err := w.adapter.Adapt(msg) if err != nil { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.cui.writer.adapt_error"), map[string]any{ // "CUI Writer: Failed to adapt message" + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.cui.writer.adapt_error"), map[string]any{ // "CUI Writer: Failed to adapt message" "error": err.Error(), "message_type": msg.Type, }) @@ -39,8 +44,8 @@ func (w *Writer) Write(msg *message.Message) error { // Send each chunk for _, chunk := range chunks { if err := w.sendChunk(chunk); err != nil { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.cui.writer.chunk_error"), map[string]any{"error": err.Error()}) // "CUI Writer: Failed to send chunk" + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.cui.writer.chunk_error"), map[string]any{"error": err.Error()}) // "CUI Writer: Failed to send chunk" } return err } @@ -56,8 +61,8 @@ func (w *Writer) WriteGroup(group *message.Group) error { // Send the group if err := w.sendChunk(group); err != nil { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.cui.writer.group_error"), map[string]any{ // "CUI Writer: Failed to send message group" + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.cui.writer.group_error"), map[string]any{ // "CUI Writer: Failed to send message group" "error": err.Error(), "group_id": group.ID, }) @@ -86,15 +91,15 @@ func (w *Writer) sendChunk(chunk interface{}) error { // Convert chunk to JSON data, err := json.Marshal(chunk) if err != nil { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.cui.writer.marshal_error"), map[string]any{"error": err.Error()}) // "CUI Writer: Failed to marshal chunk" + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.cui.writer.marshal_error"), map[string]any{"error": err.Error()}) // "CUI Writer: Failed to marshal chunk" } return err } // Log outgoing data to trace for debugging - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Debug("CUI Writer: Sending chunk to client", map[string]any{ + if w.Trace != nil { + w.Trace.Debug("CUI Writer: Sending chunk to client", map[string]any{ "data": string(data), }) } @@ -106,18 +111,31 @@ func (w *Writer) sendChunk(chunk interface{}) error { // Send via context's writer // The context knows how to send data based on the connection type (SSE, WebSocket, etc.) - if err := w.ctx.Send(sseData); err != nil { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.cui.writer.send_error"), map[string]any{"error": err.Error()}) // "CUI Writer: Failed to send data to client" + if err := w.sendData(sseData); err != nil { + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.cui.writer.send_error"), map[string]any{"error": err.Error()}) // "CUI Writer: Failed to send data to client" } return err } - // Flush immediately to ensure real-time streaming - // Cast to http.ResponseWriter and call Flush if available - if flusher, ok := w.ctx.Writer.(interface{ Flush() }); ok { - flusher.Flush() - } - + w.flush() return nil } + +func (w *Writer) flush() error { + if w.Writer == nil { + return nil // No writer, silently ignore + } + if flusher, ok := w.Writer.(interface{ Flush() }); ok { + flusher.Flush() + } + return nil +} + +func (w *Writer) sendData(data []byte) error { + if w.Writer == nil { + return nil // No writer, silently ignore + } + _, err := w.Writer.Write(data) + return err +} diff --git a/agent/output/adapters/openai/converter.go b/agent/output/adapters/openai/converter.go index 9eb7faea..91914a2e 100644 --- a/agent/output/adapters/openai/converter.go +++ b/agent/output/adapters/openai/converter.go @@ -4,7 +4,6 @@ import ( "fmt" "time" - "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/output/message" ) @@ -152,9 +151,9 @@ func convertStreamStart(msg *message.Message, config *AdapterConfig) ([]interfac } // Try to convert to StreamStartData - var startData context.StreamStartData + var startData message.StreamStartData switch v := data.(type) { - case context.StreamStartData: + case message.StreamStartData: startData = v case map[string]interface{}: // If it's a map, try to extract traceID diff --git a/agent/output/adapters/openai/factory.go b/agent/output/adapters/openai/factory.go deleted file mode 100644 index eecb720e..00000000 --- a/agent/output/adapters/openai/factory.go +++ /dev/null @@ -1,28 +0,0 @@ -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/writer.go b/agent/output/adapters/openai/writer.go index cbac69ec..40ab8be8 100644 --- a/agent/output/adapters/openai/writer.go +++ b/agent/output/adapters/openai/writer.go @@ -2,45 +2,49 @@ package openai import ( "encoding/json" + "net/http" - "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/output/message" + traceTypes "github.com/yaoapp/yao/trace/types" ) // Writer implements the message.Writer interface for OpenAI-compatible clients type Writer struct { - ctx *context.Context + Writer http.ResponseWriter + Trace traceTypes.Manager + Locale string adapter *Adapter firstChunk bool // Track if this is the first chunk to add role } // NewWriter creates a new OpenAI writer -func NewWriter(ctx *context.Context) (*Writer, error) { +func NewWriter(options message.Options) (*Writer, error) { // Get model capabilities from context (set by assistant) var capabilities *ModelCapabilities - if ctx.Capabilities != nil && ctx.Capabilities.Reasoning != nil { + if options.Capabilities != nil && options.Capabilities.Reasoning != nil { capabilities = &ModelCapabilities{ - Reasoning: ctx.Capabilities.Reasoning, + Reasoning: options.Capabilities.Reasoning, } } // Create adapter with capabilities, base URL, and locale adapter := NewAdapter( WithCapabilities(capabilities), - WithBaseURL(getBaseURL(ctx)), - WithLocale(ctx.Locale), + WithBaseURL(getBaseURL(options.BaseURL)), + WithLocale(options.Locale), ) return &Writer{ - ctx: ctx, adapter: adapter, + Writer: options.Writer, + Locale: options.Locale, firstChunk: true, // First chunk should include role }, nil } // getBaseURL gets the base URL from context or environment -func getBaseURL(ctx *context.Context) string { +func getBaseURL(baseURL string) string { // @todo: get from context metadata return "http://localhost:8000/__yao_admin_root" @@ -60,8 +64,8 @@ func (w *Writer) Write(msg *message.Message) error { // Convert message to OpenAI format using adapter chunks, err := w.adapter.Adapt(msg) if err != nil { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.openai.writer.adapt_error"), map[string]any{ // "OpenAI Writer: Failed to adapt message" + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.openai.writer.adapt_error"), map[string]any{ // "OpenAI Writer: Failed to adapt message" "error": err.Error(), "message_type": msg.Type, }) @@ -84,8 +88,8 @@ func (w *Writer) Write(msg *message.Message) error { } if err := w.sendChunk(chunk); err != nil { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.openai.writer.chunk_error"), map[string]any{"error": err.Error()}) // "OpenAI Writer: Failed to send chunk" + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.openai.writer.chunk_error"), map[string]any{"error": err.Error()}) // "OpenAI Writer: Failed to send chunk" } return err } @@ -100,8 +104,8 @@ func (w *Writer) WriteGroup(group *message.Group) error { // Just send each message individually for _, msg := range group.Messages { if err := w.Write(msg); err != nil { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.openai.writer.group_error"), map[string]any{ // "OpenAI Writer: Failed to write message in group" + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.openai.writer.group_error"), map[string]any{ // "OpenAI Writer: Failed to write message in group" "error": err.Error(), "group_id": group.ID, "message_type": msg.Type, @@ -127,13 +131,31 @@ func (w *Writer) Close() error { return w.sendDone() } +func (w *Writer) sendData(data []byte) error { + if w.Writer == nil { + return nil // No writer, silently ignore + } + _, err := w.Writer.Write(data) + return err +} + +func (w *Writer) flush() error { + if w.Writer == nil { + return nil // No writer, silently ignore + } + if flusher, ok := w.Writer.(interface{ Flush() }); ok { + flusher.Flush() + } + return nil +} + // 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 { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.openai.writer.marshal_error"), map[string]any{"error": err.Error()}) // "OpenAI Writer: Failed to marshal chunk" + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.openai.writer.marshal_error"), map[string]any{"error": err.Error()}) // "OpenAI Writer: Failed to marshal chunk" } return err } @@ -143,25 +165,23 @@ func (w *Writer) sendChunk(chunk interface{}) error { sseData = append(sseData, []byte("\n\n")...) // Log outgoing data to trace for debugging - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Debug("OpenAI Writer: Sending chunk to client", map[string]any{ + if w.Trace != nil { + w.Trace.Debug("OpenAI Writer: Sending chunk to client", map[string]any{ "data": string(data), }) } // Send via context's writer - if err := w.ctx.Send(sseData); err != nil { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.openai.writer.send_error"), map[string]any{"error": err.Error()}) // "OpenAI Writer: Failed to send data to client" + if err := w.sendData(sseData); err != nil { + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.openai.writer.send_error"), map[string]any{"error": err.Error()}) // "OpenAI Writer: Failed to send data to client" } return err } // Flush immediately to ensure real-time streaming // Cast to http.ResponseWriter and call Flush if available - if flusher, ok := w.ctx.Writer.(interface{ Flush() }); ok { - flusher.Flush() - } + w.flush() return nil } @@ -169,23 +189,20 @@ func (w *Writer) sendChunk(chunk interface{}) error { // sendDone sends the final [DONE] message func (w *Writer) sendDone() error { // Log completion to trace - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Debug("OpenAI Writer: Sending [DONE] to client") + if w.Trace != nil { + w.Trace.Debug("OpenAI Writer: Sending [DONE] to client") } // OpenAI SSE format uses "data: [DONE]\n\n" to signal completion doneData := []byte("data: [DONE]\n\n") - if err := w.ctx.Send(doneData); err != nil { - if trace, _ := w.ctx.Trace(); trace != nil { - trace.Error(i18n.T(w.ctx.Locale, "output.openai.writer.done_error"), map[string]any{"error": err.Error()}) // "OpenAI Writer: Failed to send [DONE] to client" + if err := w.sendData(doneData); err != nil { + if w.Trace != nil { + w.Trace.Error(i18n.T(w.Locale, "output.openai.writer.done_error"), map[string]any{"error": err.Error()}) // "OpenAI Writer: Failed to send [DONE] to client" } return err } // Flush the final [DONE] message - if flusher, ok := w.ctx.Writer.(interface{ Flush() }); ok { - flusher.Flush() - } - + w.flush() return nil } diff --git a/agent/output/jsapi/output.go b/agent/output/jsapi/output.go index 07acbbec..0ac7534f 100644 --- a/agent/output/jsapi/output.go +++ b/agent/output/jsapi/output.go @@ -1,412 +1,401 @@ package jsapi -import ( - "fmt" +// func init() { +// // Auto-register Output JavaScript API when package is imported +// v8.RegisterFunction("Output", ExportFunction) +// } - v8 "github.com/yaoapp/gou/runtime/v8" - "github.com/yaoapp/gou/runtime/v8/bridge" - agentContext "github.com/yaoapp/yao/agent/context" - "github.com/yaoapp/yao/agent/output" - "github.com/yaoapp/yao/agent/output/message" - "rogchap.com/v8go" -) +// // Usage from JavaScript: +// // +// // const output = new Output(ctx) +// // output.Send({ type: "text", props: { content: "Hello" } }) +// // output.Send("Hello") // shorthand for text message +// // output.SendGroup({ id: "group1", messages: [...] }) +// // +// // Objects: +// // - Output: Output manager (constructor) -func init() { - // Auto-register Output JavaScript API when package is imported - v8.RegisterFunction("Output", ExportFunction) -} +// // ExportFunction exports the Output constructor function template +// // This is used by v8.RegisterFunction +// func ExportFunction(iso *v8go.Isolate) *v8go.FunctionTemplate { +// return v8go.NewFunctionTemplate(iso, outputConstructor) +// } -// Usage from JavaScript: -// -// const output = new Output(ctx) -// output.Send({ type: "text", props: { content: "Hello" } }) -// output.Send("Hello") // shorthand for text message -// output.SendGroup({ id: "group1", messages: [...] }) -// -// Objects: -// - Output: Output manager (constructor) +// // outputConstructor is the JavaScript constructor for Output +// // Usage: new Output(ctx) +// func outputConstructor(info *v8go.FunctionCallbackInfo) *v8go.Value { +// v8ctx := info.Context() +// args := info.Args() -// ExportFunction exports the Output constructor function template -// This is used by v8.RegisterFunction -func ExportFunction(iso *v8go.Isolate) *v8go.FunctionTemplate { - return v8go.NewFunctionTemplate(iso, outputConstructor) -} +// // Require ctx argument +// if len(args) < 1 { +// return bridge.JsException(v8ctx, "Output constructor requires a context argument") +// } -// outputConstructor is the JavaScript constructor for Output -// Usage: new Output(ctx) -func outputConstructor(info *v8go.FunctionCallbackInfo) *v8go.Value { - v8ctx := info.Context() - args := info.Args() +// // Get the context object from JavaScript +// ctxObj, err := args[0].AsObject() +// if err != nil { +// return bridge.JsException(v8ctx, fmt.Sprintf("context must be an object: %s", err)) +// } - // Require ctx argument - if len(args) < 1 { - return bridge.JsException(v8ctx, "Output constructor requires a context argument") - } +// // Get the goValueID from internal field (index 0) +// if ctxObj.InternalFieldCount() < 1 { +// return bridge.JsException(v8ctx, "context object is missing internal fields") +// } - // Get the context object from JavaScript - ctxObj, err := args[0].AsObject() - if err != nil { - return bridge.JsException(v8ctx, fmt.Sprintf("context must be an object: %s", err)) - } +// goValueIDValue := ctxObj.GetInternalField(0) +// if goValueIDValue == nil || !goValueIDValue.IsString() { +// return bridge.JsException(v8ctx, "context object is missing goValueID") +// } - // Get the goValueID from internal field (index 0) - if ctxObj.InternalFieldCount() < 1 { - return bridge.JsException(v8ctx, "context object is missing internal fields") - } +// goValueID := goValueIDValue.String() - goValueIDValue := ctxObj.GetInternalField(0) - if goValueIDValue == nil || !goValueIDValue.IsString() { - return bridge.JsException(v8ctx, "context object is missing goValueID") - } +// // Retrieve the Go context object from bridge registry +// goObj := bridge.GetGoObject(goValueID) +// if goObj == nil { +// return bridge.JsException(v8ctx, "context object not found in registry") +// } - goValueID := goValueIDValue.String() +// // Type assert to *agentContext.Context +// ctx, ok := goObj.(*agentContext.Context) +// if !ok { +// return bridge.JsException(v8ctx, fmt.Sprintf("object is not a Context, got %T", goObj)) +// } - // Retrieve the Go context object from bridge registry - goObj := bridge.GetGoObject(goValueID) - if goObj == nil { - return bridge.JsException(v8ctx, "context object not found in registry") - } +// // Create output object +// outputObj, err := NewOutputObject(v8ctx, ctx) +// if err != nil { +// return bridge.JsException(v8ctx, err.Error()) +// } - // Type assert to *agentContext.Context - ctx, ok := goObj.(*agentContext.Context) - if !ok { - return bridge.JsException(v8ctx, fmt.Sprintf("object is not a Context, got %T", goObj)) - } +// return outputObj +// } - // Create output object - outputObj, err := NewOutputObject(v8ctx, ctx) - if err != nil { - return bridge.JsException(v8ctx, err.Error()) - } +// // NewOutputObject creates a JavaScript Output object +// func NewOutputObject(v8ctx *v8go.Context, ctx *agentContext.Context) (*v8go.Value, error) { +// jsObject := v8go.NewObjectTemplate(v8ctx.Isolate()) - return outputObj -} +// // Set internal field count to 1 to store the __go_id +// // Internal fields are not accessible from JavaScript, providing better security +// jsObject.SetInternalFieldCount(1) -// NewOutputObject creates a JavaScript Output object -func NewOutputObject(v8ctx *v8go.Context, ctx *agentContext.Context) (*v8go.Value, error) { - jsObject := v8go.NewObjectTemplate(v8ctx.Isolate()) +// // Register context in global bridge registry for efficient Go object retrieval +// // The goValueID will be stored in internal field (index 0) after instance creation +// goValueID := bridge.RegisterGoObject(ctx) - // Set internal field count to 1 to store the __go_id - // Internal fields are not accessible from JavaScript, providing better security - jsObject.SetInternalFieldCount(1) +// // Set methods +// jsObject.Set("Send", outputSendMethod(v8ctx.Isolate(), ctx)) +// jsObject.Set("SendGroup", outputSendGroupMethod(v8ctx.Isolate(), ctx)) - // Register context in global bridge registry for efficient Go object retrieval - // The goValueID will be stored in internal field (index 0) after instance creation - goValueID := bridge.RegisterGoObject(ctx) +// // Set release function that will be called when JavaScript object is released +// jsObject.Set("__release", outputGoRelease(v8ctx.Isolate())) - // Set methods - jsObject.Set("Send", outputSendMethod(v8ctx.Isolate(), ctx)) - jsObject.Set("SendGroup", outputSendGroupMethod(v8ctx.Isolate(), ctx)) +// // Create instance +// instance, err := jsObject.NewInstance(v8ctx) +// if err != nil { +// // Clean up: release from global registry if instance creation failed +// bridge.ReleaseGoObject(goValueID) +// return nil, err +// } - // Set release function that will be called when JavaScript object is released - jsObject.Set("__release", outputGoRelease(v8ctx.Isolate())) +// // Store the goValueID in internal field (index 0) +// // This is not accessible from JavaScript, providing better security +// obj, err := instance.Value.AsObject() +// if err != nil { +// bridge.ReleaseGoObject(goValueID) +// return nil, err +// } - // Create instance - instance, err := jsObject.NewInstance(v8ctx) - if err != nil { - // Clean up: release from global registry if instance creation failed - bridge.ReleaseGoObject(goValueID) - return nil, err - } +// err = obj.SetInternalField(0, goValueID) +// if err != nil { +// bridge.ReleaseGoObject(goValueID) +// return nil, err +// } - // Store the goValueID in internal field (index 0) - // This is not accessible from JavaScript, providing better security - obj, err := instance.Value.AsObject() - if err != nil { - bridge.ReleaseGoObject(goValueID) - return nil, err - } +// return instance.Value, nil +// } - err = obj.SetInternalField(0, goValueID) - if err != nil { - bridge.ReleaseGoObject(goValueID) - return nil, err - } +// // outputGoRelease releases the Go object from the global bridge registry +// // It retrieves the goValueID from internal field (index 0) and releases the Go object +// func outputGoRelease(iso *v8go.Isolate) *v8go.FunctionTemplate { +// return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { +// // Get the output object (this) +// thisObj, err := info.This().AsObject() +// if err == nil && thisObj.InternalFieldCount() > 0 { +// // Get goValueID from internal field (index 0) +// goValueIDValue := thisObj.GetInternalField(0) +// if goValueIDValue != nil && goValueIDValue.IsString() { +// goValueID := goValueIDValue.String() +// // Release from global bridge registry +// bridge.ReleaseGoObject(goValueID) +// } +// } - return instance.Value, nil -} +// return v8go.Undefined(info.Context().Isolate()) +// }) +// } -// outputGoRelease releases the Go object from the global bridge registry -// It retrieves the goValueID from internal field (index 0) and releases the Go object -func outputGoRelease(iso *v8go.Isolate) *v8go.FunctionTemplate { - return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { - // Get the output object (this) - thisObj, err := info.This().AsObject() - if err == nil && thisObj.InternalFieldCount() > 0 { - // Get goValueID from internal field (index 0) - goValueIDValue := thisObj.GetInternalField(0) - if goValueIDValue != nil && goValueIDValue.IsString() { - goValueID := goValueIDValue.String() - // Release from global bridge registry - bridge.ReleaseGoObject(goValueID) - } - } +// // outputSendMethod implements the Send method +// // Usage: output.Send(message) +// // message can be an object with { type: string, props: object, ... } or a simple string (will be converted to text message) +// func outputSendMethod(iso *v8go.Isolate, ctx *agentContext.Context) *v8go.FunctionTemplate { +// return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { +// v8ctx := info.Context() +// args := info.Args() - return v8go.Undefined(info.Context().Isolate()) - }) -} +// if len(args) < 1 { +// return bridge.JsException(v8ctx, "Send requires a message argument") +// } -// outputSendMethod implements the Send method -// Usage: output.Send(message) -// message can be an object with { type: string, props: object, ... } or a simple string (will be converted to text message) -func outputSendMethod(iso *v8go.Isolate, ctx *agentContext.Context) *v8go.FunctionTemplate { - return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { - v8ctx := info.Context() - args := info.Args() +// // Parse message argument +// msg, err := parseMessage(v8ctx, args[0]) +// if err != nil { +// return bridge.JsException(v8ctx, fmt.Sprintf("invalid message: %s", err)) +// } - if len(args) < 1 { - return bridge.JsException(v8ctx, "Send requires a message argument") - } +// // Call output.Send +// if err := output.Send(ctx, msg); err != nil { +// return bridge.JsException(v8ctx, fmt.Sprintf("Send failed: %s", err)) +// } - // Parse message argument - msg, err := parseMessage(v8ctx, args[0]) - if err != nil { - return bridge.JsException(v8ctx, fmt.Sprintf("invalid message: %s", err)) - } +// return info.This().Value +// }) +// } - // Call output.Send - if err := output.Send(ctx, msg); err != nil { - return bridge.JsException(v8ctx, fmt.Sprintf("Send failed: %s", err)) - } +// // outputSendGroupMethod implements the SendGroup method +// // Usage: output.SendGroup(group) +// // group must be an object with { id: string, messages: [], ... } +// func outputSendGroupMethod(iso *v8go.Isolate, ctx *agentContext.Context) *v8go.FunctionTemplate { +// return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { +// v8ctx := info.Context() +// args := info.Args() - return info.This().Value - }) -} +// if len(args) < 1 { +// return bridge.JsException(v8ctx, "SendGroup requires a group argument") +// } -// outputSendGroupMethod implements the SendGroup method -// Usage: output.SendGroup(group) -// group must be an object with { id: string, messages: [], ... } -func outputSendGroupMethod(iso *v8go.Isolate, ctx *agentContext.Context) *v8go.FunctionTemplate { - return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { - v8ctx := info.Context() - args := info.Args() +// // Parse group argument +// group, err := parseGroup(v8ctx, args[0]) +// if err != nil { +// return bridge.JsException(v8ctx, fmt.Sprintf("invalid group: %s", err)) +// } - if len(args) < 1 { - return bridge.JsException(v8ctx, "SendGroup requires a group argument") - } +// // Call output.SendGroup +// if err := output.SendGroup(ctx, group); err != nil { +// return bridge.JsException(v8ctx, fmt.Sprintf("SendGroup failed: %s", err)) +// } - // Parse group argument - group, err := parseGroup(v8ctx, args[0]) - if err != nil { - return bridge.JsException(v8ctx, fmt.Sprintf("invalid group: %s", err)) - } +// return info.This().Value +// }) +// } - // Call output.SendGroup - if err := output.SendGroup(ctx, group); err != nil { - return bridge.JsException(v8ctx, fmt.Sprintf("SendGroup failed: %s", err)) - } +// // parseMessage parses a JavaScript value into a message.Message +// func parseMessage(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Message, error) { +// // Handle string shorthand: convert to text message +// if jsValue.IsString() { +// return &message.Message{ +// Type: message.TypeText, +// Props: map[string]interface{}{ +// "content": jsValue.String(), +// }, +// }, nil +// } - return info.This().Value - }) -} +// // Handle object +// if !jsValue.IsObject() { +// return nil, fmt.Errorf("message must be a string or object") +// } -// parseMessage parses a JavaScript value into a message.Message -func parseMessage(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Message, error) { - // Handle string shorthand: convert to text message - if jsValue.IsString() { - return &message.Message{ - Type: message.TypeText, - Props: map[string]interface{}{ - "content": jsValue.String(), - }, - }, nil - } +// // Convert to Go map +// goValue, err := bridge.GoValue(jsValue, v8ctx) +// if err != nil { +// return nil, fmt.Errorf("failed to convert message: %w", err) +// } - // Handle object - if !jsValue.IsObject() { - return nil, fmt.Errorf("message must be a string or object") - } +// msgMap, ok := goValue.(map[string]interface{}) +// if !ok { +// return nil, fmt.Errorf("message must be an object") +// } - // Convert to Go map - goValue, err := bridge.GoValue(jsValue, v8ctx) - if err != nil { - return nil, fmt.Errorf("failed to convert message: %w", err) - } +// // Build message +// msg := &message.Message{} - msgMap, ok := goValue.(map[string]interface{}) - if !ok { - return nil, fmt.Errorf("message must be an object") - } +// // Type field (required) +// if msgType, ok := msgMap["type"].(string); ok { +// msg.Type = msgType +// } else { +// return nil, fmt.Errorf("message.type is required and must be a string") +// } - // Build message - msg := &message.Message{} +// // Props field (optional) +// if props, ok := msgMap["props"].(map[string]interface{}); ok { +// msg.Props = props +// } - // Type field (required) - if msgType, ok := msgMap["type"].(string); ok { - msg.Type = msgType - } else { - return nil, fmt.Errorf("message.type is required and must be a string") - } +// // Optional fields +// if id, ok := msgMap["id"].(string); ok { +// msg.ID = id +// } +// if delta, ok := msgMap["delta"].(bool); ok { +// msg.Delta = delta +// } +// if done, ok := msgMap["done"].(bool); ok { +// msg.Done = done +// } +// if deltaPath, ok := msgMap["delta_path"].(string); ok { +// msg.DeltaPath = deltaPath +// } +// if deltaAction, ok := msgMap["delta_action"].(string); ok { +// msg.DeltaAction = deltaAction +// } +// if typeChange, ok := msgMap["type_change"].(bool); ok { +// msg.TypeChange = typeChange +// } +// if groupID, ok := msgMap["group_id"].(string); ok { +// msg.GroupID = groupID +// } +// if groupStart, ok := msgMap["group_start"].(bool); ok { +// msg.GroupStart = groupStart +// } +// if groupEnd, ok := msgMap["group_end"].(bool); ok { +// msg.GroupEnd = groupEnd +// } - // Props field (optional) - if props, ok := msgMap["props"].(map[string]interface{}); ok { - msg.Props = props - } +// // Metadata (optional) +// if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok { +// metadata := &message.Metadata{} +// if timestamp, ok := metadataMap["timestamp"].(float64); ok { +// metadata.Timestamp = int64(timestamp) +// } +// if sequence, ok := metadataMap["sequence"].(float64); ok { +// metadata.Sequence = int(sequence) +// } +// if traceID, ok := metadataMap["trace_id"].(string); ok { +// metadata.TraceID = traceID +// } +// msg.Metadata = metadata +// } - // Optional fields - if id, ok := msgMap["id"].(string); ok { - msg.ID = id - } - if delta, ok := msgMap["delta"].(bool); ok { - msg.Delta = delta - } - if done, ok := msgMap["done"].(bool); ok { - msg.Done = done - } - if deltaPath, ok := msgMap["delta_path"].(string); ok { - msg.DeltaPath = deltaPath - } - if deltaAction, ok := msgMap["delta_action"].(string); ok { - msg.DeltaAction = deltaAction - } - if typeChange, ok := msgMap["type_change"].(bool); ok { - msg.TypeChange = typeChange - } - if groupID, ok := msgMap["group_id"].(string); ok { - msg.GroupID = groupID - } - if groupStart, ok := msgMap["group_start"].(bool); ok { - msg.GroupStart = groupStart - } - if groupEnd, ok := msgMap["group_end"].(bool); ok { - msg.GroupEnd = groupEnd - } +// return msg, nil +// } - // Metadata (optional) - if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok { - metadata := &message.Metadata{} - if timestamp, ok := metadataMap["timestamp"].(float64); ok { - metadata.Timestamp = int64(timestamp) - } - if sequence, ok := metadataMap["sequence"].(float64); ok { - metadata.Sequence = int(sequence) - } - if traceID, ok := metadataMap["trace_id"].(string); ok { - metadata.TraceID = traceID - } - msg.Metadata = metadata - } +// // parseGroup parses a JavaScript value into a message.Group +// func parseGroup(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Group, error) { +// // Must be an object +// if !jsValue.IsObject() { +// return nil, fmt.Errorf("group must be an object") +// } - return msg, nil -} +// // Convert to Go map +// goValue, err := bridge.GoValue(jsValue, v8ctx) +// if err != nil { +// return nil, fmt.Errorf("failed to convert group: %w", err) +// } -// parseGroup parses a JavaScript value into a message.Group -func parseGroup(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Group, error) { - // Must be an object - if !jsValue.IsObject() { - return nil, fmt.Errorf("group must be an object") - } +// groupMap, ok := goValue.(map[string]interface{}) +// if !ok { +// return nil, fmt.Errorf("group must be an object") +// } - // Convert to Go map - goValue, err := bridge.GoValue(jsValue, v8ctx) - if err != nil { - return nil, fmt.Errorf("failed to convert group: %w", err) - } +// // Build group +// group := &message.Group{} - groupMap, ok := goValue.(map[string]interface{}) - if !ok { - return nil, fmt.Errorf("group must be an object") - } +// // ID field (required) +// if id, ok := groupMap["id"].(string); ok { +// group.ID = id +// } else { +// return nil, fmt.Errorf("group.id is required and must be a string") +// } - // Build group - group := &message.Group{} +// // Messages field (required) +// if messagesArray, ok := groupMap["messages"].([]interface{}); ok { +// group.Messages = make([]*message.Message, 0, len(messagesArray)) +// for i, msgInterface := range messagesArray { +// // Convert to map +// msgMap, ok := msgInterface.(map[string]interface{}) +// if !ok { +// return nil, fmt.Errorf("group.messages[%d] must be an object", i) +// } - // ID field (required) - if id, ok := groupMap["id"].(string); ok { - group.ID = id - } else { - return nil, fmt.Errorf("group.id is required and must be a string") - } +// // Convert map to Message +// msg := &message.Message{} - // Messages field (required) - if messagesArray, ok := groupMap["messages"].([]interface{}); ok { - group.Messages = make([]*message.Message, 0, len(messagesArray)) - for i, msgInterface := range messagesArray { - // Convert to map - msgMap, ok := msgInterface.(map[string]interface{}) - if !ok { - return nil, fmt.Errorf("group.messages[%d] must be an object", i) - } +// // Type field (required) +// if msgType, ok := msgMap["type"].(string); ok { +// msg.Type = msgType +// } else { +// return nil, fmt.Errorf("group.messages[%d].type is required", i) +// } - // Convert map to Message - msg := &message.Message{} +// // Props field (optional) +// if props, ok := msgMap["props"].(map[string]interface{}); ok { +// msg.Props = props +// } - // Type field (required) - if msgType, ok := msgMap["type"].(string); ok { - msg.Type = msgType - } else { - return nil, fmt.Errorf("group.messages[%d].type is required", i) - } +// // Optional fields +// if id, ok := msgMap["id"].(string); ok { +// msg.ID = id +// } +// if delta, ok := msgMap["delta"].(bool); ok { +// msg.Delta = delta +// } +// if done, ok := msgMap["done"].(bool); ok { +// msg.Done = done +// } +// if deltaPath, ok := msgMap["delta_path"].(string); ok { +// msg.DeltaPath = deltaPath +// } +// if deltaAction, ok := msgMap["delta_action"].(string); ok { +// msg.DeltaAction = deltaAction +// } +// if typeChange, ok := msgMap["type_change"].(bool); ok { +// msg.TypeChange = typeChange +// } +// if groupID, ok := msgMap["group_id"].(string); ok { +// msg.GroupID = groupID +// } +// if groupStart, ok := msgMap["group_start"].(bool); ok { +// msg.GroupStart = groupStart +// } +// if groupEnd, ok := msgMap["group_end"].(bool); ok { +// msg.GroupEnd = groupEnd +// } - // Props field (optional) - if props, ok := msgMap["props"].(map[string]interface{}); ok { - msg.Props = props - } +// // Metadata (optional) +// if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok { +// metadata := &message.Metadata{} +// if timestamp, ok := metadataMap["timestamp"].(float64); ok { +// metadata.Timestamp = int64(timestamp) +// } +// if sequence, ok := metadataMap["sequence"].(float64); ok { +// metadata.Sequence = int(sequence) +// } +// if traceID, ok := metadataMap["trace_id"].(string); ok { +// metadata.TraceID = traceID +// } +// msg.Metadata = metadata +// } - // Optional fields - if id, ok := msgMap["id"].(string); ok { - msg.ID = id - } - if delta, ok := msgMap["delta"].(bool); ok { - msg.Delta = delta - } - if done, ok := msgMap["done"].(bool); ok { - msg.Done = done - } - if deltaPath, ok := msgMap["delta_path"].(string); ok { - msg.DeltaPath = deltaPath - } - if deltaAction, ok := msgMap["delta_action"].(string); ok { - msg.DeltaAction = deltaAction - } - if typeChange, ok := msgMap["type_change"].(bool); ok { - msg.TypeChange = typeChange - } - if groupID, ok := msgMap["group_id"].(string); ok { - msg.GroupID = groupID - } - if groupStart, ok := msgMap["group_start"].(bool); ok { - msg.GroupStart = groupStart - } - if groupEnd, ok := msgMap["group_end"].(bool); ok { - msg.GroupEnd = groupEnd - } +// group.Messages = append(group.Messages, msg) +// } +// } else { +// return nil, fmt.Errorf("group.messages is required and must be an array") +// } - // Metadata (optional) - if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok { - metadata := &message.Metadata{} - if timestamp, ok := metadataMap["timestamp"].(float64); ok { - metadata.Timestamp = int64(timestamp) - } - if sequence, ok := metadataMap["sequence"].(float64); ok { - metadata.Sequence = int(sequence) - } - if traceID, ok := metadataMap["trace_id"].(string); ok { - metadata.TraceID = traceID - } - msg.Metadata = metadata - } +// // Metadata (optional) +// if metadataMap, ok := groupMap["metadata"].(map[string]interface{}); ok { +// metadata := &message.Metadata{} +// if timestamp, ok := metadataMap["timestamp"].(float64); ok { +// metadata.Timestamp = int64(timestamp) +// } +// if sequence, ok := metadataMap["sequence"].(float64); ok { +// metadata.Sequence = int(sequence) +// } +// if traceID, ok := metadataMap["trace_id"].(string); ok { +// metadata.TraceID = traceID +// } +// group.Metadata = metadata +// } - group.Messages = append(group.Messages, msg) - } - } else { - return nil, fmt.Errorf("group.messages is required and must be an array") - } - - // Metadata (optional) - if metadataMap, ok := groupMap["metadata"].(map[string]interface{}); ok { - metadata := &message.Metadata{} - if timestamp, ok := metadataMap["timestamp"].(float64); ok { - metadata.Timestamp = int64(timestamp) - } - if sequence, ok := metadataMap["sequence"].(float64); ok { - metadata.Sequence = int(sequence) - } - if traceID, ok := metadataMap["trace_id"].(string); ok { - metadata.TraceID = traceID - } - group.Metadata = metadata - } - - return group, nil -} +// return group, nil +// } diff --git a/agent/output/jsapi/output_test.go b/agent/output/jsapi/output_test.go index 535ec035..24b51000 100644 --- a/agent/output/jsapi/output_test.go +++ b/agent/output/jsapi/output_test.go @@ -1,361 +1,349 @@ package jsapi -import ( - "context" - "net/http" - "testing" +// func TestOutputConstructor(t *testing.T) { +// test.Prepare(t, config.Conf) +// defer test.Clean() - "github.com/stretchr/testify/assert" - v8 "github.com/yaoapp/gou/runtime/v8" - agentContext "github.com/yaoapp/yao/agent/context" - "github.com/yaoapp/yao/config" - "github.com/yaoapp/yao/test" -) +// tests := []struct { +// name string +// script string +// expectError bool +// }{ +// { +// name: "Create Output with context", +// script: ` +// function test(ctx) { +// const output = new Output(ctx); +// return output !== undefined && output !== null; +// } +// `, +// expectError: false, +// }, +// { +// name: "Create Output without context should fail", +// script: ` +// function test(ctx) { +// try { +// const output = new Output(); +// return false; +// } catch (e) { +// return e.toString().includes("context argument"); +// } +// } +// `, +// expectError: false, +// }, +// } -func TestOutputConstructor(t *testing.T) { - test.Prepare(t, config.Conf) - defer test.Clean() +// for _, tt := range tests { +// t.Run(tt.name, func(t *testing.T) { +// ctx := agentContext.New(context.Background(), nil, "test-chat-123", "") +// ctx.AssistantID = "test-assistant-456" - tests := []struct { - name string - script string - expectError bool - }{ - { - name: "Create Output with context", - script: ` - function test(ctx) { - const output = new Output(ctx); - return output !== undefined && output !== null; - } - `, - expectError: false, - }, - { - name: "Create Output without context should fail", - script: ` - function test(ctx) { - try { - const output = new Output(); - return false; - } catch (e) { - return e.toString().includes("context argument"); - } - } - `, - expectError: false, - }, - } +// // Execute test script with v8.Call +// res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx) +// if tt.expectError { +// assert.Error(t, err) +// return +// } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - ctx := agentContext.New(context.Background(), nil, "test-chat-123", "") - ctx.AssistantID = "test-assistant-456" +// assert.NoError(t, err) +// assert.True(t, res.(bool)) +// }) +// } +// } - // Execute test script with v8.Call - res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx) - if tt.expectError { - assert.Error(t, err) - return - } +// func TestOutputSend(t *testing.T) { +// test.Prepare(t, config.Conf) +// defer test.Clean() - assert.NoError(t, err) - assert.True(t, res.(bool)) - }) - } -} +// tests := []struct { +// name string +// script string +// expectError bool +// validate func(*testing.T, *agentContext.Context) +// }{ +// { +// name: "Send text message with object", +// script: ` +// function test(ctx) { +// const output = new Output(ctx); +// output.Send({ +// type: "text", +// props: { content: "Hello World" } +// }); +// return true; +// } +// `, +// expectError: false, +// }, +// { +// name: "Send text message with string shorthand", +// script: ` +// function test(ctx) { +// const output = new Output(ctx); +// output.Send("Hello World"); +// return true; +// } +// `, +// expectError: false, +// }, +// { +// name: "Send message with all fields", +// script: ` +// function test(ctx) { +// const output = new Output(ctx); +// output.Send({ +// type: "text", +// props: { content: "Test" }, +// id: "msg-1", +// delta: true, +// done: false, +// delta_path: "content", +// delta_action: "append", +// metadata: { +// timestamp: 1234567890, +// sequence: 1, +// trace_id: "trace-123" +// } +// }); +// return true; +// } +// `, +// expectError: false, +// }, +// { +// name: "Send error message", +// script: ` +// function test(ctx) { +// const output = new Output(ctx); +// output.Send({ +// type: "error", +// props: { +// message: "Something went wrong", +// code: "ERR_001" +// } +// }); +// return true; +// } +// `, +// expectError: false, +// }, +// { +// name: "Send without message should fail", +// script: ` +// function test(ctx) { +// try { +// const output = new Output(ctx); +// output.Send(); +// return false; +// } catch (e) { +// return e.toString().includes("message argument"); +// } +// } +// `, +// expectError: false, +// }, +// { +// name: "Send message without type should fail", +// script: ` +// function test(ctx) { +// try { +// const output = new Output(ctx); +// output.Send({ props: { content: "test" } }); +// return false; +// } catch (e) { +// return e.toString().includes("type is required"); +// } +// } +// `, +// expectError: false, +// }, +// } -func TestOutputSend(t *testing.T) { - test.Prepare(t, config.Conf) - defer test.Clean() +// for _, tt := range tests { +// t.Run(tt.name, func(t *testing.T) { +// // Create context with mock writer +// ctx := agentContext.New(context.Background(), nil, "test-chat", "") +// ctx.Writer = &mockWriter{} - tests := []struct { - name string - script string - expectError bool - validate func(*testing.T, *agentContext.Context) - }{ - { - name: "Send text message with object", - script: ` - function test(ctx) { - const output = new Output(ctx); - output.Send({ - type: "text", - props: { content: "Hello World" } - }); - return true; - } - `, - expectError: false, - }, - { - name: "Send text message with string shorthand", - script: ` - function test(ctx) { - const output = new Output(ctx); - output.Send("Hello World"); - return true; - } - `, - expectError: false, - }, - { - name: "Send message with all fields", - script: ` - function test(ctx) { - const output = new Output(ctx); - output.Send({ - type: "text", - props: { content: "Test" }, - id: "msg-1", - delta: true, - done: false, - delta_path: "content", - delta_action: "append", - metadata: { - timestamp: 1234567890, - sequence: 1, - trace_id: "trace-123" - } - }); - return true; - } - `, - expectError: false, - }, - { - name: "Send error message", - script: ` - function test(ctx) { - const output = new Output(ctx); - output.Send({ - type: "error", - props: { - message: "Something went wrong", - code: "ERR_001" - } - }); - return true; - } - `, - expectError: false, - }, - { - name: "Send without message should fail", - script: ` - function test(ctx) { - try { - const output = new Output(ctx); - output.Send(); - return false; - } catch (e) { - return e.toString().includes("message argument"); - } - } - `, - expectError: false, - }, - { - name: "Send message without type should fail", - script: ` - function test(ctx) { - try { - const output = new Output(ctx); - output.Send({ props: { content: "test" } }); - return false; - } catch (e) { - return e.toString().includes("type is required"); - } - } - `, - expectError: false, - }, - } +// // Execute test script with v8.Call +// res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx) +// if tt.expectError { +// assert.Error(t, err) +// return +// } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - // Create context with mock writer - ctx := agentContext.New(context.Background(), nil, "test-chat", "") - ctx.Writer = &mockWriter{} +// assert.NoError(t, err) +// assert.True(t, res.(bool)) - // Execute test script with v8.Call - res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx) - if tt.expectError { - assert.Error(t, err) - return - } +// if tt.validate != nil { +// tt.validate(t, &ctx) +// } +// }) +// } +// } - assert.NoError(t, err) - assert.True(t, res.(bool)) +// func TestOutputSendGroup(t *testing.T) { +// test.Prepare(t, config.Conf) +// defer test.Clean() - if tt.validate != nil { - tt.validate(t, &ctx) - } - }) - } -} +// tests := []struct { +// name string +// script string +// expectError bool +// }{ +// { +// name: "Send message group", +// script: ` +// function test(ctx) { +// const output = new Output(ctx); +// output.SendGroup({ +// id: "group-1", +// messages: [ +// { type: "text", props: { content: "Message 1" } }, +// { type: "text", props: { content: "Message 2" } } +// ] +// }); +// return true; +// } +// `, +// expectError: false, +// }, +// { +// name: "Send group with metadata", +// script: ` +// function test(ctx) { +// const output = new Output(ctx); +// output.SendGroup({ +// id: "group-1", +// messages: [ +// { type: "text", props: { content: "Test" } } +// ], +// metadata: { +// timestamp: 1234567890, +// sequence: 1 +// } +// }); +// return true; +// } +// `, +// expectError: false, +// }, +// { +// name: "Send empty group", +// script: ` +// function test(ctx) { +// const output = new Output(ctx); +// output.SendGroup({ +// id: "group-1", +// messages: [] +// }); +// return true; +// } +// `, +// expectError: false, +// }, +// { +// name: "Send group without id should fail", +// script: ` +// function test(ctx) { +// try { +// const output = new Output(ctx); +// output.SendGroup({ +// messages: [ +// { type: "text", props: { content: "Test" } } +// ] +// }); +// return false; +// } catch (e) { +// return e.toString().includes("id is required"); +// } +// } +// `, +// expectError: false, +// }, +// { +// name: "Send group without messages should fail", +// script: ` +// function test(ctx) { +// try { +// const output = new Output(ctx); +// output.SendGroup({ id: "group-1" }); +// return false; +// } catch (e) { +// return e.toString().includes("messages is required"); +// } +// } +// `, +// expectError: false, +// }, +// } -func TestOutputSendGroup(t *testing.T) { - test.Prepare(t, config.Conf) - defer test.Clean() +// for _, tt := range tests { +// t.Run(tt.name, func(t *testing.T) { +// // Create context with mock writer +// ctx := agentContext.New(context.Background(), nil, "test-chat", "") +// ctx.Writer = &mockWriter{} - tests := []struct { - name string - script string - expectError bool - }{ - { - name: "Send message group", - script: ` - function test(ctx) { - const output = new Output(ctx); - output.SendGroup({ - id: "group-1", - messages: [ - { type: "text", props: { content: "Message 1" } }, - { type: "text", props: { content: "Message 2" } } - ] - }); - return true; - } - `, - expectError: false, - }, - { - name: "Send group with metadata", - script: ` - function test(ctx) { - const output = new Output(ctx); - output.SendGroup({ - id: "group-1", - messages: [ - { type: "text", props: { content: "Test" } } - ], - metadata: { - timestamp: 1234567890, - sequence: 1 - } - }); - return true; - } - `, - expectError: false, - }, - { - name: "Send empty group", - script: ` - function test(ctx) { - const output = new Output(ctx); - output.SendGroup({ - id: "group-1", - messages: [] - }); - return true; - } - `, - expectError: false, - }, - { - name: "Send group without id should fail", - script: ` - function test(ctx) { - try { - const output = new Output(ctx); - output.SendGroup({ - messages: [ - { type: "text", props: { content: "Test" } } - ] - }); - return false; - } catch (e) { - return e.toString().includes("id is required"); - } - } - `, - expectError: false, - }, - { - name: "Send group without messages should fail", - script: ` - function test(ctx) { - try { - const output = new Output(ctx); - output.SendGroup({ id: "group-1" }); - return false; - } catch (e) { - return e.toString().includes("messages is required"); - } - } - `, - expectError: false, - }, - } +// // Execute test script with v8.Call +// res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx) +// if tt.expectError { +// assert.Error(t, err) +// return +// } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - // Create context with mock writer - ctx := agentContext.New(context.Background(), nil, "test-chat", "") - ctx.Writer = &mockWriter{} +// assert.NoError(t, err) +// assert.True(t, res.(bool)) +// }) +// } +// } - // Execute test script with v8.Call - res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx) - if tt.expectError { - assert.Error(t, err) - return - } +// func TestOutputChaining(t *testing.T) { +// test.Prepare(t, config.Conf) +// defer test.Clean() - assert.NoError(t, err) - assert.True(t, res.(bool)) - }) - } -} +// script := ` +// function test(ctx) { +// const output = new Output(ctx); -func TestOutputChaining(t *testing.T) { - test.Prepare(t, config.Conf) - defer test.Clean() +// // Send should return the output object for chaining +// const result = output.Send("Message 1"); - script := ` - function test(ctx) { - const output = new Output(ctx); - - // Send should return the output object for chaining - const result = output.Send("Message 1"); - - // Should be able to chain sends - output.Send("Message 2").Send("Message 3"); - - return result !== undefined; - } - ` +// // Should be able to chain sends +// output.Send("Message 2").Send("Message 3"); - ctx := agentContext.New(context.Background(), nil, "test-chat", "") - ctx.Writer = &mockWriter{} +// return result !== undefined; +// } +// ` - // Execute test script with v8.Call - res, err := v8.Call(v8.CallOptions{}, script, &ctx) - assert.NoError(t, err) - assert.True(t, res.(bool)) -} +// ctx := agentContext.New(context.Background(), nil, "test-chat", "") +// ctx.Writer = &mockWriter{} -// mockWriter is a mock implementation of http.ResponseWriter for testing -type mockWriter struct { - data [][]byte - header http.Header -} +// // Execute test script with v8.Call +// res, err := v8.Call(v8.CallOptions{}, script, &ctx) +// assert.NoError(t, err) +// assert.True(t, res.(bool)) +// } -func (w *mockWriter) Header() http.Header { - if w.header == nil { - w.header = make(http.Header) - } - return w.header -} +// // mockWriter is a mock implementation of http.ResponseWriter for testing +// type mockWriter struct { +// data [][]byte +// header http.Header +// } -func (w *mockWriter) Write(p []byte) (n int, err error) { - w.data = append(w.data, p) - return len(p), nil -} +// func (w *mockWriter) Header() http.Header { +// if w.header == nil { +// w.header = make(http.Header) +// } +// return w.header +// } -func (w *mockWriter) WriteHeader(statusCode int) {} +// func (w *mockWriter) Write(p []byte) (n int, err error) { +// w.data = append(w.data, p) +// return len(p), nil +// } -func (w *mockWriter) Flush() {} +// func (w *mockWriter) WriteHeader(statusCode int) {} + +// func (w *mockWriter) Flush() {} diff --git a/agent/output/message/interfaces.go b/agent/output/message/interfaces.go index 0d06bff2..6112a407 100644 --- a/agent/output/message/interfaces.go +++ b/agent/output/message/interfaces.go @@ -1,7 +1,5 @@ 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 { @@ -29,23 +27,11 @@ type Adapter interface { 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 + Handle(chunkType StreamChunkType, data []byte) error // Flush flushes any pending messages Flush() error diff --git a/agent/output/message/types.go b/agent/output/message/types.go index 514279b3..227906ad 100644 --- a/agent/output/message/types.go +++ b/agent/output/message/types.go @@ -1,5 +1,34 @@ package message +import ( + "net/http" + + traceTypes "github.com/yaoapp/yao/trace/types" +) + +// Options are the options for the writer +type Options struct { + BaseURL string + Accept string + Writer http.ResponseWriter + Trace traceTypes.Manager + Capabilities *ModelCapabilities + Locale string +} + +// ModelCapabilities defines the capabilities of a language model +// Used by LLM to select appropriate provider and validate requests +type ModelCapabilities struct { + Vision interface{} `json:"vision,omitempty"` // Supports vision/image input: bool or VisionFormat string ("openai", "claude"/"base64", "default") + ToolCalls *bool `json:"tool_calls,omitempty"` // Supports tool/function calling + Audio *bool `json:"audio,omitempty"` // Supports audio input/output + Reasoning *bool `json:"reasoning,omitempty"` // Supports reasoning/thinking mode (o1, DeepSeek R1) + Streaming *bool `json:"streaming,omitempty"` // Supports streaming responses + JSON *bool `json:"json,omitempty"` // Supports JSON mode + Multimodal *bool `json:"multimodal,omitempty"` // Supports multimodal input (text + images + audio) + TemperatureAdjustable *bool `json:"temperature_adjustable,omitempty"` // Supports temperature adjustment (reasoning models typically don't) +} + // Message represents a universal message structure (DSL) // All messages are expressed through Type + Props, without predefining specific types type Message struct { @@ -185,3 +214,133 @@ const ( DeltaMerge = "merge" // Merge (for objects) DeltaSet = "set" // Set (for new fields) ) + +// StreamChunkType represents the type of content in a streaming chunk +type StreamChunkType string + +// Stream chunk type constants - indicates what type of content is in the current chunk +const ( + // Content chunk types - actual data from the LLM + ChunkText StreamChunkType = "text" // Regular text content + ChunkThinking StreamChunkType = "thinking" // Reasoning/thinking content (o1, DeepSeek R1) + ChunkToolCall StreamChunkType = "tool_call" // Tool/function call + ChunkRefusal StreamChunkType = "refusal" // Model refusal + ChunkMetadata StreamChunkType = "metadata" // Metadata (usage, finish_reason, etc.) + ChunkError StreamChunkType = "error" // Error chunk + ChunkUnknown StreamChunkType = "unknown" // Unknown/unrecognized chunk type + + // Lifecycle event types - stream and group boundaries + ChunkStreamStart StreamChunkType = "stream_start" // Stream begins (entire request starts) + ChunkStreamEnd StreamChunkType = "stream_end" // Stream ends (entire request completes) + ChunkGroupStart StreamChunkType = "group_start" // Message group begins (text/tool_call/thinking group starts) + ChunkGroupEnd StreamChunkType = "group_end" // Message group ends (text/tool_call/thinking group completes) +) + +// StreamFunc the streaming function callback +// Parameters: +// - chunkType: the type of content in this chunk (text, thinking, tool_call, etc.) +// - data: the actual chunk data (could be text, JSON, or other format) +// +// Returns: +// - int: status code (0 = continue, non-zero = stop streaming) +type StreamFunc func(chunkType StreamChunkType, data []byte) int + +// AssistantInfo represents the assistant information structure +type AssistantInfo struct { + ID string `json:"assistant_id"` // Assistant ID + Type string `json:"type,omitempty"` // Assistant Type, default is assistant + Name string `json:"name,omitempty"` // Assistant Name + Avatar string `json:"avatar,omitempty"` // Assistant Avatar + Description string `json:"description,omitempty"` // Assistant Description +} + +// UsageInfo represents token usage statistics +// Structure matches OpenAI API: https://platform.openai.com/docs/api-reference/chat/object#chat-object-usage +type UsageInfo struct { + PromptTokens int `json:"prompt_tokens"` // Number of tokens in the prompt + CompletionTokens int `json:"completion_tokens"` // Number of tokens in the generated completion + TotalTokens int `json:"total_tokens"` // Total number of tokens used (prompt + completion) + + // Detailed token breakdown + PromptTokensDetails *PromptTokensDetails `json:"prompt_tokens_details,omitempty"` // Breakdown of tokens used in the prompt + CompletionTokensDetails *CompletionTokensDetails `json:"completion_tokens_details,omitempty"` // Breakdown of tokens used in the completion +} + +// PromptTokensDetails provides detailed breakdown of tokens used in the prompt +type PromptTokensDetails struct { + AudioTokens int `json:"audio_tokens,omitempty"` // Audio input tokens present in the prompt + CachedTokens int `json:"cached_tokens,omitempty"` // Cached tokens present in the prompt +} + +// CompletionTokensDetails provides detailed breakdown of tokens used in the completion +type CompletionTokensDetails struct { + AcceptedPredictionTokens int `json:"accepted_prediction_tokens,omitempty"` // Tokens from predictions that appeared in the completion + AudioTokens int `json:"audio_tokens,omitempty"` // Audio input tokens generated by the model + ReasoningTokens int `json:"reasoning_tokens,omitempty"` // Tokens generated by the model for reasoning (o1, o1-mini, DeepSeek R1) + RejectedPredictionTokens int `json:"rejected_prediction_tokens,omitempty"` // Tokens from predictions that did not appear in the completion +} + +// ============================================================================ +// Stream Lifecycle Event Data Structures +// ============================================================================ +// These structures define the data format for stream lifecycle events. +// They provide a standardized way to communicate stream boundaries and metadata +// to the frontend, enabling better UI/UX (progress indicators, timing, etc.). + +// StreamStartData represents the data for stream_start event +// Sent when a streaming request begins +type StreamStartData struct { + ContextID string `json:"context_id"` // Context ID for the response + RequestID string `json:"request_id"` // Unique identifier for this request + Timestamp int64 `json:"timestamp"` // Unix timestamp when stream started + ChatID string `json:"chat_id"` // Chat ID being used (e.g., "chat-123") + TraceID string `json:"trace_id"` // Trace ID being used (e.g., "trace-123") + Assistant *AssistantInfo `json:"assistant,omitempty"` // Assistant information + Metadata map[string]interface{} `json:"metadata,omitempty"` // Metadata to pass to the page for CUI context +} + +// StreamEndData represents the data for stream_end event +// Sent when a streaming request completes (successfully or with error) +type StreamEndData struct { + RequestID string `json:"request_id"` // Corresponding request ID + ContextID string `json:"context_id"` // Context ID for the response + TraceID string `json:"trace_id"` // Trace ID being used (e.g., "trace-123") + Timestamp int64 `json:"timestamp"` // Unix timestamp when stream ended + DurationMs int64 `json:"duration_ms"` // Total duration in milliseconds + Status string `json:"status"` // "completed" | "error" | "cancelled" + Error string `json:"error,omitempty"` // Error message if status is "error" + Usage *UsageInfo `json:"usage,omitempty"` // Token usage statistics + Metadata map[string]interface{} `json:"metadata,omitempty"` // Metadata to pass to the page for CUI context +} + +// GroupStartData represents the data for group_start event +// Sent when a logical message group begins (text, tool_call, thinking, etc.) +type GroupStartData struct { + GroupID string `json:"group_id"` // Unique identifier for this group + Type string `json:"type"` // Group type: "text" | "thinking" | "tool_call" | "refusal" + Timestamp int64 `json:"timestamp"` // Unix timestamp when group started + ToolCall *GroupToolCallInfo `json:"tool_call,omitempty"` // Tool call metadata (if type is "tool_call") + Extra map[string]interface{} `json:"extra,omitempty"` // Additional metadata (for custom providers or future extensions) +} + +// GroupEndData represents the data for group_end event +// Sent when a logical message group completes +type GroupEndData struct { + GroupID string `json:"group_id"` // Corresponding group ID + Type string `json:"type"` // Group type (same as in group_start) + Timestamp int64 `json:"timestamp"` // Unix timestamp when group ended + DurationMs int64 `json:"duration_ms"` // Duration of this group in milliseconds + ChunkCount int `json:"chunk_count"` // Number of data chunks in this group + Status string `json:"status"` // "completed" | "partial" | "error" + ToolCall *GroupToolCallInfo `json:"tool_call,omitempty"` // Complete tool call info (if type is "tool_call") + Extra map[string]interface{} `json:"extra,omitempty"` // Additional metadata +} + +// GroupToolCallInfo contains tool call information for group events +// Used in both group_start (partial info) and group_end (complete info) +type GroupToolCallInfo struct { + ID string `json:"id"` // Tool call ID (e.g., "call_abc123") + Name string `json:"name"` // Function name (may be partial in group_start) + Arguments string `json:"arguments,omitempty"` // Complete arguments (only in group_end) + Index int `json:"index"` // Index in the tool calls array +} diff --git a/agent/output/output.go b/agent/output/output.go index 013de262..cd479db3 100644 --- a/agent/output/output.go +++ b/agent/output/output.go @@ -2,148 +2,212 @@ 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 +// Accept type constants +const ( + AcceptStandard = "standard" + AcceptWebCUI = "cui-web" + AccepNativeCUI = "cui-native" + AcceptDesktopCUI = "cui-desktop" ) -// 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) +// Output are the options for the output +type Output struct { + Writer message.Writer } -// SendGroup sends a message group using the appropriate writer for the context -func SendGroup(ctx *context.Context, group *message.Group) error { - writer, err := GetWriter(ctx) - if err != nil { - return err - } - return writer.WriteGroup(group) -} +// NewOutput creates a new output based on Accept type +func NewOutput(options message.Options) (*Output, error) { + var writer message.Writer + var err error -// 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() + // Create writer based on Accept type + switch options.Accept { + case AcceptStandard: + // OpenAI-compatible format + writer, err = openai.NewWriter(options) - if exists { - return writer, nil + case AcceptWebCUI, AccepNativeCUI, AcceptDesktopCUI: + // CUI format + writer, err = cui.NewWriter(options) + + default: + // Default to Standard (OpenAI) + writer, err = openai.NewWriter(options) } - // 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 + return &Output{ + Writer: 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 Standard - return openai.NewWriter(ctx) - } +// Send sends a single message using the appropriate writer for the context +func (o *Output) Send(msg *message.Message) error { + return o.Writer.Write(msg) } -// 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) +// SendGroup sends a message group using the appropriate writer for the context +func (o *Output) SendGroup(group *message.Group) error { + return o.Writer.WriteGroup(group) } // 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() +func (o *Output) Flush() error { + return o.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 +// Close closes the writer for the given context +func (o *Output) Close() error { + return o.Writer.Close() } -// 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 - } - +// SendMulti sends multiple messages using the appropriate writer for the context +func (o *Output) SendMulti(messages ...*message.Message) error { for _, msg := range messages { - if err := writer.Write(msg); err != nil { + if err := o.Writer.Write(msg); err != nil { return fmt.Errorf("failed to send message: %w", err) } } - return nil } + +// // 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.Group) 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 Standard +// return openai.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 +// } diff --git a/main.go b/main.go index eb10398b..356fb79e 100644 --- a/main.go +++ b/main.go @@ -3,7 +3,6 @@ package main import ( _ "github.com/yaoapp/gou/diff" _ "github.com/yaoapp/gou/encoding" - _ "github.com/yaoapp/yao/agent/output/jsapi" _ "github.com/yaoapp/yao/aigc" _ "github.com/yaoapp/yao/crypto" _ "github.com/yaoapp/yao/excel"