项目文件夹

文件
T
Asim Aslam e416ea4a75 Enhance agent workflows with guardrails and documentation updates (#2952)
* docs: map go-micro onto Anthropic's workflows-vs-agents taxonomy

- new guide 'Agents and Workflows': adopts Anthropic's Building Effective
  Agents vocabulary — workflow (predefined path) = flow, agent (dynamic
  self-direction) = agent — maps the augmented-LLM building block and the
  five workflow patterns onto go-micro, and shows routing (chat router)
  and orchestrator-workers (conductor + plan/delegate) are already native.
- flow package doc reframed as a workflow (predefined path) per the same
  taxonomy, with guidance on flow vs agent.
- nav + README link the new guide.

* feat: agent guardrails — step limit and tool approval hook

Anthropic's Building Effective Agents stresses stopping conditions and
human-in-the-loop checkpoints for autonomous agents. Add both as plain
options enforced at the tool-handler choke point — no provider changes,
no new abstraction:

- MaxSteps(n): bound tool executions per Ask; beyond the limit, actions
  are refused and the model is told to stop and summarize.
- ApproveTool(fn): gate each action before it runs; returning false
  blocks it and surfaces the reason to the model. The internal plan tool
  is never gated.

Exposed at the micro package (AgentMaxSteps, AgentApproveTool, ApproveFunc).
Tests cover the limit, blocking, and that plan is not gated. Guardrails
section of the agents-and-workflows guide updated from 'active work' to
documented options.

* feat: flow can dispatch to an agent (flow triggers, agent reasons)

Unify the engine without collapsing the workflow/agent distinction. A
Flow with Agent set hands each event's rendered prompt to a named
registered agent over RPC (Agent.Chat) instead of running its own LLM
step — so the workflow stays the deterministic trigger and the agent is
the reasoning engine, with its plan, delegate, memory, and guardrails.
A plain flow is unchanged (single augmented-LLM step).

- flow.Agent(name) / micro.FlowAgent(name); flow stores the client and
  skips model setup when dispatching.
- test: dispatch routes to comms.Agent.Chat with the rendered prompt and
  records the reply.
- guide: 'Flow triggers, Agent reasons' section.

---------

Co-authored-by: Claude <noreply@anthropic.com>
2026-06-08 08:32:32 +01:00

319 行
7.1 KiB
Go

// Package agent provides the Agent abstraction for Go Micro.
//
// An Agent is a service with an LLM inside it. It registers a Chat
// RPC endpoint, discovers its assigned services' tools, and
// orchestrates them intelligently.
//
// agent := micro.NewAgent("task-mgr",
// micro.AgentServices("task"),
// micro.AgentPrompt("You manage tasks."),
// micro.AgentProvider("anthropic"),
// )
// agent.Run()
package agent
import (
"context"
"encoding/json"
"fmt"
"strings"
"sync"
pb "go-micro.dev/v5/agent/proto"
"go-micro.dev/v5/ai"
"go-micro.dev/v5/server"
"go-micro.dev/v5/store"
_ "go-micro.dev/v5/ai/anthropic"
_ "go-micro.dev/v5/ai/atlascloud"
_ "go-micro.dev/v5/ai/gemini"
_ "go-micro.dev/v5/ai/groq"
_ "go-micro.dev/v5/ai/mistral"
_ "go-micro.dev/v5/ai/openai"
_ "go-micro.dev/v5/ai/together"
)
// Agent is the interface for an AI agent that manages services.
type Agent interface {
Name() string
Init(...Option)
Options() Options
Ask(ctx context.Context, message string) (*Response, error)
Run() error
Stop() error
String() string
}
// Response is what an agent returns from Chat.
type Response struct {
Reply string
ToolCalls []ai.ToolCall
Agent string
}
type agentImpl struct {
opts Options
model ai.Model
tools *ai.Tools
hist *ai.History
server server.Server
mu sync.Mutex
// ephemeral marks a short-lived sub-agent created by delegation.
// Ephemeral agents run with an isolated context: they load and
// persist no history, and have no built-in tools (so they cannot
// plan or re-delegate).
ephemeral bool
// steps counts tool executions in the current Ask, for MaxSteps.
steps int
}
// New creates a new Agent.
func New(opts ...Option) Agent {
return &agentImpl{
opts: newOptions(opts...),
}
}
// newEphemeral creates a short-lived sub-agent for a delegated subtask.
// It shares the parent's provider, model, and infrastructure but runs
// with an isolated context: it loads and persists no history and has no
// built-in tools (so it can neither plan nor re-delegate). Returns the
// concrete type because ephemeral is an internal construction detail,
// not a public option.
func newEphemeral(opts ...Option) *agentImpl {
return &agentImpl{
opts: newOptions(opts...),
ephemeral: true,
}
}
func (a *agentImpl) Name() string {
return a.opts.Name
}
func (a *agentImpl) Init(opts ...Option) {
for _, o := range opts {
o(&a.opts)
}
a.setup()
}
func (a *agentImpl) Options() Options {
return a.opts
}
func (a *agentImpl) String() string {
return "agent"
}
func (a *agentImpl) setup() {
var modelOpts []ai.Option
modelOpts = append(modelOpts, ai.WithAPIKey(a.opts.APIKey))
if a.opts.Model != "" {
modelOpts = append(modelOpts, ai.WithModel(a.opts.Model))
}
a.tools = ai.NewTools(a.opts.Registry, ai.ToolClient(a.opts.Client))
modelOpts = append(modelOpts, ai.WithToolHandler(a.toolHandler()))
a.model = ai.New(a.opts.Provider, modelOpts...)
a.hist = ai.NewHistory(a.opts.HistoryLimit)
if !a.ephemeral {
a.loadHistory()
}
}
// Ask sends a message and returns the agent's response.
// This is the programmatic API for direct use.
func (a *agentImpl) Ask(ctx context.Context, message string) (*Response, error) {
a.mu.Lock()
defer a.mu.Unlock()
if a.model == nil {
a.setup()
}
toolList, err := a.discoverTools()
if err != nil {
return nil, fmt.Errorf("discover tools: %w", err)
}
a.hist.Add("user", message)
a.steps = 0
resp, err := a.model.Generate(ctx, &ai.Request{
Prompt: message,
SystemPrompt: a.buildPrompt(),
Tools: toolList,
Messages: a.hist.Messages(),
})
if err != nil {
return nil, err
}
if resp.Reply != "" {
a.hist.Add("assistant", resp.Reply)
}
if resp.Answer != "" {
a.hist.Add("assistant", resp.Answer)
}
a.saveHistory()
reply := resp.Reply
if resp.Answer != "" {
if reply != "" {
reply += "\n\n"
}
reply += resp.Answer
}
return &Response{
Reply: reply,
ToolCalls: resp.ToolCalls,
Agent: a.opts.Name,
}, nil
}
// Chat implements the proto AgentHandler interface for RPC.
// @example {"message": "What tasks are overdue?"}
func (a *agentImpl) Chat(ctx context.Context, req *pb.ChatRequest, rsp *pb.ChatResponse) error {
resp, err := a.Ask(ctx, req.Message)
if err != nil {
return err
}
rsp.Reply = resp.Reply
rsp.Agent = resp.Agent
for _, tc := range resp.ToolCalls {
input, _ := json.Marshal(tc.Input)
rsp.ToolCalls = append(rsp.ToolCalls, &pb.ToolCall{
Id: tc.ID,
Name: tc.Name,
Input: string(input),
Result: tc.Result,
})
}
return nil
}
// Run starts the agent as a service with a Chat RPC endpoint.
func (a *agentImpl) Run() error {
if a.model == nil {
a.setup()
}
a.server = server.NewServer(
server.Name(a.opts.Name),
server.Registry(a.opts.Registry),
server.Metadata(map[string]string{
"type": "agent",
"services": strings.Join(a.opts.Services, ","),
}),
)
pb.RegisterAgentHandler(a.server, a)
if err := a.server.Start(); err != nil {
return fmt.Errorf("failed to start agent: %w", err)
}
fmt.Printf("Agent %s registered (manages: %s)\n", a.opts.Name, strings.Join(a.opts.Services, ", "))
ch := make(chan struct{})
<-ch
return nil
}
func (a *agentImpl) Stop() error {
if a.server != nil {
return a.server.Stop()
}
return nil
}
func (a *agentImpl) discoverTools() ([]ai.Tool, error) {
all, err := a.tools.Discover()
if err != nil {
return nil, err
}
var scoped []ai.Tool
for _, t := range all {
if strings.HasPrefix(t.OriginalName, a.opts.Name+".") {
continue
}
if len(a.opts.Services) == 0 {
scoped = append(scoped, t)
continue
}
for _, svc := range a.opts.Services {
if strings.HasPrefix(t.OriginalName, svc+".") {
scoped = append(scoped, t)
break
}
}
}
// Expose the agent's own capabilities (plan, delegate) as tools.
// Ephemeral sub-agents don't get them.
if !a.ephemeral {
scoped = append(scoped, builtinTools()...)
}
return scoped, nil
}
func (a *agentImpl) buildPrompt() string {
var base string
switch {
case a.opts.Prompt != "":
base = a.opts.Prompt
case len(a.opts.Services) > 0:
base = fmt.Sprintf("You are the %s agent. You manage these services: %s. Use the available tools to fulfill requests.",
a.opts.Name, strings.Join(a.opts.Services, ", "))
default:
base = fmt.Sprintf("You are the %s agent. Use the available tools to fulfill requests.", a.opts.Name)
}
// Keep the agent oriented: surface its saved plan, if any.
if !a.ephemeral {
if plan := a.loadPlan(); plan != "" {
base += "\n\nYour current plan (update it with the plan tool as you make progress):\n" + plan
}
}
return base
}
func (a *agentImpl) historyKey() string {
return "agent/" + a.opts.Name + "/history"
}
func (a *agentImpl) loadHistory() {
recs, err := a.opts.Store.Read(a.historyKey())
if err != nil || len(recs) == 0 {
return
}
var messages []ai.Message
if err := json.Unmarshal(recs[0].Value, &messages); err != nil {
return
}
for _, m := range messages {
a.hist.Add(m.Role, m.Content)
}
}
func (a *agentImpl) saveHistory() {
if a.ephemeral {
return
}
data, err := json.Marshal(a.hist.Messages())
if err != nil {
return
}
a.opts.Store.Write(&store.Record{
Key: a.historyKey(),
Value: data,
})
}