diff --git a/agent/assistant/assistant.go b/agent/assistant/assistant.go index 93bcdf6c..de8005f0 100644 --- a/agent/assistant/assistant.go +++ b/agent/assistant/assistant.go @@ -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) diff --git a/agent/caller/orchestrator.go b/agent/caller/orchestrator.go index 6a0c1210..1865ef19 100644 --- a/agent/caller/orchestrator.go +++ b/agent/caller/orchestrator.go @@ -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 diff --git a/agent/context/context.go b/agent/context/context.go index 74862779..b89b4ff6 100644 --- a/agent/context/context.go +++ b/agent/context/context.go @@ -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 { diff --git a/agent/context/jsapi.go b/agent/context/jsapi.go index aefbdac0..24ef5692 100644 --- a/agent/context/jsapi.go +++ b/agent/context/jsapi.go @@ -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 diff --git a/agent/context/jsapi_llm.go b/agent/context/jsapi_llm.go new file mode 100644 index 00000000..a187ef38 --- /dev/null +++ b/agent/context/jsapi_llm.go @@ -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) +} diff --git a/agent/context/jsapi_llm_v8_test.go b/agent/context/jsapi_llm_v8_test.go new file mode 100644 index 00000000..30b86445 --- /dev/null +++ b/agent/context/jsapi_llm_v8_test.go @@ -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") +} diff --git a/agent/llm/jsapi.go b/agent/llm/jsapi.go new file mode 100644 index 00000000..f0697d1d --- /dev/null +++ b/agent/llm/jsapi.go @@ -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) +} diff --git a/agent/llm/jsapi_types.go b/agent/llm/jsapi_types.go new file mode 100644 index 00000000..54e04c7b --- /dev/null +++ b/agent/llm/jsapi_types.go @@ -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 diff --git a/agent/testutils/testutils.go b/agent/testutils/testutils.go index 104ad06a..69a6abbf 100644 --- a/agent/testutils/testutils.go +++ b/agent/testutils/testutils.go @@ -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 { diff --git a/attachment/process_test.go b/attachment/process_test.go index 5eb24826..68f0dcc2 100644 --- a/attachment/process_test.go +++ b/attachment/process_test.go @@ -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 { diff --git a/go.mod b/go.mod index e83e49a6..58da2437 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index c0bd5669..68e1afeb 100644 --- a/go.sum +++ b/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=