fix double Unicode escape issue in streaming tool_calls parameters (#13518)

Co-authored-by: Xinyuan Tong <115166877+JustinTong0323@users.noreply.github.com>
This commit is contained in:
lw9527
2026-01-06 15:47:26 +08:00
committed by GitHub
parent 176266f358
commit 2724b1100b
2 changed files with 136 additions and 2 deletions

View File

@@ -719,6 +719,140 @@ class TestBaseFormatDetector(unittest.TestCase):
self.detector._buffer, "", "Buffer should be cleared for invalid tool"
)
def test_chinese_characters_not_double_escaped(self):
"""Test that Chinese characters in tool call parameters are not double-escaped."""
# Test with Chinese city name "杭州" (Hangzhou)
chunks = [
"<tool_call>",
'{"name": "get_weather", ',
'"arguments": {"city": "杭州"}}',
"</tool_call>",
]
accumulated_parameters = {}
for chunk in chunks:
result = self.detector.parse_streaming_increment(chunk, self.tools)
if result.calls:
for call in result.calls:
if call.parameters:
tool_idx = call.tool_index if call.tool_index is not None else 0
if tool_idx not in accumulated_parameters:
accumulated_parameters[tool_idx] = ""
accumulated_parameters[tool_idx] += call.parameters
# Verify that Chinese characters are preserved (not escaped as \uXXXX)
self.assertGreater(
len(accumulated_parameters), 0, "Should have parsed parameters"
)
final_params_str = accumulated_parameters[0]
# The parameters string should contain the actual Chinese characters, not escaped Unicode
self.assertIn(
"杭州", final_params_str, "Should contain actual Chinese characters"
)
self.assertNotIn(
"\\u676d", final_params_str, "Should not contain escaped Unicode sequences"
)
self.assertNotIn(
"\\u5dde", final_params_str, "Should not contain escaped Unicode sequences"
)
# Verify the JSON can be parsed and contains the correct value
params = json.loads(final_params_str)
self.assertEqual(
params["city"], "杭州", "Should correctly parse Chinese city name"
)
def test_chinese_characters_incremental_streaming(self):
"""Test that Chinese characters work correctly with incremental streaming."""
# Test incremental streaming with Chinese characters
chunks = [
"<tool_call>",
'{"name": "get_weather", ',
'"arguments": {"city": "',
"杭州",
'"}}',
"</tool_call>",
]
accumulated_parameters = {}
for chunk in chunks:
result = self.detector.parse_streaming_increment(chunk, self.tools)
if result.calls:
for call in result.calls:
if call.parameters:
tool_idx = call.tool_index if call.tool_index is not None else 0
if tool_idx not in accumulated_parameters:
accumulated_parameters[tool_idx] = ""
accumulated_parameters[tool_idx] += call.parameters
# Verify Chinese characters are preserved throughout streaming
self.assertGreater(
len(accumulated_parameters), 0, "Should have parsed parameters"
)
final_params_str = accumulated_parameters[0]
# Should contain actual Chinese characters, not escaped
self.assertIn(
"杭州", final_params_str, "Should contain actual Chinese characters"
)
# Parse and verify
params = json.loads(final_params_str)
self.assertEqual(
params["city"], "杭州", "Should correctly parse Chinese city name"
)
def test_multiple_chinese_parameters(self):
"""Test multiple tool calls with Chinese parameters."""
# Test with multiple tool calls containing Chinese characters
chunks = [
"<tool_call>",
'{"name": "get_weather", "arguments": {"city": "北京"}}, ',
'{"name": "get_tourist_attractions", "arguments": {"city": "上海"}}',
"</tool_call>",
]
accumulated_parameters = {}
for chunk in chunks:
result = self.detector.parse_streaming_increment(chunk, self.tools)
if result.calls:
for call in result.calls:
if call.parameters:
tool_idx = call.tool_index if call.tool_index is not None else 0
if tool_idx not in accumulated_parameters:
accumulated_parameters[tool_idx] = ""
accumulated_parameters[tool_idx] += call.parameters
# Verify both tool calls have correct Chinese characters
self.assertGreaterEqual(
len(accumulated_parameters), 1, "Should have parsed parameters"
)
# Check first tool call (北京 - Beijing)
if 0 in accumulated_parameters:
params0 = json.loads(accumulated_parameters[0])
self.assertIn(
"北京",
accumulated_parameters[0],
"Should contain actual Chinese characters",
)
self.assertEqual(
params0["city"], "北京", "Should correctly parse first Chinese city"
)
# Check second tool call (上海 - Shanghai) if present
if 1 in accumulated_parameters:
params1 = json.loads(accumulated_parameters[1])
self.assertIn(
"上海",
accumulated_parameters[1],
"Should contain actual Chinese characters",
)
self.assertEqual(
params1["city"], "上海", "Should correctly parse second Chinese city"
)
class TestLlama32Detector(unittest.TestCase):
def setUp(self):