Metrics streaming

Overview

This example pairs two ways of turning the same Polyglot streaming events into metrics, run against one shared EventDispatcher:

• a hand-rolled StreamMetricsCollector (Cognesy\Metrics\Collectors\MetricsCollector) that listens for streaming-specific events and builds derived metrics no runtime package knows about: time to first chunk, chunk count, tokens/second.
• the canonical Telemetry path — a Cognesy\Telemetry\Application\Telemetry hub wired to Cognesy\Polyglot\Telemetry\PolyglotTelemetryProjector through RuntimeEventBridge — which turns the very same InferenceStarted / InferenceAttemptSucceeded / InferenceUsageReported / InferenceCompleted events into spans, logs, and the catalog inference.client.* metrics (token usage histograms, attempt/operation counters and timers) automatically, with zero extra collector code.

The output below is grouped so the two halves stay visibly distinct: metrics from the hand-rolled collector, and metrics + spans emitted by the Telemetry projector for free from the same run.

Example

<?php
require 'examples/boot.php';

use Cognesy\Events\Dispatchers\EventDispatcher;
use Cognesy\Messages\Messages;
use Cognesy\Metrics\Collectors\MetricsCollector;
use Cognesy\Metrics\Contracts\CanExportMetrics;
use Cognesy\Metrics\Data\Metric;
use Cognesy\Metrics\Exporters\CallbackExporter;
use Cognesy\Metrics\Metrics;
use Cognesy\Polyglot\Inference\Events\InferenceCompleted;
use Cognesy\Polyglot\Inference\Events\PartialInferenceDeltaCreated;
use Cognesy\Polyglot\Inference\Events\StreamFirstChunkReceived;
use Cognesy\Polyglot\Inference\Inference;
use Cognesy\Polyglot\Inference\InferenceRuntime;
use Cognesy\Polyglot\Inference\Config\LLMConfig;
use Cognesy\Polyglot\Telemetry\PolyglotTelemetryProjector;
use Cognesy\Telemetry\Application\Projector\RuntimeEventBridge;
use Cognesy\Telemetry\Application\Registry\TraceRegistry;
use Cognesy\Telemetry\Application\Telemetry;
use Cognesy\Telemetry\Domain\Contract\CanExportObservations;
use Cognesy\Telemetry\Domain\Observation\Observation;

final class StreamMetricsCollector extends MetricsCollector
{
    private int $chunkCount = 0;

    protected function listeners(): array {
        return [
            StreamFirstChunkReceived::class => 'onFirstChunk',
            PartialInferenceDeltaCreated::class => 'onChunk',
            InferenceCompleted::class => 'onCompleted',
        ];
    }

    public function onFirstChunk(StreamFirstChunkReceived $event): void {
        $this->timer('llm.stream.ttfc_ms', $event->timeToFirstChunkMs, [
            'model' => $this->modelTag($event->model),
        ]);
    }

    public function onChunk(PartialInferenceDeltaCreated $event): void {
        $this->chunkCount += 1;
    }

    public function onCompleted(InferenceCompleted $event): void {
        $durationMs = $event->data['durationMs'] ?? 1;
        $durationSeconds = max(0.001, $durationMs / 1000);
        $outputTokens = $event->data['outputTokens'] ?? 0;
        $tokensPerSecond = $outputTokens / $durationSeconds;

        $this->timer('llm.stream.duration_ms', $durationMs);
        $this->gauge('llm.stream.chunk_count', (float) $this->chunkCount);
        $this->gauge('llm.stream.output_tokens', (float) $outputTokens);
        $this->gauge('llm.stream.output_tokens_per_second', $tokensPerSecond);

        $this->chunkCount = 0;
    }

    private function modelTag(?string $model): string {
        if ($model !== null && $model !== '') {
            return $model;
        }
        return 'default';
    }
}

/**
 * In-memory sink for the Telemetry side: no network, no credentials — just
 * keeps whatever spans/logs and metrics the projector hands it, so the
 * example can print them. A real app would pass an OTel/Langfuse/Logfire
 * exporter here instead; the projector wiring is identical either way.
 */
final class InMemoryTelemetryExporter implements CanExportObservations, CanExportMetrics
{
    /** @var list<Observation> */
    private array $observations = [];
    /** @var list<Metric> */
    private array $metrics = [];

    public function exportObservation(Observation $observation): void {
        $this->observations[] = $observation;
    }

    /** @param iterable<Metric> $metrics */
    public function export(iterable $metrics): void {
        foreach ($metrics as $metric) {
            $this->metrics[] = $metric;
        }
    }

    /** @return list<Observation> */
    public function observations(): array {
        return $this->observations;
    }

    /** @return list<Metric> */
    public function metrics(): array {
        return $this->metrics;
    }
}

$events = new EventDispatcher();
$metrics = new Metrics($events);
$metrics->collect(new StreamMetricsCollector());

// Telemetry side: same $events dispatcher, canonical projector. This turns
// InferenceStarted/InferenceAttemptSucceeded/InferenceUsageReported/InferenceCompleted
// into spans, logs and inference.client.* metrics with no extra collector code.
$telemetryExporter = new InMemoryTelemetryExporter();
$telemetry = new Telemetry(new TraceRegistry(), $telemetryExporter);
(new RuntimeEventBridge(new PolyglotTelemetryProjector($telemetry)))->attachTo($events);

$prompt = 'In one sentence, explain why streaming responses help UX.';
$runtime = InferenceRuntime::fromConfig(config: LLMConfig::fromPreset('openai'), events: $events);
$stream = Inference::fromRuntime($runtime)
    ->withMessages(Messages::fromString($prompt))
    ->withOptions(['max_tokens' => 64])
    ->withStreaming()
    ->stream()
    ->deltas();

echo "USER: {$prompt}\n";
echo "ASSISTANT: ";
foreach ($stream as $delta) {
    echo $delta->contentDelta;
}
echo "\n\n";

// Metrics buffered by Telemetry::metric() only reach an exporter that implements
// CanExportMetrics, and only once flush() runs — call it after the stream is
// fully consumed, same as span/log completion depends on the terminal events.
$telemetry->flush();

$exportedMetrics = [];
$metrics->exportTo(new CallbackExporter(function (iterable $m) use (&$exportedMetrics): void {
    foreach ($m as $metric) {
        $exportedMetrics[] = $metric;
    }
}));
$metrics->export();

echo "-- StreamMetricsCollector (hand-rolled, derived) --\n";
foreach (aggregateMetrics($exportedMetrics) as $aggregate) {
    printf(
        "[%s] %s%s = %.2f\n",
        $aggregate['type'],
        $aggregate['name'],
        formatTags($aggregate['tags']),
        aggregatedValue($aggregate),
    );
}
echo "\n";

echo "-- PolyglotTelemetryProjector metrics (catalog inference.client.*, emitted for free) --\n";
foreach ($telemetryExporter->metrics() as $metric) {
    printf("[%s] %s = %.2f\n", $metric->type(), $metric->name(), $metric->value());
}
echo "\n";

echo "-- PolyglotTelemetryProjector spans/logs --\n";
foreach ($telemetryExporter->observations() as $observation) {
    printf("[%s] %s (%s)\n", $observation->kind()->value, $observation->name(), $observation->status()->value);
}

assert(count($exportedMetrics) > 0, 'Expected non-empty hand-rolled metrics collection');
assert(count($telemetryExporter->metrics()) > 0, 'Expected Telemetry to have exported inference.client.* metrics');
assert(count($telemetryExporter->observations()) > 0, 'Expected Telemetry to have produced at least one span/log');

function formatTags(array $tags): string {
    if ($tags === []) {
        return '';
    }

    $keys = array_keys($tags);
    $values = array_values($tags);
    $tagList = array_map(
        static fn (string $key, mixed $value): string => "{$key}=\"{$value}\"",
        $keys,
        $values,
    );

    return ' {' . implode(', ', $tagList) . '}';
}

/**
 * @param iterable<Metric> $metrics
 * @return array<int, array{type: string, name: string, tags: array, count: int, sum: float, last: float}>
 */
function aggregateMetrics(iterable $metrics): array {
    $aggregates = [];
    foreach ($metrics as $metric) {
        $key = $metric->type() . '|' . $metric->name() . '|' . $metric->tags()->toKey();
        if (!array_key_exists($key, $aggregates)) {
            $aggregates[$key] = [
                'type' => $metric->type(),
                'name' => $metric->name(),
                'tags' => $metric->tags()->toArray(),
                'count' => 0,
                'sum' => 0.0,
                'last' => 0.0,
            ];
        }

        $aggregates[$key]['count'] += 1;
        $aggregates[$key]['sum'] += $metric->value();
        $aggregates[$key]['last'] = $metric->value();
    }

    return array_values($aggregates);
}

/**
 * @param array{type: string, count: int, sum: float, last: float} $aggregate
 */
function aggregatedValue(array $aggregate): float {
    return match ($aggregate['type']) {
        'counter' => $aggregate['sum'],
        'timer', 'histogram' => $aggregate['sum'] / max(1, $aggregate['count']),
        default => $aggregate['last'],
    };
}
?>