Setting the file. One moment.
Print Summary · Exploring LLM Traces · PostHog/skills · Skills Docs
Back to the top of the page scripts/ print_summary.py
Python · 151 lines · 6 KB
raw[
0
].get(
"type"
)
==
"text"
:
12 raw = json.loads(raw[ 0 ][ "text" ])
13 metadata = raw if isinstance (raw, dict ) else {}
14 results = raw.get( "results" , raw) if isinstance (raw, dict ) else raw
15 return ([results] if isinstance (results, dict ) else results), metadata
16
17
18 def summarize (val, max_len = 500 ):
19 if val is None :
20 return ""
21 s = json.dumps(val, default = str ) if not isinstance (val, str ) else val
22 return s[:max_len] + "..." if len (s) > max_len else s
23
24
25 def extract_final_output (choices):
26 """Extract the text and thinking from the last generation's output choices."""
27 if not isinstance (choices, list ):
28 return None , None
29 parts_text = []
30 parts_thinking = []
31 for choice in choices:
32 content = choice.get( "content" , "" )
33 if isinstance (content, str ):
34 parts_text.append(content)
35 elif isinstance (content, list ):
36 for item in content:
37 if isinstance (item, dict ):
38 if item.get( "type" ) == "thinking" :
39 parts_thinking.append(item.get( "thinking" , "" ))
40 elif item.get( "type" ) == "text" :
41 parts_text.append(item.get( "text" , "" ))
42 return " \n " .join(parts_text) or None , " \n " .join(parts_thinking) or None
43
44
45 def as_float (value):
46 if value is None :
47 return 0.0
48 try :
49 return float (value)
50 except ( TypeError , ValueError ):
51 return 0.0
52
53
54 def print_collection_summary (traces, metadata):
55 if len (traces) <= 1 :
56 return
57
58 print ( f " { '=' * 80 } " )
59 print ( "TRACE COLLECTION SUMMARY" )
60 print ( f " { '=' * 80 } " )
61 print ( f " Total traces: { len (traces) } " )
62 print ( f " Total latency: { sum (as_float(t.get( 'totalLatency' )) for t in traces) :.2f} s" )
63 print ( f " Total cost: $ { sum (as_float(t.get( 'totalCost' )) for t in traces) :.6f} " )
64 print ( f " Tokens in: { int ( sum (as_float(t.get( 'inputTokens' )) for t in traces)) } " )
65 print ( f " Tokens out: { int ( sum (as_float(t.get( 'outputTokens' )) for t in traces)) } " )
66 print ( f " Errors: { int ( sum (as_float(t.get( 'errorCount' )) for t in traces)) } " )
67 if metadata.get( "_posthogUrl" ):
68 print ( f " PostHog URL: { metadata[ '_posthogUrl' ] } " )
69 print ()
70
71
72 max_len = int (os.environ.get( "MAX_LEN" , "500" ))
73
74 traces, metadata = load_trace_file(sys.argv[ 1 ])
75 print_collection_summary(traces, metadata)
76 for trace in traces:
77 print ( f " { '=' * 80 } " )
78 print ( f "TRACE SUMMARY" )
79 print ( f " { '=' * 80 } " )
80 print ( f " ID: { trace.get( 'id' , '?' ) } " )
81 print ( f " Name: { trace.get( 'traceName' , '?' ) } " )
82 print ( f " Created: { trace.get( 'createdAt' , '?' ) } " )
83 print ( f " Person: { trace.get( 'distinctId' , '?' ) } " )
84 print ( f " Latency: { trace.get( 'totalLatency' , '?' ) } s" )
85 print ( f " Cost: $ { trace.get( 'totalCost' , '?' ) } " )
86 print ( f " Tokens in: { trace.get( 'inputTokens' , '?' ) } " )
87 print ( f " Tokens out: { trace.get( 'outputTokens' , '?' ) } " )
88
89 # Trace-level input/output state
90 inp = trace.get( "inputState" )
91 out = trace.get( "outputState" )
92 if inp:
93 print ( f " \n --- Trace input state ---" )
94 print ( f " { summarize(inp, max_len) } " )
95 if out:
96 print ( f " \n --- Trace output state (first { max_len } chars) ---" )
97 print ( f " { summarize(out, max_len) } " )
98
99 events = sorted (trace.get( "events" , []), key =lambda e: e.get( "createdAt" , "" ))
100
101 # Collect models used
102 models = set ()
103 for ev in events:
104 if ev.get( "event" ) == "$ai_generation" :
105 m = ev[ "properties" ].get( "$ai_model" )
106 if m:
107 models.add(m)
108 if models:
109 print ( f " \n Models: { ', ' .join( sorted (models)) } " )
110
111 # Errors
112 errors = [ev for ev in events if ev.get( "properties" , {}).get( "$ai_is_error" )]
113 if errors:
114 print ( f " \n{ '!' * 80 } " )
115 print ( f " ERRORS: { len (errors) } " )
116 for ev in errors:
117 p = ev[ "properties" ]
118 name = p.get( "$ai_span_name" , p.get( "$ai_model" , ev.get( "event" )))
119 print ( f " - { name } : { summarize(p.get( '$ai_output_state' , p.get( '$ai_error' , '?' )), max_len) } " )
120 print ( f " { '!' * 80 } " )
121 else :
122 print ( " \n Errors: None" )
123
124 # Tool calls (spans with input/output state)
125 spans = [ev for ev in events if ev.get( "event" ) == "$ai_span" and ev.get( "properties" , {}).get( "$ai_input_state" )]
126 if spans:
127 print ( f " \n{ '=' * 80 } " )
128 print ( f "TOOL CALLS ( { len (spans) } spans with I/O)" )
129 print ( f " { '=' * 80 } " )
130 for ev in spans:
131 p = ev[ "properties" ]
132 name = p.get( "$ai_span_name" , "?" )
133 latency = p.get( "$ai_latency" , "?" )
134 error = " [ERROR]" if p.get( "$ai_is_error" ) else ""
135 print ( f " \n [ { name } ] ( { latency } s) { error } " )
136 print ( f " IN: { summarize(p.get( '$ai_input_state' ), max_len) } " )
137 print ( f " OUT: { summarize(p.get( '$ai_output_state' ), max_len) } " )
138
139 # Final LLM output (last generation)
140 generations = [ev for ev in events if ev.get( "event" ) == "$ai_generation" ]
141 if generations:
142 last_gen = generations[ - 1 ]
143 p = last_gen[ "properties" ]
144 text, thinking = extract_final_output(p.get( "$ai_output_choices" , []))
145 print ( f " \n{ '=' * 80 } " )
146 print ( f "FINAL LLM OUTPUT ( { p.get( '$ai_model' , '?' ) } )" )
147 print ( f " { '=' * 80 } " )
148 if thinking:
149 print ( f " \n [thinking] { thinking[:max_len] }{ '...' if len (thinking) > max_len else '' } " )
150 if text:
151 print ( f " \n { text[:max_len * 2 ] }{ '...' if len (text) > max_len * 2 else '' } " )