Accumulate streamed tool calls across chunks in Ollama streaming

Ollama >=0.8 can stream tool_calls incrementally across NDJSON chunks;
chat_with_tools_stream did `tool_calls = Some(tcs)` per chunk, so only
the last chunk's calls survived assembly and earlier calls were silently
dropped. Append into the accumulator instead.

- ollama: append_streamed_tool_calls helper + tests covering two calls
  arriving in separate chunks and the single-chunk batch case.
- llamacpp: the SSE delta assembly was already correct (per-index
  BTreeMap, same-index argument fragments concatenate, distinct indexes
  accumulate); extracted it into apply_tool_call_deltas /
  finalize_tool_calls and added tests pinning that behavior.
- llm_client: new shared strip_think_blocks (moved from ollama's private
  extract_final_answer, which now delegates) so the tool-calling final
  content paths can reuse it; unit tests for tagged/plain/unclosed/empty
  cases.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Cameron CordesandClaude Fable 5 committed 2026-06-09 18:29:06 -04:00
1 parent 8e4f91561b
commit 592dfcb42c
3 files changed
+223 -66

No files matched your search

+119 -52
View File
@@ -590,10 +590,7 @@ impl LlmClient for LlamaCppClient {
let mut byte_stream = byte_stream;
let mut buf: Vec<u8> = Vec::new();
let mut accumulated_content = String::new();
let mut tool_state: std::collections::BTreeMap<
usize,
(Option<String>, Option<String>, String),
> = std::collections::BTreeMap::new();
let mut tool_state = ToolCallAssembly::new();
let mut role = "assistant".to_string();
let mut prompt_tokens: Option<i32> = None;
let mut completion_tokens: Option<i32> = None;
@@ -670,32 +667,7 @@ impl LlmClient for LlamaCppClient {
yield Ok(LlmStreamEvent::TextDelta(content.to_string()));
}
if let Some(tcs) = delta.get("tool_calls").and_then(|v| v.as_array()) {
for tc_delta in tcs {
let idx = tc_delta
.get("index")
.and_then(|n| n.as_u64())
.unwrap_or(0) as usize;
let entry = tool_state
.entry(idx)
.or_insert((None, None, String::new()));
if let Some(id) =
tc_delta.get("id").and_then(|v| v.as_str())
{
entry.0 = Some(id.to_string());
}
if let Some(func) = tc_delta.get("function") {
if let Some(name) =
func.get("name").and_then(|v| v.as_str())
{
entry.1 = Some(name.to_string());
}
if let Some(args) =
func.get("arguments").and_then(|v| v.as_str())
{
entry.2.push_str(args);
}
}
}
apply_tool_call_deltas(&mut tool_state, tcs);
}
}
if done_seen {
@@ -707,28 +679,7 @@ impl LlmClient for LlamaCppClient {
}
}
let tool_calls: Option<Vec<ToolCall>> = if tool_state.is_empty() {
None
} else {
let mut v = Vec::with_capacity(tool_state.len());
for (_idx, (id, name, args)) in tool_state {
let arguments: Value = if args.trim().is_empty() {
Value::Object(Default::default())
} else {
serde_json::from_str(&args).unwrap_or_else(|_| {
Value::Object(Default::default())
})
};
v.push(ToolCall {
id,
function: ToolCallFunction {
name: name.unwrap_or_default(),
arguments,
},
});
}
Some(v)
};
let tool_calls = finalize_tool_calls(tool_state);
if let Some(ref frame) = last_frame {
log_timings(frame, prompt_tokens, completion_tokens);
@@ -937,6 +888,58 @@ fn extract_error_detail(parsed: &Value) -> String {
raw.chars().take(300).collect()
}
/// Per-index assembly state for streamed OpenAI-style tool-call deltas:
/// `index → (id, name, concatenated argument fragments)`. BTreeMap so the
/// finalized calls come out in index order.
type ToolCallAssembly = std::collections::BTreeMap<usize, (Option<String>, Option<String>, String)>;
/// Fold one SSE frame's `delta.tool_calls` array into the assembly state.
/// Deltas carrying the same `index` merge into one call (llama.cpp streams a
/// call's argument JSON in fragments — they concatenate); distinct indexes
/// accumulate as separate calls.
fn apply_tool_call_deltas(state: &mut ToolCallAssembly, tcs: &[Value]) {
for tc_delta in tcs {
let idx = tc_delta.get("index").and_then(|n| n.as_u64()).unwrap_or(0) as usize;
let entry = state.entry(idx).or_insert((None, None, String::new()));
if let Some(id) = tc_delta.get("id").and_then(|v| v.as_str()) {
entry.0 = Some(id.to_string());
}
if let Some(func) = tc_delta.get("function") {
if let Some(name) = func.get("name").and_then(|v| v.as_str()) {
entry.1 = Some(name.to_string());
}
if let Some(args) = func.get("arguments").and_then(|v| v.as_str()) {
entry.2.push_str(args);
}
}
}
}
/// Convert assembled tool-call state into canonical `ToolCall`s, parsing each
/// call's concatenated argument JSON (empty / malformed → `{}`). `None` when
/// no tool-call deltas arrived.
fn finalize_tool_calls(state: ToolCallAssembly) -> Option<Vec<ToolCall>> {
if state.is_empty() {
return None;
}
let mut v = Vec::with_capacity(state.len());
for (_idx, (id, name, args)) in state {
let arguments: Value = if args.trim().is_empty() {
Value::Object(Default::default())
} else {
serde_json::from_str(&args).unwrap_or_else(|_| Value::Object(Default::default()))
};
v.push(ToolCall {
id,
function: ToolCallFunction {
name: name.unwrap_or_default(),
arguments,
},
});
}
Some(v)
}
fn find_double_newline(buf: &[u8]) -> Option<usize> {
for i in 0..buf.len().saturating_sub(1) {
if buf[i] == b'\n' && buf[i + 1] == b'\n' {
@@ -1302,4 +1305,68 @@ mod tests {
let c = LlamaCppClient::new(None, None);
assert_eq!(c.tts_model, "chatterbox");
}
#[test]
fn stream_assembly_keeps_two_tool_calls_from_separate_chunks() {
// llama.cpp emits one delta per SSE frame; two calls with distinct
// `index` values arriving in separate frames must BOTH survive.
let mut state = ToolCallAssembly::new();
apply_tool_call_deltas(
&mut state,
&[json!({
"index": 0,
"id": "call_a",
"function": { "name": "get_sms_messages", "arguments": "{\"date\":\"2019-01-01\"}" }
})],
);
apply_tool_call_deltas(
&mut state,
&[json!({
"index": 1,
"id": "call_b",
"function": { "name": "reverse_geocode", "arguments": "{\"latitude\":1.0,\"longitude\":2.0}" }
})],
);
let calls = finalize_tool_calls(state).expect("two calls assembled");
assert_eq!(calls.len(), 2);
assert_eq!(calls[0].id.as_deref(), Some("call_a"));
assert_eq!(calls[0].function.name, "get_sms_messages");
assert_eq!(calls[0].function.arguments["date"], "2019-01-01");
assert_eq!(calls[1].id.as_deref(), Some("call_b"));
assert_eq!(calls[1].function.name, "reverse_geocode");
assert_eq!(calls[1].function.arguments["latitude"], 1.0);
}
#[test]
fn stream_assembly_concatenates_argument_fragments_for_same_index() {
// A single call's argument JSON streamed across frames concatenates
// into one parseable document.
let mut state = ToolCallAssembly::new();
apply_tool_call_deltas(
&mut state,
&[json!({
"index": 0,
"id": "call_x",
"function": { "name": "search_messages", "arguments": "{\"query\":" }
})],
);
apply_tool_call_deltas(
&mut state,
&[json!({
"index": 0,
"function": { "arguments": "\"dinner\"}" }
})],
);
let calls = finalize_tool_calls(state).expect("one call assembled");
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].function.name, "search_messages");
assert_eq!(calls[0].function.arguments["query"], "dinner");
}
#[test]
fn stream_assembly_empty_state_finalizes_to_none() {
assert!(finalize_tool_calls(ToolCallAssembly::new()).is_none());
}
}