Observability¶
FlowgentraAI provides tracing and visualization tools for debugging and monitoring graph execution.
Structured Logging¶
Initialize the tracing subscriber at program start:
from flowgentra_ai import init_tracing
init_tracing() # default: "info" level
init_tracing("debug") # more verbose
init_tracing("warn") # less verbose
Execution Tracing¶
Record and inspect execution events:
from flowgentra_ai import ExecutionTracer
tracer = ExecutionTracer()
# Record events manually
tracer.trace_node_start("process")
tracer.trace_node_end("process", duration_ms=150, success=True)
tracer.trace_edge_traversal("process", "output", condition_met=True)
tracer.trace_state_update("key", "new_value")
tracer.trace_custom("my_event", details="some details")
# Export
json_str = tracer.get_events_json()
print(json_str)
# Clear
tracer.clear()
Execution Trace¶
A recorded trace of a graph execution:
from flowgentra_ai import ExecutionTrace
trace = ExecutionTrace(agent_name="my_agent")
# Inspect
path = trace.execution_path() # list of node names
duration = trace.total_duration_ms()
# Serialize
json_str = trace.to_json()
trace = ExecutionTrace.from_json(json_str)
Graph Visualization¶
Export graph structure as diagrams:
graph = builder.compile()
# Mermaid diagram (renders in GitHub, GitLab, Notion, etc.)
mermaid = graph.to_mermaid()
print(mermaid)
# Graphviz DOT format
dot = graph.to_dot()
print(dot)
# JSON structure
structure = graph.to_json()