Setting the file. One moment.
Evaluate Traces · AWS Observability · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page 69
Creating Amazon Aurora Db Cluster With Instances
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Debugging Lambda Timeouts
def _service_name
— line 512
This file
Number 37.41
Position 41 of 67
Type Python
Size 42 KB
Lines 982 scripts/cloudwatch-omni/ evaluate_traces.py
Python · 982 lines · 42 KB
14 this process (kept out of the model's context).
15
16 Scoring goes through the AgentCore data plane (`bedrock-agentcore evaluate`) on session
17 spans fetched from CloudWatch Omni. Do NOT hand-extract input/output/toolCalls or call
18 any other Evaluate API (e.g. any cloudwatch-omni Evaluate operation) or assemble
19 `structuredInput` by hand: that manual reshaping is exactly the error-prone work this
20 helper exists to replace.
21
22 Doing the span-shaping in code is deliberate: hand-shaping spans turn-by-turn is
23 the step an agent thrashes on (reshape → ValidationException → retry → loop). This makes
24 the sync path deterministic. Batch is fetch-once/score-many so scoring N traces against M
25 evaluators is N fetches + N×M scores, never N×M fetches.
26
27 Each score is also written back as a `gen_ai.evaluation.result` telemetry record (the same
28 shape online evaluation emits, minus the online-config attribute) so it is queryable later
29 instead of living only in this conversation. That step is best-effort — it never fails the
30 run, and it is skipped silently when the caller lacks logs write permission. Pass
31 `--no-writeback` to turn it off.
32
33 SENSITIVE DATA: an evaluator's `explanation` is the judge's prose, and it quotes the agent's own
34 inputs and outputs. Two paths carry it, with different defaults:
35
36 * The RECEIPT (stdout, so it reaches the calling agent's context, transcript and client logs)
37 omits it unless `--include-explanations` is passed; by default each row reports only
38 `hasExplanation`, so the caller knows one exists and can ask for it. Scores, labels and
39 error codes are always returned.
40 * The WRITEBACK record persists it as `gen_ai.evaluation.explanation` — that is the point of
41 the record (a score without its "why" is not much use when read back later). So when this
42 run CREATES the results log group it provisions it deliberately: `--retention-days`
43 (DEFAULT_RESULTS_RETENTION_DAYS; 0 leaves the account default) bounds how long they live, and
44 `--kms-key-id` encrypts the group with a customer-managed key instead of the AWS-owned
45 default. Both apply ONLY to a group this run creates — a pre-existing group keeps whatever
46 retention and key it already has, which are the owner's to set. If retention cannot be set
47 the receipt says so in `notes` rather than leaving an unbounded group unremarked. Pass
48 `--no-writeback` to persist nothing at all.
49
50 Requires boto3 + caller AWS credentials with cloudwatch-omni (Omni SQL span queries) +
51 bedrock-agentcore (data plane evaluate); the writeback additionally uses logs:PutLogEvents
52 (and logs:CreateLogGroup / logs:CreateLogStream / logs:PutRetentionPolicy the first time). A `--kms-key-id` key
53 policy must let the `logs.<region>.amazonaws.com` service principal use the key, or group
54 creation fails (the run does NOT fall back to an unencrypted group).
55
56 Examples:
57 # one trace, one evaluator
58 python evaluate_traces.py --trace-id 6a82... --evaluator-id Builtin.Helpfulness
59 # a sample of traces × two evaluators (matrix; each trace fetched once)
60 python evaluate_traces.py --trace-ids 6a82...,7b91...,8c03... \
61 --evaluator-ids Builtin.Helpfulness,Builtin.ToolSelectionAccuracy --level TOOL_CALL
62 python evaluate_traces.py --trace-id 6a82... --include-explanations # return the "why" too
63 python evaluate_traces.py --trace-id 6a82... --no-writeback # score only, don't persist
64 # persist under a customer-managed key with 90-day retention
65 python evaluate_traces.py --trace-id 6a82... --kms-key-id arn:aws:kms:... --retention-days 90
66 """
67
68 from __future__ import annotations
69
70 import argparse
71 import json
72 import re
73 import sys
74 import time
75 from typing import NoReturn
76
77 import boto3
78 from botocore.exceptions import BotoCoreError, ClientError
79
80 EVALUATE_SERVICE = "bedrock-agentcore" # data plane hosts `evaluate`
81 DEFAULT_EVALUATOR = "Builtin.Helpfulness"
82 LEVELS = ( "TRACE" , "TOOL_CALL" , "SESSION" )
83 # Batch bounds (match the on-demand evaluate-traces tool: 50 traces / 10 evaluators / 100 pairs).
84 # All THREE apply independently and are enforced by TRUNCATING with a note in the receipt — never a
85 # silent cap. `traces × evaluators` fans out into one evaluate call per pair, so a big matrix would
86 # throttle and blow the timeout.
87 MAX_TRACES = 50 # unique traces fetched per run
88 MAX_EVALUATORS = 10 # distinct evaluators per run
89 MAX_PAIRS = 100 # total (trace, evaluator) PAIRS per run
90 # Hard ceiling on ac.evaluate() calls per run. TRACE/SESSION issue 1 call per pair (so MAX_PAIRS
91 # bounds them), but TOOL_CALL chunks tool spans into ≤MAX_TARGET_IDS-id calls, so one pair can be
92 # many calls — this bounds the total fan-out (throttle/timeout guard) regardless of level. This is
93 # NOT one of the three authoritative caps above: it maps to the ~100-evaluations/minute account
94 # quota, so raising it risks throttling. With MAX_PAIRS now == 100 it equals this ceiling, so a
95 # full 100-pair TRACE/SESSION run spends exactly the budget; TOOL_CALL runs can hit it sooner.
96 MAX_EVALUATE_CALLS = 100
97 # Evaluators that require ground truth (an expected tool trajectory) and so CANNOT score a raw
98 # trace: this helper sends `sessionSpans` only, no `evaluationReferenceInputs`, so they would
99 # error or score nothing. The evaluator APIs expose NO machine-readable requirement flag
100 # (get-evaluator returns only id/name/description/evaluatorConfig/level/status), so this list is
101 # maintained by hand — the trajectory matchers are the built-ins that compare against an
102 # expected sequence. Do NOT "helpfully" delete it as a hardcoded catalog: there is no flag to
103 # read instead. Correctness / GoalSuccessRate are deliberately absent — they can optionally use
104 # ground truth but run without it, so on-demand scoring of a raw trace is valid for them.
105 _GROUND_TRUTH_REQUIRED = frozenset (
106 {
107 "Builtin.TrajectoryExactOrderMatch" ,
108 "Builtin.TrajectoryInOrderMatch" ,
109 "Builtin.TrajectoryAnyOrderMatch" ,
110 }
111 )
112 # On-demand `evaluate` returns the score inline but does NOT persist it, so a score would be
113 # lost the moment the conversation ends. Write each one back as a telemetry record — same
114 # `gen_ai.evaluation.result` shape online evaluation emits — so scores can be queried later
115 # (by a later session or another user) alongside the online ones.
116 EVAL_RESULTS_LOG_GROUP_PREFIX = "/aws/cloudwatch/evaluations/results/"
117 EVAL_RESULTS_LOG_STREAM = "on-demand"
118 EVAL_RESULT_RECORD_NAME = "gen_ai.evaluation.result"
119 # The persisted record carries the judge's explanation (which quotes agent input/output), so a
120 # group this run creates gets a BOUNDED retention rather than inheriting the account default —
121 # which can be "never expire". Overridable per run; 0 opts out for accounts that manage retention
122 # centrally. Only ever applied to a group this run creates.
123 DEFAULT_RESULTS_RETENTION_DAYS = 30
124 # The only retentionInDays values CloudWatch Logs accepts. Validated up front so a bad value fails
125 # the run instead of failing put_retention_policy AFTER the group was created — which would leave
126 # exactly the unbounded group the default exists to avoid.
127 _RETENTION_DAYS_ALLOWED = frozenset (
128 (
129 1 ,
130 3 ,
131 5 ,
132 7 ,
133 14 ,
134 30 ,
135 60 ,
136 90 ,
137 120 ,
138 150 ,
139 180 ,
140 365 ,
141 400 ,
142 545 ,
143 731 ,
144 1096 ,
145 1827 ,
146 2192 ,
147 2557 ,
148 2922 ,
149 3288 ,
150 3653 ,
151 )
152 )
153 _MAX_LOG_GROUP_LEN = 512
154 # Log-group names allow [.-_/#A-Za-z0-9]; anything else in a service name is replaced.
155 _LOG_GROUP_SAFE_RE = re.compile( r " [ ^A-Za-z0-9_. \- /# ] " )
156 # Writeback is a courtesy, not the job: if the caller simply lacks logs write permission,
157 # skip silently rather than nagging about a permission they may not want to grant.
158 _ACCESS_DENIED_CODES = frozenset (
159 ( "AccessDenied" , "AccessDeniedException" , "UnauthorizedException" , "AuthorizationError" )
160 )
161 # Distinguishes "not written, but nothing worth telling the user about" from a real success,
162 # so the receipt never claims a score was persisted when it was not.
163 _WRITEBACK_SKIPPED = object ()
164 # Used for BOTH the log-group segment and the record's service.name when the scored spans
165 # carry no service name, so the stored record always agrees with where it was filed.
166 _UNKNOWN_SERVICE = "default"
167 # fetch LIMIT+1 so an exactly-full result is detectable as truncated; paginate to collect
168 # all rows (maxResults cap is 1000 per page).
169 SPAN_ROW_LIMIT = 1000
170 QUERY_POLL_ATTEMPTS = 150 # × 2s = up to 5 min for an Omni SQL query to complete
171 _TRACE_ID_RE = re.compile( r " ^[ 0-9a-fA-F ] {1,64} $ " )
172 # Session ids are alphanumeric + a few separators; validate (reject) rather than mangle,
173 # so a malformed id never silently rewrites to a DIFFERENT real session's id.
174 _SESSION_ID_RE = re.compile( r " ^[ 0-9A-Za-z._: \- ] {1,128} $ " )
175
176 # OTLP SpanKind enum → bare OTel name. `evaluate` requires `kind` to be a STRING; the
177 # stored telemetry is polymorphic (some emitters use "SERVER", others the numeric enum,
178 # some spans omit kind / send null), so coerce every shape to the bare name.
179 _KIND_BY_NUMBER = {
180 0 : "UNSPECIFIED" ,
181 1 : "INTERNAL" ,
182 2 : "SERVER" ,
183 3 : "CLIENT" ,
184 4 : "PRODUCER" ,
185 5 : "CONSUMER" ,
186 }
187 _VALID_KINDS = frozenset ( _KIND_BY_NUMBER .values())
188 _DEFAULT_KIND = "UNSPECIFIED"
189
190 # Span attribute columns to SELECT from traces.default and their SQL aliases.
191 # Maps (OTLP attribute key, SQL column alias); only non-null values are included in
192 # the reconstructed span's attributes dict.
193 #
194 # Coverage note: Omni SQL does not expose a wildcard `attributes.*` projection, so only
195 # known keys can be fetched. This list covers all attributes the built-in AgentCore
196 # evaluators are known to consume. Custom evaluators that rely on attributes not listed
197 # here will receive incomplete span context — extend _ATTR_COLS if that is the case.
198 _ATTR_COLS = [
199 # Span kind (agent framework conventions)
200 ( "openinference.span.kind" , "oi_span_kind" ),
201 ( "aws.genai.span_kind" , "aws_genai_span_kind" ),
202 ( "gen_ai.operation.name" , "gen_ai_op_name" ),
203 # Input / output (string-valued keys; structured-array keys like gen_ai.input.messages
204 # are included as-is — evaluators that need them receive whatever value Omni stores)
205 ( "input.value" , "input_value" ),
206 ( "output.value" , "output_value" ),
207 ( "gen_ai.input_value" , "gen_ai_input_value" ),
208 ( "gen_ai.output_value" , "gen_ai_output_value" ),
209 ( "gen_ai.input.messages" , "gen_ai_input_messages" ),
210 ( "gen_ai.output.messages" , "gen_ai_output_messages" ),
211 ( "llm.input_messages" , "llm_input_messages" ),
212 ( "llm.output_messages" , "llm_output_messages" ),
213 # Tool call
214 ( "tool.name" , "tool_name" ),
215 ( "gen_ai.tool.name" , "gen_ai_tool_name" ),
216 ( "gen_ai.tool.call.id" , "gen_ai_tool_call_id" ),
217 # Session identity
218 ( "session.id" , "session_id_attr" ),
219 # LLM metadata
220 ( "gen_ai.request.model" , "gen_ai_req_model" ),
221 ( "gen_ai.response.model" , "gen_ai_resp_model" ),
222 ( "gen_ai.system" , "gen_ai_system" ),
223 # Token counts
224 ( "gen_ai.usage.input_tokens" , "usage_input_tokens" ),
225 ( "gen_ai.usage.output_tokens" , "usage_output_tokens" ),
226 ( "llm.token_count.prompt" , "prompt_tokens" ),
227 ( "llm.token_count.completion" , "completion_tokens" ),
228 ]
229
230 # Pre-built SELECT fragment — one alias per attribute key.
231 _ATTR_SELECT = ", " .join(
232 "attributes[' %s '] AS %s " % (key, alias) for key, alias in _ATTR_COLS
233 )
234
235
236 def _die (payload) -> NoReturn:
237 """Print an error receipt to stdout (so the agent sees it) and exit non-zero."""
238 print (json.dumps(payload))
239 sys.exit( 1 )
240
241
242 def _split_ids ( * values, lower: bool = False ):
243 """Flatten comma-separated id args into a de-duplicated, order-preserving list.
244
245 Pass lower=True for trace ids: Omni SQL traceId values are lowercase hex, so normalise
246 on input. Do NOT pass lower=True for evaluator ids — the evaluate API and
247 _GROUND_TRUTH_REQUIRED use the original casing.
248 """
249 ids = []
250 for v in values:
251 if v:
252 ids += [(x.strip().lower() if lower else x.strip()) for x in v.split( "," ) if x.strip()]
253 return list ( dict .fromkeys(ids))
254
255
256 def _normalize_span_kind (value):
257 """Coerce a raw span `kind` (numeric enum / SPAN_KIND_* / null / missing) to a bare
258 OTel SpanKind string. `bool` is NOT treated as a numeric enum."""
259 if isinstance (value, bool ):
260 return _DEFAULT_KIND
261 if isinstance (value, int ):
262 return _KIND_BY_NUMBER .get(value, _DEFAULT_KIND )
263 if isinstance (value, float ) and value.is_integer():
264 return _KIND_BY_NUMBER .get( int (value), _DEFAULT_KIND )
265 if isinstance (value, str ):
266 text = value.strip()
267 if not text:
268 return _DEFAULT_KIND
269 if text.isdecimal(): # NOT isdigit(): "²"/"³" are digits but int() rejects them
270 return _KIND_BY_NUMBER .get( int (text), _DEFAULT_KIND )
271 upper = text.upper()
272 for prefix in ( "SPAN_KIND_" , "SPANKIND." ):
273 if upper.startswith(prefix):
274 upper = upper[ len (prefix):]
275 break
276 return upper if upper in _VALID_KINDS else _DEFAULT_KIND
277 return _DEFAULT_KIND
278
279
280 def _row_to_span (row):
281 """Reconstruct an OTLP-compatible span dict from an Omni SQL result row.
282
283 Only non-empty values are included in the attributes dict so the evaluator receives
284 a clean document rather than a dict full of null/empty-string keys.
285 """
286 attrs = {key: row[alias] for key, alias in _ATTR_COLS if row.get(alias) not in ( None , "" )}
287 span: dict = {
288 "traceId" : row.get( "traceId" ),
289 "spanId" : row.get( "spanId" ),
290 "name" : row.get( "name" ) or "" ,
291 "kind" : _normalize_span_kind(row.get( "kind" )),
292 "attributes" : attrs,
293 "resource" : { "attributes" : {}},
294 }
295 if row.get( "parentSpanId" ):
296 span[ "parentSpanId" ] = row[ "parentSpanId" ]
297 if row.get( "startTimeUnixNano" ):
298 span[ "startTimeUnixNano" ] = row[ "startTimeUnixNano" ]
299 if row.get( "endTimeUnixNano" ):
300 span[ "endTimeUnixNano" ] = row[ "endTimeUnixNano" ]
301 if row.get( "svc_name" ):
302 span[ "resource" ][ "attributes" ][ "service.name" ] = row[ "svc_name" ]
303 if row.get( "scope_name" ):
304 span[ "scope" ] = { "name" : row[ "scope_name" ]}
305 if row.get( "status_code" ):
306 span[ "status" ] = { "code" : row[ "status_code" ]}
307 return span
308
309
310 # ---- span fetch via CloudWatch Omni SQL (traces.default) ----
311
312 def _omni_query (omni, sql, soft = False ):
313 """Run an Omni SQL query and paginate to collect all rows.
314
315 Returns (rows, error). soft=True returns (None, message) on failure so callers can
316 degrade gracefully (e.g. session resolution has a traceId fallback).
317 """
318 session_id = None
319 rows: list = []
320 try :
321 try :
322 session_id = omni.start_telemetry_query_session(
323 sessionName = "eval-traces- %d " % int (time.time())
324 )[ "sessionId" ]
325 except (BotoCoreError, ClientError) as e:
326 msg = "Omni start_telemetry_query_session failed: %s " % e
327 if soft:
328 return None , msg
329 _die({ "error" : msg})
330
331 try :
332 qid = omni.start_telemetry_query(
333 sessionId = session_id, queryString = sql
334 )[ "queryId" ]
335 except (BotoCoreError, ClientError) as e:
336 msg = "Omni start_telemetry_query failed: %s " % e
337 if soft:
338 return None , msg
339 _die({ "error" : msg})
340
341 # Phase 1: poll for completion (status only)
342 status = None
343 for _ in range ( QUERY_POLL_ATTEMPTS ):
344 time.sleep( 2 )
345 try :
346 poll = omni.get_telemetry_query_results( queryId = qid, maxResults = 1 )
347 except (BotoCoreError, ClientError) as e:
348 msg = "Omni get_telemetry_query_results failed: %s " % e
349 if soft:
350 return None , msg
351 _die({ "error" : msg})
352 status = poll.get( "status" )
353 if status in ( "Complete" , "Failed" , "Cancelled" ):
354 break
355
356 if status != "Complete" :
357 msg = (
358 "Omni SQL query did not complete (status= %s ); "
359 "narrow --window-days to speed it up" % status
360 )
361 if soft:
362 return None , msg
363 _die({ "error" : msg})
364
365 # Phase 2: paginate to collect all rows (maxResults cap is 1000 per page)
366 next_token = None
367 while len (rows) <= SPAN_ROW_LIMIT :
368 try :
369 kwargs: dict = { "queryId" : qid, "maxResults" : 1000 }
370 if next_token:
371 kwargs[ "nextToken" ] = next_token
372 page = omni.get_telemetry_query_results( ** kwargs)
373 except (BotoCoreError, ClientError) as e:
374 msg = "Omni get_telemetry_query_results (fetch) failed: %s " % e
375 if soft:
376 return None , msg
377 _die({ "error" : msg})
378 page_rows = page.get( "rows" , [])
379 rows.extend(page_rows)
380 next_token = page.get( "nextToken" )
381 if not next_token or not page_rows: # no token or empty page → done
382 break
383 finally :
384 if session_id:
385 try :
386 omni.stop_telemetry_query_session( sessionId = session_id)
387 except Exception :
388 pass
389
390 return rows, None
391
392
393 def _resolve_session_id (omni, trace_id, window_days):
394 """Find the trace's session.id via Omni SQL, or None. Non-fatal on failure."""
395 sql = (
396 "SELECT attributes['session.id'] AS session_id "
397 "FROM \" traces.default \" "
398 "WHERE traceId = ' %s ' "
399 "AND attributes['session.id'] IS NOT NULL "
400 "AND \" @timestamp \" BETWEEN NOW() - INTERVAL ' %d days' AND NOW() "
401 "LIMIT 1"
402 ) % (trace_id, window_days)
403 rows, _err = _omni_query(omni, sql, soft = True )
404 for row in rows or []:
405 sid = row.get( "session_id" )
406 if sid:
407 return sid
408 return None
409
410
411 def _fetch_spans (omni, filter_clause, window_days, soft = False ):
412 """Fetch spans from traces.default for the given WHERE clause.
413
414 Returns (spans, truncated, error). spans is a list of OTLP-compatible span dicts
415 reconstructed from Omni SQL columns. truncated is True when the SQL LIMIT was hit.
416 soft=True returns (None, False, message) on failure so one trace's error doesn't
417 sink the whole batch.
418 """
419 sql = (
420 "SELECT traceId, spanId, parentSpanId, name, kind, "
421 "startTimeUnixNano, endTimeUnixNano, "
422 "resource['attributes']['service.name'] AS svc_name, "
423 "scope['name'] AS scope_name, "
424 "status['code'] AS status_code, "
425 + _ATTR_SELECT
426 + " FROM \" traces.default \" "
427 "WHERE %s "
428 "AND \" @timestamp \" BETWEEN NOW() - INTERVAL ' %d days' AND NOW() "
429 "ORDER BY startTimeUnixNano ASC "
430 # fetch one past the cap so an exactly-full result is detectable as truncated
431 "LIMIT %d "
432 ) % (filter_clause, window_days, SPAN_ROW_LIMIT + 1 )
433
434 rows, err = _omni_query(omni, sql, soft = soft)
435 if rows is None :
436 return None , False , err
437
438 truncated = len (rows) > SPAN_ROW_LIMIT
439 if truncated:
440 rows = rows[: SPAN_ROW_LIMIT ] # drop the probe row before building spans
441
442 spans = [_row_to_span(r) for r in rows if r.get( "traceId" )]
443 return spans, truncated, None
444
445
446 def _tool_span_ids (spans):
447 """Span ids of TOOL spans (for TOOL_CALL level), matching the bespoke tool's detection."""
448 ids = []
449 for s in spans:
450 attrs = s.get( "attributes" ) or {}
451 is_tool = (
452 attrs.get( "openinference.span.kind" ) == "TOOL"
453 or attrs.get( "gen_ai.operation.name" ) == "execute_tool"
454 )
455 sid = s.get( "spanId" )
456 if is_tool and isinstance (sid, str ) and len (sid) == 16 :
457 ids.append(sid)
458 return ids
459
460
461 def _summarize (response, evaluator_id, include_explanations = False ):
462 """Map a raw evaluate response to compact rows; the raw envelope is NOT returned.
463
464 The explanation is the judge's prose and quotes the agent's own inputs and outputs, so it is
465 kept OUT of the caller-facing receipt unless `--include-explanations` asks for it: the receipt
466 goes to stdout, which lands in the calling agent's context, its transcript and client logs.
467 `hasExplanation` still tells the caller one exists and can be requested, so the score is never
468 silently un-explainable.
469
470 The text is retained on the PRIVATE `_explanation` key either way, because the writeback record
471 persists it regardless (see _eval_result_records — a stored score without its "why" is little
472 use when read back later, and that path is bounded by retention/CMK instead of by omission).
473 `_public_rows` strips every private key before the receipt is printed.
474 """
475 rows = []
476 for res in response.get( "evaluationResults" ) or []:
477 explanation = str (res.get( "explanation" ) or "" )
478 clipped = explanation[: 600 ] + ( "…" if len (explanation) > 600 else "" )
479 row = {
480 "evaluatorId" : res.get( "evaluatorId" , evaluator_id),
481 "value" : res.get( "value" ),
482 "label" : res.get( "label" ),
483 "_explanation" : clipped,
484 }
485 if include_explanations:
486 row[ "explanation" ] = clipped
487 else :
488 row[ "hasExplanation" ] = bool (explanation)
489 if res.get( "errorCode" ):
490 row[ "errorCode" ] = res.get( "errorCode" )
491 if res.get( "errorMessage" ):
492 row[ "errorMessage" ] = str (res.get( "errorMessage" ))[: 600 ]
493 rows.append(row)
494 return rows
495
496
497 def _public_rows (rows):
498 """Drop private (`_`-prefixed) keys — what the receipt may print, vs. what writeback may read.
499
500 Kept as the LAST step before printing so nothing that only exists for the writeback path can
501 reach stdout by accident.
502 """
503 return [{k: v for k, v in row.items() if not k.startswith( "_" )} for row in rows]
504
505
506 def _log_group_segment (service):
507 """Sanitize a service name into a single log-group path segment."""
508 seg = _LOG_GROUP_SAFE_RE .sub( "_" , (service or "" ).strip()) or _UNKNOWN_SERVICE
509 return seg[: _MAX_LOG_GROUP_LEN - len ( EVAL_RESULTS_LOG_GROUP_PREFIX )]
510
511
512 def _service_name (spans, trace_id):
513 """The scored trace's `resource.attributes['service.name']` (its own span wins)."""
514 fallback = None
515 for s in spans:
516 name = ((s.get( "resource" ) or {}).get( "attributes" ) or {}).get( "service.name" )
517 if not isinstance (name, str ) or not name:
518 continue
519 if s.get( "traceId" ) == trace_id:
520 return name
521 fallback = fallback or name
522 return fallback
523
524
525 def _eval_result_records (rows, trace_id, session_id, level, service, ts_ms, partial = False ):
526 """Build one `gen_ai.evaluation.result` record per SCORED row."""
527 records = []
528 for row in rows:
529 if row.get( "errorCode" ) or row.get( "value" ) is None :
530 continue
531 attributes = {
532 "gen_ai.evaluation.name" : row.get( "evaluatorId" ),
533 "gen_ai.evaluation.score.value" : str (row.get( "value" )),
534 "aws.bedrock_agentcore.evaluation_level" : level,
535 }
536 if session_id:
537 attributes[ "session.id" ] = session_id
538 if row.get( "label" ) is not None :
539 attributes[ "gen_ai.evaluation.score.label" ] = row.get( "label" )
540 if row.get( "_explanation" ):
541 attributes[ "gen_ai.evaluation.explanation" ] = row.get( "_explanation" )
542 if partial:
543 attributes[ "gen_ai.evaluation.partial" ] = "true"
544 records.append(
545 {
546 "name" : EVAL_RESULT_RECORD_NAME ,
547 "traceId" : trace_id,
548 "attributes" : attributes,
549 "resource" : { "attributes" : { "service.name" : service}},
550 "@timestamp" : ts_ms,
551 }
552 )
553 return records
554
555
556 def _err_code (e):
557 """The `Error.Code` of a ClientError, or "" when the envelope does not carry one."""
558 return (e.response.get( "Error" ) or {}).get( "Code" ) or ""
559
560
561 def _provision_group (logs, group, retention_days, kms_key_id):
562 """Create the results log group + stream, retention-bounded and optionally CMK-encrypted."""
563 warning = None
564 kwargs = { "logGroupName" : group}
565 if kms_key_id:
566 kwargs[ "kmsKeyId" ] = kms_key_id
567 try :
568 logs.create_log_group( ** kwargs)
569 created = True
570 except BotoCoreError as e:
571 return "could not persist the score for later querying: %s " % e, None
572 except ClientError as e:
573 code = _err_code(e)
574 if code in _ACCESS_DENIED_CODES :
575 return _WRITEBACK_SKIPPED , None
576 if code == "ResourceAlreadyExistsException" :
577 created = False
578 elif kms_key_id and code in ( "InvalidParameterException" , "ValidationException" ):
579 return (
580 "could not create %s with kmsKeyId %s (does the key policy allow the CloudWatch "
581 "Logs service principal to use it?): %s " % (group, kms_key_id, e)
582 ), None
583 else :
584 return "could not persist the score for later querying: %s " % e, None
585
586 if created and retention_days:
587 try :
588 logs.put_retention_policy( logGroupName = group, retentionInDays = retention_days)
589 except (ClientError, BotoCoreError) as e:
590 warning = (
591 "created %s but could not set %d -day retention (needs logs:PutRetentionPolicy), "
592 "so it keeps the account default retention — which may be 'never expire', and "
593 "these records carry the evaluator's explanation: %s " % (group, retention_days, e)
594 )
595
596 try :
597 logs.create_log_stream( logGroupName = group, logStreamName = EVAL_RESULTS_LOG_STREAM )
598 except BotoCoreError as e:
599 return "could not persist the score for later querying: %s " % e, warning
600 except ClientError as e:
601 code = _err_code(e)
602 if code in _ACCESS_DENIED_CODES :
603 return _WRITEBACK_SKIPPED , warning
604 if code != "ResourceAlreadyExistsException" :
605 return "could not persist the score for later querying: %s " % e, warning
606 return None , warning
607
608
609 def _write_back (logs, service, records, ts_ms, retention_days = None , kms_key_id = None ):
610 """Persist scores to the evaluation-results log group. Best-effort: never raises."""
611 if not records:
612 return _WRITEBACK_SKIPPED , None
613 group = EVAL_RESULTS_LOG_GROUP_PREFIX + _log_group_segment(service)
614 events = [{ "timestamp" : ts_ms, "message" : json.dumps(r, default = str )} for r in records]
615 warning = None
616 for attempt in ( 1 , 2 ):
617 try :
618 logs.put_log_events(
619 logGroupName = group, logStreamName = EVAL_RESULTS_LOG_STREAM , logEvents = events
620 )
621 return None , warning
622 except ClientError as e:
623 code = _err_code(e)
624 if code in _ACCESS_DENIED_CODES :
625 return _WRITEBACK_SKIPPED , warning
626 if code == "ResourceNotFoundException" and attempt == 1 :
627 outcome, warning = _provision_group(logs, group, retention_days, kms_key_id)
628 if outcome is not None :
629 return outcome, warning
630 continue
631 return "could not persist the score for later querying: %s " % e, warning
632 except BotoCoreError as e:
633 return "could not persist the score for later querying: %s " % e, warning
634 return None , warning
635
636
637 def _fetch_trace (omni, tid, window_days, session_id = None , span_cache = None ):
638 """Resolve one trace's session and fetch its spans via Omni SQL ONCE.
639
640 Returns {traceId, sessionId, spans, truncated} on success, or {traceId, error} on failure.
641 span_cache de-duplicates by resolved session id: two trace ids in the same session fetch
642 that session's spans only once.
643 """
644 sid = session_id or _resolve_session_id(omni, tid, window_days)
645 if sid and not _SESSION_ID_RE .match(sid):
646 sid = None
647 cache_key = sid or ( "trace:" + tid)
648 if span_cache is not None and cache_key in span_cache:
649 spans, truncated = span_cache[cache_key]
650 else :
651 filter_clause = (
652 ( "attributes['session.id'] = ' %s '" % sid) if sid else ( "traceId = ' %s '" % tid)
653 )
654 spans, truncated, err = _fetch_spans(omni, filter_clause, window_days, soft = True )
655 if spans is None :
656 return { "traceId" : tid, "error" : err or "span fetch query did not complete" }
657 if not spans:
658 return { "traceId" : tid, "error" : "no spans found in the last %d day(s)" % window_days}
659 if span_cache is not None :
660 span_cache[cache_key] = (spans, truncated)
661 if sid and truncated and not any (s.get( "traceId" ) == tid for s in spans):
662 return {
663 "traceId" : tid,
664 "error" : "session has more than %d spans and the target trace was not in the fetched "
665 "set; narrow --window-days, or the session is too large to score whole"
666 % SPAN_ROW_LIMIT ,
667 }
668 return { "traceId" : tid, "sessionId" : sid, "spans" : spans, "truncated" : truncated}
669
670
671 def _score_pair (ac, tid, spans, evaluator_id, level, max_calls, include_explanations = False ):
672 """Score ONE (trace, evaluator) pair against the AgentCore data plane."""
673 evaluation_input = { "sessionSpans" : spans}
674 calls, note = [], None
675 if level == "SESSION" :
676 calls.append({ "evaluatorId" : evaluator_id, "evaluationInput" : evaluation_input})
677 elif level == "TOOL_CALL" :
678 span_ids = _tool_span_ids(spans)
679 if not span_ids:
680 return [], [ "TOOL_CALL requested but the trace has no tool spans" ], 0 , None
681 for i in range ( 0 , len (span_ids), MAX_TARGET_IDS ):
682 calls.append(
683 {
684 "evaluatorId" : evaluator_id,
685 "evaluationInput" : evaluation_input,
686 "evaluationTarget" : { "spanIds" : span_ids[i: i + MAX_TARGET_IDS ]},
687 }
688 )
689 if len (calls) > max_calls:
690 note = "tool-span scoring truncated to %d of %d chunks (evaluate-call budget)" % (
691 max ( 0 , max_calls),
692 len (calls),
693 )
694 else : # TRACE
695 calls.append(
696 {
697 "evaluatorId" : evaluator_id,
698 "evaluationInput" : evaluation_input,
699 "evaluationTarget" : { "traceIds" : [tid]},
700 }
701 )
702 calls = calls[: max ( 0 , max_calls)]
703
704 rows, errors = [], []
705 for kwargs in calls:
706 try :
707 resp = ac.evaluate( ** kwargs)
708 except (BotoCoreError, ClientError) as e:
709 msg = str (e)
710 if "Unknown evaluator" in msg and "provider" in msg:
711 msg += (
712 " — this evaluator is listed by list-evaluators/get-evaluator but not "
713 "invocable by evaluate (a service-side discovery/evaluate mismatch, not "
714 "a payload problem); pick a different evaluator"
715 )
716 errors.append( "evaluate failed: %s " % msg)
717 continue
718 resp.pop( "ResponseMetadata" , None )
719 rows.extend(_summarize(resp, evaluator_id, include_explanations))
720 return rows, errors, len (calls), note
721
722
723 MAX_TARGET_IDS = 10 # scoring service cap on ids per call (TOOL_CALL chunking)
724
725
726 def _run (args):
727 trace_ids = _split_ids(args.trace_ids, args.trace_id, lower = True )
728 if not trace_ids:
729 _die({ "error" : "no trace ids given (use --trace-id or --trace-ids)" })
730 for t in trace_ids:
731 if not _TRACE_ID_RE .match(t):
732 _die({ "error" : "invalid trace id (expected hex): %r " % t})
733 evaluator_ids = _split_ids(args.evaluator_ids, args.evaluator_id) or [ DEFAULT_EVALUATOR ]
734 if args.session_id:
735 if len (trace_ids) > 1 :
736 _die(
737 {
738 "error" : "--session-id applies to a single trace; omit it when scoring multiple traces"
739 }
740 )
741 if not _SESSION_ID_RE .match(args.session_id):
742 _die({ "error" : "invalid session id: %r " % args.session_id})
743 if args.retention_days and args.retention_days not in _RETENTION_DAYS_ALLOWED :
744 _die(
745 {
746 "error" : "invalid --retention-days %r ; CloudWatch Logs accepts one of %s (or 0 to "
747 "leave the account default)"
748 % (args.retention_days, sorted ( _RETENTION_DAYS_ALLOWED )),
749 }
750 )
751
752 notes = []
753 gt = [e for e in evaluator_ids if e in _GROUND_TRUTH_REQUIRED ]
754 if gt:
755 evaluator_ids = [e for e in evaluator_ids if e not in _GROUND_TRUTH_REQUIRED ]
756 msg = (
757 "evaluator(s) %s require ground truth (an expected tool trajectory), which a raw "
758 "trace does not carry — on-demand scoring here supplies no reference inputs. Use "
759 "ground-truth-free evaluators (e.g. Builtin.ToolSelectionAccuracy for tool quality), "
760 "or evaluate against a dataset whose examples carry the expected trajectory."
761 % ", " .join(gt)
762 )
763 if not evaluator_ids:
764 _die({ "error" : msg, "evaluatorIds" : gt})
765 notes.append( "skipped " + msg)
766
767 if len (evaluator_ids) > MAX_EVALUATORS :
768 dropped_evals = evaluator_ids[ MAX_EVALUATORS :]
769 evaluator_ids = evaluator_ids[: MAX_EVALUATORS ]
770 notes.append(
771 "capped to %d evaluators (≤ %d -evaluator limit); NOT run: %s "
772 % ( len (evaluator_ids), MAX_EVALUATORS , ", " .join(dropped_evals))
773 )
774 n_eval = len (evaluator_ids)
775 allowed_traces = min ( MAX_TRACES , max ( 1 , MAX_PAIRS // n_eval))
776 if len (trace_ids) > allowed_traces:
777 dropped = trace_ids[allowed_traces:]
778 trace_ids = trace_ids[:allowed_traces]
779 notes.append(
780 "capped to %d traces (≤ %d -trace / ≤ %d -pair limit at %d evaluators); NOT scored: %s "
781 % ( len (trace_ids), MAX_TRACES , MAX_PAIRS , n_eval, ", " .join(dropped))
782 )
783
784 session = boto3.Session( region_name = args.region)
785 try :
786 omni = session.client( "cloudwatchomni" ) # span queries via Omni SQL
787 except (
788 Exception
789 ) as e: # noqa: BLE001 — e.g. UnknownServiceError on a boto3 too old for this service
790 _die(
791 {
792 "error" : "could not create a cloudwatchomni client (is boto3 recent enough?): %s "
793 % e
794 }
795 )
796 logs = session.client( "logs" ) # writeback only (put_log_events)
797 try :
798 ac = session.client( EVALUATE_SERVICE )
799 except (
800 Exception
801 ) as e: # noqa: BLE001 — e.g. UnknownServiceError on a boto3 too old for this service
802 _die(
803 {
804 "error" : "could not create a %s client (is boto3 recent enough?): %s "
805 % ( EVALUATE_SERVICE , e)
806 }
807 )
808
809 fetched: dict = {}
810 fetch_errors: list = []
811 span_cache: dict = {}
812 for tid in trace_ids:
813 f = _fetch_trace(
814 omni,
815 tid,
816 args.window_days,
817 session_id = args.session_id if len (trace_ids) == 1 else None ,
818 span_cache = span_cache,
819 )
820 if f.get( "error" ):
821 fetch_errors.append({ "traceId" : tid, "error" : f[ "error" ]})
822 else :
823 fetched[tid] = f
824
825 results: list = []
826 score_errors: list = []
827 pairs_scored = 0
828 persisted_traces = 0
829 calls_budget = MAX_EVALUATE_CALLS
830 budget_hit = False
831 scored_sessions = set ()
832 ts_ms = int (time.time() * 1000 )
833 for tid in trace_ids:
834 if budget_hit:
835 break
836 f = fetched.get(tid)
837 if not f:
838 continue
839 trace_rows = []
840 for ev in evaluator_ids:
841 if args.level == "SESSION" and f[ "sessionId" ]:
842 skey = (f[ "sessionId" ], ev)
843 if skey in scored_sessions:
844 continue
845 scored_sessions.add(skey)
846 if calls_budget <= 0 :
847 notes.append(
848 "evaluate-call budget ( %d ) reached; remaining (trace, evaluator) pairs "
849 "not scored" % MAX_EVALUATE_CALLS
850 )
851 budget_hit = True
852 break
853 rows, errs, made, pair_note = _score_pair(
854 ac, tid, f[ "spans" ], ev, args.level, calls_budget, args.include_explanations
855 )
856 calls_budget -= made
857 for r in rows:
858 r[ "traceId" ] = tid
859 if any (r.get( "value" ) is not None and not r.get( "errorCode" ) for r in rows):
860 pairs_scored += 1
861 trace_rows.extend(rows)
862 for err in errs:
863 score_errors.append({ "traceId" : tid, "evaluatorId" : ev, "error" : err})
864 if pair_note:
865 notes.append( "trace %s / %s : %s " % (tid, ev, pair_note))
866 results.extend(trace_rows)
867
868 if trace_rows and not args.no_writeback:
869 service = _service_name(f[ "spans" ], tid) or _UNKNOWN_SERVICE
870 records = _eval_result_records(
871 trace_rows, tid, f[ "sessionId" ], args.level, service, ts_ms, partial = f[ "truncated" ]
872 )
873 if records:
874 outcome, warning = _write_back(
875 logs, service, records, ts_ms, args.retention_days, args.kms_key_id
876 )
877 if warning and warning not in notes:
878 notes.append(warning)
879 if outcome is None :
880 persisted_traces += 1
881 elif outcome is not _WRITEBACK_SKIPPED :
882 notes.append( "trace %s : %s " % (tid, outcome))
883 if f[ "truncated" ]:
884 notes.append(
885 "trace %s : span fetch hit the %d -row cap; scored on a partial span set"
886 % (tid, SPAN_ROW_LIMIT )
887 )
888
889 if not results:
890 _die(
891 {
892 "error" : "no (trace, evaluator) pair produced a score" ,
893 "level" : args.level,
894 "evaluators" : evaluator_ids,
895 "fetchErrors" : fetch_errors,
896 "scoreErrors" : score_errors,
897 }
898 )
899
900 scored_trace_ids = {
901 r[ "traceId" ] for r in results if r.get( "value" ) is not None and not r.get( "errorCode" )
902 }
903 receipt = {
904 "level" : args.level,
905 "evaluators" : evaluator_ids,
906 "tracesFetched" : len (fetched),
907 "tracesScored" : len (scored_trace_ids),
908 "pairsScored" : pairs_scored,
909 "results" : _public_rows(results),
910 }
911 if not args.no_writeback:
912 receipt[ "persistedTraces" ] = persisted_traces
913 if persisted_traces:
914 receipt[ "resultsLogGroupPrefix" ] = EVAL_RESULTS_LOG_GROUP_PREFIX
915 receipt[ "resultsRetentionDaysOnCreate" ] = args.retention_days or "account default"
916 receipt[ "resultsKmsKeyOnCreate" ] = args.kms_key_id or "aws-owned"
917 if notes:
918 receipt[ "notes" ] = notes
919 if fetch_errors:
920 receipt[ "fetchErrors" ] = fetch_errors
921 if score_errors:
922 receipt[ "scoreErrors" ] = score_errors
923 print (json.dumps(receipt, default = str , indent = 2 ))
924
925
926 def main ():
927 ap = argparse.ArgumentParser(
928 description = "Score CloudWatch Omni traces on demand with AgentCore evaluators."
929 )
930 ap.add_argument(
931 "--trace-id" , help = "a single trace id to score (or use --trace-ids for a batch)"
932 )
933 ap.add_argument( "--trace-ids" , help = "comma-separated trace ids to score (each fetched once)" )
934 ap.add_argument(
935 "--session-id" ,
936 help = "the trace's session id (single-trace only; resolved automatically if omitted)" ,
937 )
938 ap.add_argument( "--evaluator-id" , help = "a single evaluator id, e.g. Builtin.Helpfulness" )
939 ap.add_argument(
940 "--evaluator-ids" ,
941 help = "comma-separated evaluator ids; every one is run against every trace" ,
942 )
943 ap.add_argument( "--level" , choices = LEVELS , default = "TRACE" )
944 ap.add_argument( "--region" , default = "us-east-1" )
945 ap.add_argument( "--window-days" , type = int , default = 30 , help = "trace lookback window" )
946 ap.add_argument(
947 "--include-explanations" ,
948 action = "store_true" ,
949 help = "return the evaluator's explanation text in the receipt. Off by default because it "
950 "quotes the agent's own inputs and outputs, and the receipt reaches the calling agent's "
951 "context and transcript; without it each row reports hasExplanation instead. The "
952 "persisted record carries the explanation either way" ,
953 )
954 ap.add_argument(
955 "--no-writeback" ,
956 action = "store_true" ,
957 help = "do not persist scores to the evaluation-results log group (persisted by default "
958 "so they can be queried later)" ,
959 )
960 ap.add_argument(
961 "--retention-days" ,
962 type = int ,
963 default = DEFAULT_RESULTS_RETENTION_DAYS ,
964 help = "retention for the evaluation-results log group when this run CREATES it, bounding "
965 "how long the persisted explanations live (default %d , 0 leaves the account default; a "
966 "pre-existing group keeps its own)" % DEFAULT_RESULTS_RETENTION_DAYS ,
967 )
968 ap.add_argument(
969 "--kms-key-id" ,
970 help = "customer-managed KMS key (arn or id) to encrypt the evaluation-results log group "
971 "with when this run CREATES it; its key policy must allow the CloudWatch Logs service "
972 "principal (default: AWS-owned key)" ,
973 )
974 args = ap.parse_args()
975 try :
976 _run(args)
977 except Exception as e: # noqa: BLE001
978 _die({ "error" : "unexpected error: %s " % e})
979
980
981 if __name__ == "__main__" :
982 main()