fix(bedrock): handle additional Nexus text tool call formats

This commit is contained in:
GitHub Copilot 2026-05-22 21:22:45 +08:00
parent 26f3216db8
commit bd6388b2c3
2 changed files with 155 additions and 0 deletions

View file

@ -5,6 +5,7 @@ Translating between OpenAI's `/chat/completion` format and Amazon's `/converse`
import copy
import json
import re
import ast
import time
import types
from typing import Any, List, Literal, Optional, Tuple, Union, cast, overload
@ -1863,6 +1864,78 @@ class AmazonConverseConfig(BaseConfig):
)
)
tool_call_match = re.search(
r"<tool_call>\s*(.*?)\s*</tool_call>",
content,
flags=re.DOTALL | re.IGNORECASE,
)
if tool_call_match is not None:
parsed_tool_call = self._parse_json_string_if_possible(
tool_call_match.group(1)
)
if not isinstance(parsed_tool_call, dict):
return None
resolved_tool_name = self._resolve_nexus_tool_name(
cast(Optional[str], parsed_tool_call.get("name")), tool_call_names
)
if resolved_tool_name is None:
return None
parsed_tool_input: Any = parsed_tool_call.get(
"arguments", parsed_tool_call.get("input", {})
)
if isinstance(parsed_tool_input, str):
parsed_tool_input = self._parse_json_string_if_possible(
parsed_tool_input
)
if not isinstance(parsed_tool_input, dict):
return None
return ChatCompletionMessageToolCall(
function=Function(
name=resolved_tool_name,
arguments=json.dumps(parsed_tool_input),
)
)
direct_function_call_match = re.match(
r"^\s*([A-Za-z0-9_.-]+)\s*\((.*)\)\s*$",
content,
flags=re.DOTALL,
)
if direct_function_call_match is not None:
resolved_tool_name = self._resolve_nexus_tool_name(
direct_function_call_match.group(1), tool_call_names
)
if resolved_tool_name is None:
return None
raw_tool_input = direct_function_call_match.group(2).strip()
parsed_tool_input: Any = {}
if raw_tool_input:
if raw_tool_input.startswith("{") and raw_tool_input.endswith("}"):
parsed_tool_input = self._parse_json_string_if_possible(
raw_tool_input
)
else:
parsed_tool_input = {}
for key, value in re.findall(
r"([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(\"(?:[^\"\\]|\\.)*\"|'(?:[^'\\]|\\.)*')",
raw_tool_input,
):
parsed_tool_input[key] = ast.literal_eval(value)
if not isinstance(parsed_tool_input, dict):
return None
return ChatCompletionMessageToolCall(
function=Function(
name=resolved_tool_name,
arguments=json.dumps(parsed_tool_input),
)
)
tool_use_match = re.search(
r"<tool_use>\s*(.*?)\s*</tool_use>",
content,

View file

@ -329,6 +329,88 @@ def test_apply_tool_call_transformation_if_needed_with_nexus_tool_use_xml_format
)
def test_apply_tool_call_transformation_if_needed_with_nexus_tool_call_json_format():
from litellm.types.utils import Message
config = AmazonConverseConfig()
tools = [
{
"type": "function",
"function": {
"name": "read_file",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string"},
},
"required": ["path"],
},
},
}
]
message = Message(
role="assistant",
content='<tool_call>{"name":"read_file","arguments":{"path":"README.md"}}</tool_call>',
)
transformed_message, transformed_finish_reason = (
config.apply_tool_call_transformation_if_needed(
message=message,
tools=tools,
initial_finish_reason="stop",
)
)
assert transformed_message.content is None
assert transformed_finish_reason == "tool_calls"
assert transformed_message.tool_calls is not None
assert transformed_message.tool_calls[0].function.name == "read_file"
assert transformed_message.tool_calls[0].function.arguments == json.dumps(
{"path": "README.md"}
)
def test_apply_tool_call_transformation_if_needed_with_direct_function_call_format():
from litellm.types.utils import Message
config = AmazonConverseConfig()
tools = [
{
"type": "function",
"function": {
"name": "read_file",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string"},
},
"required": ["path"],
},
},
}
]
message = Message(
role="assistant",
content='read_file(path="README.md")',
)
transformed_message, transformed_finish_reason = (
config.apply_tool_call_transformation_if_needed(
message=message,
tools=tools,
initial_finish_reason="stop",
)
)
assert transformed_message.content is None
assert transformed_finish_reason == "tool_calls"
assert transformed_message.tool_calls is not None
assert transformed_message.tool_calls[0].function.name == "read_file"
assert transformed_message.tool_calls[0].function.arguments == json.dumps(
{"path": "README.md"}
)
def test_apply_tool_call_transformation_if_needed_ignores_unknown_nexus_tool_name():
from litellm.types.utils import Message