Merge pull request #1346 from trheyi/main

Refactor: Decouple Output From Context
This commit is contained in:
Max 2025-11-25 22:08:34 +08:00 committed by GitHub
commit 3af007bd86
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
31 changed files with 2081 additions and 1165 deletions

View file

@ -7,17 +7,18 @@ import (
jsoniter "github.com/json-iterator/go" jsoniter "github.com/json-iterator/go"
"github.com/yaoapp/gou/connector" "github.com/yaoapp/gou/connector"
"github.com/yaoapp/kun/log" "github.com/yaoapp/kun/log"
"github.com/yaoapp/yao/agent/assistant/handlers"
"github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/i18n"
"github.com/yaoapp/yao/agent/llm" "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/trace/types"
"github.com/yaoapp/yao/utils/jsonschema" "github.com/yaoapp/yao/utils/jsonschema"
) )
// Stream stream the agent // Stream stream the agent
// handler is optional, if not provided, a default handler will be used // 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) 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) 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) ast.sendAgentStreamEnd(ctx, streamHandler, streamStartTime, "completed", nil, completionResponse)
// Close the output writer to send [DONE] marker // 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 { 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" 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 // Info get the assistant information
func (ast *Assistant) Info(locale ...string) *context.AssistantInfo { func (ast *Assistant) Info(locale ...string) *message.AssistantInfo {
lc := "en" lc := "en"
if len(locale) > 0 { if len(locale) > 0 {
lc = locale[0] lc = locale[0]
} }
return &context.AssistantInfo{ return &message.AssistantInfo{
ID: ast.ID, ID: ast.ID,
Type: ast.Type, Type: ast.Type,
Name: i18n.Tr(ast.ID, lc, ast.Name), 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 // 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 { if len(handler) > 0 && handler[0] != nil {
return handler[0] return handler[0]
} }
return llm.DefaultStreamHandler(ctx) return handlers.DefaultStreamHandler(ctx)
} }
// sendAgentStreamStart sends ChunkStreamStart for root stack only (agent-level stream start) // 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 // 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 { if ctx.Stack == nil || !ctx.Stack.IsRoot() || handler == nil {
return return
} }
// Build the start data // Build the start data
startData := context.StreamStartData{ startData := message.StreamStartData{
ContextID: ctx.ID, ContextID: ctx.ID,
ChatID: ctx.ChatID, ChatID: ctx.ChatID,
TraceID: ctx.TraceID(), TraceID: ctx.TraceID(),
@ -714,12 +715,12 @@ func (ast *Assistant) sendAgentStreamStart(ctx *context.Context, handler context
} }
if startJSON, err := jsoniter.Marshal(startData); err == nil { 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) // 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 { if ctx.Stack == nil || !ctx.Stack.IsRoot() || handler == nil {
return return
} }
@ -730,7 +731,7 @@ func (ast *Assistant) sendAgentStreamEnd(ctx *context.Context, handler context.S
return return
} }
endData := &context.StreamEndData{ endData := &message.StreamEndData{
RequestID: ctx.RequestID(), RequestID: ctx.RequestID(),
ContextID: ctx.ID, ContextID: ctx.ID,
Timestamp: time.Now().UnixMilli(), 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 { 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 // 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) ast.sendAgentStreamEnd(ctx, handler, startTime, "error", err, nil)
} }

View file

@ -11,7 +11,7 @@ import (
// DefaultStreamHandler creates a default stream handler that sends messages via context // DefaultStreamHandler creates a default stream handler that sends messages via context
// This handler is used when no custom handler is provided // 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 // Create stream state manager
state := &streamState{ state := &streamState{
@ -20,7 +20,7 @@ func DefaultStreamHandler(ctx *context.Context) context.StreamFunc {
currentID: "", currentID: "",
} }
return func(chunkType context.StreamChunkType, data []byte) int { return func(chunkType message.StreamChunkType, data []byte) int {
trace, _ := ctx.Trace() trace, _ := ctx.Trace()
if trace != nil { if trace != nil {
trace.Info(i18n.T(ctx.Locale, "llm.handlers.stream.info"), map[string]any{"data": string(data)}) 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 // Handle different chunk types
switch chunkType { switch chunkType {
case context.ChunkStreamStart: case message.ChunkStreamStart:
return state.handleStreamStart(data) return state.handleStreamStart(data)
case context.ChunkGroupStart: case message.ChunkGroupStart:
return state.handleGroupStart(data) return state.handleGroupStart(data)
case context.ChunkText: case message.ChunkText:
return state.handleText(data) return state.handleText(data)
case context.ChunkThinking: case message.ChunkThinking:
return state.handleThinking(data) return state.handleThinking(data)
case context.ChunkToolCall: case message.ChunkToolCall:
return state.handleToolCall(data) return state.handleToolCall(data)
case context.ChunkMetadata: case message.ChunkMetadata:
return state.handleMetadata(data) return state.handleMetadata(data)
case context.ChunkError: case message.ChunkError:
return state.handleError(data) return state.handleError(data)
case context.ChunkGroupEnd: case message.ChunkGroupEnd:
return state.handleGroupEnd(data) return state.handleGroupEnd(data)
case context.ChunkStreamEnd: case message.ChunkStreamEnd:
return state.handleStreamEnd(data) return state.handleStreamEnd(data)
default: default:
@ -75,13 +75,13 @@ type streamState struct {
func (s *streamState) handleStreamStart(data []byte) int { func (s *streamState) handleStreamStart(data []byte) int {
// Send event message to indicate stream has started // Send event message to indicate stream has started
// This is a lifecycle event, CUI clients can show it, OpenAI clients will ignore it // 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) err := jsoniter.Unmarshal(data, &startData)
if err != nil { if err != nil {
log.Error("Failed to unmarshal stream start data: %v", err) log.Error("Failed to unmarshal stream start data: %v", err)
} }
msg := output.NewEventMessage("stream_start", "Stream started", startData) msg := output.NewEventMessage("stream_start", "Stream started", startData)
output.Send(s.ctx, msg) s.ctx.Send(msg)
return 0 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 // Log error but continue streaming
return 0 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 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 return 0 // Continue
} }
@ -191,7 +191,7 @@ func (s *streamState) handleMetadata(data []byte) int {
func (s *streamState) handleError(data []byte) int { func (s *streamState) handleError(data []byte) int {
// Send error message // Send error message
msg := output.NewErrorMessage(string(data), "stream_error") msg := output.NewErrorMessage(string(data), "stream_error")
output.Send(s.ctx, msg) s.ctx.Send(msg)
return 1 // Stop streaming on error return 1 // Stop streaming on error
} }
@ -218,7 +218,7 @@ func (s *streamState) handleGroupEnd(data []byte) int {
"content": string(s.buffer), "content": string(s.buffer),
}, },
} }
output.Send(s.ctx, msg) s.ctx.Send(msg)
} }
// Reset state // Reset state
@ -233,19 +233,19 @@ func (s *streamState) handleGroupEnd(data []byte) int {
// handleStreamEnd handles stream end event // handleStreamEnd handles stream end event
func (s *streamState) handleStreamEnd(data []byte) int { func (s *streamState) handleStreamEnd(data []byte) int {
// Parse the stream end data // Parse the stream end data
var endData context.StreamEndData var endData message.StreamEndData
if err := jsoniter.Unmarshal(data, &endData); err != nil { if err := jsoniter.Unmarshal(data, &endData); err != nil {
log.Error("Failed to parse stream_end data: %v", err) log.Error("Failed to parse stream_end data: %v", err)
output.Flush(s.ctx) s.ctx.Flush()
return 0 return 0
} }
// Send stream_end event as a message to frontend // Send stream_end event as a message to frontend
msg := output.NewEventMessage("stream_end", "Stream completed", endData) msg := output.NewEventMessage("stream_end", "Stream completed", endData)
output.Send(s.ctx, msg) s.ctx.Send(msg)
// Flush any remaining data // Flush any remaining data
output.Flush(s.ctx) s.ctx.Flush()
return 0 // Continue (stream will end naturally) return 0 // Continue (stream will end naturally)
} }

View file

@ -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

View file

@ -157,14 +157,14 @@ func (ctx *Context) Release() {
// Send sends data to the context's writer // Send sends data to the context's writer
// This is used by the output module to send messages to the client // This is used by the output module to send messages to the client
func (ctx *Context) Send(data []byte) error { // func (ctx *Context) Send(data []byte) error {
if ctx.Writer == nil { // if ctx.Writer == nil {
return nil // No writer, silently ignore // return nil // No writer, silently ignore
} // }
_, err := ctx.Writer.Write(data) // _, err := ctx.Writer.Write(data)
return err // return err
} // }
// Trace returns the trace manager for this context, lazily initialized on first call // 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 // Uses the TraceID from ctx.Stack if available, or generates a new one

View file

@ -1,6 +1,10 @@
package context 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 // StreamChunkType represents the type of content in a streaming chunk
type StreamChunkType string type StreamChunkType string
@ -23,15 +27,6 @@ const (
ChunkGroupEnd StreamChunkType = "group_end" // Message group ends (text/tool_call/thinking group completes) 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. // 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. // A Writer may not be used after the agent execution has completed.
type Writer = http.ResponseWriter type Writer = http.ResponseWriter
@ -40,7 +35,7 @@ type Writer = http.ResponseWriter
type Agent interface { type Agent interface {
// Stream stream the agent // 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 run the agent
Run(ctx *Context, messages []Message) (*Response, error) Run(ctx *Context, messages []Message) (*Response, error)

View file

@ -45,6 +45,9 @@ func (ctx *Context) NewObject(v8ctx *v8go.Context) (*v8go.Value, error) {
// Set methods // Set methods
jsObject.Set("Trace", ctx.traceMethod(v8ctx.Isolate())) 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 // Create instance
instance, err := jsObject.NewInstance(v8ctx) instance, err := jsObject.NewInstance(v8ctx)
@ -157,3 +160,71 @@ func (ctx *Context) traceMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
return traceObj 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)
})
}

View file

@ -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
}

79
agent/context/output.go Normal file
View file

@ -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
}

View file

@ -6,6 +6,7 @@ import (
"github.com/yaoapp/gou/plan" "github.com/yaoapp/gou/plan"
"github.com/yaoapp/gou/store" "github.com/yaoapp/gou/store"
"github.com/yaoapp/yao/agent/output"
"github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/openapi/oauth/types"
traceTypes "github.com/yaoapp/yao/trace/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 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 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 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) // Model capabilities (set by assistant, used by output adapters)
Capabilities *ModelCapabilities `json:"-"` // Model capabilities for the current connector Capabilities *ModelCapabilities `json:"-"` // Model capabilities for the current connector

View file

@ -1,5 +1,7 @@
package context package context
import "github.com/yaoapp/yao/agent/output/message"
// Uses represents the wrapper configurations for assistant // Uses represents the wrapper configurations for assistant
// Used to specify which assistant or MCP server to use for vision, audio, search, and fetch operations // Used to specify which assistant or MCP server to use for vision, audio, search, and fetch operations
type Uses struct { type Uses struct {
@ -28,16 +30,7 @@ const (
// ModelCapabilities defines the capabilities of a language model // ModelCapabilities defines the capabilities of a language model
// Used by LLM to select appropriate provider and validate requests // Used by LLM to select appropriate provider and validate requests
type ModelCapabilities struct { type ModelCapabilities message.ModelCapabilities
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)
}
// GetVisionSupport returns whether vision is supported and the format // GetVisionSupport returns whether vision is supported and the format
func (m *ModelCapabilities) GetVisionSupport() (bool, VisionFormat) { 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.) FinishReason string `json:"finish_reason"` // Why generation stopped (stop, length, tool_calls, content_filter, etc.)
// Usage statistics // Usage statistics
Usage *UsageInfo `json:"usage,omitempty"` // Token usage statistics Usage *message.UsageInfo `json:"usage,omitempty"` // Token usage statistics
// Additional metadata // Additional metadata
SystemFingerprint string `json:"system_fingerprint,omitempty"` // System fingerprint for reproducibility 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 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 // FinishReason constants - why the model stopped generating tokens
const ( const (
FinishReasonStop = "stop" // Natural stop point or provided stop sequence reached 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{}) 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) 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
}

View file

@ -1,9 +1,12 @@
package llm 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 // LLM the LLM interface
type 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) Post(ctx *context.Context, messages []context.Message, options *context.CompletionOptions) (*context.CompletionResponse, error)
} }

View file

@ -6,11 +6,12 @@ import (
"github.com/yaoapp/gou/connector" "github.com/yaoapp/gou/connector"
"github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/llm/providers/openai" "github.com/yaoapp/yao/agent/llm/providers/openai"
"github.com/yaoapp/yao/agent/output/message"
) )
// LLM interface (copied to avoid import cycle) // LLM interface (copied to avoid import cycle)
type 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) Post(ctx *context.Context, messages []context.Message, options *context.CompletionOptions) (*context.CompletionResponse, error)
} }

View file

@ -8,6 +8,7 @@ import (
"github.com/yaoapp/gou/plan" "github.com/yaoapp/gou/plan"
"github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/llm" "github.com/yaoapp/yao/agent/llm"
"github.com/yaoapp/yao/agent/output/message"
"github.com/yaoapp/yao/config" "github.com/yaoapp/yao/config"
"github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/openapi/oauth/types"
"github.com/yaoapp/yao/test" "github.com/yaoapp/yao/test"
@ -89,7 +90,7 @@ func TestClaudeSonnet4StreamBasic(t *testing.T) {
ctx := newClaudeTestContext("test-claude-sonnet4-basic", "claude.sonnet-4_0") ctx := newClaudeTestContext("test-claude-sonnet4-basic", "claude.sonnet-4_0")
var chunks []string var chunks []string
handler := func(chunkType context.StreamChunkType, data []byte) int { handler := func(chunkType message.StreamChunkType, data []byte) int {
chunks = append(chunks, string(data)) chunks = append(chunks, string(data))
t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) t.Logf("Stream chunk [%s]: %s", chunkType, string(data))
return 0 return 0
@ -406,11 +407,11 @@ func TestClaudeSonnet4ThinkingStream(t *testing.T) {
var thinkingChunks []string var thinkingChunks []string
var textChunks []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)) t.Logf("Stream chunk [%s]: %s", chunkType, string(data))
if chunkType == context.ChunkThinking { if chunkType == message.ChunkThinking {
thinkingChunks = append(thinkingChunks, string(data)) thinkingChunks = append(thinkingChunks, string(data))
} else if chunkType == context.ChunkText { } else if chunkType == message.ChunkText {
textChunks = append(textChunks, string(data)) textChunks = append(textChunks, string(data))
} }
return 0 return 0

View file

@ -10,6 +10,7 @@ import (
"github.com/yaoapp/gou/plan" "github.com/yaoapp/gou/plan"
"github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/llm" "github.com/yaoapp/yao/agent/llm"
"github.com/yaoapp/yao/agent/output/message"
"github.com/yaoapp/yao/config" "github.com/yaoapp/yao/config"
"github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/openapi/oauth/types"
"github.com/yaoapp/yao/test" "github.com/yaoapp/yao/test"
@ -66,20 +67,20 @@ func TestDeepSeekR1StreamBasic(t *testing.T) {
var thinkingGroupEnded bool var thinkingGroupEnded bool
var textGroupEnded bool var textGroupEnded bool
handler := func(chunkType context.StreamChunkType, data []byte) int { handler := func(chunkType message.StreamChunkType, data []byte) int {
dataStr := string(data) dataStr := string(data)
t.Logf("Stream chunk [%s]: %s", chunkType, dataStr) t.Logf("Stream chunk [%s]: %s", chunkType, dataStr)
// Track different chunk types // Track different chunk types
switch chunkType { switch chunkType {
case context.ChunkThinking: case message.ChunkThinking:
reasoningChunks = append(reasoningChunks, dataStr) reasoningChunks = append(reasoningChunks, dataStr)
case context.ChunkText: case message.ChunkText:
contentChunks = append(contentChunks, dataStr) contentChunks = append(contentChunks, dataStr)
} }
// Track group_end events to verify type field // 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 // Parse the group_end data to check the type field
var groupEndData struct { var groupEndData struct {
GroupID string `json:"group_id"` GroupID string `json:"group_id"`
@ -327,10 +328,10 @@ func TestDeepSeekR1LogicPuzzle(t *testing.T) {
// Track reasoning and content separately // Track reasoning and content separately
var hasReasoning, hasContent bool var hasReasoning, hasContent bool
handler := func(chunkType context.StreamChunkType, data []byte) int { handler := func(chunkType message.StreamChunkType, data []byte) int {
if chunkType == context.ChunkThinking && len(data) > 0 { if chunkType == message.ChunkThinking && len(data) > 0 {
hasReasoning = true hasReasoning = true
} else if chunkType == context.ChunkText && len(data) > 0 { } else if chunkType == message.ChunkText && len(data) > 0 {
hasContent = true hasContent = true
} }
return 0 return 0

View file

@ -8,6 +8,7 @@ import (
"github.com/yaoapp/gou/plan" "github.com/yaoapp/gou/plan"
"github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/llm" "github.com/yaoapp/yao/agent/llm"
"github.com/yaoapp/yao/agent/output/message"
"github.com/yaoapp/yao/config" "github.com/yaoapp/yao/config"
"github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/openapi/oauth/types"
"github.com/yaoapp/yao/test" "github.com/yaoapp/yao/test"
@ -57,11 +58,11 @@ func TestDeepSeekV3StreamBasic(t *testing.T) {
// Track streaming chunks // Track streaming chunks
var contentChunks []string var contentChunks []string
handler := func(chunkType context.StreamChunkType, data []byte) int { handler := func(chunkType message.StreamChunkType, data []byte) int {
dataStr := string(data) dataStr := string(data)
t.Logf("Stream chunk [%s]: %s", chunkType, dataStr) t.Logf("Stream chunk [%s]: %s", chunkType, dataStr)
if chunkType == context.ChunkText { if chunkType == message.ChunkText {
contentChunks = append(contentChunks, dataStr) contentChunks = append(contentChunks, dataStr)
} }

View file

@ -8,6 +8,7 @@ import (
"github.com/yaoapp/gou/plan" "github.com/yaoapp/gou/plan"
"github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/llm" "github.com/yaoapp/yao/agent/llm"
"github.com/yaoapp/yao/agent/output/message"
"github.com/yaoapp/yao/config" "github.com/yaoapp/yao/config"
"github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/openapi/oauth/types"
"github.com/yaoapp/yao/test" "github.com/yaoapp/yao/test"
@ -52,7 +53,7 @@ func TestGPT5StreamBasic(t *testing.T) {
ctx := newGPT5TestContext("test-gpt5-basic", "openai.gpt-5") ctx := newGPT5TestContext("test-gpt5-basic", "openai.gpt-5")
var chunks []string var chunks []string
handler := func(chunkType context.StreamChunkType, data []byte) int { handler := func(chunkType message.StreamChunkType, data []byte) int {
chunks = append(chunks, string(data)) chunks = append(chunks, string(data))
t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) t.Logf("Stream chunk [%s]: %s", chunkType, string(data))
return 0 return 0

View file

@ -14,11 +14,12 @@ import (
"github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/i18n"
"github.com/yaoapp/yao/agent/llm/adapters" "github.com/yaoapp/yao/agent/llm/adapters"
"github.com/yaoapp/yao/agent/llm/providers/base" "github.com/yaoapp/yao/agent/llm/providers/base"
"github.com/yaoapp/yao/agent/output/message"
"github.com/yaoapp/yao/utils/jsonschema" "github.com/yaoapp/yao/utils/jsonschema"
) )
// startGroup starts a new group and sends group_start event // 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 { if gt.active {
// End previous group first // End previous group first
gt.endGroup(handler) gt.endGroup(handler)
@ -32,39 +33,39 @@ func (gt *groupTracker) startGroup(groupType context.StreamChunkType, handler co
gt.toolCallInfo = nil gt.toolCallInfo = nil
if handler != nil { if handler != nil {
startData := &context.GroupStartData{ startData := &message.GroupStartData{
GroupID: gt.groupID, GroupID: gt.groupID,
Type: string(groupType), Type: string(groupType),
Timestamp: gt.startTime, Timestamp: gt.startTime,
} }
if startJSON, err := jsoniter.Marshal(startData); err == nil { 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 // 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 { if gt.active {
gt.endGroup(handler) gt.endGroup(handler)
} }
gt.active = true gt.active = true
gt.groupID = fmt.Sprintf("grp_tool_%d", time.Now().UnixNano()) gt.groupID = fmt.Sprintf("grp_tool_%d", time.Now().UnixNano())
gt.groupType = context.ChunkToolCall gt.groupType = message.ChunkToolCall
gt.startTime = time.Now().UnixMilli() gt.startTime = time.Now().UnixMilli()
gt.chunkCount = 0 gt.chunkCount = 0
gt.toolCallInfo = toolCallInfo gt.toolCallInfo = toolCallInfo
if handler != nil { if handler != nil {
startData := &context.GroupStartData{ startData := &message.GroupStartData{
GroupID: gt.groupID, GroupID: gt.groupID,
Type: string(context.ChunkToolCall), Type: string(message.ChunkToolCall),
Timestamp: gt.startTime, Timestamp: gt.startTime,
ToolCall: toolCallInfo, ToolCall: toolCallInfo,
} }
if startJSON, err := jsoniter.Marshal(startData); err == nil { 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 // 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 { if !gt.active {
return return
} }
if handler != nil { if handler != nil {
endData := &context.GroupEndData{ endData := &message.GroupEndData{
GroupID: gt.groupID, GroupID: gt.groupID,
Type: string(gt.groupType), Type: string(gt.groupType),
Timestamp: time.Now().UnixMilli(), Timestamp: time.Now().UnixMilli(),
@ -95,7 +96,7 @@ func (gt *groupTracker) endGroup(handler context.StreamFunc) {
endData.ToolCall = gt.toolCallInfo endData.ToolCall = gt.toolCallInfo
} }
if endJSON, err := jsoniter.Marshal(endData); err == nil { 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 // 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 // Add debug log
trace, _ := ctx.Trace() trace, _ := ctx.Trace()
if trace != nil { 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 // 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() streamStartTime := time.Now()
requestID := fmt.Sprintf("req_%d", streamStartTime.UnixNano()) 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) newMessages, err := adapter.PreprocessMessages(processedMessages)
if err != nil { if err != nil {
if handler != 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) 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 { if err != nil {
// Send error to handler // Send error to handler
if handler != nil { 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) 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) // Handle reasoning content (DeepSeek R1)
if delta.ReasoningContent != "" { if delta.ReasoningContent != "" {
// Start thinking group if not active // Start thinking group if not active
if !groupTracker.active || groupTracker.groupType != context.ChunkThinking { if !groupTracker.active || groupTracker.groupType != message.ChunkThinking {
groupTracker.startGroup(context.ChunkThinking, handler) groupTracker.startGroup(message.ChunkThinking, handler)
} }
accumulator.reasoningContent += delta.ReasoningContent accumulator.reasoningContent += delta.ReasoningContent
if handler != nil { if handler != nil {
handler(context.ChunkThinking, []byte(delta.ReasoningContent)) handler(message.ChunkThinking, []byte(delta.ReasoningContent))
groupTracker.incrementChunk() groupTracker.incrementChunk()
} }
} }
@ -532,13 +533,13 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess
// Handle content // Handle content
if delta.Content != "" { if delta.Content != "" {
// Start text group if not active // Start text group if not active
if !groupTracker.active || groupTracker.groupType != context.ChunkText { if !groupTracker.active || groupTracker.groupType != message.ChunkText {
groupTracker.startGroup(context.ChunkText, handler) groupTracker.startGroup(message.ChunkText, handler)
} }
accumulator.content += delta.Content accumulator.content += delta.Content
if handler != nil { if handler != nil {
handler(context.ChunkText, []byte(delta.Content)) handler(message.ChunkText, []byte(delta.Content))
groupTracker.incrementChunk() groupTracker.incrementChunk()
} }
} }
@ -546,13 +547,13 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess
// Handle refusal // Handle refusal
if delta.Refusal != "" { if delta.Refusal != "" {
// Start refusal group if not active // Start refusal group if not active
if !groupTracker.active || groupTracker.groupType != context.ChunkRefusal { if !groupTracker.active || groupTracker.groupType != message.ChunkRefusal {
groupTracker.startGroup(context.ChunkRefusal, handler) groupTracker.startGroup(message.ChunkRefusal, handler)
} }
accumulator.refusal += delta.Refusal accumulator.refusal += delta.Refusal
if handler != nil { if handler != nil {
handler(context.ChunkRefusal, []byte(delta.Refusal)) handler(message.ChunkRefusal, []byte(delta.Refusal))
groupTracker.incrementChunk() 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 // Start new tool call group when we first see this tool call
if tc.ID != "" { if tc.ID != "" {
toolCallInfo := &context.GroupToolCallInfo{ toolCallInfo := &message.GroupToolCallInfo{
ID: tc.ID, ID: tc.ID,
Name: tc.Function.Name, // May be partial or empty initially Name: tc.Function.Name, // May be partial or empty initially
Index: tc.Index, Index: tc.Index,
@ -600,7 +601,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess
// Notify handler of tool call progress // Notify handler of tool call progress
if handler != nil { if handler != nil {
toolCallData, _ := jsoniter.Marshal(delta.ToolCalls) toolCallData, _ := jsoniter.Marshal(delta.ToolCalls)
handler(context.ChunkToolCall, toolCallData) handler(message.ChunkToolCall, toolCallData)
groupTracker.incrementChunk() groupTracker.incrementChunk()
} }
} }
@ -612,7 +613,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess
// Handle usage (in choices, for older API versions) // Handle usage (in choices, for older API versions)
if chunk.Usage != nil { if chunk.Usage != nil {
accumulator.usage = &context.UsageInfo{ accumulator.usage = &message.UsageInfo{
PromptTokens: chunk.Usage.PromptTokens, PromptTokens: chunk.Usage.PromptTokens,
CompletionTokens: chunk.Usage.CompletionTokens, CompletionTokens: chunk.Usage.CompletionTokens,
TotalTokens: chunk.Usage.TotalTokens, 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) // Check for usage at the top level (newer API versions with stream_options)
if chunk.Usage != nil && accumulator.usage == nil { if chunk.Usage != nil && accumulator.usage == nil {
accumulator.usage = &context.UsageInfo{ accumulator.usage = &message.UsageInfo{
PromptTokens: chunk.Usage.PromptTokens, PromptTokens: chunk.Usage.PromptTokens,
CompletionTokens: chunk.Usage.CompletionTokens, CompletionTokens: chunk.Usage.CompletionTokens,
TotalTokens: chunk.Usage.TotalTokens, TotalTokens: chunk.Usage.TotalTokens,
@ -718,7 +719,7 @@ func (p *Provider) streamWithRetry(ctx *context.Context, messages []context.Mess
// Notify handler of error if provided // Notify handler of error if provided
if handler != nil { if handler != nil {
errData := []byte(err.Error()) errData := []byte(err.Error())
handler(context.ChunkError, errData) handler(message.ChunkError, errData)
} }
return nil, fmt.Errorf("streaming request failed: %w", err) 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 // Notify handler of error if provided
if handler != nil { if handler != nil {
errData := []byte(err.Error()) errData := []byte(err.Error())
handler(context.ChunkError, errData) handler(message.ChunkError, errData)
} }
return nil, err return nil, err
} }

View file

@ -11,6 +11,7 @@ import (
"github.com/yaoapp/gou/plan" "github.com/yaoapp/gou/plan"
"github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/llm" "github.com/yaoapp/yao/agent/llm"
"github.com/yaoapp/yao/agent/output/message"
"github.com/yaoapp/yao/config" "github.com/yaoapp/yao/config"
"github.com/yaoapp/yao/openapi/oauth/types" "github.com/yaoapp/yao/openapi/oauth/types"
"github.com/yaoapp/yao/test" "github.com/yaoapp/yao/test"
@ -58,7 +59,7 @@ func TestOpenAIStreamBasic(t *testing.T) {
// Track streaming chunks // Track streaming chunks
var chunks []string var chunks []string
handler := func(chunkType context.StreamChunkType, data []byte) int { handler := func(chunkType message.StreamChunkType, data []byte) int {
chunks = append(chunks, string(data)) chunks = append(chunks, string(data))
t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) t.Logf("Stream chunk [%s]: %s", chunkType, string(data))
return 0 // Continue return 0 // Continue
@ -243,8 +244,8 @@ func TestOpenAIStreamWithToolCalls(t *testing.T) {
// Track streaming chunks // Track streaming chunks
var toolCallChunks int var toolCallChunks int
handler := func(chunkType context.StreamChunkType, data []byte) int { handler := func(chunkType message.StreamChunkType, data []byte) int {
if chunkType == context.ChunkToolCall { if chunkType == message.ChunkToolCall {
toolCallChunks++ toolCallChunks++
} }
t.Logf("Stream chunk [%s]: %s", chunkType, string(data)) t.Logf("Stream chunk [%s]: %s", chunkType, string(data))
@ -482,7 +483,7 @@ func TestOpenAIStreamWithInvalidToolCall(t *testing.T) {
// Create context // Create context
ctx := newTestContext("test-stream-basic", "openai.gpt-4o") 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 return 0 // Continue
} }
@ -604,8 +605,8 @@ func TestOpenAIStreamChunkTypes(t *testing.T) {
ctx := newTestContext("test-chunk-types", "openai.gpt-4o") ctx := newTestContext("test-chunk-types", "openai.gpt-4o")
// Track chunk types // Track chunk types
chunkTypes := make(map[context.StreamChunkType]int) chunkTypes := make(map[message.StreamChunkType]int)
handler := func(chunkType context.StreamChunkType, data []byte) int { handler := func(chunkType message.StreamChunkType, data []byte) int {
chunkTypes[chunkType]++ chunkTypes[chunkType]++
t.Logf("Received chunk type: %s, data length: %d", chunkType, len(data)) t.Logf("Received chunk type: %s, data length: %d", chunkType, len(data))
return 1 // Continue return 1 // Continue
@ -621,7 +622,7 @@ func TestOpenAIStreamChunkTypes(t *testing.T) {
} }
// Validate chunk types received // Validate chunk types received
if chunkTypes[context.ChunkText] == 0 { if chunkTypes[message.ChunkText] == 0 {
t.Error("Expected to receive ChunkText, but got 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 // Track if error chunk was received
receivedError := false receivedError := false
var errorMessage string var errorMessage string
handler := func(chunkType context.StreamChunkType, data []byte) int { handler := func(chunkType message.StreamChunkType, data []byte) int {
if chunkType == context.ChunkError { if chunkType == message.ChunkError {
receivedError = true receivedError = true
errorMessage = string(data) errorMessage = string(data)
t.Logf("Received error chunk: %s", errorMessage) t.Logf("Received error chunk: %s", errorMessage)
@ -1287,19 +1288,19 @@ func TestOpenAIStreamLifecycleEvents(t *testing.T) {
var events []string var events []string
var groupStartReceived, groupEndReceived bool 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)) events = append(events, string(chunkType))
switch chunkType { switch chunkType {
case context.ChunkStreamStart: case message.ChunkStreamStart:
t.Error("❌ LLM layer should NOT send stream_start (now sent at Agent level)") 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)") t.Error("❌ LLM layer should NOT send stream_end (now sent at Agent level)")
case context.ChunkGroupStart: case message.ChunkGroupStart:
groupStartReceived = true groupStartReceived = true
var startData context.GroupStartData var startData message.GroupStartData
if err := json.Unmarshal(data, &startData); err == nil { if err := json.Unmarshal(data, &startData); err == nil {
t.Logf("✓ group_start: type=%s, group_id=%s", startData.Type, startData.GroupID) t.Logf("✓ group_start: type=%s, group_id=%s", startData.Type, startData.GroupID)
if startData.GroupID == "" { if startData.GroupID == "" {
@ -1309,9 +1310,9 @@ func TestOpenAIStreamLifecycleEvents(t *testing.T) {
t.Errorf("Failed to parse group_start data: %v", err) t.Errorf("Failed to parse group_start data: %v", err)
} }
case context.ChunkGroupEnd: case message.ChunkGroupEnd:
groupEndReceived = true groupEndReceived = true
var endData context.GroupEndData var endData message.GroupEndData
if err := json.Unmarshal(data, &endData); err == nil { if err := json.Unmarshal(data, &endData); err == nil {
t.Logf("✓ group_end: type=%s, chunks=%d, duration=%dms", t.Logf("✓ group_end: type=%s, chunks=%d, duration=%dms",
endData.Type, endData.ChunkCount, endData.DurationMs) 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) t.Errorf("Failed to parse group_end data: %v", err)
} }
case context.ChunkText: case message.ChunkText:
t.Logf(" text chunk: %s", string(data)) t.Logf(" text chunk: %s", string(data))
} }
@ -1394,12 +1395,12 @@ func TestOpenAIStreamContextCancellation(t *testing.T) {
var receivedChunks int var receivedChunks int
handler := func(chunkType context.StreamChunkType, data []byte) int { handler := func(chunkType message.StreamChunkType, data []byte) int {
if chunkType == context.ChunkText || chunkType == context.ChunkToolCall { if chunkType == message.ChunkText || chunkType == message.ChunkToolCall {
receivedChunks++ receivedChunks++
} }
// Note: stream_end is now sent at Agent level, not LLM level // 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)") t.Error("❌ LLM layer should NOT send stream_end (now sent at Agent level)")
} }
return 0 return 0
@ -1467,7 +1468,7 @@ func TestOpenAIStreamWithTemperature(t *testing.T) {
// Use callback to collect chunks // Use callback to collect chunks
chunkCount := 0 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++ chunkCount++
return 1 // Continue return 1 // Continue
} }

View file

@ -1,6 +1,9 @@
package openai 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 // StreamChunk represents a chunk from OpenAI's streaming response
type StreamChunk struct { type StreamChunk struct {
@ -63,7 +66,7 @@ type CompletionResponseFull struct {
} `json:"message"` } `json:"message"`
FinishReason string `json:"finish_reason"` FinishReason string `json:"finish_reason"`
} `json:"choices"` } `json:"choices"`
Usage *context.UsageInfo `json:"usage,omitempty"` Usage *message.UsageInfo `json:"usage,omitempty"`
SystemFingerprint string `json:"system_fingerprint,omitempty"` SystemFingerprint string `json:"system_fingerprint,omitempty"`
} }
@ -78,7 +81,7 @@ type streamAccumulator struct {
refusal string refusal string
toolCalls map[int]*accumulatedToolCall toolCalls map[int]*accumulatedToolCall
finishReason string finishReason string
usage *context.UsageInfo usage *message.UsageInfo
} }
// accumulatedToolCall accumulates a single tool call // accumulatedToolCall accumulates a single tool call
@ -93,8 +96,8 @@ type accumulatedToolCall struct {
type groupTracker struct { type groupTracker struct {
active bool // Whether a group is currently active active bool // Whether a group is currently active
groupID string // Current group ID groupID string // Current group ID
groupType context.StreamChunkType // Current group type groupType message.StreamChunkType // Current group type
startTime int64 // Group start timestamp startTime int64 // Group start timestamp
chunkCount int // Number of chunks in this group 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
} }

View file

@ -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)
}

View file

@ -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
}

View file

@ -2,22 +2,27 @@ package cui
import ( import (
"encoding/json" "encoding/json"
"net/http"
"github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/i18n"
"github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/agent/output/message"
traceTypes "github.com/yaoapp/yao/trace/types"
) )
// Writer implements the message.Writer interface for CUI clients // Writer implements the message.Writer interface for CUI clients
type Writer struct { type Writer struct {
ctx *context.Context Writer http.ResponseWriter
Trace traceTypes.Manager
Locale string
adapter *Adapter adapter *Adapter
} }
// NewWriter creates a new CUI writer // NewWriter creates a new CUI writer
func NewWriter(ctx *context.Context) (*Writer, error) { func NewWriter(options message.Options) (*Writer, error) {
return &Writer{ return &Writer{
ctx: ctx, Writer: options.Writer,
Trace: options.Trace,
Locale: options.Locale,
adapter: NewAdapter(), adapter: NewAdapter(),
}, nil }, nil
} }
@ -27,8 +32,8 @@ func (w *Writer) Write(msg *message.Message) error {
// CUI adapter passes messages through as-is // CUI adapter passes messages through as-is
chunks, err := w.adapter.Adapt(msg) chunks, err := w.adapter.Adapt(msg)
if err != nil { if err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.Trace != nil {
trace.Error(i18n.T(w.ctx.Locale, "output.cui.writer.adapt_error"), map[string]any{ // "CUI Writer: Failed to adapt message" w.Trace.Error(i18n.T(w.Locale, "output.cui.writer.adapt_error"), map[string]any{ // "CUI Writer: Failed to adapt message"
"error": err.Error(), "error": err.Error(),
"message_type": msg.Type, "message_type": msg.Type,
}) })
@ -39,8 +44,8 @@ func (w *Writer) Write(msg *message.Message) error {
// Send each chunk // Send each chunk
for _, chunk := range chunks { for _, chunk := range chunks {
if err := w.sendChunk(chunk); err != nil { if err := w.sendChunk(chunk); err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.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" 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 return err
} }
@ -56,8 +61,8 @@ func (w *Writer) WriteGroup(group *message.Group) error {
// Send the group // Send the group
if err := w.sendChunk(group); err != nil { if err := w.sendChunk(group); err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.Trace != nil {
trace.Error(i18n.T(w.ctx.Locale, "output.cui.writer.group_error"), map[string]any{ // "CUI Writer: Failed to send message group" 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(), "error": err.Error(),
"group_id": group.ID, "group_id": group.ID,
}) })
@ -86,15 +91,15 @@ func (w *Writer) sendChunk(chunk interface{}) error {
// Convert chunk to JSON // Convert chunk to JSON
data, err := json.Marshal(chunk) data, err := json.Marshal(chunk)
if err != nil { if err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.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" 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 return err
} }
// Log outgoing data to trace for debugging // Log outgoing data to trace for debugging
if trace, _ := w.ctx.Trace(); trace != nil { if w.Trace != nil {
trace.Debug("CUI Writer: Sending chunk to client", map[string]any{ w.Trace.Debug("CUI Writer: Sending chunk to client", map[string]any{
"data": string(data), "data": string(data),
}) })
} }
@ -106,18 +111,31 @@ func (w *Writer) sendChunk(chunk interface{}) error {
// Send via context's writer // Send via context's writer
// The context knows how to send data based on the connection type (SSE, WebSocket, etc.) // The context knows how to send data based on the connection type (SSE, WebSocket, etc.)
if err := w.ctx.Send(sseData); err != nil { if err := w.sendData(sseData); err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.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" 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 return err
} }
// Flush immediately to ensure real-time streaming w.flush()
// Cast to http.ResponseWriter and call Flush if available
if flusher, ok := w.ctx.Writer.(interface{ Flush() }); ok {
flusher.Flush()
}
return nil 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
}

View file

@ -4,7 +4,6 @@ import (
"fmt" "fmt"
"time" "time"
"github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/i18n"
"github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/agent/output/message"
) )
@ -152,9 +151,9 @@ func convertStreamStart(msg *message.Message, config *AdapterConfig) ([]interfac
} }
// Try to convert to StreamStartData // Try to convert to StreamStartData
var startData context.StreamStartData var startData message.StreamStartData
switch v := data.(type) { switch v := data.(type) {
case context.StreamStartData: case message.StreamStartData:
startData = v startData = v
case map[string]interface{}: case map[string]interface{}:
// If it's a map, try to extract traceID // If it's a map, try to extract traceID

View file

@ -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
}

View file

@ -2,45 +2,49 @@ package openai
import ( import (
"encoding/json" "encoding/json"
"net/http"
"github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/i18n" "github.com/yaoapp/yao/agent/i18n"
"github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/agent/output/message"
traceTypes "github.com/yaoapp/yao/trace/types"
) )
// Writer implements the message.Writer interface for OpenAI-compatible clients // Writer implements the message.Writer interface for OpenAI-compatible clients
type Writer struct { type Writer struct {
ctx *context.Context Writer http.ResponseWriter
Trace traceTypes.Manager
Locale string
adapter *Adapter adapter *Adapter
firstChunk bool // Track if this is the first chunk to add role firstChunk bool // Track if this is the first chunk to add role
} }
// NewWriter creates a new OpenAI writer // 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) // Get model capabilities from context (set by assistant)
var capabilities *ModelCapabilities var capabilities *ModelCapabilities
if ctx.Capabilities != nil && ctx.Capabilities.Reasoning != nil { if options.Capabilities != nil && options.Capabilities.Reasoning != nil {
capabilities = &ModelCapabilities{ capabilities = &ModelCapabilities{
Reasoning: ctx.Capabilities.Reasoning, Reasoning: options.Capabilities.Reasoning,
} }
} }
// Create adapter with capabilities, base URL, and locale // Create adapter with capabilities, base URL, and locale
adapter := NewAdapter( adapter := NewAdapter(
WithCapabilities(capabilities), WithCapabilities(capabilities),
WithBaseURL(getBaseURL(ctx)), WithBaseURL(getBaseURL(options.BaseURL)),
WithLocale(ctx.Locale), WithLocale(options.Locale),
) )
return &Writer{ return &Writer{
ctx: ctx,
adapter: adapter, adapter: adapter,
Writer: options.Writer,
Locale: options.Locale,
firstChunk: true, // First chunk should include role firstChunk: true, // First chunk should include role
}, nil }, nil
} }
// getBaseURL gets the base URL from context or environment // 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 // @todo: get from context metadata
return "http://localhost:8000/__yao_admin_root" 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 // Convert message to OpenAI format using adapter
chunks, err := w.adapter.Adapt(msg) chunks, err := w.adapter.Adapt(msg)
if err != nil { if err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.Trace != nil {
trace.Error(i18n.T(w.ctx.Locale, "output.openai.writer.adapt_error"), map[string]any{ // "OpenAI Writer: Failed to adapt message" w.Trace.Error(i18n.T(w.Locale, "output.openai.writer.adapt_error"), map[string]any{ // "OpenAI Writer: Failed to adapt message"
"error": err.Error(), "error": err.Error(),
"message_type": msg.Type, "message_type": msg.Type,
}) })
@ -84,8 +88,8 @@ func (w *Writer) Write(msg *message.Message) error {
} }
if err := w.sendChunk(chunk); err != nil { if err := w.sendChunk(chunk); err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.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" 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 return err
} }
@ -100,8 +104,8 @@ func (w *Writer) WriteGroup(group *message.Group) error {
// Just send each message individually // Just send each message individually
for _, msg := range group.Messages { for _, msg := range group.Messages {
if err := w.Write(msg); err != nil { if err := w.Write(msg); err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.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" 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(), "error": err.Error(),
"group_id": group.ID, "group_id": group.ID,
"message_type": msg.Type, "message_type": msg.Type,
@ -127,13 +131,31 @@ func (w *Writer) Close() error {
return w.sendDone() 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 // sendChunk sends a chunk to the output stream in SSE format
func (w *Writer) sendChunk(chunk interface{}) error { func (w *Writer) sendChunk(chunk interface{}) error {
// Convert chunk to JSON // Convert chunk to JSON
data, err := json.Marshal(chunk) data, err := json.Marshal(chunk)
if err != nil { if err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.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" 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 return err
} }
@ -143,25 +165,23 @@ func (w *Writer) sendChunk(chunk interface{}) error {
sseData = append(sseData, []byte("\n\n")...) sseData = append(sseData, []byte("\n\n")...)
// Log outgoing data to trace for debugging // Log outgoing data to trace for debugging
if trace, _ := w.ctx.Trace(); trace != nil { if w.Trace != nil {
trace.Debug("OpenAI Writer: Sending chunk to client", map[string]any{ w.Trace.Debug("OpenAI Writer: Sending chunk to client", map[string]any{
"data": string(data), "data": string(data),
}) })
} }
// Send via context's writer // Send via context's writer
if err := w.ctx.Send(sseData); err != nil { if err := w.sendData(sseData); err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.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" 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 return err
} }
// Flush immediately to ensure real-time streaming // Flush immediately to ensure real-time streaming
// Cast to http.ResponseWriter and call Flush if available // Cast to http.ResponseWriter and call Flush if available
if flusher, ok := w.ctx.Writer.(interface{ Flush() }); ok { w.flush()
flusher.Flush()
}
return nil return nil
} }
@ -169,23 +189,20 @@ func (w *Writer) sendChunk(chunk interface{}) error {
// sendDone sends the final [DONE] message // sendDone sends the final [DONE] message
func (w *Writer) sendDone() error { func (w *Writer) sendDone() error {
// Log completion to trace // Log completion to trace
if trace, _ := w.ctx.Trace(); trace != nil { if w.Trace != nil {
trace.Debug("OpenAI Writer: Sending [DONE] to client") w.Trace.Debug("OpenAI Writer: Sending [DONE] to client")
} }
// OpenAI SSE format uses "data: [DONE]\n\n" to signal completion // OpenAI SSE format uses "data: [DONE]\n\n" to signal completion
doneData := []byte("data: [DONE]\n\n") doneData := []byte("data: [DONE]\n\n")
if err := w.ctx.Send(doneData); err != nil { if err := w.sendData(doneData); err != nil {
if trace, _ := w.ctx.Trace(); trace != nil { if w.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" 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 return err
} }
// Flush the final [DONE] message // Flush the final [DONE] message
if flusher, ok := w.ctx.Writer.(interface{ Flush() }); ok { w.flush()
flusher.Flush()
}
return nil return nil
} }

View file

@ -1,412 +1,401 @@
package jsapi package jsapi
import ( // func init() {
"fmt" // // Auto-register Output JavaScript API when package is imported
// v8.RegisterFunction("Output", ExportFunction)
// }
v8 "github.com/yaoapp/gou/runtime/v8" // // Usage from JavaScript:
"github.com/yaoapp/gou/runtime/v8/bridge" // //
agentContext "github.com/yaoapp/yao/agent/context" // // const output = new Output(ctx)
"github.com/yaoapp/yao/agent/output" // // output.Send({ type: "text", props: { content: "Hello" } })
"github.com/yaoapp/yao/agent/output/message" // // output.Send("Hello") // shorthand for text message
"rogchap.com/v8go" // // output.SendGroup({ id: "group1", messages: [...] })
) // //
// // Objects:
// // - Output: Output manager (constructor)
func init() { // // ExportFunction exports the Output constructor function template
// Auto-register Output JavaScript API when package is imported // // This is used by v8.RegisterFunction
v8.RegisterFunction("Output", ExportFunction) // func ExportFunction(iso *v8go.Isolate) *v8go.FunctionTemplate {
} // return v8go.NewFunctionTemplate(iso, outputConstructor)
// }
// Usage from JavaScript: // // outputConstructor is the JavaScript constructor for Output
// // // Usage: new Output(ctx)
// const output = new Output(ctx) // func outputConstructor(info *v8go.FunctionCallbackInfo) *v8go.Value {
// output.Send({ type: "text", props: { content: "Hello" } }) // v8ctx := info.Context()
// output.Send("Hello") // shorthand for text message // args := info.Args()
// output.SendGroup({ id: "group1", messages: [...] })
//
// Objects:
// - Output: Output manager (constructor)
// ExportFunction exports the Output constructor function template // // Require ctx argument
// This is used by v8.RegisterFunction // if len(args) < 1 {
func ExportFunction(iso *v8go.Isolate) *v8go.FunctionTemplate { // return bridge.JsException(v8ctx, "Output constructor requires a context argument")
return v8go.NewFunctionTemplate(iso, outputConstructor) // }
}
// outputConstructor is the JavaScript constructor for Output // // Get the context object from JavaScript
// Usage: new Output(ctx) // ctxObj, err := args[0].AsObject()
func outputConstructor(info *v8go.FunctionCallbackInfo) *v8go.Value { // if err != nil {
v8ctx := info.Context() // return bridge.JsException(v8ctx, fmt.Sprintf("context must be an object: %s", err))
args := info.Args() // }
// Require ctx argument // // Get the goValueID from internal field (index 0)
if len(args) < 1 { // if ctxObj.InternalFieldCount() < 1 {
return bridge.JsException(v8ctx, "Output constructor requires a context argument") // return bridge.JsException(v8ctx, "context object is missing internal fields")
} // }
// Get the context object from JavaScript // goValueIDValue := ctxObj.GetInternalField(0)
ctxObj, err := args[0].AsObject() // if goValueIDValue == nil || !goValueIDValue.IsString() {
if err != nil { // return bridge.JsException(v8ctx, "context object is missing goValueID")
return bridge.JsException(v8ctx, fmt.Sprintf("context must be an object: %s", err)) // }
}
// Get the goValueID from internal field (index 0) // goValueID := goValueIDValue.String()
if ctxObj.InternalFieldCount() < 1 {
return bridge.JsException(v8ctx, "context object is missing internal fields")
}
goValueIDValue := ctxObj.GetInternalField(0) // // Retrieve the Go context object from bridge registry
if goValueIDValue == nil || !goValueIDValue.IsString() { // goObj := bridge.GetGoObject(goValueID)
return bridge.JsException(v8ctx, "context object is missing 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 // // Create output object
goObj := bridge.GetGoObject(goValueID) // outputObj, err := NewOutputObject(v8ctx, ctx)
if goObj == nil { // if err != nil {
return bridge.JsException(v8ctx, "context object not found in registry") // return bridge.JsException(v8ctx, err.Error())
} // }
// Type assert to *agentContext.Context // return outputObj
ctx, ok := goObj.(*agentContext.Context) // }
if !ok {
return bridge.JsException(v8ctx, fmt.Sprintf("object is not a Context, got %T", goObj))
}
// Create output object // // NewOutputObject creates a JavaScript Output object
outputObj, err := NewOutputObject(v8ctx, ctx) // func NewOutputObject(v8ctx *v8go.Context, ctx *agentContext.Context) (*v8go.Value, error) {
if err != nil { // jsObject := v8go.NewObjectTemplate(v8ctx.Isolate())
return bridge.JsException(v8ctx, err.Error())
}
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 // // Register context in global bridge registry for efficient Go object retrieval
func NewOutputObject(v8ctx *v8go.Context, ctx *agentContext.Context) (*v8go.Value, error) { // // The goValueID will be stored in internal field (index 0) after instance creation
jsObject := v8go.NewObjectTemplate(v8ctx.Isolate()) // goValueID := bridge.RegisterGoObject(ctx)
// Set internal field count to 1 to store the __go_id // // Set methods
// Internal fields are not accessible from JavaScript, providing better security // jsObject.Set("Send", outputSendMethod(v8ctx.Isolate(), ctx))
jsObject.SetInternalFieldCount(1) // jsObject.Set("SendGroup", outputSendGroupMethod(v8ctx.Isolate(), ctx))
// Register context in global bridge registry for efficient Go object retrieval // // Set release function that will be called when JavaScript object is released
// The goValueID will be stored in internal field (index 0) after instance creation // jsObject.Set("__release", outputGoRelease(v8ctx.Isolate()))
goValueID := bridge.RegisterGoObject(ctx)
// Set methods // // Create instance
jsObject.Set("Send", outputSendMethod(v8ctx.Isolate(), ctx)) // instance, err := jsObject.NewInstance(v8ctx)
jsObject.Set("SendGroup", outputSendGroupMethod(v8ctx.Isolate(), ctx)) // 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 // // Store the goValueID in internal field (index 0)
jsObject.Set("__release", outputGoRelease(v8ctx.Isolate())) // // 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 // err = obj.SetInternalField(0, goValueID)
instance, err := jsObject.NewInstance(v8ctx) // if err != nil {
if err != nil { // bridge.ReleaseGoObject(goValueID)
// Clean up: release from global registry if instance creation failed // return nil, err
bridge.ReleaseGoObject(goValueID) // }
return nil, err
}
// Store the goValueID in internal field (index 0) // return instance.Value, nil
// This is not accessible from JavaScript, providing better security // }
obj, err := instance.Value.AsObject()
if err != nil {
bridge.ReleaseGoObject(goValueID)
return nil, err
}
err = obj.SetInternalField(0, goValueID) // // outputGoRelease releases the Go object from the global bridge registry
if err != nil { // // It retrieves the goValueID from internal field (index 0) and releases the Go object
bridge.ReleaseGoObject(goValueID) // func outputGoRelease(iso *v8go.Isolate) *v8go.FunctionTemplate {
return nil, err // 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 // // outputSendMethod implements the Send method
// It retrieves the goValueID from internal field (index 0) and releases the Go object // // Usage: output.Send(message)
func outputGoRelease(iso *v8go.Isolate) *v8go.FunctionTemplate { // // message can be an object with { type: string, props: object, ... } or a simple string (will be converted to text message)
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { // func outputSendMethod(iso *v8go.Isolate, ctx *agentContext.Context) *v8go.FunctionTemplate {
// Get the output object (this) // return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
thisObj, err := info.This().AsObject() // v8ctx := info.Context()
if err == nil && thisObj.InternalFieldCount() > 0 { // args := info.Args()
// 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 v8go.Undefined(info.Context().Isolate()) // if len(args) < 1 {
}) // return bridge.JsException(v8ctx, "Send requires a message argument")
} // }
// outputSendMethod implements the Send method // // Parse message argument
// Usage: output.Send(message) // msg, err := parseMessage(v8ctx, args[0])
// message can be an object with { type: string, props: object, ... } or a simple string (will be converted to text message) // if err != nil {
func outputSendMethod(iso *v8go.Isolate, ctx *agentContext.Context) *v8go.FunctionTemplate { // return bridge.JsException(v8ctx, fmt.Sprintf("invalid message: %s", err))
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { // }
v8ctx := info.Context()
args := info.Args()
if len(args) < 1 { // // Call output.Send
return bridge.JsException(v8ctx, "Send requires a message argument") // if err := output.Send(ctx, msg); err != nil {
} // return bridge.JsException(v8ctx, fmt.Sprintf("Send failed: %s", err))
// }
// Parse message argument // return info.This().Value
msg, err := parseMessage(v8ctx, args[0]) // })
if err != nil { // }
return bridge.JsException(v8ctx, fmt.Sprintf("invalid message: %s", err))
}
// Call output.Send // // outputSendGroupMethod implements the SendGroup method
if err := output.Send(ctx, msg); err != nil { // // Usage: output.SendGroup(group)
return bridge.JsException(v8ctx, fmt.Sprintf("Send failed: %s", err)) // // 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 // // Parse group argument
// Usage: output.SendGroup(group) // group, err := parseGroup(v8ctx, args[0])
// group must be an object with { id: string, messages: [], ... } // if err != nil {
func outputSendGroupMethod(iso *v8go.Isolate, ctx *agentContext.Context) *v8go.FunctionTemplate { // return bridge.JsException(v8ctx, fmt.Sprintf("invalid group: %s", err))
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value { // }
v8ctx := info.Context()
args := info.Args()
if len(args) < 1 { // // Call output.SendGroup
return bridge.JsException(v8ctx, "SendGroup requires a group argument") // if err := output.SendGroup(ctx, group); err != nil {
} // return bridge.JsException(v8ctx, fmt.Sprintf("SendGroup failed: %s", err))
// }
// Parse group argument // return info.This().Value
group, err := parseGroup(v8ctx, args[0]) // })
if err != nil { // }
return bridge.JsException(v8ctx, fmt.Sprintf("invalid group: %s", err))
}
// Call output.SendGroup // // parseMessage parses a JavaScript value into a message.Message
if err := output.SendGroup(ctx, group); err != nil { // func parseMessage(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Message, error) {
return bridge.JsException(v8ctx, fmt.Sprintf("SendGroup failed: %s", err)) // // 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 // // Convert to Go map
func parseMessage(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Message, error) { // goValue, err := bridge.GoValue(jsValue, v8ctx)
// Handle string shorthand: convert to text message // if err != nil {
if jsValue.IsString() { // return nil, fmt.Errorf("failed to convert message: %w", err)
return &message.Message{ // }
Type: message.TypeText,
Props: map[string]interface{}{
"content": jsValue.String(),
},
}, nil
}
// Handle object // msgMap, ok := goValue.(map[string]interface{})
if !jsValue.IsObject() { // if !ok {
return nil, fmt.Errorf("message must be a string or object") // return nil, fmt.Errorf("message must be an object")
} // }
// Convert to Go map // // Build message
goValue, err := bridge.GoValue(jsValue, v8ctx) // msg := &message.Message{}
if err != nil {
return nil, fmt.Errorf("failed to convert message: %w", err)
}
msgMap, ok := goValue.(map[string]interface{}) // // Type field (required)
if !ok { // if msgType, ok := msgMap["type"].(string); ok {
return nil, fmt.Errorf("message must be an object") // msg.Type = msgType
} // } else {
// return nil, fmt.Errorf("message.type is required and must be a string")
// }
// Build message // // Props field (optional)
msg := &message.Message{} // if props, ok := msgMap["props"].(map[string]interface{}); ok {
// msg.Props = props
// }
// Type field (required) // // Optional fields
if msgType, ok := msgMap["type"].(string); ok { // if id, ok := msgMap["id"].(string); ok {
msg.Type = msgType // msg.ID = id
} else { // }
return nil, fmt.Errorf("message.type is required and must be a string") // 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) // // Metadata (optional)
if props, ok := msgMap["props"].(map[string]interface{}); ok { // if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok {
msg.Props = props // 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 // return msg, nil
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) // // parseGroup parses a JavaScript value into a message.Group
if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok { // func parseGroup(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Group, error) {
metadata := &message.Metadata{} // // Must be an object
if timestamp, ok := metadataMap["timestamp"].(float64); ok { // if !jsValue.IsObject() {
metadata.Timestamp = int64(timestamp) // return nil, fmt.Errorf("group must be an object")
} // }
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 // // 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 // groupMap, ok := goValue.(map[string]interface{})
func parseGroup(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Group, error) { // if !ok {
// Must be an object // return nil, fmt.Errorf("group must be an object")
if !jsValue.IsObject() { // }
return nil, fmt.Errorf("group must be an object")
}
// Convert to Go map // // Build group
goValue, err := bridge.GoValue(jsValue, v8ctx) // group := &message.Group{}
if err != nil {
return nil, fmt.Errorf("failed to convert group: %w", err)
}
groupMap, ok := goValue.(map[string]interface{}) // // ID field (required)
if !ok { // if id, ok := groupMap["id"].(string); ok {
return nil, fmt.Errorf("group must be an object") // group.ID = id
} // } else {
// return nil, fmt.Errorf("group.id is required and must be a string")
// }
// Build group // // Messages field (required)
group := &message.Group{} // 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) // // Convert map to Message
if id, ok := groupMap["id"].(string); ok { // msg := &message.Message{}
group.ID = id
} else {
return nil, fmt.Errorf("group.id is required and must be a string")
}
// Messages field (required) // // Type field (required)
if messagesArray, ok := groupMap["messages"].([]interface{}); ok { // if msgType, ok := msgMap["type"].(string); ok {
group.Messages = make([]*message.Message, 0, len(messagesArray)) // msg.Type = msgType
for i, msgInterface := range messagesArray { // } else {
// Convert to map // return nil, fmt.Errorf("group.messages[%d].type is required", i)
msgMap, ok := msgInterface.(map[string]interface{}) // }
if !ok {
return nil, fmt.Errorf("group.messages[%d] must be an object", i)
}
// Convert map to Message // // Props field (optional)
msg := &message.Message{} // if props, ok := msgMap["props"].(map[string]interface{}); ok {
// msg.Props = props
// }
// Type field (required) // // Optional fields
if msgType, ok := msgMap["type"].(string); ok { // if id, ok := msgMap["id"].(string); ok {
msg.Type = msgType // msg.ID = id
} else { // }
return nil, fmt.Errorf("group.messages[%d].type is required", i) // 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) // // Metadata (optional)
if props, ok := msgMap["props"].(map[string]interface{}); ok { // if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok {
msg.Props = props // 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 // group.Messages = append(group.Messages, msg)
if id, ok := msgMap["id"].(string); ok { // }
msg.ID = id // } else {
} // return nil, fmt.Errorf("group.messages is required and must be an array")
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) // // Metadata (optional)
if metadataMap, ok := msgMap["metadata"].(map[string]interface{}); ok { // if metadataMap, ok := groupMap["metadata"].(map[string]interface{}); ok {
metadata := &message.Metadata{} // metadata := &message.Metadata{}
if timestamp, ok := metadataMap["timestamp"].(float64); ok { // if timestamp, ok := metadataMap["timestamp"].(float64); ok {
metadata.Timestamp = int64(timestamp) // metadata.Timestamp = int64(timestamp)
} // }
if sequence, ok := metadataMap["sequence"].(float64); ok { // if sequence, ok := metadataMap["sequence"].(float64); ok {
metadata.Sequence = int(sequence) // metadata.Sequence = int(sequence)
} // }
if traceID, ok := metadataMap["trace_id"].(string); ok { // if traceID, ok := metadataMap["trace_id"].(string); ok {
metadata.TraceID = traceID // metadata.TraceID = traceID
} // }
msg.Metadata = metadata // group.Metadata = metadata
} // }
group.Messages = append(group.Messages, msg) // return group, nil
} // }
} 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
}

View file

@ -1,361 +1,349 @@
package jsapi package jsapi
import ( // func TestOutputConstructor(t *testing.T) {
"context" // test.Prepare(t, config.Conf)
"net/http" // defer test.Clean()
"testing"
"github.com/stretchr/testify/assert" // tests := []struct {
v8 "github.com/yaoapp/gou/runtime/v8" // name string
agentContext "github.com/yaoapp/yao/agent/context" // script string
"github.com/yaoapp/yao/config" // expectError bool
"github.com/yaoapp/yao/test" // }{
) // {
// 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) { // for _, tt := range tests {
test.Prepare(t, config.Conf) // t.Run(tt.name, func(t *testing.T) {
defer test.Clean() // ctx := agentContext.New(context.Background(), nil, "test-chat-123", "")
// ctx.AssistantID = "test-assistant-456"
tests := []struct { // // Execute test script with v8.Call
name string // res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx)
script string // if tt.expectError {
expectError bool // assert.Error(t, err)
}{ // return
{ // }
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,
},
}
for _, tt := range tests { // assert.NoError(t, err)
t.Run(tt.name, func(t *testing.T) { // assert.True(t, res.(bool))
ctx := agentContext.New(context.Background(), nil, "test-chat-123", "") // })
ctx.AssistantID = "test-assistant-456" // }
// }
// Execute test script with v8.Call // func TestOutputSend(t *testing.T) {
res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx) // test.Prepare(t, config.Conf)
if tt.expectError { // defer test.Clean()
assert.Error(t, err)
return
}
assert.NoError(t, err) // tests := []struct {
assert.True(t, res.(bool)) // 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) { // for _, tt := range tests {
test.Prepare(t, config.Conf) // t.Run(tt.name, func(t *testing.T) {
defer test.Clean() // // Create context with mock writer
// ctx := agentContext.New(context.Background(), nil, "test-chat", "")
// ctx.Writer = &mockWriter{}
tests := []struct { // // Execute test script with v8.Call
name string // res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx)
script string // if tt.expectError {
expectError bool // assert.Error(t, err)
validate func(*testing.T, *agentContext.Context) // return
}{ // }
{
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,
},
}
for _, tt := range tests { // assert.NoError(t, err)
t.Run(tt.name, func(t *testing.T) { // assert.True(t, res.(bool))
// Create context with mock writer
ctx := agentContext.New(context.Background(), nil, "test-chat", "")
ctx.Writer = &mockWriter{}
// Execute test script with v8.Call // if tt.validate != nil {
res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx) // tt.validate(t, &ctx)
if tt.expectError { // }
assert.Error(t, err) // })
return // }
} // }
assert.NoError(t, err) // func TestOutputSendGroup(t *testing.T) {
assert.True(t, res.(bool)) // test.Prepare(t, config.Conf)
// defer test.Clean()
if tt.validate != nil { // tests := []struct {
tt.validate(t, &ctx) // 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) { // for _, tt := range tests {
test.Prepare(t, config.Conf) // t.Run(tt.name, func(t *testing.T) {
defer test.Clean() // // Create context with mock writer
// ctx := agentContext.New(context.Background(), nil, "test-chat", "")
// ctx.Writer = &mockWriter{}
tests := []struct { // // Execute test script with v8.Call
name string // res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx)
script string // if tt.expectError {
expectError bool // assert.Error(t, err)
}{ // return
{ // }
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,
},
}
for _, tt := range tests { // assert.NoError(t, err)
t.Run(tt.name, func(t *testing.T) { // assert.True(t, res.(bool))
// Create context with mock writer // })
ctx := agentContext.New(context.Background(), nil, "test-chat", "") // }
ctx.Writer = &mockWriter{} // }
// Execute test script with v8.Call // func TestOutputChaining(t *testing.T) {
res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx) // test.Prepare(t, config.Conf)
if tt.expectError { // defer test.Clean()
assert.Error(t, err)
return
}
assert.NoError(t, err) // script := `
assert.True(t, res.(bool)) // function test(ctx) {
}) // const output = new Output(ctx);
}
}
func TestOutputChaining(t *testing.T) { // // Send should return the output object for chaining
test.Prepare(t, config.Conf) // const result = output.Send("Message 1");
defer test.Clean()
script := ` // // Should be able to chain sends
function test(ctx) { // output.Send("Message 2").Send("Message 3");
const output = new Output(ctx);
// Send should return the output object for chaining // return result !== undefined;
const result = output.Send("Message 1"); // }
// `
// Should be able to chain sends // ctx := agentContext.New(context.Background(), nil, "test-chat", "")
output.Send("Message 2").Send("Message 3"); // 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", "") // // mockWriter is a mock implementation of http.ResponseWriter for testing
ctx.Writer = &mockWriter{} // type mockWriter struct {
// data [][]byte
// header http.Header
// }
// Execute test script with v8.Call // func (w *mockWriter) Header() http.Header {
res, err := v8.Call(v8.CallOptions{}, script, &ctx) // if w.header == nil {
assert.NoError(t, err) // w.header = make(http.Header)
assert.True(t, res.(bool)) // }
} // return w.header
// }
// mockWriter is a mock implementation of http.ResponseWriter for testing // func (w *mockWriter) Write(p []byte) (n int, err error) {
type mockWriter struct { // w.data = append(w.data, p)
data [][]byte // return len(p), nil
header http.Header // }
}
func (w *mockWriter) Header() http.Header { // func (w *mockWriter) WriteHeader(statusCode int) {}
if w.header == nil {
w.header = make(http.Header)
}
return w.header
}
func (w *mockWriter) Write(p []byte) (n int, err error) { // func (w *mockWriter) Flush() {}
w.data = append(w.data, p)
return len(p), nil
}
func (w *mockWriter) WriteHeader(statusCode int) {}
func (w *mockWriter) Flush() {}

View file

@ -1,7 +1,5 @@
package message package message
import "github.com/yaoapp/yao/agent/context"
// Writer is the interface for writing output messages // Writer is the interface for writing output messages
// Different writers handle different output formats (SSE, WebSocket, Standard, etc.) // Different writers handle different output formats (SSE, WebSocket, Standard, etc.)
type Writer interface { type Writer interface {
@ -29,23 +27,11 @@ type Adapter interface {
SupportsType(msgType string) bool 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 // StreamHandler handles streaming message processing
// It bridges between LLM streaming chunks and output messages // It bridges between LLM streaming chunks and output messages
type StreamHandler interface { type StreamHandler interface {
// Handle processes a streaming chunk from LLM // 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 flushes any pending messages
Flush() error Flush() error

View file

@ -1,5 +1,34 @@
package message 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) // Message represents a universal message structure (DSL)
// All messages are expressed through Type + Props, without predefining specific types // All messages are expressed through Type + Props, without predefining specific types
type Message struct { type Message struct {
@ -185,3 +214,133 @@ const (
DeltaMerge = "merge" // Merge (for objects) DeltaMerge = "merge" // Merge (for objects)
DeltaSet = "set" // Set (for new fields) 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
}

View file

@ -2,148 +2,212 @@ package output
import ( import (
"fmt" "fmt"
"sync"
"github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/output/adapters/cui" "github.com/yaoapp/yao/agent/output/adapters/cui"
"github.com/yaoapp/yao/agent/output/adapters/openai" "github.com/yaoapp/yao/agent/output/adapters/openai"
"github.com/yaoapp/yao/agent/output/message" "github.com/yaoapp/yao/agent/output/message"
) )
var ( // Accept type constants
writerCache = make(map[*context.Context]message.Writer) const (
writerMutex sync.RWMutex AcceptStandard = "standard"
globalFactory message.WriterFactory AcceptWebCUI = "cui-web"
AccepNativeCUI = "cui-native"
AcceptDesktopCUI = "cui-desktop"
) )
// Send sends a single message using the appropriate writer for the context // Output are the options for the output
func Send(ctx *context.Context, msg *message.Message) error { type Output struct {
writer, err := GetWriter(ctx) Writer message.Writer
if err != nil {
return err
}
return writer.Write(msg)
} }
// SendGroup sends a message group using the appropriate writer for the context // NewOutput creates a new output based on Accept type
func SendGroup(ctx *context.Context, group *message.Group) error { func NewOutput(options message.Options) (*Output, error) {
writer, err := GetWriter(ctx) var writer message.Writer
if err != nil { var err error
return err
}
return writer.WriteGroup(group)
}
// GetWriter gets or creates a writer for the given context // Create writer based on Accept type
// Writers are cached per context to avoid recreating them switch options.Accept {
func GetWriter(ctx *context.Context) (message.Writer, error) { case AcceptStandard:
// Try to get cached writer // OpenAI-compatible format
writerMutex.RLock() writer, err = openai.NewWriter(options)
writer, exists := writerCache[ctx]
writerMutex.RUnlock()
if exists { case AcceptWebCUI, AccepNativeCUI, AcceptDesktopCUI:
return writer, nil // 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 { if err != nil {
return nil, err return nil, err
} }
// Cache the writer return &Output{
writerCache[ctx] = writer Writer: writer,
}, nil
return writer, nil
} }
// createWriter creates a writer based on context.Accept // Send sends a single message using the appropriate writer for the context
func createWriter(ctx *context.Context) (message.Writer, error) { func (o *Output) Send(msg *message.Message) error {
// If global factory is set, use it return o.Writer.Write(msg)
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 // SendGroup sends a message group using the appropriate writer for the context
// This allows applications to provide their own writer implementations func (o *Output) SendGroup(group *message.Group) error {
func SetWriterFactory(factory message.WriterFactory) { return o.Writer.WriteGroup(group)
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 // Flush flushes the writer for the given context
func Flush(ctx *context.Context) error { func (o *Output) Flush() error {
writer, err := GetWriter(ctx) return o.Writer.Flush()
if err != nil {
return err
}
return writer.Flush()
} }
// Close closes the writer for the given context and removes it from cache // Close closes the writer for the given context
func Close(ctx *context.Context) error { func (o *Output) Close() error {
writer, err := GetWriter(ctx) return o.Writer.Close()
if err != nil {
return err
}
err = writer.Close()
ClearWriterCache(ctx)
return err
} }
// SendMulti is a convenience function to send multiple messages // SendMulti sends multiple messages using the appropriate writer for the context
func SendMulti(ctx *context.Context, messages ...*message.Message) error { func (o *Output) SendMulti(messages ...*message.Message) error {
writer, err := GetWriter(ctx)
if err != nil {
return err
}
for _, msg := range messages { 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 fmt.Errorf("failed to send message: %w", err)
} }
} }
return nil 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
// }

View file

@ -3,7 +3,6 @@ package main
import ( import (
_ "github.com/yaoapp/gou/diff" _ "github.com/yaoapp/gou/diff"
_ "github.com/yaoapp/gou/encoding" _ "github.com/yaoapp/gou/encoding"
_ "github.com/yaoapp/yao/agent/output/jsapi"
_ "github.com/yaoapp/yao/aigc" _ "github.com/yaoapp/yao/aigc"
_ "github.com/yaoapp/yao/crypto" _ "github.com/yaoapp/yao/crypto"
_ "github.com/yaoapp/yao/excel" _ "github.com/yaoapp/yao/excel"