yao/agent/context/context.go
Max 91aa109f5b Add streaming support and message handling in DefaultStreamHandler
- Implemented a state management system for handling different streaming chunk types, including stream start, group start, text, thinking, tool calls, metadata, errors, and stream end.
- Introduced a new Send method in the context to facilitate message sending to the client.
- Enhanced the output package with message sending capabilities, including support for message groups and flushing.
- Improved error handling and message ID generation for better streaming reliability and user feedback.
2025-11-15 17:40:03 +08:00

165 lines
3.6 KiB
Go

package context
import (
"context"
"time"
jsoniter "github.com/json-iterator/go"
"github.com/yaoapp/gou/plan"
"github.com/yaoapp/kun/log"
"github.com/yaoapp/yao/openapi/oauth/types"
)
// New create a new context
func New(parent context.Context, authorized *types.AuthorizedInfo, chatID, payload string) Context {
if parent == nil {
parent = context.Background()
}
// Validate the client type
ctx := Context{
Context: parent,
Space: plan.NewMemorySharedSpace(),
ChatID: chatID,
}
if payload == "" {
return ctx
}
err := jsoniter.Unmarshal([]byte(payload), &ctx)
if err != nil {
log.Error("%s", err.Error())
}
return ctx
}
// NewWithCancel create a new context with cancel
func NewWithCancel(parent context.Context, authorized *types.AuthorizedInfo, chatID, payload string) (Context, context.CancelFunc) {
ctx := New(parent, authorized, chatID, payload)
return WithCancel(ctx)
}
// NewWithTimeout create a new context with timeout
func NewWithTimeout(parent context.Context, authorized *types.AuthorizedInfo, chatID, payload string, timeout time.Duration) (Context, context.CancelFunc) {
ctx := New(parent, authorized, chatID, payload)
return WithTimeout(ctx, timeout)
}
// WithCancel create a new context
func WithCancel(parent Context) (Context, context.CancelFunc) {
new, cancel := context.WithCancel(parent.Context)
parent.Context = new
return parent, cancel
}
// WithTimeout create a new context
func WithTimeout(parent Context, timeout time.Duration) (Context, context.CancelFunc) {
new, cancel := context.WithTimeout(parent.Context, timeout)
parent.Context = new
return parent, cancel
}
// Release the context and clean up all resources including stacks
func (ctx *Context) Release() {
// Clear space
if ctx.Space != nil {
ctx.Space.Clear()
ctx.Space = nil
}
// Clear stacks
if ctx.Stacks != nil {
for k := range ctx.Stacks {
delete(ctx.Stacks, k)
}
ctx.Stacks = nil
}
// Clear current stack reference
ctx.Stack = nil
// Clear writer reference
ctx.Writer = nil
ctx = nil
}
// Send sends data to the context's writer
// This is used by the output module to send messages to the client
func (ctx *Context) Send(data []byte) error {
if ctx.Writer == nil {
return nil // No writer, silently ignore
}
_, err := ctx.Writer.Write(data)
return err
}
// Map the context to a map
func (ctx *Context) Map() map[string]interface{} {
data := map[string]interface{}{}
// Authorized information
if ctx.Authorized != nil {
data["authorized"] = ctx.Authorized
}
if ctx.ChatID != "" {
data["chat_id"] = ctx.ChatID
}
if ctx.AssistantID != "" {
data["assistant_id"] = ctx.AssistantID
}
if ctx.Connector != "" {
data["connector"] = ctx.Connector
}
if ctx.Search != nil {
data["search"] = *ctx.Search
}
// Arguments for call
if len(ctx.Args) > 0 {
data["args"] = ctx.Args
}
if ctx.Retry {
data["retry"] = ctx.Retry
}
if ctx.RetryTimes > 0 {
data["retry_times"] = ctx.RetryTimes
}
// Locale information
if ctx.Locale != "" {
data["locale"] = ctx.Locale
}
if ctx.Theme != "" {
data["theme"] = ctx.Theme
}
// Request information
if ctx.Client.Type != "" || ctx.Client.UserAgent != "" || ctx.Client.IP != "" {
data["client"] = map[string]interface{}{
"type": ctx.Client.Type,
"user_agent": ctx.Client.UserAgent,
"ip": ctx.Client.IP,
}
}
if ctx.Referer != "" {
data["referer"] = ctx.Referer
}
if ctx.Accept != "" {
data["accept"] = ctx.Accept
}
// CUI Context information
if ctx.Route != "" {
data["route"] = ctx.Route
}
if len(ctx.Metadata) > 0 {
data["metadata"] = ctx.Metadata
}
return data
}