Setting the file. One moment.
Extract Conversation · Exploring LLM Traces · PostHog/skills · Skills Docs
Back to the top of the page scripts/extract_conversation.py
scripts/ extract_conversation.py
Python · 106 lines · 4 KB
json.load(f)
15 # Claude Code persists large MCP tool results as [{"type": "text", "text": "<json>"}] — unwrap to get the actual trace data.
16 if isinstance (raw, list ) and raw and raw[ 0 ].get( "type" ) == "text" :
17 raw = json.loads(raw[ 0 ][ "text" ])
18 # Both query-llm-trace and query-llm-traces-list return {"results": [...]}, but handle a bare trace object too.
19 results = raw.get( "results" , raw)
20 return [results] if isinstance (results, dict ) else results
21
22
23 def truncate (text, max_len):
24 if max_len <= 0 or len (text) <= max_len:
25 return text
26 half = max_len // 2
27 return text[:half] + f " \n ... [ { len (text) } chars] ... \n " + text[ - half:]
28
29
30 def format_content (content, max_len):
31 """Format message content, preserving thinking/text/tool_use structure."""
32 if isinstance (content, str ):
33 return truncate(content, max_len)
34 if not isinstance (content, list ):
35 return str (content)
36
37 parts = []
38 for item in content:
39 if not isinstance (item, dict ):
40 parts.append( str (item))
41 continue
42 item_type = item.get( "type" , "" )
43 if item_type == "thinking" :
44 thinking = item.get( "thinking" , "" )
45 parts.append( f " [thinking] { truncate(thinking, max_len) } " )
46 elif item_type == "text" :
47 parts.append( f " { truncate(item.get( 'text' , '' ), max_len) } " )
48 elif item_type == "tool_use" :
49 name = item.get( "name" , "?" )
50 tool_input = json.dumps(item.get( "input" , {}), default = str )
51 parts.append( f " [tool_use: { name } ] { truncate(tool_input, max_len) } " )
52 elif item_type == "tool_result" :
53 tool_id = item.get( "tool_use_id" , "?" )
54 result_content = item.get( "content" , "" )
55 if isinstance (result_content, list ):
56 result_content = " " .join(
57 p.get( "text" , "" ) for p in result_content if isinstance (p, dict )
58 )
59 parts.append( f " [tool_result: { tool_id } ] { truncate( str (result_content), max_len) } " )
60 else :
61 parts.append( f " [ { item_type } ] { truncate(json.dumps(item, default = str ), max_len) } " )
62 return " \n " .join(parts)
63
64
65 max_len = int (os.environ.get( "MAX_LEN" , "500" ))
66
67 traces = load_trace_file(sys.argv[ 1 ])
68 for trace in traces:
69 for ev in sorted (trace.get( "events" , []), key =lambda e: e.get( "createdAt" , "" )):
70 if ev.get( "event" ) != "$ai_generation" :
71 continue
72 p = ev.get( "properties" , {})
73 messages = p.get( "$ai_input" )
74 if not isinstance (messages, list ):
75 continue
76 model = p.get( "$ai_model" , "?" )
77 print ( f " \n{ '=' * 80 } " )
78 print ( f "Generation: { model } ( { ev.get( 'createdAt' , '?' ) } )" )
79 print ( f " { '=' * 80 } " )
80 for msg in messages:
81 role = msg.get( "role" , "?" )
82 content = msg.get( "content" , "" )
83
84 # Show tool_calls on assistant messages
85 tool_calls = msg.get( "tool_calls" , [])
86
87 print ( f " \n [ { role.upper() } ]" )
88 print (format_content(content, max_len))
89
90 if tool_calls:
91 for tc in tool_calls:
92 fn = tc.get( "function" , tc)
93 name = fn.get( "name" , "?" )
94 args = fn.get( "arguments" , " {} " )
95 if isinstance (args, str ):
96 args_str = args
97 else :
98 args_str = json.dumps(args, default = str )
99 print ( f " [tool_call: { name } ] { truncate(args_str, max_len) } " )
100
101 # Show output choices
102 choices = p.get( "$ai_output_choices" , [])
103 if choices:
104 print ( f " \n [ASSISTANT (output)]" )
105 for choice in choices:
106 print (format_content(choice.get( "content" , "" ), max_len))