From 3979bf0e75379bb2403ef73cf89a62ecc2edb80a Mon Sep 17 00:00:00 2001 From: kamilbenkirane Date: Mon, 4 May 2026 22:32:43 +0200 Subject: [PATCH 1/6] feat(telemetry): GenAI OpenTelemetry spans on _predict and _stream celeste-python now emits GenAI v1.41 semantic-convention spans on every ModalityClient._predict and ._stream call when a TracerProvider is configured. No changes to public API; default installs gain zero new deps; spans no-op when no provider is set. - New src/celeste/telemetry.py: lazy OTel import + no-op fallback, provider/modality maps, request_attributes, span_name, output_attributes, trace_stream async generator - client.py _predict: wrap body in tracer.start_as_current_span; OTel's context manager handles exception/lifecycle - client.py _stream: detached span + use_span(end_on_exit=True) inside the iterator wrapper; records time_to_first_chunk on first chunk - pyproject: optional [otel] extra adding opentelemetry-api>=1.30; per-file ANN401 ignore for telemetry.py (mirrors OTel API surface) Closes #269. --- pyproject.toml | 2 + src/celeste/client.py | 75 +++++++++++------ src/celeste/telemetry.py | 170 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 224 insertions(+), 23 deletions(-) create mode 100644 src/celeste/telemetry.py diff --git a/pyproject.toml b/pyproject.toml index f82578e..f12b87d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ dependencies = [ [project.optional-dependencies] gcp = ["google-auth[requests]>=2.0.0"] +otel = ["opentelemetry-api>=1.30"] [project.urls] Homepage = "https://withceleste.ai" @@ -121,6 +122,7 @@ known-first-party = ["celeste"] [tool.ruff.lint.per-file-ignores] "tests/*" = ["D"] # No docstrings required in tests "**/client.py" = ["ANN401"] # Allow Any return types in mixin client classes +"**/telemetry.py" = ["ANN401"] # Allow Any: mirrors OTel API surface for forwarding [tool.mypy] python_version = "3.12" diff --git a/src/celeste/client.py b/src/celeste/client.py index a8e9f49..0f8f5af 100644 --- a/src/celeste/client.py +++ b/src/celeste/client.py @@ -9,6 +9,7 @@ import httpx from pydantic import BaseModel, ConfigDict, Field +from celeste import telemetry from celeste.auth import Authentication from celeste.core import Modality, Protocol, Provider from celeste.exceptions import ( @@ -212,29 +213,44 @@ async def _predict( Returns: Output of the parameterized type. """ - inputs, parameters = self._validate_artifacts(inputs, **parameters) - request_body = self._build_request(inputs, extra_body=extra_body, **parameters) - response_data = await self._make_request( - request_body, endpoint=endpoint, extra_headers=extra_headers, **parameters - ) - content = self._parse_content(response_data) - content = self._transform_output(content, **parameters) - tool_calls = self._parse_tool_calls(response_data) - reasoning, signature = self._parse_reasoning(response_data) - kwargs: dict[str, Any] = {} - if reasoning is not None: - kwargs["reasoning"] = reasoning - if signature: - kwargs["signature"] = signature - output = self._output_class()( - content=content, - usage=self._get_usage(response_data), - finish_reason=self._get_finish_reason(response_data), - metadata=self._build_metadata(response_data), - tool_calls=tool_calls, - **kwargs, - ) - return output + with telemetry.tracer.start_as_current_span( + telemetry.span_name(self.modality, self.model), + attributes=telemetry.request_attributes( + model=self.model, + provider=self.provider, + protocol=self.protocol, + modality=self.modality, + ), + ) as span: + inputs, parameters = self._validate_artifacts(inputs, **parameters) + request_body = self._build_request( + inputs, extra_body=extra_body, **parameters + ) + response_data = await self._make_request( + request_body, + endpoint=endpoint, + extra_headers=extra_headers, + **parameters, + ) + content = self._parse_content(response_data) + content = self._transform_output(content, **parameters) + tool_calls = self._parse_tool_calls(response_data) + reasoning, signature = self._parse_reasoning(response_data) + kwargs: dict[str, Any] = {} + if reasoning is not None: + kwargs["reasoning"] = reasoning + if signature: + kwargs["signature"] = signature + output = self._output_class()( + content=content, + usage=self._get_usage(response_data), + finish_reason=self._get_finish_reason(response_data), + metadata=self._build_metadata(response_data), + tool_calls=tool_calls, + **kwargs, + ) + span.set_attributes(telemetry.output_attributes(output)) + return output def _parse_tool_calls(self, response_data: dict[str, Any]) -> list[ToolCall]: """Parse tool calls from response. Override in providers that support tools.""" @@ -281,6 +297,18 @@ def _stream( request_body = self._build_request( inputs, extra_body=extra_body, streaming=True, **parameters ) + span = telemetry.tracer.start_span( + telemetry.span_name(self.modality, self.model), + attributes={ + **telemetry.request_attributes( + model=self.model, + provider=self.provider, + protocol=self.protocol, + modality=self.modality, + ), + "gen_ai.request.stream": True, + }, + ) sse_iterator = self._make_stream_request( request_body, endpoint=endpoint, @@ -288,6 +316,7 @@ def _stream( **parameters, ) sse_iterator = enrich_stream_errors(sse_iterator, self._handle_error_response) + sse_iterator = telemetry.trace_stream(sse_iterator, span) return stream_class( sse_iterator, transform_output=self._transform_output, diff --git a/src/celeste/telemetry.py b/src/celeste/telemetry.py new file mode 100644 index 0000000..8b60abc --- /dev/null +++ b/src/celeste/telemetry.py @@ -0,0 +1,170 @@ +"""OpenTelemetry GenAI telemetry for Celeste.""" + +import time +from collections.abc import AsyncIterator, Iterator +from contextlib import contextmanager +from typing import Any + +from celeste.core import Modality, Protocol, Provider +from celeste.io import Output +from celeste.models import Model + +_PROVIDER_NAME_MAP: dict[Provider, str] = { + Provider.OPENAI: "openai", + Provider.ANTHROPIC: "anthropic", + Provider.GOOGLE: "gcp.gemini", + Provider.MISTRAL: "mistral_ai", + Provider.XAI: "x_ai", + Provider.GROQ: "groq", + Provider.DEEPSEEK: "deepseek", + Provider.PERPLEXITY: "perplexity", + Provider.COHERE: "cohere", +} + +_MODALITY_OPERATION: dict[Modality, str] = { + Modality.TEXT: "chat", + Modality.EMBEDDINGS: "embeddings", +} + + +class _NoOpSpan: + """Span stand-in used when ``opentelemetry-api`` is not installed.""" + + def set_attribute(self, key: str, value: Any) -> None: + """Discard the attribute.""" + + def set_attributes(self, attributes: dict[str, Any]) -> None: + """Discard the attribute mapping.""" + + def set_status(self, *args: Any, **kwargs: Any) -> None: + """Discard the status update.""" + + def record_exception(self, *args: Any, **kwargs: Any) -> None: + """Discard the exception.""" + + def end(self) -> None: + """Discard the end signal.""" + + +class _NoOpTracer: + """Tracer stand-in used when ``opentelemetry-api`` is not installed.""" + + @contextmanager + def start_as_current_span(self, name: str, **kwargs: Any) -> Iterator[_NoOpSpan]: + """Yield a no-op span as a context manager.""" + yield _NoOpSpan() + + def start_span(self, name: str, **kwargs: Any) -> _NoOpSpan: + """Return a detached no-op span.""" + return _NoOpSpan() + + +@contextmanager +def _noop_use_span(span: Any, end_on_exit: bool = False) -> Iterator[Any]: + """No-op stand-in for ``opentelemetry.trace.use_span``.""" + yield span + + +try: + from opentelemetry import trace as _otel_trace + + tracer: Any = _otel_trace.get_tracer("celeste") + use_span: Any = _otel_trace.use_span +except ImportError: + tracer = _NoOpTracer() + use_span = _noop_use_span + + +def request_attributes( + *, + model: Model, + provider: Provider | None, + protocol: Protocol | None, + modality: Modality, +) -> dict[str, Any]: + """Build GenAI request attributes for a span. + + Off-spec providers/modalities get ``celeste.*`` keys so the data is + queryable without polluting the ``gen_ai.*`` namespace. + + Args: + model: Resolved model. + provider: Resolved provider, if any. + protocol: Resolved protocol, if any. + modality: Operation modality. + + Returns: + Attribute dict suitable for passing to ``start_as_current_span``. + """ + attrs: dict[str, Any] = {"gen_ai.request.model": model.id} + provider_name = _PROVIDER_NAME_MAP.get(provider) if provider is not None else None + if provider_name is not None: + attrs["gen_ai.provider.name"] = provider_name + elif protocol is not None: + attrs["gen_ai.provider.name"] = protocol.value + operation = _MODALITY_OPERATION.get(modality) + if operation is not None: + attrs["gen_ai.operation.name"] = operation + else: + attrs["celeste.modality"] = modality.value + return attrs + + +def span_name(modality: Modality, model: Model) -> str: + """Build the span name per GenAI semconv ``"{operation} {model}"``. + + Falls back to ``"celeste.{modality} {model}"`` for modalities the spec + does not yet cover. + """ + operation = _MODALITY_OPERATION.get(modality) + if operation is not None: + return f"{operation} {model.id}" + return f"celeste.{modality.value} {model.id}" + + +def output_attributes(output: Output[Any]) -> dict[str, Any]: + """Extract GenAI response attributes from a typed Output.""" + attrs: dict[str, Any] = {} + input_tokens = getattr(output.usage, "input_tokens", None) + output_tokens = getattr(output.usage, "output_tokens", None) + if isinstance(input_tokens, int): + attrs["gen_ai.usage.input_tokens"] = input_tokens + if isinstance(output_tokens, int): + attrs["gen_ai.usage.output_tokens"] = output_tokens + if output.finish_reason is not None and output.finish_reason.reason is not None: + attrs["gen_ai.response.finish_reasons"] = (output.finish_reason.reason,) + return attrs + + +async def trace_stream( + sse_iterator: AsyncIterator[dict[str, Any]], + span: Any, +) -> AsyncIterator[dict[str, Any]]: + """Wrap an SSE iterator to record TTFC and end the span at iteration's end. + + OTel's ``use_span(span, end_on_exit=True)`` context manager makes the + span current, auto-records exceptions, sets ERROR status on exception, + and ends the span on exit (including ``GeneratorExit`` from consumer + abandonment). No try/except needed. + """ + with use_span(span, end_on_exit=True): + started = time.monotonic() + seen_first = False + async for event in sse_iterator: + if not seen_first: + seen_first = True + span.set_attribute( + "gen_ai.response.time_to_first_chunk", + time.monotonic() - started, + ) + yield event + + +__all__ = [ + "output_attributes", + "request_attributes", + "span_name", + "trace_stream", + "tracer", + "use_span", +] From b87c05db2af875fc2fb6cec75ab0b32a60f49ff4 Mon Sep 17 00:00:00 2001 From: kamilbenkirane Date: Mon, 4 May 2026 22:33:00 +0200 Subject: [PATCH 2/6] docs(telemetry): document OpenTelemetry tracing setup Add a 'Tracing with OpenTelemetry' README section covering: - uv add 'celeste-ai[otel]' - minimal TracerProvider + OTLP/Console exporter snippet - expected span name and attribute schema - OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental opt-in - attribute drift caveat until GenAI semconv stabilizes --- README.md | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/README.md b/README.md index 08fd1df..c49b9e4 100644 --- a/README.md +++ b/README.md @@ -218,6 +218,38 @@ Catch errors **before** production. --- +## 🔭 Tracing with OpenTelemetry + +Celeste emits [GenAI semantic-convention](https://opentelemetry.io/docs/specs/semconv/gen-ai/) spans on every modality call when a `TracerProvider` is configured. Spans are zero-cost when no provider is set. + +```bash +uv add 'celeste-ai[otel]' +``` + +Configure a provider once at startup (SDK setup is the consumer's responsibility): + +```python +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter + +provider = TracerProvider() +provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) +trace.set_tracer_provider(provider) +``` + +Every `celeste.text.generate(...)` and `celeste.text.stream.generate(...)` call (and the equivalents on other modalities) emits a span named `chat ` (or `embeddings ` for embeddings; `celeste. ` for image/video/audio operations the spec doesn't yet cover). Attributes follow GenAI v1.41.0: `gen_ai.provider.name`, `gen_ai.request.model`, `gen_ai.operation.name`, `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, `gen_ai.response.finish_reasons`, plus `gen_ai.request.stream` and `gen_ai.response.time_to_first_chunk` on streaming spans. + +The GenAI semconv is currently `Status: Development`. Pin attribute shapes via the standard opt-in: + +```bash +export OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental +``` + +Attribute names may drift between celeste-ai minor versions until the spec stabilizes. + +--- + ## 🤝 Contributing We welcome contributions! See [CONTRIBUTING.md](CONTRIBUTING.md). From 9d143610004dc5e60a25e2226af718ad97d3e1c3 Mon Sep 17 00:00:00 2001 From: kamilbenkirane Date: Mon, 4 May 2026 22:38:02 +0200 Subject: [PATCH 3/6] fix(telemetry): add mypy ignore_missing_imports for opentelemetry CI runs mypy without the [otel] extra installed, so 'from opentelemetry import trace' fails the type check even though the runtime import is guarded by try/except. Mirror the existing pattern used for httpx / httpx_sse / google.* / etc. --- pyproject.toml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index f12b87d..28b59c0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -165,6 +165,10 @@ ignore_missing_imports = true module = "google.*" ignore_missing_imports = true +[[tool.mypy.overrides]] +module = "opentelemetry.*" +ignore_missing_imports = true + [[tool.mypy.overrides]] module = [ "celeste.modalities.text.client", From 34416d796699cae9df1c9d5744561f1d3c61f715 Mon Sep 17 00:00:00 2001 From: kamilbenkirane Date: Mon, 4 May 2026 22:38:28 +0200 Subject: [PATCH 4/6] Revert "docs(telemetry): document OpenTelemetry tracing setup" This reverts commit b87c05db2af875fc2fb6cec75ab0b32a60f49ff4. --- README.md | 32 -------------------------------- 1 file changed, 32 deletions(-) diff --git a/README.md b/README.md index c49b9e4..08fd1df 100644 --- a/README.md +++ b/README.md @@ -218,38 +218,6 @@ Catch errors **before** production. --- -## 🔭 Tracing with OpenTelemetry - -Celeste emits [GenAI semantic-convention](https://opentelemetry.io/docs/specs/semconv/gen-ai/) spans on every modality call when a `TracerProvider` is configured. Spans are zero-cost when no provider is set. - -```bash -uv add 'celeste-ai[otel]' -``` - -Configure a provider once at startup (SDK setup is the consumer's responsibility): - -```python -from opentelemetry import trace -from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter - -provider = TracerProvider() -provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) -trace.set_tracer_provider(provider) -``` - -Every `celeste.text.generate(...)` and `celeste.text.stream.generate(...)` call (and the equivalents on other modalities) emits a span named `chat ` (or `embeddings ` for embeddings; `celeste. ` for image/video/audio operations the spec doesn't yet cover). Attributes follow GenAI v1.41.0: `gen_ai.provider.name`, `gen_ai.request.model`, `gen_ai.operation.name`, `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, `gen_ai.response.finish_reasons`, plus `gen_ai.request.stream` and `gen_ai.response.time_to_first_chunk` on streaming spans. - -The GenAI semconv is currently `Status: Development`. Pin attribute shapes via the standard opt-in: - -```bash -export OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental -``` - -Attribute names may drift between celeste-ai minor versions until the spec stabilizes. - ---- - ## 🤝 Contributing We welcome contributions! See [CONTRIBUTING.md](CONTRIBUTING.md). From 08af74842ae5fe6ef12664d066c88a871d77a546 Mon Sep 17 00:00:00 2001 From: kamilbenkirane Date: Mon, 4 May 2026 22:39:59 +0200 Subject: [PATCH 5/6] fix(telemetry): install opentelemetry-api in dev group instead of mypy ignore The previous fix (`ignore_missing_imports` for opentelemetry.*) was a workaround that disabled real type-checking on OTel symbols. The opentelemetry-api package ships py.typed and is fully type-checkable when installed. Add opentelemetry-api to the dev dependency group so CI's mypy sees the real types. Drop the ignore_missing_imports override. --- pyproject.toml | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 28b59c0..af1ec9f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,6 +50,7 @@ dev = [ "bandit[toml]>=1.7.5", "pre-commit>=3.5.0", "pytest-rerunfailures>=16.1", + "opentelemetry-api>=1.30", ] [build-system] @@ -165,10 +166,6 @@ ignore_missing_imports = true module = "google.*" ignore_missing_imports = true -[[tool.mypy.overrides]] -module = "opentelemetry.*" -ignore_missing_imports = true - [[tool.mypy.overrides]] module = [ "celeste.modalities.text.client", From b7ee8d5efe32029639fd325e0afd164d3d37d9ab Mon Sep 17 00:00:00 2001 From: kamilbenkirane Date: Mon, 4 May 2026 22:42:16 +0200 Subject: [PATCH 6/6] fix(ci): install otel extra in type-check job Earlier commit moved opentelemetry-api into the dev group as a workaround. Cleaner: mirror the existing 'extras-per-job' pattern (the test job already does `uv sync --extra gcp`). The type-check job now `uv sync --extra otel` so mypy sees the real OTel types. Reverts the dev-group addition. --- .github/workflows/ci.yml | 1 + pyproject.toml | 1 - 2 files changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 36b7aba..651f3d1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,6 +52,7 @@ jobs: - uses: ./.github/actions/setup-python-uv with: python-version: ${{ inputs.python-version || '3.12' }} + - run: uv sync --extra otel - run: uv run mypy -p celeste && uv run mypy tests/ security: diff --git a/pyproject.toml b/pyproject.toml index af1ec9f..f12b87d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,7 +50,6 @@ dev = [ "bandit[toml]>=1.7.5", "pre-commit>=3.5.0", "pytest-rerunfailures>=16.1", - "opentelemetry-api>=1.30", ] [build-system]