From 70212c0c32b9975598a78ec11f1f5605a2c28be1 Mon Sep 17 00:00:00 2001 From: LHMQ878 <72402929@cityu-dg.edu.cn> Date: Fri, 14 Aug 2026 14:15:27 +0800 Subject: [PATCH] fix: assemble Anthropic streamed tool_use blocks before replay Co-authored-by: Cursor --- .../flow/anthropic_tool_content.py | 135 ++++++++++++++++++ apps/application/flow/tests/__init__.py | 0 .../flow/tests/test_anthropic_tool_content.py | 64 +++++++++ apps/application/flow/tools.py | 30 +++- 4 files changed, 227 insertions(+), 2 deletions(-) create mode 100644 apps/application/flow/anthropic_tool_content.py create mode 100644 apps/application/flow/tests/__init__.py create mode 100644 apps/application/flow/tests/test_anthropic_tool_content.py diff --git a/apps/application/flow/anthropic_tool_content.py b/apps/application/flow/anthropic_tool_content.py new file mode 100644 index 00000000000..9827447e635 --- /dev/null +++ b/apps/application/flow/anthropic_tool_content.py @@ -0,0 +1,135 @@ +# coding=utf-8 +"""Assemble Anthropic streamed tool_use blocks before they are replayed. + +Anthropic streams tool arguments as `input_json_delta` events. Those deltas are +valid on the wire, but they are not legal `messages[].content` block types. +Replaying them on the next turn yields: + + Input tag 'input_json_delta' found using 'type' does not match any of the expected tags + +Empty `text` blocks from the same stream also fail with: + + messages: text content blocks must be non-empty +""" + +from __future__ import annotations + +import json + +ANTHROPIC_TOOL_STOP_REASONS = ("tool_use", "end_turn") +_STREAM_DELTA_TYPES = {"input_json_delta"} + + +def is_anthropic_tool_finish(response_metadata, chunk_position=None) -> bool: + """True when this chunk closes an Anthropic (or OpenAI-mapped) tool turn.""" + meta = response_metadata or {} + if meta.get("finish_reason") == "tool_calls": + return True + if meta.get("stop_reason") in ANTHROPIC_TOOL_STOP_REASONS: + return True + return chunk_position == "last" + + +def collect_input_json_deltas(content) -> dict: + """Concatenate `input_json_delta` fragments keyed by content-block index.""" + collected = {} + if not isinstance(content, list): + return collected + for block in content: + if not isinstance(block, dict) or block.get("type") != "input_json_delta": + continue + index = block.get("index") + if index is None: + continue + piece = block.get("partial_json") + if piece is None: + piece = block.get("input") or "" + if not isinstance(piece, str): + piece = json.dumps(piece, ensure_ascii=False) if piece else "" + collected[index] = collected.get(index, "") + piece + return collected + + +def _parse_tool_input(raw): + if raw is None or raw == "": + return None + if isinstance(raw, (dict, list)): + return raw + if not isinstance(raw, str): + return None + try: + parsed = json.loads(raw) + except (json.JSONDecodeError, TypeError, ValueError): + return None + return parsed if isinstance(parsed, (dict, list)) else None + + +def _input_from_tool_calls(block, tool_calls): + block_id = block.get("id") + for tool_call in tool_calls or []: + if not isinstance(tool_call, dict): + continue + if block_id and tool_call.get("id") == block_id: + return _parse_tool_input(tool_call.get("args") or tool_call.get("arguments")) + if block.get("index") is not None and tool_call.get("index") == block.get("index"): + return _parse_tool_input(tool_call.get("args") or tool_call.get("arguments")) + return None + + +def _input_from_fragments(block, fragments): + if not fragments: + return None + block_id = block.get("id") + block_index = block.get("index") + for entry in fragments.values(): + if not isinstance(entry, dict): + continue + if block_id and entry.get("id") == block_id: + return _parse_tool_input(entry.get("arguments")) + if block_index is not None and entry.get("index") == block_index: + return _parse_tool_input(entry.get("arguments")) + return None + + +def _is_empty_text_block(block) -> bool: + if not isinstance(block, dict): + return False + if block.get("type") not in ("text", "text_delta"): + return False + return not str(block.get("text") or "").strip() + + +def finalize_anthropic_assistant_content(content, tool_calls=None, fragments=None): + """Drop streamed deltas and fill completed `tool_use.input` values. + + Non-list content (plain strings) is returned unchanged. + """ + if not isinstance(content, list): + return content + + json_by_index = collect_input_json_deltas(content) + finalized = [] + for block in content: + if not isinstance(block, dict): + finalized.append(block) + continue + if block.get("type") in _STREAM_DELTA_TYPES: + continue + if _is_empty_text_block(block): + continue + if block.get("type") == "tool_use": + new_block = dict(block) + filled = ( + _input_from_fragments(new_block, fragments) + or _input_from_tool_calls(new_block, tool_calls) + or _parse_tool_input(json_by_index.get(new_block.get("index"))) + ) + current = new_block.get("input") + if filled is not None and (current in (None, "", {}, []) or not current): + new_block["input"] = filled + elif current == "": + new_block["input"] = {} + finalized.append(new_block) + continue + finalized.append(block) + return finalized diff --git a/apps/application/flow/tests/__init__.py b/apps/application/flow/tests/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/apps/application/flow/tests/test_anthropic_tool_content.py b/apps/application/flow/tests/test_anthropic_tool_content.py new file mode 100644 index 00000000000..7605826eb62 --- /dev/null +++ b/apps/application/flow/tests/test_anthropic_tool_content.py @@ -0,0 +1,64 @@ +import sys +import unittest +from pathlib import Path + +# Allow `python -m unittest` without Django by putting `apps/` on sys.path. +sys.path.insert(0, str(Path(__file__).resolve().parents[3])) + +from application.flow.anthropic_tool_content import ( + collect_input_json_deltas, + finalize_anthropic_assistant_content, + is_anthropic_tool_finish, +) + + +class AnthropicToolContentTests(unittest.TestCase): + def test_finish_detects_anthropic_stop_reason(self): + self.assertTrue(is_anthropic_tool_finish({"stop_reason": "tool_use"})) + self.assertTrue(is_anthropic_tool_finish({"finish_reason": "tool_calls"})) + self.assertTrue(is_anthropic_tool_finish({}, chunk_position="last")) + self.assertFalse(is_anthropic_tool_finish({"stop_reason": "pause_turn"})) + + def test_strips_input_json_delta_and_empty_text(self): + content = [ + {"type": "text", "text": ""}, + {"type": "tool_use", "id": "toolu_1", "name": "live_price", "input": {}, "index": 1}, + {"type": "input_json_delta", "index": 1, "partial_json": '{"query":'}, + {"type": "input_json_delta", "index": 1, "partial_json": ' "DJI Neo"}'}, + ] + finalized = finalize_anthropic_assistant_content(content) + self.assertEqual(len(finalized), 1) + self.assertEqual(finalized[0]["type"], "tool_use") + self.assertEqual(finalized[0]["id"], "toolu_1") + self.assertEqual(finalized[0]["input"], {"query": "DJI Neo"}) + self.assertNotIn("input_json_delta", [b.get("type") for b in finalized]) + + def test_fills_tool_use_input_from_tool_calls(self): + content = [ + {"type": "tool_use", "id": "toolu_2", "name": "live_price", "input": ""}, + ] + tool_calls = [{"id": "toolu_2", "name": "live_price", "args": {"query": "in stock"}}] + finalized = finalize_anthropic_assistant_content(content, tool_calls=tool_calls) + self.assertEqual(finalized[0]["input"], {"query": "in stock"}) + + def test_fills_tool_use_input_from_fragments(self): + content = [ + {"type": "tool_use", "id": "toolu_3", "name": "live_price", "input": {}}, + ] + fragments = {"1": {"id": "toolu_3", "name": "live_price", "arguments": '{"query": "GEL"}'}} + finalized = finalize_anthropic_assistant_content(content, fragments=fragments) + self.assertEqual(finalized[0]["input"], {"query": "GEL"}) + + def test_collect_input_json_deltas_concatenates_partial_json(self): + content = [ + {"type": "input_json_delta", "index": 0, "partial_json": '{"a":'}, + {"type": "input_json_delta", "index": 0, "input": " 1}"}, + ] + self.assertEqual(collect_input_json_deltas(content), {0: '{"a": 1}'}) + + def test_plain_string_content_is_unchanged(self): + self.assertEqual(finalize_anthropic_assistant_content("hello"), "hello") + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/application/flow/tools.py b/apps/application/flow/tools.py index 42fc2ca2241..2883c487237 100644 --- a/apps/application/flow/tools.py +++ b/apps/application/flow/tools.py @@ -46,6 +46,11 @@ from langchain_core.tools import StructuredTool from langchain_core.utils._merge import merge_lists as _original_merge_lists from langchain_mcp_adapters.client import MultiServerMCPClient +from .anthropic_tool_content import ( + collect_input_json_deltas, + finalize_anthropic_assistant_content, + is_anthropic_tool_finish, +) from langgraph.checkpoint.memory import MemorySaver from maxkb.const import CONFIG from pydantic import Field, create_model @@ -589,11 +594,24 @@ def _upsert_fragment(key, raw_id, func_name, part_args): key = _get_fragment_key(tool_call.get("index"), raw_id) _upsert_fragment(key, raw_id, func_name, part_args) + # ---------------------------------------------------------------- + # 2.1 Anthropic streams tool arguments as input_json_delta content + # blocks. Fold those fragments into _tool_fragments so finish + # handling can assemble a legal tool_use.input. + # ---------------------------------------------------------------- + if isinstance(chunk[0].content, list): + for index, partial_json in collect_input_json_deltas(chunk[0].content).items(): + key = _get_fragment_key(index, None) + if key is not None: + _upsert_fragment(key, None, None, partial_json) + # ---------------------------------------------------------------- # 3. 检测工具调用结束，更新 tool_calls_info + # Anthropic uses stop_reason=tool_use rather than OpenAI's + # finish_reason=tool_calls. # ---------------------------------------------------------------- - is_finish_chunk = ( - chunk[0].response_metadata.get("finish_reason") == "tool_calls" or chunk[0].chunk_position == "last" + is_finish_chunk = is_anthropic_tool_finish( + chunk[0].response_metadata, chunk[0].chunk_position ) if is_finish_chunk: @@ -675,6 +693,14 @@ def _upsert_fragment(key, raw_id, func_name, part_args): fixed_tool_calls.append(tc) chunk[0].additional_kwargs["tool_calls"] = fixed_tool_calls + # Anthropic: never replay input_json_delta / empty text blocks. + if isinstance(chunk[0].content, list): + chunk[0].content = finalize_anthropic_assistant_content( + chunk[0].content, + tool_calls=chunk[0].tool_calls, + fragments=_tool_fragments, + ) + yield chunk[0] if mcp_output_enable and isinstance(chunk[0], ToolMessage):