diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ebbf564..950905da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 * The `snowflake` extra (i.e., `pip install "chatlas[snowflake]"`) no longer installs `snowflake-ml-python`, which pulled in the heavy torch ecosystem and blocked installation on Python 3.13+. `ChatSnowflake()` has used the Cortex REST API (`snowflake.core`) since v0.11.0, so it only needs `snowflake-core` and `snowflake-snowpark-python`. +### New features + +* Streamed responses now record the time to first token (in seconds) as the `gen_ai.response.time_to_first_chunk` attribute on the OpenTelemetry `chat` span. + ### Bug fixes * `ChatBedrock()` (with `api="messages"` or `api="responses"`) no longer sends a URL for `content_pdf_url()`/`content_document_url()` content, which `bedrock-mantle` rejects since it can't fetch URLs itself. The bytes are downloaded and sent instead, as they already were before URL passthrough was added. `ChatBedrockAnthropic()` gets the same fix, since Bedrock's InvokeModel API can't fetch `url` sources either. (#410) diff --git a/chatlas/_chat.py b/chatlas/_chat.py index 82d742bd..49fbb717 100644 --- a/chatlas/_chat.py +++ b/chatlas/_chat.py @@ -65,6 +65,7 @@ activate_span, end_span, record_chat_result, + record_chat_ttft_attr, record_error, start_agent_span, start_chat_span, @@ -3007,6 +3008,9 @@ def emit(x: str | Content): # provider's HTTP instrumentor spans nest under it -- never # across a `yield`, which would leak the context into the # consumer's scope. + # Measured from request issuance; chat_perform() blocks until + # headers arrive. + stream_start = time.monotonic() with activate_span(chat_span): response = self.provider.chat_perform( stream=True, @@ -3034,8 +3038,12 @@ def emit(x: str | Content): for content in self.provider.stream_content( chunk, result, turns=provider_turns ): + text = display_text(content) + if stream_start is not None and text: + record_chat_ttft_attr(chat_span, stream_start) + stream_start = None yield from acc.process_content( - content, display_text(content), content_mode, emit + content, text, content_mode, emit ) yield from acc.flush_thinking(content_mode, emit) @@ -3167,6 +3175,9 @@ def emit(x: str | Content): # provider's HTTP instrumentor spans nest under it -- never # across a `yield`, which would leak the context into the # consumer's scope. + # Measured from request issuance; chat_perform_async() blocks + # until headers arrive. + stream_start = time.monotonic() with activate_span(chat_span): response = await self.provider.chat_perform_async( stream=True, @@ -3194,8 +3205,12 @@ def emit(x: str | Content): for content in self.provider.stream_content( chunk, result, turns=provider_turns ): + text = display_text(content) + if stream_start is not None and text: + record_chat_ttft_attr(chat_span, stream_start) + stream_start = None for item in acc.process_content( - content, display_text(content), content_mode, emit + content, text, content_mode, emit ): yield item diff --git a/chatlas/_otel.py b/chatlas/_otel.py index 9d76e6bc..ddc8f6f0 100644 --- a/chatlas/_otel.py +++ b/chatlas/_otel.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import time from contextlib import AbstractContextManager, nullcontext from typing import TYPE_CHECKING, Any, Optional @@ -101,6 +102,17 @@ def start_tool_span( ) +def record_chat_ttft_attr(span: Span, start: float) -> None: + """Record the time to first token (in seconds) on the chat span as the + `gen_ai.response.time_to_first_chunk` semconv attribute.""" + if not span.is_recording(): + return + span.set_attribute( + "gen_ai.response.time_to_first_chunk", + time.monotonic() - start, + ) + + def record_chat_result( span: Span, turn: AssistantTurn[Any], diff --git a/chatlas/types/anthropic/_client.py b/chatlas/types/anthropic/_client.py index 908e52a1..e70693d6 100644 --- a/chatlas/types/anthropic/_client.py +++ b/chatlas/types/anthropic/_client.py @@ -7,6 +7,7 @@ Any, Awaitable, Callable, + Literal, Mapping, Optional, Sequence, @@ -37,6 +38,7 @@ class ChatClientArgs(TypedDict, total=False): max_retries: int default_headers: Optional[Mapping[str, str]] default_query: Optional[Mapping[str, object]] + open_telemetry: Union[anthropic.OpenTelemetryOptions, Literal[False], None] http_client: httpx2.AsyncClient | None middleware: Optional[ Sequence[ diff --git a/chatlas/types/anthropic/_client_bedrock.py b/chatlas/types/anthropic/_client_bedrock.py index fafe1447..b7e8711e 100644 --- a/chatlas/types/anthropic/_client_bedrock.py +++ b/chatlas/types/anthropic/_client_bedrock.py @@ -6,6 +6,7 @@ Any, Awaitable, Callable, + Literal, Mapping, Optional, Sequence, @@ -30,6 +31,7 @@ class ChatBedrockClientArgs(TypedDict, total=False): max_retries: int default_headers: Optional[Mapping[str, str]] default_query: Optional[Mapping[str, object]] + open_telemetry: Union[anthropic.OpenTelemetryOptions, Literal[False], None] http_client: httpx2.AsyncClient | None middleware: Optional[ Sequence[ diff --git a/chatlas/types/anthropic/_submit.py b/chatlas/types/anthropic/_submit.py index cc150293..5392718e 100644 --- a/chatlas/types/anthropic/_submit.py +++ b/chatlas/types/anthropic/_submit.py @@ -48,6 +48,7 @@ class SubmitInputArgs(TypedDict, total=False): messages: Iterable[anthropic.types.message_param.MessageParam] model: Union[ Literal[ + "claude-haiku-5-5", "claude-sonnet-5-5", "claude-fable-5-1", "claude-opus-5-5", @@ -86,6 +87,7 @@ class SubmitInputArgs(TypedDict, total=False): anthropic.types.output_config_param.OutputConfigParam | anthropic.Omit ) service_tier: Union[Literal["auto", "standard_only"], anthropic.Omit] + speed: Union[Literal["fast", "standard"], None, anthropic.Omit] stop_sequences: Union[Sequence[str], anthropic.Omit] stream: Union[Literal[False], Literal[True], anthropic.Omit] system: Union[ diff --git a/pyproject.toml b/pyproject.toml index 9ca0836d..bb1e6daa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -104,7 +104,9 @@ mcp = [ "mcp>=2.0.0;python_version>='3.10'" ] eval = [ - "inspect-ai;python_version>='3.10'" + "inspect-ai;python_version>='3.10'", + # Inspect's Anthropic adapter constructs MessageParam, a union in SDK 2.0. + "anthropic<2.0;python_version>='3.10'", ] # Provider extras ---- anthropic = ["anthropic>=1.0.0"] diff --git a/tests/test_otel.py b/tests/test_otel.py index 7c7248d4..f9959218 100644 --- a/tests/test_otel.py +++ b/tests/test_otel.py @@ -1,16 +1,25 @@ from __future__ import annotations import json -from typing import Any, cast +from collections.abc import AsyncIterator, Iterator, Sequence +from types import SimpleNamespace +from typing import cast import pytest pytest.importorskip("opentelemetry.sdk") -from chatlas import ChatOpenAI, _otel -from chatlas._content import ContentText, ContentToolRequest, ContentToolResult, ToolInfo -from chatlas._turn import AssistantTurn +from chatlas import ChatOpenAI, _chat, _otel +from chatlas._content import ( + Content, + ContentText, + ContentThinkingDelta, + ContentToolRequest, + ContentToolResult, + ToolInfo, +) from chatlas._tools import ToolBuiltIn +from chatlas._turn import AssistantTurn, Turn from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter @@ -269,6 +278,93 @@ def test_streaming_span_lifecycle(otel_setup: InMemorySpanExporter): assert chat_span.status.status_code.name == "UNSET", ( f"Successful chat span should be UNSET, got {chat_span.status.status_code.name}" ) + # Time to first token is recorded on the streamed chat span. + attrs = chat_span.attributes or {} + ttft = attrs.get("gen_ai.response.time_to_first_chunk") + assert isinstance(ttft, float) and ttft > 0, ( + f"Expected gen_ai.response.time_to_first_chunk > 0, got {ttft!r}" + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize( + "first_content", + [ContentText(text="hi"), ContentThinkingDelta(thinking="hmm")], + ids=["text", "thinking"], +) +async def test_ttft_recorded_at_first_non_empty_token( + otel_setup: InMemorySpanExporter, + monkeypatch: pytest.MonkeyPatch, + async_mode: bool, + first_content: ContentText | ContentThinkingDelta, +) -> None: + chat = ChatOpenAI(model="gpt-4o-mini") + now = 0.0 + clock = SimpleNamespace(monotonic=lambda: now) + monkeypatch.setattr(_chat, "time", clock) + monkeypatch.setattr(_otel, "time", clock) + + def chunks() -> Iterator[object]: + nonlocal now + for timestamp in (2.0, 5.0, 9.0): + now = timestamp + yield object() + + def streaming_perform(*args: object, **kwargs: object) -> Iterator[object]: + nonlocal now + now = 1.0 + return chunks() + + async def streaming_perform_async( + *args: object, **kwargs: object + ) -> AsyncIterator[object]: + nonlocal now + now = 1.0 + + async def gen() -> AsyncIterator[object]: + for chunk in chunks(): + yield chunk + + return gen() + + content_chunks: list[list[Content]] = [ + [ContentText.model_construct(text="")], + [first_content], + [ContentText(text="there")], + ] + stream_contents = iter(content_chunks) + + def stream_content( + chunk: object, completion: object, turns: Sequence[Turn] = () + ) -> list[Content]: + return next(stream_contents) + + def stream_merge_chunks(result: object, chunk: object) -> object: + return chunk + + def stream_turn( + completion: object, has_data_model: bool, turns: Sequence[Turn] = () + ) -> AssistantTurn[object]: + return AssistantTurn("hi there") + + monkeypatch.setattr(chat.provider, "chat_perform", streaming_perform) + monkeypatch.setattr(chat.provider, "chat_perform_async", streaming_perform_async) + monkeypatch.setattr(chat.provider, "stream_content", stream_content) + monkeypatch.setattr(chat.provider, "stream_merge_chunks", stream_merge_chunks) + monkeypatch.setattr(chat.provider, "stream_turn", stream_turn) + + if async_mode: + async for _ in await chat.stream_async("Say hello."): + pass + else: + list(chat.stream("Say hello.")) + + spans = otel_setup.get_finished_spans() + chat_spans = [s for s in spans if s.name.startswith("chat ")] + assert len(chat_spans) == 1 + attrs = chat_spans[0].attributes or {} + assert attrs.get("gen_ai.response.time_to_first_chunk") == 5.0 def test_chat_error_recorded(otel_setup: InMemorySpanExporter): @@ -534,6 +630,12 @@ async def gen(): assert _parent_span_id(iter_span) == chat_span.context.span_id, ( "provider span created while iterating the async stream should nest under the chat span" ) + # Time to first token is recorded on the async streamed chat span. + attrs = chat_span.attributes or {} + ttft = attrs.get("gen_ai.response.time_to_first_chunk") + assert isinstance(ttft, float) and ttft > 0, ( + f"Expected gen_ai.response.time_to_first_chunk > 0, got {ttft!r}" + ) @pytest.mark.vcr