执行摘要
- 一句话：修复并增强 Anthropic /v1/messages 兼容性
- 推荐动作：此 PR 价值极高，建议负责 API 兼容层的开发者精读。重点关注：
- streaming 生成器中 content_block 状态管理设计（serving.py 的 _ensure_content_block_events）
- 思考参数在 Anthropic 与 OpenAI 之间的映射机制（serving_chat.py 的 apply_reasoning_enabled）
- 协议模型从普通 Pydantic 模型向 discriminated union 的迁移方式（protocol.py）
- 测试用例的构造模式，尤其是如何用 fake serving 模拟底层行为。
功能与动机
Anthropic /v1/messages 端点存在多处兼容性问题：流式输出中 text_delta 可能误打到 tool_use 块索引；reasoning_content 未被映射为 thinking 块；缓存 token 重复计入 input_tokens；web_search 内置工具因缺少 input_schema 导致 422 验证错误；OpenAI handler 错误被吞掉返回通用 500。这些问题导致 Claude Code、anthropic-sdk-python 等客户端无法正常使用 SGLang 作为后端。此 PR 旨在系统性修复这些缺陷，使 Anthropic 兼容层达到可投产水平。
实现拆解
- 协议模型重构（protocol.py）：将原有单一的 AnthropicContentBlock 拆分为 TextBlock、ToolUseBlock、ThinkingBlock 等带歧视联合的独立模型，采用 Pydantic Discriminator 实现类型安全解析。新增 ThinkingDelta、SignatureDelta 等流式事件类型。
- 服务层核心逻辑重写（serving.py）：重写 _generate_anthropic_stream 流式生成器，通过 _ensure_content_block_events 管理块打开/关闭状态，确保 text_delta/thinking_delta/input_json_delta 路由到正确的块索引。新增 _anthropic_usage_from_openai 函数统一处理 token 用量转换（减去缓存 token），新增 _convert_assistant_thinking_blocks 将 prior-turn thinking 历史重包装为 标签。接入 continuous_usage_stats 以获取实时流式用量。
- OpenAI 侧新接口（serving_chat.py）：新增 wrap_reasoning_history、_reasoning_default_mode、_get_reasoning_toggle_param、apply_reasoning_enabled 四个方法，使 Anthropic 层能通过统一接口控制模型的思考模式，并保证与 _get_reasoning_from_request 的读取逻辑一致。
- HTTP 错误处理适配（http_server.py）：针对 /v1/messages 路径的 RequestValidationError 和 HTTPException，返回 Anthropic 格式的 {"type":"error","error":{...}} 响应，并利用 _scrub_error_message 清洗 5xx 内部错误细节。
- 测试覆盖（test_serving.py）：新增 1300+ 行单元测试，涵盖工具流块切换、思考流事件序列、缓存用量、web_search 展平、错误转发、thinking 参数多回合历史等场景，所有测试通过 CI（base-a-test-cpu）。
关键文件：
python/sglang/srt/entrypoints/anthropic/serving.py（模块 适配层；类别 source；类型 core-logic；符号 _cached_prompt_tokens, _anthropic_input_tokens, _anthropic_usage_from_openai, _scrub_error_message）: 核心服务层，实现 Anthropic 请求到 OpenAI 的转换、流式响应生成、错误处理、token 用量计算等关键逻辑。改动量最大（+843/-208）。python/sglang/srt/entrypoints/anthropic/protocol.py（模块 协议模型；类别 source；类型 data-contract；符号 AnthropicContentBlock, TextBlock, ImageBlock, ToolUseBlock）: 协议模型文件，将原本单一的 AnthropicContentBlock 拆分为 TextBlock、ToolUseBlock、ThinkingBlock 等 discriminated union 模型，新增 WebSearchTool、ThinkingDelta、SignatureDelta 等类型。是保障类型安全的基础。python/sglang/srt/entrypoints/openai/serving_chat.py（模块 推理控制；类别 source；类型 core-logic；符号 wrap_reasoning_history, _reasoning_default_mode, _get_reasoning_toggle_param, apply_reasoning_enabled）: OpenAI 端适配接口，新增 wrap_reasoning_history、apply_reasoning_enabled 等方法，使 Anthropic 层能与底层聊天模板和推理检测器正确交互，是 thinking 参数映射的关键依赖。test/registered/unit/entrypoints/anthropic/test_serving.py（模块 测试；类别 test；类型 test-coverage；符号 _FakeOpenAIServingChat, _generate_chat_stream, apply_reasoning_enabled, wrap_reasoning_history）: 全新的 1339 行单元测试，覆盖了工具流块切换、思考流事件序列、缓存用量、search_result 展平、错误转发、thinking 参数等多回合场景，使用 fake OpenAI 服务模拟底层行为。python/sglang/srt/entrypoints/http_server.py（模块 错误处理；类别 source；类型 core-logic；符号 _anthropic_validation_message, _anthropic_error_response）: 修改通用 HTTP 错误处理器，对 /v1/messages 路径返回 Anthropic 格式的错误响应，并清洗 5xx 内部细节。新增 _anthropic_validation_message 和 _anthropic_error_response 辅助函数。
关键符号：_cached_prompt_tokens, _anthropic_input_tokens, _anthropic_usage_from_openai, _scrub_error_message, _text_from_search_result, _convert_assistant_thinking_blocks, _emit_user_message, _message_start_event, wrap_reasoning_history, _reasoning_default_mode, _get_reasoning_toggle_param, apply_reasoning_enabled, _anthropic_validation_message, _anthropic_error_response
关键源码片段
python/sglang/srt/entrypoints/anthropic/serving.py
核心服务层，实现 Anthropic 请求到 OpenAI 的转换、流式响应生成、错误处理、token 用量计算等关键逻辑。改动量最大（+843/-208）。
def _cached_prompt_tokens(usage) -> int:
# 从 OpenAI usage 对象中提取缓存 token 数量
prompt_tokens_details = getattr(usage, 'prompt_tokens_details', None)
return getattr(prompt_tokens_details, 'cached_tokens', 0) or 0
def _anthropic_input_tokens(usage) -> int:
# 计算 Anthropic 可见的 input_tokens：prompt_tokens 减去缓存命中部分，
# 避免双倍计费。若缓存超过 prompt（上游 telemetry 异常），钳位为 0 并记录 warning。
prompt = getattr(usage, 'prompt_tokens', 0) or 0
cached = _cached_prompt_tokens(usage)
if cached > prompt:
logger.warning(
'Cached tokens (%d) exceed prompt tokens (%d); clamping '
'input_tokens to 0. This usually indicates an upstream '
'telemetry bug.',
cached,
prompt,
)
return max(prompt - cached, 0)
def _anthropic_usage_from_openai(
usage,
*,
include_input: bool,
include_output: bool,
force_zero_output: bool = False,
) -> AnthropicUsage:
# 将 OpenAI usage 转换为 AnthropicUsage，可选包含 input/output。
# 流式 message_delta 事件应省略 input_tokens（仅 output_tokens），
# 非流式响应两者都包含。force_zero_output 用于 streaming 首条 chunk。
if usage is None:
return AnthropicUsage(
input_tokens=0 if include_input else None,
output_tokens=0 if include_output else None,
)
usage_fields: dict[str, int] = {}
cached_tokens = _cached_prompt_tokens(usage)
if include_input:
usage_fields['input_tokens'] = _anthropic_input_tokens(usage)
if cached_tokens:
usage_fields['cache_read_input_tokens'] = cached_tokens
if include_output:
usage_fields['output_tokens'] = (
0 if force_zero_output else (getattr(usage, 'completion_tokens', 0) or 0)
)
return AnthropicUsage(**usage_fields)
python/sglang/srt/entrypoints/anthropic/protocol.py
协议模型文件，将原本单一的 AnthropicContentBlock 拆分为 TextBlock、ToolUseBlock、ThinkingBlock 等 discriminated union 模型，新增 WebSearchTool、ThinkingDelta、SignatureDelta 等类型。是保障类型安全的基础。
# 使用 Pyadntic Discriminator 实现 discriminated union
# 每个变体只携带自身使用的字段，而非一个包含所有可选字段的大模型
from typing import Annotated, Literal, Optional, Union
from pydantic import BaseModel, Field, Tag
class TextBlock(BaseModel):
type: Literal['text'] = 'text'
text: str
class ToolUseBlock(BaseModel):
type: Literal['tool_use'] = 'tool_use'
id: str
name: str
input: dict[str, Any] = Field(default_factory=dict)
class ThinkingBlock(BaseModel):
type: Literal['thinking'] = 'thinking'
thinking: str
signature: Optional[str] = None
class RedactedThinkingBlock(BaseModel):
type: Literal['redacted_thinking'] = 'redacted_thinking'
data: Optional[str] = None
AnthropicContentBlock = Annotated[
Union[
TextBlock, ImageBlock, ToolUseBlock, ToolResultBlock,
ToolReferenceBlock, SearchResultBlock, ThinkingBlock,
RedactedThinkingBlock,
],
Field(discriminator='type'),
]
python/sglang/srt/entrypoints/openai/serving_chat.py
OpenAI 端适配接口，新增 wrap_reasoning_history、apply_reasoning_enabled 等方法，使 Anthropic 层能与底层聊天模板和推理检测器正确交互，是 thinking 参数映射的关键依赖。
def apply_reasoning_enabled(
self, request: ChatCompletionRequest, enabled: bool
) -> None:
"""强制设置请求的推理开启/关闭状态，与 _get_reasoning_from_request 保持对称。"""
if not self.reasoning_parser:
if enabled:
raise ValueError(
'Anthropic thinking is not supported for models without '
'a reasoning parser'
)
return
# Hunyuan 用 reasoning_effort=medium/no_think
if self.reasoning_parser == 'hunyuan':
request.reasoning_effort = 'medium' if enabled else 'no_think'
return
# Mistral 用 reasoning_effort=medium/none
config = self.template_manager.reasoning_config
is_mistral = (config is not None and config.special_case == 'mistral') or (
config is None and self._reasoning_default_mode() == 'mistral'
)
if is_mistral:
request.reasoning_effort = 'medium' if enabled else 'none'
return
# 始终开启的模型无法关闭
is_always_on = (config is not None and config.special_case == 'always') or (
config is None and self._reasoning_default_mode() == 'always'
)
if is_always_on:
if not enabled:
raise ValueError(
f'Reasoning parser "{self.reasoning_parser}" is always-on '
f'and cannot be disabled via Anthropic thinking'
)
return
# 通用 toggle_param 方式：设置 request.chat_template_kwargs
toggle_param = self._get_reasoning_toggle_param()
request.chat_template_kwargs = request.chat_template_kwargs or {}
request.chat_template_kwargs[toggle_param] = enabled
评论区精华
在 Code Review 中，gemini-code-assist[bot] 指出基于 name.startswith('web_search') 检测内置工具可能过于宽泛，若用户自定义工具名以同样前缀开头会有冲突。作者 JustinTong0323 回应已采用 Pydantic 歧视联合体，通过
我们已从名称检测切换为基于 type 字段的 Pydantic 歧视联合体。类型模式 ^web_search_\d{8}$ 也在 Pydantic Field(pattern=...) 中强制执行，只有符合日期格式的 Anthropic 类型才会匹配。同时，自定义工具即使 name 以 web_search 开头，只要其 type 不为空或为 'custom'，就不会被误判为内置工具。
该决策已被接受，未产生额外修改。
- 内置工具检测方式：基于 name 前缀 vs 基于 type 字段 (correctness): 作者已改为使用 Pydantic discriminated union，通过 type 字段的模式匹配（^web_search_\d{8}$）识别内置工具，同时自定义工具即使 name 以 web_search 开头，只要 type 不为空或为 'custom'，就不会被误判。
风险与影响
关联脉络
- PR #24294 fix(anthropic): track block type and surface reasoning_content as thinking (#24293): 被整合的 PR：流式内容块类型追踪和 reasoning_content 映射为 thinking 块。
- PR #20830 fix: populate input_tokens in Anthropic streaming message_start: 被整合的 PR：在 message_start 中提供真实的 input_tokens。
- PR #20856 fix: populate cache_read_input_tokens in Anthropic /v1/messages responses: 被整合的 PR：将缓存 token 映射为 cache_read_input_tokens。
- PR #22030 [fix] Anthropic /v1/messages endpoint: forward original error instead of returning generic 500: 被整合的 PR：转发原始错误而非返回通用 500。
- PR #23024 fix: support Anthropic web_search built-in tools in /v1/messages: 被整合的 PR：支持 web_search 内置工具和 search_result 块。
- PR #19334 [Anthropic] Support thinking/reasoning tokens in /v1/messages: 被整合的 PR：thinking 参数支持的基础提案。
- PR #21902 feat(anthropic): add thinking/reasoning support and cache token usage to /v1/messages: 被整合的 PR：综合 thinking + cache token 用量，多轮思考历史重构模式。
- PR #22061 [Anthropic] Add thinking support for /v1/messages: 被整合的 PR：AnthropicThinkingParam、apply_reasoning_enabled 分解、count_tokens thinking 传播。
- PR #22135 [Anthropic] Support thinking/reasoning tokens in /v1/messages: 被整合的 PR：独立的 thinking/reasoning 支持方案，为合并设计提供了信息。
- PR #22375 fix(anthropic): use monotonic_time in Anthropic API entrypoint for accurate Prometheus metrics: 被整合的 PR：清理已死的 received_time_perf/validation_time 计时重复代码。
- PR #25271 feat(anthropic): add Claude 4.7 adaptive thinking support: 后续 PR：在基本 thinking 支持之上扩展 adaptive thinking 等特性（未包含在本 PR 中）。
参与讨论