micro--go-micro
9fdcc24cce
* docs: design note for flow steps + Checkpoint durable execution
* docs: fold in durable-execution decisions (State struct, single Step, run retention, retry)
* docs: rename State.Payload to State.Data
* feat(flow): ordered steps + Checkpoint durable execution
A flow can now be an ordered list of steps (a task with stages) instead
of a single LLM turn. State carries typed Data plus a Stage marker; each
step is checkpointed before and after via a pluggable Checkpoint
(store-backed by default), so a run survives a crash and resumes where it
stopped without re-running completed steps. Flow-level Retry with a
per-step override; runs retained for audit unless DeleteOnSuccess.
Step actions: Call (RPC), LLM (augmented turn), Dispatch (to an agent),
or any StepFunc. Single-step and agent-dispatch flows are unchanged.
* feat(flow): top-level re-exports + durable flow example
Expose the step/checkpoint API from the micro package (FlowSteps,
FlowStep, FlowState, FlowRetry, FlowWithCheckpoint, FlowCall/LLM/Dispatch,
Checkpoint, StoreCheckpoint) and add a runnable, key-free example
demonstrating crash + resume.
* docs: document durable flow steps (guide, README, CLI help)
* docs: blog post + changelog for durable workflows
* fix(flow): scope checkpoint keys by flow name (flow/{name}/runs/{id})
Run keys were flow/runs/{id} — a single global keyspace shared by every
flow on the default store. Namespace them by flow name so each flow's
state is kept apart. StoreCheckpoint now takes a scope argument (the flow
passes its name by default).
* feat(store): Scope handle; scope agent and flow state by name
Add store.Scope(s, database, table) — a store handle that confines every
operation to a database/table without mutating the shared store, so
co-located components don't clobber each other's table (the failure mode
of the global Init(Table(...)) approach).
Use it to keep each agent's memory and plan in its own table
(agent/{name}) and each flow's runs in its own (flow/{name}), instead of
one global table partitioned only by key prefix. Services already scope
by service name.
* feat: consistent state model — service store scoping, flow registry, list/history CLI
- service: scope store via store.Scope (database service / table name),
retiring the Init(store.Table(name)) global-mutation hack; bridge the
default store so handlers using store.DefaultStore stay isolated.
- flow: register in the registry as type=flow while running (with trigger
and step count), deregister on Stop. Live discovery, like agents.
- cli: micro flow list (registry), micro flow runs <name> (durable store),
micro agent history <name> (durable store). list = running, runs/history
= durable, mirroring the service model.
* test: mini-universe end-to-end harness + scheduled GitHub Action
internal/harness/universe boots a small but real go-micro world — four
services, a durable checkout flow that crashes at payment and resumes,
and a guardrailed agent with a tool wrapper reached over RPC — drives the
scenario, asserts the end state (10 checks), and shuts down. Everything
is real except the LLM (mocked), so it's deterministic and needs no key;
-provider anthropic runs it live. Exits non-zero on failure, so it's an
end-to-end test, not just a demo.
Adds .github/workflows/universe.yml (push/PR/daily/dispatch) running the
universe + existing harnesses on the mock provider, plus an opt-in job
that runs live when ANTHROPIC_API_KEY is set. 'make harness' runs them
locally.
* ci: run the live universe job against AtlasCloud (ATLASCLOUD_API_KEY)
* ci: run the live universe job only on schedule or manual dispatch
The deterministic mock job still runs on push/PR/daily; the live
(AtlasCloud) job runs daily and on manual workflow_dispatch only, so
changes don't burn API credits on every PR but can still be checked
against a real model on demand.
---------
Co-authored-by: Claude <noreply@anthropic.com>
313 行
7.5 KiB
Go
313 行
7.5 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
|
|
mem Memory
|
|
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
|
|
// calls counts identical tool calls (name+args) in the current Ask,
|
|
// for LoopLimit.
|
|
calls map[string]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...)
|
|
|
|
// Memory is pluggable. Use the configured one, otherwise the default
|
|
// store-backed memory — except ephemeral sub-agents, which keep an
|
|
// isolated, non-persistent context.
|
|
switch {
|
|
case a.opts.Memory != nil:
|
|
a.mem = a.opts.Memory
|
|
case a.ephemeral:
|
|
a.mem = NewInMemory(a.opts.HistoryLimit)
|
|
default:
|
|
a.mem = NewMemory(a.stateStore(), "history", a.opts.HistoryLimit)
|
|
}
|
|
}
|
|
|
|
// stateStore returns the agent's own state store, scoped to its name so
|
|
// memory and plan live in their own table ("agent/{name}") rather than a
|
|
// shared global one. The scoped handle injects the database/table per
|
|
// operation without mutating the underlying store.
|
|
func (a *agentImpl) stateStore() store.Store {
|
|
s := a.opts.Store
|
|
if s == nil {
|
|
s = store.DefaultStore
|
|
}
|
|
return store.Scope(s, "agent", a.opts.Name)
|
|
}
|
|
|
|
// 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.mem.Add("user", message)
|
|
a.steps = 0
|
|
a.calls = map[string]int{}
|
|
|
|
resp, err := a.model.Generate(ctx, &ai.Request{
|
|
Prompt: message,
|
|
SystemPrompt: a.buildPrompt(),
|
|
Tools: toolList,
|
|
Messages: a.mem.Messages(),
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if resp.Reply != "" {
|
|
a.mem.Add("assistant", resp.Reply)
|
|
}
|
|
if resp.Answer != "" {
|
|
a.mem.Add("assistant", resp.Answer)
|
|
}
|
|
|
|
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
|
|
}
|
|
}
|
|
}
|
|
|
|
// Developer-registered custom tools (WithTool).
|
|
for i := range a.opts.tools {
|
|
scoped = append(scoped, a.opts.tools[i].def)
|
|
}
|
|
|
|
// 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
|
|
}
|