fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-18 10:00:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(anthropic): Anthropic API tool_choice schema conversion (#8328)

Browse Source 
* fix(anthropic): 修复 Anthropic API tool_choice 格式转换及参数支持

- 将 tool_choice 从简单的 auto/required 逻辑改为遵循 Anthropic API 规范，支持 auto/any/none/tool 四种原生值
- 兼容 OpenAI 风格的 tool_choice="required"，自动映射为 {"type": "any"}
- 允许直接传入 dict 类型的 tool_choice 以实现指定工具调用
- 更新 text_chat 和 stream_chat 入口的参数类型标注，扩大可接收的 tool_choice 类型
- 新增 tool_choice 格式转换的单元测试，覆盖各类输入场景

Closes #8319

* Clean up test cases and remove unused mocks

Removed unused mock classes and tests for tool_choice conversion.

* fix(anthropic): 修复 Anthropic API tool_choice="tool" 参数处理及重构格式转换逻辑

- 提取静态方法 _normalize_tool_choice 统一处理 tool_choice 格式转换，消除重复代码
- 处理字符串 "tool" 值时，因无法指定具体工具名而回退为 auto 并记录警告，避免无效请求
- 在 _query 和 _stream_query 中采用默认值 auto 并应用规范化逻辑，确保一致性

* test(anthropic): 添加空工具集时跳过工具参数设置的测试

- 新增 _EmptyToolSet 模拟类，模拟无工具场景
- 新增测试用例 test_tool_choice_empty_tool_list_skips_tool_choice
- 验证当 ToolSet 存在但工具列表为空时，请求不包含 tools 和 tool_choice 参数
- 完善边缘情况测试覆盖，确保与现有逻辑一致

* style: ruff 格式化一下

---------

Co-authored-by: Weilong Liao <37870767+Soulter@users.noreply.github.com>
This commit is contained in:
NayukiChiba
2026-05-30 13:44:45 +08:00
committed by GitHub
parent 01a47b8360
commit 8353fe1608
2 changed files with 238 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
astrbot/core/provider/sources/anthropic_source.py
View File

@@ -318,15 +318,44 @@ class ProviderAnthropic(Provider):
if usage.output_tokens is not None:
token_usage.output = usage.output_tokens
@staticmethod
def _normalize_tool_choice(tool_choice) -> dict:
"""将 tool_choice 转换为 Anthropic API 要求的格式
参考: https://platform.claude.com/docs/en/agents-and-tools/tool-use/define-tools#controlling-claudes-output
Args:
tool_choice: 原始 tool_choice 值，支持 str 或 dict
Returns:
Anthropic API 格式的 tool_choice 字典
"""
if isinstance(tool_choice, dict):
return tool_choice
if tool_choice == "required":
# 兼容 OpenAI 命名：required → any
return {"type": "any"}
if tool_choice in ("auto", "any", "none"):
return {"type": tool_choice}
if tool_choice == "tool":
# {"type": "tool"} 必须配合 name 字段指定具体工具
# 纯字符串 "tool" 无法指定工具名，回退为 auto
logger.warning("tool_choice='tool' 无法指定工具名，已回退为 'auto'")
return {"type": "auto"}
logger.warning(f"未知的 tool_choice 值: {tool_choice}，已回退为 'auto'")
return {"type": "auto"}
async def _query(self, payloads: dict, tools: ToolSet | None) -> LLMResponse:
if tools:
if tool_list := tools.get_func_desc_anthropic_style():
payloads["tools"] = tool_list
payloads["tool_choice"] = {
"type": "any"
if payloads.get("tool_choice") == "required"
else "auto"
}
payloads["tool_choice"] = self._normalize_tool_choice(
payloads.get("tool_choice", "auto")
)
extra_body = self.provider_config.get("custom_extra_body", {})
@@ -409,11 +438,9 @@ class ProviderAnthropic(Provider):
if tools:
if tool_list := tools.get_func_desc_anthropic_style():
payloads["tools"] = tool_list
payloads["tool_choice"] = {
"type": "any"
if payloads.get("tool_choice") == "required"
else "auto"
}
payloads["tool_choice"] = self._normalize_tool_choice(
payloads.get("tool_choice", "auto")
)
# 用于累积工具调用信息
tool_use_buffer = {}
@@ -569,7 +596,7 @@ class ProviderAnthropic(Provider):
tool_calls_result=None,
model=None,
extra_user_content_parts=None,
tool_choice: Literal["auto", "required"] = "auto",
tool_choice: Literal["auto", "any", "tool", "none"] | dict[str, str] = "auto",
**kwargs,
) -> LLMResponse:
if contexts is None:
@@ -598,8 +625,8 @@ class ProviderAnthropic(Provider):
if not isinstance(tool_calls_result, list):
context_query.extend(tool_calls_result.to_openai_messages())
else:
for tcr in tool_calls_result:
context_query.extend(tcr.to_openai_messages())
for tool_call_result in tool_calls_result:
context_query.extend(tool_call_result.to_openai_messages())
system_prompt, new_messages = self._prepare_payload(context_query)
@@ -637,7 +664,7 @@ class ProviderAnthropic(Provider):
tool_calls_result=None,
model=None,
extra_user_content_parts=None,
tool_choice: Literal["auto", "required"] = "auto",
tool_choice: Literal["auto", "any", "tool", "none"] | dict[str, str] = "auto",
**kwargs,
):
if contexts is None:
@@ -665,8 +692,8 @@ class ProviderAnthropic(Provider):
if not isinstance(tool_calls_result, list):
context_query.extend(tool_calls_result.to_openai_messages())
else:
for tcr in tool_calls_result:
context_query.extend(tcr.to_openai_messages())
for tool_call_result in tool_calls_result:
context_query.extend(tool_call_result.to_openai_messages())
system_prompt, new_messages = self._prepare_payload(context_query)

195
tests/test_anthropic_kimi_code_provider.py
View File

@@ -416,3 +416,198 @@ def test_prepare_payload_does_not_merge_non_consecutive_tool_results():
],
},
]
# ---- tool_choice 转换测试 ----
class _FakeToolSet:
"""模拟包含工具的 ToolSet"""
def get_func_desc_anthropic_style(self):
return [{"name": "get_weather", "description": "Get weather"}]
def empty(self):
return False
class _EmptyToolSet:
"""模拟空工具列表的 ToolSet，用于验证无工具时不设置 tool_choice"""
def get_func_desc_anthropic_style(self):
return []
def empty(self):
return True
class _FakeMessages:
"""模拟 AsyncAnthropic.messages 命名空间"""
async def _capture_payloads_create(**kwargs):
"""捕获 payloads 并返回一个真实的 Message 实例"""
from anthropic.types import Message, TextBlock, Usage
_capture_payloads_create.last_kwargs = kwargs
return Message(
id="msg_fake",
content=[TextBlock(type="text", text="Hello")],
model="claude-test",
role="assistant",
stop_reason=None,
stop_sequence=None,
type="message",
usage=Usage(input_tokens=10, output_tokens=5),
)
def _setup_provider_with_mock_client(monkeypatch) -> anthropic_source.ProviderAnthropic:
"""创建 provider 并 mock 底层 API 调用"""
monkeypatch.setattr(anthropic_source, "AsyncAnthropic", _FakeAsyncAnthropic)
provider = anthropic_source.ProviderAnthropic(
provider_config={
"id": "anthropic-test",
"type": "anthropic_chat_completion",
"model": "claude-test",
"key": ["test-key"],
},
provider_settings={},
)
fakeMessages = _FakeMessages()
fakeMessages.create = _capture_payloads_create
provider.client.messages = fakeMessages
return provider
@pytest.mark.asyncio
async def test_tool_choice_auto_converts_to_dict(monkeypatch):
"""tool_choice='auto' 应转换为 {'type': 'auto'}"""
provider = _setup_provider_with_mock_client(monkeypatch)
await provider.text_chat(
prompt="hello",
func_tool=_FakeToolSet(),
tool_choice="auto",
)
assert _capture_payloads_create.last_kwargs["tool_choice"] == {"type": "auto"}
@pytest.mark.asyncio
async def test_tool_choice_any_converts_to_dict(monkeypatch):
"""tool_choice='any' 应转换为 {'type': 'any'}"""
provider = _setup_provider_with_mock_client(monkeypatch)
await provider.text_chat(
prompt="hello",
func_tool=_FakeToolSet(),
tool_choice="any",
)
assert _capture_payloads_create.last_kwargs["tool_choice"] == {"type": "any"}
@pytest.mark.asyncio
async def test_tool_choice_none_converts_to_dict(monkeypatch):
"""tool_choice='none' 应转换为 {'type': 'none'}"""
provider = _setup_provider_with_mock_client(monkeypatch)
await provider.text_chat(
prompt="hello",
func_tool=_FakeToolSet(),
tool_choice="none",
)
assert _capture_payloads_create.last_kwargs["tool_choice"] == {"type": "none"}
@pytest.mark.asyncio
async def test_tool_choice_required_legacy_compat(monkeypatch):
"""tool_choice='required'(OpenAI 命名) 应兼容转换为 {'type': 'any'}"""
provider = _setup_provider_with_mock_client(monkeypatch)
await provider.text_chat(
prompt="hello",
func_tool=_FakeToolSet(),
tool_choice="required",
)
assert _capture_payloads_create.last_kwargs["tool_choice"] == {"type": "any"}
@pytest.mark.asyncio
async def test_tool_choice_dict_passthrough(monkeypatch):
"""tool_choice 为 dict 时应直接透传"""
provider = _setup_provider_with_mock_client(monkeypatch)
await provider.text_chat(
prompt="hello",
func_tool=_FakeToolSet(),
tool_choice={"type": "tool", "name": "get_weather"},
)
assert _capture_payloads_create.last_kwargs["tool_choice"] == {
"type": "tool",
"name": "get_weather",
}
@pytest.mark.asyncio
async def test_tool_choice_default_when_not_set(monkeypatch):
"""未传 tool_choice 时，默认应为 {'type': 'auto'}"""
provider = _setup_provider_with_mock_client(monkeypatch)
await provider.text_chat(
prompt="hello",
func_tool=_FakeToolSet(),
)
assert _capture_payloads_create.last_kwargs["tool_choice"] == {"type": "auto"}
@pytest.mark.asyncio
async def test_tool_choice_invalid_string_falls_back_to_auto(monkeypatch):
"""无效的 tool_choice 字符串应回退为 {'type': 'auto'}"""
provider = _setup_provider_with_mock_client(monkeypatch)
await provider.text_chat(
prompt="hello",
func_tool=_FakeToolSet(),
tool_choice="invalid_value",
)
assert _capture_payloads_create.last_kwargs["tool_choice"] == {"type": "auto"}
@pytest.mark.asyncio
async def test_tool_choice_no_tools_skips_tool_choice(monkeypatch):
"""无工具时不应设置 tool_choice"""
provider = _setup_provider_with_mock_client(monkeypatch)
await provider.text_chat(
prompt="hello",
func_tool=None,
tool_choice="any",
)
assert "tool_choice" not in _capture_payloads_create.last_kwargs
@pytest.mark.asyncio
async def test_tool_choice_empty_tool_list_skips_tool_choice(monkeypatch):
"""ToolSet 存在但工具列表为空时，不应设置 tools 和 tool_choice"""
provider = _setup_provider_with_mock_client(monkeypatch)
await provider.text_chat(
prompt="hello",
func_tool=_EmptyToolSet(),
tool_choice="any",
)
kwargs = _capture_payloads_create.last_kwargs
assert "tools" not in kwargs
assert "tool_choice" not in kwargs