Scenario A: AI agent observability¶

Part of Streaming Transformations › Scenarios.

Goal¶

An AI agent works in multi-step runs: it calls a model, then one or more tools. Each run sends one telemetry record with token counts, latency, and the list of tool calls it made. To see what agents do and what they cost, teams need one row per tool call, with the model provider and an estimated cost, and without health checks and internal evaluation runs.

Transformations used¶

1. Lateral Flatten on the stream: expand the array of tool calls in each agent run, before the dimension join (derived table).
2. Stream enrichment: map each model to its provider and token prices.
3. Stream filtering on stream columns: drop health-check and evaluation runs.

Source data¶

Stream¶

Each streamed record is one agent run. The example uses these five runs:

{"ts": "2026-10-09 11:00:00", "agent": "support-bot", "trace_id": "t-1001", "model": "claude-sonnet", "input_tokens": 1200, "output_tokens": 340, "latency_ms": 2150, "status": "OK", "tool_calls": [{"name": "search_kb", "ms": 420}, {"name": "create_ticket", "ms": 180}]}
{"ts": "2026-10-09 11:00:05", "agent": "code-review-bot", "trace_id": "t-1002", "model": "claude-haiku", "input_tokens": 3000, "output_tokens": 800, "latency_ms": 4100, "status": "OK", "tool_calls": [{"name": "read_file", "ms": 90}, {"name": "run_tests", "ms": 2300}, {"name": "post_comment", "ms": 150}]}
{"ts": "2026-10-09 11:00:10", "agent": "support-bot", "trace_id": "t-1003", "model": "claude-sonnet", "input_tokens": 50, "output_tokens": 5, "latency_ms": 120, "status": "HEALTHCHECK", "tool_calls": [{"name": "ping", "ms": 5}]}
{"ts": "2026-10-09 11:00:15", "agent": "research-bot", "trace_id": "t-1004", "model": "custom-llm", "input_tokens": 2000, "output_tokens": 500, "latency_ms": 3300, "status": "OK", "tool_calls": [{"name": "web_search", "ms": 900}, {"name": "summarize", "ms": 700}]}
{"ts": "2026-10-09 11:00:20", "agent": "support-bot", "trace_id": "t-1005", "model": "claude-sonnet", "input_tokens": 800, "output_tokens": 200, "latency_ms": 1500, "status": "EVAL", "tool_calls": [{"name": "search_kb", "ms": 300}]}

Dimension table¶

The PIPE joins this dimension table, refreshed periodically:

CREATE OR REPLACE TABLE dim_models (
  model STRING PRIMARY KEY RELY,
  provider STRING,
  input_price_per_1k FLOAT,
  output_price_per_1k FLOAT);

INSERT INTO dim_models VALUES
  ('claude-sonnet', 'anthropic', 0.003, 0.015),
  ('claude-haiku', 'anthropic', 0.0008, 0.004);

Target table¶

The PIPE lands transformed and filtered rows in this target table, with one row per tool call:

CREATE OR REPLACE TABLE agent_tool_calls (
  event_ts TIMESTAMP_NTZ, agent STRING, trace_id STRING, model STRING,
  provider STRING, tool_name STRING, tool_ms NUMBER, estimated_cost_usd FLOAT);

Transformation¶

The PIPE flattens streamed records, enriches by joining the dimension table, and filters rows before landing by applying a WHERE on a stream column:

CREATE OR REPLACE PIPE agent_tool_calls_pipe AS
  COPY INTO agent_tool_calls (event_ts, agent, trace_id, model, provider, tool_name, tool_ms, estimated_cost_usd) FROM (
    SELECT s.event_ts, s.agent, s.trace_id, s.model, m.provider, s.tool_name, s.tool_ms,
           (s.input_tokens * m.input_price_per_1k + s.output_tokens * m.output_price_per_1k) / 1000
    FROM (
      SELECT
        e.$1:ts::TIMESTAMP_NTZ     AS event_ts,
        e.$1:agent::STRING         AS agent,
        e.$1:trace_id::STRING      AS trace_id,
        e.$1:model::STRING         AS model,
        e.$1:input_tokens::NUMBER  AS input_tokens,
        e.$1:output_tokens::NUMBER AS output_tokens,
        e.$1:status::STRING        AS status,
        t.value:name::STRING       AS tool_name,
        t.value:ms::NUMBER         AS tool_ms
      FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING')) e,
           LATERAL FLATTEN(input => e.$1:tool_calls) t) s
    LEFT OUTER JOIN dim_models m
      ON s.model = m.model
    WHERE s.status NOT IN ('HEALTHCHECK', 'EVAL'));

Result¶

Querying the target table returns rows like these:

The health-check run t-1003 and the evaluation run t-1005 are filtered out. Each tool call gets its own row, and the estimated cost of the run repeats on every row of the trace. The model custom-llm has no entry in dim_models, so its provider and cost are NULL.

Considerations¶

1. Dimension data is a snapshot that refreshes periodically, so updates can take several minutes to appear. This works well for slow-moving reference data. Dynamic tables can be used as dimension tables.
2. A flatten before the dimension join must be wrapped in a derived table.
3. ERROR_LOGGING currently can’t be enabled on the target table of a pipe that uses LATERAL FLATTEN.

See also¶

See Stream filtering, Stream enrichment, and Lateral Flatten for more information on the transformations used in this scenario.

Previous: Scenarios · Next: Scenario B: Observability logs and OpenTelemetry