fix(anthropic): support the Claude 4.7+ and 5 model generations

livekit-plugins/livekit-plugins-anthropic/livekit/plugins/anthropic/llm.py

@@ -34,16 +34,24 @@
 )
 from livekit.agents.utils import is_given
 
-from .models import ChatModels
+from .log import logger
+from .models import (
+    ChatModels,
+    _model_supports_prefill,
+    _model_supports_sampling_params,
+    _model_thinking_support,
+)
 from .utils import CACHE_CONTROL_EPHEMERAL
 
-# Claude 4.6+ no longer supports prefilling (trailing assistant messages).
-_NO_PREFILL_PATTERNS = ("claude-sonnet-4-6", "claude-opus-4-6")
-
+# Rejected by Claude 4.7 and later. `top_p` has no constructor argument: it can only
+# reach the request through `extra_kwargs`.
+_SAMPLING_PARAMS = ("temperature", "top_p", "top_k")
 
-def _model_disables_prefill(model: str) -> bool:
-    """Return True if the model does not support assistant message prefilling."""
-    return any(model.startswith(p) for p in _NO_PREFILL_PATTERNS)
+# `max_tokens` is a single ceiling over thinking *and* the reply. 1024 is plenty for a
+# spoken turn, but a model that always thinks would spend most of it reasoning and
+# return a truncated (or empty) answer, so those get more headroom.
+_DEFAULT_MAX_TOKENS = 1024
+_DEFAULT_MAX_TOKENS_THINKING = 8192
 
 
 @dataclass
@@ -99,9 +107,19 @@ def __init__(
             Pass a custom ``httpx.Timeout`` to override (e.g. ``httpx.Timeout(5.0, read=60.0)``
             for very large contexts or extended thinking budgets).
         temperature (float, optional): The temperature for the Anthropic API. Defaults to None.
+            Claude 4.7 and later reject sampling parameters; on those models ``temperature``
+            and ``top_k`` are dropped from the request (with a one-time warning) instead of
+            failing it. Steer those models with the system prompt instead.
         parallel_tool_calls (bool, optional): Whether to parallelize tool calls. Defaults to None.
         tool_choice (ToolChoice, optional): The tool choice for the Anthropic API. Defaults to "auto".
         caching (Literal["ephemeral"], optional): If set to "ephemeral", caching will be enabled for the system prompt, tools, and chat history.
+
+        Thinking is disabled by default on every model that accepts the parameter: a voice
+        agent pays for it in latency and in ``max_tokens``, which caps thinking and the
+        reply together. Pass ``extra_kwargs={"thinking": {"type": "adaptive"}}`` to
+        ``chat()`` to opt back in — but note that thinking blocks are not kept in the chat
+        context, so a turn carrying a tool call is replayed without them, which Anthropic
+        can reject.
         """  # noqa: E501
 
         super().__init__()
@@ -117,6 +135,8 @@ def __init__(
             max_tokens=max_tokens,
             strict_tool_schema=_strict_tool_schema,
         )
+        self._sampling_params_warned = False
+        self._thinking_tools_warned = False
         anthropic_api_key = api_key if is_given(api_key) else os.environ.get("ANTHROPIC_API_KEY")
         if not anthropic_api_key:
             raise ValueError(
@@ -168,13 +188,35 @@ def chat(
         if is_given(self._opts.user):
             extra["user"] = self._opts.user
 
-        if is_given(self._opts.temperature):
-            extra["temperature"] = self._opts.temperature
+        self._apply_sampling_params(extra)
 
-        if is_given(self._opts.top_k):
-            extra["top_k"] = self._opts.top_k
+        # Thinking costs latency and eats into max_tokens, so it is off wherever it can
+        # be turned off. `extra_kwargs` wins: it is the escape hatch for callers that do
+        # want the model to reason before answering.
+        thinking_support = _model_thinking_support(self._opts.model)
+        thinking: Any = extra.get("thinking")
+        if thinking is None and thinking_support == "configurable":
+            thinking = extra["thinking"] = {"type": "disabled"}
 
-        extra["max_tokens"] = self._opts.max_tokens if is_given(self._opts.max_tokens) else 1024
+        thinking_enabled = thinking_support == "always_on" or (
+            isinstance(thinking, dict) and thinking.get("type") != "disabled"
+        )
+
+        if thinking_enabled and tools and not self._thinking_tools_warned:
+            self._thinking_tools_warned = True
+            logger.warning(
+                "%s will reason before answering, but thinking blocks are not kept in the "
+                "chat context: the next turn replays the assistant turn without them, "
+                "which Anthropic can reject when that turn contains a tool call",
+                self._opts.model,
+            )
+
+        if is_given(self._opts.max_tokens):
+            extra["max_tokens"] = self._opts.max_tokens
+        elif "max_tokens" not in extra:
+            extra["max_tokens"] = (
+                _DEFAULT_MAX_TOKENS_THINKING if thinking_enabled else _DEFAULT_MAX_TOKENS
+            )
 
         beta_flag: str | None = None
         if tools:
@@ -218,7 +260,7 @@ def chat(
                     extra["tool_choice"] = anthropic_tool_choice
 
         # Claude 4.6+ does not support prefilling (trailing assistant messages).
-        inject_trailing = _model_disables_prefill(self._opts.model)
+        inject_trailing = not _model_supports_prefill(self._opts.model)
         anthropic_ctx, extra_data = chat_ctx.to_provider_format(
             format="anthropic", inject_trailing_user_message=inject_trailing
         )
@@ -279,8 +321,49 @@ async def create_anthropic_stream() -> anthropic.AsyncStream[
             chat_ctx=chat_ctx,
             tools=tools or [],
             conn_options=conn_options,
+            thinking_expected=thinking_enabled,
         )
 
+    def _apply_sampling_params(self, extra: dict[str, Any]) -> None:
+        """Add `temperature`/`top_k` to the request, unless the model rejects them.
+
+        Claude 4.7 and later answer a request carrying sampling parameters with a 400, so
+        they are dropped instead — including the ones that came in through
+        ``extra_kwargs``, which are already in ``extra`` by the time this runs. The
+        warning is logged once per instance to stay visible without flooding a
+        long-running session.
+        """
+        if _model_supports_sampling_params(self._opts.model):
+            if is_given(self._opts.temperature):
+                extra["temperature"] = self._opts.temperature
+
+            if is_given(self._opts.top_k):
+                extra["top_k"] = self._opts.top_k
+
+            return
+
+        from_extra = {name for name in _SAMPLING_PARAMS if name in extra}
+        for name in from_extra:
+            del extra[name]
+
+        from_opts = {
+            name
+            for name, value in (
+                ("temperature", self._opts.temperature),
+                ("top_k", self._opts.top_k),
+            )
+            if is_given(value)
+        }
+        dropped = [name for name in _SAMPLING_PARAMS if name in from_extra | from_opts]
+        if dropped and not self._sampling_params_warned:
+            self._sampling_params_warned = True
+            logger.warning(
+                "%s rejects sampling parameters, dropping %s from the request; "
+                "steer the model with the system prompt instead",
+                self._opts.model,
+                ", ".join(dropped),
+            )
+
 
 class LLMStream(llm.LLMStream):
     def __init__(
@@ -293,9 +376,11 @@ def __init__(
         chat_ctx: llm.ChatContext,
         tools: list[Tool],
         conn_options: APIConnectOptions,
+        thinking_expected: bool = False,
     ) -> None:
         super().__init__(llm, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options)
         self._create_anthropic_stream = create_anthropic_stream
+        self._thinking_expected = thinking_expected
 
         # current function call that we're waiting for full completion (args are streamed)
         self._tool_call_id: str | None = None
@@ -304,13 +389,36 @@ def __init__(
 
         self._request_id: str = ""
         self._ignoring_cot = False  # ignore chain of thought
+        self._thinking_blocks = 0
+        self._thinking_chars = 0
         self._input_tokens = 0
         self._cache_creation_tokens = 0
         self._cache_read_tokens = 0
         self._output_tokens = 0
 
+    def _reset_stream_state(self) -> None:
+        """Clear the state a single attempt builds up.
+
+        The base class re-enters `_run` on the same instance, and it only retries when the
+        failed attempt emitted no chunk — a stream that died mid-thinking, mid
+        chain-of-thought or mid tool-call would otherwise carry that state into the next
+        attempt, including the `_ignoring_cot` latch that swallows text. `_input_tokens`
+        and `_output_tokens` are reassigned unconditionally at `message_start`, but the
+        cache counters are only assigned there when the new attempt reports a non-zero
+        value, so they are cleared here.
+        """
+        self._tool_call_id = None
+        self._fnc_name = None
+        self._fnc_raw_arguments = None
+        self._ignoring_cot = False
+        self._thinking_blocks = 0
+        self._thinking_chars = 0
+        self._cache_creation_tokens = 0
+        self._cache_read_tokens = 0
+
     async def _run(self) -> None:
         retryable = True
+        self._reset_stream_state()
         try:
             async with await self._create_anthropic_stream() as stream:
                 async for event in stream:
@@ -319,6 +427,14 @@ async def _run(self) -> None:
                         self._event_ch.send_nowait(chat_chunk)
                         retryable = False
 
+                if self._thinking_blocks:
+                    logger.debug(
+                        "anthropic: %s returned %d thinking block(s), %d characters",
+                        self._llm.model,
+                        self._thinking_blocks,
+                        self._thinking_chars,
+                    )
+
                 # https://docs.anthropic.com/en/docs/build-with-claude/prompt-caching#tracking-cache-performance
                 prompt_token = (
                     self._input_tokens + self._cache_creation_tokens + self._cache_read_tokens
@@ -364,9 +480,18 @@ def _parse_event(self, event: anthropic.types.RawMessageStreamEvent) -> llm.Chat
                 self._tool_call_id = event.content_block.id
                 self._fnc_name = event.content_block.name
                 self._fnc_raw_arguments = ""
+            elif event.content_block.type in ("thinking", "redacted_thinking"):
+                self._note_thinking_block()
         elif event.type == "content_block_delta":
             delta = event.delta
-            if delta.type == "text_delta":
+            if delta.type == "thinking_delta":
+                # Reasoning is not part of the answer: it must not reach the caller (a
+                # voice agent would speak it), but it is billed and it is counted below.
+                self._thinking_chars += len(delta.thinking)
+                return None
+            elif delta.type == "signature_delta":
+                return None
+            elif delta.type == "text_delta":
                 text = delta.text
 
                 if self._tools is not None:
@@ -410,3 +535,13 @@ def _parse_event(self, event: anthropic.types.RawMessageStreamEvent) -> llm.Chat
                 return chat_chunk
 
         return None
+
+    def _note_thinking_block(self) -> None:
+        """Record a thinking block, warning once when one was not expected."""
+        self._thinking_blocks += 1
+        if self._thinking_blocks == 1 and not self._thinking_expected:
+            logger.warning(
+                "anthropic: %s returned a thinking block although thinking was not "
+                "requested; the reasoning is dropped but still counts against max_tokens",
+                self._llm.model,
+            )