Enhance streaming and grouping functionality in Assistant
- Introduced SendGroupStart and SendGroupEnd methods for better control over message grouping, allowing for automatic ID generation and event handling. - Updated streamState structure to track group IDs, message sequences, and chunk counts, improving message organization during streaming. - Refactored message handling to eliminate the 'done' field, signaling message completion through group_end events instead. - Revised JSAPI documentation to reflect new methods and usage patterns, enhancing clarity for developers. - Improved test cases to validate new group handling features and ensure proper functionality.
This commit is contained in:
Max
parent
afab9c2173
commit
f0495ab990
8 changed files with 433 additions and 164 deletions
|
|
@ -1,6 +1,9 @@
|
|||
package handlers
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
"github.com/yaoapp/kun/log"
|
||||
"github.com/yaoapp/yao/agent/context"
|
||||
|
|
@ -15,9 +18,10 @@ func DefaultStreamHandler(ctx *context.Context) message.StreamFunc {
|
|||
|
||||
// Create stream state manager
|
||||
state := &streamState{
|
||||
ctx: ctx,
|
||||
inGroup: false,
|
||||
currentID: "",
|
||||
ctx: ctx,
|
||||
inGroup: false,
|
||||
currentGroupID: "",
|
||||
messageSeq: 0,
|
||||
}
|
||||
|
||||
return func(chunkType message.StreamChunkType, data []byte) int {
|
||||
|
|
@ -64,11 +68,14 @@ func DefaultStreamHandler(ctx *context.Context) message.StreamFunc {
|
|||
|
||||
// streamState manages the state of the streaming process
|
||||
type streamState struct {
|
||||
ctx *context.Context
|
||||
inGroup bool
|
||||
currentID string
|
||||
currentType string // Track the current message type (text, thinking, tool_call)
|
||||
buffer []byte
|
||||
ctx *context.Context
|
||||
inGroup bool
|
||||
currentGroupID string // Current group ID (shared by all chunks in the group)
|
||||
currentType string // Track the current message type (text, thinking, tool_call)
|
||||
buffer []byte
|
||||
chunkCount int // Track number of chunks in current group
|
||||
messageSeq int // Message sequence number (for generating readable IDs)
|
||||
groupStartTime time.Time // Track when group started
|
||||
}
|
||||
|
||||
// handleStreamStart handles stream start event
|
||||
|
|
@ -87,9 +94,32 @@ func (s *streamState) handleStreamStart(data []byte) int {
|
|||
|
||||
// handleGroupStart handles group start event
|
||||
func (s *streamState) handleGroupStart(data []byte) int {
|
||||
// Parse group start data first to get the group ID
|
||||
var startData message.GroupStartData
|
||||
if err := jsoniter.Unmarshal(data, &startData); err != nil {
|
||||
log.Error("Failed to unmarshal group start data: %v", err)
|
||||
return 0
|
||||
}
|
||||
|
||||
// Use the group ID from the start data, or generate one if not provided
|
||||
groupID := startData.GroupID
|
||||
if groupID == "" {
|
||||
groupID = generateMessageID()
|
||||
startData.GroupID = groupID
|
||||
}
|
||||
|
||||
// Initialize group state with the correct group ID
|
||||
s.inGroup = true
|
||||
s.currentID = generateMessageID()
|
||||
s.currentGroupID = groupID
|
||||
s.buffer = []byte{}
|
||||
s.chunkCount = 0
|
||||
s.messageSeq = 0 // Reset message sequence for each group
|
||||
s.groupStartTime = time.Now()
|
||||
|
||||
// Send group_start event
|
||||
msg := output.NewEventMessage(message.EventGroupStart, "Group started", startData)
|
||||
s.ctx.Send(msg)
|
||||
|
||||
return 0 // Continue
|
||||
}
|
||||
|
||||
|
|
@ -99,22 +129,22 @@ func (s *streamState) handleText(data []byte) int {
|
|||
return 0
|
||||
}
|
||||
|
||||
// Ensure we have a message ID
|
||||
if s.currentID == "" {
|
||||
s.currentID = generateMessageID()
|
||||
}
|
||||
|
||||
// Track current message type
|
||||
s.currentType = message.TypeText
|
||||
|
||||
// Append to buffer
|
||||
s.buffer = append(s.buffer, data...)
|
||||
s.chunkCount++
|
||||
s.messageSeq++
|
||||
|
||||
// Send delta message
|
||||
// - ID: Sequential message ID (e.g., msg_001, msg_002) for readability
|
||||
// - GroupID: Same for all chunks of this logical message (frontend merges by group_id)
|
||||
msg := &message.Message{
|
||||
ID: s.currentID,
|
||||
Type: message.TypeText,
|
||||
Delta: true,
|
||||
ID: s.generateSequentialID(), // Sequential ID for this chunk
|
||||
GroupID: s.currentGroupID, // Group ID for merging (all chunks share this)
|
||||
Type: message.TypeText,
|
||||
Delta: true,
|
||||
Props: map[string]interface{}{
|
||||
"content": string(data),
|
||||
},
|
||||
|
|
@ -134,22 +164,22 @@ func (s *streamState) handleThinking(data []byte) int {
|
|||
return 0
|
||||
}
|
||||
|
||||
// Ensure we have a message ID
|
||||
if s.currentID == "" {
|
||||
s.currentID = generateMessageID()
|
||||
}
|
||||
|
||||
// Track current message type
|
||||
s.currentType = message.TypeThinking
|
||||
|
||||
// Append to buffer
|
||||
s.buffer = append(s.buffer, data...)
|
||||
s.chunkCount++
|
||||
s.messageSeq++
|
||||
|
||||
// Send delta message
|
||||
// - ID: Sequential message ID (e.g., msg_001, msg_002) for readability
|
||||
// - GroupID: Same for all chunks of this logical message (frontend merges by group_id)
|
||||
msg := &message.Message{
|
||||
ID: s.currentID,
|
||||
Type: message.TypeThinking,
|
||||
Delta: true,
|
||||
ID: s.generateSequentialID(), // Sequential ID for this chunk
|
||||
GroupID: s.currentGroupID, // Group ID for merging (all chunks share this)
|
||||
Type: message.TypeThinking,
|
||||
Delta: true,
|
||||
Props: map[string]interface{}{
|
||||
"content": string(data),
|
||||
},
|
||||
|
|
@ -202,30 +232,38 @@ func (s *streamState) handleGroupEnd(data []byte) int {
|
|||
return 0
|
||||
}
|
||||
|
||||
// Send done message with complete content
|
||||
if s.currentID != "" && len(s.buffer) > 0 {
|
||||
// Use the tracked message type (thinking, text, tool_call, etc.)
|
||||
msgType := s.currentType
|
||||
if msgType == "" {
|
||||
msgType = message.TypeText // Fallback to text if type not set
|
||||
}
|
||||
// Calculate duration
|
||||
durationMs := time.Since(s.groupStartTime).Milliseconds()
|
||||
|
||||
msg := &message.Message{
|
||||
ID: s.currentID,
|
||||
Type: msgType, // Use the actual message type from the group
|
||||
Done: true,
|
||||
Props: map[string]interface{}{
|
||||
"content": string(s.buffer),
|
||||
},
|
||||
}
|
||||
s.ctx.Send(msg)
|
||||
// Use the tracked message type (thinking, text, tool_call, etc.)
|
||||
msgType := s.currentType
|
||||
if msgType == "" {
|
||||
msgType = message.TypeText // Fallback to text if type not set
|
||||
}
|
||||
|
||||
// Build GroupEndData with complete content
|
||||
endData := message.GroupEndData{
|
||||
GroupID: s.currentGroupID, // Use the group ID, not message ID
|
||||
Type: msgType,
|
||||
Timestamp: time.Now().UnixMilli(),
|
||||
DurationMs: durationMs,
|
||||
ChunkCount: s.chunkCount,
|
||||
Status: "completed",
|
||||
Extra: map[string]interface{}{
|
||||
"content": string(s.buffer), // Include complete content in the event
|
||||
},
|
||||
}
|
||||
|
||||
// Send group_end event
|
||||
msg := output.NewEventMessage(message.EventGroupEnd, "Group completed", endData)
|
||||
s.ctx.Send(msg)
|
||||
|
||||
// Reset state
|
||||
s.inGroup = false
|
||||
s.currentID = ""
|
||||
s.currentGroupID = ""
|
||||
s.currentType = ""
|
||||
s.buffer = []byte{}
|
||||
s.chunkCount = 0
|
||||
|
||||
return 0 // Continue
|
||||
}
|
||||
|
|
@ -249,6 +287,13 @@ func (s *streamState) handleStreamEnd(data []byte) int {
|
|||
return 0 // Continue (stream will end naturally)
|
||||
}
|
||||
|
||||
// generateSequentialID generates a sequential message ID for better readability
|
||||
func (s *streamState) generateSequentialID() string {
|
||||
// Format: 1, 2, 3, etc.
|
||||
// This makes it easier for developers to track message order in logs
|
||||
return fmt.Sprintf("%d", s.messageSeq)
|
||||
}
|
||||
|
||||
// generateMessageID generates a unique message ID
|
||||
func generateMessageID() string {
|
||||
// TODO: Implement proper ID generation
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
# Context Output JS API
|
||||
|
||||
The Context object provides `Send`, `SendGroup`, and `Flush` methods for sending messages to clients from JavaScript within Agent Hook functions.
|
||||
The Context object provides `Send`, `SendGroup`, `SendGroupStart`, and `SendGroupEnd` methods for sending messages to clients from JavaScript within Agent Hook functions.
|
||||
|
||||
## Hook Functions Overview
|
||||
|
||||
|
|
@ -23,16 +23,14 @@ Agent Hook functions are lifecycle callbacks that allow you to customize the beh
|
|||
* Create hook - send initial messages to client
|
||||
*/
|
||||
function Create(ctx, messages) {
|
||||
// Send welcome message (string shorthand)
|
||||
// Send welcome message (string shorthand, auto-flushes)
|
||||
ctx.Send("Welcome! Let me help you with that...");
|
||||
ctx.Flush();
|
||||
|
||||
// Send loading indicator
|
||||
// Send loading indicator (auto-flushes)
|
||||
ctx.Send({
|
||||
type: "loading",
|
||||
props: { message: "Analyzing your request..." },
|
||||
});
|
||||
ctx.Flush();
|
||||
|
||||
// Continue with normal processing
|
||||
return { messages };
|
||||
|
|
@ -154,13 +152,43 @@ ctx.SendGroup({
|
|||
});
|
||||
```
|
||||
|
||||
### ctx.Flush()
|
||||
### ctx.SendGroupStart(type?, id?)
|
||||
|
||||
Flush the output buffer to ensure messages are sent immediately.
|
||||
Start a message group and return the group ID. Messages sent after this should include the returned `group_id`.
|
||||
|
||||
**Parameters:**
|
||||
|
||||
- `type` (optional): Group type (`"text"`, `"thinking"`, `"tool_call"`, `"mixed"`), defaults to `"mixed"`
|
||||
- `id` (optional): Custom group ID, auto-generates if not provided
|
||||
|
||||
**Returns:** Group ID (string)
|
||||
|
||||
```javascript
|
||||
ctx.Send("Processing...");
|
||||
ctx.Flush(); // Send immediately to client
|
||||
// Auto-generate ID with default type
|
||||
const groupId = ctx.SendGroupStart();
|
||||
|
||||
// Specify type, auto-generate ID
|
||||
const groupId = ctx.SendGroupStart("text");
|
||||
|
||||
// Specify both type and custom ID
|
||||
const groupId = ctx.SendGroupStart("thinking", "my-group-123");
|
||||
```
|
||||
|
||||
### ctx.SendGroupEnd(id, chunkCount?)
|
||||
|
||||
End a message group.
|
||||
|
||||
**Parameters:**
|
||||
|
||||
- `id` (required): Group ID returned from `SendGroupStart`
|
||||
- `chunkCount` (optional): Number of messages in the group
|
||||
|
||||
```javascript
|
||||
// Basic usage
|
||||
ctx.SendGroupEnd(groupId);
|
||||
|
||||
// With chunk count
|
||||
ctx.SendGroupEnd(groupId, 5);
|
||||
```
|
||||
|
||||
## Complete Hook Examples
|
||||
|
|
@ -172,9 +200,8 @@ ctx.Flush(); // Send immediately to client
|
|||
* Send welcome message when conversation starts
|
||||
*/
|
||||
function Create(ctx, messages) {
|
||||
// Send welcome message
|
||||
// Send welcome message (auto-flushes)
|
||||
ctx.Send("Welcome to AI Assistant! How can I help you today?");
|
||||
ctx.Flush();
|
||||
|
||||
// Return messages to continue processing
|
||||
return { messages };
|
||||
|
|
@ -409,16 +436,15 @@ function Create(ctx, messages) {
|
|||
}
|
||||
```
|
||||
|
||||
### 9. Message Groups for Related Content
|
||||
### 9. Message Groups for Related Content (High-level API)
|
||||
|
||||
```javascript
|
||||
/**
|
||||
* Send groups of related messages together
|
||||
* Send groups of related messages together using SendGroup (auto-handles events)
|
||||
*/
|
||||
function Before(ctx, messages, response) {
|
||||
// Send a group of context information
|
||||
// SendGroup automatically sends group_start and group_end events
|
||||
ctx.SendGroup({
|
||||
id: "context_info",
|
||||
messages: [
|
||||
{
|
||||
type: "text",
|
||||
|
|
@ -442,12 +468,86 @@ function Before(ctx, messages, response) {
|
|||
type: "context",
|
||||
},
|
||||
});
|
||||
ctx.Flush();
|
||||
|
||||
return { response };
|
||||
}
|
||||
```
|
||||
|
||||
### 10. Manual Group Control (Low-level API)
|
||||
|
||||
```javascript
|
||||
/**
|
||||
* Manually control group boundaries with SendGroupStart and SendGroupEnd
|
||||
*/
|
||||
function Create(ctx, messages) {
|
||||
// Start a text group
|
||||
const groupId = ctx.SendGroupStart("text");
|
||||
|
||||
// Send messages with group_id
|
||||
ctx.Send({
|
||||
type: "text",
|
||||
props: { content: "First message in group" },
|
||||
group_id: groupId,
|
||||
});
|
||||
|
||||
ctx.Send({
|
||||
type: "text",
|
||||
props: { content: "Second message in group" },
|
||||
group_id: groupId,
|
||||
});
|
||||
|
||||
// End the group
|
||||
ctx.SendGroupEnd(groupId, 2);
|
||||
|
||||
return { messages };
|
||||
}
|
||||
```
|
||||
|
||||
### 11. Streaming with Groups
|
||||
|
||||
```javascript
|
||||
/**
|
||||
* Stream delta updates within a group
|
||||
*/
|
||||
function Create(ctx, messages) {
|
||||
// Start thinking group
|
||||
const thinkingId = ctx.SendGroupStart("thinking");
|
||||
|
||||
// Stream thinking process
|
||||
const steps = ["Analyzing", "Processing", "Generating"];
|
||||
const msgId = "thinking_msg";
|
||||
|
||||
steps.forEach((step, i) => {
|
||||
if (i === 0) {
|
||||
// First message
|
||||
ctx.Send({
|
||||
type: "thinking",
|
||||
props: { content: step },
|
||||
id: msgId,
|
||||
group_id: thinkingId,
|
||||
delta: false,
|
||||
});
|
||||
} else {
|
||||
// Delta updates
|
||||
ctx.Send({
|
||||
type: "thinking",
|
||||
props: { content: ` → ${step}` },
|
||||
id: msgId,
|
||||
group_id: thinkingId,
|
||||
delta: true,
|
||||
delta_path: "content",
|
||||
delta_action: "append",
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
// End thinking group
|
||||
ctx.SendGroupEnd(thinkingId, steps.length);
|
||||
|
||||
return { messages };
|
||||
}
|
||||
```
|
||||
|
||||
## Message Types
|
||||
|
||||
Built-in message types supported:
|
||||
|
|
@ -617,9 +717,8 @@ ctx.Send({
|
|||
|
||||
```javascript
|
||||
function Create(ctx, messages) {
|
||||
ctx.Send("Starting processing...");
|
||||
ctx.Flush();
|
||||
// No need to wait, continue processing
|
||||
ctx.Send("Starting processing..."); // Auto-flushes
|
||||
// Continue processing immediately
|
||||
return { messages };
|
||||
}
|
||||
```
|
||||
|
|
@ -630,8 +729,7 @@ function Create(ctx, messages) {
|
|||
function Create(ctx, messages) {
|
||||
const stages = ["validate", "analyze", "prepare"];
|
||||
stages.forEach((stage) => {
|
||||
ctx.Send({ type: "loading", props: { message: `${stage}...` } });
|
||||
ctx.Flush();
|
||||
ctx.Send({ type: "loading", props: { message: `${stage}...` } }); // Auto-flushes
|
||||
performStage(stage);
|
||||
});
|
||||
return { messages };
|
||||
|
|
@ -647,8 +745,7 @@ function Before(ctx, messages, response) {
|
|||
ctx.Send({
|
||||
type: "thinking",
|
||||
props: { content: "Analyzing complex query..." },
|
||||
});
|
||||
ctx.Flush();
|
||||
}); // Auto-flushes
|
||||
}
|
||||
return { response };
|
||||
}
|
||||
|
|
@ -659,8 +756,7 @@ function Before(ctx, messages, response) {
|
|||
```javascript
|
||||
function Error(ctx, messages, error) {
|
||||
if (error.code === "RATE_LIMIT") {
|
||||
ctx.Send("Service is busy, retrying...");
|
||||
ctx.Flush();
|
||||
ctx.Send("Service is busy, retrying..."); // Auto-flushes
|
||||
time.Sleep(1000);
|
||||
return { retry: true };
|
||||
}
|
||||
|
|
@ -668,8 +764,7 @@ function Error(ctx, messages, error) {
|
|||
ctx.Send({
|
||||
type: "error",
|
||||
props: { message: "Sorry, something went wrong.", code: error.code },
|
||||
});
|
||||
ctx.Flush();
|
||||
}); // Auto-flushes
|
||||
return { error };
|
||||
}
|
||||
```
|
||||
|
|
@ -686,16 +781,16 @@ Each hook receives different parameters:
|
|||
- `Done(ctx, messages, response)` - Context, messages, and final response
|
||||
- `Error(ctx, messages, error)` - Context, messages, and error object
|
||||
|
||||
### 2. Always Flush for Real-time Updates
|
||||
### 2. Messages Auto-Flush for Real-time Updates
|
||||
|
||||
```javascript
|
||||
// Good - user sees message immediately
|
||||
ctx.Send("Processing...");
|
||||
ctx.Flush();
|
||||
// Messages are automatically flushed after each Send
|
||||
ctx.Send("Processing..."); // Sent immediately to client
|
||||
|
||||
// Bad - message buffered until hook returns
|
||||
ctx.Send("Processing...");
|
||||
// ... hook continues ...
|
||||
// Multiple sends work seamlessly
|
||||
ctx.Send("Step 1"); // Flushed
|
||||
ctx.Send("Step 2"); // Flushed
|
||||
ctx.Send("Step 3"); // Flushed
|
||||
```
|
||||
|
||||
### 3. Delta Updates Require Unique IDs
|
||||
|
|
@ -722,9 +817,10 @@ ctx.Send({
|
|||
|
||||
### 5. Performance Considerations
|
||||
|
||||
- Use `Flush()` sparingly - only when immediate delivery is needed
|
||||
- Batch related messages with `SendGroup()` when possible
|
||||
- Avoid sending too many small updates (combine them)
|
||||
- Messages auto-flush after each `Send()` for real-time delivery
|
||||
- Batch related messages with `SendGroup()` when possible for better performance
|
||||
- Avoid sending too many small updates (combine them when feasible)
|
||||
- Use `SendGroupStart`/`SendGroupEnd` for fine-grained control over grouping
|
||||
|
||||
### 6. Context Information Available
|
||||
|
||||
|
|
@ -757,9 +853,8 @@ function Create(ctx, messages) {
|
|||
|
||||
```javascript
|
||||
function Create(ctx, messages) {
|
||||
ctx.Send("Hello");
|
||||
ctx.SendGroup({ id: "grp1", messages: [...] });
|
||||
ctx.Flush();
|
||||
ctx.Send("Hello"); // Auto-flushes
|
||||
ctx.SendGroup({ messages: [...] }); // Auto-handles events and flushing
|
||||
return { messages };
|
||||
}
|
||||
```
|
||||
|
|
@ -768,17 +863,24 @@ function Create(ctx, messages) {
|
|||
|
||||
1. **Use String Shorthand**: `ctx.Send("Hello")` is simpler than `ctx.Send({ type: "text", props: { content: "Hello" } })`
|
||||
|
||||
2. **Flush After Each Step**: Ensure users see progress in real-time
|
||||
2. **Messages Auto-Flush**: Each `Send()` automatically flushes for real-time delivery - no manual flushing needed
|
||||
|
||||
3. **Handle Errors Gracefully**: Always provide user-friendly error messages
|
||||
3. **Choose the Right API Level**:
|
||||
|
||||
4. **Show Progress for Long Operations**: Use loading indicators for better UX
|
||||
- **High-level**: Use `SendGroup()` for simple grouped messages (auto-handles events)
|
||||
- **Low-level**: Use `SendGroupStart()`/`SendGroupEnd()` for fine-grained control
|
||||
|
||||
5. **Return Hook Results**: Always return required objects from hooks:
|
||||
4. **Handle Errors Gracefully**: Always provide user-friendly error messages
|
||||
|
||||
5. **Show Progress for Long Operations**: Use loading indicators for better UX
|
||||
|
||||
6. **Return Hook Results**: Always return required objects from hooks:
|
||||
|
||||
- `Create`: `{ messages }`
|
||||
- `Before/After`: `{ response }`
|
||||
- `Done`: `{}` or `{ response }`
|
||||
- `Error`: `{ error }` or `{ retry: true }`
|
||||
|
||||
6. **Test with Different Clients**: Verify behavior with both OpenAI and CUI clients
|
||||
7. **Test with Different Clients**: Verify behavior with both OpenAI and CUI clients
|
||||
|
||||
8. **Group Related Messages**: Use groups to organize related content for better frontend rendering
|
||||
|
|
|
|||
|
|
@ -1,7 +1,11 @@
|
|||
package context
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/yaoapp/gou/runtime/v8/bridge"
|
||||
"github.com/yaoapp/yao/agent/output"
|
||||
"github.com/yaoapp/yao/agent/output/message"
|
||||
traceJsapi "github.com/yaoapp/yao/trace/jsapi"
|
||||
"rogchap.com/v8go"
|
||||
)
|
||||
|
|
@ -47,7 +51,8 @@ func (ctx *Context) NewObject(v8ctx *v8go.Context) (*v8go.Value, error) {
|
|||
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()))
|
||||
jsObject.Set("SendGroupStart", ctx.sendGroupStartMethod(v8ctx.Isolate()))
|
||||
jsObject.Set("SendGroupEnd", ctx.sendGroupEndMethod(v8ctx.Isolate()))
|
||||
|
||||
// Create instance
|
||||
instance, err := jsObject.NewInstance(v8ctx)
|
||||
|
|
@ -164,6 +169,7 @@ func (ctx *Context) traceMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
|||
// sendMethod implements ctx.Send(message)
|
||||
// Usage: ctx.Send({ type: "text", props: { content: "Hello" } })
|
||||
// Usage: ctx.Send("Hello") // shorthand for text message
|
||||
// Automatically generates ID and flushes output
|
||||
func (ctx *Context) sendMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
||||
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
|
||||
v8ctx := info.Context()
|
||||
|
|
@ -179,17 +185,28 @@ func (ctx *Context) sendMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
|||
return bridge.JsException(v8ctx, "invalid message: "+err.Error())
|
||||
}
|
||||
|
||||
// Generate unique ID if not provided
|
||||
if msg.ID == "" {
|
||||
msg.ID = output.GenerateID()
|
||||
}
|
||||
|
||||
// Call ctx.Send
|
||||
if err := ctx.Send(msg); err != nil {
|
||||
return bridge.JsException(v8ctx, "Send failed: "+err.Error())
|
||||
}
|
||||
|
||||
// Automatically flush after sending
|
||||
if err := ctx.Flush(); err != nil {
|
||||
return bridge.JsException(v8ctx, "Flush failed: "+err.Error())
|
||||
}
|
||||
|
||||
return v8go.Undefined(iso)
|
||||
})
|
||||
}
|
||||
|
||||
// sendGroupMethod implements ctx.SendGroup(group)
|
||||
// Usage: ctx.SendGroup({ id: "group1", messages: [...] })
|
||||
// Automatically generates IDs, sends group_start/group_end events, and flushes output
|
||||
func (ctx *Context) sendGroupMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
||||
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
|
||||
v8ctx := info.Context()
|
||||
|
|
@ -205,24 +222,159 @@ func (ctx *Context) sendGroupMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
|||
return bridge.JsException(v8ctx, "invalid group: "+err.Error())
|
||||
}
|
||||
|
||||
// Generate group ID if not provided
|
||||
if group.ID == "" {
|
||||
group.ID = output.GenerateID()
|
||||
}
|
||||
|
||||
// Send group_start event
|
||||
startTime := time.Now()
|
||||
startEvent := output.NewEventMessage(
|
||||
message.EventGroupStart,
|
||||
"Group started",
|
||||
message.GroupStartData{
|
||||
GroupID: group.ID,
|
||||
Type: "mixed", // Mixed types in group
|
||||
Timestamp: startTime.UnixMilli(),
|
||||
},
|
||||
)
|
||||
if err := ctx.Send(startEvent); err != nil {
|
||||
return bridge.JsException(v8ctx, "Failed to send group_start event: "+err.Error())
|
||||
}
|
||||
if err := ctx.Flush(); err != nil {
|
||||
return bridge.JsException(v8ctx, "Flush failed after group_start: "+err.Error())
|
||||
}
|
||||
|
||||
// Generate IDs for messages and set group_id
|
||||
for _, msg := range group.Messages {
|
||||
if msg.ID == "" {
|
||||
msg.ID = output.GenerateID()
|
||||
}
|
||||
if msg.GroupID == "" {
|
||||
msg.GroupID = group.ID
|
||||
}
|
||||
}
|
||||
|
||||
// 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 bridge.JsException(v8ctx, "Flush failed after SendGroup: "+err.Error())
|
||||
}
|
||||
|
||||
// Send group_end event
|
||||
endEvent := output.NewEventMessage(
|
||||
message.EventGroupEnd,
|
||||
"Group completed",
|
||||
message.GroupEndData{
|
||||
GroupID: group.ID,
|
||||
Type: "mixed",
|
||||
Timestamp: time.Now().UnixMilli(),
|
||||
DurationMs: time.Since(startTime).Milliseconds(),
|
||||
ChunkCount: len(group.Messages),
|
||||
Status: "completed",
|
||||
},
|
||||
)
|
||||
if err := ctx.Send(endEvent); err != nil {
|
||||
return bridge.JsException(v8ctx, "Failed to send group_end event: "+err.Error())
|
||||
}
|
||||
if err := ctx.Flush(); err != nil {
|
||||
return bridge.JsException(v8ctx, "Flush failed after group_end: "+err.Error())
|
||||
}
|
||||
|
||||
return v8go.Undefined(iso)
|
||||
})
|
||||
}
|
||||
|
||||
// sendGroupStartMethod implements ctx.SendGroupStart(type?, id?)
|
||||
// Usage: const groupId = ctx.SendGroupStart() // type="mixed", auto-generate ID
|
||||
// Usage: const groupId = ctx.SendGroupStart("text") // type="text", auto-generate ID
|
||||
// Usage: const groupId = ctx.SendGroupStart("text", "my-group-id") // type="text", use provided ID
|
||||
// Returns the group ID (generated or provided)
|
||||
func (ctx *Context) sendGroupStartMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
||||
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
|
||||
v8ctx := info.Context()
|
||||
args := info.Args()
|
||||
|
||||
// Get type (default: "mixed")
|
||||
groupType := "mixed"
|
||||
if len(args) > 0 && args[0].IsString() {
|
||||
groupType = args[0].String()
|
||||
}
|
||||
|
||||
// Get or generate group ID
|
||||
var groupID string
|
||||
if len(args) > 1 && args[1].IsString() {
|
||||
groupID = args[1].String()
|
||||
} else {
|
||||
groupID = output.GenerateID()
|
||||
}
|
||||
|
||||
// Send group_start event
|
||||
startEvent := output.NewEventMessage(
|
||||
message.EventGroupStart,
|
||||
"Group started",
|
||||
message.GroupStartData{
|
||||
GroupID: groupID,
|
||||
Type: groupType,
|
||||
Timestamp: time.Now().UnixMilli(),
|
||||
},
|
||||
)
|
||||
if err := ctx.Send(startEvent); err != nil {
|
||||
return bridge.JsException(v8ctx, "Failed to send group_start event: "+err.Error())
|
||||
}
|
||||
if err := ctx.Flush(); err != nil {
|
||||
return bridge.JsException(v8ctx, "Flush failed after group_start: "+err.Error())
|
||||
}
|
||||
|
||||
// Return the group ID
|
||||
groupIDVal, err := v8go.NewValue(iso, groupID)
|
||||
if err != nil {
|
||||
return bridge.JsException(v8ctx, "Failed to create return value: "+err.Error())
|
||||
}
|
||||
return groupIDVal
|
||||
})
|
||||
}
|
||||
|
||||
// sendGroupEndMethod implements ctx.SendGroupEnd(id, chunkCount?)
|
||||
// Usage: ctx.SendGroupEnd(groupId)
|
||||
// Usage: ctx.SendGroupEnd(groupId, 10) // With chunk count
|
||||
func (ctx *Context) sendGroupEndMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
||||
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
|
||||
v8ctx := info.Context()
|
||||
args := info.Args()
|
||||
|
||||
// Group ID is required
|
||||
if len(args) < 1 || !args[0].IsString() {
|
||||
return bridge.JsException(v8ctx, "SendGroupEnd requires a group ID (string) as first argument")
|
||||
}
|
||||
groupID := args[0].String()
|
||||
|
||||
// Optional chunk count
|
||||
chunkCount := 0
|
||||
if len(args) > 1 && args[1].IsNumber() {
|
||||
chunkCount = int(args[1].Integer())
|
||||
}
|
||||
|
||||
// Send group_end event
|
||||
endEvent := output.NewEventMessage(
|
||||
message.EventGroupEnd,
|
||||
"Group completed",
|
||||
message.GroupEndData{
|
||||
GroupID: groupID,
|
||||
Type: "mixed",
|
||||
Timestamp: time.Now().UnixMilli(),
|
||||
DurationMs: 0, // Duration not tracked at this level
|
||||
ChunkCount: chunkCount,
|
||||
Status: "completed",
|
||||
},
|
||||
)
|
||||
if err := ctx.Send(endEvent); err != nil {
|
||||
return bridge.JsException(v8ctx, "Failed to send group_end event: "+err.Error())
|
||||
}
|
||||
if err := ctx.Flush(); err != nil {
|
||||
return bridge.JsException(v8ctx, "Flush failed after group_end: "+err.Error())
|
||||
}
|
||||
|
||||
return v8go.Undefined(iso)
|
||||
|
|
|
|||
|
|
@ -58,9 +58,6 @@ func parseMessage(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Message, e
|
|||
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
|
||||
}
|
||||
|
|
@ -73,12 +70,6 @@ func parseMessage(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Message, e
|
|||
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 {
|
||||
|
|
@ -158,9 +149,6 @@ func parseGroup(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Group, error
|
|||
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
|
||||
}
|
||||
|
|
@ -173,12 +161,6 @@ func parseGroup(v8ctx *v8go.Context, jsValue *v8go.Value) (*message.Group, error
|
|||
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 {
|
||||
|
|
|
|||
|
|
@ -163,8 +163,8 @@ func TestJsValueSendGroup(t *testing.T) {
|
|||
assert.Equal(t, true, result["success"], "SendGroup should succeed")
|
||||
}
|
||||
|
||||
// TestJsValueFlush test the Flush method on Context
|
||||
func TestJsValueFlush(t *testing.T) {
|
||||
// TestJsValueSendGroupStartEnd test the SendGroupStart and SendGroupEnd methods
|
||||
func TestJsValueSendGroupStartEnd(t *testing.T) {
|
||||
|
||||
test.Prepare(t, config.Conf)
|
||||
defer test.Clean()
|
||||
|
|
@ -181,13 +181,17 @@ func TestJsValueFlush(t *testing.T) {
|
|||
res, err := v8.Call(v8.CallOptions{}, `
|
||||
function test(ctx) {
|
||||
try {
|
||||
// Send a message
|
||||
ctx.Send("Processing...");
|
||||
// Start a group with auto-generated ID
|
||||
const groupId = ctx.SendGroupStart("text");
|
||||
|
||||
// Flush output
|
||||
ctx.Flush();
|
||||
// Send messages in the group
|
||||
ctx.Send({ type: "text", props: { content: "Message 1" }, group_id: groupId });
|
||||
ctx.Send({ type: "text", props: { content: "Message 2" }, group_id: groupId });
|
||||
|
||||
return { success: true };
|
||||
// End the group
|
||||
ctx.SendGroupEnd(groupId, 2);
|
||||
|
||||
return { success: true, groupId: groupId };
|
||||
} catch (error) {
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
|
|
@ -200,7 +204,8 @@ func TestJsValueFlush(t *testing.T) {
|
|||
if !ok {
|
||||
t.Fatalf("Expected map result, got %T", res)
|
||||
}
|
||||
assert.Equal(t, true, result["success"], "Flush should succeed")
|
||||
assert.Equal(t, true, result["success"], "SendGroupStart/End should succeed")
|
||||
assert.NotEmpty(t, result["groupId"], "Should return group ID")
|
||||
}
|
||||
|
||||
// TestJsValueSendDeltaUpdates test delta updates in Send
|
||||
|
|
@ -239,13 +244,7 @@ func TestJsValueSendDeltaUpdates(t *testing.T) {
|
|||
delta_action: "append"
|
||||
});
|
||||
|
||||
// Mark as complete
|
||||
ctx.Send({
|
||||
type: "text",
|
||||
props: {},
|
||||
id: "msg_1",
|
||||
done: true
|
||||
});
|
||||
// Send completion (no done field needed)
|
||||
|
||||
return { success: true };
|
||||
} catch (error) {
|
||||
|
|
@ -329,8 +328,6 @@ func TestJsValueSendMultipleTypes(t *testing.T) {
|
|||
}
|
||||
});
|
||||
|
||||
ctx.Flush();
|
||||
|
||||
return { success: true };
|
||||
} catch (error) {
|
||||
return { success: false, error: error.message };
|
||||
|
|
@ -467,7 +464,6 @@ func TestJsValueSendWithCUIAccept(t *testing.T) {
|
|||
type: "text",
|
||||
props: { content: "Hello CUI" }
|
||||
});
|
||||
ctx.Flush();
|
||||
return { success: true };
|
||||
} catch (error) {
|
||||
return { success: false, error: error.message };
|
||||
|
|
@ -566,15 +562,11 @@ func TestJsValueSendChainedCalls(t *testing.T) {
|
|||
res, err := v8.Call(v8.CallOptions{}, `
|
||||
function test(ctx) {
|
||||
try {
|
||||
// Multiple sequential sends
|
||||
// Multiple sequential sends (each auto-flushes)
|
||||
ctx.Send("Step 1");
|
||||
ctx.Send("Step 2");
|
||||
ctx.Send("Step 3");
|
||||
ctx.Flush();
|
||||
|
||||
// Send after flush
|
||||
ctx.Send("Step 4");
|
||||
ctx.Flush();
|
||||
|
||||
return { success: true };
|
||||
} catch (error) {
|
||||
|
|
|
|||
|
|
@ -39,7 +39,6 @@ type Message struct {
|
|||
// Streaming control
|
||||
ID string `json:"id,omitempty"` // Unique message ID (for merging in streaming)
|
||||
Delta bool `json:"delta,omitempty"` // Whether this is an incremental update
|
||||
Done bool `json:"done,omitempty"` // Whether the message is complete
|
||||
|
||||
// Delta update control (for incremental props updates)
|
||||
DeltaPath string `json:"delta_path,omitempty"` // Which field to update (e.g., "content", "items.0.name")
|
||||
|
|
@ -85,11 +84,7 @@ type Message struct {
|
|||
- `true`: Append/update to existing message with same ID
|
||||
- `false`: Complete message (default)
|
||||
- Used for streaming LLM responses
|
||||
|
||||
- **`Done`** (optional): Marks message as complete
|
||||
- `true`: No more updates will come for this message ID
|
||||
- `false`: More updates may follow
|
||||
- Typically sent as final message in a delta sequence
|
||||
- Message completion is signaled via `group_end` event instead
|
||||
|
||||
#### Delta Update Control
|
||||
|
||||
|
|
@ -379,11 +374,22 @@ msg := &message.Message{
|
|||
}
|
||||
output.Send(ctx, msg)
|
||||
|
||||
// Mark as complete
|
||||
msg.Delta = false
|
||||
msg.Done = true
|
||||
msg.Props["content"] = "Hello world!" // Full content
|
||||
// Send more delta updates...
|
||||
msg.Props["content"] = " world"
|
||||
output.Send(ctx, msg)
|
||||
|
||||
// Mark completion with group_end event
|
||||
endData := message.GroupEndData{
|
||||
GroupID: "msg_123",
|
||||
Type: "text",
|
||||
Status: "completed",
|
||||
ChunkCount: 2,
|
||||
Extra: map[string]interface{}{
|
||||
"content": "Hello world!", // Full content
|
||||
},
|
||||
}
|
||||
eventMsg := output.NewEventMessage(message.EventGroupEnd, "Group completed", endData)
|
||||
output.Send(ctx, eventMsg)
|
||||
```
|
||||
|
||||
### Custom Writers
|
||||
|
|
|
|||
|
|
@ -90,13 +90,6 @@ func (a *Adapter) Adapt(msg *message.Message) ([]interface{}, error) {
|
|||
return []interface{}{}, nil // Return empty array, nothing to send
|
||||
}
|
||||
|
||||
// Skip "done" messages that are not delta updates
|
||||
// These are final confirmation messages for CUI clients
|
||||
// OpenAI clients don't need them - they use finish_reason instead
|
||||
if msg.Done && !msg.Delta {
|
||||
return []interface{}{}, nil // Return empty array, nothing to send
|
||||
}
|
||||
|
||||
// Get converter for this message type
|
||||
converter, exists := a.registry.GetConverter(msg.Type)
|
||||
if !exists {
|
||||
|
|
|
|||
|
|
@ -37,9 +37,8 @@ type Message struct {
|
|||
Props map[string]interface{} `json:"props,omitempty"` // Message properties (passed to frontend component)
|
||||
|
||||
// Streaming control
|
||||
ID string `json:"id,omitempty"` // Message ID (used for merging messages in streaming scenarios)
|
||||
ID string `json:"id,omitempty"` // Unique chunk/message ID (each chunk has unique ID; use group_id for merging)
|
||||
Delta bool `json:"delta,omitempty"` // Whether this is an incremental update
|
||||
Done bool `json:"done,omitempty"` // Whether the message is complete
|
||||
|
||||
// Delta update control
|
||||
DeltaPath string `json:"delta_path,omitempty"` // Update path (e.g., "content", "data", "items.0.name")
|
||||
|
|
@ -49,9 +48,7 @@ type Message struct {
|
|||
TypeChange bool `json:"type_change,omitempty"` // Marks this as a type correction message
|
||||
|
||||
// Message group
|
||||
GroupID string `json:"group_id,omitempty"` // Parent message group ID
|
||||
GroupStart bool `json:"group_start,omitempty"` // Marks the start of a message group
|
||||
GroupEnd bool `json:"group_end,omitempty"` // Marks the end of a message group
|
||||
GroupID string `json:"group_id,omitempty"` // Group ID (all delta chunks of same logical message share this; used for merging)
|
||||
|
||||
// Metadata
|
||||
Metadata *Metadata `json:"metadata,omitempty"` // Additional metadata
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue