opencode/yao
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor: Decouple Output From Context

Browse source
This commit is contained in:
Max 2025-11-25 22:08:00 +08:00
parent 824757aec7
commit 673e36a7c9
31 changed files with 2081 additions and 1165 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
agent/assistant/agent.go
View file

@ -7,17 +7,18 @@ import (
jsoniter "github.com/json-iterator/go"
"github.com/yaoapp/gou/connector"
"github.com/yaoapp/kun/log"
"github.com/yaoapp/yao/agent/assistant/handlers"
"github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/i18n"
"github.com/yaoapp/yao/agent/llm"
"github.com/yaoapp/yao/agent/output"
"github.com/yaoapp/yao/agent/output/message"
"github.com/yaoapp/yao/trace/types"
"github.com/yaoapp/yao/utils/jsonschema"
)
// Stream stream the agent
// handler is optional, if not provided, a default handler will be used
func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Message, handler ...context.StreamFunc) (*context.Response, error) {
func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Message, handler ...message.StreamFunc) (*context.Response, error) {
log.Trace("[AGENT] Stream started: assistant=%s, contextID=%s", ast.ID, ctx.ID)
defer log.Trace("[AGENT] Stream ended: assistant=%s, contextID=%s", ast.ID, ctx.ID)
@ -232,7 +233,7 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa
ast.sendAgentStreamEnd(ctx, streamHandler, streamStartTime, "completed", nil, completionResponse)
// Close the output writer to send [DONE] marker
if err := output.Close(ctx); err != nil {
if err := ctx.CloseOutput(); err != nil {
if trace, _ := ctx.Trace(); trace != nil {
trace.Error(i18n.Tr(ast.ID, ctx.Locale, "assistant.agent.stream.close_error"), map[string]any{"error": err.Error()}) // "Failed to close output"
}
@ -361,12 +362,12 @@ func (ast *Assistant) BuildRequest(ctx *context.Context, messages []context.Mess
}
// Info get the assistant information
func (ast *Assistant) Info(locale ...string) *context.AssistantInfo {
func (ast *Assistant) Info(locale ...string) *message.AssistantInfo {
lc := "en"
if len(locale) > 0 {
lc = locale[0]
}
return &context.AssistantInfo{
return &message.AssistantInfo{
ID: ast.ID,
Type: ast.Type,
Name: i18n.Tr(ast.ID, lc, ast.Name),
@ -688,22 +689,22 @@ func (ast *Assistant) WithHistory(ctx *context.Context, messages []context.Messa
}
// getStreamHandler returns the stream handler from the provided handlers or a default one
func (ast *Assistant) getStreamHandler(ctx *context.Context, handler ...context.StreamFunc) context.StreamFunc {
func (ast *Assistant) getStreamHandler(ctx *context.Context, handler ...message.StreamFunc) message.StreamFunc {
if len(handler) > 0 && handler[0] != nil {
return handler[0]
}
return llm.DefaultStreamHandler(ctx)
return handlers.DefaultStreamHandler(ctx)
}
// sendAgentStreamStart sends ChunkStreamStart for root stack only (agent-level stream start)
// This ensures only one stream_start per agent execution, even with multiple LLM calls
func (ast *Assistant) sendAgentStreamStart(ctx *context.Context, handler context.StreamFunc, startTime time.Time) {
func (ast *Assistant) sendAgentStreamStart(ctx *context.Context, handler message.StreamFunc, startTime time.Time) {
if ctx.Stack == nil || !ctx.Stack.IsRoot() || handler == nil {
return
}
// Build the start data
startData := context.StreamStartData{
startData := message.StreamStartData{
ContextID: ctx.ID,
ChatID: ctx.ChatID,
TraceID: ctx.TraceID(),
@ -714,12 +715,12 @@ func (ast *Assistant) sendAgentStreamStart(ctx *context.Context, handler context
}
if startJSON, err := jsoniter.Marshal(startData); err == nil {
handler(context.ChunkStreamStart, startJSON)
handler(message.ChunkStreamStart, startJSON)
}
}
// sendAgentStreamEnd sends ChunkStreamEnd for root stack only (agent-level stream completion)
func (ast *Assistant) sendAgentStreamEnd(ctx *context.Context, handler context.StreamFunc, startTime time.Time, status string, err error, response *context.CompletionResponse) {
func (ast *Assistant) sendAgentStreamEnd(ctx *context.Context, handler message.StreamFunc, startTime time.Time, status string, err error, response *context.CompletionResponse) {
if ctx.Stack == nil || !ctx.Stack.IsRoot() || handler == nil {
return
}
@ -730,7 +731,7 @@ func (ast *Assistant) sendAgentStreamEnd(ctx *context.Context, handler context.S
return
}
endData := &context.StreamEndData{
endData := &message.StreamEndData{
RequestID: ctx.RequestID(),
ContextID: ctx.ID,
Timestamp: time.Now().UnixMilli(),
@ -749,12 +750,12 @@ func (ast *Assistant) sendAgentStreamEnd(ctx *context.Context, handler context.S
}
if endJSON, marshalErr := jsoniter.Marshal(endData); marshalErr == nil {
handler(context.ChunkStreamEnd, endJSON)
handler(message.ChunkStreamEnd, endJSON)
}
}
// sendStreamEndOnError sends ChunkStreamEnd with error status for root stack only
func (ast *Assistant) sendStreamEndOnError(ctx *context.Context, handler context.StreamFunc, startTime time.Time, err error) {
func (ast *Assistant) sendStreamEndOnError(ctx *context.Context, handler message.StreamFunc, startTime time.Time, err error) {
ast.sendAgentStreamEnd(ctx, handler, startTime, "error", err, nil)
}

44
agent/llm/handlers/stream.go → agent/assistant/handlers/stream.go
View file

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

477
agent/context/JSAPI_OUTPUT.md Normal file
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

14
agent/context/context.go
View file

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

17
agent/context/interfaces.go
View file

@ -1,6 +1,10 @@
package context
import "net/http"
import (
"net/http"
"github.com/yaoapp/yao/agent/output/message"
)
// StreamChunkType represents the type of content in a streaming chunk
type StreamChunkType string
@ -23,15 +27,6 @@ const (
ChunkGroupEnd StreamChunkType = "group_end" // Message group ends (text/tool_call/thinking group completes)
)
// StreamFunc the streaming function callback
// Parameters:
// - chunkType: the type of content in this chunk (text, thinking, tool_call, etc.)
// - data: the actual chunk data (could be text, JSON, or other format)
//
// Returns:
// - int: status code (0 = continue, non-zero = stop streaming)
type StreamFunc func(chunkType StreamChunkType, data []byte) int
// Writer is an alias for http.ResponseWriter interface used by an agent to construct a response.
// A Writer may not be used after the agent execution has completed.
type Writer = http.ResponseWriter
@ -40,7 +35,7 @@ type Writer = http.ResponseWriter
type Agent interface {
// Stream stream the agent
Stream(ctx *Context, messages []Message, handler StreamFunc) error
Stream(ctx *Context, messages []Message, handler message.StreamFunc) error
// Run run the agent
Run(ctx *Context, messages []Message) (*Response, error)

71
agent/context/jsapi.go
View file

@ -45,6 +45,9 @@ func (ctx *Context) NewObject(v8ctx *v8go.Context) (*v8go.Value, error) {
// Set methods
jsObject.Set("Trace", ctx.traceMethod(v8ctx.Isolate()))
jsObject.Set("Send", ctx.sendMethod(v8ctx.Isolate()))
jsObject.Set("SendGroup", ctx.sendGroupMethod(v8ctx.Isolate()))
jsObject.Set("Flush", ctx.flushMethod(v8ctx.Isolate()))
// Create instance
instance, err := jsObject.NewInstance(v8ctx)
@ -157,3 +160,71 @@ func (ctx *Context) traceMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
return traceObj
})
}
// sendMethod implements ctx.Send(message)
// Usage: ctx.Send({ type: "text", props: { content: "Hello" } })
// Usage: ctx.Send("Hello") // shorthand for text message
func (ctx *Context) sendMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
v8ctx := info.Context()
args := info.Args()
if len(args) < 1 {
return bridge.JsException(v8ctx, "Send requires a message argument")
}
// Parse message argument
msg, err := parseMessage(v8ctx, args[0])
if err != nil {
return bridge.JsException(v8ctx, "invalid message: "+err.Error())
}
// Call ctx.Send
if err := ctx.Send(msg); err != nil {
return bridge.JsException(v8ctx, "Send failed: "+err.Error())
}
return v8go.Undefined(iso)
})
}
// sendGroupMethod implements ctx.SendGroup(group)
// Usage: ctx.SendGroup({ id: "group1", messages: [...] })
func (ctx *Context) sendGroupMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
v8ctx := info.Context()
args := info.Args()
if len(args) < 1 {
return bridge.JsException(v8ctx, "SendGroup requires a group argument")
}
// Parse group argument
group, err := parseGroup(v8ctx, args[0])
if err != nil {
return bridge.JsException(v8ctx, "invalid group: "+err.Error())
}
// Call ctx.SendGroup
if err := ctx.SendGroup(group); err != nil {
return bridge.JsException(v8ctx, "SendGroup failed: "+err.Error())
}
return v8go.Undefined(iso)
})
}
// flushMethod implements ctx.Flush()
// Usage: ctx.Flush()
func (ctx *Context) flushMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
v8ctx := info.Context()
// Call ctx.Flush
if err := ctx.Flush(); err != nil {
return bridge.JsException(v8ctx, "Flush failed: "+err.Error())
}
return v8go.Undefined(iso)
})
}

221
agent/context/jsapi_helpers.go Normal file
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
}

2
agent/context/types.go
View file

@ -6,6 +6,7 @@ import (
"github.com/yaoapp/gou/plan"
"github.com/yaoapp/gou/store"
"github.com/yaoapp/yao/agent/output"
"github.com/yaoapp/yao/openapi/oauth/types"
traceTypes "github.com/yaoapp/yao/trace/types"
)
@ -205,6 +206,7 @@ type Context struct {
Writer Writer `json:"-"` // Writer, it will be used to write response data to the client
Skip *Skip `json:"skip,omitempty"` // Skip configuration (history, trace, etc.), nil means don't skip anything
trace traceTypes.Manager `json:"-"` // Trace manager, lazy initialized on first access
output *output.Output `json:"-"` // Output, it will be used to write response data to the client
// Model capabilities (set by assistant, used by output adapters)
Capabilities *ModelCapabilities `json:"-"` // Model capabilities for the current connector

106
agent/context/types_llm.go
View file

@ -1,5 +1,7 @@
package context
import "github.com/yaoapp/yao/agent/output/message"
// Uses represents the wrapper configurations for assistant
// Used to specify which assistant or MCP server to use for vision, audio, search, and fetch operations
type Uses struct {
@ -28,16 +30,7 @@ const (
// ModelCapabilities defines the capabilities of a language model
// Used by LLM to select appropriate provider and validate requests
type ModelCapabilities struct {
Vision interface{} `json:"vision,omitempty"` // Supports vision/image input: bool or VisionFormat string ("openai", "claude"/"base64", "default")
ToolCalls *bool `json:"tool_calls,omitempty"` // Supports tool/function calling
Audio *bool `json:"audio,omitempty"` // Supports audio input/output
Reasoning *bool `json:"reasoning,omitempty"` // Supports reasoning/thinking mode (o1, DeepSeek R1)
Streaming *bool `json:"streaming,omitempty"` // Supports streaming responses
JSON *bool `json:"json,omitempty"` // Supports JSON mode
Multimodal *bool `json:"multimodal,omitempty"` // Supports multimodal input (text + images + audio)
TemperatureAdjustable *bool `json:"temperature_adjustable,omitempty"` // Supports temperature adjustment (reasoning models typically don't)
}
type ModelCapabilities message.ModelCapabilities
// GetVisionSupport returns whether vision is supported and the format
func (m *ModelCapabilities) GetVisionSupport() (bool, VisionFormat) {
@ -140,7 +133,7 @@ type CompletionResponse struct {
FinishReason string `json:"finish_reason"` // Why generation stopped (stop, length, tool_calls, content_filter, etc.)
// Usage statistics
Usage *UsageInfo `json:"usage,omitempty"` // Token usage statistics
Usage *message.UsageInfo `json:"usage,omitempty"` // Token usage statistics
// Additional metadata
SystemFingerprint string `json:"system_fingerprint,omitempty"` // System fingerprint for reproducibility
@ -150,32 +143,6 @@ type CompletionResponse struct {
Raw interface{} `json:"raw,omitempty"` // Original raw response from the LLM provider
}
// UsageInfo represents token usage statistics
// Structure matches OpenAI API: https://platform.openai.com/docs/api-reference/chat/object#chat-object-usage
type UsageInfo struct {
PromptTokens int `json:"prompt_tokens"` // Number of tokens in the prompt
CompletionTokens int `json:"completion_tokens"` // Number of tokens in the generated completion
TotalTokens int `json:"total_tokens"` // Total number of tokens used (prompt + completion)
// Detailed token breakdown
PromptTokensDetails *PromptTokensDetails `json:"prompt_tokens_details,omitempty"` // Breakdown of tokens used in the prompt
CompletionTokensDetails *CompletionTokensDetails `json:"completion_tokens_details,omitempty"` // Breakdown of tokens used in the completion
}
// PromptTokensDetails provides detailed breakdown of tokens used in the prompt
type PromptTokensDetails struct {
AudioTokens int `json:"audio_tokens,omitempty"` // Audio input tokens present in the prompt
CachedTokens int `json:"cached_tokens,omitempty"` // Cached tokens present in the prompt
}
// CompletionTokensDetails provides detailed breakdown of tokens used in the completion
type CompletionTokensDetails struct {
AcceptedPredictionTokens int `json:"accepted_prediction_tokens,omitempty"` // Tokens from predictions that appeared in the completion
AudioTokens int `json:"audio_tokens,omitempty"` // Audio input tokens generated by the model
ReasoningTokens int `json:"reasoning_tokens,omitempty"` // Tokens generated by the model for reasoning (o1, o1-mini, DeepSeek R1)
RejectedPredictionTokens int `json:"rejected_prediction_tokens,omitempty"` // Tokens from predictions that did not appear in the completion
}
// FinishReason constants - why the model stopped generating tokens
const (
FinishReasonStop = "stop" // Natural stop point or provided stop sequence reached
@ -210,68 +177,3 @@ type JSONSchema struct {
Schema interface{} `json:"schema"` // Required: JSON schema (*jsonschema.Schema or map[string]interface{})
Strict *bool `json:"strict,omitempty"` // Optional: whether to enforce strict schema validation (default: true)
}
// ============================================================================
// Stream Lifecycle Event Data Structures
// ============================================================================
// These structures define the data format for stream lifecycle events.
// They provide a standardized way to communicate stream boundaries and metadata
// to the frontend, enabling better UI/UX (progress indicators, timing, etc.).
// StreamStartData represents the data for stream_start event
// Sent when a streaming request begins
type StreamStartData struct {
ContextID string `json:"context_id"` // Context ID for the response
RequestID string `json:"request_id"` // Unique identifier for this request
Timestamp int64 `json:"timestamp"` // Unix timestamp when stream started
ChatID string `json:"chat_id"` // Chat ID being used (e.g., "chat-123")
TraceID string `json:"trace_id"` // Trace ID being used (e.g., "trace-123")
Assistant *AssistantInfo `json:"assistant,omitempty"` // Assistant information
Metadata map[string]interface{} `json:"metadata,omitempty"` // Metadata to pass to the page for CUI context
}
// StreamEndData represents the data for stream_end event
// Sent when a streaming request completes (successfully or with error)
type StreamEndData struct {
RequestID string `json:"request_id"` // Corresponding request ID
ContextID string `json:"context_id"` // Context ID for the response
TraceID string `json:"trace_id"` // Trace ID being used (e.g., "trace-123")
Timestamp int64 `json:"timestamp"` // Unix timestamp when stream ended
DurationMs int64 `json:"duration_ms"` // Total duration in milliseconds
Status string `json:"status"` // "completed" | "error" | "cancelled"
Error string `json:"error,omitempty"` // Error message if status is "error"
Usage *UsageInfo `json:"usage,omitempty"` // Token usage statistics
Metadata map[string]interface{} `json:"metadata,omitempty"` // Metadata to pass to the page for CUI context
}
// GroupStartData represents the data for group_start event
// Sent when a logical message group begins (text, tool_call, thinking, etc.)
type GroupStartData struct {
GroupID string `json:"group_id"` // Unique identifier for this group
Type string `json:"type"` // Group type: "text" | "thinking" | "tool_call" | "refusal"
Timestamp int64 `json:"timestamp"` // Unix timestamp when group started
ToolCall *GroupToolCallInfo `json:"tool_call,omitempty"` // Tool call metadata (if type is "tool_call")
Extra map[string]interface{} `json:"extra,omitempty"` // Additional metadata (for custom providers or future extensions)
}
// GroupEndData represents the data for group_end event
// Sent when a logical message group completes
type GroupEndData struct {
GroupID string `json:"group_id"` // Corresponding group ID
Type string `json:"type"` // Group type (same as in group_start)
Timestamp int64 `json:"timestamp"` // Unix timestamp when group ended
DurationMs int64 `json:"duration_ms"` // Duration of this group in milliseconds
ChunkCount int `json:"chunk_count"` // Number of data chunks in this group
Status string `json:"status"` // "completed" | "partial" | "error"
ToolCall *GroupToolCallInfo `json:"tool_call,omitempty"` // Complete tool call info (if type is "tool_call")
Extra map[string]interface{} `json:"extra,omitempty"` // Additional metadata
}
// GroupToolCallInfo contains tool call information for group events
// Used in both group_start (partial info) and group_end (complete info)
type GroupToolCallInfo struct {
ID string `json:"id"` // Tool call ID (e.g., "call_abc123")
Name string `json:"name"` // Function name (may be partial in group_start)
Arguments string `json:"arguments,omitempty"` // Complete arguments (only in group_end)
Index int `json:"index"` // Index in the tool calls array
}

7
agent/llm/interfaces.go
View file

@ -1,9 +1,12 @@
package llm
import "github.com/yaoapp/yao/agent/context"
import (
"github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/output/message"
)
// LLM the LLM interface
type LLM interface {
Stream(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler context.StreamFunc) (*context.CompletionResponse, error)
Stream(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, handler message.StreamFunc) (*context.CompletionResponse, error)
Post(ctx *context.Context, messages []context.Message, options *context.CompletionOptions) (*context.CompletionResponse, error)
}

3
agent/llm/providers/factory.go
View file

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

9
agent/llm/providers/openai/claude_test.go
View file

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

15
agent/llm/providers/openai/deepseek_r1_test.go
View file

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

5
agent/llm/providers/openai/deepseek_v3_test.go
View file

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

3
agent/llm/providers/openai/gpt5_test.go
View file

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

61
agent/llm/providers/openai/openai.go
View file

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

43
agent/llm/providers/openai/openai_test.go
View file

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

13
agent/llm/providers/openai/types.go
View file

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

12
agent/llm/stream.go
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)
}

24
agent/output/adapters/cui/factory.go
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
}

64
agent/output/adapters/cui/writer.go
View file

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

5
agent/output/adapters/openai/converter.go
View file

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

28
agent/output/adapters/openai/factory.go
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
}

85
agent/output/adapters/openai/writer.go
View file

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

695
agent/output/jsapi/output.go
View file

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

654
agent/output/jsapi/output_test.go
View file

@ -1,361 +1,349 @@
package jsapi
import (
"context"
"net/http"
"testing"
// func TestOutputConstructor(t *testing.T) {
// test.Prepare(t, config.Conf)
// defer test.Clean()
"github.com/stretchr/testify/assert"
v8 "github.com/yaoapp/gou/runtime/v8"
agentContext "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/config"
"github.com/yaoapp/yao/test"
)
// tests := []struct {
// name string
// script string
// expectError bool
// }{
// {
// name: "Create Output with context",
// script: `
// function test(ctx) {
// const output = new Output(ctx);
// return output !== undefined && output !== null;
// }
// `,
// expectError: false,
// },
// {
// name: "Create Output without context should fail",
// script: `
// function test(ctx) {
// try {
// const output = new Output();
// return false;
// } catch (e) {
// return e.toString().includes("context argument");
// }
// }
// `,
// expectError: false,
// },
// }
func TestOutputConstructor(t *testing.T) {
test.Prepare(t, config.Conf)
defer test.Clean()
// for _, tt := range tests {
// t.Run(tt.name, func(t *testing.T) {
// ctx := agentContext.New(context.Background(), nil, "test-chat-123", "")
// ctx.AssistantID = "test-assistant-456"
tests := []struct {
name string
script string
expectError bool
}{
{
name: "Create Output with context",
script: `
function test(ctx) {
const output = new Output(ctx);
return output !== undefined && output !== null;
}
`,
expectError: false,
},
{
name: "Create Output without context should fail",
script: `
function test(ctx) {
try {
const output = new Output();
return false;
} catch (e) {
return e.toString().includes("context argument");
}
}
`,
expectError: false,
},
}
// // Execute test script with v8.Call
// res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx)
// if tt.expectError {
// assert.Error(t, err)
// return
// }
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ctx := agentContext.New(context.Background(), nil, "test-chat-123", "")
ctx.AssistantID = "test-assistant-456"
// assert.NoError(t, err)
// assert.True(t, res.(bool))
// })
// }
// }
// Execute test script with v8.Call
res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx)
if tt.expectError {
assert.Error(t, err)
return
}
// func TestOutputSend(t *testing.T) {
// test.Prepare(t, config.Conf)
// defer test.Clean()
assert.NoError(t, err)
assert.True(t, res.(bool))
})
}
}
// tests := []struct {
// name string
// script string
// expectError bool
// validate func(*testing.T, *agentContext.Context)
// }{
// {
// name: "Send text message with object",
// script: `
// function test(ctx) {
// const output = new Output(ctx);
// output.Send({
// type: "text",
// props: { content: "Hello World" }
// });
// return true;
// }
// `,
// expectError: false,
// },
// {
// name: "Send text message with string shorthand",
// script: `
// function test(ctx) {
// const output = new Output(ctx);
// output.Send("Hello World");
// return true;
// }
// `,
// expectError: false,
// },
// {
// name: "Send message with all fields",
// script: `
// function test(ctx) {
// const output = new Output(ctx);
// output.Send({
// type: "text",
// props: { content: "Test" },
// id: "msg-1",
// delta: true,
// done: false,
// delta_path: "content",
// delta_action: "append",
// metadata: {
// timestamp: 1234567890,
// sequence: 1,
// trace_id: "trace-123"
// }
// });
// return true;
// }
// `,
// expectError: false,
// },
// {
// name: "Send error message",
// script: `
// function test(ctx) {
// const output = new Output(ctx);
// output.Send({
// type: "error",
// props: {
// message: "Something went wrong",
// code: "ERR_001"
// }
// });
// return true;
// }
// `,
// expectError: false,
// },
// {
// name: "Send without message should fail",
// script: `
// function test(ctx) {
// try {
// const output = new Output(ctx);
// output.Send();
// return false;
// } catch (e) {
// return e.toString().includes("message argument");
// }
// }
// `,
// expectError: false,
// },
// {
// name: "Send message without type should fail",
// script: `
// function test(ctx) {
// try {
// const output = new Output(ctx);
// output.Send({ props: { content: "test" } });
// return false;
// } catch (e) {
// return e.toString().includes("type is required");
// }
// }
// `,
// expectError: false,
// },
// }
func TestOutputSend(t *testing.T) {
test.Prepare(t, config.Conf)
defer test.Clean()
// for _, tt := range tests {
// t.Run(tt.name, func(t *testing.T) {
// // Create context with mock writer
// ctx := agentContext.New(context.Background(), nil, "test-chat", "")
// ctx.Writer = &mockWriter{}
tests := []struct {
name string
script string
expectError bool
validate func(*testing.T, *agentContext.Context)
}{
{
name: "Send text message with object",
script: `
function test(ctx) {
const output = new Output(ctx);
output.Send({
type: "text",
props: { content: "Hello World" }
});
return true;
}
`,
expectError: false,
},
{
name: "Send text message with string shorthand",
script: `
function test(ctx) {
const output = new Output(ctx);
output.Send("Hello World");
return true;
}
`,
expectError: false,
},
{
name: "Send message with all fields",
script: `
function test(ctx) {
const output = new Output(ctx);
output.Send({
type: "text",
props: { content: "Test" },
id: "msg-1",
delta: true,
done: false,
delta_path: "content",
delta_action: "append",
metadata: {
timestamp: 1234567890,
sequence: 1,
trace_id: "trace-123"
}
});
return true;
}
`,
expectError: false,
},
{
name: "Send error message",
script: `
function test(ctx) {
const output = new Output(ctx);
output.Send({
type: "error",
props: {
message: "Something went wrong",
code: "ERR_001"
}
});
return true;
}
`,
expectError: false,
},
{
name: "Send without message should fail",
script: `
function test(ctx) {
try {
const output = new Output(ctx);
output.Send();
return false;
} catch (e) {
return e.toString().includes("message argument");
}
}
`,
expectError: false,
},
{
name: "Send message without type should fail",
script: `
function test(ctx) {
try {
const output = new Output(ctx);
output.Send({ props: { content: "test" } });
return false;
} catch (e) {
return e.toString().includes("type is required");
}
}
`,
expectError: false,
},
}
// // Execute test script with v8.Call
// res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx)
// if tt.expectError {
// assert.Error(t, err)
// return
// }
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Create context with mock writer
ctx := agentContext.New(context.Background(), nil, "test-chat", "")
ctx.Writer = &mockWriter{}
// assert.NoError(t, err)
// assert.True(t, res.(bool))
// Execute test script with v8.Call
res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx)
if tt.expectError {
assert.Error(t, err)
return
}
// if tt.validate != nil {
// tt.validate(t, &ctx)
// }
// })
// }
// }
assert.NoError(t, err)
assert.True(t, res.(bool))
// func TestOutputSendGroup(t *testing.T) {
// test.Prepare(t, config.Conf)
// defer test.Clean()
if tt.validate != nil {
tt.validate(t, &ctx)
}
})
}
}
// tests := []struct {
// name string
// script string
// expectError bool
// }{
// {
// name: "Send message group",
// script: `
// function test(ctx) {
// const output = new Output(ctx);
// output.SendGroup({
// id: "group-1",
// messages: [
// { type: "text", props: { content: "Message 1" } },
// { type: "text", props: { content: "Message 2" } }
// ]
// });
// return true;
// }
// `,
// expectError: false,
// },
// {
// name: "Send group with metadata",
// script: `
// function test(ctx) {
// const output = new Output(ctx);
// output.SendGroup({
// id: "group-1",
// messages: [
// { type: "text", props: { content: "Test" } }
// ],
// metadata: {
// timestamp: 1234567890,
// sequence: 1
// }
// });
// return true;
// }
// `,
// expectError: false,
// },
// {
// name: "Send empty group",
// script: `
// function test(ctx) {
// const output = new Output(ctx);
// output.SendGroup({
// id: "group-1",
// messages: []
// });
// return true;
// }
// `,
// expectError: false,
// },
// {
// name: "Send group without id should fail",
// script: `
// function test(ctx) {
// try {
// const output = new Output(ctx);
// output.SendGroup({
// messages: [
// { type: "text", props: { content: "Test" } }
// ]
// });
// return false;
// } catch (e) {
// return e.toString().includes("id is required");
// }
// }
// `,
// expectError: false,
// },
// {
// name: "Send group without messages should fail",
// script: `
// function test(ctx) {
// try {
// const output = new Output(ctx);
// output.SendGroup({ id: "group-1" });
// return false;
// } catch (e) {
// return e.toString().includes("messages is required");
// }
// }
// `,
// expectError: false,
// },
// }
func TestOutputSendGroup(t *testing.T) {
test.Prepare(t, config.Conf)
defer test.Clean()
// for _, tt := range tests {
// t.Run(tt.name, func(t *testing.T) {
// // Create context with mock writer
// ctx := agentContext.New(context.Background(), nil, "test-chat", "")
// ctx.Writer = &mockWriter{}
tests := []struct {
name string
script string
expectError bool
}{
{
name: "Send message group",
script: `
function test(ctx) {
const output = new Output(ctx);
output.SendGroup({
id: "group-1",
messages: [
{ type: "text", props: { content: "Message 1" } },
{ type: "text", props: { content: "Message 2" } }
]
});
return true;
}
`,
expectError: false,
},
{
name: "Send group with metadata",
script: `
function test(ctx) {
const output = new Output(ctx);
output.SendGroup({
id: "group-1",
messages: [
{ type: "text", props: { content: "Test" } }
],
metadata: {
timestamp: 1234567890,
sequence: 1
}
});
return true;
}
`,
expectError: false,
},
{
name: "Send empty group",
script: `
function test(ctx) {
const output = new Output(ctx);
output.SendGroup({
id: "group-1",
messages: []
});
return true;
}
`,
expectError: false,
},
{
name: "Send group without id should fail",
script: `
function test(ctx) {
try {
const output = new Output(ctx);
output.SendGroup({
messages: [
{ type: "text", props: { content: "Test" } }
]
});
return false;
} catch (e) {
return e.toString().includes("id is required");
}
}
`,
expectError: false,
},
{
name: "Send group without messages should fail",
script: `
function test(ctx) {
try {
const output = new Output(ctx);
output.SendGroup({ id: "group-1" });
return false;
} catch (e) {
return e.toString().includes("messages is required");
}
}
`,
expectError: false,
},
}
// // Execute test script with v8.Call
// res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx)
// if tt.expectError {
// assert.Error(t, err)
// return
// }
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Create context with mock writer
ctx := agentContext.New(context.Background(), nil, "test-chat", "")
ctx.Writer = &mockWriter{}
// assert.NoError(t, err)
// assert.True(t, res.(bool))
// })
// }
// }
// Execute test script with v8.Call
res, err := v8.Call(v8.CallOptions{}, tt.script, &ctx)
if tt.expectError {
assert.Error(t, err)
return
}
// func TestOutputChaining(t *testing.T) {
// test.Prepare(t, config.Conf)
// defer test.Clean()
assert.NoError(t, err)
assert.True(t, res.(bool))
})
}
}
// script := `
// function test(ctx) {
// const output = new Output(ctx);
func TestOutputChaining(t *testing.T) {
test.Prepare(t, config.Conf)
defer test.Clean()
// // Send should return the output object for chaining
// const result = output.Send("Message 1");
script := `
function test(ctx) {
const output = new Output(ctx);
// Send should return the output object for chaining
const result = output.Send("Message 1");
// Should be able to chain sends
output.Send("Message 2").Send("Message 3");
return result !== undefined;
}
`
// // Should be able to chain sends
// output.Send("Message 2").Send("Message 3");
ctx := agentContext.New(context.Background(), nil, "test-chat", "")
ctx.Writer = &mockWriter{}
// return result !== undefined;
// }
// `
// Execute test script with v8.Call
res, err := v8.Call(v8.CallOptions{}, script, &ctx)
assert.NoError(t, err)
assert.True(t, res.(bool))
}
// ctx := agentContext.New(context.Background(), nil, "test-chat", "")
// ctx.Writer = &mockWriter{}
// mockWriter is a mock implementation of http.ResponseWriter for testing
type mockWriter struct {
data [][]byte
header http.Header
}
// // Execute test script with v8.Call
// res, err := v8.Call(v8.CallOptions{}, script, &ctx)
// assert.NoError(t, err)
// assert.True(t, res.(bool))
// }
func (w *mockWriter) Header() http.Header {
if w.header == nil {
w.header = make(http.Header)
}
return w.header
}
// // mockWriter is a mock implementation of http.ResponseWriter for testing
// type mockWriter struct {
// data [][]byte
// header http.Header
// }
func (w *mockWriter) Write(p []byte) (n int, err error) {
w.data = append(w.data, p)
return len(p), nil
}
// func (w *mockWriter) Header() http.Header {
// if w.header == nil {
// w.header = make(http.Header)
// }
// return w.header
// }
func (w *mockWriter) WriteHeader(statusCode int) {}
// func (w *mockWriter) Write(p []byte) (n int, err error) {
// w.data = append(w.data, p)
// return len(p), nil
// }
func (w *mockWriter) Flush() {}
// func (w *mockWriter) WriteHeader(statusCode int) {}
// func (w *mockWriter) Flush() {}

16
agent/output/message/interfaces.go
View file

@ -1,7 +1,5 @@
package message
import "github.com/yaoapp/yao/agent/context"
// Writer is the interface for writing output messages
// Different writers handle different output formats (SSE, WebSocket, Standard, etc.)
type Writer interface {
@ -29,23 +27,11 @@ type Adapter interface {
SupportsType(msgType string) bool
}
// WriterFactory creates writers based on context
type WriterFactory interface {
// NewWriter creates a writer for the given context
NewWriter(ctx *context.Context, adapter Adapter) (Writer, error)
}
// AdapterFactory creates adapters based on context
type AdapterFactory interface {
// NewAdapter creates an adapter for the given context
NewAdapter(ctx *context.Context) (Adapter, error)
}
// StreamHandler handles streaming message processing
// It bridges between LLM streaming chunks and output messages
type StreamHandler interface {
// Handle processes a streaming chunk from LLM
Handle(chunkType context.StreamChunkType, data []byte) error
Handle(chunkType StreamChunkType, data []byte) error
// Flush flushes any pending messages
Flush() error

159
agent/output/message/types.go
View file

@ -1,5 +1,34 @@
package message
import (
"net/http"
traceTypes "github.com/yaoapp/yao/trace/types"
)
// Options are the options for the writer
type Options struct {
BaseURL string
Accept string
Writer http.ResponseWriter
Trace traceTypes.Manager
Capabilities *ModelCapabilities
Locale string
}
// ModelCapabilities defines the capabilities of a language model
// Used by LLM to select appropriate provider and validate requests
type ModelCapabilities struct {
Vision interface{} `json:"vision,omitempty"` // Supports vision/image input: bool or VisionFormat string ("openai", "claude"/"base64", "default")
ToolCalls *bool `json:"tool_calls,omitempty"` // Supports tool/function calling
Audio *bool `json:"audio,omitempty"` // Supports audio input/output
Reasoning *bool `json:"reasoning,omitempty"` // Supports reasoning/thinking mode (o1, DeepSeek R1)
Streaming *bool `json:"streaming,omitempty"` // Supports streaming responses
JSON *bool `json:"json,omitempty"` // Supports JSON mode
Multimodal *bool `json:"multimodal,omitempty"` // Supports multimodal input (text + images + audio)
TemperatureAdjustable *bool `json:"temperature_adjustable,omitempty"` // Supports temperature adjustment (reasoning models typically don't)
}
// Message represents a universal message structure (DSL)
// All messages are expressed through Type + Props, without predefining specific types
type Message struct {
@ -185,3 +214,133 @@ const (
DeltaMerge = "merge" // Merge (for objects)
DeltaSet = "set" // Set (for new fields)
)
// StreamChunkType represents the type of content in a streaming chunk
type StreamChunkType string
// Stream chunk type constants - indicates what type of content is in the current chunk
const (
// Content chunk types - actual data from the LLM
ChunkText StreamChunkType = "text" // Regular text content
ChunkThinking StreamChunkType = "thinking" // Reasoning/thinking content (o1, DeepSeek R1)
ChunkToolCall StreamChunkType = "tool_call" // Tool/function call
ChunkRefusal StreamChunkType = "refusal" // Model refusal
ChunkMetadata StreamChunkType = "metadata" // Metadata (usage, finish_reason, etc.)
ChunkError StreamChunkType = "error" // Error chunk
ChunkUnknown StreamChunkType = "unknown" // Unknown/unrecognized chunk type
// Lifecycle event types - stream and group boundaries
ChunkStreamStart StreamChunkType = "stream_start" // Stream begins (entire request starts)
ChunkStreamEnd StreamChunkType = "stream_end" // Stream ends (entire request completes)
ChunkGroupStart StreamChunkType = "group_start" // Message group begins (text/tool_call/thinking group starts)
ChunkGroupEnd StreamChunkType = "group_end" // Message group ends (text/tool_call/thinking group completes)
)
// StreamFunc the streaming function callback
// Parameters:
// - chunkType: the type of content in this chunk (text, thinking, tool_call, etc.)
// - data: the actual chunk data (could be text, JSON, or other format)
//
// Returns:
// - int: status code (0 = continue, non-zero = stop streaming)
type StreamFunc func(chunkType StreamChunkType, data []byte) int
// AssistantInfo represents the assistant information structure
type AssistantInfo struct {
ID string `json:"assistant_id"` // Assistant ID
Type string `json:"type,omitempty"` // Assistant Type, default is assistant
Name string `json:"name,omitempty"` // Assistant Name
Avatar string `json:"avatar,omitempty"` // Assistant Avatar
Description string `json:"description,omitempty"` // Assistant Description
}
// UsageInfo represents token usage statistics
// Structure matches OpenAI API: https://platform.openai.com/docs/api-reference/chat/object#chat-object-usage
type UsageInfo struct {
PromptTokens int `json:"prompt_tokens"` // Number of tokens in the prompt
CompletionTokens int `json:"completion_tokens"` // Number of tokens in the generated completion
TotalTokens int `json:"total_tokens"` // Total number of tokens used (prompt + completion)
// Detailed token breakdown
PromptTokensDetails *PromptTokensDetails `json:"prompt_tokens_details,omitempty"` // Breakdown of tokens used in the prompt
CompletionTokensDetails *CompletionTokensDetails `json:"completion_tokens_details,omitempty"` // Breakdown of tokens used in the completion
}
// PromptTokensDetails provides detailed breakdown of tokens used in the prompt
type PromptTokensDetails struct {
AudioTokens int `json:"audio_tokens,omitempty"` // Audio input tokens present in the prompt
CachedTokens int `json:"cached_tokens,omitempty"` // Cached tokens present in the prompt
}
// CompletionTokensDetails provides detailed breakdown of tokens used in the completion
type CompletionTokensDetails struct {
AcceptedPredictionTokens int `json:"accepted_prediction_tokens,omitempty"` // Tokens from predictions that appeared in the completion
AudioTokens int `json:"audio_tokens,omitempty"` // Audio input tokens generated by the model
ReasoningTokens int `json:"reasoning_tokens,omitempty"` // Tokens generated by the model for reasoning (o1, o1-mini, DeepSeek R1)
RejectedPredictionTokens int `json:"rejected_prediction_tokens,omitempty"` // Tokens from predictions that did not appear in the completion
}
// ============================================================================
// Stream Lifecycle Event Data Structures
// ============================================================================
// These structures define the data format for stream lifecycle events.
// They provide a standardized way to communicate stream boundaries and metadata
// to the frontend, enabling better UI/UX (progress indicators, timing, etc.).
// StreamStartData represents the data for stream_start event
// Sent when a streaming request begins
type StreamStartData struct {
ContextID string `json:"context_id"` // Context ID for the response
RequestID string `json:"request_id"` // Unique identifier for this request
Timestamp int64 `json:"timestamp"` // Unix timestamp when stream started
ChatID string `json:"chat_id"` // Chat ID being used (e.g., "chat-123")
TraceID string `json:"trace_id"` // Trace ID being used (e.g., "trace-123")
Assistant *AssistantInfo `json:"assistant,omitempty"` // Assistant information
Metadata map[string]interface{} `json:"metadata,omitempty"` // Metadata to pass to the page for CUI context
}
// StreamEndData represents the data for stream_end event
// Sent when a streaming request completes (successfully or with error)
type StreamEndData struct {
RequestID string `json:"request_id"` // Corresponding request ID
ContextID string `json:"context_id"` // Context ID for the response
TraceID string `json:"trace_id"` // Trace ID being used (e.g., "trace-123")
Timestamp int64 `json:"timestamp"` // Unix timestamp when stream ended
DurationMs int64 `json:"duration_ms"` // Total duration in milliseconds
Status string `json:"status"` // "completed" | "error" | "cancelled"
Error string `json:"error,omitempty"` // Error message if status is "error"
Usage *UsageInfo `json:"usage,omitempty"` // Token usage statistics
Metadata map[string]interface{} `json:"metadata,omitempty"` // Metadata to pass to the page for CUI context
}
// GroupStartData represents the data for group_start event
// Sent when a logical message group begins (text, tool_call, thinking, etc.)
type GroupStartData struct {
GroupID string `json:"group_id"` // Unique identifier for this group
Type string `json:"type"` // Group type: "text" | "thinking" | "tool_call" | "refusal"
Timestamp int64 `json:"timestamp"` // Unix timestamp when group started
ToolCall *GroupToolCallInfo `json:"tool_call,omitempty"` // Tool call metadata (if type is "tool_call")
Extra map[string]interface{} `json:"extra,omitempty"` // Additional metadata (for custom providers or future extensions)
}
// GroupEndData represents the data for group_end event
// Sent when a logical message group completes
type GroupEndData struct {
GroupID string `json:"group_id"` // Corresponding group ID
Type string `json:"type"` // Group type (same as in group_start)
Timestamp int64 `json:"timestamp"` // Unix timestamp when group ended
DurationMs int64 `json:"duration_ms"` // Duration of this group in milliseconds
ChunkCount int `json:"chunk_count"` // Number of data chunks in this group
Status string `json:"status"` // "completed" | "partial" | "error"
ToolCall *GroupToolCallInfo `json:"tool_call,omitempty"` // Complete tool call info (if type is "tool_call")
Extra map[string]interface{} `json:"extra,omitempty"` // Additional metadata
}
// GroupToolCallInfo contains tool call information for group events
// Used in both group_start (partial info) and group_end (complete info)
type GroupToolCallInfo struct {
ID string `json:"id"` // Tool call ID (e.g., "call_abc123")
Name string `json:"name"` // Function name (may be partial in group_start)
Arguments string `json:"arguments,omitempty"` // Complete arguments (only in group_end)
Index int `json:"index"` // Index in the tool calls array
}

284
agent/output/output.go
View file

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

1
main.go
View file

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