Module: Lyman::Workers

Extended by:
Shifty::DSL
Defined in:
lib/lyman/workers/tool_execution.rb,
lib/lyman/workers/chat_completion.rb

Class Method Summary collapse

Class Method Details

.apply_delta(message, tool_calls, delta, on_delta, state) ⇒ Object

Content deltas concatenate; tool-call deltas arrive as fragments keyed by index, with the arguments string dribbling in over many chunks. Reasoning-field deltas stream to on_delta wrapped in synthesized tags (see chat_completion).



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
 
# File 'lib/lyman/workers/chat_completion.rb', line 117

def self.apply_delta(message, tool_calls, delta, on_delta, state)
  reasoning = delta["reasoning"] || delta["reasoning_content"]
  if reasoning && !reasoning.empty?
    unless state[:thinking]
      state[:thinking] = true
      on_delta.call("<think>")
    end
    message["reasoning"] = (message["reasoning"] || +"") + reasoning
    on_delta.call(reasoning)
  end

  if (text = delta["content"]) && !text.empty?
    if state[:thinking]
      state[:thinking] = false
      on_delta.call("</think>")
    end
    message["content"] = (message["content"] || +"") + text
    on_delta.call(text)
  end

  (delta["tool_calls"] || []).each do |fragment|
    entry = tool_calls[fragment["index"]] ||= {
      "id" => nil,
      "type" => "function",
      "function" => {"name" => +"", "arguments" => +""}
    }
    entry["id"] = fragment["id"] if fragment["id"]
    entry["type"] = fragment["type"] if fragment["type"]

    function = fragment["function"] || {}
    entry["function"]["name"] << function["name"] if function["name"]
    entry["function"]["arguments"] << function["arguments"] if function["arguments"]
  end
end

.chat_completion(base_url:, model:, tools: nil, read_timeout: 300, on_delta: nil) ⇒ Object

Relay worker: sends the conversation to an OpenAI-compatible chat completions endpoint and hands off a new conversation with the assistant's reply (which may include tool calls) appended.

Pass an on_delta callable to stream: it receives each content fragment as it arrives. The worker still returns the conversation with the fully-assembled message appended, so the circuit sees no difference between streamed and unstreamed rounds.

Servers disagree about where thinking goes: some put tags inline in the content, others (e.g. Ollama) stream a separate reasoning field. The worker normalizes the latter to the former, so on_delta always sees one convention: ... inline. The raw reasoning text still lands on the message unaltered.

This is the only part of lyman that knows HTTP exists.



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/lyman/workers/chat_completion.rb', line 26

def self.chat_completion(base_url:, model:, tools: nil, read_timeout: 300, on_delta: nil)
  uri = URI("#{base_url.chomp("/")}/chat/completions")

  relay_worker do |conversation|
    payload = {"model" => model, "messages" => wire_messages(conversation.messages)}
    payload["tools"] = tools if tools && !tools.empty?
    payload["stream"] = true if on_delta

    http = Net::HTTP.new(uri.host, uri.port)
    http.use_ssl = (uri.scheme == "https")
    http.read_timeout = read_timeout

    message =
      if on_delta
        stream_message(http, uri, payload, on_delta)
      else
        fetch_message(http, uri, payload)
      end

    conversation.with_assistant_message(message)
  end
end

.fetch_message(http, uri, payload) ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/lyman/workers/chat_completion.rb', line 59

def self.fetch_message(http, uri, payload)
  response = http.post(
    uri.path,
    JSON.generate(payload),
    "Content-Type" => "application/json"
  )

  unless response.is_a?(Net::HTTPSuccess)
    raise "chat completion failed (#{response.code}): #{response.body}"
  end

  message = JSON.parse(response.body).dig("choices", 0, "message")
  raise "chat completion response had no message: #{response.body}" unless message
  message
end

.parse_arguments(raw) ⇒ Object 



43
44
45
46
47
48
 
# File 'lib/lyman/workers/tool_execution.rb', line 43

def self.parse_arguments(raw)
  return raw if raw.is_a?(Hash)
  JSON.parse(raw.to_s)
rescue JSON::ParserError
  {}
end

.stream_message(http, uri, payload, on_delta) ⇒ Object 



75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/lyman/workers/chat_completion.rb', line 75

def self.stream_message(http, uri, payload, on_delta)
  request = Net::HTTP::Post.new(uri.path, "Content-Type" => "application/json")
  request.body = JSON.generate(payload)

  message = {"role" => "assistant", "content" => nil}
  tool_calls = {}
  state = {thinking: false}
  buffer = +""

  http.request(request) do |response|
    unless response.is_a?(Net::HTTPSuccess)
      raise "chat completion failed (#{response.code}): #{response.read_body}"
    end

    response.read_body do |chunk|
      buffer << chunk
      # SSE events arrive as "data: {json}" lines, but chunks don't
      # align with line boundaries.
      while (newline = buffer.index("\n"))
        line = buffer.slice!(0..newline).strip
        next unless line.start_with?("data:")

        data = line.delete_prefix("data:").strip
        next if data == "[DONE]"

        delta = JSON.parse(data).dig("choices", 0, "delta")
        apply_delta(message, tool_calls, delta, on_delta, state) if delta
      end
    end
  end

  on_delta.call("</think>") if state[:thinking] # reasoning ran to stream end
  unless tool_calls.empty?
    message["tool_calls"] = tool_calls.keys.sort.map { |index| tool_calls[index] }
  end
  message
end

.tool_execution(handlers) ⇒ Object

Relay worker: executes any tool calls pending on the conversation and hands off a new conversation with their results appended. Acts only when the turn isn't already finished and the model has asked for tools; otherwise passes through.

handlers is a hash of tool name => callable taking a Hash of arguments and returning something stringable.



15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/lyman/workers/tool_execution.rb', line 15

def self.tool_execution(handlers)
  relay_worker do |conversation|
    # Fold results onto the accumulator while iterating the *original*
    # conversation's pending calls: each with_tool_result appends a
    # tool message, which empties pending_tool_calls on the new value.
    conversation.pending_tool_calls.reduce(conversation) do |convo, tool_call|
      break convo if convo.finished?

      name = tool_call.dig("function", "name")
      args = parse_arguments(tool_call.dig("function", "arguments"))
      handler = handlers[name]

      result =
        if handler
          begin
            handler.call(args).to_s
          rescue => e
            "Tool #{name} raised #{e.class}: #{e.message}"
          end
        else
          "Unknown tool: #{name}"
        end

      convo.with_tool_result(tool_call["id"], result)
    end
  end
end

.wire_messages(messages) ⇒ Object

The conversation keeps each message's reasoning (it's useful to observability), but it never rides back to the model: providers either ignore it, reject it outright (DeepSeek), or would burn context re-reading thoughts the model already finished thinking.



53
54
55
56
57
 
# File 'lib/lyman/workers/chat_completion.rb', line 53

def self.wire_messages(messages)
  messages.map do |message|
    message.reject { |key, _| key == "reasoning" || key == "reasoning_content" }
  end
end