Incremental Usage Reporting for Streaming APIs

Hello! Welcome back to our module on LLM Token Economics and Billing Systems.

In our previous lesson, we built a robust, idempotent API for handling single, non-streaming usage reports. By using an Idempotency-Key, we ensured that network retries wouldn't lead to duplicate charges—a critical feature for any billing system.

However, the world of LLMs is increasingly moving towards streaming responses to improve the user experience. Instead of waiting seconds for a complete answer, users see text appear token by token. This poses a new challenge for our billing system: usage data no longer arrives in a single, neat package. It arrives as a flow of incremental updates.

Today's lesson tackles this head-on. Our learning outcome is to adapt the API design to support incremental usage reporting for streaming responses. We will explore how to reliably capture usage from a stream of events without sacrificing accuracy or performance.

Given your work on Bedrock's infrastructure and the Rudi scheduler, you're deeply familiar with handling continuous data flows and ensuring system reliability. We'll apply those same principles to design a resilient billing pipeline for streaming LLM services.

1. The Mechanics of Streaming LLM Responses

Before we can bill for a stream, we must understand how it works. In a non-streaming API call, you make a request and wait for the entire response to be generated before you receive it. With streaming, the server starts sending back data as soon as the first pieces are ready.

This is typically achieved using technologies like Server-Sent Events (SSE), where the server holds a connection open and pushes a series of small data chunks to the client.

To see this in action, let's watch a short video that demonstrates both the user-facing benefit and the core backend implementation difference.

As the video shows, the key change is that the server iterates through a generator of response "chunks" and yields them, rather than returning a single object. Our billing system must now contend with this sequence of events.

2. How Usage is Reported in a Stream

Fortunately, we don't have to count tokens ourselves from the text chunks. Modern LLM APIs, like Anthropic's, embed usage information directly within the event stream. This is a crucial design detail.

Let's examine the official Anthropic documentation to see exactly how this is structured.

The fact that the usage field is cumulative is a game-changer. It means each message_delta event gives us the total output tokens generated so far for that stream. We don't have to sum up deltas on the client side, which greatly simplifies our logic and reduces the risk of miscalculation.

The stream gives us a sequence of state snapshots:

• message_start: Contains the initial input_tokens.
• message_delta: Periodically provides a cumulative count of output_tokens.
• message_stop: Signals the end of the stream.

This is visually represented in the API's class diagram, where the Usage object is a core part of the response structure.

3. Architectural Design: An Asynchronous, Decoupled Pipeline

Our API service that calls the LLM should not be blocked waiting for the billing system. Just as we discussed for payment systems, critical operations like billing should be handled asynchronously and off the main request path.

When dealing with a stream of usage events, this becomes even more important. We don't want to add network latency inside the tight loop that's processing response chunks. The best practice is to "fire-and-forget" billing events to a durable message queue.

This leads to a simple and robust architecture:

1. API Service: The service making the call to the Claude API. It processes the stream and emits usage events.
2. Message Queue: A durable queue (e.g., AWS SQS, Kafka, Google Pub/Sub) that receives the events.
3. Billing Ingestion Service: A separate service that consumes events from the queue, deduplicates them, and aggregates them into a database.

This decoupling ensures that even if the billing service is slow or down, the API service remains unaffected, and no billing data is lost (as it's buffered in the queue).

4. Adapting the API for Incremental Reports

Our previous POST /v1/usage-events endpoint was designed for a single event. We now need a way to submit a sequence of related events. We'll design a new schema for these incremental reports.

4.1. The Incremental Event Schema

An incremental event needs to contain enough information to be uniquely identified and correctly ordered.

from pydantic import BaseModel, Field
from datetime import datetime
import uuid

class IncrementalUsageEvent(BaseModel):



    # The unique ID for the entire streaming session.
    # Could be the 'message.id' from the LLM API response.
    stream_id: uuid.UUID
    



    # A monotonically increasing number for each report in this stream.
    sequence_number: int = Field(gt=0)




    # Tenant/User identifiers
    tenant_id: str
    user_id: str
    



    # Cumulative token counts, reflecting the latest state.
    cumulative_input_tokens: int
    cumulative_output_tokens: int
    



    # Timestamp of when this event was generated.
    timestamp: datetime
    



    # Flag to signal the final event of the stream.
    is_final: bool = False

4.2. Idempotency for Streams

In our last lesson, the Idempotency-Key was tied to a single logical operation. Here, a single logical operation (one LLM query) produces many usage reports. We can't reuse the same key for all of them.

The solution is a composite idempotency key. The unique identifier for an incremental report is the combination of (stream_id, sequence_number). This allows the client to safely retry sending report number N for a given stream without fear of it being double-counted.

Your ingestion service's logic would be:

1. Receive an IncrementalUsageEvent.
2. Check a database (like DynamoDB) for a record with the key (stream_id, sequence_number).
3. If it exists, the event is a duplicate. Discard it.
4. If it doesn't exist, process it and store the key to prevent future duplicates.

This pattern ensures "exactly-once" processing for each individual report within the stream.

5. Client-Side Reporting Strategy

The service calling the LLM now has a choice: when should it send an incremental report?

• Strategy A: Report on Every Update

  • Send an IncrementalUsageEvent for every message_delta received from the LLM API.
  • Pros: Real-time, highly accurate data.
  • Cons: Very high network overhead. Might generate hundreds of events for a single query.

• Strategy B: Report on Final Update Only

  • Wait for the message_stop event and send a single report with the final cumulative usage.
  • Pros: Minimal overhead.
  • Cons: Not real-time. If the service crashes before the stream finishes, all usage data is lost. This is unacceptable for long-running streams.

• Strategy C: Throttled "Heartbeat" Reports (Recommended)

  • Send an IncrementalUsageEvent periodically (e.g., every 5 seconds or every 500 new tokens).
  • Crucially, always send a final report upon receiving the message_stop event.
  • Pros: Balances accuracy and overhead. Provides good real-time visibility and dramatically reduces data loss risk in case of a crash.
  • Cons: Slightly more complex client logic.

The "heartbeat" approach is the most production-ready. Let's sketch out what this client logic would look like.

import time
import uuid
from datetime import datetime, timezone




# Assume `anthropic_client` is an initialized Anthropic SDK client
# Assume `billing_client.report_usage(event)` sends the event asynchronously

def process_streaming_query(prompt: str):
    stream_id = None
    sequence_number = 0
    last_report_time = 0
    



    # Use a real message queue in production!
    # This is a placeholder for fire-and-forget logic.

    with anthropic_client.messages.stream(
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}],
        model="claude-3-5-sonnet-20240620",
    ) as stream:
        for event in stream:



            # Capture the stream_id on the first event
            if event.type == "message_start":
                stream_id = uuid.UUID(event.message.id)
                



            # Process content for the user
            if event.type == "content_block_delta":
                print(event.delta.text, end="", flush=True)




            # Check for a usage update
            if event.type == "message_delta" and stream_id:
                current_time = time.time()



                # Throttle: send report every 5 seconds
                if current_time - last_report_time > 5:
                    sequence_number += 1
                    usage_event = IncrementalUsageEvent(
                        stream_id=stream_id,
                        sequence_number=sequence_number,



                        # ... tenant_id, user_id from context ...
                        cumulative_input_tokens=stream.get_final_message().usage.input_tokens,
                        cumulative_output_tokens=event.usage.output_tokens,
                        timestamp=datetime.now(timezone.utc),
                        is_final=False
                    )



                    # billing_client.report_usage(usage_event)
                    print(f"\n[Sent heartbeat report #{sequence_number}]")
                    last_report_time = current_time




    # After the stream is fully processed, send the final, definitive report
    if stream_id:
        final_message = stream.get_final_message()
        sequence_number += 1
        final_event = IncrementalUsageEvent(
            stream_id=stream_id,
            sequence_number=sequence_number,



            # ... tenant_id, user_id ...
            cumulative_input_tokens=final_message.usage.input_tokens,
            cumulative_output_tokens=final_message.usage.output_tokens,
            timestamp=datetime.now(timezone.utc),
            is_final=True
        )



        # billing_client.report_usage(final_event)
        print(f"\n[Sent FINAL report #{sequence_number}]")

Note: The code uses stream.get_final_message() to access the input_tokens. This helper accumulates the stream internally. A more optimized approach might cache the input token count from the message_start event.

Conclusion

We have successfully adapted our billing architecture to handle the complexities of streaming responses. By understanding the event-driven nature of LLM streams and leveraging the cumulative usage data they provide, we designed a robust and scalable system.

Key Takeaways:

• Streaming APIs use event streams (SSE) to send data in chunks, including cumulative usage information.
• Billing for streams should be asynchronous, using a message queue to decouple the application from the billing service.
• A new event schema is needed for incremental reports, including a stream_id, sequence_number, and an is_final flag.
• Idempotency for streams is achieved with a composite key (stream_id, sequence_number), ensuring each partial report is processed exactly once.
• A "heartbeat" reporting strategy (throttled updates + a guaranteed final report) provides the best trade-off between real-time accuracy and system overhead.

Next Lesson Preview

We have now designed a reliable pipeline to capture both single and incremental usage events. But where does this data go? Raw event logs are not suitable for fast, user-facing queries like "How much have I spent this month?". In our next lesson, we will design a data model for storing aggregated token counts that supports efficient querying by user, model, and time.