fix(tools): never deliver a tool call twice, and make an unparsed one executable
Docker / publish (push) Canceled after 0s
Release / crates-io (push) Canceled after 0s
Release / release (macos-14, aarch64-apple-darwin) (push) Canceled after 0s
Release / release (ubuntu-latest, x86_64-unknown-linux-gnu) (push) Canceled after 0s
Release / release (windows-latest, x86_64-pc-windows-msvc) (push) Canceled after 0s
Docker / publish (push) Canceled after 0s
Release / crates-io (push) Canceled after 0s
Release / release (macos-14, aarch64-apple-darwin) (push) Canceled after 0s
Release / release (ubuntu-latest, x86_64-unknown-linux-gnu) (push) Canceled after 0s
Release / release (windows-latest, x86_64-pc-windows-msvc) (push) Canceled after 0s
Qwen-family templates write a function call into the generated text as
<tool_call>{...}</tool_call>. Runtimes disagree on what to do with it: some
parse it into structured tool_calls AND still stream the raw markup as
content - the client then gets the same call twice, once as prose it must not
show and once as the real thing - while others never parse it at all, so a
perfectly good call arrives as text no tool loop will run.
Settle it in the gateway. A streaming-aware filter strips the markup from
content (handling a marker split across SSE chunks, leaving ordinary text
containing "<" alone, and handing an unterminated block back as text), and
when nothing structured came with it the markup is promoted into a real
tool_calls entry with a generated call_... id. Markup that is not a JSON
object with a name is ignored so a stray marker cannot become a bogus call.
Streamed tool_calls are also brought to the OpenAI shape: index on every
entry, id and type exactly once per call so continuation deltas carrying only
an argument fragment stay legal, and function.arguments always a string. A
reply that carries tool calls now always ends with finish_reason "tool_calls",
whatever the upstream called it. The same cleanup runs on the non-streaming
path, so every protocol adapter benefits.
Verified end to end: Open WebUI 0.11.0 -> gateway -> managed CMF/Qwen with
native function calling. The model returns get_weather({"city":"Paris"}), Open
WebUI executes the tool, feeds the result back, and the assistant answers from
it - no markup in content, finish_reason tool_calls, stream closed with [DONE].
This commit is contained in:
@@ -3,6 +3,33 @@
|
||||
All notable changes to this project are documented here.
|
||||
The format is based on [Keep a Changelog](https://keepachangelog.com/).
|
||||
|
||||
## [0.2.48] - 2026-08-21
|
||||
|
||||
Verified end to end: Open WebUI 0.11.0 → gateway → managed CMF/Qwen, native
|
||||
function calling, tool executed and its result fed back into the reply.
|
||||
|
||||
### Fixed
|
||||
- **A tool call is never delivered twice.** Qwen-family templates write the call into the
|
||||
generated text as `<tool_call>{…}</tool_call>`; some runtimes parse it into structured
|
||||
`tool_calls` *and* still stream the raw markup as `content`, so the client sees the call
|
||||
once as prose it must not show and once as the real thing. The gateway now strips the
|
||||
markup from `content` — including when the marker is split across SSE chunks — while
|
||||
passing the structured call through untouched. Ordinary text containing `<` is not
|
||||
affected, and an unterminated block is handed back as text rather than swallowed.
|
||||
- **`finish_reason` agrees with the reply.** A response carrying tool calls now always ends
|
||||
with `finish_reason: "tool_calls"`, whatever the upstream called it — clients branch on
|
||||
that field to decide whether to run the tool.
|
||||
- **Streamed `tool_calls` are brought to the OpenAI shape**: `index` on every entry, `id`
|
||||
and `type` exactly once per call (continuation deltas that carry only an argument
|
||||
fragment are left alone, as the spec requires), and `function.arguments` always a string
|
||||
even when the runtime sends an object.
|
||||
|
||||
### Added
|
||||
- **Unparsed `<tool_call>` markup is promoted into a real call.** When the upstream never
|
||||
parsed it, the gateway turns it into a proper `tool_calls` entry with a generated
|
||||
`call_…` id, so the call is executable instead of decorative. Markup that is not a JSON
|
||||
object with a `name` is ignored — a stray marker must not become a bogus call.
|
||||
|
||||
## [0.2.47] - 2026-08-21
|
||||
|
||||
### Changed
|
||||
|
||||
Generated
+1
-1
@@ -285,7 +285,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
|
||||
|
||||
[[package]]
|
||||
name = "cortiq-gateway"
|
||||
version = "0.2.47"
|
||||
version = "0.2.48"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"arc-swap",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "cortiq-gateway"
|
||||
version = "0.2.47"
|
||||
version = "0.2.48"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "Apache-2.0"
|
||||
|
||||
@@ -121,6 +121,20 @@ opens with a `role` delta, carries exactly one terminal `finish_reason`, and end
|
||||
failure arrives as an SSE `error` event. A client can therefore never be left waiting on a
|
||||
half-formed stream.
|
||||
|
||||
### Tool calls
|
||||
|
||||
`tools` and `tool_calls` are proxied through, with the wire kept clean:
|
||||
|
||||
* raw `<tool_call>…</tool_call>` markup (how Qwen-family templates express a call in the
|
||||
generated text) is stripped from `content`, so a client never receives the same call
|
||||
twice — once as prose and once structured. Markers split across SSE chunks are handled;
|
||||
ordinary text containing `<` is not touched;
|
||||
* when the upstream never parsed that markup, it is promoted into a real `tool_calls`
|
||||
entry with a generated `call_…` id, so the call is executable rather than decorative;
|
||||
* streamed entries carry `index` always, `id`/`type` once per call, and
|
||||
`function.arguments` as a string even if the runtime sent an object;
|
||||
* a reply carrying tool calls always ends with `finish_reason: "tool_calls"`.
|
||||
|
||||
### Metadata in response headers (always present)
|
||||
|
||||
```
|
||||
|
||||
@@ -119,6 +119,21 @@ HTTP 400 — но не «голый» 422, который OpenAI-клиенты
|
||||
`data: [DONE]` — даже если upstream оборвался или упал на середине (тогда ошибка приходит
|
||||
SSE-событием `error`). Клиент никогда не остаётся ждать недосказанный поток.
|
||||
|
||||
### Tool calls
|
||||
|
||||
`tools` и `tool_calls` проксируются, но провод держится чистым:
|
||||
|
||||
* сырая разметка `<tool_call>…</tool_call>` (так Qwen-подобные шаблоны выражают вызов
|
||||
прямо в тексте) вырезается из `content` — клиент никогда не получает один и тот же
|
||||
вызов дважды, прозой и структурой. Маркер, разорванный между SSE-чанками, обрабатывается
|
||||
корректно; обычный текст с символом `<` не трогается;
|
||||
* если апстрим эту разметку не распарсил, она повышается до настоящего элемента
|
||||
`tool_calls` со сгенерированным `call_…` id — вызов становится исполнимым, а не
|
||||
декоративным;
|
||||
* в стриме у каждого элемента всегда есть `index`, `id`/`type` — ровно один раз на вызов,
|
||||
а `function.arguments` — всегда строка, даже если рантайм прислал объект;
|
||||
* ответ с tool-вызовами всегда завершается `finish_reason: "tool_calls"`.
|
||||
|
||||
### Метаданные в заголовках (всегда)
|
||||
|
||||
```
|
||||
|
||||
@@ -29,6 +29,7 @@ mod secrets;
|
||||
mod state;
|
||||
mod stats;
|
||||
mod sysmon;
|
||||
mod toolcalls;
|
||||
|
||||
use axum::{routing::get, Router};
|
||||
use clap::Parser;
|
||||
|
||||
@@ -341,6 +341,12 @@ async fn handler(
|
||||
/// to end with `data: [DONE]` — even when the upstream dies mid-flight, in
|
||||
/// which case the failure is delivered as an SSE `error` event so the client
|
||||
/// shows a message instead of waiting forever.
|
||||
///
|
||||
/// Tool calls are settled here too (see [`crate::toolcalls`]): raw
|
||||
/// `<tool_call>` markup never reaches `content`, structured deltas are brought
|
||||
/// to the OpenAI shape, markup is promoted to a real call when the upstream
|
||||
/// produced none, and a stream that emitted tool calls always finishes with
|
||||
/// `finish_reason: "tool_calls"`.
|
||||
fn normalize_sse(
|
||||
stream: crate::providers::ChatStream,
|
||||
model: String,
|
||||
@@ -355,6 +361,13 @@ fn normalize_sse(
|
||||
let mut roled: std::collections::HashSet<u64> = std::collections::HashSet::new();
|
||||
let mut sent_finish = false;
|
||||
let mut failure: Option<String> = None;
|
||||
// Tool-call hygiene: one markup filter per choice index, the set of
|
||||
// tool-call indices already opened, and whether anything structured
|
||||
// was emitted at all.
|
||||
let mut markup: std::collections::HashMap<u64, crate::toolcalls::ToolMarkupFilter> =
|
||||
std::collections::HashMap::new();
|
||||
let mut tool_indices: std::collections::HashSet<u64> = std::collections::HashSet::new();
|
||||
let mut emitted_tool_calls = false;
|
||||
|
||||
// Emit one normalised event; `None` means "nothing worth sending".
|
||||
macro_rules! chunk {
|
||||
@@ -417,8 +430,15 @@ fn normalize_sse(
|
||||
delta.insert("role".into(), serde_json::json!("assistant"));
|
||||
}
|
||||
if let Some(text) = src["content"].as_str() {
|
||||
if !text.is_empty() {
|
||||
delta.insert("content".into(), serde_json::json!(text));
|
||||
// Strip any `<tool_call>` markup before it can
|
||||
// reach the client as prose — the structured
|
||||
// call is the only form a tool loop may see.
|
||||
let visible = markup
|
||||
.entry(index)
|
||||
.or_default()
|
||||
.push(text);
|
||||
if !visible.is_empty() {
|
||||
delta.insert("content".into(), serde_json::json!(visible));
|
||||
}
|
||||
}
|
||||
// Reasoning traces and citations are not in the base
|
||||
@@ -430,10 +450,45 @@ fn normalize_sse(
|
||||
}
|
||||
}
|
||||
if let Some(tc) = src["tool_calls"].as_array().filter(|a| !a.is_empty()) {
|
||||
delta.insert("tool_calls".into(), serde_json::Value::Array(tc.clone()));
|
||||
let normalized =
|
||||
crate::toolcalls::normalize_delta_tool_calls(tc, &mut tool_indices);
|
||||
if !normalized.is_empty() {
|
||||
emitted_tool_calls = true;
|
||||
delta.insert(
|
||||
"tool_calls".into(),
|
||||
serde_json::Value::Array(normalized),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let finish = c["finish_reason"].as_str().map(normalize_finish_reason);
|
||||
|
||||
// Last chance to promote `<tool_call>` markup into a
|
||||
// real call: an upstream that never parsed it would
|
||||
// otherwise hand the client prose no tool loop runs.
|
||||
if finish.is_some() && !emitted_tool_calls {
|
||||
if let Some(filter) = markup.get_mut(&index) {
|
||||
let leftover = filter.finish();
|
||||
if !leftover.is_empty() {
|
||||
let merged = match delta.get("content").and_then(|v| v.as_str()) {
|
||||
Some(prev) => format!("{prev}{leftover}"),
|
||||
None => leftover,
|
||||
};
|
||||
delta.insert("content".into(), serde_json::json!(merged));
|
||||
}
|
||||
let calls = crate::toolcalls::markup_to_tool_calls(filter.captured(), 0);
|
||||
if !calls.is_empty() {
|
||||
emitted_tool_calls = true;
|
||||
let tc = serde_json::json!({
|
||||
"index": index,
|
||||
"delta": { "tool_calls": calls },
|
||||
});
|
||||
let extra: Vec<(&str, serde_json::Value)> = Vec::new();
|
||||
yield Ok(chunk!(vec![tc], extra));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if finish.is_none() && delta.is_empty() {
|
||||
continue;
|
||||
}
|
||||
@@ -442,6 +497,10 @@ fn normalize_sse(
|
||||
"delta": serde_json::Value::Object(delta),
|
||||
});
|
||||
if let Some(f) = finish {
|
||||
// A reply that carries tool calls stops for that
|
||||
// reason, whatever the upstream called it —
|
||||
// clients branch on this field to run the tool.
|
||||
let f = if emitted_tool_calls { "tool_calls" } else { f };
|
||||
choice["finish_reason"] = serde_json::json!(f);
|
||||
sent_finish = true;
|
||||
}
|
||||
@@ -482,10 +541,24 @@ fn normalize_sse(
|
||||
if roled.insert(0) {
|
||||
delta.insert("role".into(), serde_json::json!("assistant"));
|
||||
}
|
||||
// Held-back text and unparsed markup must not be lost with the stream.
|
||||
for (index, filter) in markup.iter_mut() {
|
||||
let leftover = filter.finish();
|
||||
if !leftover.is_empty() && *index == 0 {
|
||||
delta.insert("content".into(), serde_json::json!(leftover));
|
||||
}
|
||||
if !emitted_tool_calls {
|
||||
let calls = crate::toolcalls::markup_to_tool_calls(filter.captured(), 0);
|
||||
if !calls.is_empty() {
|
||||
emitted_tool_calls = true;
|
||||
delta.insert("tool_calls".into(), serde_json::Value::Array(calls));
|
||||
}
|
||||
}
|
||||
}
|
||||
let choices = vec![serde_json::json!({
|
||||
"index": 0,
|
||||
"delta": serde_json::Value::Object(delta),
|
||||
"finish_reason": "stop",
|
||||
"finish_reason": if emitted_tool_calls { "tool_calls" } else { "stop" },
|
||||
})];
|
||||
let extra: Vec<(&str, serde_json::Value)> = Vec::new();
|
||||
yield Ok(chunk!(choices, extra));
|
||||
@@ -530,6 +603,82 @@ mod tests {
|
||||
assert_eq!(normalize_finish_reason("tool_use"), "tool_calls");
|
||||
}
|
||||
|
||||
/// Collect a normalised stream into one string for assertions.
|
||||
async fn drain(chunks: Vec<&'static str>) -> String {
|
||||
use futures::StreamExt;
|
||||
let upstream = futures::stream::iter(
|
||||
chunks
|
||||
.into_iter()
|
||||
.map(|c| Ok(bytes::Bytes::from(c)))
|
||||
.collect::<Vec<_>>(),
|
||||
);
|
||||
normalize_sse(Box::pin(upstream), "cortiq-auto".into())
|
||||
.map(|b| String::from_utf8_lossy(&b.unwrap()).to_string())
|
||||
.collect::<Vec<_>>()
|
||||
.await
|
||||
.join("")
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tool_markup_is_not_duplicated_into_content() {
|
||||
// The reported shape: the upstream streams the call as prose AND then
|
||||
// sends it properly structured.
|
||||
let out = drain(vec![
|
||||
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"Sure. <tool_call>\"}}]}\n\n",
|
||||
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"{\\\"name\\\":\\\"get_weather\\\",\\\"arguments\\\":{\\\"city\\\":\\\"Paris\\\"}}\"}}]}\n\n",
|
||||
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"</tool_call>\"}}]}\n\n",
|
||||
"data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"get_weather\",\"arguments\":\"{\\\"city\\\":\\\"Paris\\\"}\"}}]}}]}\n\n",
|
||||
"data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
|
||||
"data: [DONE]\n\n",
|
||||
])
|
||||
.await;
|
||||
assert!(
|
||||
!out.contains("tool_call>"),
|
||||
"markup leaked into content: {out}"
|
||||
);
|
||||
assert!(
|
||||
out.contains("\"id\":\"call_1\""),
|
||||
"structured call missing: {out}"
|
||||
);
|
||||
// exactly one structured call, and the stop reason agrees with it
|
||||
assert_eq!(out.matches("\"tool_calls\":[").count(), 1);
|
||||
assert!(out.contains("\"finish_reason\":\"tool_calls\""));
|
||||
assert!(out.trim_end().ends_with("data: [DONE]"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unparsed_markup_is_promoted_to_a_real_tool_call() {
|
||||
// An upstream that never parses the markup: the call must still be
|
||||
// executable by the client rather than arriving as prose.
|
||||
let out = drain(vec![
|
||||
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"<tool_call>{\\\"name\\\":\\\"get_weather\\\",\\\"arguments\\\":{\\\"city\\\":\\\"Paris\\\"}}</tool_call>\"}}]}\n\n",
|
||||
"data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
|
||||
"data: [DONE]\n\n",
|
||||
])
|
||||
.await;
|
||||
assert!(!out.contains("tool_call>"), "markup leaked: {out}");
|
||||
assert!(
|
||||
out.contains("\"name\":\"get_weather\""),
|
||||
"call not promoted: {out}"
|
||||
);
|
||||
assert!(
|
||||
out.contains("\"finish_reason\":\"tool_calls\""),
|
||||
"wrong stop reason: {out}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ordinary_text_is_untouched_by_the_tool_filter() {
|
||||
let out = drain(vec![
|
||||
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"a < b and 2<3\"}}]}\n\n",
|
||||
"data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
|
||||
"data: [DONE]\n\n",
|
||||
])
|
||||
.await;
|
||||
assert!(out.contains("a < b and 2<3"), "text mangled: {out}");
|
||||
assert!(out.contains("\"finish_reason\":\"stop\""));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stream_is_closed_even_when_the_upstream_stops_early() {
|
||||
use futures::StreamExt;
|
||||
|
||||
+17
-3
@@ -122,13 +122,23 @@ impl Provider for OpenAiProvider {
|
||||
for (i, c) in choices_arr.iter().enumerate() {
|
||||
let msg_val = &c["message"];
|
||||
let role = msg_val["role"].as_str().unwrap_or("assistant").to_string();
|
||||
let content = msg_val["content"].as_str().unwrap_or("").to_string();
|
||||
let raw_content = msg_val["content"].as_str().unwrap_or("");
|
||||
// Qwen-style templates put the call in the text as `<tool_call>…`.
|
||||
// Strip it: the structured array is the only form a client may act
|
||||
// on, and prose that merely looks like a call is worse than none.
|
||||
let (content, markup) = crate::toolcalls::split_tool_markup(raw_content);
|
||||
|
||||
let tool_calls = msg_val["tool_calls"]
|
||||
let mut tool_calls = msg_val["tool_calls"]
|
||||
.as_array()
|
||||
.cloned()
|
||||
.unwrap_or_default();
|
||||
if tool_calls.is_empty() {
|
||||
// Upstream never parsed the markup — promote it, so the call is
|
||||
// executable instead of decorative.
|
||||
tool_calls = crate::toolcalls::markup_to_tool_calls(&markup, 0);
|
||||
}
|
||||
|
||||
let tool_calls_present = !tool_calls.is_empty();
|
||||
choices.push(crate::model::Choice {
|
||||
index: c["index"].as_u64().unwrap_or(i as u64) as u32,
|
||||
message: crate::model::Message {
|
||||
@@ -137,7 +147,11 @@ impl Provider for OpenAiProvider {
|
||||
tool_calls,
|
||||
..Default::default()
|
||||
},
|
||||
finish_reason: c["finish_reason"].as_str().unwrap_or("stop").to_string(),
|
||||
finish_reason: if tool_calls_present {
|
||||
"tool_calls".to_string()
|
||||
} else {
|
||||
c["finish_reason"].as_str().unwrap_or("stop").to_string()
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
//! Tool-call hygiene for OpenAI-compatible responses.
|
||||
//!
|
||||
//! Qwen-family chat templates express a function call as literal markup in the
|
||||
//! generated text:
|
||||
//!
|
||||
//! ```text
|
||||
//! <tool_call>
|
||||
//! {"name": "get_weather", "arguments": {"city": "Paris"}}
|
||||
//! </tool_call>
|
||||
//! ```
|
||||
//!
|
||||
//! Runtimes handle that inconsistently. Some parse it into structured
|
||||
//! `tool_calls` **and** still stream the raw markup as `content` — the client
|
||||
//! then sees the call twice, once as prose it must not show and once as the real
|
||||
//! thing, and tool-loop implementations (Open WebUI's among them) get confused.
|
||||
//! Others never parse it at all, so a perfectly good call reaches the client as
|
||||
//! text no tool loop will ever execute.
|
||||
//!
|
||||
//! This module makes the gateway the place where that is settled: markup is
|
||||
//! stripped from the text, and when nothing structured came with it the markup
|
||||
//! is promoted into a real `tool_calls` array.
|
||||
|
||||
const OPEN: &str = "<tool_call>";
|
||||
const CLOSE: &str = "</tool_call>";
|
||||
|
||||
/// Longest suffix of `s` that is a proper prefix of `pat`. Used to hold back the
|
||||
/// tail of a streamed chunk that might be the beginning of a marker split across
|
||||
/// chunk boundaries (`"<tool"` + `"_call>"`).
|
||||
fn partial_marker_len(s: &str, pat: &str) -> usize {
|
||||
let max = pat.len().saturating_sub(1).min(s.len());
|
||||
(1..=max)
|
||||
.rev()
|
||||
.find(|&n| {
|
||||
s.is_char_boundary(s.len() - n)
|
||||
&& pat.as_bytes().starts_with(&s.as_bytes()[s.len() - n..])
|
||||
})
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Incremental `<tool_call>` filter for SSE deltas.
|
||||
///
|
||||
/// Feed it content fragments in order; it returns the text that is safe to
|
||||
/// forward and keeps the markup payloads for [`Self::captured`].
|
||||
#[derive(Default)]
|
||||
pub struct ToolMarkupFilter {
|
||||
/// Text held back: either an unterminated marker prefix or the inside of an
|
||||
/// open `<tool_call>` block.
|
||||
pending: String,
|
||||
inside: bool,
|
||||
captured: Vec<String>,
|
||||
}
|
||||
|
||||
impl ToolMarkupFilter {
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
/// Consume one content fragment; returns the part that may be forwarded now.
|
||||
pub fn push(&mut self, chunk: &str) -> String {
|
||||
self.pending.push_str(chunk);
|
||||
let mut out = String::new();
|
||||
loop {
|
||||
if self.inside {
|
||||
let Some(end) = self.pending.find(CLOSE) else {
|
||||
// still inside the block — nothing to forward
|
||||
return out;
|
||||
};
|
||||
let payload: String = self.pending[..end].to_string();
|
||||
self.captured.push(payload);
|
||||
self.pending = self.pending[end + CLOSE.len()..].to_string();
|
||||
self.inside = false;
|
||||
continue;
|
||||
}
|
||||
match self.pending.find(OPEN) {
|
||||
Some(start) => {
|
||||
out.push_str(&self.pending[..start]);
|
||||
self.pending = self.pending[start + OPEN.len()..].to_string();
|
||||
self.inside = true;
|
||||
}
|
||||
None => {
|
||||
// Hold back a trailing partial marker so a split like
|
||||
// "…<tool" / "_call>…" is not forwarded as visible text.
|
||||
let hold = partial_marker_len(&self.pending, OPEN);
|
||||
let cut = self.pending.len() - hold;
|
||||
out.push_str(&self.pending[..cut]);
|
||||
self.pending = self.pending[cut..].to_string();
|
||||
return out;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// End of stream: an unterminated block was never a tool call after all, so
|
||||
/// give the text back rather than swallowing output.
|
||||
pub fn finish(&mut self) -> String {
|
||||
let mut out = String::new();
|
||||
if self.inside {
|
||||
out.push_str(OPEN);
|
||||
self.inside = false;
|
||||
}
|
||||
out.push_str(&std::mem::take(&mut self.pending));
|
||||
out
|
||||
}
|
||||
|
||||
/// Payloads found between `<tool_call>` and `</tool_call>`.
|
||||
pub fn captured(&self) -> &[String] {
|
||||
&self.captured
|
||||
}
|
||||
}
|
||||
|
||||
/// Strip every complete `<tool_call>` block from a finished string.
|
||||
/// Returns the cleaned text and the payloads that were removed.
|
||||
pub fn split_tool_markup(text: &str) -> (String, Vec<String>) {
|
||||
let mut f = ToolMarkupFilter::new();
|
||||
let mut out = f.push(text);
|
||||
out.push_str(&f.finish());
|
||||
(out, f.captured)
|
||||
}
|
||||
|
||||
/// Turn `<tool_call>` payloads into OpenAI `tool_calls` entries, numbered from
|
||||
/// `start_index`. Payloads that are not a JSON object with a `name` are skipped —
|
||||
/// a stray marker must not become a bogus call.
|
||||
pub fn markup_to_tool_calls(payloads: &[String], start_index: usize) -> Vec<serde_json::Value> {
|
||||
let mut out = Vec::new();
|
||||
for payload in payloads {
|
||||
let Ok(v) = serde_json::from_str::<serde_json::Value>(payload.trim()) else {
|
||||
continue;
|
||||
};
|
||||
let Some(name) = v.get("name").and_then(|n| n.as_str()) else {
|
||||
continue;
|
||||
};
|
||||
// `arguments` is a JSON string on the wire; templates emit an object.
|
||||
let arguments = match v.get("arguments").or_else(|| v.get("parameters")) {
|
||||
Some(serde_json::Value::String(s)) => s.clone(),
|
||||
Some(other) => other.to_string(),
|
||||
None => "{}".to_string(),
|
||||
};
|
||||
out.push(serde_json::json!({
|
||||
"index": start_index + out.len(),
|
||||
"id": new_call_id(),
|
||||
"type": "function",
|
||||
"function": { "name": name, "arguments": arguments },
|
||||
}));
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
pub fn new_call_id() -> String {
|
||||
format!("call_{}", crate::admin::random_token(12))
|
||||
}
|
||||
|
||||
/// Bring one streamed `tool_calls` array up to the OpenAI shape.
|
||||
///
|
||||
/// `seen` tracks which choice-local indices already had their opening delta, so
|
||||
/// the identity fields (`id`, `type`) are added once and continuation deltas —
|
||||
/// which legitimately carry nothing but an argument fragment — are left alone.
|
||||
pub fn normalize_delta_tool_calls(
|
||||
raw: &[serde_json::Value],
|
||||
seen: &mut std::collections::HashSet<u64>,
|
||||
) -> Vec<serde_json::Value> {
|
||||
let mut out = Vec::new();
|
||||
for (i, tc) in raw.iter().enumerate() {
|
||||
let index = tc.get("index").and_then(|v| v.as_u64()).unwrap_or(i as u64);
|
||||
let mut entry = serde_json::Map::new();
|
||||
entry.insert("index".into(), serde_json::json!(index));
|
||||
|
||||
let first = seen.insert(index);
|
||||
if let Some(id) = tc
|
||||
.get("id")
|
||||
.and_then(|v| v.as_str())
|
||||
.filter(|s| !s.is_empty())
|
||||
{
|
||||
entry.insert("id".into(), serde_json::json!(id));
|
||||
} else if first {
|
||||
entry.insert("id".into(), serde_json::json!(new_call_id()));
|
||||
}
|
||||
if first {
|
||||
let kind = tc
|
||||
.get("type")
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("function");
|
||||
entry.insert("type".into(), serde_json::json!(kind));
|
||||
}
|
||||
|
||||
if let Some(f) = tc.get("function") {
|
||||
let mut func = serde_json::Map::new();
|
||||
if let Some(name) = f.get("name").and_then(|v| v.as_str()) {
|
||||
func.insert("name".into(), serde_json::json!(name));
|
||||
} else if first {
|
||||
func.insert("name".into(), serde_json::json!(""));
|
||||
}
|
||||
match f.get("arguments") {
|
||||
// Arguments are a string on the wire; some runtimes send the object.
|
||||
Some(serde_json::Value::String(s)) => {
|
||||
func.insert("arguments".into(), serde_json::json!(s));
|
||||
}
|
||||
Some(serde_json::Value::Null) | None => {
|
||||
if first {
|
||||
func.insert("arguments".into(), serde_json::json!(""));
|
||||
}
|
||||
}
|
||||
Some(other) => {
|
||||
func.insert("arguments".into(), serde_json::json!(other.to_string()));
|
||||
}
|
||||
}
|
||||
entry.insert("function".into(), serde_json::Value::Object(func));
|
||||
}
|
||||
out.push(serde_json::Value::Object(entry));
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
const CALL: &str = r#"<tool_call>
|
||||
{"name": "get_weather", "arguments": {"city": "Paris"}}
|
||||
</tool_call>"#;
|
||||
|
||||
#[test]
|
||||
fn markup_is_stripped_from_finished_text() {
|
||||
let (clean, payloads) = split_tool_markup(&format!("Sure!{CALL}"));
|
||||
assert_eq!(clean, "Sure!");
|
||||
assert_eq!(payloads.len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn markup_split_across_stream_chunks_never_leaks() {
|
||||
let mut f = ToolMarkupFilter::new();
|
||||
let mut seen = String::new();
|
||||
// the marker itself is broken across three fragments
|
||||
for chunk in [
|
||||
"Let me check. <tool",
|
||||
"_call>",
|
||||
"{\"name\":\"get_weather\",",
|
||||
"\"arguments\":{\"city\":\"Paris\"}}",
|
||||
"</tool",
|
||||
"_call>",
|
||||
" done",
|
||||
] {
|
||||
seen.push_str(&f.push(chunk));
|
||||
}
|
||||
seen.push_str(&f.finish());
|
||||
assert_eq!(seen, "Let me check. done");
|
||||
assert!(!seen.contains("tool_call"));
|
||||
assert_eq!(f.captured().len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn captured_markup_becomes_a_real_tool_call() {
|
||||
let (_, payloads) = split_tool_markup(CALL);
|
||||
let calls = markup_to_tool_calls(&payloads, 0);
|
||||
assert_eq!(calls.len(), 1);
|
||||
assert_eq!(calls[0]["type"], "function");
|
||||
assert_eq!(calls[0]["index"], 0);
|
||||
assert_eq!(calls[0]["function"]["name"], "get_weather");
|
||||
// arguments must be a STRING, not an object
|
||||
assert_eq!(
|
||||
calls[0]["function"]["arguments"].as_str().unwrap(),
|
||||
r#"{"city":"Paris"}"#
|
||||
);
|
||||
assert!(calls[0]["id"].as_str().unwrap().starts_with("call_"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_stray_marker_does_not_become_a_call() {
|
||||
let calls = markup_to_tool_calls(&["not json at all".into(), "{\"no_name\":1}".into()], 0);
|
||||
assert!(calls.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_unterminated_block_is_given_back_as_text() {
|
||||
let mut f = ToolMarkupFilter::new();
|
||||
let mut out = f.push("thinking <tool_call>{\"name\":\"x\"");
|
||||
out.push_str(&f.finish());
|
||||
assert_eq!(out, "thinking <tool_call>{\"name\":\"x\"");
|
||||
assert!(f.captured().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn delta_tool_calls_get_identity_once_and_object_arguments_become_strings() {
|
||||
let mut seen = std::collections::HashSet::new();
|
||||
let opening = normalize_delta_tool_calls(
|
||||
&[serde_json::json!({"index": 0, "function": {"name": "f", "arguments": {"a": 1}}})],
|
||||
&mut seen,
|
||||
);
|
||||
assert_eq!(opening[0]["type"], "function");
|
||||
assert!(opening[0]["id"].as_str().unwrap().starts_with("call_"));
|
||||
assert_eq!(opening[0]["function"]["arguments"], r#"{"a":1}"#);
|
||||
|
||||
// a continuation delta must not be handed a second id
|
||||
let cont = normalize_delta_tool_calls(
|
||||
&[serde_json::json!({"index": 0, "function": {"arguments": "\"}"}})],
|
||||
&mut seen,
|
||||
);
|
||||
assert!(cont[0].get("id").is_none());
|
||||
assert!(cont[0].get("type").is_none());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user