Class: LangGraphRB::Observers::StructuredObserver

Inherits:
BaseObserver
  • Object
show all
Defined in:
lib/langgraph_rb/observers/structured.rb

Overview

Structured data observer for APM/monitoring integration

Instance Method Summary collapse

Methods inherited from BaseObserver

#on_checkpoint_saved, #on_interrupt, #on_state_change

Constructor Details

#initialize(sink: nil, format: :json, include_state: false, async: false) ⇒ StructuredObserver

Returns a new instance of StructuredObserver.



7
8
9
10
11
12
13
14
 
# File 'lib/langgraph_rb/observers/structured.rb', line 7

def initialize(sink: nil, format: :json, include_state: false, async: false)
  @sink = sink || $stdout
  @format = format
  @include_state = include_state
  @async = async
  @event_queue = async ? Queue.new : nil
  @worker_thread = start_worker_thread if async
end

Instance Method Details

#on_command_processed(event) ⇒ Object 



40
41
42
 
# File 'lib/langgraph_rb/observers/structured.rb', line 40

def on_command_processed(event)
  emit_event(:command_processed, event)
end

#on_graph_end(event) ⇒ Object 



20
21
22
 
# File 'lib/langgraph_rb/observers/structured.rb', line 20

def on_graph_end(event)
  emit_event(:graph_end, event.to_h)
end

#on_graph_start(event) ⇒ Object 



16
17
18
 
# File 'lib/langgraph_rb/observers/structured.rb', line 16

def on_graph_start(event)
  emit_event(:graph_start, event.to_h)
end

#on_node_end(event) ⇒ Object 



28
29
30
 
# File 'lib/langgraph_rb/observers/structured.rb', line 28

def on_node_end(event)
  emit_event(:node_end, sanitize_event(event.to_h))
end

#on_node_error(event) ⇒ Object 



32
33
34
 
# File 'lib/langgraph_rb/observers/structured.rb', line 32

def on_node_error(event)
  emit_event(:node_error, sanitize_event(event.to_h))
end

#on_node_start(event) ⇒ Object 



24
25
26
 
# File 'lib/langgraph_rb/observers/structured.rb', line 24

def on_node_start(event)
  emit_event(:node_start, sanitize_event(event.to_h))
end

#on_step_complete(event) ⇒ Object 



36
37
38
 
# File 'lib/langgraph_rb/observers/structured.rb', line 36

def on_step_complete(event)
  emit_event(:step_complete, sanitize_event(event.to_h))
end

#shutdown ⇒ Object 



44
45
46
47
48
49
 
# File 'lib/langgraph_rb/observers/structured.rb', line 44

def shutdown
  if @async && @worker_thread
    @event_queue << :shutdown
    @worker_thread.join
  end
end