Class: SwarmSDK::V3::Agent

Inherits:
Object
  • Object
show all
Defined in:
lib/swarm_sdk/v3/agent.rb

Overview

Runtime agent with built-in memory

The Agent ties together RubyLLM::Chat, the memory system, and tools. The LLM’s context window is a staging area — older turns get consolidated into memory cards, and retrieval brings relevant memory back on demand.

## Interruption Safety

The agent supports safe interruption via #interrupt!, which raises ‘Async::Stop` in the fiber running #ask. When adding new features to the agent, follow these rules:

  1. **Interruptible phases** (LLM calls, tool execution, streaming): Code here can be interrupted at any fiber yield point. Do NOT leave shared state half-updated — use snapshot/restore or flags to detect incomplete operations in ‘ensure` blocks.

  2. **Uninterruptible phases** (memory writes, STM capture, eviction): Wrap in ‘Async::Task.current.defer_stop { }` to defer `Async::Stop` until the block completes. Use this for any multi-step I/O that must be atomic (e.g., writing cards + saving the index).

  3. **New instance state**: If a new feature adds state that is modified during #ask, ensure it is either rolled back on interruption (via ‘ensure`) or protected with `defer_stop`.

  4. Subprocesses: Any code that spawns a subprocess (Open3, etc.) must terminate it in an ‘ensure` block. `Async::Stop` bypasses `rescue StandardError` — only `ensure` is guaranteed to run.

Examples:

Basic usage

definition = AgentDefinition.new(
  name: :assistant,
  description: "A helpful assistant",
  model: "claude-sonnet-4",
  tools: [:Read, :Write, :Edit, :Bash, :Grep, :Glob],
  memory_directory: ".swarm/memory",
)

agent = Agent.new(definition)
response = agent.ask("Build a login page")

Without memory

definition = AgentDefinition.new(
  name: :chat,
  description: "Simple chat",
)
agent = Agent.new(definition)
response = agent.ask("Hello!")

Interrupting a running agent

Async do |parent|
  task = parent.async { agent.ask("Long running task") }
  agent.running?      # => true
  agent.interrupt!    # => true
  result = task.wait  # => nil
  agent.running?      # => false
end

See Also:

Direct Known Subclasses

SubTaskAgent

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(definition) ⇒ Agent

Create a new agent

Lazy-initializes the RubyLLM::Chat and memory system on first ask().

Parameters:



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/swarm_sdk/v3/agent.rb', line 81

def initialize(definition)
  @definition = definition
  @id = "#{definition.name}_#{SecureRandom.hex(3)}"
  @chat = nil
  @memory_store = nil
  @stm_buffer = []
  @turn_counter = 0
  @total_input_tokens = 0
  @total_output_tokens = 0
  @initialized = false
  @semaphore = Async::Semaphore.new(1)
  @current_task = nil
  @pending_ingestion = nil
  @steering_queue = []
  @loaded_skills = nil
  @base_system_prompt = nil
  @mcp_connectors = []
  @hooks = Hooks::Runner.new(definition.hooks)
end

Instance Attribute Details

#definition ⇒ AgentDefinition (readonly)

Returns Immutable agent configuration.

Returns:



68
69
70
 
# File 'lib/swarm_sdk/v3/agent.rb', line 68

def definition
  @definition
end

#id ⇒ String (readonly)

Returns Unique instance identifier (name_<hex>).

Returns:

  • (String) —

    Unique instance identifier (name_<hex>)



71
72
73
 
# File 'lib/swarm_sdk/v3/agent.rb', line 71

def id
  @id
end

#loaded_skills ⇒ Array<Skills::Manifest> (readonly)

Returns Loaded skills (available after first ask).

Returns:



74
75
76
 
# File 'lib/swarm_sdk/v3/agent.rb', line 74

def loaded_skills
  @loaded_skills
end

Instance Method Details

#ask(prompt, output_schema: nil) {|event| ... } ⇒ RubyLLM::Message?

Send a message to the agent and get a response

The ask() flow:

  1. Lazy-initialize (create chat, memory, tools)

  2. Retrieve relevant memory cards for the prompt

  3. Build working context (system prompt + memory + recent turns)

  4. Execute via RubyLLM::Chat (handles tool loop internally)

  5. Capture turn in STM buffer

  6. Ingest turn into memory (async)

  7. Evict old turns from STM if buffer exceeds limit

  8. Emit events

Supports safe interruption via #interrupt!. When interrupted, returns nil and leaves the agent in a consistent state for the next ask() call. Check #running? to see if an ask is in progress.

Examples:

Simple ask

response = agent.ask("What is 2+2?")
puts response.content

Receive all events via block

agent.ask("Tell me a story") do |event|
  case event[:type]
  when "content_chunk"
    print event[:content]
  when "tool_call"
    puts "Calling #{event[:tool]}..."
  end
end

With structured output

schema = { type: "object", properties: { answer: { type: "integer" } } }
response = agent.ask("What is 2+2?", output_schema: schema)
response.content  # => { "answer" => 4 }

Parameters:

  • prompt (String) —

    User message

  • output_schema (Hash, Object, nil) (defaults to: nil) —

    Per-call schema override (nil = use definition default)

Yields:

  • (event) —

    Optional block receives ALL events (content_chunk, tool_call, etc.)

Yield Parameters:

  • event (Hash) —

    Event hash with :type, :timestamp, and event-specific fields

Returns:

  • (RubyLLM::Message, nil) —

    LLM response, or nil if interrupted



141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/swarm_sdk/v3/agent.rb', line 141

def ask(prompt, output_schema: nil, &block)
  with_block_emitter(block) do
    Sync do |task|
      @semaphore.acquire do
        @current_task = task
        begin
          lazy_initialize!

          before_result = @hooks.run(:before_ask, Hooks::Context.new(
            event: :before_ask, agent_name: @definition.name, prompt: prompt,
          ))

          if before_result.halt?
            nil
          else
            prompt = before_result.value if before_result.replace?
            response = execute_turn(prompt, output_schema: output_schema)

            @hooks.run(:after_ask, Hooks::Context.new(
              event: :after_ask, agent_name: @definition.name, prompt: prompt, response: response,
            ))

            @hooks.run(:on_stop, Hooks::Context.new(
              event: :on_stop, agent_name: @definition.name, response: response,
            ))

            response
          end
        rescue Async::Stop
          EventStream.emit(type: "agent_interrupted", agent: @id, turn: @turn_counter)
          nil
        ensure
          @current_task = nil
        end
      end
    end
  end
end

#clear(clear_memory: false) ⇒ void

This method returns an undefined value.

Reset conversation and optionally clear memory

Parameters:

  • clear_memory (Boolean) (defaults to: false) —

    Also clear memory storage



256
257
258
259
260
261
262
263
264
265
266
267
268
 
# File 'lib/swarm_sdk/v3/agent.rb', line 256

def clear(clear_memory: false)
  wait_for_pending_ingestion
  disconnect_mcp_servers
  @stm_buffer.clear
  @steering_queue.clear
  @turn_counter = 0
  @chat&.reset_messages!(preserve_system_prompt: true)

  return unless clear_memory && @memory_store

  # Clear memory by reinitializing the adapter
  @memory_store = nil
end

#clear_steering_queue ⇒ void

This method returns an undefined value.

Clear all queued steering messages



241
242
243
 
# File 'lib/swarm_sdk/v3/agent.rb', line 241

def clear_steering_queue
  @steering_queue.clear
end

#defrag! {|event| ... } ⇒ Hash?

Run memory defragmentation (compression, consolidation, promotion, pruning)

Call this between sessions, on a schedule, or whenever appropriate. Does not run during conversation turns — the SDK user controls when.

Examples:

Run defrag after a session

agent.defrag!
#=> { duplicates_merged: 0, conflicts_detected: 0,
#     cards_compressed: 3, cards_promoted: 1, cards_pruned: 0 }

Run defrag with progress display

agent.defrag! do |event|
  case event[:type]
  when "memory_defrag_progress"
    puts "#{event[:phase]}: #{event[:phase_current]}/#{event[:phase_total]}"
  end
end

Yields:

  • (event) —

    Optional block receives defrag events

Yield Parameters:

  • event (Hash) —

    Event hash with :type, :timestamp, and event-specific fields

Returns:

  • (Hash, nil) —

    Defragmentation results, or nil if memory not enabled



400
401
402
403
404
405
406
407
408
409
410
411
412
 
# File 'lib/swarm_sdk/v3/agent.rb', line 400

def defrag!(&block)
  return unless @definition.memory_enabled?

  with_block_emitter(block) do
    Sync do
      @semaphore.acquire do
        lazy_initialize!
        wait_for_pending_ingestion
        @memory_store.defrag!
      end
    end
  end
end

#initialized? ⇒ Boolean

Whether the agent has been initialized

Returns:

  • (Boolean) 


287
288
289
 
# File 'lib/swarm_sdk/v3/agent.rb', line 287

def initialized?
  @initialized
end

#interrupt! ⇒ Boolean?

Stop whatever the agent is doing

Raises Async::Stop in the fiber running ask(). Safe to call from another Async fiber in the same reactor. Idempotent — returns nil if the agent is idle.

Examples:

Interrupt from another fiber

Async do |parent|
  task = parent.async { agent.ask("Long task") }
  sleep 1
  agent.interrupt!  # => true
  result = task.wait  # => nil
end

No-op when idle

agent.interrupt!  # => nil

Returns:

  • (Boolean, nil) —

    true if a running task was stopped, nil if idle



198
199
200
201
202
203
 
# File 'lib/swarm_sdk/v3/agent.rb', line 198

def interrupt!
  return unless @current_task

  @current_task.stop
  true
end

#loop(kickoff:, iterate:, max_iterations: 10, convergence_threshold: 0.95, converge: true) {|event| ... } ⇒ Loop::Result

Run an iterative refinement loop over the agent

Executes a kickoff prompt followed by repeated iterate prompts, optionally checking for convergence via embedding similarity between consecutive responses. Each iteration is a normal ask() call — hooks fire, memory ingests, and events stream normally.

Examples:

Basic iterative refinement

result = agent.loop(
  kickoff: "Write a poem about the sea",
  iterate: "Improve the poem, making it more vivid",
  max_iterations: 5,
)
puts result.final_response.content
puts "Converged: #{result.converged?}"

Without convergence checking

result = agent.loop(
  kickoff: "Draft an outline",
  iterate: "Expand the next section",
  max_iterations: 3,
  converge: false,
)

With event streaming

agent.loop(kickoff: "Start", iterate: "Continue", max_iterations: 5) do |event|
  case event[:type]
  when "loop_iteration_completed"
    puts "Iteration #{event[:iteration]}, delta: #{event[:delta_score]}"
  when "content_chunk"
    print event[:content]
  end
end

Parameters:

  • kickoff (String) —

    Prompt for the first iteration

  • iterate (String) —

    Prompt for subsequent iterations

  • max_iterations (Integer) (defaults to: 10) —

    Maximum number of iterations (>= 1)

  • convergence_threshold (Float) (defaults to: 0.95) —

    Similarity threshold for convergence (0.0..1.0)

  • converge (Boolean) (defaults to: true) —

    Whether to check for convergence via embeddings

Yields:

  • (event) —

    Optional block receives ALL events (content_chunk, loop_*, etc.)

Yield Parameters:

  • event (Hash) —

    Event hash with :type, :timestamp, and event-specific fields

Returns:

  • (Loop::Result) —

    Aggregate result with iterations and convergence status

Raises:

  • (ArgumentError) —

    If max_iterations < 1 or convergence_threshold out of range



352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
 
# File 'lib/swarm_sdk/v3/agent.rb', line 352

def loop(kickoff:, iterate:, max_iterations: 10, convergence_threshold: 0.95, converge: true, &block)
  validate_loop_params!(max_iterations, convergence_threshold)

  embedder = converge ? loop_embedder : nil
  ask_callable = ->(prompt) { ask(prompt, &block) }

  executor = Loop::Executor.new(
    ask_callable: ask_callable,
    embedder: embedder,
    agent_id: @id,
  )

  # Wrap with block emitter so loop lifecycle events
  # (loop_started, loop_iteration_completed, loop_completed)
  # reach the caller's block. Each ask() inside the executor
  # will also set/restore the block emitter for its own events.
  with_block_emitter(block) do
    executor.run(
      kickoff: kickoff,
      iterate: iterate,
      max_iterations: max_iterations,
      convergence_threshold: convergence_threshold,
      converge: converge,
    )
  end
end

#memory ⇒ Memory::Store?

Read-only access to the memory store

Returns:



294
295
296
 
# File 'lib/swarm_sdk/v3/agent.rb', line 294

def memory
  @memory_store
end

#memory_read_only? ⇒ Boolean

Whether memory operations are read-only

Subtask agents override this to return true, preventing access counter updates during context building.

Returns:

  • (Boolean) 


304
305
306
 
# File 'lib/swarm_sdk/v3/agent.rb', line 304

def memory_read_only?
  false
end

#messages ⇒ Array<Hash>

Get recent messages (STM buffer)

Returns:

  • (Array<Hash>) —

    Recent conversation turns



248
249
250
 
# File 'lib/swarm_sdk/v3/agent.rb', line 248

def messages
  @stm_buffer.dup
end

#name ⇒ Symbol

Agent name from definition

Returns:

  • (Symbol) 


280
281
282
 
# File 'lib/swarm_sdk/v3/agent.rb', line 280

def name
  @definition.name
end

#running? ⇒ Boolean

Whether the agent is currently executing an ask() call

Returns true only while the agent holds the semaphore and is actively processing a turn. Useful for deciding whether to call #interrupt!.

Examples:

Guard an interrupt call

agent.interrupt! if agent.running?

Returns:

  • (Boolean) 


215
216
217
 
# File 'lib/swarm_sdk/v3/agent.rb', line 215

def running?
  !@current_task.nil?
end

#steer(message) ⇒ void

This method returns an undefined value.

Queue a high-priority message that interrupts the current tool batch

Steering messages are injected after the current tool completes, skipping any remaining tools in the batch. The message is delivered as a ‘role: user` message before the next LLM call.

Use this for urgent interruptions that should preempt normal execution.

Examples:

Inject an urgent update while the agent is working

agent.on_tool_result do |tool_call, result|
  agent.steer("Priority update: server is down")
end

Parameters:

  • message (String) —

    Message content to inject



234
235
236
 
# File 'lib/swarm_sdk/v3/agent.rb', line 234

def steer(message)
  @steering_queue << message
end

#tokens ⇒ Hash

Token usage statistics

Returns:

  • (Hash) —

    Input and output token counts



273
274
275
 
# File 'lib/swarm_sdk/v3/agent.rb', line 273

def tokens
  { input: @total_input_tokens, output: @total_output_tokens }
end