stormlog.infer.events

Inference profiling event records and JSONL persistence.

Classes

InferenceRequestEvent(session_id, ...[, ...])

Client-observed request trace for one chat completion.

InferenceSummaryEvent(session_id, ...)

Aggregate summary emitted at the end of a profiling run.

InferenceSystemSample(session_id, ...[, ...])

Best-effort system telemetry sampled during an inference profiling run.

JsonlEventWriter(path)

Append inference profiling records to a JSONL artifact.
class stormlog.infer.events.InferenceRequestEvent(session_id, request_id, case_id, phase, started_at_ns, ended_at_ns, endpoint, model, concurrency, target_input_tokens, target_output_tokens, stream, status, e2e_latency_ms, ttft_ms, first_chunk_latency_ms, chunk_interarrival_ms=<factory>, prompt_tokens=None, prompt_token_source='unknown', prompt_token_exact=False, output_tokens=None, output_token_source='unknown', output_token_exact=False, total_tokens=None, finish_reason=None, error_type=None, error_message=None)[source]

Bases: object

Client-observed request trace for one chat completion.

Parameters:

  • session_id (str)

  • request_id (str)

  • case_id (str)

  • phase (str)

  • started_at_ns (int)

  • ended_at_ns (int)

  • endpoint (str)

  • model (str)

  • concurrency (int)

  • target_input_tokens (int)

  • target_output_tokens (int)

  • stream (bool)

  • status (str)

  • e2e_latency_ms (float | None)

  • ttft_ms (float | None)

  • first_chunk_latency_ms (float | None)

  • chunk_interarrival_ms (list[float])

  • prompt_tokens (int | None)

  • prompt_token_source (str)

  • prompt_token_exact (bool)

  • output_tokens (int | None)

  • output_token_source (str)

  • output_token_exact (bool)

  • total_tokens (int | None)

  • finish_reason (str | None)

  • error_type (str | None)

  • error_message (str | None)

session_id: str
request_id: str
case_id: str
phase: str
started_at_ns: int
ended_at_ns: int
endpoint: str
model: str
concurrency: int
target_input_tokens: int
target_output_tokens: int
stream: bool
status: str
e2e_latency_ms: float | None
ttft_ms: float | None
first_chunk_latency_ms: float | None
chunk_interarrival_ms: list[float]
prompt_tokens: int | None = None
prompt_token_source: str = 'unknown'
prompt_token_exact: bool = False
output_tokens: int | None = None
output_token_source: str = 'unknown'
output_token_exact: bool = False
total_tokens: int | None = None
finish_reason: str | None = None
error_type: str | None = None
error_message: str | None = None
to_record()[source]
Return type:

dict[str, Any]

class stormlog.infer.events.InferenceSystemSample(session_id, timestamp_ns, sampler, device_id=None, device_name=None, device_used_bytes=None, device_free_bytes=None, device_total_bytes=None, gpu_utilization_percent=None, process_rss_bytes=None, metadata=<factory>)[source]

Bases: object

Best-effort system telemetry sampled during an inference profiling run.

Parameters:

  • session_id (str)

  • timestamp_ns (int)

  • sampler (str)

  • device_id (int | None)

  • device_name (str | None)

  • device_used_bytes (int | None)

  • device_free_bytes (int | None)

  • device_total_bytes (int | None)

  • gpu_utilization_percent (float | None)

  • process_rss_bytes (int | None)

  • metadata (dict[str, Any])

session_id: str
timestamp_ns: int
sampler: str
device_id: int | None = None
device_name: str | None = None
device_used_bytes: int | None = None
device_free_bytes: int | None = None
device_total_bytes: int | None = None
gpu_utilization_percent: float | None = None
process_rss_bytes: int | None = None
metadata: dict[str, Any]
to_record()[source]
Return type:

dict[str, Any]

class stormlog.infer.events.InferenceSummaryEvent(session_id, timestamp_ns, summary)[source]

Bases: object

Aggregate summary emitted at the end of a profiling run.

Parameters:

  • session_id (str)

  • timestamp_ns (int)

  • summary (dict[str, Any])

session_id: str
timestamp_ns: int
summary: dict[str, Any]
to_record()[source]
Return type:

dict[str, Any]

class stormlog.infer.events.JsonlEventWriter(path)[source]

Bases: object

Append inference profiling records to a JSONL artifact.

Parameters:

path (str | Path)

append(record)[source]
Parameters:

record (dict[str, Any])

Return type:

None