use std::collections::HashMap;
use async_trait::async_trait;
use openai_protocol::common::Tool;
use regex::Regex;
use serde_json::Value;
use crate::{
errors::{ParserError, ParserResult},
parsers::helpers,
traits::ToolParser,
types::{FunctionCall, StreamingParseResult, ToolCall, ToolCallItem},
};
pub struct QwenXmlParser {
extractor: Regex,
buffer: String,
prev_tool_call_arr: Vec<Value>,
current_tool_id: i32,
current_tool_name_sent: bool,
streamed_args_for_tool: Vec<String>,
tool_call_start_token: &'static str,
tool_call_end_token: &'static str,
in_tool_call: bool,
current_function_name: String,
current_parameters: serde_json::Map<String, Value>,
xml_function_pattern: Regex,
xml_param_pattern: Regex,
xml_param_open_pattern: Regex,
open_parameter: Option<OpenParameter>,
}
const CLOSING_TAGS: [&str; 3] = ["</parameter>", "</function>", "</tool_call>"];
struct OpenParameter {
key: String,
value_start: usize,
streamed_end: Option<usize>,
sent: String,
literal: bool,
}
fn json_string_body(text: &str) -> String {
let quoted = serde_json::to_string(text).unwrap_or_default();
quoted
.strip_prefix('"')
.and_then(|body| body.strip_suffix('"'))
.unwrap_or_default()
.to_string()
}
fn safe_val(raw: &str) -> Value {
let trimmed = raw.trim();
if let Ok(v) = serde_json::from_str::<Value>(trimmed) {
return v;
}
match trimmed {
"True" => return Value::Bool(true),
"False" => return Value::Bool(false),
"None" => return Value::Null,
_ => {}
}
Value::String(trimmed.to_string())
}
fn coerce_value(raw: &str, declared_type: Option<&str>) -> Value {
let trimmed = raw.trim();
helpers::coerce_by_schema_type(trimmed, declared_type).unwrap_or_else(|| safe_val(raw))
}
impl QwenXmlParser {
#[expect(
clippy::expect_used,
reason = "regex patterns are compile-time string literals"
)]
pub fn new() -> Self {
let pattern = r"(?s)<tool_call>\s*(.*?)\s*</tool_call>";
let extractor = Regex::new(pattern).expect("Valid regex pattern");
let xml_function_pattern =
Regex::new(r"<function=([^>]+)>").expect("Valid XML function pattern");
let xml_param_pattern = Regex::new(r"(?s)<parameter=([^>]+)>(.*?)</parameter>")
.expect("Valid XML parameter pattern");
let xml_param_open_pattern =
Regex::new(r"<parameter=([^>]+)>").expect("Valid XML parameter start pattern");
Self {
extractor,
buffer: String::new(),
prev_tool_call_arr: Vec::new(),
current_tool_id: -1,
current_tool_name_sent: false,
streamed_args_for_tool: Vec::new(),
tool_call_start_token: "<tool_call>",
tool_call_end_token: "</tool_call>",
in_tool_call: false,
current_function_name: String::new(),
current_parameters: serde_json::Map::new(),
xml_function_pattern,
xml_param_pattern,
xml_param_open_pattern,
open_parameter: None,
}
}
fn parse_xml_format(&self, content: &str, tools: &[Tool]) -> ParserResult<Option<ToolCall>> {
let function_captures = self
.xml_function_pattern
.captures(content)
.ok_or_else(|| ParserError::ParsingFailed("No function name found".to_string()))?;
let function_name = function_captures
.get(1)
.ok_or_else(|| ParserError::ParsingFailed("Function name capture failed".to_string()))?
.as_str()
.trim()
.to_string();
if function_name.is_empty() {
return Ok(None);
}
let param_types = helpers::param_types_for_function(tools, &function_name);
let mut parameters = serde_json::Map::new();
for cap in self.xml_param_pattern.captures_iter(content) {
if let (Some(key_match), Some(value_match)) = (cap.get(1), cap.get(2)) {
let key = key_match.as_str().trim().to_string();
let value = value_match.as_str();
let json_value = coerce_value(value, param_types.get(&key).map(String::as_str));
parameters.insert(key, json_value);
}
}
let arguments = serde_json::to_string(¶meters)
.map_err(|e| ParserError::ParsingFailed(e.to_string()))?;
Ok(Some(ToolCall {
function: FunctionCall {
name: function_name,
arguments,
},
}))
}
fn parse_and_stream_parameters(
&mut self,
tools: &[Tool],
parameter_end: usize,
) -> Vec<ToolCallItem> {
let mut calls: Vec<ToolCallItem> = vec![];
let param_types = helpers::param_types_for_function(tools, &self.current_function_name);
let mut new_params = serde_json::Map::new();
for cap in self
.xml_param_pattern
.captures_iter(&self.buffer[..parameter_end])
{
if let (Some(key_match), Some(value_match)) = (cap.get(1), cap.get(2)) {
let key = key_match.as_str().trim().to_string();
let value = value_match.as_str();
let json_value = coerce_value(value, param_types.get(&key).map(String::as_str));
new_params.insert(key, json_value);
}
}
if new_params != self.current_parameters {
let current_args = &mut self.streamed_args_for_tool[self.current_tool_id as usize];
if self.current_parameters.is_empty() {
let mut items = Vec::new();
for (key, value) in &new_params {
let key_json =
serde_json::to_string(key).unwrap_or_else(|_| format!("\"{key}\""));
let value_json = serde_json::to_string(value).unwrap_or_default();
items.push(format!("{key_json}: {value_json}"));
}
let json_fragment = format!("{{{}", items.join(", "));
calls.push(ToolCallItem {
tool_index: self.current_tool_id as usize,
name: None,
parameters: json_fragment.clone(),
});
*current_args = json_fragment;
} else {
let new_keys: Vec<_> = new_params
.keys()
.filter(|k| !self.current_parameters.contains_key(*k))
.collect();
if !new_keys.is_empty() {
let mut continuation_parts = Vec::new();
for key in new_keys {
if let Some(value) = new_params.get(key) {
let key_json =
serde_json::to_string(key).unwrap_or_else(|_| format!("\"{key}\""));
let value_json = serde_json::to_string(value).unwrap_or_default();
continuation_parts.push(format!("{key_json}: {value_json}"));
}
}
let json_fragment = format!(", {}", continuation_parts.join(", "));
calls.push(ToolCallItem {
tool_index: self.current_tool_id as usize,
name: None,
parameters: json_fragment.clone(),
});
current_args.push_str(&json_fragment);
}
}
self.current_parameters.clone_from(&new_params);
if let Some(tool_obj) =
self.prev_tool_call_arr[self.current_tool_id as usize].as_object_mut()
{
tool_obj.insert("arguments".to_string(), Value::Object(new_params));
}
}
calls
}
fn finish_closed_parameter(
&mut self,
tools: &[Tool],
parameter_end: usize,
) -> Vec<ToolCallItem> {
let closed_at = self.open_parameter.as_ref().and_then(|open| {
self.buffer[open.value_start..parameter_end]
.find("</parameter>")
.map(|close| open.value_start + close)
});
let Some(value_end) = closed_at else {
return Vec::new();
};
let param_types = helpers::param_types_for_function(tools, &self.current_function_name);
self.finish_open_parameter(¶m_types, value_end)
}
fn stream_open_parameter(&mut self, tools: &[Tool], parameter_end: usize) -> Vec<ToolCallItem> {
let mut calls = Vec::new();
let param_types = helpers::param_types_for_function(tools, &self.current_function_name);
let region = &self.buffer[..parameter_end];
let Some(open_tag) = self.xml_param_open_pattern.captures_iter(region).last() else {
return calls;
};
let value_start = open_tag.get(0).map_or(parameter_end, |m| m.end());
let value_region = ®ion[value_start..];
if value_region.contains("</parameter>") {
return calls;
}
let key = open_tag.get(1).map_or("", |m| m.as_str()).trim();
if param_types.get(key).map(String::as_str) != Some("string") {
return calls;
}
if self
.open_parameter
.as_ref()
.is_none_or(|open| open.key != key || open.value_start != value_start)
{
self.open_parameter = Some(OpenParameter {
key: key.to_string(),
value_start,
streamed_end: None,
sent: String::new(),
literal: false,
});
}
let region_end = match value_region.find("</function>") {
Some(end) => value_start + end,
None => {
let held = CLOSING_TAGS
.iter()
.filter_map(|tag| helpers::ends_with_partial_token(value_region, tag))
.max()
.unwrap_or(0);
parameter_end - held
}
};
let tool_index = self.current_tool_id as usize;
let Some(open) = self.open_parameter.as_mut() else {
return calls;
};
if open.literal {
return calls;
}
let from = match open.streamed_end {
Some(end) => end,
None => {
let raw = &self.buffer[value_start..region_end];
let from = value_start + (raw.len() - raw.trim_start().len());
if from >= region_end {
return calls;
}
if self.buffer[from..].starts_with('"') {
open.literal = true;
return calls;
}
from
}
};
if region_end <= from {
return calls;
}
let fresh = self.buffer[from..region_end].trim_end();
if fresh.is_empty() {
return calls;
}
let body = json_string_body(fresh);
let mut delta = String::new();
if open.streamed_end.is_none() {
let key_json =
serde_json::to_string(&open.key).unwrap_or_else(|_| format!("\"{}\"", open.key));
delta.push_str(if self.streamed_args_for_tool[tool_index].is_empty() {
"{"
} else {
", "
});
delta.push_str(&key_json);
delta.push_str(": \"");
}
delta.push_str(&body);
open.sent.push_str(&body);
open.streamed_end = Some(from + fresh.len());
self.streamed_args_for_tool[tool_index].push_str(&delta);
calls.push(ToolCallItem {
tool_index,
name: None,
parameters: delta,
});
calls
}
fn finish_open_parameter(
&mut self,
param_types: &HashMap<String, String>,
value_end: usize,
) -> Vec<ToolCallItem> {
let Some(open) = self.open_parameter.take() else {
return Vec::new();
};
if open.streamed_end.is_none() {
return Vec::new();
}
let tool_index = self.current_tool_id as usize;
let value = coerce_value(
&self.buffer[open.value_start..value_end],
param_types.get(&open.key).map(String::as_str),
);
let full = serde_json::to_string(&value).unwrap_or_default();
let body = full
.strip_prefix('"')
.and_then(|body| body.strip_suffix('"'))
.unwrap_or_default();
let mut tail = body
.strip_prefix(open.sent.as_str())
.unwrap_or_default()
.to_string();
tail.push('"');
self.streamed_args_for_tool[tool_index].push_str(&tail);
if let Some(arguments) = self.prev_tool_call_arr[tool_index]
.get_mut("arguments")
.and_then(Value::as_object_mut)
{
arguments.insert(open.key.clone(), value.clone());
}
self.current_parameters.insert(open.key, value);
vec![ToolCallItem {
tool_index,
name: None,
parameters: tail,
}]
}
fn parse_complete_inner(
&self,
text: &str,
tools: &[Tool],
) -> ParserResult<(String, Vec<ToolCall>)> {
if !self.has_tool_markers(text) {
return Ok((text.to_string(), vec![]));
}
let idx = text
.find(self.tool_call_start_token)
.ok_or_else(|| ParserError::ParsingFailed("tool call marker not found".to_string()))?;
let normal_text = text[..idx].to_string();
let mut parsed = Vec::new();
for captures in self.extractor.captures_iter(text) {
if let Some(content_str) = captures.get(1) {
let content = content_str.as_str().trim();
match self.parse_xml_format(content, tools) {
Ok(Some(tool)) => parsed.push(tool),
Ok(None) => continue,
Err(e) => {
tracing::warn!("Failed to parse XML tool call: {:?}", e);
continue;
}
}
}
}
if parsed.is_empty() {
return Ok((text.to_string(), vec![]));
}
Ok((normal_text, parsed))
}
fn reset_streaming_state(&mut self) {
self.in_tool_call = false;
self.current_tool_name_sent = false;
self.current_function_name.clear();
self.current_parameters.clear();
self.open_parameter = None;
}
}
impl Default for QwenXmlParser {
fn default() -> Self {
Self::new()
}
}
#[async_trait]
impl ToolParser for QwenXmlParser {
async fn parse_complete(&self, text: &str) -> ParserResult<(String, Vec<ToolCall>)> {
self.parse_complete_inner(text, &[])
}
async fn parse_complete_with_tools(
&self,
text: &str,
tools: &[Tool],
) -> ParserResult<(String, Vec<ToolCall>)> {
self.parse_complete_inner(text, tools)
}
async fn parse_incremental(
&mut self,
chunk: &str,
tools: &[Tool],
) -> ParserResult<StreamingParseResult> {
self.buffer.push_str(chunk);
let mut normal_text = String::new();
let mut calls: Vec<ToolCallItem> = vec![];
let tool_indices = helpers::get_tool_indices(tools);
loop {
if !self.in_tool_call && !self.buffer.contains(self.tool_call_start_token) {
if helpers::ends_with_partial_token(&self.buffer, self.tool_call_start_token)
.is_none()
{
normal_text.push_str(&self.buffer);
self.buffer.clear();
}
break;
}
if !self.in_tool_call {
if let Some(s) = self.buffer.find(self.tool_call_start_token) {
normal_text.push_str(&self.buffer[..s]);
self.buffer = self.buffer[s + self.tool_call_start_token.len()..].to_string();
self.in_tool_call = true;
self.current_tool_name_sent = false;
self.current_function_name.clear();
self.current_parameters.clear();
continue;
} else {
break;
}
}
if !self.current_tool_name_sent {
if let Some(captures) = self.xml_function_pattern.captures(&self.buffer) {
if let Some(name_match) = captures.get(1) {
let function_name = name_match.as_str().trim().to_string();
if tool_indices.contains_key(&function_name) {
self.current_function_name.clone_from(&function_name);
self.current_tool_name_sent = true;
if self.current_tool_id == -1 {
self.current_tool_id = 0;
}
helpers::ensure_capacity(
self.current_tool_id,
&mut self.prev_tool_call_arr,
&mut self.streamed_args_for_tool,
);
self.prev_tool_call_arr[self.current_tool_id as usize] = serde_json::json!({
"name": function_name,
"arguments": {}
});
calls.push(ToolCallItem {
tool_index: self.current_tool_id as usize,
name: Some(function_name),
parameters: String::new(),
});
self.buffer =
self.buffer[captures.get(0).map_or(0, |m| m.end())..].to_string();
continue;
} else {
tracing::warn!("Invalid function name: {}", function_name);
self.reset_streaming_state();
normal_text.push_str(&self.buffer);
self.buffer.clear();
break;
}
}
} else {
break;
}
}
if self.current_tool_name_sent {
let end_pos = self.buffer.find(self.tool_call_end_token);
let parameter_end = end_pos.unwrap_or(self.buffer.len());
calls.extend(self.finish_closed_parameter(tools, parameter_end));
calls.extend(self.parse_and_stream_parameters(tools, parameter_end));
calls.extend(self.stream_open_parameter(tools, parameter_end));
if let Some(end_pos) = end_pos {
let unclosed_string = self
.open_parameter
.take()
.is_some_and(|open| open.streamed_end.is_some());
let current_args =
&mut self.streamed_args_for_tool[self.current_tool_id as usize];
let mut closing = String::new();
if unclosed_string {
closing.push('"');
}
closing.push_str(if current_args.is_empty() { "{}" } else { "}" });
calls.push(ToolCallItem {
tool_index: self.current_tool_id as usize,
name: None,
parameters: closing.clone(),
});
current_args.push_str(&closing);
self.buffer =
self.buffer[end_pos + self.tool_call_end_token.len()..].to_string();
self.reset_streaming_state();
self.current_tool_id += 1;
continue;
} else {
break;
}
}
break;
}
Ok(StreamingParseResult { normal_text, calls })
}
fn has_tool_markers(&self, text: &str) -> bool {
text.contains(self.tool_call_start_token)
}
fn get_unstreamed_tool_args(&self) -> Option<Vec<ToolCallItem>> {
let tool_index = self.prev_tool_call_arr.len().checked_sub(1)?;
if self
.open_parameter
.as_ref()
.is_some_and(|open| open.streamed_end.is_some())
{
return Some(vec![ToolCallItem {
tool_index,
name: None,
parameters: "\"}".to_string(),
}]);
}
let actual = self.streamed_args_for_tool.get(tool_index)?;
let expected = self.prev_tool_call_arr[tool_index].get("arguments")?;
if !actual.is_empty()
&& serde_json::from_str::<Value>(&format!("{actual}}}"))
.ok()
.as_ref()
== Some(expected)
{
return Some(vec![ToolCallItem {
tool_index,
name: None,
parameters: "}".to_string(),
}]);
}
helpers::get_unstreamed_args(&self.prev_tool_call_arr, &self.streamed_args_for_tool)
}
fn reset(&mut self) {
helpers::reset_parser_state(
&mut self.buffer,
&mut self.prev_tool_call_arr,
&mut self.current_tool_id,
&mut self.current_tool_name_sent,
&mut self.streamed_args_for_tool,
);
self.reset_streaming_state();
}
}
#[cfg(test)]
mod tests {
use openai_protocol::common::Function;
use super::*;
#[test]
fn test_safe_val_json() {
assert_eq!(safe_val("42"), Value::Number(42.into()));
assert_eq!(safe_val("1.5"), serde_json::json!(1.5));
assert_eq!(safe_val("true"), Value::Bool(true));
assert_eq!(safe_val("false"), Value::Bool(false));
assert_eq!(safe_val("null"), Value::Null);
assert_eq!(
safe_val(r#"{"key": "value"}"#),
serde_json::json!({"key": "value"})
);
assert_eq!(safe_val(r"[1, 2, 3]"), serde_json::json!([1, 2, 3]));
}
#[test]
fn test_safe_val_python_literals() {
assert_eq!(safe_val("True"), Value::Bool(true));
assert_eq!(safe_val("False"), Value::Bool(false));
assert_eq!(safe_val("None"), Value::Null);
}
#[test]
fn test_safe_val_string_fallback() {
assert_eq!(
safe_val("hello world"),
Value::String("hello world".to_string())
);
assert_eq!(safe_val(" spaces "), Value::String("spaces".to_string()));
}
#[test]
fn test_safe_val_preserves_html_entities() {
assert_eq!(
safe_val("<div>"),
Value::String("<div>".to_string())
);
assert_eq!(
safe_val("Tom & Jerry"),
Value::String("Tom & Jerry".to_string())
);
assert_eq!(safe_val("it's"), Value::String("it's".to_string()));
assert_eq!(safe_val("<"), Value::String("<".to_string()));
}
fn tool_with_props(props: Value) -> Vec<Tool> {
vec![Tool {
tool_type: "function".to_string(),
function: Function {
name: "f".to_string(),
description: None,
parameters: serde_json::json!({"type": "object", "properties": props}),
strict: None,
extra: Default::default(),
},
}]
}
#[tokio::test]
async fn test_schema_aware_coercion_keeps_strings() {
let tools = tool_with_props(serde_json::json!({
"limit": {"type": "string"},
"flag": {"type": "string"},
"coords": {"type": "string"},
"cfg": {"type": "string"},
"count": {"type": "integer"},
}));
let text = "<tool_call>\n<function=f>\n\
<parameter=limit>4</parameter>\n\
<parameter=flag>true</parameter>\n\
<parameter=coords>[60,30]</parameter>\n\
<parameter=cfg>{\"a\": 1}</parameter>\n\
<parameter=count>5</parameter>\n\
</function>\n</tool_call>";
let (_, calls) = QwenXmlParser::new()
.parse_complete_with_tools(text, &tools)
.await
.unwrap();
assert_eq!(calls.len(), 1);
let args: Value = serde_json::from_str(&calls[0].function.arguments).unwrap();
assert_eq!(args["limit"], Value::String("4".to_string()));
assert_eq!(args["flag"], Value::String("true".to_string()));
assert_eq!(args["coords"], Value::String("[60,30]".to_string()));
assert_eq!(args["cfg"], Value::String("{\"a\": 1}".to_string()));
assert_eq!(args["count"], Value::Number(5.into()));
}
#[tokio::test]
async fn test_streaming_schema_aware_coercion() {
let tools = tool_with_props(serde_json::json!({
"limit": {"type": "string"},
"count": {"type": "integer"},
}));
let text = "<tool_call>\n<function=f>\n\
<parameter=limit>4</parameter>\n\
<parameter=count>5</parameter>\n\
</function>\n</tool_call>";
let result = QwenXmlParser::new()
.parse_incremental(text, &tools)
.await
.unwrap();
let args: String = result.calls.iter().map(|c| c.parameters.as_str()).collect();
assert!(
args.contains(r#""limit": "4""#),
"string param must stay string: {args}"
);
assert!(
args.contains(r#""count": 5"#),
"int param must coerce: {args}"
);
}
#[tokio::test]
async fn test_arg_values_with_entities_roundtrip_unchanged() {
let literal = "<a>Tom & Jerry</a> <x> it's";
let text = format!(
"<tool_call>\n<function=f>\n\
<parameter=content>\n{literal}\n</parameter>\n\
</function>\n</tool_call>"
);
let (_, calls) = QwenXmlParser::new().parse_complete(&text).await.unwrap();
assert_eq!(calls.len(), 1);
let args: Value = serde_json::from_str(&calls[0].function.arguments).unwrap();
assert_eq!(args["content"], Value::String(literal.to_string()));
let tools = tool_with_props(serde_json::json!({
"content": {"type": "string"},
}));
let (_, calls) = QwenXmlParser::new()
.parse_complete_with_tools(&text, &tools)
.await
.unwrap();
let args: Value = serde_json::from_str(&calls[0].function.arguments).unwrap();
assert_eq!(args["content"], Value::String(literal.to_string()));
}
fn bash_tool() -> Vec<Tool> {
vec![Tool {
tool_type: "function".to_string(),
function: Function {
name: "bash".to_string(),
description: None,
parameters: serde_json::json!({
"type": "object",
"properties": {
"command": {"type": "string"},
"timeout": {"type": "integer"}
}
}),
strict: None,
extra: Default::default(),
},
}]
}
async fn argument_deltas(
parser: &mut QwenXmlParser,
chunks: &[&str],
tools: &[Tool],
) -> Vec<(usize, String)> {
let mut deltas = Vec::new();
for (index, chunk) in chunks.iter().enumerate() {
for call in parser.parse_incremental(chunk, tools).await.unwrap().calls {
if call.name.is_none() && !call.parameters.is_empty() {
deltas.push((index, call.parameters));
}
}
}
deltas
}
#[tokio::test]
async fn test_string_values_stream_before_the_parameter_closes() {
let tools = bash_tool();
let chunks = [
"<tool_call>",
"\n",
"<function=",
"bash",
">\n",
"<parameter=",
"command",
">\n",
"ls",
" -",
"la",
" /",
"tmp",
"\n",
"</parameter>",
"\n</function>",
"\n</tool_call>",
];
let close_at = chunks.iter().position(|c| *c == "</parameter>").unwrap();
let mut parser = QwenXmlParser::new();
let deltas = argument_deltas(&mut parser, &chunks, &tools).await;
assert!(deltas.len() > 2, "arguments arrived as {deltas:?}");
assert_eq!(deltas[0], (8, "{\"command\": \"ls".to_string()));
let before_close = deltas.iter().filter(|(at, _)| *at < close_at).count();
assert_eq!(before_close, 5, "{deltas:?}");
let joined: String = deltas.iter().map(|(_, d)| d.as_str()).collect();
assert_eq!(joined, "{\"command\": \"ls -la /tmp\"}");
assert!(parser.get_unstreamed_tool_args().is_none());
assert!(parser.take_unstreamed_normal_text().is_empty());
}
#[tokio::test]
async fn test_streamed_string_values_match_the_complete_parse_at_every_split() {
let tools = bash_tool();
let command = "<parameter=command>\n echo \"a\\b\"\t<tag> \n\n</parameter>\n";
let timeout = "<parameter=timeout>30</parameter>\n";
for text in [
format!("<tool_call>\n<function=bash>\n{command}{timeout}</function>\n</tool_call>"),
format!("<tool_call>\n<function=bash>\n{timeout}{command}</function>\n</tool_call>"),
] {
let (_, complete) = QwenXmlParser::new()
.parse_complete_with_tools(&text, &tools)
.await
.unwrap();
let expected: Value = serde_json::from_str(&complete[0].function.arguments).unwrap();
assert_eq!(expected["command"], "echo \"a\\b\"\t<tag>");
assert_eq!(expected["timeout"], 30);
let mut feeds: Vec<Vec<String>> = vec![text.chars().map(|c| c.to_string()).collect()];
for (split, _) in text.char_indices().skip(1) {
feeds.push(vec![text[..split].to_string(), text[split..].to_string()]);
}
for chunks in feeds {
let mut parser = QwenXmlParser::new();
let mut args = String::new();
for chunk in &chunks {
for call in parser.parse_incremental(chunk, &tools).await.unwrap().calls {
args.push_str(&call.parameters);
}
}
assert!(parser.get_unstreamed_tool_args().is_none(), "{chunks:?}");
assert_eq!(
serde_json::from_str::<Value>(&args).ok().as_ref(),
Some(&expected),
"args {args}; chunks {chunks:?}"
);
}
}
}
#[tokio::test]
async fn test_values_that_need_the_complete_text_arrive_whole() {
let tools = bash_tool();
for (text, expected) in [
(
"<tool_call><function=bash><parameter=timeout>30</parameter></function></tool_call>",
["{\"timeout\": 30", "}"],
),
(
"<tool_call><function=bash><parameter=command>\"ls\"</parameter></function></tool_call>",
["{\"command\": \"ls\"", "}"],
),
] {
let chunks: Vec<String> = text.chars().map(|c| c.to_string()).collect();
let chunk_refs: Vec<&str> = chunks.iter().map(String::as_str).collect();
let mut parser = QwenXmlParser::new();
let deltas: Vec<String> = argument_deltas(&mut parser, &chunk_refs, &tools)
.await
.into_iter()
.map(|(_, delta)| delta)
.collect();
assert_eq!(deltas, expected);
}
}
#[tokio::test]
async fn test_end_of_stream_closes_a_partially_streamed_value() {
let tools = bash_tool();
let mut parser = QwenXmlParser::new();
let deltas = argument_deltas(
&mut parser,
&[
"<tool_call><function=bash><parameter=command>ls -l",
"a /tm",
],
&tools,
)
.await;
let mut args: String = deltas.into_iter().map(|(_, d)| d).collect();
assert_eq!(args, "{\"command\": \"ls -la /tm");
for item in parser.get_unstreamed_tool_args().unwrap() {
args.push_str(&item.parameters);
}
assert_eq!(args, "{\"command\": \"ls -la /tm\"}");
assert!(parser.take_unstreamed_normal_text().is_empty());
let mut parser = QwenXmlParser::new();
let deltas = argument_deltas(
&mut parser,
&["<tool_call><function=bash><parameter=command>\n"],
&tools,
)
.await;
assert!(deltas.is_empty(), "{deltas:?}");
let pending: String = parser
.get_unstreamed_tool_args()
.unwrap()
.into_iter()
.map(|item| item.parameters)
.collect();
assert_eq!(pending, "{}");
}
}