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 calls a reserved handoff tool to select an allowed next member.

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

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 lifecycle 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, #interrupt, #interrupt_response, #prompt_locals, #start_execution, stream_ask, #stream_ask, to_builder, validate_step_policy!

Constructor Details

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

:nodoc:



162
163
164
165
166
 
# File 'lib/little_ghost/swarm.rb', line 162

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) 


73
74
75
76
77
78
79
80
81
82
 
# File 'lib/little_ghost/swarm.rb', line 73

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 the repeated directed-handoff limit.

Raises:

  • (ArgumentError) 


95
96
97
98
99
100
101
102
 
# File 'lib/little_ghost/swarm.rb', line 95

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) 


85
86
87
88
89
90
91
92
 
# File 'lib/little_ghost/swarm.rb', line 85

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:



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/little_ghost/swarm.rb', line 47

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.



66
67
68
69
70
 
# File 'lib/little_ghost/swarm.rb', line 66

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:



110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/little_ghost/swarm.rb', line 110

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 member set and topology and returns this Swarm class.



105
106
107
108
 
# File 'lib/little_ghost/swarm.rb', line 105

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) 


169
170
171
172
173
174
175
176
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
 
# File 'lib/little_ghost/swarm.rb', line 169

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