📓 Evaluate Streaming Apps¶

This notebook shows how to trace and evaluate an app with streaming generation.

It also shows the use of the dummy feedback function provider which behaves like the HuggingFace provider except it does not actually perform any network calls and just produces constant results. It can be used to prototype feedback function wiring for your apps before invoking potentially slow (to run/to load) feedback functions.

Import libraries¶

In [ ]:

# !pip install trulens trulens-providers-huggingface

# !pip install trulens trulens-providers-huggingface

Set keys¶

In [ ]:

import os

if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = "sk-proj-..."

import os if "OPENAI_API_KEY" not in os.environ: os.environ["OPENAI_API_KEY"] = "sk-proj-..."

In [ ]:

from trulens.core import Metric
from trulens.core import Selector
from trulens.core import TruSession

session = TruSession()
session.reset_database()

from trulens.core import Metric from trulens.core import Selector from trulens.core import TruSession session = TruSession() session.reset_database()

In [ ]:

from trulens.dashboard import run_dashboard

run_dashboard(session)

from trulens.dashboard import run_dashboard run_dashboard(session)

Build the app¶

In [ ]:

from openai import OpenAI
from trulens.apps.app import instrument

oai_client = OpenAI()


class APP:
    @instrument
    def stream_completion(self, prompt):
        completion = oai_client.chat.completions.create(
            model="gpt-4.1-mini",
            stream=True,
            # Asks OpenAI for a final usage chunk. Without it a streamed
            # response carries no token counts, so cost and tokens/sec cannot
            # be reported.
            stream_options={"include_usage": True},
            temperature=0,
            messages=[
                {
                    "role": "user",
                    "content": f"Please answer the question: {prompt}",
                }
            ],
        )
        for chunk in completion:
            if (
                len(choices := chunk.choices) > 0
                and (content := choices[0].delta.content) is not None
            ):
                yield content


llm_app = APP()

from openai import OpenAI from trulens.apps.app import instrument oai_client = OpenAI() class APP: @instrument def stream_completion(self, prompt): completion = oai_client.chat.completions.create( model="gpt-4.1-mini", stream=True, # Asks OpenAI for a final usage chunk. Without it a streamed # response carries no token counts, so cost and tokens/sec cannot # be reported. stream_options={"include_usage": True}, temperature=0, messages=[ { "role": "user", "content": f"Please answer the question: {prompt}", } ], ) for chunk in completion: if ( len(choices := chunk.choices) > 0 and (content := choices[0].delta.content) is not None ): yield content llm_app = APP()

Create dummy feedback¶

By setting the provider as Dummy(), you can erect your evaluation suite and then easily substitute in a real model provider (e.g. OpenAI) later.

In [ ]:

from trulens.providers.huggingface.provider import Dummy

hugs = Dummy()

f_positive_sentiment = Metric(
    implementation=hugs.positive_sentiment,
    selectors={"response": Selector.select_record_output()},
)

from trulens.providers.huggingface.provider import Dummy hugs = Dummy() f_positive_sentiment = Metric( implementation=hugs.positive_sentiment, selectors={"response": Selector.select_record_output()}, )

Create the app¶

In [ ]:

# add trulens as a context manager for llm_app with dummy feedback
from trulens.apps.app import TruApp

tru_app = TruApp(
    llm_app,
    app_name="LLM App",
    app_version="v1",
    feedbacks=[f_positive_sentiment],
)

# add trulens as a context manager for llm_app with dummy feedback from trulens.apps.app import TruApp tru_app = TruApp( llm_app, app_name="LLM App", app_version="v1", feedbacks=[f_positive_sentiment], )

Run the app¶

In [ ]:

with tru_app as recording:
    for chunk in llm_app.stream_completion(
        "give me a good name for a colorful sock company and the store behind its founding"
    ):
        print(chunk, end="")

record = recording.get()

with tru_app as recording: for chunk in llm_app.stream_completion( "give me a good name for a colorful sock company and the store behind its founding" ): print(chunk, end="") record = recording.get()

In [ ]:

# Check full output:

record.main_output

# Check full output: record.main_output

Inspect the streaming metrics¶

Streaming a response makes the total latency a poor summary on its own: most of it is time spent streaming, not time the user spent waiting to see anything. TruLens records the split as span attributes, so you can see how long the first chunk took separately from how fast the rest arrived.

In [ ]:

import json

from trulens.otel.semconv.trace import SpanAttributes

events = session.get_events(
    app_name="LLM App", app_version="v1", record_ids=[record.record_id]
)

reported = [
    SpanAttributes.GENERATION.IS_STREAMING,
    SpanAttributes.GENERATION.TIME_TO_FIRST_TOKEN_MS,
    SpanAttributes.GENERATION.TOKENS_PER_SECOND,
    SpanAttributes.GENERATION.CHUNKS_RECEIVED,
    SpanAttributes.COST.NUM_COMPLETION_TOKENS,
    SpanAttributes.COST.COST,
]

for record_attributes in events["record_attributes"]:
    if isinstance(record_attributes, str):
        record_attributes = json.loads(record_attributes)

    if not record_attributes.get(SpanAttributes.GENERATION.IS_STREAMING):
        continue

    for name in reported:
        print(f"{name}: {record_attributes.get(name)}")

import json from trulens.otel.semconv.trace import SpanAttributes events = session.get_events( app_name="LLM App", app_version="v1", record_ids=[record.record_id] ) reported = [ SpanAttributes.GENERATION.IS_STREAMING, SpanAttributes.GENERATION.TIME_TO_FIRST_TOKEN_MS, SpanAttributes.GENERATION.TOKENS_PER_SECOND, SpanAttributes.GENERATION.CHUNKS_RECEIVED, SpanAttributes.COST.NUM_COMPLETION_TOKENS, SpanAttributes.COST.COST, ] for record_attributes in events["record_attributes"]: if isinstance(record_attributes, str): record_attributes = json.loads(record_attributes) if not record_attributes.get(SpanAttributes.GENERATION.IS_STREAMING): continue for name in reported: print(f"{name}: {record_attributes.get(name)}")

In [ ]:

session.get_leaderboard(app_ids=[tru_app.app_id])

session.get_leaderboard(app_ids=[tru_app.app_id])