Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d46a49b50c |
+11
-4
@@ -207,6 +207,17 @@ func (a *agentImpl) Stream(ctx context.Context, message string) (ai.Stream, erro
|
||||
})
|
||||
}
|
||||
|
||||
// Resume returns the response for a checkpointed agent run. Completed runs are
|
||||
// returned from the checkpoint without calling the model or replaying tool
|
||||
// calls; failed or in-progress runs continue from the saved input message.
|
||||
func Resume(ctx context.Context, ag Agent, runID string) (*Response, error) {
|
||||
a, ok := ag.(*agentImpl)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("agent resume: unsupported agent implementation %T", ag)
|
||||
}
|
||||
return a.resume(ctx, runID)
|
||||
}
|
||||
|
||||
// Pending returns checkpointed agent runs that have not completed. It mirrors
|
||||
// flow.Pending for startup recovery loops that drain durable agent work.
|
||||
func Pending(ctx context.Context, ag Agent) ([]flow.Run, error) {
|
||||
@@ -289,10 +300,6 @@ func (a *agentImpl) askLocked(ctx context.Context, runID, message, parentRunID s
|
||||
run.Status = "paused"
|
||||
run.State.Stage = agentApprovalStep
|
||||
run.State.Data = []byte(message)
|
||||
if a.pause.Tool == toolHumanInput {
|
||||
run.State.Stage = agentInputStep
|
||||
_ = run.State.Set(inputPause{OriginalMessage: message, Prompt: a.pause.Message})
|
||||
}
|
||||
run.Steps[0].Status = "paused"
|
||||
run.Steps[0].Error = a.pause.Message
|
||||
run.Steps[0].Result = a.pause.Tool
|
||||
|
||||
+2
-37
@@ -20,9 +20,8 @@ import (
|
||||
// the discovered service tools. There is no separate harness or graph:
|
||||
// the LLM calls them like any other tool.
|
||||
const (
|
||||
toolPlan = "plan"
|
||||
toolDelegate = "delegate"
|
||||
toolHumanInput = "request_input"
|
||||
toolPlan = "plan"
|
||||
toolDelegate = "delegate"
|
||||
)
|
||||
|
||||
// builtinTools returns the tool definitions exposed to the model in
|
||||
@@ -42,18 +41,6 @@ func builtinTools() []ai.Tool {
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: toolHumanInput,
|
||||
OriginalName: toolHumanInput,
|
||||
Description: "Pause this agent run when you need missing information, a decision, or other human input before you can continue. " +
|
||||
"The run is checkpointed as input-required and can be resumed with the human response without losing completed tool history.",
|
||||
Properties: map[string]any{
|
||||
"prompt": map[string]any{
|
||||
"type": "string",
|
||||
"description": "The specific question, decision, or instruction needed from the human operator.",
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Name: toolDelegate,
|
||||
OriginalName: toolDelegate,
|
||||
@@ -91,9 +78,6 @@ func Builtins(opts ...Option) (tools []ai.Tool, handle func(name string, input m
|
||||
case toolPlan:
|
||||
r := a.handlePlan(ai.ToolCall{Name: name, Input: input})
|
||||
return r.Value, r.Content, true
|
||||
case toolHumanInput:
|
||||
r := a.handleHumanInput(ai.ToolCall{Name: name, Input: input})
|
||||
return r.Value, r.Content, true
|
||||
case toolDelegate:
|
||||
r := a.handleDelegate(context.Background(), ai.ToolCall{Name: name, Input: input})
|
||||
return r.Value, r.Content, true
|
||||
@@ -162,9 +146,6 @@ func (a *agentImpl) baseHandler() ai.ToolHandler {
|
||||
return ai.ToolResult{ID: call.ID, Value: out, Content: out}
|
||||
}
|
||||
}
|
||||
if call.Name == toolHumanInput {
|
||||
return a.handleHumanInput(call)
|
||||
}
|
||||
if call.Name == toolDelegate {
|
||||
return a.handleDelegate(ctx, call)
|
||||
}
|
||||
@@ -225,11 +206,6 @@ type approvalPause struct {
|
||||
Message string
|
||||
}
|
||||
|
||||
type inputPause struct {
|
||||
OriginalMessage string `json:"original_message"`
|
||||
Prompt string `json:"prompt"`
|
||||
}
|
||||
|
||||
func (a *agentImpl) approveWrap(next ai.ToolHandler) ai.ToolHandler {
|
||||
return func(ctx context.Context, call ai.ToolCall) ai.ToolResult {
|
||||
if a.opts.Approve != nil {
|
||||
@@ -257,17 +233,6 @@ func (a *agentImpl) handlePlan(call ai.ToolCall) ai.ToolResult {
|
||||
return ai.ToolResult{ID: call.ID, Value: call.Input, Content: string(data)}
|
||||
}
|
||||
|
||||
// handleHumanInput records that the model needs operator input before it can continue.
|
||||
func (a *agentImpl) handleHumanInput(call ai.ToolCall) ai.ToolResult {
|
||||
prompt, _ := call.Input["prompt"].(string)
|
||||
prompt = strings.TrimSpace(prompt)
|
||||
if prompt == "" {
|
||||
prompt = "human input required"
|
||||
}
|
||||
a.pause = &approvalPause{Tool: toolHumanInput, Message: prompt}
|
||||
return refused(call.ID, ai.RefusedApproval, "input-required: "+prompt)
|
||||
}
|
||||
|
||||
// handleDelegate hands a subtask to another agent. Delegate-first:
|
||||
// if 'to' names a registered agent, it is called via RPC. Otherwise an
|
||||
// ephemeral sub-agent is created with a fresh, isolated context, asked
|
||||
|
||||
@@ -11,15 +11,15 @@ import (
|
||||
|
||||
func TestBuiltinTools(t *testing.T) {
|
||||
tools := builtinTools()
|
||||
if len(tools) != 3 {
|
||||
t.Fatalf("builtinTools() = %d tools, want 3", len(tools))
|
||||
if len(tools) != 2 {
|
||||
t.Fatalf("builtinTools() = %d tools, want 2", len(tools))
|
||||
}
|
||||
names := map[string]bool{}
|
||||
for _, tl := range tools {
|
||||
names[tl.Name] = true
|
||||
}
|
||||
if !names[toolPlan] || !names[toolDelegate] || !names[toolHumanInput] {
|
||||
t.Errorf("builtin tools = %v, want plan, request_input, and delegate", names)
|
||||
if !names[toolPlan] || !names[toolDelegate] {
|
||||
t.Errorf("builtin tools = %v, want plan and delegate", names)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -109,8 +109,8 @@ func TestBuiltinsAccessor(t *testing.T) {
|
||||
WithRegistry(registry.NewMemoryRegistry()),
|
||||
)
|
||||
|
||||
if len(tools) != 3 {
|
||||
t.Fatalf("Builtins() returned %d tools, want 3", len(tools))
|
||||
if len(tools) != 2 {
|
||||
t.Fatalf("Builtins() returned %d tools, want 2", len(tools))
|
||||
}
|
||||
|
||||
// A name that isn't a built-in falls through (ok == false).
|
||||
|
||||
@@ -13,7 +13,6 @@ import (
|
||||
const (
|
||||
agentAskStep = "ask"
|
||||
agentApprovalStep = "approval"
|
||||
agentInputStep = "input-required"
|
||||
)
|
||||
|
||||
func (a *agentImpl) newCheckpointRun(runID, message, parentRunID string, existing *flow.Run) flow.Run {
|
||||
@@ -52,17 +51,6 @@ func (a *agentImpl) saveRun(ctx context.Context, run flow.Run) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Resume returns the response for a checkpointed agent run. Completed runs are
|
||||
// returned from the checkpoint without calling the model or replaying tool
|
||||
// calls; failed or in-progress runs continue from the saved input message.
|
||||
func Resume(ctx context.Context, ag Agent, runID string) (*Response, error) {
|
||||
a, ok := ag.(*agentImpl)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("agent resume: unsupported agent implementation %T", ag)
|
||||
}
|
||||
return a.resume(ctx, runID)
|
||||
}
|
||||
|
||||
func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error) {
|
||||
if a.opts.Checkpoint == nil {
|
||||
return nil, fmt.Errorf("agent %s has no checkpoint configured", a.opts.Name)
|
||||
@@ -75,9 +63,6 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
|
||||
return nil, fmt.Errorf("agent run %s not found", runID)
|
||||
}
|
||||
if run.Status == "paused" {
|
||||
if run.State.Stage == agentInputStep {
|
||||
return nil, fmt.Errorf("agent run %s is input-required; resume with ResumeInput", runID)
|
||||
}
|
||||
run.Status = "running"
|
||||
run.State.Stage = agentAskStep
|
||||
}
|
||||
@@ -98,51 +83,6 @@ func (a *agentImpl) resume(ctx context.Context, runID string) (*Response, error)
|
||||
return a.askLocked(ctx, run.ID, message, parentID, &run)
|
||||
}
|
||||
|
||||
// ResumeInput resumes a checkpointed agent run that paused via the built-in
|
||||
// request_input tool. The supplied input is appended to the original request so
|
||||
// the same run can continue with durable checkpoint and completed tool history.
|
||||
func ResumeInput(ctx context.Context, ag Agent, runID, input string) (*Response, error) {
|
||||
a, ok := ag.(*agentImpl)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("agent resume input: unsupported agent implementation %T", ag)
|
||||
}
|
||||
return a.resumeInput(ctx, runID, input)
|
||||
}
|
||||
|
||||
func (a *agentImpl) resumeInput(ctx context.Context, runID, input string) (*Response, error) {
|
||||
if a.opts.Checkpoint == nil {
|
||||
return nil, fmt.Errorf("agent %s has no checkpoint configured", a.opts.Name)
|
||||
}
|
||||
run, ok, err := a.opts.Checkpoint.Load(ctx, runID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("agent run %s not found", runID)
|
||||
}
|
||||
if run.Status != "paused" || run.State.Stage != agentInputStep {
|
||||
return nil, fmt.Errorf("agent run %s is not waiting for human input", runID)
|
||||
}
|
||||
var p inputPause
|
||||
if err := run.State.Scan(&p); err != nil {
|
||||
return nil, fmt.Errorf("agent run %s input state decode: %w", runID, err)
|
||||
}
|
||||
message := p.OriginalMessage
|
||||
if message == "" {
|
||||
message = string(run.State.Data)
|
||||
}
|
||||
message += "\n\nHuman input: " + input
|
||||
run.Status = "running"
|
||||
run.State.Stage = agentAskStep
|
||||
run.State.Data = []byte(message)
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
if a.model == nil {
|
||||
a.setup()
|
||||
}
|
||||
return a.askLocked(ctx, run.ID, message, run.ParentID, &run)
|
||||
}
|
||||
|
||||
func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) {
|
||||
if a.opts.Checkpoint == nil {
|
||||
return nil, nil
|
||||
|
||||
@@ -3,7 +3,6 @@ package agent
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"go-micro.dev/v6/ai"
|
||||
@@ -119,64 +118,6 @@ func TestPendingReturnsUnfinishedAgentRuns(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHumanInputPauseResumesSameRunWithInput(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cp := flow.StoreCheckpoint(store.NewStore(), "input-agent")
|
||||
calls := 0
|
||||
fakeGen = func(ctx context.Context, opts ai.Options, req *ai.Request) (*ai.Response, error) {
|
||||
calls++
|
||||
if calls == 1 {
|
||||
if opts.ToolHandler != nil {
|
||||
opts.ToolHandler(ctx, ai.ToolCall{ID: "input-1", Name: toolHumanInput, Input: map[string]any{"prompt": "Which region should I deploy to?"}})
|
||||
}
|
||||
return &ai.Response{Reply: "waiting"}, nil
|
||||
}
|
||||
if !strings.Contains(req.Prompt, "Human input: us-east-1") {
|
||||
t.Fatalf("resumed prompt = %q, want human input", req.Prompt)
|
||||
}
|
||||
return &ai.Response{Reply: "deploying to us-east-1"}, nil
|
||||
}
|
||||
defer func() { fakeGen = nil }()
|
||||
|
||||
a := newTestAgent(Name("input-agent"), WithCheckpoint(cp))
|
||||
_, err := a.Ask(ctx, "deploy the service")
|
||||
if err == nil {
|
||||
t.Fatal("Ask succeeded, want input-required pause")
|
||||
}
|
||||
runs, err := Pending(ctx, a)
|
||||
if err != nil {
|
||||
t.Fatalf("Pending: %v", err)
|
||||
}
|
||||
if len(runs) != 1 || runs[0].Status != "paused" || runs[0].State.Stage != agentInputStep {
|
||||
t.Fatalf("paused runs = %#v, want one input-required run", runs)
|
||||
}
|
||||
var pause inputPause
|
||||
if err := runs[0].State.Scan(&pause); err != nil {
|
||||
t.Fatalf("Scan pause: %v", err)
|
||||
}
|
||||
if pause.OriginalMessage != "deploy the service" || pause.Prompt != "Which region should I deploy to?" {
|
||||
t.Fatalf("pause = %#v", pause)
|
||||
}
|
||||
|
||||
if _, err := Resume(ctx, a, runs[0].ID); err == nil || !strings.Contains(err.Error(), "ResumeInput") {
|
||||
t.Fatalf("Resume input-required err = %v, want guidance", err)
|
||||
}
|
||||
resp, err := ResumeInput(ctx, a, runs[0].ID, "us-east-1")
|
||||
if err != nil {
|
||||
t.Fatalf("ResumeInput: %v", err)
|
||||
}
|
||||
if resp.RunID != runs[0].ID || resp.Reply != "deploying to us-east-1" {
|
||||
t.Fatalf("response = %#v", resp)
|
||||
}
|
||||
loaded, ok, err := cp.Load(ctx, runs[0].ID)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("Load resumed run ok=%v err=%v", ok, err)
|
||||
}
|
||||
if loaded.Status != "done" {
|
||||
t.Fatalf("resumed run status = %q, want done", loaded.Status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApprovalDenialPausesCheckpointedRunAndResumeContinues(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cp := flow.StoreCheckpoint(store.NewStore(), "approval-agent")
|
||||
|
||||
+8
-34
@@ -3,7 +3,6 @@ package agent
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
@@ -124,31 +123,17 @@ func (m *storeMemory) Recall(query string, limit int) []ai.Message {
|
||||
limit = 5
|
||||
}
|
||||
terms := recallTerms(query)
|
||||
type match struct {
|
||||
msg ai.Message
|
||||
score int
|
||||
index int
|
||||
}
|
||||
matches := make([]match, 0, len(m.archive))
|
||||
for i := len(m.archive) - 1; i >= 0; i-- {
|
||||
var out []ai.Message
|
||||
for i := len(m.archive) - 1; i >= 0 && len(out) < limit; i-- {
|
||||
msg := m.archive[i]
|
||||
if score := recallScore(msg, terms); score > 0 {
|
||||
matches = append(matches, match{msg: msg, score: score, index: i})
|
||||
text := strings.ToLower(fmt.Sprint(msg.Content))
|
||||
for _, term := range terms {
|
||||
if strings.Contains(text, term) {
|
||||
out = append(out, msg)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
sort.SliceStable(matches, func(i, j int) bool {
|
||||
if matches[i].score != matches[j].score {
|
||||
return matches[i].score > matches[j].score
|
||||
}
|
||||
return matches[i].index > matches[j].index
|
||||
})
|
||||
if len(matches) > limit {
|
||||
matches = matches[:limit]
|
||||
}
|
||||
out := make([]ai.Message, 0, len(matches))
|
||||
for _, match := range matches {
|
||||
out = append(out, match.msg)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -243,17 +228,6 @@ func compactText(s string, max int) string {
|
||||
return s
|
||||
}
|
||||
|
||||
func recallScore(msg ai.Message, terms []string) int {
|
||||
text := strings.ToLower(fmt.Sprint(msg.Content))
|
||||
score := 0
|
||||
for _, term := range terms {
|
||||
if strings.Contains(text, term) {
|
||||
score++
|
||||
}
|
||||
}
|
||||
return score
|
||||
}
|
||||
|
||||
func recallTerms(query string) []string {
|
||||
seen := map[string]bool{}
|
||||
var terms []string
|
||||
|
||||
@@ -62,46 +62,6 @@ func TestWithMemoryUsed(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompactingMemoryRecallRanksSpecificMatches(t *testing.T) {
|
||||
m := NewCompactingMemory(store.NewMemoryStore(), "agent/rank/history", 3, 1).(MemoryRecall)
|
||||
writer := m.(Memory)
|
||||
writer.Add("user", "alpha budget is 42")
|
||||
writer.Add("assistant", "noted")
|
||||
writer.Add("user", "beta budget is 7")
|
||||
writer.Add("assistant", "noted")
|
||||
writer.Add("user", "alpha owner is sam")
|
||||
|
||||
recalled := m.Recall("alpha budget", 2)
|
||||
if len(recalled) == 0 {
|
||||
t.Fatal("expected recalled messages")
|
||||
}
|
||||
if got := recalled[0].Content.(string); !strings.Contains(got, "alpha budget is 42") {
|
||||
t.Fatalf("top recall = %q, want alpha budget match", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompactingMemoryArchivePersistsAndReloads(t *testing.T) {
|
||||
st := store.NewMemoryStore()
|
||||
m := NewCompactingMemory(st, "agent/reload/history", 3, 1)
|
||||
m.Add("user", "alpha budget is 42")
|
||||
m.Add("assistant", "noted")
|
||||
m.Add("user", "beta budget is 7")
|
||||
m.Add("assistant", "noted")
|
||||
|
||||
reloaded := NewCompactingMemory(st, "agent/reload/history", 3, 1)
|
||||
recall, ok := reloaded.(MemoryRecall)
|
||||
if !ok {
|
||||
t.Fatal("compacting memory should support recall")
|
||||
}
|
||||
recalled := recall.Recall("alpha budget", 1)
|
||||
if len(recalled) != 1 {
|
||||
t.Fatalf("recalled %d messages, want 1", len(recalled))
|
||||
}
|
||||
if got := recalled[0].Content.(string); !strings.Contains(got, "alpha budget is 42") {
|
||||
t.Fatalf("reloaded recall = %q, want alpha budget", got)
|
||||
}
|
||||
}
|
||||
|
||||
// A custom tool is offered to the model and dispatched to its handler.
|
||||
func TestWithToolExposedAndDispatched(t *testing.T) {
|
||||
var got map[string]any
|
||||
|
||||
@@ -77,9 +77,11 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
// Build initial request
|
||||
apiReq := map[string]any{
|
||||
"model": p.opts.Model,
|
||||
"max_tokens": anthropicMaxTokens(p.opts),
|
||||
"max_tokens": 8192,
|
||||
"system": req.SystemPrompt,
|
||||
"messages": threadAnthropicMessages(req),
|
||||
"messages": []map[string]any{
|
||||
{"role": "user", "content": req.Prompt},
|
||||
},
|
||||
}
|
||||
|
||||
if len(anthropicTools) > 0 {
|
||||
@@ -99,9 +101,10 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
|
||||
// Tool execution loop: execute tools, send results back, repeat
|
||||
// until the model responds with text only (no more tool calls)
|
||||
messages := append(threadAnthropicMessages(req),
|
||||
map[string]any{"role": "assistant", "content": cleanContent(rawContent)},
|
||||
)
|
||||
messages := []map[string]any{
|
||||
{"role": "user", "content": req.Prompt},
|
||||
{"role": "assistant", "content": cleanContent(rawContent)},
|
||||
}
|
||||
|
||||
pendingCalls := resp.ToolCalls
|
||||
|
||||
@@ -124,7 +127,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
|
||||
followUpReq := map[string]any{
|
||||
"model": p.opts.Model,
|
||||
"max_tokens": anthropicMaxTokens(p.opts),
|
||||
"max_tokens": 8192,
|
||||
"system": req.SystemPrompt,
|
||||
"messages": messages,
|
||||
}
|
||||
@@ -267,25 +270,3 @@ func cleanContent(raw any) any {
|
||||
}
|
||||
return cleaned
|
||||
}
|
||||
|
||||
|
||||
// threadAnthropicMessages builds the Anthropic messages array from the
|
||||
// conversation history (req.Messages) followed by the current prompt. The
|
||||
// system prompt is sent separately via the top-level "system" field.
|
||||
func threadAnthropicMessages(req *ai.Request) []map[string]any {
|
||||
msgs := make([]map[string]any, 0, len(req.Messages)+1)
|
||||
for _, m := range req.Messages {
|
||||
msgs = append(msgs, map[string]any{"role": m.Role, "content": m.Content})
|
||||
}
|
||||
if req.Prompt != "" {
|
||||
msgs = append(msgs, map[string]any{"role": "user", "content": req.Prompt})
|
||||
}
|
||||
return msgs
|
||||
}
|
||||
|
||||
func anthropicMaxTokens(o ai.Options) int {
|
||||
if o.MaxTokens > 0 {
|
||||
return o.MaxTokens
|
||||
}
|
||||
return 8192
|
||||
}
|
||||
|
||||
@@ -91,21 +91,13 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
|
||||
messages := []map[string]any{
|
||||
{"role": "system", "content": req.SystemPrompt},
|
||||
}
|
||||
for _, m := range req.Messages {
|
||||
messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
|
||||
}
|
||||
if req.Prompt != "" {
|
||||
messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
|
||||
{"role": "user", "content": req.Prompt},
|
||||
}
|
||||
|
||||
apiReq := map[string]any{
|
||||
"model": p.opts.Model,
|
||||
"messages": messages,
|
||||
}
|
||||
if p.opts.MaxTokens > 0 {
|
||||
apiReq["max_tokens"] = p.opts.MaxTokens
|
||||
}
|
||||
|
||||
if len(tools) > 0 {
|
||||
apiReq["tools"] = tools
|
||||
|
||||
+2
-18
@@ -85,12 +85,7 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
// Build messages
|
||||
messages := []map[string]any{
|
||||
{"role": "system", "content": req.SystemPrompt},
|
||||
}
|
||||
for _, m := range req.Messages {
|
||||
messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
|
||||
}
|
||||
if req.Prompt != "" {
|
||||
messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
|
||||
{"role": "user", "content": req.Prompt},
|
||||
}
|
||||
|
||||
// Build initial request
|
||||
@@ -98,9 +93,6 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
"model": p.opts.Model,
|
||||
"messages": messages,
|
||||
}
|
||||
if p.opts.MaxTokens > 0 {
|
||||
apiReq["max_tokens"] = p.opts.MaxTokens
|
||||
}
|
||||
|
||||
if len(openaiTools) > 0 {
|
||||
apiReq["tools"] = openaiTools
|
||||
@@ -154,21 +146,13 @@ func (p *Provider) Generate(ctx context.Context, req *ai.Request, opts ...ai.Gen
|
||||
func (p *Provider) Stream(ctx context.Context, req *ai.Request, opts ...ai.GenerateOption) (ai.Stream, error) {
|
||||
messages := []map[string]any{
|
||||
{"role": "system", "content": req.SystemPrompt},
|
||||
}
|
||||
for _, m := range req.Messages {
|
||||
messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
|
||||
}
|
||||
if req.Prompt != "" {
|
||||
messages = append(messages, map[string]any{"role": "user", "content": req.Prompt})
|
||||
{"role": "user", "content": req.Prompt},
|
||||
}
|
||||
apiReq := map[string]any{
|
||||
"model": p.opts.Model,
|
||||
"messages": messages,
|
||||
"stream": true,
|
||||
}
|
||||
if p.opts.MaxTokens > 0 {
|
||||
apiReq["max_tokens"] = p.opts.MaxTokens
|
||||
}
|
||||
reqBody, err := json.Marshal(apiReq)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to marshal stream request: %w", err)
|
||||
|
||||
@@ -16,8 +16,6 @@ type Options struct {
|
||||
BaseURL string
|
||||
// ToolHandler handles tool calls (optional, for automatic tool execution)
|
||||
ToolHandler ToolHandler
|
||||
// MaxTokens caps the length of the response (0 = provider default)
|
||||
MaxTokens int
|
||||
}
|
||||
|
||||
// GenerateOptions for generate call
|
||||
@@ -93,11 +91,3 @@ func WithTools(t *Tools) Option {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithMaxTokens caps the number of tokens in the response. 0 leaves the
|
||||
// provider default in place.
|
||||
func WithMaxTokens(n int) Option {
|
||||
return func(o *Options) {
|
||||
o.MaxTokens = n
|
||||
}
|
||||
}
|
||||
|
||||
@@ -42,28 +42,14 @@ func Deploy(c *cli.Context) error {
|
||||
return showDeployHelp()
|
||||
}
|
||||
|
||||
target, remotePath := resolveDeployTarget(c, target, cfg)
|
||||
|
||||
return deploySSH(c, target, cfg, remotePath)
|
||||
}
|
||||
|
||||
func resolveDeployTarget(c *cli.Context, target string, cfg *config.Config) (string, string) {
|
||||
remotePath := c.String("path")
|
||||
if remotePath == "" {
|
||||
remotePath = defaultRemotePath
|
||||
}
|
||||
|
||||
// Check if target is a named target from config
|
||||
if cfg != nil {
|
||||
if dt, ok := cfg.Deploy[target]; ok {
|
||||
target = dt.SSH
|
||||
if dt.Path != "" && !c.IsSet("path") {
|
||||
remotePath = dt.Path
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return target, remotePath
|
||||
return deploySSH(c, target, cfg)
|
||||
}
|
||||
|
||||
func showDeployHelp() error {
|
||||
@@ -96,7 +82,7 @@ func showDeployTargets(cfg *config.Config) error {
|
||||
return fmt.Errorf("%s", sb.String())
|
||||
}
|
||||
|
||||
func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath string) error {
|
||||
func deploySSH(c *cli.Context, target string, cfg *config.Config) error {
|
||||
dir := c.Args().Get(1)
|
||||
if dir == "" {
|
||||
dir = "."
|
||||
@@ -112,6 +98,7 @@ func deploySSH(c *cli.Context, target string, cfg *config.Config, remotePath str
|
||||
cfg, _ = config.Load(absDir)
|
||||
}
|
||||
|
||||
remotePath := c.String("path")
|
||||
if remotePath == "" {
|
||||
remotePath = defaultRemotePath
|
||||
}
|
||||
|
||||
@@ -1,120 +0,0 @@
|
||||
package deploy
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/urfave/cli/v2"
|
||||
"go-micro.dev/v6/cmd/micro/run/config"
|
||||
)
|
||||
|
||||
func newDeployTestContext(t *testing.T, args ...string) *cli.Context {
|
||||
t.Helper()
|
||||
set := flag.NewFlagSet("deploy", flag.ContinueOnError)
|
||||
set.String("path", defaultRemotePath, "")
|
||||
set.String("ssh", "", "")
|
||||
set.String("service", "", "")
|
||||
set.Bool("build", false, "")
|
||||
if err := set.Parse(args); err != nil {
|
||||
t.Fatalf("parse flags: %v", err)
|
||||
}
|
||||
return cli.NewContext(cli.NewApp(), set, nil)
|
||||
}
|
||||
|
||||
func TestDeployNoTargetExplainsInitAndDeployHandoff(t *testing.T) {
|
||||
err := showDeployHelp()
|
||||
if err == nil {
|
||||
t.Fatal("expected missing target guidance")
|
||||
}
|
||||
msg := err.Error()
|
||||
for _, want := range []string{
|
||||
"no deployment target specified",
|
||||
"sudo micro init --server",
|
||||
"micro deploy user@your-server",
|
||||
"deploy prod",
|
||||
} {
|
||||
if !strings.Contains(msg, want) {
|
||||
t.Fatalf("missing %q in guidance:\n%s", want, msg)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeployListsConfiguredTargetsWhenNoTargetProvided(t *testing.T) {
|
||||
err := showDeployTargets(&config.Config{Deploy: map[string]*config.DeployTarget{
|
||||
"prod": {Name: "prod", SSH: "deploy@prod.example.com"},
|
||||
"staging": {Name: "staging", SSH: "deploy@staging.example.com"},
|
||||
}})
|
||||
if err == nil {
|
||||
t.Fatal("expected configured target guidance")
|
||||
}
|
||||
msg := err.Error()
|
||||
for _, want := range []string{
|
||||
"Available deploy targets:",
|
||||
"prod -> deploy@prod.example.com",
|
||||
"staging -> deploy@staging.example.com",
|
||||
"micro deploy <target>",
|
||||
} {
|
||||
if !strings.Contains(msg, want) {
|
||||
t.Fatalf("missing %q in configured target guidance:\n%s", want, msg)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveDeployTargetUsesConfigTargetAndPath(t *testing.T) {
|
||||
ctx := newDeployTestContext(t, "prod")
|
||||
cfg := &config.Config{Deploy: map[string]*config.DeployTarget{
|
||||
"prod": {Name: "prod", SSH: "deploy@prod.example.com", Path: "/srv/micro"},
|
||||
}}
|
||||
|
||||
target, remotePath := resolveDeployTarget(ctx, ctx.Args().First(), cfg)
|
||||
if target != "deploy@prod.example.com" {
|
||||
t.Fatalf("target = %q, want configured SSH", target)
|
||||
}
|
||||
if remotePath != "/srv/micro" {
|
||||
t.Fatalf("remotePath = %q, want configured path", remotePath)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveDeployTargetAllowsCLIPathOverride(t *testing.T) {
|
||||
ctx := newDeployTestContext(t, "--path", "/tmp/micro", "prod")
|
||||
cfg := &config.Config{Deploy: map[string]*config.DeployTarget{
|
||||
"prod": {Name: "prod", SSH: "deploy@prod.example.com", Path: "/srv/micro"},
|
||||
}}
|
||||
|
||||
target, remotePath := resolveDeployTarget(ctx, ctx.Args().First(), cfg)
|
||||
if target != "deploy@prod.example.com" {
|
||||
t.Fatalf("target = %q, want configured SSH", target)
|
||||
}
|
||||
if remotePath != "/tmp/micro" {
|
||||
t.Fatalf("remotePath = %q, want CLI override", remotePath)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeployConfigParserSupportsDeployTargets(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := dir + "/micro.mu"
|
||||
content := `service api
|
||||
path ./api
|
||||
|
||||
deploy prod
|
||||
ssh deploy@prod.example.com
|
||||
path /srv/micro
|
||||
`
|
||||
if err := os.WriteFile(path, []byte(content), 0644); err != nil {
|
||||
t.Fatalf("write config: %v", err)
|
||||
}
|
||||
|
||||
cfg, err := config.ParseMu(path)
|
||||
if err != nil {
|
||||
t.Fatalf("parse config: %v", err)
|
||||
}
|
||||
prod := cfg.Deploy["prod"]
|
||||
if prod == nil {
|
||||
t.Fatal("missing prod deploy target")
|
||||
}
|
||||
if prod.SSH != "deploy@prod.example.com" || prod.Path != "/srv/micro" {
|
||||
t.Fatalf("deploy target = %#v", prod)
|
||||
}
|
||||
}
|
||||
@@ -1,36 +0,0 @@
|
||||
# Agent Human Input Pause/Resume
|
||||
|
||||
Agents can pause a durable run when the model needs a human decision before it
|
||||
can continue. This keeps the services → agents → workflows lifecycle in one
|
||||
runtime: services expose tools, the agent decides it needs operator input, and
|
||||
the same checkpointed run resumes once that input arrives.
|
||||
|
||||
## Pattern
|
||||
|
||||
```go
|
||||
cp := flow.StoreCheckpoint(nil, "deploy-agent")
|
||||
ag := agent.New(
|
||||
agent.Name("deploy-agent"),
|
||||
agent.WithCheckpoint(cp),
|
||||
)
|
||||
|
||||
resp, err := ag.Ask(ctx, "Deploy the service")
|
||||
if err != nil {
|
||||
// If the model called the built-in request_input tool, the run is saved as
|
||||
// paused/input-required instead of losing state or completing early.
|
||||
pending, _ := agent.Pending(ctx, ag)
|
||||
runID := pending[0].ID
|
||||
|
||||
// Later, after an operator supplies the missing answer, the same run ID
|
||||
// continues with the original prompt, human input, memory, and completed
|
||||
// tool history intact.
|
||||
resp, err = agent.ResumeInput(ctx, ag, runID, "Deploy to us-east-1")
|
||||
}
|
||||
_ = resp
|
||||
```
|
||||
|
||||
The model sees a built-in `request_input` tool with a `prompt` argument. When it
|
||||
calls that tool, Go Micro persists the run with status `paused` and stage
|
||||
`input-required`. Plain `agent.Resume` continues to support completed, failed,
|
||||
and approval-paused runs; input-required runs are resumed with
|
||||
`agent.ResumeInput` so the human response is explicit.
|
||||
+19
-31
@@ -212,39 +212,37 @@ func waitFor(reg registry.Registry, names ...string) {
|
||||
}
|
||||
}
|
||||
|
||||
func runSupport(provider string) error {
|
||||
func main() {
|
||||
provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, ...")
|
||||
flag.Parse()
|
||||
|
||||
apiKey := ""
|
||||
if provider == "mock" {
|
||||
if *provider == "mock" {
|
||||
ai.Register("mock", newMock)
|
||||
} else if apiKey = providerKey(provider); apiKey == "" {
|
||||
return fmt.Errorf("no API key for provider %q — set MICRO_AI_API_KEY or the provider's key env", provider)
|
||||
} else if apiKey = providerKey(*provider); apiKey == "" {
|
||||
fmt.Printf("no API key for provider %q — set MICRO_AI_API_KEY or the provider's key env\n", *provider)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
fmt.Printf("\n\033[1mSupport desk (provider: %s)\033[0m\n\n", provider)
|
||||
fmt.Printf("\n\033[1mSupport desk (provider: %s)\033[0m\n\n", *provider)
|
||||
|
||||
// Shared in-memory infrastructure so the demo runs in one process.
|
||||
reg := registry.NewMemoryRegistry()
|
||||
br := broker.NewMemoryBroker()
|
||||
if err := br.Connect(); err != nil {
|
||||
return fmt.Errorf("broker connect: %w", err)
|
||||
fmt.Println("broker connect:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
cl := client.NewClient(client.Registry(reg), client.Selector(selector.NewSelector(selector.Registry(reg))))
|
||||
|
||||
// Services.
|
||||
tickets := new(TicketService)
|
||||
notify := new(NotifyService)
|
||||
var services []service.Service
|
||||
for name, h := range map[string]any{"customers": new(CustomerService), "tickets": tickets, "notify": notify} {
|
||||
svc := service.New(service.Name(name), service.Address("127.0.0.1:0"), service.Registry(reg), service.Client(cl), service.HandleSignal(false))
|
||||
svc := service.New(service.Name(name), service.Registry(reg), service.Client(cl))
|
||||
_ = svc.Handle(h)
|
||||
services = append(services, svc)
|
||||
go svc.Run()
|
||||
}
|
||||
defer func() {
|
||||
for _, svc := range services {
|
||||
_ = svc.Server().Stop()
|
||||
}
|
||||
}()
|
||||
|
||||
// The support agent manages the three services. The approval gate is
|
||||
// the human-in-the-loop: it can read and triage freely, but emailing a
|
||||
@@ -252,11 +250,10 @@ func runSupport(provider string) error {
|
||||
// it for a person or a policy; here we approve and log.
|
||||
support := agent.New(
|
||||
agent.Name("support"),
|
||||
agent.Address("127.0.0.1:0"),
|
||||
agent.Services("customers", "tickets", "notify"),
|
||||
agent.Prompt("You are a support agent. For each ticket, look up the customer, set an "+
|
||||
"appropriate priority, and reply to them. Escalate billing issues."),
|
||||
agent.Provider(provider), agent.APIKey(apiKey),
|
||||
agent.Provider(*provider), agent.APIKey(apiKey),
|
||||
agent.ApproveTool(func(tool string, input map[string]any) (bool, string) {
|
||||
if strings.Contains(tool, "Send") {
|
||||
fmt.Printf(" \033[33m▣ approval gate\033[0m %s(%v) — approved\n", tool, input["to"])
|
||||
@@ -278,7 +275,8 @@ func runSupport(provider string) error {
|
||||
flow.Prompt("A new support ticket arrived: {{.Data}}. Handle it."),
|
||||
)
|
||||
if err := intake.Register(reg, br, cl); err != nil {
|
||||
return fmt.Errorf("flow register: %w", err)
|
||||
fmt.Println("flow register:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer intake.Stop()
|
||||
|
||||
@@ -289,7 +287,8 @@ func runSupport(provider string) error {
|
||||
fmt.Println("\033[1m> event:\033[0m events.ticket.created", string(body))
|
||||
fmt.Println()
|
||||
if err := br.Publish("events.ticket.created", &broker.Message{Body: body}); err != nil {
|
||||
return fmt.Errorf("publish: %w", err)
|
||||
fmt.Println("publish:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Wait for the agent to act.
|
||||
@@ -306,18 +305,7 @@ func runSupport(provider string) error {
|
||||
}
|
||||
if notify.sent >= 1 {
|
||||
fmt.Println("\n\033[32m✓ ticket triaged and the customer was replied to — triggered by an event\033[0m")
|
||||
return nil
|
||||
}
|
||||
fmt.Println("\n\033[31m✗ the agent did not complete the triage\033[0m")
|
||||
return fmt.Errorf("support agent did not complete triage")
|
||||
}
|
||||
|
||||
func main() {
|
||||
provider := flag.String("provider", "mock", "LLM provider: mock (default), anthropic, openai, ...")
|
||||
flag.Parse()
|
||||
|
||||
if err := runSupport(*provider); err != nil {
|
||||
fmt.Println(err)
|
||||
os.Exit(1)
|
||||
} else {
|
||||
fmt.Println("\n\033[31m✗ the agent did not complete the triage\033[0m")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,9 +0,0 @@
|
||||
package main
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestRunSupportMockSmoke(t *testing.T) {
|
||||
if err := runSupport("mock"); err != nil {
|
||||
t.Fatalf("support example failed: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -19,11 +19,9 @@ redirect the loop; direction always wins.
|
||||
items the loop can auto-merge): brand/positioning copy, breaking public-API
|
||||
changes, architectural rewrites. Those go to the human.
|
||||
|
||||
## Developer experience (ranked)
|
||||
## Later (ranked)
|
||||
|
||||
1. **Schedule cross-provider agent conformance** ([#3295](https://github.com/micro/go-micro/issues/3295)) — after the 0→hero reference and provider-focused unit coverage shipped, the highest-value remaining Now-roadmap hardening gap is a scheduled, key-gated provider matrix that proves the same agent/tool workflow keeps working across supported models without blocking contributors who lack secrets.
|
||||
2. **Harden agent failure and cancellation semantics** ([#3296](https://github.com/micro/go-micro/issues/3296)) — the harness is increasingly durable, streaming, observable, and human-in-the-loop; the next operability seam is making timeouts, cancellation, rate-limit errors, and retry/backoff behavior predictable across agent, AI provider, service-tool, and flow boundaries.
|
||||
3. **Expose run inspection in the CLI inner loop** ([#3297](https://github.com/micro/go-micro/issues/3297)) — the canon promises scaffold → run → chat → inspect → deploy, and recent work has strengthened scaffold/run/chat/deploy; inspection remains the most visible DX gap for turning agent/flow activity into actionable breadcrumbs during local development.
|
||||
1. **Add agent memory summarization and retrieval** ([#3273](https://github.com/micro/go-micro/issues/3273)) — with A2A resubscribe/input-required handoffs now shipped, the highest-value remaining roadmap gap is durable memory beyond a fixed context buffer: compact older turns/tool results and retrieve relevant prior facts using the existing agent/store primitives so longer-running agents remain coherent without becoming a separate product layer.
|
||||
|
||||
_Seeded by Claude Code from the roadmap + open issues; thereafter maintained by the
|
||||
architecture-review pass._
|
||||
|
||||
Reference in New Issue
Block a user