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.