Enhance Assistant methods with options parameter for improved context management
- Updated the Stream, BuildContent, and LLM execution methods to accept an Options parameter, allowing for more flexible context handling. - Removed debug print statements to clean up the code and improve readability. - Enhanced locale handling in the loadMap function to automatically inject assistant name and description into all locales, ensuring better localization support. - Introduced output skipping functionality in context options to manage internal A2A calls more effectively. - Improved output writer resolution logic to prioritize context settings, enhancing output management during agent calls.
This commit is contained in:
parent
a96946d8bb
commit
311350bfbc
10 changed files with 93 additions and 27 deletions
|
|
@ -51,13 +51,6 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa
|
|||
_, _, done := context.EnterStack(ctx, ast.ID, opts)
|
||||
defer done()
|
||||
|
||||
fmt.Println("--- Stack debug ---")
|
||||
if ctx.Stack != nil {
|
||||
fmt.Println(ctx.Stack.IsRoot())
|
||||
utils.Dump(ctx.Stack)
|
||||
}
|
||||
fmt.Println("------ end stack debug ------")
|
||||
|
||||
// Determine stream handler
|
||||
streamHandler := ast.getStreamHandler(ctx, opts)
|
||||
|
||||
|
|
@ -123,7 +116,7 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa
|
|||
}
|
||||
|
||||
// Build content - convert extended types (file, data) to standard LLM types (text, image_url, input_audio)
|
||||
completionMessages, err = ast.BuildContent(ctx, completionMessages, completionOptions)
|
||||
completionMessages, err = ast.BuildContent(ctx, completionMessages, completionOptions, opts)
|
||||
if err != nil {
|
||||
ast.traceAgentFail(agentNode, err)
|
||||
ast.sendStreamEndOnError(ctx, streamHandler, streamStartTime, err)
|
||||
|
|
@ -131,7 +124,7 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa
|
|||
}
|
||||
|
||||
// Execute the LLM streaming call
|
||||
completionResponse, err = ast.executeLLMStream(ctx, completionMessages, completionOptions, agentNode, streamHandler)
|
||||
completionResponse, err = ast.executeLLMStream(ctx, completionMessages, completionOptions, agentNode, streamHandler, opts)
|
||||
if err != nil {
|
||||
ast.traceAgentFail(agentNode, err)
|
||||
// Send error stream_end for root stack
|
||||
|
|
@ -210,7 +203,7 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa
|
|||
|
||||
// Retry LLM call (streaming to keep user informed)
|
||||
log.Trace("[AGENT] Retrying LLM for tool call correction (attempt %d/%d)", attempt+1, maxToolRetries-1)
|
||||
currentResponse, err = ast.executeLLMForToolRetry(ctx, retryMessages, completionOptions, agentNode, streamHandler)
|
||||
currentResponse, err = ast.executeLLMForToolRetry(ctx, retryMessages, completionOptions, agentNode, streamHandler, opts)
|
||||
if err != nil {
|
||||
log.Error("[AGENT] LLM retry failed: %v", err)
|
||||
ast.traceAgentFail(agentNode, err)
|
||||
|
|
@ -479,6 +472,10 @@ func (ast *Assistant) initializeCapabilities(ctx *context.Context, opts *context
|
|||
return err
|
||||
}
|
||||
|
||||
fmt.Println("--- initializeCapabilities debug ---")
|
||||
utils.Dump(capabilities)
|
||||
fmt.Println("--- end initializeCapabilities debug ---")
|
||||
|
||||
// Set capabilities in context for output adapters to use
|
||||
if capabilities != nil {
|
||||
ctx.Capabilities = capabilities
|
||||
|
|
|
|||
|
|
@ -11,9 +11,9 @@ import (
|
|||
// (file, data) to standard LLM-compatible types (text, image_url, input_audio)
|
||||
//
|
||||
// This should be called after BuildRequest and before executing LLM call
|
||||
func (ast *Assistant) BuildContent(ctx *context.Context, messages []context.Message, options *context.CompletionOptions) ([]context.Message, error) {
|
||||
func (ast *Assistant) BuildContent(ctx *context.Context, messages []context.Message, options *context.CompletionOptions, opts *context.Options) ([]context.Message, error) {
|
||||
// Get connector and capabilities
|
||||
_, capabilities, err := ast.GetConnector(ctx)
|
||||
_, capabilities, err := ast.GetConnector(ctx, opts)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get connector: %w", err)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ func (ast *Assistant) executeLLMStream(
|
|||
completionOptions *context.CompletionOptions,
|
||||
agentNode types.Node,
|
||||
streamHandler message.StreamFunc,
|
||||
opts *context.Options,
|
||||
) (*context.CompletionResponse, error) {
|
||||
|
||||
// === Debug LLM Stream Start ===
|
||||
|
|
@ -27,7 +28,7 @@ func (ast *Assistant) executeLLMStream(
|
|||
// === End Debug ===
|
||||
|
||||
// Get connector object (capabilities were already set above, before stream_start)
|
||||
conn, capabilities, err := ast.GetConnector(ctx)
|
||||
conn, capabilities, err := ast.GetConnector(ctx, opts)
|
||||
if err != nil {
|
||||
ast.traceAgentFail(agentNode, err)
|
||||
return nil, err
|
||||
|
|
@ -92,10 +93,11 @@ func (ast *Assistant) executeLLMForToolRetry(
|
|||
completionOptions *context.CompletionOptions,
|
||||
agentNode types.Node,
|
||||
streamHandler message.StreamFunc,
|
||||
opts *context.Options,
|
||||
) (*context.CompletionResponse, error) {
|
||||
|
||||
// Get connector object
|
||||
conn, capabilities, err := ast.GetConnector(ctx)
|
||||
conn, capabilities, err := ast.GetConnector(ctx, opts)
|
||||
if err != nil {
|
||||
ast.traceAgentFail(agentNode, err)
|
||||
return nil, err
|
||||
|
|
|
|||
|
|
@ -531,7 +531,38 @@ func loadMap(data map[string]interface{}) (*Assistant, error) {
|
|||
// locales
|
||||
if locales, ok := data["locales"].(i18n.Map); ok {
|
||||
assistant.Locales = locales
|
||||
i18n.Locales[id] = locales.FlattenWithGlobal()
|
||||
flattened := locales.FlattenWithGlobal()
|
||||
|
||||
// Auto-inject assistant name and description into all locales
|
||||
// so that {{name}} and {{description}} templates can be resolved
|
||||
for locale, i18nObj := range flattened {
|
||||
if i18nObj.Messages == nil {
|
||||
i18nObj.Messages = make(map[string]any)
|
||||
}
|
||||
// Add name and description if not already present
|
||||
if _, exists := i18nObj.Messages["name"]; !exists && assistant.Name != "" {
|
||||
i18nObj.Messages["name"] = assistant.Name
|
||||
}
|
||||
if _, exists := i18nObj.Messages["description"]; !exists && assistant.Description != "" {
|
||||
i18nObj.Messages["description"] = assistant.Description
|
||||
}
|
||||
flattened[locale] = i18nObj
|
||||
}
|
||||
|
||||
i18n.Locales[id] = flattened
|
||||
} else {
|
||||
// No locales defined, create default with name and description
|
||||
if assistant.Name != "" || assistant.Description != "" {
|
||||
defaultLocales := make(map[string]i18n.I18n)
|
||||
defaultLocales["en"] = i18n.I18n{
|
||||
Locale: "en",
|
||||
Messages: map[string]any{
|
||||
"name": assistant.Name,
|
||||
"description": assistant.Description,
|
||||
},
|
||||
}
|
||||
i18n.Locales[id] = defaultLocales
|
||||
}
|
||||
}
|
||||
|
||||
// Search options
|
||||
|
|
|
|||
|
|
@ -36,8 +36,8 @@ func CallAgent(ctx *agentContext.Context, agentID string, message agentContext.M
|
|||
messages := []agentContext.Message{message}
|
||||
|
||||
// Note: Connector is now in Options (call-level parameter), not Context
|
||||
// For A2A calls, we use an empty Connector to let the agent use its default
|
||||
opts := &agentContext.Options{Skip: &agentContext.Skip{History: true}, Writer: nil} // Skip history and output to the caller
|
||||
// For A2A calls, skip history and output (we only need the response data)
|
||||
opts := &agentContext.Options{Skip: &agentContext.Skip{History: true, Output: true}} // Skip history and output
|
||||
response, err := agent.Stream(ctx, messages, opts)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to call agent %s: %w", agentID, err)
|
||||
|
|
|
|||
|
|
@ -58,6 +58,9 @@ func OptionsFromMap(m map[string]interface{}) *Options {
|
|||
if trace, ok := skipMap["trace"].(bool); ok {
|
||||
skip.Trace = trace
|
||||
}
|
||||
if output, ok := skipMap["output"].(bool); ok {
|
||||
skip.Output = output
|
||||
}
|
||||
opts.Skip = skip
|
||||
}
|
||||
if disableGlobalPrompts, ok := m["disable_global_prompts"].(bool); ok {
|
||||
|
|
|
|||
|
|
@ -297,16 +297,33 @@ func (ctx *Context) sendRaw(msg *message.Message) error {
|
|||
return out.Send(msg)
|
||||
}
|
||||
|
||||
// getWriter gets the effective Writer for the current context
|
||||
// Priority: Skip.Output > Stack.Options.Writer > ctx.Writer
|
||||
func (ctx *Context) getWriter() Writer {
|
||||
// Check if output is explicitly skipped (for internal A2A calls)
|
||||
if ctx.Stack != nil && ctx.Stack.Options != nil && ctx.Stack.Options.Skip != nil && ctx.Stack.Options.Skip.Output {
|
||||
return nil // Explicitly disable output
|
||||
}
|
||||
|
||||
// Check if current Stack has a Writer override
|
||||
if ctx.Stack != nil && ctx.Stack.Options != nil && ctx.Stack.Options.Writer != nil {
|
||||
return ctx.Stack.Options.Writer
|
||||
}
|
||||
|
||||
return ctx.Writer
|
||||
}
|
||||
|
||||
// getOutput gets the output writer for the context
|
||||
func (ctx *Context) getOutput() (*output.Output, error) {
|
||||
if ctx.output != nil {
|
||||
return ctx.output, nil
|
||||
// Check if current Stack has cached output
|
||||
if ctx.Stack != nil && ctx.Stack.output != nil {
|
||||
return ctx.Stack.output, nil
|
||||
}
|
||||
|
||||
trace, _ := ctx.Trace()
|
||||
var options message.Options = message.Options{
|
||||
BaseURL: "/",
|
||||
Writer: ctx.Writer,
|
||||
Writer: ctx.getWriter(), // Use getWriter() to resolve Writer priority
|
||||
Trace: trace,
|
||||
Locale: ctx.Locale,
|
||||
Accept: string(ctx.Accept),
|
||||
|
|
@ -318,10 +335,15 @@ func (ctx *Context) getOutput() (*output.Output, error) {
|
|||
options.Capabilities = &caps
|
||||
}
|
||||
|
||||
var err error
|
||||
ctx.output, err = output.NewOutput(options)
|
||||
out, err := output.NewOutput(options)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return ctx.output, nil
|
||||
|
||||
// Cache to current Stack (each Stack has its own output with its own Writer)
|
||||
if ctx.Stack != nil {
|
||||
ctx.Stack.output = out
|
||||
}
|
||||
|
||||
return out, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -194,6 +194,7 @@ type AssistantInfo struct {
|
|||
type Skip struct {
|
||||
History bool `json:"history"` // Skip saving chat history (for internal calls like title/prompt generation)
|
||||
Trace bool `json:"trace"` // Skip trace logging
|
||||
Output bool `json:"output"` // Skip output to client (for internal A2A calls that only need response data)
|
||||
}
|
||||
|
||||
// MessageMetadata stores metadata for sent messages
|
||||
|
|
@ -232,7 +233,6 @@ type Context struct {
|
|||
|
||||
// Internal
|
||||
trace traceTypes.Manager `json:"-"` // Trace manager, lazy initialized on first access
|
||||
output *output.Output `json:"-"` // Output, it will be used to write response data to the client
|
||||
messageMetadata *messageMetadataStore `json:"-"` // Thread-safe message metadata store for delta operations
|
||||
|
||||
// Model capabilities (set by assistant, used by output adapters)
|
||||
|
|
@ -312,6 +312,9 @@ type Stack struct {
|
|||
|
||||
// Metrics
|
||||
DurationMs *int64 `json:"duration_ms,omitempty"` // Duration in milliseconds (calculated when completed)
|
||||
|
||||
// Runtime cache (not serialized)
|
||||
output *output.Output `json:"-"` // Cached output instance for this stack
|
||||
}
|
||||
|
||||
// Response the response
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ func init() {
|
|||
"assistant.agent.stream.skipping": "Skipping output close (nested call)",
|
||||
"assistant.agent.stream.close_error": "Failed to close output",
|
||||
"assistant.agent.completion.label": "Agent Completion",
|
||||
"assistant.agent.completion.description": "Final output from assistant",
|
||||
"assistant.agent.completion.description": "Final output from {{name}}",
|
||||
|
||||
// LLM: providers/openai/openai.go Stream() function
|
||||
"llm.openai.stream.label": "LLM %s",
|
||||
|
|
@ -111,7 +111,7 @@ func init() {
|
|||
"assistant.agent.stream.skipping": "跳过输出关闭(嵌套调用)",
|
||||
"assistant.agent.stream.close_error": "关闭输出失败",
|
||||
"assistant.agent.completion.label": "智能体完成",
|
||||
"assistant.agent.completion.description": "智能体最终输出",
|
||||
"assistant.agent.completion.description": "{{name}} 最终输出",
|
||||
|
||||
// LLM: providers/openai/openai.go Stream() function
|
||||
"llm.openai.stream.label": "LLM %s",
|
||||
|
|
@ -173,7 +173,7 @@ func init() {
|
|||
"assistant.agent.stream.skipping": "跳过输出关闭(嵌套调用)",
|
||||
"assistant.agent.stream.close_error": "关闭输出失败",
|
||||
"assistant.agent.completion.label": "智能体完成",
|
||||
"assistant.agent.completion.description": "智能体最终输出",
|
||||
"assistant.agent.completion.description": "{{name}} 最终输出",
|
||||
|
||||
// LLM: providers/openai/openai.go Stream() function
|
||||
"llm.openai.stream.label": "LLM %s",
|
||||
|
|
|
|||
|
|
@ -211,7 +211,11 @@ func (p *Provider) Stream(ctx *context.Context, messages []context.Message, opti
|
|||
var lastErr error
|
||||
|
||||
// Get Go context for cancellation support
|
||||
// Read from Stack.Options if available (call-level override)
|
||||
goCtx := ctx.Context
|
||||
if ctx.Stack != nil && ctx.Stack.Options != nil && ctx.Stack.Options.Context != nil {
|
||||
goCtx = ctx.Stack.Options.Context
|
||||
}
|
||||
if goCtx == nil {
|
||||
goCtx = gocontext.Background()
|
||||
}
|
||||
|
|
@ -808,7 +812,11 @@ func (p *Provider) Post(ctx *context.Context, messages []context.Message, option
|
|||
var lastErr error
|
||||
|
||||
// Get Go context for cancellation support
|
||||
// Read from Stack.Options if available (call-level override)
|
||||
goCtx := ctx.Context
|
||||
if ctx.Stack != nil && ctx.Stack.Options != nil && ctx.Stack.Options.Context != nil {
|
||||
goCtx = ctx.Stack.Options.Context
|
||||
}
|
||||
if goCtx == nil {
|
||||
goCtx = gocontext.Background()
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue