Update Dependencies and Enhance Agent Context with Forked Context Support
- Updated `logrus` dependency from v1.9.3 to v1.9.4 and `golang.org/x/sys` from v0.38.0 to v0.40.0 for improved functionality and security. - Introduced a `Fork` method in the agent context to create child contexts for concurrent agent and LLM calls, preventing race conditions on shared state during batch operations. - Enhanced the `Orchestrator` methods to utilize forked contexts, ensuring independent execution of agent calls without interference.
This commit is contained in:
Max
parent
3a9f32af12
commit
02e813af0b
12 changed files with 1732 additions and 20 deletions
|
|
@ -8,6 +8,7 @@ import (
|
|||
"github.com/yaoapp/yao/agent/caller"
|
||||
agentContext "github.com/yaoapp/yao/agent/context"
|
||||
"github.com/yaoapp/yao/agent/i18n"
|
||||
"github.com/yaoapp/yao/agent/llm"
|
||||
"github.com/yaoapp/yao/agent/search"
|
||||
searchTypes "github.com/yaoapp/yao/agent/search/types"
|
||||
store "github.com/yaoapp/yao/agent/store/types"
|
||||
|
|
@ -28,6 +29,9 @@ func init() {
|
|||
// Initialize Agent JSAPI factory for ctx.agent.* methods
|
||||
caller.SetJSAPIFactory()
|
||||
|
||||
// Initialize LLM JSAPI factory for ctx.llm.* methods
|
||||
llm.SetJSAPIFactory()
|
||||
|
||||
// Initialize Search JSAPI factory with config getter
|
||||
search.SetJSAPIFactory(func(assistantID string) (*searchTypes.Config, *search.Uses) {
|
||||
ast, err := Get(assistantID)
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ type callResult struct {
|
|||
|
||||
// All executes all agent calls and waits for all to complete (like Promise.all)
|
||||
// Returns results in the same order as requests, regardless of completion order
|
||||
// Each call uses a forked context to avoid race conditions on shared state
|
||||
func (o *Orchestrator) All(reqs []*Request) []*Result {
|
||||
if len(reqs) == 0 {
|
||||
return []*Result{}
|
||||
|
|
@ -49,7 +50,8 @@ func (o *Orchestrator) All(reqs []*Request) []*Result {
|
|||
}
|
||||
}()
|
||||
|
||||
result := o.callAgent(r)
|
||||
// Use forked context to avoid race conditions
|
||||
result := o.callAgentWithForkedContext(r)
|
||||
mu.Lock()
|
||||
results[idx] = result
|
||||
mu.Unlock()
|
||||
|
|
@ -63,6 +65,7 @@ func (o *Orchestrator) All(reqs []*Request) []*Result {
|
|||
// Any returns as soon as any agent call succeeds (has non-error result) (like Promise.any)
|
||||
// Other calls continue in background but results are discarded after first success
|
||||
// Returns all results received so far when first success is found
|
||||
// Each call uses a forked context to avoid race conditions on shared state
|
||||
func (o *Orchestrator) Any(reqs []*Request) []*Result {
|
||||
if len(reqs) == 0 {
|
||||
return []*Result{}
|
||||
|
|
@ -98,7 +101,8 @@ func (o *Orchestrator) Any(reqs []*Request) []*Result {
|
|||
default:
|
||||
}
|
||||
|
||||
result := o.callAgent(r)
|
||||
// Use forked context to avoid race conditions
|
||||
result := o.callAgentWithForkedContext(r)
|
||||
|
||||
// Try to send result
|
||||
select {
|
||||
|
|
@ -132,6 +136,7 @@ func (o *Orchestrator) Any(reqs []*Request) []*Result {
|
|||
// Race returns as soon as any agent call completes (like Promise.race)
|
||||
// Returns immediately when first result arrives, regardless of success/failure
|
||||
// Note: Still waits for all goroutines to complete before returning to avoid resource leaks
|
||||
// Each call uses a forked context to avoid race conditions on shared state
|
||||
func (o *Orchestrator) Race(reqs []*Request) []*Result {
|
||||
if len(reqs) == 0 {
|
||||
return []*Result{}
|
||||
|
|
@ -167,7 +172,8 @@ func (o *Orchestrator) Race(reqs []*Request) []*Result {
|
|||
default:
|
||||
}
|
||||
|
||||
result := o.callAgent(r)
|
||||
// Use forked context to avoid race conditions
|
||||
result := o.callAgentWithForkedContext(r)
|
||||
|
||||
// Try to send result
|
||||
select {
|
||||
|
|
@ -200,6 +206,21 @@ func (o *Orchestrator) Race(reqs []*Request) []*Result {
|
|||
// callAgent executes a single agent call using the AgentGetterFunc
|
||||
// This method handles context sharing and result extraction
|
||||
func (o *Orchestrator) callAgent(req *Request) *Result {
|
||||
return o.callAgentWithContext(o.ctx, req)
|
||||
}
|
||||
|
||||
// callAgentWithForkedContext executes a single agent call with a forked context
|
||||
// This is used by batch operations (All/Any/Race) to avoid race conditions
|
||||
// when multiple goroutines modify shared context state (Stack, Logger, etc.)
|
||||
func (o *Orchestrator) callAgentWithForkedContext(req *Request) *Result {
|
||||
// Fork the context to get independent Stack and Logger
|
||||
forkedCtx := o.ctx.Fork()
|
||||
return o.callAgentWithContext(forkedCtx, req)
|
||||
}
|
||||
|
||||
// callAgentWithContext executes a single agent call with the given context
|
||||
// This is the core implementation used by both callAgent and callAgentWithForkedContext
|
||||
func (o *Orchestrator) callAgentWithContext(ctx *agentContext.Context, req *Request) *Result {
|
||||
if req == nil {
|
||||
return &Result{Error: "nil request"}
|
||||
}
|
||||
|
|
@ -237,9 +258,9 @@ func (o *Orchestrator) callAgent(req *Request) *Result {
|
|||
ctxOpts.OnMessage = req.Handler
|
||||
}
|
||||
|
||||
// Execute the agent call with shared context
|
||||
// The agent.Stream method will use the parent context's Writer for output
|
||||
resp, err := agent.Stream(o.ctx, req.Messages, ctxOpts)
|
||||
// Execute the agent call with the provided context
|
||||
// The agent.Stream method will use the context's Writer for output
|
||||
resp, err := agent.Stream(ctx, req.Messages, ctxOpts)
|
||||
if err != nil {
|
||||
result.Error = "agent call failed: " + err.Error()
|
||||
return result
|
||||
|
|
|
|||
|
|
@ -132,6 +132,62 @@ func (ctx *Context) GetAuthorizedMap() map[string]interface{} {
|
|||
return ctx.Authorized.AuthorizedToMap()
|
||||
}
|
||||
|
||||
// Fork creates a child context for concurrent agent/LLM calls
|
||||
// The forked context shares read-only resources (Memory, Authorized, Cache, Writer)
|
||||
// but has its own independent Stack and Logger to avoid race conditions
|
||||
//
|
||||
// This is essential for batch operations (All/Any/Race) where multiple goroutines
|
||||
// need to execute concurrently without interfering with each other's Stack state.
|
||||
//
|
||||
// The forked context does NOT need to be released separately - the parent context
|
||||
// manages shared resources. However, the child's Stack will be collected in parent's Stacks map.
|
||||
func (ctx *Context) Fork() *Context {
|
||||
childID := generateContextID()
|
||||
|
||||
child := &Context{
|
||||
// Inherit parent's standard context
|
||||
Context: ctx.Context,
|
||||
|
||||
// New unique ID for this forked context
|
||||
ID: childID,
|
||||
|
||||
// Share read-only/thread-safe resources with parent
|
||||
Memory: ctx.Memory, // Memory is designed to be shared
|
||||
Cache: ctx.Cache, // Cache store is thread-safe
|
||||
Writer: ctx.Writer, // Output writer is thread-safe (output module handles concurrency)
|
||||
Authorized: ctx.Authorized, // Read-only auth info
|
||||
Capabilities: ctx.Capabilities, // Read-only model capabilities
|
||||
|
||||
// Share reference to parent's Stacks map for trace collection
|
||||
// Child stacks will be added here by EnterStack
|
||||
Stacks: ctx.Stacks,
|
||||
|
||||
// Create independent resources to avoid race conditions
|
||||
Stack: nil, // Will be set by EnterStack
|
||||
IDGenerator: message.NewIDGenerator(),
|
||||
Logger: NewRequestLogger(ctx.AssistantID, ctx.ChatID, childID),
|
||||
messageMetadata: newMessageMetadataStore(),
|
||||
|
||||
// Inherit context metadata
|
||||
ChatID: ctx.ChatID,
|
||||
AssistantID: ctx.AssistantID,
|
||||
Locale: ctx.Locale,
|
||||
Theme: ctx.Theme,
|
||||
Client: ctx.Client,
|
||||
Referer: ctx.Referer,
|
||||
Accept: ctx.Accept,
|
||||
Route: ctx.Route,
|
||||
Metadata: ctx.Metadata,
|
||||
|
||||
// Don't inherit these - they are request-specific
|
||||
Buffer: nil, // Buffer belongs to root context
|
||||
Interrupt: nil, // Interrupt controller belongs to root context
|
||||
trace: nil, // Trace will be inherited via TraceID in Stack
|
||||
}
|
||||
|
||||
return child
|
||||
}
|
||||
|
||||
// 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 {
|
||||
|
|
|
|||
|
|
@ -71,6 +71,9 @@ func (ctx *Context) NewObject(v8ctx *v8go.Context) (*v8go.Value, error) {
|
|||
// Set agent object for calling other agents
|
||||
jsObject.Set("agent", ctx.newAgentObject(v8ctx.Isolate()))
|
||||
|
||||
// Set llm object for direct LLM calls
|
||||
jsObject.Set("llm", ctx.newLlmObject(v8ctx.Isolate()))
|
||||
|
||||
// Note: Space object will be set after instance creation (requires v8ctx)
|
||||
|
||||
// Create instance
|
||||
|
|
|
|||
360
agent/context/jsapi_llm.go
Normal file
360
agent/context/jsapi_llm.go
Normal file
|
|
@ -0,0 +1,360 @@
|
|||
package context
|
||||
|
||||
import (
|
||||
"github.com/yaoapp/gou/runtime/v8/bridge"
|
||||
"github.com/yaoapp/yao/agent/output/message"
|
||||
"rogchap.com/v8go"
|
||||
)
|
||||
|
||||
// LlmAPI defines the LLM JSAPI interface for ctx.llm.*
|
||||
// This interface is defined here to avoid circular dependency between context and llm packages.
|
||||
// The actual implementation is in agent/llm/jsapi.go
|
||||
type LlmAPI interface {
|
||||
// Stream calls LLM with streaming output to ctx.Writer
|
||||
// Returns *llm.Result or error information
|
||||
Stream(connector string, messages []interface{}, opts map[string]interface{}) interface{}
|
||||
|
||||
// Parallel LLM call methods - inspired by JavaScript Promise
|
||||
// All waits for all LLM calls to complete (like Promise.all)
|
||||
All(requests []interface{}) []interface{}
|
||||
// Any returns when any LLM call succeeds (like Promise.any)
|
||||
Any(requests []interface{}) []interface{}
|
||||
// Race returns when any LLM call completes (like Promise.race)
|
||||
Race(requests []interface{}) []interface{}
|
||||
}
|
||||
|
||||
// LlmAPIWithCallback extends LlmAPI with callback support
|
||||
// This interface provides methods that accept OnMessage handlers for real-time message processing
|
||||
type LlmAPIWithCallback interface {
|
||||
LlmAPI
|
||||
|
||||
// StreamWithHandler calls LLM with an OnMessage handler
|
||||
// handler receives SSE messages: func(msg *message.Message) int
|
||||
StreamWithHandler(connector string, messages []interface{}, opts map[string]interface{}, handler OnMessageFunc) interface{}
|
||||
|
||||
// AllWithHandler executes all LLM calls with handlers
|
||||
// globalHandler receives messages with connectorID and index: func(connectorID, index, msg) int
|
||||
AllWithHandler(requests []interface{}, globalHandler LlmBatchOnMessageFunc) []interface{}
|
||||
|
||||
// AnyWithHandler executes LLM calls and returns on first success, with handlers
|
||||
AnyWithHandler(requests []interface{}, globalHandler LlmBatchOnMessageFunc) []interface{}
|
||||
|
||||
// RaceWithHandler executes LLM calls and returns on first completion, with handlers
|
||||
RaceWithHandler(requests []interface{}, globalHandler LlmBatchOnMessageFunc) []interface{}
|
||||
}
|
||||
|
||||
// LlmBatchOnMessageFunc is the OnMessage function for batch LLM calls
|
||||
// It includes connectorID and index to identify the source of each message
|
||||
type LlmBatchOnMessageFunc func(connectorID string, index int, msg *message.Message) int
|
||||
|
||||
// LlmAPIFactory is a function type that creates a LlmAPI for a context
|
||||
// This is set by the llm package during initialization
|
||||
var LlmAPIFactory func(ctx *Context) LlmAPI
|
||||
|
||||
// Llm returns the LLM API for this context
|
||||
// Returns nil if LlmAPIFactory is not set
|
||||
func (ctx *Context) Llm() LlmAPI {
|
||||
if LlmAPIFactory == nil {
|
||||
return nil
|
||||
}
|
||||
return LlmAPIFactory(ctx)
|
||||
}
|
||||
|
||||
// newLlmObject creates a new llm object with all llm methods
|
||||
// This is called from jsapi.go NewObject() to mount ctx.llm
|
||||
func (ctx *Context) newLlmObject(iso *v8go.Isolate) *v8go.ObjectTemplate {
|
||||
llmObj := v8go.NewObjectTemplate(iso)
|
||||
|
||||
// Single LLM call method
|
||||
llmObj.Set("Stream", ctx.llmStreamMethod(iso))
|
||||
|
||||
// Parallel LLM call methods - inspired by JavaScript Promise
|
||||
llmObj.Set("All", ctx.llmAllMethod(iso))
|
||||
llmObj.Set("Any", ctx.llmAnyMethod(iso))
|
||||
llmObj.Set("Race", ctx.llmRaceMethod(iso))
|
||||
|
||||
return llmObj
|
||||
}
|
||||
|
||||
// llmStreamMethod implements ctx.llm.Stream(connector, messages, options?)
|
||||
// Usage: const result = ctx.llm.Stream("gpt-4o", [{ role: "user", content: "Hello" }], { temperature: 0.7, onChunk: (msg) => 0 })
|
||||
// Returns: { connector, response, content, error }
|
||||
func (ctx *Context) llmStreamMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
||||
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
|
||||
v8ctx := info.Context()
|
||||
args := info.Args()
|
||||
|
||||
// Validate arguments
|
||||
if len(args) < 2 {
|
||||
return bridge.JsException(v8ctx, "Stream requires connector and messages parameters")
|
||||
}
|
||||
|
||||
// Get connector ID (first argument)
|
||||
if !args[0].IsString() {
|
||||
return bridge.JsException(v8ctx, "connector must be a string")
|
||||
}
|
||||
connector := args[0].String()
|
||||
|
||||
// Parse messages (second argument)
|
||||
messagesVal, err := bridge.GoValue(args[1], v8ctx)
|
||||
if err != nil {
|
||||
return bridge.JsException(v8ctx, "invalid messages: "+err.Error())
|
||||
}
|
||||
messages, ok := messagesVal.([]interface{})
|
||||
if !ok {
|
||||
return bridge.JsException(v8ctx, "messages must be an array")
|
||||
}
|
||||
|
||||
// Parse options (optional third argument) - extract onChunk separately
|
||||
var opts map[string]interface{}
|
||||
var onChunkFn *v8go.Function
|
||||
|
||||
if len(args) >= 3 && !args[2].IsUndefined() && !args[2].IsNull() {
|
||||
optsObj, err := args[2].AsObject()
|
||||
if err == nil && optsObj != nil {
|
||||
// Extract onChunk callback before converting to Go value
|
||||
onChunkVal, _ := optsObj.Get("onChunk")
|
||||
if onChunkVal != nil && onChunkVal.IsFunction() {
|
||||
onChunkFn, _ = onChunkVal.AsFunction()
|
||||
}
|
||||
|
||||
// Convert the rest of options to Go map
|
||||
goVal, err := bridge.GoValue(args[2], v8ctx)
|
||||
if err == nil {
|
||||
if optsMap, ok := goVal.(map[string]interface{}); ok {
|
||||
// Remove onChunk from the map (it's handled separately)
|
||||
delete(optsMap, "onChunk")
|
||||
opts = optsMap
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Get LLM API
|
||||
llmAPI := ctx.Llm()
|
||||
if llmAPI == nil {
|
||||
return bridge.JsException(v8ctx, "LLM API not available")
|
||||
}
|
||||
|
||||
var result interface{}
|
||||
|
||||
// If onChunk callback is provided and API supports it, use StreamWithHandler
|
||||
if onChunkFn != nil {
|
||||
if apiWithCb, ok := llmAPI.(LlmAPIWithCallback); ok {
|
||||
// Create Go OnMessageFunc that calls JS callback
|
||||
handler := createJSStreamHandler(v8ctx, onChunkFn)
|
||||
result = apiWithCb.StreamWithHandler(connector, messages, opts, handler)
|
||||
} else {
|
||||
// Fallback: ignore callback if API doesn't support it
|
||||
result = llmAPI.Stream(connector, messages, opts)
|
||||
}
|
||||
} else {
|
||||
// No callback, use regular Stream
|
||||
result = llmAPI.Stream(connector, messages, opts)
|
||||
}
|
||||
|
||||
// Convert result to JS value
|
||||
jsVal, err := bridge.JsValue(v8ctx, result)
|
||||
if err != nil {
|
||||
return bridge.JsException(v8ctx, "failed to convert result: "+err.Error())
|
||||
}
|
||||
|
||||
return jsVal
|
||||
})
|
||||
}
|
||||
|
||||
// llmAllMethod implements ctx.llm.All(requests, options?)
|
||||
// Usage: const results = ctx.llm.All([
|
||||
//
|
||||
// { connector: "gpt-4o", messages: [...], options: {...} },
|
||||
// { connector: "claude-3", messages: [...] }
|
||||
//
|
||||
// ], { onChunk: (connectorID, index, msg) => 0 })
|
||||
// Returns: [{ connector, response, content, error }, ...]
|
||||
func (ctx *Context) llmAllMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
||||
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
|
||||
return ctx.executeLlmBatchMethod(info, LlmBatchMethodAll)
|
||||
})
|
||||
}
|
||||
|
||||
// llmAnyMethod implements ctx.llm.Any(requests, options?)
|
||||
// Returns first successful result
|
||||
func (ctx *Context) llmAnyMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
||||
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
|
||||
return ctx.executeLlmBatchMethod(info, LlmBatchMethodAny)
|
||||
})
|
||||
}
|
||||
|
||||
// llmRaceMethod implements ctx.llm.Race(requests, options?)
|
||||
// Returns first completed result (success or failure)
|
||||
func (ctx *Context) llmRaceMethod(iso *v8go.Isolate) *v8go.FunctionTemplate {
|
||||
return v8go.NewFunctionTemplate(iso, func(info *v8go.FunctionCallbackInfo) *v8go.Value {
|
||||
return ctx.executeLlmBatchMethod(info, LlmBatchMethodRace)
|
||||
})
|
||||
}
|
||||
|
||||
// LlmBatchMethod represents the type of batch LLM operation
|
||||
type LlmBatchMethod int
|
||||
|
||||
const (
|
||||
LlmBatchMethodAll LlmBatchMethod = iota
|
||||
LlmBatchMethodAny
|
||||
LlmBatchMethodRace
|
||||
)
|
||||
|
||||
// executeLlmBatchMethod handles All/Any/Race batch LLM calls
|
||||
func (ctx *Context) executeLlmBatchMethod(info *v8go.FunctionCallbackInfo, method LlmBatchMethod) *v8go.Value {
|
||||
v8ctx := info.Context()
|
||||
args := info.Args()
|
||||
|
||||
// Validate arguments
|
||||
if len(args) < 1 {
|
||||
return bridge.JsException(v8ctx, "requires requests array parameter")
|
||||
}
|
||||
|
||||
// Parse requests array (first argument)
|
||||
requestsVal, err := bridge.GoValue(args[0], v8ctx)
|
||||
if err != nil {
|
||||
return bridge.JsException(v8ctx, "invalid requests: "+err.Error())
|
||||
}
|
||||
requests, ok := requestsVal.([]interface{})
|
||||
if !ok {
|
||||
return bridge.JsException(v8ctx, "requests must be an array")
|
||||
}
|
||||
|
||||
// Get LLM API
|
||||
llmAPI := ctx.Llm()
|
||||
if llmAPI == nil {
|
||||
return bridge.JsException(v8ctx, "LLM API not available")
|
||||
}
|
||||
|
||||
// Parse optional global callback from second argument (options object)
|
||||
var globalCallback *v8go.Function
|
||||
if len(args) >= 2 && !args[1].IsUndefined() && !args[1].IsNull() {
|
||||
optsObj, err := args[1].AsObject()
|
||||
if err == nil && optsObj != nil {
|
||||
onChunkVal, _ := optsObj.Get("onChunk")
|
||||
if onChunkVal != nil && onChunkVal.IsFunction() {
|
||||
globalCallback, _ = onChunkVal.AsFunction()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var results []interface{}
|
||||
|
||||
// If callback is provided and API supports it, use channel-based execution
|
||||
if globalCallback != nil {
|
||||
if apiWithCb, ok := llmAPI.(LlmAPIWithCallback); ok {
|
||||
results = ctx.executeLlmBatchWithCallback(method, requests, globalCallback, v8ctx, apiWithCb)
|
||||
} else {
|
||||
// Fallback: execute without callback
|
||||
results = ctx.executeLlmBatchWithoutCallback(method, requests, llmAPI)
|
||||
}
|
||||
} else {
|
||||
// No callback, use regular batch methods
|
||||
results = ctx.executeLlmBatchWithoutCallback(method, requests, llmAPI)
|
||||
}
|
||||
|
||||
// Convert results to JS value
|
||||
jsVal, err := bridge.JsValue(v8ctx, results)
|
||||
if err != nil {
|
||||
return bridge.JsException(v8ctx, "failed to convert results: "+err.Error())
|
||||
}
|
||||
|
||||
return jsVal
|
||||
}
|
||||
|
||||
// executeLlmBatchWithoutCallback executes batch LLM calls without callback
|
||||
func (ctx *Context) executeLlmBatchWithoutCallback(method LlmBatchMethod, requests []interface{}, llmAPI LlmAPI) []interface{} {
|
||||
switch method {
|
||||
case LlmBatchMethodAll:
|
||||
return llmAPI.All(requests)
|
||||
case LlmBatchMethodAny:
|
||||
return llmAPI.Any(requests)
|
||||
case LlmBatchMethodRace:
|
||||
return llmAPI.Race(requests)
|
||||
default:
|
||||
return llmAPI.All(requests)
|
||||
}
|
||||
}
|
||||
|
||||
// llmBatchMessage is used for channel communication in batch LLM calls
|
||||
type llmBatchMessage struct {
|
||||
ConnectorID string
|
||||
Index int
|
||||
Message *message.Message
|
||||
}
|
||||
|
||||
// executeLlmBatchWithCallback executes batch LLM calls with callback using channel
|
||||
// This ensures V8 thread safety by serializing callback invocations
|
||||
func (ctx *Context) executeLlmBatchWithCallback(method LlmBatchMethod, requests []interface{}, callback *v8go.Function, v8ctx *v8go.Context, apiWithCb LlmAPIWithCallback) []interface{} {
|
||||
// Create a buffered channel for messages
|
||||
// Using blocking send to ensure all messages are delivered
|
||||
msgChan := make(chan llmBatchMessage, 1000)
|
||||
doneChan := make(chan []interface{}, 1)
|
||||
|
||||
// Create Go handler that sends to channel
|
||||
goHandler := func(connectorID string, index int, msg *message.Message) int {
|
||||
msgChan <- llmBatchMessage{
|
||||
ConnectorID: connectorID,
|
||||
Index: index,
|
||||
Message: msg,
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// Execute batch calls in background goroutine
|
||||
go func() {
|
||||
defer close(msgChan)
|
||||
|
||||
var results []interface{}
|
||||
switch method {
|
||||
case LlmBatchMethodAll:
|
||||
results = apiWithCb.AllWithHandler(requests, goHandler)
|
||||
case LlmBatchMethodAny:
|
||||
results = apiWithCb.AnyWithHandler(requests, goHandler)
|
||||
case LlmBatchMethodRace:
|
||||
results = apiWithCb.RaceWithHandler(requests, goHandler)
|
||||
default:
|
||||
results = apiWithCb.AllWithHandler(requests, goHandler)
|
||||
}
|
||||
doneChan <- results
|
||||
}()
|
||||
|
||||
// Process messages in main goroutine (V8 thread)
|
||||
for msg := range msgChan {
|
||||
callJSLlmBatchCallback(v8ctx, callback, msg.ConnectorID, msg.Index, msg.Message)
|
||||
}
|
||||
|
||||
// Wait for results
|
||||
return <-doneChan
|
||||
}
|
||||
|
||||
// callJSLlmBatchCallback calls the JS callback function for batch LLM calls
|
||||
func callJSLlmBatchCallback(v8ctx *v8go.Context, callback *v8go.Function, connectorID string, index int, msg *message.Message) {
|
||||
if callback == nil || v8ctx == nil {
|
||||
return
|
||||
}
|
||||
|
||||
iso := v8ctx.Isolate()
|
||||
|
||||
// Create arguments: connectorID, index, message
|
||||
connectorVal, err := v8go.NewValue(iso, connectorID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
indexVal, err := v8go.NewValue(iso, int32(index))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Convert message to JS object
|
||||
msgVal, err := bridge.JsValue(v8ctx, msg)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Call the callback
|
||||
_, _ = callback.Call(v8go.Undefined(iso), connectorVal, indexVal, msgVal)
|
||||
}
|
||||
497
agent/context/jsapi_llm_v8_test.go
Normal file
497
agent/context/jsapi_llm_v8_test.go
Normal file
|
|
@ -0,0 +1,497 @@
|
|||
package context_test
|
||||
|
||||
import (
|
||||
stdContext "context"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
v8 "github.com/yaoapp/gou/runtime/v8"
|
||||
"github.com/yaoapp/yao/agent/context"
|
||||
"github.com/yaoapp/yao/agent/testutils"
|
||||
"github.com/yaoapp/yao/openapi/oauth/types"
|
||||
|
||||
// Import assistant package to register LlmAPIFactory
|
||||
_ "github.com/yaoapp/yao/agent/assistant"
|
||||
)
|
||||
|
||||
// TestLlm_Stream_V8 tests basic ctx.llm.Stream functionality with real V8 execution
|
||||
func TestLlm_Stream_V8(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping integration test in short mode")
|
||||
}
|
||||
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
// Create authorized info for the context
|
||||
authorized := &types.AuthorizedInfo{
|
||||
Subject: "test-user",
|
||||
UserID: "test-123",
|
||||
TenantID: "test-tenant",
|
||||
}
|
||||
|
||||
// Create a context
|
||||
ctx := context.New(stdContext.Background(), authorized, "test-chat-llm-stream")
|
||||
ctx.AssistantID = "tests.simple-greeting"
|
||||
defer ctx.Release()
|
||||
|
||||
// Test basic Stream call with real connector
|
||||
res, err := v8.Call(v8.CallOptions{}, `
|
||||
function test(ctx) {
|
||||
try {
|
||||
const result = ctx.llm.Stream("gpt-4o-mini", [
|
||||
{ role: "user", content: "Say hello in one word" }
|
||||
], {
|
||||
temperature: 0.1,
|
||||
max_tokens: 10
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
connector: result.connector,
|
||||
has_content: result.content && result.content.length > 0,
|
||||
has_response: result.response !== undefined,
|
||||
error: result.error || ""
|
||||
};
|
||||
} catch (error) {
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}`, ctx)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, res)
|
||||
|
||||
result, ok := res.(map[string]interface{})
|
||||
require.True(t, ok, "result should be a map")
|
||||
|
||||
success, _ := result["success"].(bool)
|
||||
if !success {
|
||||
t.Logf("Test result: %v", result)
|
||||
}
|
||||
require.True(t, success, "Test should succeed, error: %v", result["error"])
|
||||
|
||||
assert.Equal(t, "gpt-4o-mini", result["connector"])
|
||||
|
||||
hasContent, _ := result["has_content"].(bool)
|
||||
assert.True(t, hasContent, "Should have content in response")
|
||||
|
||||
hasResponse, _ := result["has_response"].(bool)
|
||||
assert.True(t, hasResponse, "Should have response object")
|
||||
}
|
||||
|
||||
// TestLlm_Stream_WithCallback_V8 tests ctx.llm.Stream with onChunk callback
|
||||
func TestLlm_Stream_WithCallback_V8(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping integration test in short mode")
|
||||
}
|
||||
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
authorized := &types.AuthorizedInfo{
|
||||
Subject: "test-user",
|
||||
UserID: "test-123",
|
||||
TenantID: "test-tenant",
|
||||
}
|
||||
|
||||
ctx := context.New(stdContext.Background(), authorized, "test-chat-llm-callback")
|
||||
ctx.AssistantID = "tests.simple-greeting"
|
||||
defer ctx.Release()
|
||||
|
||||
// Test Stream call with callback
|
||||
res, err := v8.Call(v8.CallOptions{}, `
|
||||
function test(ctx) {
|
||||
try {
|
||||
let callbackCount = 0;
|
||||
let receivedTypes = [];
|
||||
|
||||
const result = ctx.llm.Stream("gpt-4o-mini", [
|
||||
{ role: "user", content: "Say hi" }
|
||||
], {
|
||||
temperature: 0.1,
|
||||
max_tokens: 10,
|
||||
onChunk: function(msg) {
|
||||
callbackCount++;
|
||||
if (msg && msg.type) {
|
||||
receivedTypes.push(msg.type);
|
||||
}
|
||||
return 0; // Continue
|
||||
}
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
connector: result.connector,
|
||||
callbackCount: callbackCount,
|
||||
receivedTypes: receivedTypes,
|
||||
has_content: result.content && result.content.length > 0,
|
||||
error: result.error || ""
|
||||
};
|
||||
} catch (error) {
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}`, ctx)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, res)
|
||||
|
||||
result, ok := res.(map[string]interface{})
|
||||
require.True(t, ok, "result should be a map")
|
||||
|
||||
success, _ := result["success"].(bool)
|
||||
if !success {
|
||||
t.Logf("Test result: %v", result)
|
||||
}
|
||||
require.True(t, success, "Test should succeed, error: %v", result["error"])
|
||||
|
||||
// Callback should have been called at least once
|
||||
callbackCount, _ := result["callbackCount"].(float64)
|
||||
assert.Greater(t, callbackCount, float64(0), "Callback should be called at least once")
|
||||
}
|
||||
|
||||
// TestLlm_All_V8 tests ctx.llm.All with multiple connectors
|
||||
func TestLlm_All_V8(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping integration test in short mode")
|
||||
}
|
||||
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
authorized := &types.AuthorizedInfo{
|
||||
Subject: "test-user",
|
||||
UserID: "test-123",
|
||||
TenantID: "test-tenant",
|
||||
}
|
||||
|
||||
ctx := context.New(stdContext.Background(), authorized, "test-chat-llm-all")
|
||||
ctx.AssistantID = "tests.simple-greeting"
|
||||
defer ctx.Release()
|
||||
|
||||
// Test All with multiple requests to same connector (different prompts)
|
||||
res, err := v8.Call(v8.CallOptions{}, `
|
||||
function test(ctx) {
|
||||
try {
|
||||
const results = ctx.llm.All([
|
||||
{
|
||||
connector: "gpt-4o-mini",
|
||||
messages: [{ role: "user", content: "Say 'one'" }],
|
||||
options: { temperature: 0.1, max_tokens: 5 }
|
||||
},
|
||||
{
|
||||
connector: "gpt-4o-mini",
|
||||
messages: [{ role: "user", content: "Say 'two'" }],
|
||||
options: { temperature: 0.1, max_tokens: 5 }
|
||||
}
|
||||
]);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
count: results.length,
|
||||
results: results.map(r => ({
|
||||
connector: r.connector,
|
||||
has_content: r.content && r.content.length > 0,
|
||||
error: r.error || ""
|
||||
}))
|
||||
};
|
||||
} catch (error) {
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}`, ctx)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, res)
|
||||
|
||||
result, ok := res.(map[string]interface{})
|
||||
require.True(t, ok, "result should be a map")
|
||||
|
||||
success, _ := result["success"].(bool)
|
||||
if !success {
|
||||
t.Logf("Test result: %v", result)
|
||||
}
|
||||
require.True(t, success, "Test should succeed, error: %v", result["error"])
|
||||
|
||||
// Should have 2 results
|
||||
count, _ := result["count"].(float64)
|
||||
assert.Equal(t, float64(2), count, "Should have 2 results")
|
||||
|
||||
// Check individual results
|
||||
results, _ := result["results"].([]interface{})
|
||||
require.Len(t, results, 2)
|
||||
|
||||
for i, r := range results {
|
||||
rMap, _ := r.(map[string]interface{})
|
||||
hasContent, _ := rMap["has_content"].(bool)
|
||||
assert.True(t, hasContent, "Result %d should have content", i)
|
||||
errorStr, _ := rMap["error"].(string)
|
||||
assert.Empty(t, errorStr, "Result %d should not have error", i)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLlm_All_WithCallback_V8 tests ctx.llm.All with global callback
|
||||
func TestLlm_All_WithCallback_V8(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping integration test in short mode")
|
||||
}
|
||||
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
authorized := &types.AuthorizedInfo{
|
||||
Subject: "test-user",
|
||||
UserID: "test-123",
|
||||
TenantID: "test-tenant",
|
||||
}
|
||||
|
||||
ctx := context.New(stdContext.Background(), authorized, "test-chat-llm-all-callback")
|
||||
ctx.AssistantID = "tests.simple-greeting"
|
||||
defer ctx.Release()
|
||||
|
||||
// Test All with global callback
|
||||
res, err := v8.Call(v8.CallOptions{}, `
|
||||
function test(ctx) {
|
||||
try {
|
||||
let callbackCount = 0;
|
||||
let indexesSeen = new Set();
|
||||
|
||||
const results = ctx.llm.All([
|
||||
{
|
||||
connector: "gpt-4o-mini",
|
||||
messages: [{ role: "user", content: "Say 'A'" }],
|
||||
options: { temperature: 0.1, max_tokens: 5 }
|
||||
},
|
||||
{
|
||||
connector: "gpt-4o-mini",
|
||||
messages: [{ role: "user", content: "Say 'B'" }],
|
||||
options: { temperature: 0.1, max_tokens: 5 }
|
||||
}
|
||||
], {
|
||||
onChunk: function(connectorID, index, msg) {
|
||||
callbackCount++;
|
||||
indexesSeen.add(index);
|
||||
return 0;
|
||||
}
|
||||
});
|
||||
|
||||
return {
|
||||
success: true,
|
||||
count: results.length,
|
||||
callbackCount: callbackCount,
|
||||
indexesSeen: indexesSeen.size
|
||||
};
|
||||
} catch (error) {
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}`, ctx)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, res)
|
||||
|
||||
result, ok := res.(map[string]interface{})
|
||||
require.True(t, ok, "result should be a map")
|
||||
|
||||
success, _ := result["success"].(bool)
|
||||
if !success {
|
||||
t.Logf("Test result: %v", result)
|
||||
}
|
||||
require.True(t, success, "Test should succeed, error: %v", result["error"])
|
||||
|
||||
// Callback should have been called
|
||||
callbackCount, _ := result["callbackCount"].(float64)
|
||||
assert.Greater(t, callbackCount, float64(0), "Callback should be called")
|
||||
|
||||
// Should have seen at least one index (both requests may complete so fast that only one is tracked)
|
||||
// Note: Due to V8 thread safety with channel-based approach, callbacks are serialized
|
||||
indexesSeen, _ := result["indexesSeen"].(float64)
|
||||
assert.GreaterOrEqual(t, indexesSeen, float64(1), "Should have seen callbacks from at least one request")
|
||||
}
|
||||
|
||||
// TestLlm_Any_V8 tests ctx.llm.Any - returns first success
|
||||
func TestLlm_Any_V8(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping integration test in short mode")
|
||||
}
|
||||
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
authorized := &types.AuthorizedInfo{
|
||||
Subject: "test-user",
|
||||
UserID: "test-123",
|
||||
TenantID: "test-tenant",
|
||||
}
|
||||
|
||||
ctx := context.New(stdContext.Background(), authorized, "test-chat-llm-any")
|
||||
ctx.AssistantID = "tests.simple-greeting"
|
||||
defer ctx.Release()
|
||||
|
||||
// Test Any - should return first successful result
|
||||
res, err := v8.Call(v8.CallOptions{}, `
|
||||
function test(ctx) {
|
||||
try {
|
||||
const results = ctx.llm.Any([
|
||||
{
|
||||
connector: "gpt-4o-mini",
|
||||
messages: [{ role: "user", content: "Say 'hello'" }],
|
||||
options: { temperature: 0.1, max_tokens: 5 }
|
||||
},
|
||||
{
|
||||
connector: "gpt-4o-mini",
|
||||
messages: [{ role: "user", content: "Say 'world'" }],
|
||||
options: { temperature: 0.1, max_tokens: 5 }
|
||||
}
|
||||
]);
|
||||
|
||||
// Any returns array with single successful result
|
||||
return {
|
||||
success: true,
|
||||
count: results.length,
|
||||
first_has_content: results[0] && results[0].content && results[0].content.length > 0,
|
||||
first_error: results[0] ? (results[0].error || "") : "no result"
|
||||
};
|
||||
} catch (error) {
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}`, ctx)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, res)
|
||||
|
||||
result, ok := res.(map[string]interface{})
|
||||
require.True(t, ok, "result should be a map")
|
||||
|
||||
success, _ := result["success"].(bool)
|
||||
if !success {
|
||||
t.Logf("Test result: %v", result)
|
||||
}
|
||||
require.True(t, success, "Test should succeed, error: %v", result["error"])
|
||||
|
||||
// Any returns single result on success
|
||||
count, _ := result["count"].(float64)
|
||||
assert.Equal(t, float64(1), count, "Should have 1 result (first success)")
|
||||
|
||||
firstHasContent, _ := result["first_has_content"].(bool)
|
||||
assert.True(t, firstHasContent, "First result should have content")
|
||||
|
||||
firstError, _ := result["first_error"].(string)
|
||||
assert.Empty(t, firstError, "First result should not have error")
|
||||
}
|
||||
|
||||
// TestLlm_Race_V8 tests ctx.llm.Race - returns first completion
|
||||
func TestLlm_Race_V8(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping integration test in short mode")
|
||||
}
|
||||
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
authorized := &types.AuthorizedInfo{
|
||||
Subject: "test-user",
|
||||
UserID: "test-123",
|
||||
TenantID: "test-tenant",
|
||||
}
|
||||
|
||||
ctx := context.New(stdContext.Background(), authorized, "test-chat-llm-race")
|
||||
ctx.AssistantID = "tests.simple-greeting"
|
||||
defer ctx.Release()
|
||||
|
||||
// Test Race - should return first completed result
|
||||
res, err := v8.Call(v8.CallOptions{}, `
|
||||
function test(ctx) {
|
||||
try {
|
||||
const results = ctx.llm.Race([
|
||||
{
|
||||
connector: "gpt-4o-mini",
|
||||
messages: [{ role: "user", content: "Say 'fast'" }],
|
||||
options: { temperature: 0.1, max_tokens: 5 }
|
||||
},
|
||||
{
|
||||
connector: "gpt-4o-mini",
|
||||
messages: [{ role: "user", content: "Say 'slow'" }],
|
||||
options: { temperature: 0.1, max_tokens: 5 }
|
||||
}
|
||||
]);
|
||||
|
||||
// Race returns array with single result (first to complete)
|
||||
return {
|
||||
success: true,
|
||||
count: results.length,
|
||||
has_result: results[0] !== undefined,
|
||||
first_connector: results[0] ? results[0].connector : "",
|
||||
first_has_content: results[0] && results[0].content && results[0].content.length > 0
|
||||
};
|
||||
} catch (error) {
|
||||
return { success: false, error: error.message };
|
||||
}
|
||||
}`, ctx)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, res)
|
||||
|
||||
result, ok := res.(map[string]interface{})
|
||||
require.True(t, ok, "result should be a map")
|
||||
|
||||
success, _ := result["success"].(bool)
|
||||
if !success {
|
||||
t.Logf("Test result: %v", result)
|
||||
}
|
||||
require.True(t, success, "Test should succeed, error: %v", result["error"])
|
||||
|
||||
// Race returns single result
|
||||
count, _ := result["count"].(float64)
|
||||
assert.Equal(t, float64(1), count, "Should have 1 result (first to complete)")
|
||||
|
||||
hasResult, _ := result["has_result"].(bool)
|
||||
assert.True(t, hasResult, "Should have a result")
|
||||
|
||||
firstConnector, _ := result["first_connector"].(string)
|
||||
assert.Equal(t, "gpt-4o-mini", firstConnector, "First result should have connector")
|
||||
}
|
||||
|
||||
// TestLlm_Stream_InvalidConnector_V8 tests error handling for invalid connector
|
||||
func TestLlm_Stream_InvalidConnector_V8(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping integration test in short mode")
|
||||
}
|
||||
|
||||
testutils.Prepare(t)
|
||||
defer testutils.Clean(t)
|
||||
|
||||
authorized := &types.AuthorizedInfo{
|
||||
Subject: "test-user",
|
||||
UserID: "test-123",
|
||||
TenantID: "test-tenant",
|
||||
}
|
||||
|
||||
ctx := context.New(stdContext.Background(), authorized, "test-chat-llm-invalid")
|
||||
ctx.AssistantID = "tests.simple-greeting"
|
||||
defer ctx.Release()
|
||||
|
||||
// Test Stream with invalid connector
|
||||
res, err := v8.Call(v8.CallOptions{}, `
|
||||
function test(ctx) {
|
||||
try {
|
||||
const result = ctx.llm.Stream("invalid-connector-that-does-not-exist", [
|
||||
{ role: "user", content: "Hello" }
|
||||
]);
|
||||
|
||||
return {
|
||||
has_error: result.error && result.error.length > 0,
|
||||
error: result.error || ""
|
||||
};
|
||||
} catch (error) {
|
||||
return { has_error: true, error: error.message };
|
||||
}
|
||||
}`, ctx)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, res)
|
||||
|
||||
result, ok := res.(map[string]interface{})
|
||||
require.True(t, ok, "result should be a map")
|
||||
|
||||
hasError, _ := result["has_error"].(bool)
|
||||
assert.True(t, hasError, "Should have error for invalid connector")
|
||||
}
|
||||
737
agent/llm/jsapi.go
Normal file
737
agent/llm/jsapi.go
Normal file
|
|
@ -0,0 +1,737 @@
|
|||
// Package llm provides the LLM JSAPI implementation
|
||||
package llm
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/yaoapp/gou/connector"
|
||||
agentContext "github.com/yaoapp/yao/agent/context"
|
||||
"github.com/yaoapp/yao/agent/output/message"
|
||||
)
|
||||
|
||||
// JSAPI implements LlmAPI interface for ctx.llm.* methods
|
||||
type JSAPI struct {
|
||||
ctx *agentContext.Context
|
||||
}
|
||||
|
||||
// Ensure JSAPI implements both interfaces
|
||||
var _ agentContext.LlmAPI = (*JSAPI)(nil)
|
||||
var _ agentContext.LlmAPIWithCallback = (*JSAPI)(nil)
|
||||
|
||||
// NewJSAPI creates a new JSAPI for the given context
|
||||
func NewJSAPI(ctx *agentContext.Context) *JSAPI {
|
||||
return &JSAPI{ctx: ctx}
|
||||
}
|
||||
|
||||
// SetJSAPIFactory registers the JSAPI factory with the context package
|
||||
// This should be called during initialization
|
||||
func SetJSAPIFactory() {
|
||||
agentContext.LlmAPIFactory = func(ctx *agentContext.Context) agentContext.LlmAPI {
|
||||
return NewJSAPI(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
// Stream implements LlmAPI.Stream - calls LLM with streaming output to ctx.Writer
|
||||
func (api *JSAPI) Stream(connectorID string, messages []interface{}, opts map[string]interface{}) interface{} {
|
||||
return api.StreamWithHandler(connectorID, messages, opts, nil)
|
||||
}
|
||||
|
||||
// StreamWithHandler implements LlmAPIWithCallback.StreamWithHandler - calls LLM with OnMessage handler
|
||||
func (api *JSAPI) StreamWithHandler(connectorID string, messages []interface{}, opts map[string]interface{}, handler agentContext.OnMessageFunc) interface{} {
|
||||
result := &Result{
|
||||
Connector: connectorID,
|
||||
}
|
||||
|
||||
// Validate context
|
||||
if api.ctx == nil {
|
||||
result.Error = "context is nil"
|
||||
return result
|
||||
}
|
||||
|
||||
// Get connector
|
||||
conn, err := connector.Select(connectorID)
|
||||
if err != nil {
|
||||
result.Error = fmt.Sprintf("failed to select connector %s: %v", connectorID, err)
|
||||
return result
|
||||
}
|
||||
|
||||
// Parse messages to context.Message format
|
||||
ctxMessages, err := parseMessages(messages)
|
||||
if err != nil {
|
||||
result.Error = fmt.Sprintf("failed to parse messages: %v", err)
|
||||
return result
|
||||
}
|
||||
|
||||
// Build CompletionOptions from opts
|
||||
completionOptions := buildCompletionOptions(conn, opts)
|
||||
|
||||
// Create LLM instance
|
||||
llmInstance, err := New(conn, completionOptions)
|
||||
if err != nil {
|
||||
result.Error = fmt.Sprintf("failed to create LLM instance: %v", err)
|
||||
return result
|
||||
}
|
||||
|
||||
// Create stream handler with the provided callback
|
||||
// Note: We pass handler directly to the stream handler instead of setting ctx.Stack.Options.OnMessage
|
||||
// This avoids race conditions in concurrent batch calls where multiple goroutines
|
||||
// would otherwise overwrite the same ctx.Stack.Options.OnMessage
|
||||
streamHandler := createStreamHandlerWithCallback(api.ctx, handler)
|
||||
|
||||
// Execute LLM stream call
|
||||
response, err := llmInstance.Stream(api.ctx, ctxMessages, completionOptions, streamHandler)
|
||||
if err != nil {
|
||||
result.Error = fmt.Sprintf("LLM stream failed: %v", err)
|
||||
return result
|
||||
}
|
||||
|
||||
// Set response
|
||||
result.Response = response
|
||||
|
||||
// Extract text content from response
|
||||
if response != nil {
|
||||
result.Content = extractContent(response)
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// createStreamHandlerWithCallback creates a stream handler that uses the provided callback directly
|
||||
// This is used instead of setting ctx.Stack.Options.OnMessage to avoid race conditions
|
||||
// in concurrent batch calls
|
||||
func createStreamHandlerWithCallback(ctx *agentContext.Context, handler agentContext.OnMessageFunc) message.StreamFunc {
|
||||
// Handle nil context
|
||||
if ctx == nil {
|
||||
return func(chunkType message.StreamChunkType, data []byte) int {
|
||||
return 0 // No-op handler when context is nil
|
||||
}
|
||||
}
|
||||
|
||||
// Stream state for tracking message groups
|
||||
state := &streamState{
|
||||
ctx: ctx,
|
||||
buffer: []byte{},
|
||||
handler: handler, // Store the handler directly in state
|
||||
}
|
||||
|
||||
return func(chunkType message.StreamChunkType, data []byte) int {
|
||||
return state.handleChunk(chunkType, data)
|
||||
}
|
||||
}
|
||||
|
||||
// parseMessages converts JS message array to context.Message slice
|
||||
func parseMessages(messages []interface{}) ([]agentContext.Message, error) {
|
||||
result := make([]agentContext.Message, 0, len(messages))
|
||||
|
||||
for i, msg := range messages {
|
||||
msgMap, ok := msg.(map[string]interface{})
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("message %d is not an object", i)
|
||||
}
|
||||
|
||||
ctxMsg := agentContext.Message{}
|
||||
|
||||
// Required: role
|
||||
if role, ok := msgMap["role"].(string); ok {
|
||||
ctxMsg.Role = agentContext.MessageRole(role)
|
||||
} else {
|
||||
return nil, fmt.Errorf("message %d missing role", i)
|
||||
}
|
||||
|
||||
// Optional: content (can be string or array for multimodal)
|
||||
if content, ok := msgMap["content"]; ok {
|
||||
ctxMsg.Content = content
|
||||
}
|
||||
|
||||
// Optional: name
|
||||
if name, ok := msgMap["name"].(string); ok {
|
||||
ctxMsg.Name = &name
|
||||
}
|
||||
|
||||
// Optional: tool_calls
|
||||
if toolCalls, ok := msgMap["tool_calls"]; ok {
|
||||
if tcArray, ok := toolCalls.([]interface{}); ok {
|
||||
ctxMsg.ToolCalls = parseToolCalls(tcArray)
|
||||
}
|
||||
}
|
||||
|
||||
// Optional: tool_call_id (for tool response messages)
|
||||
if toolCallID, ok := msgMap["tool_call_id"].(string); ok {
|
||||
ctxMsg.ToolCallID = &toolCallID
|
||||
}
|
||||
|
||||
result = append(result, ctxMsg)
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// parseToolCalls converts JS tool_calls array to context.ToolCall slice
|
||||
func parseToolCalls(toolCalls []interface{}) []agentContext.ToolCall {
|
||||
result := make([]agentContext.ToolCall, 0, len(toolCalls))
|
||||
|
||||
for _, tc := range toolCalls {
|
||||
tcMap, ok := tc.(map[string]interface{})
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
toolCall := agentContext.ToolCall{}
|
||||
|
||||
if id, ok := tcMap["id"].(string); ok {
|
||||
toolCall.ID = id
|
||||
}
|
||||
if typ, ok := tcMap["type"].(string); ok {
|
||||
toolCall.Type = agentContext.ToolCallType(typ)
|
||||
}
|
||||
if fn, ok := tcMap["function"].(map[string]interface{}); ok {
|
||||
toolCall.Function = agentContext.Function{}
|
||||
if name, ok := fn["name"].(string); ok {
|
||||
toolCall.Function.Name = name
|
||||
}
|
||||
if args, ok := fn["arguments"].(string); ok {
|
||||
toolCall.Function.Arguments = args
|
||||
}
|
||||
}
|
||||
|
||||
result = append(result, toolCall)
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// buildCompletionOptions creates CompletionOptions from JS opts map
|
||||
func buildCompletionOptions(conn connector.Connector, opts map[string]interface{}) *agentContext.CompletionOptions {
|
||||
// Get capabilities from connector
|
||||
capabilities := GetCapabilitiesFromConn(conn, nil)
|
||||
|
||||
completionOptions := &agentContext.CompletionOptions{
|
||||
Capabilities: capabilities,
|
||||
}
|
||||
|
||||
if opts == nil {
|
||||
return completionOptions
|
||||
}
|
||||
|
||||
// Temperature
|
||||
if temp, ok := opts["temperature"].(float64); ok {
|
||||
completionOptions.Temperature = &temp
|
||||
}
|
||||
|
||||
// Max tokens
|
||||
if maxTokens, ok := opts["max_tokens"].(float64); ok {
|
||||
mt := int(maxTokens)
|
||||
completionOptions.MaxTokens = &mt
|
||||
}
|
||||
if maxCompletionTokens, ok := opts["max_completion_tokens"].(float64); ok {
|
||||
mct := int(maxCompletionTokens)
|
||||
completionOptions.MaxCompletionTokens = &mct
|
||||
}
|
||||
|
||||
// Top P
|
||||
if topP, ok := opts["top_p"].(float64); ok {
|
||||
completionOptions.TopP = &topP
|
||||
}
|
||||
|
||||
// Presence penalty
|
||||
if presencePenalty, ok := opts["presence_penalty"].(float64); ok {
|
||||
completionOptions.PresencePenalty = &presencePenalty
|
||||
}
|
||||
|
||||
// Frequency penalty
|
||||
if frequencyPenalty, ok := opts["frequency_penalty"].(float64); ok {
|
||||
completionOptions.FrequencyPenalty = &frequencyPenalty
|
||||
}
|
||||
|
||||
// Stop sequences
|
||||
if stop, ok := opts["stop"]; ok {
|
||||
completionOptions.Stop = stop
|
||||
}
|
||||
|
||||
// User
|
||||
if user, ok := opts["user"].(string); ok {
|
||||
completionOptions.User = user
|
||||
}
|
||||
|
||||
// Seed
|
||||
if seed, ok := opts["seed"].(float64); ok {
|
||||
s := int(seed)
|
||||
completionOptions.Seed = &s
|
||||
}
|
||||
|
||||
// Tools
|
||||
if tools, ok := opts["tools"].([]interface{}); ok {
|
||||
completionOptions.Tools = make([]map[string]interface{}, 0, len(tools))
|
||||
for _, tool := range tools {
|
||||
if toolMap, ok := tool.(map[string]interface{}); ok {
|
||||
completionOptions.Tools = append(completionOptions.Tools, toolMap)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Tool choice
|
||||
if toolChoice, ok := opts["tool_choice"]; ok {
|
||||
completionOptions.ToolChoice = toolChoice
|
||||
}
|
||||
|
||||
// Response format
|
||||
if responseFormat, ok := opts["response_format"].(map[string]interface{}); ok {
|
||||
rf := &agentContext.ResponseFormat{}
|
||||
if rfType, ok := responseFormat["type"].(string); ok {
|
||||
rf.Type = agentContext.ResponseFormatType(rfType)
|
||||
}
|
||||
if jsonSchema, ok := responseFormat["json_schema"].(map[string]interface{}); ok {
|
||||
rf.JSONSchema = &agentContext.JSONSchema{}
|
||||
if name, ok := jsonSchema["name"].(string); ok {
|
||||
rf.JSONSchema.Name = name
|
||||
}
|
||||
if desc, ok := jsonSchema["description"].(string); ok {
|
||||
rf.JSONSchema.Description = desc
|
||||
}
|
||||
if schema, ok := jsonSchema["schema"]; ok {
|
||||
rf.JSONSchema.Schema = schema
|
||||
}
|
||||
if strict, ok := jsonSchema["strict"].(bool); ok {
|
||||
rf.JSONSchema.Strict = &strict
|
||||
}
|
||||
}
|
||||
completionOptions.ResponseFormat = rf
|
||||
}
|
||||
|
||||
// Reasoning effort (for reasoning models)
|
||||
if reasoningEffort, ok := opts["reasoning_effort"].(string); ok {
|
||||
completionOptions.ReasoningEffort = &reasoningEffort
|
||||
}
|
||||
|
||||
return completionOptions
|
||||
}
|
||||
|
||||
// streamState manages stream handler state
|
||||
type streamState struct {
|
||||
ctx *agentContext.Context
|
||||
inMessage bool
|
||||
currentMsgID string
|
||||
currentMsgType string
|
||||
buffer []byte
|
||||
msgCounter int // Counter for generating message IDs when IDGenerator is nil
|
||||
chunkCounter int // Counter for generating chunk IDs when IDGenerator is nil
|
||||
handler agentContext.OnMessageFunc // Direct handler reference (avoids race condition via ctx.Stack.Options)
|
||||
}
|
||||
|
||||
// generateMessageID generates a unique message ID
|
||||
func (s *streamState) generateMessageID() string {
|
||||
if s.ctx != nil && s.ctx.IDGenerator != nil {
|
||||
return s.ctx.IDGenerator.GenerateMessageID()
|
||||
}
|
||||
s.msgCounter++
|
||||
return fmt.Sprintf("M%d", s.msgCounter)
|
||||
}
|
||||
|
||||
// generateChunkID generates a unique chunk ID
|
||||
func (s *streamState) generateChunkID() string {
|
||||
if s.ctx != nil && s.ctx.IDGenerator != nil {
|
||||
return s.ctx.IDGenerator.GenerateChunkID()
|
||||
}
|
||||
s.chunkCounter++
|
||||
return fmt.Sprintf("C%d", s.chunkCounter)
|
||||
}
|
||||
|
||||
// handleChunk processes a single stream chunk
|
||||
func (s *streamState) handleChunk(chunkType message.StreamChunkType, data []byte) int {
|
||||
switch chunkType {
|
||||
case message.ChunkMessageStart:
|
||||
s.inMessage = true
|
||||
s.currentMsgID = s.generateMessageID()
|
||||
s.buffer = []byte{}
|
||||
return 0
|
||||
|
||||
case message.ChunkText:
|
||||
if !s.inMessage {
|
||||
s.inMessage = true
|
||||
s.currentMsgID = s.generateMessageID()
|
||||
}
|
||||
s.currentMsgType = message.TypeText
|
||||
s.buffer = append(s.buffer, data...)
|
||||
|
||||
// Create message
|
||||
msg := &message.Message{
|
||||
ChunkID: s.generateChunkID(),
|
||||
MessageID: s.currentMsgID,
|
||||
Type: message.TypeText,
|
||||
Delta: true,
|
||||
Props: map[string]interface{}{
|
||||
"content": string(data),
|
||||
},
|
||||
}
|
||||
|
||||
// Call handler directly if provided (for batch calls and single calls with callback)
|
||||
// We use direct handler instead of ctx.Stack.Options.OnMessage to avoid race conditions
|
||||
// in concurrent batch calls where multiple goroutines would overwrite the shared OnMessage
|
||||
if s.handler != nil {
|
||||
if ret := s.handler(msg); ret != 0 {
|
||||
return ret
|
||||
}
|
||||
}
|
||||
|
||||
// Send to output for actual message delivery to client
|
||||
// Note: ctx.Send may also call ctx.Stack.Options.OnMessage if set (for agent calls),
|
||||
// but for LLM calls we don't set OnMessage, so no double callback occurs
|
||||
if err := s.ctx.Send(msg); err != nil {
|
||||
// Log error but continue streaming
|
||||
return 0
|
||||
}
|
||||
return 0
|
||||
|
||||
case message.ChunkThinking:
|
||||
if !s.inMessage {
|
||||
s.inMessage = true
|
||||
s.currentMsgID = s.generateMessageID()
|
||||
}
|
||||
s.currentMsgType = message.TypeThinking
|
||||
s.buffer = append(s.buffer, data...)
|
||||
|
||||
msg := &message.Message{
|
||||
ChunkID: s.generateChunkID(),
|
||||
MessageID: s.currentMsgID,
|
||||
Type: message.TypeThinking,
|
||||
Delta: true,
|
||||
Props: map[string]interface{}{
|
||||
"content": string(data),
|
||||
},
|
||||
}
|
||||
|
||||
// Call handler directly if provided
|
||||
if s.handler != nil {
|
||||
if ret := s.handler(msg); ret != 0 {
|
||||
return ret
|
||||
}
|
||||
}
|
||||
|
||||
if err := s.ctx.Send(msg); err != nil {
|
||||
return 0
|
||||
}
|
||||
return 0
|
||||
|
||||
case message.ChunkToolCall:
|
||||
if !s.inMessage {
|
||||
s.inMessage = true
|
||||
s.currentMsgID = s.generateMessageID()
|
||||
}
|
||||
s.currentMsgType = message.TypeToolCall
|
||||
s.buffer = append(s.buffer, data...)
|
||||
|
||||
// Tool call chunks are more complex - parse and forward
|
||||
msg := &message.Message{
|
||||
ChunkID: s.generateChunkID(),
|
||||
MessageID: s.currentMsgID,
|
||||
Type: message.TypeToolCall,
|
||||
Delta: true,
|
||||
Props: map[string]interface{}{
|
||||
"raw": string(data),
|
||||
},
|
||||
}
|
||||
|
||||
// Call handler directly if provided
|
||||
if s.handler != nil {
|
||||
if ret := s.handler(msg); ret != 0 {
|
||||
return ret
|
||||
}
|
||||
}
|
||||
|
||||
if err := s.ctx.Send(msg); err != nil {
|
||||
return 0
|
||||
}
|
||||
return 0
|
||||
|
||||
case message.ChunkMessageEnd:
|
||||
if s.inMessage {
|
||||
s.inMessage = false
|
||||
s.currentMsgID = ""
|
||||
s.buffer = []byte{}
|
||||
}
|
||||
return 0
|
||||
|
||||
case message.ChunkError:
|
||||
// Send error and stop
|
||||
msg := &message.Message{
|
||||
Type: message.TypeError,
|
||||
Props: map[string]interface{}{
|
||||
"error": string(data),
|
||||
},
|
||||
}
|
||||
|
||||
// Call handler directly if provided
|
||||
if s.handler != nil {
|
||||
s.handler(msg)
|
||||
}
|
||||
|
||||
_ = s.ctx.Send(msg) // Ignore error on error message
|
||||
return 1 // Stop on error
|
||||
|
||||
default:
|
||||
// Other chunk types (stream_start, stream_end, metadata) - ignore
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
// extractContent extracts text content from CompletionResponse
|
||||
func extractContent(response *agentContext.CompletionResponse) string {
|
||||
if response == nil || response.Content == nil {
|
||||
return ""
|
||||
}
|
||||
|
||||
switch content := response.Content.(type) {
|
||||
case string:
|
||||
return content
|
||||
case []interface{}:
|
||||
// Multimodal response - extract text parts
|
||||
var text string
|
||||
for _, part := range content {
|
||||
if partMap, ok := part.(map[string]interface{}); ok {
|
||||
if partMap["type"] == "text" {
|
||||
if t, ok := partMap["text"].(string); ok {
|
||||
text += t
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return text
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Batch LLM Methods: All, Any, Race
|
||||
// ============================================================================
|
||||
|
||||
// All executes all LLM requests concurrently and returns all results
|
||||
func (api *JSAPI) All(requests []interface{}) []interface{} {
|
||||
return api.AllWithHandler(requests, nil)
|
||||
}
|
||||
|
||||
// Any executes LLM requests concurrently and returns first successful result
|
||||
func (api *JSAPI) Any(requests []interface{}) []interface{} {
|
||||
return api.AnyWithHandler(requests, nil)
|
||||
}
|
||||
|
||||
// Race executes LLM requests concurrently and returns first completed result
|
||||
func (api *JSAPI) Race(requests []interface{}) []interface{} {
|
||||
return api.RaceWithHandler(requests, nil)
|
||||
}
|
||||
|
||||
// AllWithHandler executes all LLM requests with global handler
|
||||
func (api *JSAPI) AllWithHandler(requests []interface{}, globalHandler agentContext.LlmBatchOnMessageFunc) []interface{} {
|
||||
parsedRequests := api.parseRequests(requests, globalHandler)
|
||||
return api.executeAll(parsedRequests)
|
||||
}
|
||||
|
||||
// AnyWithHandler executes LLM requests and returns first success with handler
|
||||
func (api *JSAPI) AnyWithHandler(requests []interface{}, globalHandler agentContext.LlmBatchOnMessageFunc) []interface{} {
|
||||
parsedRequests := api.parseRequests(requests, globalHandler)
|
||||
return api.executeAny(parsedRequests)
|
||||
}
|
||||
|
||||
// RaceWithHandler executes LLM requests and returns first completion with handler
|
||||
func (api *JSAPI) RaceWithHandler(requests []interface{}, globalHandler agentContext.LlmBatchOnMessageFunc) []interface{} {
|
||||
parsedRequests := api.parseRequests(requests, globalHandler)
|
||||
return api.executeRace(parsedRequests)
|
||||
}
|
||||
|
||||
// parseRequests converts JS request array to internal Request slice
|
||||
func (api *JSAPI) parseRequests(requests []interface{}, globalHandler agentContext.LlmBatchOnMessageFunc) []*Request {
|
||||
result := make([]*Request, 0, len(requests))
|
||||
|
||||
for i, req := range requests {
|
||||
reqMap, ok := req.(map[string]interface{})
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
request := &Request{}
|
||||
|
||||
// Required: connector
|
||||
if connector, ok := reqMap["connector"].(string); ok {
|
||||
request.Connector = connector
|
||||
} else {
|
||||
continue // Skip invalid request
|
||||
}
|
||||
|
||||
// Required: messages
|
||||
if messages, ok := reqMap["messages"].([]interface{}); ok {
|
||||
request.Messages = messages
|
||||
} else {
|
||||
continue // Skip invalid request
|
||||
}
|
||||
|
||||
// Optional: options
|
||||
if options, ok := reqMap["options"].(map[string]interface{}); ok {
|
||||
// Remove onChunk from options if present (handled via globalHandler)
|
||||
delete(options, "onChunk")
|
||||
request.Options = options
|
||||
}
|
||||
|
||||
// Set handler based on globalHandler
|
||||
if globalHandler != nil {
|
||||
index := i
|
||||
connectorID := request.Connector
|
||||
request.Handler = func(msg *message.Message) int {
|
||||
return globalHandler(connectorID, index, msg)
|
||||
}
|
||||
}
|
||||
|
||||
result = append(result, request)
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
// executeAll executes all requests concurrently and waits for all to complete
|
||||
// Each request uses a forked context to avoid race conditions on shared state
|
||||
func (api *JSAPI) executeAll(requests []*Request) []interface{} {
|
||||
if len(requests) == 0 {
|
||||
return []interface{}{}
|
||||
}
|
||||
|
||||
results := make([]interface{}, len(requests))
|
||||
done := make(chan struct{})
|
||||
remaining := len(requests)
|
||||
|
||||
for i, req := range requests {
|
||||
go func(index int, request *Request) {
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
results[index] = &Result{
|
||||
Connector: request.Connector,
|
||||
Error: fmt.Sprintf("panic: %v", err),
|
||||
}
|
||||
}
|
||||
done <- struct{}{}
|
||||
}()
|
||||
|
||||
// Use forked context to avoid race conditions
|
||||
results[index] = api.executeSingleRequestWithForkedContext(request)
|
||||
}(i, req)
|
||||
}
|
||||
|
||||
// Wait for all to complete
|
||||
for remaining > 0 {
|
||||
<-done
|
||||
remaining--
|
||||
}
|
||||
|
||||
return results
|
||||
}
|
||||
|
||||
// executeAny executes requests and returns first successful result
|
||||
// Each request uses a forked context to avoid race conditions on shared state
|
||||
func (api *JSAPI) executeAny(requests []*Request) []interface{} {
|
||||
if len(requests) == 0 {
|
||||
return []interface{}{}
|
||||
}
|
||||
|
||||
type indexedResult struct {
|
||||
index int
|
||||
result *Result
|
||||
}
|
||||
|
||||
resultChan := make(chan indexedResult, len(requests))
|
||||
remaining := len(requests)
|
||||
|
||||
for i, req := range requests {
|
||||
go func(index int, request *Request) {
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
resultChan <- indexedResult{
|
||||
index: index,
|
||||
result: &Result{
|
||||
Connector: request.Connector,
|
||||
Error: fmt.Sprintf("panic: %v", err),
|
||||
},
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Use forked context to avoid race conditions
|
||||
res := api.executeSingleRequestWithForkedContext(request)
|
||||
resultChan <- indexedResult{index: index, result: res.(*Result)}
|
||||
}(i, req)
|
||||
}
|
||||
|
||||
// Wait for first success or all failures
|
||||
var firstSuccess *indexedResult
|
||||
errors := make([]*indexedResult, 0)
|
||||
|
||||
for remaining > 0 {
|
||||
ir := <-resultChan
|
||||
remaining--
|
||||
|
||||
if ir.result.Error == "" {
|
||||
// Success!
|
||||
firstSuccess = &ir
|
||||
break
|
||||
}
|
||||
errors = append(errors, &ir)
|
||||
}
|
||||
|
||||
if firstSuccess != nil {
|
||||
return []interface{}{firstSuccess.result}
|
||||
}
|
||||
|
||||
// All failed - return all errors
|
||||
results := make([]interface{}, len(errors))
|
||||
for i, e := range errors {
|
||||
results[i] = e.result
|
||||
}
|
||||
return results
|
||||
}
|
||||
|
||||
// executeRace executes requests and returns first completed result (success or failure)
|
||||
// Each request uses a forked context to avoid race conditions on shared state
|
||||
func (api *JSAPI) executeRace(requests []*Request) []interface{} {
|
||||
if len(requests) == 0 {
|
||||
return []interface{}{}
|
||||
}
|
||||
|
||||
resultChan := make(chan *Result, len(requests))
|
||||
|
||||
for _, req := range requests {
|
||||
go func(request *Request) {
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
resultChan <- &Result{
|
||||
Connector: request.Connector,
|
||||
Error: fmt.Sprintf("panic: %v", err),
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Use forked context to avoid race conditions
|
||||
res := api.executeSingleRequestWithForkedContext(request)
|
||||
resultChan <- res.(*Result)
|
||||
}(req)
|
||||
}
|
||||
|
||||
// Return first result
|
||||
result := <-resultChan
|
||||
return []interface{}{result}
|
||||
}
|
||||
|
||||
// executeSingleRequest executes a single LLM request using the original context
|
||||
// This is used for single calls (not batch)
|
||||
func (api *JSAPI) executeSingleRequest(request *Request) interface{} {
|
||||
return api.StreamWithHandler(request.Connector, request.Messages, request.Options, request.Handler)
|
||||
}
|
||||
|
||||
// executeSingleRequestWithForkedContext executes a single LLM request with a forked context
|
||||
// This is used by batch operations (All/Any/Race) to avoid race conditions
|
||||
// when multiple goroutines access shared context state
|
||||
func (api *JSAPI) executeSingleRequestWithForkedContext(request *Request) interface{} {
|
||||
// Fork the context to get independent resources (IDGenerator, Logger, etc.)
|
||||
forkedCtx := api.ctx.Fork()
|
||||
|
||||
// Create a temporary JSAPI with the forked context
|
||||
forkedAPI := &JSAPI{ctx: forkedCtx}
|
||||
|
||||
return forkedAPI.StreamWithHandler(request.Connector, request.Messages, request.Options, request.Handler)
|
||||
}
|
||||
25
agent/llm/jsapi_types.go
Normal file
25
agent/llm/jsapi_types.go
Normal file
|
|
@ -0,0 +1,25 @@
|
|||
// Package llm provides types and utilities for LLM JSAPI
|
||||
package llm
|
||||
|
||||
import (
|
||||
agentContext "github.com/yaoapp/yao/agent/context"
|
||||
)
|
||||
|
||||
// Request represents a request to call an LLM connector
|
||||
type Request struct {
|
||||
Connector string `json:"connector"` // Target connector ID
|
||||
Messages []interface{} `json:"messages"` // Messages to send
|
||||
Options map[string]interface{} `json:"options,omitempty"` // LLM call options (temperature, max_tokens, etc.)
|
||||
Handler agentContext.OnMessageFunc `json:"-"` // OnMessage handler for this request (not serialized)
|
||||
}
|
||||
|
||||
// Result represents the result of a LLM call via JSAPI
|
||||
type Result struct {
|
||||
Connector string `json:"connector"` // Connector ID that was used
|
||||
Response *agentContext.CompletionResponse `json:"response,omitempty"` // Full LLM response
|
||||
Content string `json:"content,omitempty"` // Extracted text content
|
||||
Error string `json:"error,omitempty"` // Error message if call failed
|
||||
}
|
||||
|
||||
// Note: LlmBatchOnMessageFunc is defined in agent/context/jsapi_llm.go
|
||||
// to avoid circular dependencies
|
||||
|
|
@ -10,9 +10,14 @@ import (
|
|||
_ "github.com/yaoapp/gou/text"
|
||||
"github.com/yaoapp/xun/capsule"
|
||||
"github.com/yaoapp/yao/agent"
|
||||
"github.com/yaoapp/yao/agent/caller"
|
||||
"github.com/yaoapp/yao/agent/llm"
|
||||
"github.com/yaoapp/yao/config"
|
||||
"github.com/yaoapp/yao/kb"
|
||||
"github.com/yaoapp/yao/test"
|
||||
|
||||
// Import assistant to trigger init() which registers AgentGetterFunc
|
||||
_ "github.com/yaoapp/yao/agent/assistant"
|
||||
)
|
||||
|
||||
// Prepare prepare the test environment with optional V8 mode configuration
|
||||
|
|
@ -35,6 +40,11 @@ func Prepare(t *testing.T, opts ...interface{}) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Ensure JSAPI factories are registered (may be called multiple times, idempotent)
|
||||
// This is needed because Go's init() order is not guaranteed across packages
|
||||
caller.SetJSAPIFactory()
|
||||
llm.SetJSAPIFactory()
|
||||
|
||||
// Register default query engine (required for DB search)
|
||||
// capsule.Global is initialized by test.Prepare
|
||||
if _, has := query.Engines["default"]; !has && capsule.Global != nil {
|
||||
|
|
|
|||
|
|
@ -1034,18 +1034,17 @@ func TestParseDataURI(t *testing.T) {
|
|||
}
|
||||
})
|
||||
|
||||
// Test 2: Plain base64 (no data URI header)
|
||||
t.Run("PlainBase64", func(t *testing.T) {
|
||||
content := "Plain base64"
|
||||
base64Content := base64.StdEncoding.EncodeToString([]byte(content))
|
||||
// Test 2: Plain text (no data URI header) - treated as plain text, not base64
|
||||
t.Run("PlainText", func(t *testing.T) {
|
||||
content := "Plain text content"
|
||||
|
||||
contentType, data, err := parseDataURI(base64Content)
|
||||
contentType, data, err := parseDataURI(content)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to parse plain base64: %v", err)
|
||||
t.Fatalf("Failed to parse plain text: %v", err)
|
||||
}
|
||||
|
||||
if contentType != "application/octet-stream" {
|
||||
t.Errorf("Expected content type 'application/octet-stream', got '%s'", contentType)
|
||||
if contentType != "text/plain" {
|
||||
t.Errorf("Expected content type 'text/plain', got '%s'", contentType)
|
||||
}
|
||||
|
||||
if string(data) != content {
|
||||
|
|
|
|||
4
go.mod
4
go.mod
|
|
@ -125,7 +125,7 @@ require (
|
|||
github.com/rivo/uniseg v0.4.7 // indirect
|
||||
github.com/robfig/cron/v3 v3.0.1 // indirect
|
||||
github.com/sergi/go-diff v1.3.1 // indirect
|
||||
github.com/sirupsen/logrus v1.9.3 // indirect
|
||||
github.com/sirupsen/logrus v1.9.4 // indirect
|
||||
github.com/spf13/pflag v1.0.6 // indirect
|
||||
github.com/tcnksm/go-gitconfig v0.1.2 // indirect
|
||||
github.com/tidwall/btree v1.7.0 // indirect
|
||||
|
|
@ -153,7 +153,7 @@ require (
|
|||
golang.org/x/mod v0.29.0 // indirect
|
||||
golang.org/x/oauth2 v0.30.0 // indirect
|
||||
golang.org/x/sync v0.18.0 // indirect
|
||||
golang.org/x/sys v0.38.0 // indirect
|
||||
golang.org/x/sys v0.40.0 // indirect
|
||||
golang.org/x/tools v0.38.0 // indirect
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20250519155744-55703ea1f237 // indirect
|
||||
google.golang.org/grpc v1.72.1 // indirect
|
||||
|
|
|
|||
8
go.sum
8
go.sum
|
|
@ -274,8 +274,8 @@ github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWN
|
|||
github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
|
||||
github.com/sergi/go-diff v1.3.1 h1:xkr+Oxo4BOQKmkn/B9eMK0g5Kg/983T9DqqPHwYqD+8=
|
||||
github.com/sergi/go-diff v1.3.1/go.mod h1:aMJSSKb2lpPvRNec0+w3fl7LP9IOFzdc9Pa4NFbPK1I=
|
||||
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
|
||||
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
|
||||
github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w=
|
||||
github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g=
|
||||
github.com/spf13/cast v1.9.2 h1:SsGfm7M8QOFtEzumm7UZrZdLLquNdzFYfIbEXntcFbE=
|
||||
github.com/spf13/cast v1.9.2/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo=
|
||||
github.com/spf13/cobra v1.9.1 h1:CXSaggrXdbHK9CF+8ywj8Amf7PBRmPCOJugH954Nnlo=
|
||||
|
|
@ -427,8 +427,8 @@ golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
|||
golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=
|
||||
golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
|
||||
golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ=
|
||||
golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
|
||||
golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE=
|
||||
golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue