Class: LittleGhost::Swarm

Inherits:
Assembly show all
Extended by:
LittleGhost::Support::ClassAttributes
Defined in:
lib/little_ghost/swarm.rb

Overview

Lets configured Agent members hand one request directly to one another.

A swarm is an Assembly for model-selected routing. One member is active at a time. It either produces the final answer or uses a model-visible handoff tool to choose one of the next members allowed by the application.

class ProblemSolverSwarm < LittleGhost::Swarm
member TriageAgent
member BillingAgent
member AccountAgent

start TriageAgent
handoff TriageAgent, to: [BillingAgent, AccountAgent]
max_steps 12
end

run = ProblemSolverSwarm.ask("Why was I charged twice?")
run.response

Call a named Swarm with ask for its final Run, or the streaming entrypoint for coordination and final-response events.

Swarm members are Agent definitions rather than arbitrary assemblies so a handoff remains a direct model-to-model transition. Original conversation history and application context stay isolated unless a member opts in with history: true or context: true. Streams expose coordination events and the final member response, but not intermediate model text.

Defined Under Namespace

Classes: Handoff, Member

Constant Summary collapse

MAX_BUFFERED_EVENTS =

:nodoc:

10_000
MAX_BUFFERED_EVENT_BYTES =

:nodoc:

10 * 1024 * 1024

Constants inherited from Assembly

Assembly::MAX_STEP_EVENTS, Assembly::MAX_STEP_EVENT_BYTES, Assembly::MAX_STEP_OUTPUT_BYTES

Instance Attribute Summary

Attributes inherited from Assembly

#run, #runtime, #sandbox, #workspace

Class Method Summary collapse

Instance Method Summary collapse

Methods included from LittleGhost::Support::ClassAttributes

class_attribute, included

Methods inherited from Assembly

#as_tool, ask, #ask, assembly_id, assembly_kind, #build_run, #call, #close, definition, description, #entrypoint_name, #interject, #prompt_locals, #start_execution, stream_ask, #stream_ask, to_builder, validate_step_policy!

Constructor Details

#initialize(run: nil, runtime: nil) ⇒ Swarm

:nodoc:



170
171
172
173
174
 
# File 'lib/little_ghost/swarm.rb', line 170

def initialize(run: nil, runtime: nil) # :nodoc:
  super(run:, runtime:, standalone: run.nil?)
  @swarm_mutex = Mutex.new
  @swarm_started = false
end

Class Method Details

.handoff(from, to:) ⇒ Object

Restricts one member to the declared handoff targets.

Raises:

  • (ArgumentError) 


78
79
80
81
82
83
84
85
86
87
 
# File 'lib/little_ghost/swarm.rb', line 78

def handoff(from, to:)
  from = normalize_member_id(member_reference_id(from))
  targets = Array(to).map { |target| normalize_member_id(member_reference_id(target)) }
  raise ArgumentError, "handoff requires at least one target" if targets.empty?
  raise ArgumentError, "handoff targets must be unique" unless targets.uniq.length == targets.length

  declaration = Handoff.new(from:, to: targets.freeze)
  self.swarm_handoffs_value = (swarm_handoffs_value + [declaration]).freeze
  declaration
end

.max_handoff_repeats(value = nil) ⇒ Object

Reads or assigns how often the same directed handoff may repeat.

For example, a value of 2 allows the transition from triage to billing twice during one Swarm run. max_steps still limits total member calls.

Raises:

  • (ArgumentError) 


103
104
105
106
107
108
109
110
 
# File 'lib/little_ghost/swarm.rb', line 103

def max_handoff_repeats(value = nil)
  return swarm_max_handoff_repeats_value if value.nil?

  value = Integer(value)
  raise ArgumentError, "max_handoff_repeats must be at least 1" if value < 1

  self.swarm_max_handoff_repeats_value = value
end

.max_steps(value = nil) ⇒ Object

Reads or assigns the maximum member executions.

Raises:

  • (ArgumentError) 


90
91
92
93
94
95
96
97
 
# File 'lib/little_ghost/swarm.rb', line 90

def max_steps(value = nil)
  return swarm_max_steps_value if value.nil?

  value = Integer(value)
  raise ArgumentError, "max_steps must be at least 1" if value < 1

  self.swarm_max_steps_value = value
end

.member(agent, as: nil, timeout: nil, retries: 0, retry_on: nil, retry_delay: 0, history: false, context: false) ⇒ Object

Declares one Agent member and its optional execution policy.

Raises:



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/little_ghost/swarm.rb', line 52

def member(agent, as: nil, timeout: nil, retries: 0, retry_on: nil, retry_delay: 0,
  history: false, context: false)
  validate_agent_reference!(agent)
  id = normalize_member_id(as || agent_reference_id(agent))
  raise ConfigurationError, "swarm member #{id.inspect} is already declared" if swarm_members_value.key?(id)
  unless [history, context].all? { |value| value == true || value == false }
    raise ArgumentError, "swarm member history and context options must be true or false"
  end

  policies = {timeout:, retries:, retry_on:, retry_delay:}.freeze
  declaration = Member.new(
    id:, agent:, policies:,
    inherit_history: history,
    inherit_context: context
  )
  self.swarm_members_value = swarm_members_value.merge(id => declaration).freeze
end

.start(member = nil) ⇒ Object

Reads or assigns the initial Agent member.



71
72
73
74
75
 
# File 'lib/little_ghost/swarm.rb', line 71

def start(member = nil)
  return swarm_start_value if member.nil?

  self.swarm_start_value = normalize_member_id(member_reference_id(member))
end

.swarm_definition! ⇒ Object

:nodoc:

Raises:



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/little_ghost/swarm.rb', line 118

def swarm_definition! # :nodoc:
  members = swarm_members_value
  start_id = swarm_start_value
  raise ConfigurationError, "swarm must declare at least two members" if members.length < 2
  raise ConfigurationError, "swarm must declare a start member" unless start_id
  raise ConfigurationError, "swarm start member #{start_id.inspect} is not declared" unless members.key?(start_id)

  seen_sources = {}
  members.each_value { |member| validate_step_policy!(member.policies) }
  swarm_handoffs_value.each do |declaration|
    raise ConfigurationError, "swarm handoff source #{declaration.from.inspect} is not declared" unless members.key?(declaration.from)
    if seen_sources[declaration.from]
      raise ConfigurationError, "swarm member #{declaration.from.inspect} has more than one handoff declaration"
    end
    seen_sources[declaration.from] = true
    declaration.to.each do |target|
      raise ConfigurationError, "swarm handoff target #{target.inspect} is not declared" unless members.key?(target)
      raise ConfigurationError, "swarm member #{target.inspect} cannot hand off to itself" if target == declaration.from
    end
  end
  [members, start_id, swarm_handoffs_value]
end

.validate! ⇒ Object

Validates the members and allowed handoff routes, then returns this class.



113
114
115
116
 
# File 'lib/little_ghost/swarm.rb', line 113

def validate!
  swarm_definition!
  self
end

Instance Method Details

#stream(input = nil, history: nil, context: nil, cancellation_token: Support::CancellationToken.new, deadline: nil, settings: nil, template_locals: nil, template_paths: nil, parent_operation_id: nil, checkpoint: nil, **_options) ⇒ Object

Streams lifecycle events and only the final member's answer events.

Raises:

  • (ArgumentError) 


177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
 
# File 'lib/little_ghost/swarm.rb', line 177

def stream(input = nil, history: nil, context: nil,
  cancellation_token: Support::CancellationToken.new, deadline: nil,
  settings: nil, template_locals: nil, template_paths: nil,
  parent_operation_id: nil, checkpoint: nil, **_options)
  raise ArgumentError, "input is required" if input.nil?
  if standalone?
    return build_run(entrypoint_payload(input, {
      history:, context:, settings:, template_paths:,
      deadline_at: deadline, cancellation_token:
    }.compact)).each
  end

  reserve_execution!
  current_input = input.is_a?(Message) ? input : Message.new(role: :user, content: input)
  original_history = normalize_history(history)
  original_context = isolated_assembly_state(context || {})
  settings ||= {}
  template_locals ||= {}
  template_paths ||= []

  usage = Usage.new
  error_emitted = false
  Enumerator.new do |events|
    members, current, topology = self.class.swarm_definition!
    steps = []
    transitions = Hash.new(0)
    step_number = 0
    previous_step_id = nil

    loop do
      check_swarm_control!(cancellation_token, deadline, step_number)
      step_number += 1
      member = members.fetch(current)
      allowed = allowed_targets(current, members, topology)
      handoff_tool = handoff_tool_for(current, allowed, members) if allowed.any?
      step_id = SecureRandom.uuid
      events << StreamEvent.build(
        :assembly_step_start,
        assembly_id: self.class.assembly_id,
        assembly_kind: :swarm,
        step: step_number,
        participant: current,
        step_id:
      )
      execution = execute_assembly_step(
        reference: member.agent,
        participant: current,
        input: current_input,
        history: member.inherit_history ? original_history : [],
        context: member.inherit_context ? original_context : {},
        cancellation_token:, deadline:, settings:, template_locals:,
        template_paths:, parent_operation_id:,
        policies: member.policies,
        predecessor_ids: Array(previous_step_id),
        checkpoint: nil,
        build_options: {tools: [handoff_tool].compact},
        step_id:
      ) { |event| events << event }
      result = execution.result
      previous_step_id = execution.step.id
      transition = active_transition(execution, allowed)
      events << StreamEvent.build(
        :assembly_step_stop,
        assembly_id: self.class.assembly_id,
        assembly_kind: :swarm,
        step: step_number,
        participant: current,
        step_id: execution.step.id,
        usage: execution.step.usage
      )

      if transition
        usage += execution.step.usage
        target = transition.fetch(:agent_id)
        key = [current, target]
        transitions[key] += 1
        if transitions[key] > self.class.max_handoff_repeats
          events << StreamEvent.build(
            :assembly_handoff_loop,
            assembly_id: self.class.assembly_id,
            assembly_kind: :swarm,
            from: current,
            to: target,
            count: transitions[key]
          )
          raise AssemblyLimitError, "swarm repeated handoff #{current.inspect} -> #{target.inspect} too many times"
        end
        sanitized = sanitized_handoff_step(execution.step, transition)
        steps << sanitized
        steps.concat(result.steps.drop(1))
        events << StreamEvent.build(
          :assembly_transition,
          assembly_id: self.class.assembly_id,
          assembly_kind: :swarm,
          step: step_number,
          from: current,
          to: target
        )
        current_input = handoff_input(from: current, transition:)
        current = target
        next
      end

      usage += execution.step.usage
      steps.concat(result.steps)
      final = copy_run_result(result, usage:, steps: steps.freeze)
      release_final_events(execution.events, final).each { |event| events << event }
      break
    end
  rescue => error
    usage += step_error_usage(error)
    unless error_emitted
      events << StreamEvent.build(:invocation_error, error:, usage:, metadata: {})
    end
    raise
  end
end