Help us shape the future of Go! Take the 2026 Go Developer Survey

crew

package
v0.9.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 22, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func GetOutput 

func GetOutput[T any](r *Result, taskID string) (T, error)

GetOutput extracts and unmarshals a typed result from a crew execution result.

Types

type Awaitable added in v0.8.0 

type Awaitable struct {
	// contains filtered or unexported fields
}

Awaitable is a future representing an async crew execution. It delivers the result once (to any number of awaiters) via a closed channel.

func (*Awaitable) Await added in v0.8.0 

func (a *Awaitable) Await(ctx context.Context) (*Result, error)

Await blocks until the crew finishes or the context is cancelled. Returns the cached result immediately on subsequent calls.

func (*Awaitable) AwaitWithTimeout added in v0.8.0 

func (a *Awaitable) AwaitWithTimeout(ctx context.Context, d time.Duration) (*Result, error)

AwaitWithTimeout blocks until the crew finishes or the timeout expires.

func (*Awaitable) State added in v0.8.0 

func (a *Awaitable) State() AwaitableState

State returns the current execution state without blocking.

type AwaitableState added in v0.8.0 

type AwaitableState uint32

AwaitableState represents the execution state of an async crew. 

const (
	AwaitablePending   AwaitableState = 0
	AwaitableRunning   AwaitableState = 1
	AwaitableCompleted AwaitableState = 2
	AwaitableFailed    AwaitableState = 3
)

func (AwaitableState) String added in v0.8.0 

func (s AwaitableState) String() string

type Checkpoint 

type Checkpoint struct {
	ID             string
	ProcessType    string
	Phase          string
	NodeStates     map[string]*NodeState
	TaskStates     map[string]*NodeState // new code populates both; readers prefer TaskStates, fall back to NodeStates
	OrderedTaskIDs []string              // preserves task execution order for resume
	CrewID         string                // identifies which crew context this checkpoint belongs to
	CreatedAt      time.Time
	UpdatedAt      time.Time
}

Checkpoint represents a snapshot of crew execution state.

type CheckpointStore 

type CheckpointStore interface {
	Save(ctx context.Context, ckpt *Checkpoint) error
	Load(ctx context.Context, id string) (*Checkpoint, error)
	List(ctx context.Context, filter map[string]any) ([]*Checkpoint, error)
	Delete(ctx context.Context, id string) error
	ListCheckpoints(ctx context.Context, crewID string) ([]*Checkpoint, error)
}

CheckpointStore defines the interface for checkpoint persistence. The implementation is provided by the checkpoint package to avoid circular dependency.

type Config 

type Config struct {
	Agents          []*agent.Agent
	Tasks           []*task.Task
	Process         ProcessType
	ManagerLLM      llm.Client
	ManagerAgent    *agent.Agent
	Verbose         bool
	MaxRPM          int
	MaxCycles       int
	StepCallback    callback.Callback
	CheckpointStore CheckpointStore
	// @sk-task knowledge-sources#T4.1: shared knowledge sources (AC-008)
	KnowledgeSources []knowledge.KnowledgeSource
	FlowEdges        []flow.Edge
	// @sk-task output-streaming#T2.1: streaming flag (AC-001, AC-002)
	Stream bool
	// @sk-task unified-memory#T3.1: crew-level memory (AC-004)
	Memory *memory.UnifiedMemory
	// @sk-task tool-caching#T3.1: crew-level cache override (AC-005)
	Cache bool
	// @sk-task async-crew-execution#T1.4: MaxParallel for KickoffForEachAsync (AC-013)
	MaxParallel int
	// @sk-task replay-system#T3.2: CrewID for checkpoint scoping (AC-007)
	CrewID string
	// @sk-task skills-system#T2.1: shared skill paths for all agents (AC-005, AC-006)
	Skills []string
	// @sk-task planner#T3.1: crew-level planning config (AC-007, AC-008)
	Planning    bool
	PlanningLLM llm.Client
}

Config defines the configuration for a Crew.

type Crew 

type Crew struct {
	// contains filtered or unexported fields
}

Crew orchestrates a group of agents to execute a collection of tasks.

func New 

func New(cfg Config) *Crew

New creates a new Crew with the given configuration.

func (*Crew) CacheHitCount added in v0.8.0 

func (c *Crew) CacheHitCount() int64

CacheHitCount returns the total cache hits across all agents.

func (*Crew) CacheMissCount added in v0.8.0 

func (c *Crew) CacheMissCount() int64

CacheMissCount returns the total cache misses across all agents.

func (*Crew) ClearToolCache added in v0.8.0 

func (c *Crew) ClearToolCache()

ClearToolCache clears all cached tool results across all agents.

func (*Crew) Fork added in v0.8.0 

func (c *Crew) Fork(checkpointID string) (string, error)

Fork creates a deep copy of the checkpoint with the given ID and saves it under a new ID.

func (*Crew) FrameStream added in v0.7.0 

func (c *Crew) FrameStream() <-chan stream.Frame

FrameStream returns a merged frame channel from all agents. Returns nil if no agents have frame channels.

func (*Crew) InjectTask added in v0.6.0 

func (c *Crew) InjectTask(t *task.Task) error

InjectTask queues a new task for execution. Returns an error if the queue is full.

func (*Crew) IsPaused added in v0.6.0 

func (c *Crew) IsPaused() bool

IsPaused returns whether the crew is currently paused.

func (*Crew) Kickoff 

func (c *Crew) Kickoff(ctx context.Context) (*Result, error)

Kickoff starts the crew's task execution using the configured process type.

func (*Crew) KickoffAsync added in v0.8.0 

func (c *Crew) KickoffAsync(ctx context.Context) (*Awaitable, <-chan CrewStreamingOutput, error)

KickoffAsync starts the crew's task execution in a background goroutine. Returns an Awaitable future and a channel of per-task streaming outputs. The stream channel is closed after all tasks complete. Works for all process types.

func (*Crew) KickoffForEach added in v0.8.0 

func (c *Crew) KickoffForEach(ctx context.Context, inputs []map[string]any) ([]*Result, error)

KickoffForEach runs the crew sequentially for each input set. Each iteration clones the crew, substitutes placeholders, and calls Kickoff. Returns a slice of results in input order. Halts and returns the first error if any iteration fails.

func (*Crew) KickoffForEachAsync added in v0.8.0 

func (c *Crew) KickoffForEachAsync(ctx context.Context, inputs []map[string]any) (<-chan ForEachResult, error)

KickoffForEachAsync runs the crew for each input set in parallel using a bounded worker pool. Returns a channel of ForEachResult. The channel is closed after all iterations complete. MaxParallel controls the maximum number of concurrent executions (0 = NumCPU).

func (*Crew) KickoffStream added in v0.7.0 

func (c *Crew) KickoffStream(ctx context.Context) (<-chan CrewStreamingOutput, error)

KickoffStream starts the crew's task execution and returns a channel of per-task chunks. Stream must be true in Config, otherwise KickoffStream returns an error. The channel is closed after the last task. Only sequential process is supported for streaming.

func (*Crew) Pause added in v0.6.0 

func (c *Crew) Pause()

Pause suspends crew execution at the next task boundary.

func (*Crew) Replay added in v0.8.0 

func (c *Crew) Replay(ctx context.Context, freshCrew *Crew, taskID string) (*Result, error)

Replay re-executes from the given taskID forward using the latest checkpoint. Tasks before taskID retain their checkpoint output without re-execution.

func (*Crew) ReplayFromCheckpoint added in v0.8.0 

func (c *Crew) ReplayFromCheckpoint(ctx context.Context, freshCrew *Crew, checkpointID, taskID string) (*Result, error)

ReplayFromCheckpoint re-executes from the given taskID forward using a specific checkpoint.

func (*Crew) Resume 

func (c *Crew) Resume(ctx context.Context, freshCrew *Crew, checkpointID string) (*Result, error)

Resume restores crew execution from a checkpoint, using a fresh crew for agents/LLMs. Completed tasks are not re-executed (matched by index position in the tasks slice); remaining tasks execute normally.

func (*Crew) ResumeExecution added in v0.6.0 

func (c *Crew) ResumeExecution()

ResumeExecution continues a paused crew.

func (*Crew) SetEventBus added in v0.8.0 

func (c *Crew) SetEventBus(bus *events.EventBus)

SetEventBus sets the event bus for publishing async lifecycle events.

func (*Crew) SetStepCallback 

func (c *Crew) SetStepCallback(cb callback.Callback)

SetStepCallback replaces the crew's step callback. Used by telemetry bridge.

func (*Crew) StepCallback 

func (c *Crew) StepCallback() callback.Callback

StepCallback returns the crew's current step callback.

func (*Crew) Train added in v0.4.0 

func (c *Crew) Train(ctx context.Context, iterations int, inputs map[string]any,
	trainingDir string,
	feedbackFn func(iteration int, result *Result) training.Feedback,
) (*TrainingReport, error)

Train runs the crew N iterations, collects human feedback after each iteration, persists negative feedback (Rating < 0), and consolidates advice per agent after all iterations. Only sequential process is supported; other processes return ErrInvalidProcess.

type CrewStreamingOutput added in v0.7.0 

type CrewStreamingOutput struct {
	TaskID     string
	AgentRole  string
	Content    string
	IsLast     bool
	TokenUsage llm.Usage
	Error      string
}

CrewStreamingOutput is a per-task output chunk delivered in real time.

type ForEachResult added in v0.8.0 

type ForEachResult struct {
	Index  int
	Input  map[string]any
	Result *Result
	Error  error
}

ForEachResult holds the result of a single iteration of KickoffForEachAsync.

type NodeState 

type NodeState struct {
	TaskID          string
	Processed       bool
	Failed          bool
	Output          string
	Error           string
	InputContext    map[string]string // context passed to task execution
	AgentID         string            // Role of the agent that executed this task
	TokenUsage      map[string]int    // per-agent token usage at checkpoint time
	TaskDescription string            // task description at time of checkpoint
	ExpectedOutput  string            // expected output at time of checkpoint
	// @sk-task planner#T3.1: plan text for replay/audit (AC-009)
	Plan string
	// @sk-task conditional-tasks#T1.3: skip state for checkpoint persistence (AC-008)
	Skipped bool
}

NodeState represents the execution state of a single task or flow node within a checkpoint.

type ProcessType 

type ProcessType string

ProcessType defines the execution strategy for a crew. 

const (
	ProcessSequential   ProcessType = "sequential"
	ProcessHierarchical ProcessType = "hierarchical"
	ProcessConsensual   ProcessType = "consensual"
	ProcessReflective   ProcessType = "reflective"
	ProcessFlow         ProcessType = "flow"
	ProcessStateMachine ProcessType = "state_machine"
)

Supported crew process types.

type Result 

type Result struct {
	Tasks map[string]*task.Task
	Usage map[string]int
}

Result contains the outputs and usage metrics from a crew execution.

type TrainingReport added in v0.4.0 

type TrainingReport struct {
	AgentAdvice    map[string]string
	IterationCount int
	TotalDuration  time.Duration
	FeedbackCount  int
}

TrainingReport summarizes a training run.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.