From 9a49fb2bb37f2c6f7396800d53058a32676b8479 Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Sun, 1 Nov 2026 13:41:34 +0900 Subject: [PATCH] fix(clarification): keep each question when a provider reuses a call ID (#7762) * fix(clarification): keep each question when a provider reuses a call ID The clarification ToolMessage ID, which is also the human-input request_id, was clarification:{tool_call_id}. Providers that number calls per response reuse call_0 across turns, so a later question took the earlier one's ID: add_messages replaced the earlier question in place and left the new call without a result. Add a digest of the issuing AIMessage ID. A retry of the same call comes from the same AIMessage and keeps its ID; without that message in state the ID keeps its previous form. * docs(changelog): reference #7762 for the clarification call ID fix --- CHANGELOG.md | 9 ++ CHANGELOG_zh.md | 6 + .../middlewares/clarification_middleware.py | 48 +++++-- .../tests/test_clarification_middleware.py | 49 +++++++ ...cation_reused_call_id_graph_integration.py | 124 ++++++++++++++++++ 5 files changed, 227 insertions(+), 9 deletions(-) create mode 100644 backend/tests/test_clarification_reused_call_id_graph_integration.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c9e6c2d7ad..12d523c3ac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,14 @@ This section accumulates work toward the **2.3.0** milestone ### Fixed +- **agents:** A clarification no longer overwrites an earlier question when the + provider reuses a tool-call ID across turns (for example `call_0` on every + response). Its message ID, which is also the human-input `request_id`, named + only the call ID, so the later question replaced the earlier one in place and + the later call was left without a result. The ID now also names the + AIMessage that issued the call; a retry of the same call keeps its ID. + ([#7762]) + - **projects:** The trash retention sweep no longer deletes live document originals when the data directory, `users/`, a user's directory or a user's `projects/` is a symlink. It protected each document by its resolved @@ -15957,3 +15965,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7727]: https://github.com/bytedance/deer-flow/pull/7727 [#7729]: https://github.com/bytedance/deer-flow/pull/7729 [#7732]: https://github.com/bytedance/deer-flow/pull/7732 +[#7762]: https://github.com/bytedance/deer-flow/pull/7762 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index b240372e8e..c4431785c6 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -20,6 +20,11 @@ ### 修复 +- **智能体：** 当模型提供方在不同轮次复用同一个工具调用 ID（例如每次响应都是 `call_0`）时， + 澄清提问不再覆盖之前的问题。此前其消息 ID（同时也是人工输入的 `request_id`）只包含调用 ID， + 后一个问题会原地替换前一个问题，后一次调用也因此缺少结果。现在 ID 还会包含发起该调用的 + AIMessage；同一调用的重试仍保持原 ID。([#7762]) + - **项目：** 当数据目录、`users/`、某个用户目录或用户的 `projects/` 是符号链接时， 回收站保留期清扫不再删除仍在使用的文档原件。此前它按解析后的路径保护每个文档， 却遍历未解析的目录树，导致没有任何文档能匹配上，所有超过 24 小时的原件都被当作 @@ -12719,3 +12724,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7727]: https://github.com/bytedance/deer-flow/pull/7727 [#7729]: https://github.com/bytedance/deer-flow/pull/7729 [#7732]: https://github.com/bytedance/deer-flow/pull/7732 +[#7762]: https://github.com/bytedance/deer-flow/pull/7762 diff --git a/backend/packages/harness/deerflow/agents/middlewares/clarification_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/clarification_middleware.py index 712aae7ea5..d05b66fb4c 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/clarification_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/clarification_middleware.py @@ -3,7 +3,7 @@ import json import logging import re -from collections.abc import Callable +from collections.abc import Callable, Mapping from hashlib import sha256 from typing import Any, override @@ -96,12 +96,42 @@ class ClarificationMiddleware(AgentMiddleware[ClarificationMiddlewareState]): state_schema = ClarificationMiddlewareState - def _stable_message_id(self, tool_call_id: str, formatted_message: str) -> str: - """Build a deterministic message ID so retried clarification calls replace, not append.""" - if tool_call_id: - return f"clarification:{tool_call_id}" - digest = sha256(formatted_message.encode("utf-8")).hexdigest()[:16] - return f"clarification:{digest}" + def _stable_message_id(self, tool_call_id: str, formatted_message: str, issuing_message_id: str | None = None) -> str: + """Build a deterministic message ID so retried clarification calls replace, not append. + + The ID doubles as the human-input ``request_id``. Providers may reuse + call IDs across turns, so the ID also names the AIMessage that issued + the call: a later question must not take an earlier one's ID, or + ``add_messages`` would replace the earlier question in place and leave + the new call without a result. A retry of the same call comes from the + same AIMessage and keeps its ID. + """ + base = tool_call_id or sha256(formatted_message.encode("utf-8")).hexdigest()[:16] + if not issuing_message_id: + return f"clarification:{base}" + # A digest rather than the raw AIMessage ID keeps the request ID short + # enough for keys built from it, such as the 128-character + # notification source key. + return f"clarification:{base}:{sha256(issuing_message_id.encode('utf-8')).hexdigest()[:12]}" + + @staticmethod + def _issuing_message_id(request: ToolCallRequest) -> str | None: + """ID of the latest AIMessage in state that carries this tool call, if any. + + Message IDs are unique within state, since ``add_messages`` merges + messages that share one, so this identifies the call's turn. + """ + state = getattr(request, "state", None) + if not isinstance(state, Mapping): + return None + messages = state.get("messages") + if not isinstance(messages, list): + return None + tool_call_id = request.tool_call.get("id") or "" + for message in reversed(messages): + if isinstance(message, AIMessage) and any((tool_call.get("id") or "") == tool_call_id for tool_call in message.tool_calls): + return message.id or None + return None def _normalize_options(self, raw_options: Any) -> list[str]: """Normalize tool-provided options into displayable string values.""" @@ -457,7 +487,7 @@ class ClarificationMiddleware(AgentMiddleware[ClarificationMiddlewareState]): tool_call_id = request.tool_call.get("id", "") logger.info("ask_clarification suppressed by run interaction policy; applying unattended risk guidance") return ToolMessage( - id=self._stable_message_id(tool_call_id, "proceed-without-clarification"), + id=self._stable_message_id(tool_call_id, "proceed-without-clarification", self._issuing_message_id(request)), content=( "Clarification is disabled in this context — the human is not present " "to answer synchronously. Do not ask for confirmation or wait for a human response. " @@ -497,7 +527,7 @@ class ClarificationMiddleware(AgentMiddleware[ClarificationMiddlewareState]): # Get the tool call ID tool_call_id = request.tool_call.get("id", "") - request_id = self._stable_message_id(tool_call_id, formatted_message) + request_id = self._stable_message_id(tool_call_id, formatted_message, self._issuing_message_id(request)) human_input_payload = self._build_human_input_payload(args, tool_call_id=tool_call_id, request_id=request_id, fields=fields) # Create a ToolMessage with the formatted question diff --git a/backend/tests/test_clarification_middleware.py b/backend/tests/test_clarification_middleware.py index d11702bf1f..37a8ce7526 100644 --- a/backend/tests/test_clarification_middleware.py +++ b/backend/tests/test_clarification_middleware.py @@ -4,6 +4,7 @@ import json from types import SimpleNamespace import pytest +from langchain_core.messages import AIMessage, HumanMessage from langgraph.graph.message import add_messages from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware @@ -674,6 +675,54 @@ class TestClarificationCommandIdempotency: assert merged[0].content == first_message.content assert merged[0].artifact == first_message.artifact + @staticmethod + def _turn(ai_message_id, call_id, question): + return AIMessage( + id=ai_message_id, + content="", + tool_calls=[{"name": "ask_clarification", "id": call_id, "args": {"question": question, "clarification_type": "missing_info"}}], + ) + + @staticmethod + def _result(middleware, messages, *, disabled): + call = messages[-1].tool_calls[0] + request = SimpleNamespace( + tool_call=call, + state={"messages": messages}, + runtime=SimpleNamespace(context={"disable_clarification": True} if disabled else {}), + ) + result = middleware.wrap_tool_call(request, lambda _req: pytest.fail("handler should not be called")) + return result if disabled else result.update["messages"][0] + + @pytest.mark.parametrize("disabled", [False, True], ids=["interactive", "unattended"]) + def test_retry_from_the_same_ai_message_keeps_its_message_id(self, middleware, disabled): + messages = [HumanMessage(content="Plan a trip"), self._turn("ai-1", "call_0", "Which city?")] + + first = self._result(middleware, messages, disabled=disabled) + second = self._result(middleware, messages, disabled=disabled) + + assert second.id == first.id + assert len(add_messages(add_messages(messages, [first]), [second])) == 3 + + @pytest.mark.parametrize("call_id", ["call_0", ""], ids=["reused-id", "missing-id"]) + @pytest.mark.parametrize("disabled", [False, True], ids=["interactive", "unattended"]) + def test_reused_call_id_in_a_later_turn_gets_its_own_message_id(self, middleware, call_id, disabled): + """Providers that number calls per response reuse ``call_0`` across turns.""" + history = [HumanMessage(content="Plan a trip"), self._turn("ai-1", call_id, "Which city?")] + first = self._result(middleware, history, disabled=disabled) + history = add_messages(history, [first, HumanMessage(content="Seoul"), self._turn("ai-2", call_id, "Which city?")]) + + second = self._result(middleware, history, disabled=disabled) + merged = add_messages(history, [second]) + + assert second.id != first.id + # The earlier question stays in place and the new call gets its result. + assert [message.id for message in merged] == [merged[0].id, "ai-1", first.id, merged[3].id, "ai-2", second.id] + assert merged[2].content == first.content + if not disabled: + assert first.artifact["human_input"]["request_id"] == first.id + assert second.artifact["human_input"]["request_id"] == second.id + def test_tool_message_model_dump_preserves_human_input_artifact(self, middleware): request = SimpleNamespace( tool_call={ diff --git a/backend/tests/test_clarification_reused_call_id_graph_integration.py b/backend/tests/test_clarification_reused_call_id_graph_integration.py new file mode 100644 index 0000000000..93bc9bcb1d --- /dev/null +++ b/backend/tests/test_clarification_reused_call_id_graph_integration.py @@ -0,0 +1,124 @@ +"""Graph-level test for clarifications whose provider reuses a call ID. + +Some providers number tool calls per response (``call_0``), so two turns can +issue ``ask_clarification`` under the same call ID. Each question's ToolMessage +must keep its own ID in a real ``create_agent`` graph with a checkpointer: +``add_messages`` replaces a message that shares an ID in place, which would +overwrite the earlier question and leave the later call without a result. +""" + +from __future__ import annotations + +from langchain.agents import create_agent +from langchain_core.language_models import BaseChatModel +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage +from langchain_core.outputs import ChatGeneration, ChatResult +from langgraph.checkpoint.memory import InMemorySaver + +from deerflow.agents.human_input import HUMAN_INPUT_RESPONSE_KEY, extract_unanswered_human_input_request +from deerflow.agents.middlewares.clarification_middleware import ClarificationMiddleware +from deerflow.tools.builtins.clarification_tool import ask_clarification_tool + +_REUSED_CALL_ID = "call_0" + + +def _ask(question: str) -> AIMessage: + return AIMessage( + content="", + tool_calls=[ + { + "id": _REUSED_CALL_ID, + "name": "ask_clarification", + "args": {"question": question, "clarification_type": "missing_info"}, + } + ], + ) + + +class _ScriptedModel(BaseChatModel): + """Replays ``replies`` in order and records the messages of each call.""" + + replies: list[AIMessage] + seen: list[list] = [] + + @property + def _llm_type(self) -> str: + return "fake-reused-call-id" + + def bind_tools(self, tools, **kwargs): + return self + + def _generate(self, messages, stop=None, run_manager=None, **kwargs): + self.seen.append(list(messages)) + return ChatResult(generations=[ChatGeneration(message=self.replies[len(self.seen) - 1])]) + + async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs): + return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs) + + +def _answer(request: dict[str, str], value: str) -> HumanMessage: + return HumanMessage( + content=value, + additional_kwargs={ + HUMAN_INPUT_RESPONSE_KEY: { + "version": 1, + "kind": "human_input_response", + "source": "ask_clarification", + "request_id": request["request_id"], + "response_kind": "text", + "value": value, + } + }, + ) + + +def _assert_every_call_answered_in_place(messages: list) -> None: + """Each tool call is followed by its own result, as providers require.""" + for index, message in enumerate(messages): + if isinstance(message, AIMessage) and message.tool_calls: + result = messages[index + 1] + assert isinstance(result, ToolMessage), f"call at {index} has no result: {messages}" + assert result.tool_call_id == _REUSED_CALL_ID + + +def test_reused_call_id_keeps_each_clarification_question(): + model = _ScriptedModel(replies=[_ask("Which city?"), _ask("How many days?"), AIMessage(content="done")], seen=[]) + agent = create_agent(model=model, tools=[ask_clarification_tool], middleware=[ClarificationMiddleware()], checkpointer=InMemorySaver()) + config = {"configurable": {"thread_id": "reused-call-id"}} + + agent.invoke({"messages": [HumanMessage(content="Plan a trip")]}, config) + first = extract_unanswered_human_input_request(agent.get_state(config).values["messages"]) + assert first is not None and first["question"] == "Which city?" + + agent.invoke({"messages": [_answer(first, "Seoul")]}, config) + second = extract_unanswered_human_input_request(agent.get_state(config).values["messages"]) + assert second is not None and second["question"] == "How many days?" + assert second["request_id"] != first["request_id"] + + agent.invoke({"messages": [_answer(second, "3 days")]}, config) + + messages = agent.get_state(config).values["messages"] + _assert_every_call_answered_in_place(messages) + questions = [message.artifact["human_input"]["question"] for message in messages if isinstance(message, ToolMessage)] + assert questions == ["Which city?", "How many days?"] + # The model sees both questions next to the answers given to them. + final_call = model.seen[-1] + _assert_every_call_answered_in_place(final_call) + assert [message.content for message in final_call if isinstance(message, HumanMessage)] == ["Plan a trip", "Seoul", "3 days"] + + +def test_reused_call_id_keeps_each_unattended_clarification_result(): + model = _ScriptedModel(replies=[_ask("Which city?"), _ask("How many days?"), AIMessage(content="done")], seen=[]) + agent = create_agent(model=model, tools=[ask_clarification_tool], middleware=[ClarificationMiddleware()], checkpointer=InMemorySaver()) + config = {"configurable": {"thread_id": "reused-call-id-unattended"}} + unattended = {"disable_clarification": True} + + # ``ask_clarification`` returns directly, so each unattended run ends at + # its "proceed" result and the next run continues the thread. + for text in ["Plan a trip", "Continue", "Continue"]: + agent.invoke({"messages": [HumanMessage(content=text)]}, config, context=unattended) + + messages = agent.get_state(config).values["messages"] + _assert_every_call_answered_in_place(messages) + assert len([message for message in messages if isinstance(message, ToolMessage)]) == 2 + _assert_every_call_answered_in_place(model.seen[-1])