mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-06 17:38:35 +00:00
b94bc18e59
- Do not send reasoning_effort="none" to APIs (prevents 400 on gemma/Gemini) - Treat "none" as thinking disabled in thinking_style, Kimi, and reasoning_content backfill paths - Fix Anthropic extended thinking not respecting "none" - Fix Azure OpenAI temperature suppression and reasoning body for "none" - Fix Codex reasoning body for "none" - Add "gemma" keyword to Gemini ProviderSpec for correct auto routing
608 lines
23 KiB
Python
608 lines
23 KiB
Python
"""Anthropic provider — direct SDK integration for Claude models."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import re
|
|
import secrets
|
|
import string
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
import json_repair
|
|
|
|
from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
|
|
|
|
_ALNUM = string.ascii_letters + string.digits
|
|
|
|
|
|
def _gen_tool_id() -> str:
|
|
return "toolu_" + "".join(secrets.choice(_ALNUM) for _ in range(22))
|
|
|
|
|
|
class AnthropicProvider(LLMProvider):
|
|
"""LLM provider using the native Anthropic SDK for Claude models.
|
|
|
|
Handles message format conversion (OpenAI → Anthropic Messages API),
|
|
prompt caching, extended thinking, tool calls, and streaming.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
api_key: str | None = None,
|
|
api_base: str | None = None,
|
|
default_model: str = "claude-sonnet-4-20250514",
|
|
extra_headers: dict[str, str] | None = None,
|
|
):
|
|
super().__init__(api_key, api_base)
|
|
self.default_model = default_model
|
|
self.extra_headers = extra_headers or {}
|
|
|
|
from anthropic import AsyncAnthropic
|
|
|
|
client_kw: dict[str, Any] = {}
|
|
if api_key:
|
|
client_kw["api_key"] = api_key
|
|
if api_base:
|
|
client_kw["base_url"] = api_base
|
|
if extra_headers:
|
|
client_kw["default_headers"] = extra_headers
|
|
# Keep retries centralized in LLMProvider._run_with_retry to avoid retry amplification.
|
|
client_kw["max_retries"] = 0
|
|
self._client = AsyncAnthropic(**client_kw)
|
|
|
|
@classmethod
|
|
def _handle_error(cls, e: Exception) -> LLMResponse:
|
|
response = getattr(e, "response", None)
|
|
headers = getattr(response, "headers", None)
|
|
payload = (
|
|
getattr(e, "body", None)
|
|
or getattr(e, "doc", None)
|
|
or getattr(response, "text", None)
|
|
)
|
|
if payload is None and response is not None:
|
|
response_json = getattr(response, "json", None)
|
|
if callable(response_json):
|
|
try:
|
|
payload = response_json()
|
|
except Exception:
|
|
payload = None
|
|
payload_text = payload if isinstance(payload, str) else str(payload) if payload is not None else ""
|
|
msg = f"Error: {payload_text.strip()[:500]}" if payload_text.strip() else f"Error calling LLM: {e}"
|
|
retry_after = cls._extract_retry_after_from_headers(headers)
|
|
if retry_after is None:
|
|
retry_after = LLMProvider._extract_retry_after(msg)
|
|
|
|
status_code = getattr(e, "status_code", None)
|
|
if status_code is None and response is not None:
|
|
status_code = getattr(response, "status_code", None)
|
|
|
|
should_retry: bool | None = None
|
|
if headers is not None:
|
|
raw = headers.get("x-should-retry")
|
|
if isinstance(raw, str):
|
|
lowered = raw.strip().lower()
|
|
if lowered == "true":
|
|
should_retry = True
|
|
elif lowered == "false":
|
|
should_retry = False
|
|
|
|
error_kind: str | None = None
|
|
error_name = e.__class__.__name__.lower()
|
|
if "timeout" in error_name:
|
|
error_kind = "timeout"
|
|
elif "connection" in error_name:
|
|
error_kind = "connection"
|
|
error_type, error_code = LLMProvider._extract_error_type_code(payload)
|
|
|
|
return LLMResponse(
|
|
content=msg,
|
|
finish_reason="error",
|
|
retry_after=retry_after,
|
|
error_status_code=int(status_code) if status_code is not None else None,
|
|
error_kind=error_kind,
|
|
error_type=error_type,
|
|
error_code=error_code,
|
|
error_retry_after_s=retry_after,
|
|
error_should_retry=should_retry,
|
|
)
|
|
|
|
@staticmethod
|
|
def _strip_prefix(model: str) -> str:
|
|
if model.startswith("anthropic/"):
|
|
return model[len("anthropic/"):]
|
|
return model
|
|
|
|
# ------------------------------------------------------------------
|
|
# Message conversion: OpenAI chat format → Anthropic Messages API
|
|
# ------------------------------------------------------------------
|
|
|
|
def _convert_messages(
|
|
self, messages: list[dict[str, Any]],
|
|
) -> tuple[str | list[dict[str, Any]], list[dict[str, Any]]]:
|
|
"""Return ``(system, anthropic_messages)``."""
|
|
system: str | list[dict[str, Any]] = ""
|
|
raw: list[dict[str, Any]] = []
|
|
|
|
for msg in messages:
|
|
role = msg.get("role", "")
|
|
content = msg.get("content")
|
|
|
|
if role == "system":
|
|
system = content if isinstance(content, (str, list)) else str(content or "")
|
|
continue
|
|
|
|
if role == "tool":
|
|
block = self._tool_result_block(msg)
|
|
if raw and raw[-1]["role"] == "user":
|
|
prev_c = raw[-1]["content"]
|
|
if isinstance(prev_c, list):
|
|
prev_c.append(block)
|
|
else:
|
|
raw[-1]["content"] = [
|
|
{"type": "text", "text": prev_c or ""}, block,
|
|
]
|
|
else:
|
|
raw.append({"role": "user", "content": [block]})
|
|
continue
|
|
|
|
if role == "assistant":
|
|
raw.append({"role": "assistant", "content": self._assistant_blocks(msg)})
|
|
continue
|
|
|
|
if role == "user":
|
|
raw.append({
|
|
"role": "user",
|
|
"content": self._convert_user_content(content),
|
|
})
|
|
continue
|
|
|
|
return system, self._merge_consecutive(raw)
|
|
|
|
@staticmethod
|
|
def _tool_result_block(msg: dict[str, Any]) -> dict[str, Any]:
|
|
content = msg.get("content")
|
|
block: dict[str, Any] = {
|
|
"type": "tool_result",
|
|
"tool_use_id": msg.get("tool_call_id", ""),
|
|
}
|
|
if isinstance(content, list):
|
|
block["content"] = AnthropicProvider._convert_user_content(content)
|
|
elif isinstance(content, str):
|
|
block["content"] = content
|
|
else:
|
|
block["content"] = str(content) if content else ""
|
|
return block
|
|
|
|
@staticmethod
|
|
def _assistant_blocks(msg: dict[str, Any]) -> list[dict[str, Any]]:
|
|
blocks: list[dict[str, Any]] = []
|
|
content = msg.get("content")
|
|
|
|
for tb in msg.get("thinking_blocks") or []:
|
|
if isinstance(tb, dict) and tb.get("type") == "thinking":
|
|
blocks.append({
|
|
"type": "thinking",
|
|
"thinking": tb.get("thinking", ""),
|
|
"signature": tb.get("signature", ""),
|
|
})
|
|
|
|
if isinstance(content, str) and content:
|
|
blocks.append({"type": "text", "text": content})
|
|
elif isinstance(content, list):
|
|
for item in content:
|
|
blocks.append(item if isinstance(item, dict) else {"type": "text", "text": str(item)})
|
|
|
|
for tc in msg.get("tool_calls") or []:
|
|
if not isinstance(tc, dict):
|
|
continue
|
|
func = tc.get("function", {})
|
|
args = func.get("arguments", "{}")
|
|
if isinstance(args, str):
|
|
args = json_repair.loads(args)
|
|
blocks.append({
|
|
"type": "tool_use",
|
|
"id": tc.get("id") or _gen_tool_id(),
|
|
"name": func.get("name", ""),
|
|
"input": args,
|
|
})
|
|
|
|
return blocks or [{"type": "text", "text": ""}]
|
|
|
|
@staticmethod
|
|
def _convert_user_content(content: Any) -> Any:
|
|
"""Convert user message content, translating image_url blocks."""
|
|
if isinstance(content, str) or content is None:
|
|
return content or "(empty)"
|
|
if not isinstance(content, list):
|
|
return str(content)
|
|
|
|
result: list[dict[str, Any]] = []
|
|
for item in content:
|
|
if not isinstance(item, dict):
|
|
result.append({"type": "text", "text": str(item)})
|
|
continue
|
|
if item.get("type") == "image_url":
|
|
converted = AnthropicProvider._convert_image_block(item)
|
|
if converted:
|
|
result.append(converted)
|
|
continue
|
|
result.append(item)
|
|
return result or "(empty)"
|
|
|
|
@staticmethod
|
|
def _convert_image_block(block: dict[str, Any]) -> dict[str, Any] | None:
|
|
"""Convert OpenAI image_url block to Anthropic image block."""
|
|
url = (block.get("image_url") or {}).get("url", "")
|
|
if not url:
|
|
return None
|
|
m = re.match(r"data:(image/\w+);base64,(.+)", url, re.DOTALL)
|
|
if m:
|
|
return {
|
|
"type": "image",
|
|
"source": {"type": "base64", "media_type": m.group(1), "data": m.group(2)},
|
|
}
|
|
return {
|
|
"type": "image",
|
|
"source": {"type": "url", "url": url},
|
|
}
|
|
|
|
@staticmethod
|
|
def _has_tool_use(msg: dict[str, Any]) -> bool:
|
|
"""True if ``msg.content`` carries any ``tool_use`` block.
|
|
|
|
Anthropic forbids ``tool_use`` inside ``user`` turns, so messages that
|
|
issued a tool call cannot be safely rerouted when we patch the role.
|
|
"""
|
|
content = msg.get("content")
|
|
if not isinstance(content, list):
|
|
return False
|
|
return any(
|
|
isinstance(block, dict) and block.get("type") == "tool_use"
|
|
for block in content
|
|
)
|
|
|
|
@staticmethod
|
|
def _merge_consecutive(msgs: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Normalize a message sequence for Anthropic's ``/messages`` endpoint.
|
|
|
|
Anthropic's contract is stricter than OpenAI's:
|
|
|
|
1. Consecutive same-role turns must be collapsed into one.
|
|
2. The conversation cannot end with an ``assistant`` turn — Anthropic
|
|
does not support assistant-message prefill and returns 400.
|
|
3. The conversation cannot start with an ``assistant`` turn — the
|
|
first message must be ``user``.
|
|
|
|
Rules 2 and 3 mirror ``LLMProvider._enforce_role_alternation`` in
|
|
``base.py``, which applies the equivalent invariants to OpenAI-compat
|
|
providers. The only Anthropic-specific wrinkle: ``tool_use`` blocks
|
|
live inside ``content`` (not a separate ``tool_calls`` field) and are
|
|
invalid inside ``user`` turns, so the recovery paths below must skip
|
|
any message carrying them rather than silently producing a malformed
|
|
request.
|
|
"""
|
|
merged: list[dict[str, Any]] = []
|
|
for msg in msgs:
|
|
if merged and merged[-1]["role"] == msg["role"]:
|
|
prev_c = merged[-1]["content"]
|
|
cur_c = msg["content"]
|
|
if isinstance(prev_c, str):
|
|
prev_c = [{"type": "text", "text": prev_c}]
|
|
if isinstance(cur_c, str):
|
|
cur_c = [{"type": "text", "text": cur_c}]
|
|
if isinstance(cur_c, list):
|
|
prev_c.extend(cur_c)
|
|
merged[-1]["content"] = prev_c
|
|
else:
|
|
merged.append(msg)
|
|
|
|
# Rule 2: strip trailing assistant turns — Anthropic rejects prefill.
|
|
last_popped: dict[str, Any] | None = None
|
|
while merged and merged[-1].get("role") == "assistant":
|
|
last_popped = merged.pop()
|
|
|
|
# Recovery for rule 2: if stripping removed every turn, reroute the
|
|
# last popped assistant as a user turn so upstream code still gets a
|
|
# valid request instead of a secondary "messages array empty" 400.
|
|
# Skip when the message carried ``tool_use`` blocks (see _has_tool_use).
|
|
if (
|
|
not merged
|
|
and last_popped is not None
|
|
and not AnthropicProvider._has_tool_use(last_popped)
|
|
):
|
|
merged.append({"role": "user", "content": last_popped.get("content")})
|
|
|
|
# Rule 3: prepend a synthetic opener if the first surviving turn is an
|
|
# assistant (e.g. upstream history truncation dropped the original
|
|
# user request). ``tool_use``-carrying assistants are left alone —
|
|
# that message will still fail validation, but injecting an opener
|
|
# before it would orphan the tool_use/tool_result pair that follows,
|
|
# turning a recoverable 400 into a harder-to-diagnose one.
|
|
if (
|
|
merged
|
|
and merged[0].get("role") == "assistant"
|
|
and not AnthropicProvider._has_tool_use(merged[0])
|
|
):
|
|
merged.insert(0, {"role": "user", "content": "(conversation continued)"})
|
|
|
|
return merged
|
|
|
|
# ------------------------------------------------------------------
|
|
# Tool definition conversion
|
|
# ------------------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def _convert_tools(tools: list[dict[str, Any]] | None) -> list[dict[str, Any]] | None:
|
|
if not tools:
|
|
return None
|
|
result = []
|
|
for tool in tools:
|
|
func = tool.get("function", tool)
|
|
entry: dict[str, Any] = {
|
|
"name": func.get("name", ""),
|
|
"input_schema": func.get("parameters", {"type": "object", "properties": {}}),
|
|
}
|
|
desc = func.get("description")
|
|
if desc:
|
|
entry["description"] = desc
|
|
if "cache_control" in tool:
|
|
entry["cache_control"] = tool["cache_control"]
|
|
result.append(entry)
|
|
return result
|
|
|
|
@staticmethod
|
|
def _convert_tool_choice(
|
|
tool_choice: str | dict[str, Any] | None,
|
|
thinking_enabled: bool = False,
|
|
) -> dict[str, Any] | None:
|
|
if thinking_enabled:
|
|
return {"type": "auto"}
|
|
if tool_choice is None or tool_choice == "auto":
|
|
return {"type": "auto"}
|
|
if tool_choice == "required":
|
|
return {"type": "any"}
|
|
if tool_choice == "none":
|
|
return None
|
|
if isinstance(tool_choice, dict):
|
|
name = tool_choice.get("function", {}).get("name")
|
|
if name:
|
|
return {"type": "tool", "name": name}
|
|
return {"type": "auto"}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Prompt caching
|
|
# ------------------------------------------------------------------
|
|
|
|
@classmethod
|
|
def _apply_cache_control(
|
|
cls,
|
|
system: str | list[dict[str, Any]],
|
|
messages: list[dict[str, Any]],
|
|
tools: list[dict[str, Any]] | None,
|
|
) -> tuple[str | list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]] | None]:
|
|
marker = {"type": "ephemeral"}
|
|
|
|
if isinstance(system, str) and system:
|
|
system = [{"type": "text", "text": system, "cache_control": marker}]
|
|
elif isinstance(system, list) and system:
|
|
system = list(system)
|
|
system[-1] = {**system[-1], "cache_control": marker}
|
|
|
|
new_msgs = list(messages)
|
|
if len(new_msgs) >= 3:
|
|
m = new_msgs[-2]
|
|
c = m.get("content")
|
|
if isinstance(c, str):
|
|
new_msgs[-2] = {**m, "content": [{"type": "text", "text": c, "cache_control": marker}]}
|
|
elif isinstance(c, list) and c:
|
|
nc = list(c)
|
|
nc[-1] = {**nc[-1], "cache_control": marker}
|
|
new_msgs[-2] = {**m, "content": nc}
|
|
|
|
new_tools = tools
|
|
if tools:
|
|
new_tools = list(tools)
|
|
for idx in cls._tool_cache_marker_indices(new_tools):
|
|
new_tools[idx] = {**new_tools[idx], "cache_control": marker}
|
|
|
|
return system, new_msgs, new_tools
|
|
|
|
# ------------------------------------------------------------------
|
|
# Build API kwargs
|
|
# ------------------------------------------------------------------
|
|
|
|
def _build_kwargs(
|
|
self,
|
|
messages: list[dict[str, Any]],
|
|
tools: list[dict[str, Any]] | None,
|
|
model: str | None,
|
|
max_tokens: int,
|
|
temperature: float,
|
|
reasoning_effort: str | None,
|
|
tool_choice: str | dict[str, Any] | None,
|
|
supports_caching: bool = True,
|
|
) -> dict[str, Any]:
|
|
model_name = self._strip_prefix(model or self.default_model)
|
|
system, anthropic_msgs = self._convert_messages(self._sanitize_empty_content(messages))
|
|
anthropic_tools = self._convert_tools(tools)
|
|
|
|
if supports_caching:
|
|
system, anthropic_msgs, anthropic_tools = self._apply_cache_control(
|
|
system, anthropic_msgs, anthropic_tools,
|
|
)
|
|
|
|
max_tokens = max(1, max_tokens)
|
|
thinking_enabled = bool(reasoning_effort) and reasoning_effort.lower() != "none"
|
|
|
|
# claude-opus-4-7 deprecated the `temperature` parameter entirely — the
|
|
# API returns 400 if it is present, on any code path.
|
|
omit_temperature = "opus-4-7" in model_name
|
|
|
|
kwargs: dict[str, Any] = {
|
|
"model": model_name,
|
|
"messages": anthropic_msgs,
|
|
"max_tokens": max_tokens,
|
|
}
|
|
|
|
if system:
|
|
kwargs["system"] = system
|
|
|
|
if reasoning_effort == "adaptive":
|
|
# Adaptive thinking: model decides when and how much to think
|
|
# Supported on claude-sonnet-4-6 and claude-opus-4-6.
|
|
# Also auto-enables interleaved thinking between tool calls.
|
|
kwargs["thinking"] = {"type": "adaptive"}
|
|
if not omit_temperature:
|
|
kwargs["temperature"] = 1.0
|
|
elif thinking_enabled:
|
|
budget_map = {"low": 1024, "medium": 4096, "high": max(8192, max_tokens)}
|
|
budget = budget_map.get(reasoning_effort.lower(), 4096)
|
|
kwargs["thinking"] = {"type": "enabled", "budget_tokens": budget}
|
|
kwargs["max_tokens"] = max(max_tokens, budget + 4096)
|
|
if not omit_temperature:
|
|
kwargs["temperature"] = 1.0
|
|
elif not omit_temperature:
|
|
kwargs["temperature"] = temperature
|
|
|
|
if anthropic_tools:
|
|
kwargs["tools"] = anthropic_tools
|
|
tc = self._convert_tool_choice(tool_choice, thinking_enabled)
|
|
if tc:
|
|
kwargs["tool_choice"] = tc
|
|
|
|
if self.extra_headers:
|
|
kwargs["extra_headers"] = self.extra_headers
|
|
|
|
return kwargs
|
|
|
|
# ------------------------------------------------------------------
|
|
# Response parsing
|
|
# ------------------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def _parse_response(response: Any) -> LLMResponse:
|
|
content_parts: list[str] = []
|
|
tool_calls: list[ToolCallRequest] = []
|
|
thinking_blocks: list[dict[str, Any]] = []
|
|
|
|
for block in response.content:
|
|
if block.type == "text":
|
|
content_parts.append(block.text)
|
|
elif block.type == "tool_use":
|
|
tool_calls.append(ToolCallRequest(
|
|
id=block.id,
|
|
name=block.name,
|
|
arguments=block.input if isinstance(block.input, dict) else {},
|
|
))
|
|
elif block.type == "thinking":
|
|
thinking_blocks.append({
|
|
"type": "thinking",
|
|
"thinking": block.thinking,
|
|
"signature": getattr(block, "signature", ""),
|
|
})
|
|
|
|
stop_map = {"tool_use": "tool_calls", "end_turn": "stop", "max_tokens": "length"}
|
|
finish_reason = stop_map.get(response.stop_reason or "", response.stop_reason or "stop")
|
|
|
|
usage: dict[str, int] = {}
|
|
if response.usage:
|
|
input_tokens = response.usage.input_tokens
|
|
cache_creation = getattr(response.usage, "cache_creation_input_tokens", 0) or 0
|
|
cache_read = getattr(response.usage, "cache_read_input_tokens", 0) or 0
|
|
total_prompt_tokens = input_tokens + cache_creation + cache_read
|
|
usage = {
|
|
"prompt_tokens": total_prompt_tokens,
|
|
"completion_tokens": response.usage.output_tokens,
|
|
"total_tokens": total_prompt_tokens + response.usage.output_tokens,
|
|
}
|
|
for attr in ("cache_creation_input_tokens", "cache_read_input_tokens"):
|
|
val = getattr(response.usage, attr, 0)
|
|
if val:
|
|
usage[attr] = val
|
|
# Normalize to cached_tokens for downstream consistency.
|
|
if cache_read:
|
|
usage["cached_tokens"] = cache_read
|
|
|
|
return LLMResponse(
|
|
content="".join(content_parts) or None,
|
|
tool_calls=tool_calls,
|
|
finish_reason=finish_reason,
|
|
usage=usage,
|
|
thinking_blocks=thinking_blocks or None,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public API
|
|
# ------------------------------------------------------------------
|
|
|
|
async def chat(
|
|
self,
|
|
messages: list[dict[str, Any]],
|
|
tools: list[dict[str, Any]] | None = None,
|
|
model: str | None = None,
|
|
max_tokens: int = 4096,
|
|
temperature: float = 0.7,
|
|
reasoning_effort: str | None = None,
|
|
tool_choice: str | dict[str, Any] | None = None,
|
|
) -> LLMResponse:
|
|
kwargs = self._build_kwargs(
|
|
messages, tools, model, max_tokens, temperature,
|
|
reasoning_effort, tool_choice,
|
|
)
|
|
try:
|
|
response = await self._client.messages.create(**kwargs)
|
|
return self._parse_response(response)
|
|
except Exception as e:
|
|
return self._handle_error(e)
|
|
|
|
async def chat_stream(
|
|
self,
|
|
messages: list[dict[str, Any]],
|
|
tools: list[dict[str, Any]] | None = None,
|
|
model: str | None = None,
|
|
max_tokens: int = 4096,
|
|
temperature: float = 0.7,
|
|
reasoning_effort: str | None = None,
|
|
tool_choice: str | dict[str, Any] | None = None,
|
|
on_content_delta: Callable[[str], Awaitable[None]] | None = None,
|
|
) -> LLMResponse:
|
|
kwargs = self._build_kwargs(
|
|
messages, tools, model, max_tokens, temperature,
|
|
reasoning_effort, tool_choice,
|
|
)
|
|
idle_timeout_s = int(os.environ.get("NANOBOT_STREAM_IDLE_TIMEOUT_S", "90"))
|
|
try:
|
|
async with self._client.messages.stream(**kwargs) as stream:
|
|
if on_content_delta:
|
|
stream_iter = stream.text_stream.__aiter__()
|
|
while True:
|
|
try:
|
|
text = await asyncio.wait_for(
|
|
stream_iter.__anext__(),
|
|
timeout=idle_timeout_s,
|
|
)
|
|
except StopAsyncIteration:
|
|
break
|
|
await on_content_delta(text)
|
|
response = await asyncio.wait_for(
|
|
stream.get_final_message(),
|
|
timeout=idle_timeout_s,
|
|
)
|
|
return self._parse_response(response)
|
|
except asyncio.TimeoutError:
|
|
return LLMResponse(
|
|
content=(
|
|
f"Error calling LLM: stream stalled for more than "
|
|
f"{idle_timeout_s} seconds"
|
|
),
|
|
finish_reason="error",
|
|
error_kind="timeout",
|
|
)
|
|
except Exception as e:
|
|
return self._handle_error(e)
|
|
|
|
def get_default_model(self) -> str:
|
|
return self.default_model
|