diff --git a/agent/agent.go b/agent/agent.go index d25f41224d..aaf63c1785 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -19,6 +19,7 @@ import ( "fmt" "io" "net/http" + "sort" "strings" "sync" "time" @@ -27,6 +28,7 @@ import ( pb "go-micro.dev/v6/agent/proto" "go-micro.dev/v6/flow" "go-micro.dev/v6/gateway/a2a" + "go-micro.dev/v6/logger" "go-micro.dev/v6/model" "go-micro.dev/v6/server" "go-micro.dev/v6/store" @@ -170,6 +172,11 @@ func (a *agentImpl) setupWithToolHandler(handler model.ToolHandler) { modelOpts = append(modelOpts, model.WithBaseURL(a.opts.BaseURL)) } + modelOpts = append(modelOpts, model.WithMaxTokens(a.opts.MaxTokens), model.WithEffort(a.opts.Effort)) + if a.opts.Temperature != nil { + modelOpts = append(modelOpts, model.WithTemperature(*a.opts.Temperature)) + } + // Reuse the existing tools instance: its name map is populated by // discoverTools, and rebuilding it here would orphan a base handler that // already captured the old instance (breaking StreamAsk tool resolution). @@ -746,6 +753,20 @@ func (a *agentImpl) discoverTools() ([]model.Tool, error) { if !a.ephemeral { scoped = append(scoped, builtinTools()...) } + if limit := a.opts.MaxTools; limit > 0 && len(scoped) > limit { + sort.SliceStable(scoped, func(i, j int) bool { + if scoped[i].Name == scoped[j].Name { + return scoped[i].OriginalName < scoped[j].OriginalName + } + return scoped[i].Name < scoped[j].Name + }) + dropped := make([]string, 0, len(scoped)-limit) + for _, tool := range scoped[limit:] { + dropped = append(dropped, tool.Name) + } + logger.Warnf("agent %s: MaxTools=%d omitted tools: %s", a.opts.Name, limit, strings.Join(dropped, ", ")) + scoped = scoped[:limit] + } return scoped, nil } diff --git a/agent/builtin.go b/agent/builtin.go index b22c262b82..ad3f2328ad 100644 --- a/agent/builtin.go +++ b/agent/builtin.go @@ -750,6 +750,15 @@ func (a *agentImpl) handleDelegate(ctx context.Context, call model.ToolCall) (re "Complete it using the available tools and report the result concisely."), Provider(a.opts.Provider), Model(a.opts.Model), + BaseURL(a.opts.BaseURL), + MaxTokens(a.opts.MaxTokens), + Effort(a.opts.Effort), + MaxTools(a.opts.MaxTools), + func(o *Options) { + if a.opts.Temperature != nil { + Temperature(*a.opts.Temperature)(o) + } + }, APIKey(a.opts.APIKey), WithRegistry(a.opts.Registry), WithClient(a.opts.Client), diff --git a/agent/model_controls_test.go b/agent/model_controls_test.go new file mode 100644 index 0000000000..713c25d482 --- /dev/null +++ b/agent/model_controls_test.go @@ -0,0 +1,106 @@ +package agent + +import ( + "context" + "io" + "reflect" + "testing" + + "go-micro.dev/v6/model" +) + +func TestAgentModelControlsAcrossCallPaths(t *testing.T) { + for _, mode := range []string{"ask", "stream_ask", "stream", "delegate"} { + t.Run(mode, func(t *testing.T) { + calls := 0 + check := func(opts model.Options) { + calls++ + if opts.MaxTokens != 1024 || opts.Effort != "low" || opts.Temperature == nil || *opts.Temperature != 0 || opts.BaseURL != "http://unused.test" { + t.Errorf("options lost: %+v", opts) + } + } + fakeGen = func(_ context.Context, opts model.Options, _ *model.Request) (*model.Response, error) { + check(opts) + return &model.Response{Reply: "ok"}, nil + } + fakeStream = func(_ context.Context, opts model.Options, _ *model.Request) (model.Stream, error) { + check(opts) + return &sliceStream{chunks: []string{"ok"}}, nil + } + defer func() { fakeGen = nil; fakeStream = nil }() + a := newTestAgent(Name("controls"), BaseURL("http://unused.test"), MaxTokens(1024), Effort("low"), Temperature(0), MaxTools(1)) + ctx := context.Background() + switch mode { + case "ask": + if _, err := a.Ask(ctx, "hello"); err != nil { + t.Fatal(err) + } + case "stream_ask": + s, err := a.StreamAsk(ctx, "hello") + if err != nil { + t.Fatal(err) + } + defer s.Close() + for { + _, err = s.Recv() + if err == io.EOF { + break + } + if err != nil { + t.Fatal(err) + } + } + case "stream": + s, err := a.Stream(ctx, "hello") + if err != nil { + t.Fatal(err) + } + defer s.Close() + for { + _, err = s.Recv() + if err == io.EOF { + break + } + if err != nil { + t.Fatal(err) + } + } + case "delegate": + result := a.toolHandler()(ctx, model.ToolCall{ID: "delegate", Name: "delegate", Input: map[string]any{"task": "hello"}}) + if result.Content == "" { + t.Fatal("empty delegate result") + } + } + if calls != 1 { + t.Fatalf("provider calls=%d", calls) + } + }) + } +} + +func TestMaxToolsDeterministicAcrossDiscoveryOrder(t *testing.T) { + for _, names := range [][]string{{"zeta", "alpha", "beta"}, {"beta", "zeta", "alpha"}} { + for _, limit := range []int{0, 2, 10} { + opts := []Option{Name("cap"), Services(), MaxTools(limit)} + for _, name := range names { + opts = append(opts, WithTool(name, name, nil, func(context.Context, map[string]any) (string, error) { return "ok", nil })) + } + a := newTestAgent(opts...) + tools, err := a.discoverTools() + if err != nil { + t.Fatal(err) + } + var got []string + for _, tool := range tools { + got = append(got, tool.Name) + } + if limit == 2 { + if !reflect.DeepEqual(got, []string{"alpha", "beta"}) { + t.Fatalf("capped=%v", got) + } + } else if len(got) != len(names)+len(builtinTools()) { + t.Fatalf("uncapped=%v", got) + } + } + } +} diff --git a/agent/options.go b/agent/options.go index 7396bec174..a619a71e7d 100644 --- a/agent/options.go +++ b/agent/options.go @@ -50,6 +50,14 @@ type Options struct { Store store.Store HistoryLimit int + // MaxTokens and Effort configure provider output and reasoning budgets. + MaxTokens int + Effort string + // Temperature is optional; nil leaves the provider default unchanged. + Temperature *float64 + // MaxTools caps advertised tools, including custom and built-in tools (0 = unlimited). + MaxTools int + // ModelTimeout bounds each provider Generate call (0 disables). ModelTimeout time.Duration // ModelMaxAttempts bounds provider Generate attempts including the first @@ -454,3 +462,17 @@ type RunEventFunc func(RunEvent) func OnRunEvent(f RunEventFunc) Option { return func(o *Options) { o.OnRunEvent = f } } + +// MaxTokens caps provider output tokens (0 leaves the provider default). +func MaxTokens(n int) Option { return func(o *Options) { o.MaxTokens = n } } + +// Effort sets provider-specific reasoning effort. Empty leaves the default. +func Effort(level string) Option { return func(o *Options) { o.Effort = level } } + +// Temperature sets sampling temperature for providers that support it, including zero. +func Temperature(t float64) Option { return func(o *Options) { v := t; o.Temperature = &v } } + +// MaxTools limits advertised tools to the first n names in lexical order when +// the limit is exceeded, logging omitted names. Zero leaves tools unlimited. +// This includes service, custom and built-in tools; MaxSteps separately limits executions. +func MaxTools(n int) Option { return func(o *Options) { o.MaxTools = n } } diff --git a/internal/website/content/en/docs/guides/debugging-agents.md b/internal/website/content/en/docs/guides/debugging-agents.md index 14c40d5375..e67bec9bbf 100644 --- a/internal/website/content/en/docs/guides/debugging-agents.md +++ b/internal/website/content/en/docs/guides/debugging-agents.md @@ -286,3 +286,36 @@ agent.ToolCallTimeout(60 * time.Second) The caller's context and RPC request deadline must also allow the whole turn to finish. Increasing a per-call timeout cannot extend an earlier caller deadline. + +## Configure model budgets and tool counts + +Agent options pass generation settings to the provider for both `Ask` and +streaming calls. Ephemeral delegates inherit these settings; registered domain +agents retain their own configuration. + +```go +agent.New( + agent.Provider("groq"), + agent.MaxTokens(1024), + agent.Effort("low"), + agent.Temperature(0), + agent.MaxTools(128), +) +``` + +`MaxTokens(0)`, an empty effort, and an unset temperature keep provider defaults. +Temperature zero is explicitly sent. Valid effort and temperature values depend +on the selected model; Groq and OpenAI adapters send all three settings on initial +requests, tool follow-ups, and text streams. + +`MaxTools` limits the advertised set, including service, custom, and built-in +plan/delegate tools. When exceeded, tools are sorted by name, the first N are +retained, and omitted names are logged. Zero means unlimited. Use `Services` to +narrow the selection before applying the cap. `MaxSteps` remains a separate +limit on executions. + +Groq/OpenAI responses that finish with `length` without visible text or tool +calls return `model.ErrOutputLimit` (detectable with `errors.Is`), including in +text streams. Increase the output budget or adjust the model's reasoning effort +before retrying. Partial visible responses remain usable and carry +`StopReason: "length"`. diff --git a/micro.go b/micro.go index b2be32de33..233d57e8fd 100644 --- a/micro.go +++ b/micro.go @@ -105,6 +105,18 @@ func AgentAPIKey(k string) AgentOption { return agent.APIKey(k) } // the provider at a non-default endpoint (e.g., local Ollama, a proxy). func AgentBaseURL(url string) AgentOption { return agent.BaseURL(url) } +// AgentMaxTokens caps provider output tokens (0 leaves the provider default). +func AgentMaxTokens(n int) AgentOption { return agent.MaxTokens(n) } + +// AgentEffort sets provider-specific reasoning effort. +func AgentEffort(level string) AgentOption { return agent.Effort(level) } + +// AgentTemperature sets sampling temperature, including zero, for supporting providers. +func AgentTemperature(t float64) AgentOption { return agent.Temperature(t) } + +// AgentMaxTools caps the advertised tool count (0 leaves it unlimited). +func AgentMaxTools(n int) AgentOption { return agent.MaxTools(n) } + // ApproveFunc gates an agent's tool calls before they run. type ApproveFunc = agent.ApproveFunc diff --git a/model/groq/groq.go b/model/groq/groq.go index 74efbb7a02..0c9d0b175a 100644 --- a/model/groq/groq.go +++ b/model/groq/groq.go @@ -106,6 +106,7 @@ func (p *Provider) Generate(ctx context.Context, req *model.Request, opts ...mod if followUpResp.Reply != "" { resp.Answer = followUpResp.Reply } + resp.StopReason = followUpResp.StopReason pending, raw = followUpResp.ToolCalls, followUpRaw resp.ToolCalls = append(resp.ToolCalls, followUpResp.ToolCalls...) } @@ -152,7 +153,8 @@ func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*model.Resp var chatResp struct { Choices []struct { - Message struct { + FinishReason string `json:"finish_reason"` + Message struct { Content string `json:"content"` ToolCalls []struct { ID string `json:"id"` @@ -174,8 +176,13 @@ func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*model.Resp } choice := chatResp.Choices[0] + if choice.FinishReason == "length" && strings.TrimSpace(choice.Message.Content) == "" && len(choice.Message.ToolCalls) == 0 { + return nil, nil, model.ErrOutputLimit + } response := &model.Response{Reply: choice.Message.Content} + response.StopReason = choice.FinishReason + for _, tc := range choice.Message.ToolCalls { var input map[string]any if err := json.Unmarshal([]byte(tc.Function.Arguments), &input); err != nil { diff --git a/model/internal/openaiapi/request.go b/model/internal/openaiapi/request.go index 12b9affeec..a580f09028 100644 --- a/model/internal/openaiapi/request.go +++ b/model/internal/openaiapi/request.go @@ -24,6 +24,9 @@ func Request(opts model.Options, messages []map[string]any, tools []model.Tool) if opts.Effort != "" { request["reasoning_effort"] = opts.Effort } + if opts.Temperature != nil { + request["temperature"] = *opts.Temperature + } if len(tools) > 0 { definitions := make([]map[string]any, 0, len(tools)) for _, tool := range tools { diff --git a/model/internal/openaiapi/request_integration_test.go b/model/internal/openaiapi/request_integration_test.go index 5a0c3e70ee..f0b557f925 100644 --- a/model/internal/openaiapi/request_integration_test.go +++ b/model/internal/openaiapi/request_integration_test.go @@ -27,7 +27,7 @@ func TestChatRequestParity(t *testing.T) { t.Error(err) return } - if body["max_tokens"] != float64(1024) || body["reasoning_effort"] != "high" || body["model"] != "test" { + if body["max_tokens"] != float64(1024) || body["reasoning_effort"] != "high" || body["model"] != "test" || body["temperature"] != float64(0) { t.Errorf("lost options: %v", body) } messages := body["messages"].([]any) @@ -73,7 +73,7 @@ func TestChatRequestParity(t *testing.T) { fmt.Fprint(w, `{"choices":[{"message":{"content":"done"}}]}`) })) defer ts.Close() - p := factory(model.WithBaseURL(ts.URL), model.WithAPIKey("test"), model.WithModel("test"), model.WithMaxTokens(1024), model.WithEffort("high"), model.WithToolHandler(func(_ context.Context, c model.ToolCall) model.ToolResult { + p := factory(model.WithBaseURL(ts.URL), model.WithAPIKey("test"), model.WithModel("test"), model.WithMaxTokens(1024), model.WithTemperature(0), model.WithEffort("high"), model.WithToolHandler(func(_ context.Context, c model.ToolCall) model.ToolResult { calls++ return model.ToolResult{ID: c.ID, Content: "result"} })) @@ -116,3 +116,58 @@ func TestChatRequestParity(t *testing.T) { } } } + +func TestChatEmptyOutputLimit(t *testing.T) { + for name, factory := range map[string]func(...model.Option) model.Model{ + "groq": func(opts ...model.Option) model.Model { return groq.NewProvider(opts...) }, + "openai": func(opts ...model.Option) model.Model { return openai.NewProvider(opts...) }, + } { + for _, streaming := range []bool{false, true} { + for _, visible := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/stream=%v/visible=%v", name, streaming, visible), func(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + content := "" + if visible { + content = "partial" + } + if streaming { + fmt.Fprintf(w, "data: {\"choices\":[{\"delta\":{\"content\":%q}}]}\n\n", content) + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"length\"}]}\n\ndata: [DONE]\n\n") + } else { + fmt.Fprintf(w, `{"choices":[{"message":{"content":%q},"finish_reason":"length"}]}`, content) + } + })) + defer ts.Close() + p := factory(model.WithBaseURL(ts.URL), model.WithAPIKey("test")) + var err error + if streaming { + var s model.Stream + s, err = p.Stream(context.Background(), &model.Request{Prompt: "hello"}) + if err != nil { + t.Fatal(err) + } + defer s.Close() + for { + _, err = s.Recv() + if err != nil { + break + } + } + if err == io.EOF { + err = nil + } + } else { + _, err = p.Generate(context.Background(), &model.Request{Prompt: "hello"}) + } + if visible { + if err != nil { + t.Fatal(err) + } + } else if !errors.Is(err, model.ErrOutputLimit) { + t.Fatalf("error=%v", err) + } + }) + } + } + } +} diff --git a/model/internal/openaiapi/stream.go b/model/internal/openaiapi/stream.go index 8d13e55678..03d3ecb9af 100644 --- a/model/internal/openaiapi/stream.go +++ b/model/internal/openaiapi/stream.go @@ -45,9 +45,10 @@ func Stream(ctx context.Context, opts model.Options, req *model.Request, basePat // StreamReader reads OpenAI-compatible server-sent event chunks. type StreamReader struct { - body io.ReadCloser - scanner *bufio.Scanner - closed bool + body io.ReadCloser + scanner *bufio.Scanner + closed bool + hasContent bool } func (s *StreamReader) Recv() (*model.Response, error) { @@ -65,7 +66,8 @@ func (s *StreamReader) Recv() (*model.Response, error) { } var chunk struct { Choices []struct { - Delta struct { + FinishReason string `json:"finish_reason"` + Delta struct { Content string `json:"content"` } `json:"delta"` } `json:"choices"` @@ -78,15 +80,22 @@ func (s *StreamReader) Recv() (*model.Response, error) { if err := json.Unmarshal([]byte(data), &chunk); err != nil { return nil, fmt.Errorf("failed to parse stream chunk: %w", err) } - if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" { - return &model.Response{Reply: chunk.Choices[0].Delta.Content}, nil + response := &model.Response{} + if len(chunk.Choices) > 0 { + choice := chunk.Choices[0] + if strings.TrimSpace(choice.Delta.Content) != "" { + s.hasContent = true + } + if choice.FinishReason == "length" && !s.hasContent { + return nil, model.ErrOutputLimit + } + response.Reply, response.StopReason = choice.Delta.Content, choice.FinishReason } if chunk.Usage != nil { - return &model.Response{Usage: model.Usage{ - InputTokens: chunk.Usage.PromptTokens, - OutputTokens: chunk.Usage.CompletionTokens, - TotalTokens: chunk.Usage.TotalTokens, - }}, nil + response.Usage = model.Usage{InputTokens: chunk.Usage.PromptTokens, OutputTokens: chunk.Usage.CompletionTokens, TotalTokens: chunk.Usage.TotalTokens} + } + if response.Reply != "" || response.StopReason != "" || chunk.Usage != nil { + return response, nil } } if err := s.scanner.Err(); err != nil { diff --git a/model/model.go b/model/model.go index d71ab32f04..61d7b3b943 100644 --- a/model/model.go +++ b/model/model.go @@ -291,3 +291,7 @@ func Generate(ctx context.Context, req *Request, opts ...GenerateOption) (*Respo } return DefaultModel.Generate(ctx, req, opts...) } + +// ErrOutputLimit means a provider exhausted its output budget without visible +// content or tool calls. Increase MaxTokens or adjust reasoning effort before retrying. +var ErrOutputLimit = errors.New("model: output token limit reached without visible content") diff --git a/model/openai/openai.go b/model/openai/openai.go index 4511d2b954..90fb0a4e19 100644 --- a/model/openai/openai.go +++ b/model/openai/openai.go @@ -117,6 +117,7 @@ func (p *Provider) Generate(ctx context.Context, req *model.Request, opts ...mod if followUpResp.Reply != "" { resp.Answer = followUpResp.Reply } + resp.StopReason = followUpResp.StopReason pending, raw = followUpResp.ToolCalls, followUpRaw resp.ToolCalls = append(resp.ToolCalls, followUpResp.ToolCalls...) } @@ -162,9 +163,10 @@ func (p *Provider) Stream(ctx context.Context, req *model.Request, opts ...model } type openAIStream struct { - body io.ReadCloser - scanner *bufio.Scanner - closed bool + body io.ReadCloser + scanner *bufio.Scanner + closed bool + hasContent bool } func (s *openAIStream) Recv() (*model.Response, error) { @@ -182,7 +184,8 @@ func (s *openAIStream) Recv() (*model.Response, error) { } var chunk struct { Choices []struct { - Delta struct { + FinishReason string `json:"finish_reason"` + Delta struct { Content string `json:"content"` } `json:"delta"` } `json:"choices"` @@ -195,18 +198,23 @@ func (s *openAIStream) Recv() (*model.Response, error) { if err := json.Unmarshal([]byte(data), &chunk); err != nil { return nil, fmt.Errorf("failed to parse stream chunk: %w", err) } - if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" { - return &model.Response{Reply: chunk.Choices[0].Delta.Content}, nil + response := &model.Response{} + if len(chunk.Choices) > 0 { + choice := chunk.Choices[0] + if strings.TrimSpace(choice.Delta.Content) != "" { + s.hasContent = true + } + if choice.FinishReason == "length" && !s.hasContent { + return nil, model.ErrOutputLimit + } + response.Reply, response.StopReason = choice.Delta.Content, choice.FinishReason } - // Final chunk (after include_usage) carries token usage and no content. if chunk.Usage != nil { - return &model.Response{Usage: model.Usage{ - InputTokens: chunk.Usage.PromptTokens, - OutputTokens: chunk.Usage.CompletionTokens, - TotalTokens: chunk.Usage.TotalTokens, - }}, nil + response.Usage = model.Usage{InputTokens: chunk.Usage.PromptTokens, OutputTokens: chunk.Usage.CompletionTokens, TotalTokens: chunk.Usage.TotalTokens} + } + if response.Reply != "" || response.StopReason != "" || chunk.Usage != nil { + return response, nil } - continue } if err := s.scanner.Err(); err != nil { return nil, err @@ -262,7 +270,8 @@ func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*model.Resp TotalTokens int `json:"total_tokens"` } `json:"usage"` Choices []struct { - Message struct { + FinishReason string `json:"finish_reason"` + Message struct { Content string `json:"content"` ToolCalls []struct { ID string `json:"id"` @@ -285,11 +294,16 @@ func (p *Provider) callAPI(ctx context.Context, req map[string]any) (*model.Resp } choice := chatResp.Choices[0] + if choice.FinishReason == "length" && strings.TrimSpace(choice.Message.Content) == "" && len(choice.Message.ToolCalls) == 0 { + return nil, nil, model.ErrOutputLimit + } response := &model.Response{ Reply: choice.Message.Content, Usage: model.Usage{InputTokens: chatResp.Usage.PromptTokens, OutputTokens: chatResp.Usage.CompletionTokens, TotalTokens: chatResp.Usage.TotalTokens}, } + response.StopReason = choice.FinishReason + // Extract tool calls for _, tc := range choice.Message.ToolCalls { var input map[string]any diff --git a/model/options.go b/model/options.go index b955ee1481..d5a76a3bd6 100644 --- a/model/options.go +++ b/model/options.go @@ -23,6 +23,8 @@ type Options struct { Thinking ThinkingMode // Effort controls reasoning depth for providers that support it. Effort string + // Temperature is optional; nil leaves provider defaults unchanged. + Temperature *float64 // NoCache disables prompt-prefix caching for providers that support it // (e.g. Anthropic cache_control). Caching is on by default because the // dominant caller — the agent tool loop — re-sends an identical prefix on @@ -143,3 +145,6 @@ func WithEffort(effort string) Option { o.Effort = effort } } + +// WithTemperature sets sampling temperature for supporting providers, including zero. +func WithTemperature(t float64) Option { return func(o *Options) { v := t; o.Temperature = &v } }