Analytics overview

​

Mental model

• Everything that runs on VectorShift emits events (the wire format calls them spans). An event is one Pipeline run, Chatbot message, KB fetch, Form submission, Session message, etc. — each carrying status, latency, token counts, costs, and a trace_id that links it to siblings.
• There’s one entry point: Analytics.query(...) returns an immutable, chainable Query. Scope it by object and event kind, refine it with .where(...) and .group_by(...), then call a terminal (.count(), .sum(...), .events(), .table(...), .export(...)) to make the HTTP call and get a typed result.
• Time scoping is a .where() predicate — q.where(EventField.EVENT_START_TIME > cutoff). There is no since= / start= / end= shortcut. Datetimes must be timezone-aware. Aggregations (sum, mean, percentile, raw_aggregate, …) require a lower time bound, or they raise AnalyticsInvalidQuery.
• Scope is a single object. object= / object_ids= take one resource or one id. To analyse several, query each and merge client-side — passing more than one raises AnalyticsInvalidQuery.
• Single-object lookups live outside the chain: Analytics.trace(trace_id), Analytics.event(event_id), and Analytics.run(object=…, run_id=…).
• Every terminal has an async sibling (acount, aevents, asum, atable, aexport, …), and Analytics.aquery(...) returns a Query whose terminals route to them.

​

Quick start

from datetime import datetime, timedelta, timezone

from vectorshift import Analytics
from vectorshift.analytics import EventField

# Scope: org-wide, last 7 days (time bound via .where()).
q = Analytics.query().where(
    EventField.EVENT_START_TIME > datetime.now(timezone.utc) - timedelta(days=7)
)

# Terminals make the call and return typed results.
print(f"runs:         {q.count()['count']}")
print(f"total tokens: {q.sum('tokens')}")
print(f"mean latency: {q.mean('latency'):.2f} ms")
print(f"failures:     {q.where(status='failure').count()['count']}")

import asyncio
from datetime import datetime, timedelta, timezone

from vectorshift import Analytics
from vectorshift.analytics import EventField

async def main():
    q = Analytics.aquery().where(
        EventField.EVENT_START_TIME > datetime.now(timezone.utc) - timedelta(days=7)
    )

    # Fan out a dashboard concurrently.
    runs, failures, latency = await asyncio.gather(
        q.acount(),
        q.where(status="failure").acount(),
        q.amean("latency"),
    )
    print(f"runs:         {runs['count']}")
    print(f"failures:     {failures['count']}")
    print(f"mean latency: {latency:.2f} ms")

asyncio.run(main())

​

Scoping a query

​

Terminals

​

Filtering

• Equality kwargs — q.where(status="failure", interface_type="chatbot"). Shortest for the common case.
• Operator overloading on EventField — q.where(EventField.EVENT_START_TIME > cutoff, EventField.ERROR_MESSAGE.matches("rate limit")). SQLAlchemy / Peewee style.
• Dataclass helpers — Filter (scope fields), FieldFilter (data columns), and FilterGroup for OR groups via logical_op=LogicalOp.OR.

​

What’s next