Setting the file. One moment.
Langsmith Usage · N8n:testmax N8n AI · n8n-io/n8n · Skills Docs
Back to the top of the page Previous
Script Derive Checks
scripts/ langsmith_usage.py
Python · 161 lines · 6 KB
17 each run-start event is also its LangSmith trace id). Use it when other Instance AI work runs
18 at the same time.
19
20 The script also writes context.json: for each agent, the system prompt, the messages before
21 the first model response, and the tools, from the first LLM call of that agent. Tools that the
22 agent loads later with load_tool come from its last call.
23 """
24
25 import argparse
26 import json
27 import urllib.parse
28 import urllib.request
29 from datetime import datetime, timedelta
30 from pathlib import Path
31
32 from n8n_api import setting
33
34
35 def langsmith_client ():
36 key = setting( 'LANGSMITH_API_KEY' )
37 if not key:
38 raise SystemExit ( 'LANGSMITH_API_KEY is not set. Instance AI cost and tokens are not available.' )
39 base = setting( 'LANGSMITH_ENDPOINT' , default = 'https://api.smith.langchain.com' ).rstrip( '/' )
40
41 def call (method, path, body = None ):
42 req = urllib.request.Request(
43 base + path, method = method,
44 data = json.dumps(body).encode() if body is not None else None ,
45 headers = { 'x-api-key' : key, 'content-type' : 'application/json' },
46 )
47 with urllib.request.urlopen(req, timeout = 60 ) as res:
48 return json.load(res)
49
50 return call
51
52
53 def fetch_calls (started_at, finished_at, traces = None ):
54 """LLM calls in the LangSmith project between two ISO times. traces keeps only the calls of these traces."""
55 call = langsmith_client()
56 project = setting( 'LANGSMITH_PROJECT' , 'LANGCHAIN_PROJECT' , default = 'instance-ai' )
57
58 sessions = call( 'GET' , '/api/v1/sessions?' + urllib.parse.urlencode({ 'name' : project}))
59 if not sessions:
60 raise SystemExit ( f 'LangSmith project " { project } " not found.' )
61 session_id = sessions[ 0 ][ 'id' ]
62
63 start = datetime.fromisoformat(started_at) - timedelta( seconds = 5 )
64 end = datetime.fromisoformat(finished_at) + timedelta( seconds = 5 )
65 calls, cursor = [], None
66 while True :
67 body = {
68 'session' : [session_id], 'run_type' : 'llm' , 'limit' : 100 ,
69 'start_time' : start.isoformat(),
70 'select' : [ 'id' , 'name' , 'start_time' , 'end_time' , 'prompt_tokens' , 'completion_tokens' ,
71 'total_cost' , 'prompt_token_details' , 'extra' , 'trace_id' ],
72 }
73 if cursor:
74 body[ 'cursor' ] = cursor
75 res = call( 'POST' , '/api/v1/runs/query' , body)
76 for run in res.get( 'runs' , []):
77 started = datetime.fromisoformat(run[ 'start_time' ].replace( 'Z' , '+00:00' ))
78 if started.tzinfo is None :
79 started = started.replace( tzinfo = start.tzinfo)
80 if started > end or (traces is not None and run.get( 'trace_id' ) not in traces):
81 continue
82 details = run.get( 'prompt_token_details' ) or {}
83 metadata = (run.get( 'extra' ) or {}).get( 'metadata' ) or {}
84 calls.append({
85 'id' : run.get( 'id' ), 'trace' : run.get( 'trace_id' ), 'name' : run.get( 'name' ),
86 'start' : started.isoformat(), 'end' : run.get( 'end_time' ),
87 'in' : run.get( 'prompt_tokens' ) or 0 ,
88 'cache_read' : details.get( 'cache_read' ) or 0 ,
89 'cache_write' : details.get( 'cache_creation' ) or 0 ,
90 'out' : run.get( 'completion_tokens' ) or 0 ,
91 'cost' : float (run.get( 'total_cost' ) or 0 ),
92 'model' : metadata.get( 'ls_model_name' ),
93 })
94 cursor = (res.get( 'cursors' ) or {}).get( 'next' )
95 if not cursor:
96 break
97 return sorted (calls, key =lambda c: c[ 'start' ])
98
99
100 def message_text (content):
101 if isinstance (content, str ):
102 return content
103 return ' \n ' .join(p.get( 'text' , '' ) for p in content or [] if isinstance (p, dict ) and p.get( 'type' ) == 'text' )
104
105
106 def tool_spec (t):
107 return { 'name' : t.get( 'name' ) or t.get( 'id' ), 'description' : t.get( 'description' ) or '' ,
108 'schema' : t.get( 'inputSchema' ) or t.get( 'parameters' ), 'type' : t.get( 'type' )}
109
110
111 def fetch_contexts (calls):
112 """What each agent got before its first model response, in the order that the agents started."""
113 call = langsmith_client()
114 by_agent = {}
115 for c in calls:
116 if c.get( 'id' ):
117 by_agent.setdefault(c[ 'name' ], []).append(c)
118 contexts = []
119 for name, agent_calls in by_agent.items():
120 first = call( 'GET' , f '/api/v1/runs/ { agent_calls[ 0 ][ "id" ] } ' ).get( 'inputs' ) or {}
121 messages = []
122 for m in first.get( 'input' ) or []:
123 if m.get( 'role' ) not in ( 'system' , 'user' ):
124 break
125 messages.append({ 'role' : m[ 'role' ], 'text' : message_text(m.get( 'content' ))})
126 tools = [tool_spec(t) for t in first.get( 'tools' ) or []]
127 loaded = []
128 if len (agent_calls) > 1 :
129 last = call( 'GET' , f '/api/v1/runs/ { agent_calls[ - 1 ][ "id" ] } ' ).get( 'inputs' ) or {}
130 known = {t[ 'name' ] for t in tools}
131 loaded = [tool_spec(t) for t in last.get( 'tools' ) or [] if (t.get( 'name' ) or t.get( 'id' )) not in known]
132 contexts.append({ 'agent' : name, 'messages' : messages, 'tools' : tools, 'loadedTools' : loaded})
133 return contexts
134
135
136 def totals (calls):
137 return {
138 'calls' : len (calls), 'models' : sorted ({c[ 'model' ] for c in calls if c[ 'model' ]}),
139 'input' : sum (c[ 'in' ] for c in calls), 'output' : sum (c[ 'out' ] for c in calls),
140 'cost' : round ( sum (c[ 'cost' ] for c in calls), 4 ),
141 }
142
143
144 def main ():
145 ap = argparse.ArgumentParser()
146 ap.add_argument( '--run-dir' , required = True )
147 ap.add_argument( '--by-trace' , action = 'store_true' )
148 args = ap.parse_args()
149 run_dir = Path(args.run_dir)
150 meta = json.loads((run_dir / 'run.json' ).read_text())
151 traces = set (meta.get( 'langsmithRunIds' ) or [])
152 if args.by_trace and not traces:
153 raise SystemExit ( 'run.json has no langsmithRunIds. Is LANGSMITH_TRACING=true on the instance?' )
154 calls = fetch_calls(meta[ 'startedAt' ], meta.get( 'usageFinishedAt' ) or meta[ 'finishedAt' ], traces if args.by_trace else None )
155 (run_dir / 'usage.json' ).write_text(json.dumps(calls, indent = 1 ))
156 (run_dir / 'context.json' ).write_text(json.dumps(fetch_contexts(calls), indent = 1 ))
157 print (json.dumps(totals(calls), indent = 2 ))
158
159
160 if __name__ == '__main__' :
161 main()