Class: LLM::Function::Group

Inherits:
Object
  • Object
show all
Defined in:
lib/llm/function/group.rb

Overview

Abstract superclass for all concurrency strategy groups. Each strategy (sequential, thread, fiber, async, fork, ractor) subclasses this and implements #spawn, #alive? and #interrupt!. #wait is shared, and a strategy whose tasks report themselves answers with one of its own.

Instance Method Summary collapse

Instance Method Details

#spawn ⇒ nil

This method is abstract.

Returns:

  • (nil)

Raises:

  • (NotImplementedError) 


14
15
16
 
# File 'lib/llm/function/group.rb', line 14

def spawn
  raise NotImplementedError
end

#alive? ⇒ Boolean

This method is abstract.

Returns:

  • (Boolean)

Raises:

  • (NotImplementedError) 


21
22
23
 
# File 'lib/llm/function/group.rb', line 21

def alive?
  raise NotImplementedError
end

#interrupt! ⇒ nil Also known as: cancel!

This method is abstract.

Returns:

  • (nil)

Raises:

  • (NotImplementedError) 


28
29
30
 
# File 'lib/llm/function/group.rb', line 28

def interrupt!
  raise NotImplementedError
end

#wait ⇒ Array<LLM::Function::Return> Also known as: value

Note:

This method might raise an Interrupt if one or more tasks were interrupted but only after the tracer has accounted for each task and collected an exit status from each of them.

This method waits for one or more tasks to return. Within this method, the tracer associated with a task closes the on_tool_start span that was opened when the task spawned.

Returns:

Raises:



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/llm/function/group.rb', line 47

def wait
  spawn unless @spawned
  interrupt = nil
  vals = @tasks.map do |task|
    val, started_at, finished_at = task.wait, task.started_at, task.finished_at
    if LLM::Interrupt === val
      interrupt = val
      task.tracer&.on_tool_interrupt(ex: val, span: task.span, started_at:, finished_at:)
    else
      if val.error?
        task.tracer&.on_tool_error(ex: val, span: task.span, started_at:, finished_at:)
      else
        task.tracer&.on_tool_finish(result: val, span: task.span, started_at:, finished_at:)
      end
      val
    end
  end
  interrupt ? raise(interrupt) : vals
end