mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(subagents): exclude ambiguous tool calls from parent context snapshots (#7372)
Co-authored-by: Cordis798 <284918518+Cordis798@users.noreply.github.com>
This commit is contained in:
parent
f481c7e19f
commit
4583c46311
@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from collections import Counter
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
@ -118,11 +119,14 @@ class ParentContextSnapshot:
|
||||
# Providers may reuse call ids across turns. Match each result to the
|
||||
# preceding call, including hidden frames so their results cannot be
|
||||
# reassigned to a visible call. Only retained pairs count as completed.
|
||||
call_positions: dict[str, int] = {}
|
||||
call_positions: dict[str, int | None] = {}
|
||||
completed_calls: set[tuple[int, str]] = set()
|
||||
for index, message in enumerate(messages):
|
||||
if isinstance(message, AIMessage):
|
||||
call_positions.update((call["id"], index) for call in message.tool_calls)
|
||||
# A result cannot identify one of several calls with the same
|
||||
# id in this message. Claim the id without completing any call,
|
||||
# so it cannot fall back to an earlier occurrence either.
|
||||
call_positions.update((call_id, index if count == 1 else None) for call_id, count in Counter(call["id"] for call in message.tool_calls).items())
|
||||
elif isinstance(message, ToolMessage):
|
||||
call_index = call_positions.pop(message.tool_call_id, None)
|
||||
if call_index is not None and call_index in retained_positions and index in retained_positions:
|
||||
|
||||
@ -371,6 +371,51 @@ def test_hidden_tool_frames_do_not_complete_visible_calls(hidden_part, tool_name
|
||||
assert ("TOOL_RESULT" in snapshot.content_json) is (hidden_part == "call")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("result_count", [1, 2])
|
||||
@pytest.mark.parametrize("same_name", [False, True])
|
||||
def test_snapshot_omits_ambiguous_calls_without_losing_other_history(result_count, same_name):
|
||||
messages = [
|
||||
AIMessage(content="", tool_calls=[{"name": "bash", "args": {"input": "EARLIER_COMPLETED"}, "id": "reused"}]),
|
||||
ToolMessage(content="Earlier result", tool_call_id="reused"),
|
||||
AIMessage(
|
||||
content="Keep this explanation",
|
||||
tool_calls=[
|
||||
{"name": "bash", "args": {"input": "AMBIGUOUS_FIRST"}, "id": "reused"},
|
||||
{"name": "bash" if same_name else "task", "args": {"input": "AMBIGUOUS_SECOND"}, "id": "reused"},
|
||||
{"name": "read_file", "args": {"path": "UNIQUE_COMPLETED"}, "id": "unique"},
|
||||
],
|
||||
),
|
||||
*[ToolMessage(content=f"Observed result {index}", tool_call_id="reused") for index in range(result_count)],
|
||||
ToolMessage(content="Unique result", tool_call_id="unique"),
|
||||
]
|
||||
before = [message.model_dump() for message in messages]
|
||||
snapshot = ParentContextSnapshot.from_state({"messages": messages})
|
||||
assert "AMBIGUOUS" not in snapshot.content_json
|
||||
assert "EARLIER_COMPLETED" in snapshot.content_json
|
||||
assert "UNIQUE_COMPLETED" in snapshot.content_json
|
||||
assert "Keep this explanation" in snapshot.content_json
|
||||
assert "Observed result 0" in snapshot.content_json
|
||||
assert [message.model_dump() for message in messages] == before
|
||||
|
||||
|
||||
def test_ambiguous_calls_do_not_reassign_results_to_an_earlier_pending_call():
|
||||
messages = [
|
||||
AIMessage(content="", tool_calls=[{"name": "task", "args": {"input": "EARLIER_PENDING"}, "id": "reused"}]),
|
||||
AIMessage(
|
||||
content="Current history",
|
||||
tool_calls=[
|
||||
{"name": "bash", "args": {"input": "AMBIGUOUS_FIRST"}, "id": "reused"},
|
||||
{"name": "task", "args": {"input": "AMBIGUOUS_SECOND"}, "id": "reused"},
|
||||
],
|
||||
),
|
||||
ToolMessage(content="Current result", tool_call_id="reused"),
|
||||
]
|
||||
snapshot = ParentContextSnapshot.from_state({"messages": messages})
|
||||
assert "EARLIER_PENDING" not in snapshot.content_json
|
||||
assert "AMBIGUOUS" not in snapshot.content_json
|
||||
assert "Current result" in snapshot.content_json
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tool_name", ["task", "batch_task", "write_file", "bash", "custom_lookup"])
|
||||
@pytest.mark.parametrize("result_state", ["pending", "success", "error", "hidden"])
|
||||
def test_snapshot_keeps_only_result_paired_calls_in_mixed_dispatch(tool_name, result_state):
|
||||
|
||||
@ -1098,6 +1098,49 @@ def test_task_tool_context_mode_captures_dispatch_time_history(monkeypatch, cont
|
||||
assert captured.get("context_snapshot") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("duplicate_id", [False, True])
|
||||
def test_task_dispatch_snapshot_does_not_include_unfinished_ambiguous_calls(monkeypatch, duplicate_id):
|
||||
from langchain_core.messages import AIMessage, HumanMessage
|
||||
|
||||
runtime = _make_runtime()
|
||||
runtime.state["messages"] = [
|
||||
HumanMessage(content="Keep the user requirement"),
|
||||
AIMessage(
|
||||
content="",
|
||||
tool_calls=[
|
||||
{"name": "bash", "args": {"command": "COMPLETED_COMMAND"}, "id": "finished"},
|
||||
{"name": "task", "args": {"prompt": "UNFINISHED_DELEGATION"}, "id": "finished" if duplicate_id else "pending"},
|
||||
],
|
||||
),
|
||||
ToolMessage(content="Observed command result", tool_call_id="finished", name="bash"),
|
||||
]
|
||||
before = [message.model_dump() for message in runtime.state["messages"]]
|
||||
captured = {}
|
||||
|
||||
class DummyExecutor:
|
||||
def __init__(self, **kwargs):
|
||||
captured.update(kwargs)
|
||||
|
||||
def execute_async(self, prompt, task_id=None):
|
||||
return task_id
|
||||
|
||||
monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
|
||||
monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor)
|
||||
monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: _make_subagent_config())
|
||||
monkeypatch.setattr(task_tool_module, "get_background_task_result", lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done"))
|
||||
monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda event: None)
|
||||
monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep)
|
||||
monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [])
|
||||
result = _run_task_tool(runtime=runtime, prompt="Current delegated task", subagent_type="general-purpose", tool_call_id="current-dispatch", context_mode="snapshot")
|
||||
assert _task_tool_message(result).additional_kwargs[SUBAGENT_STATUS_KEY] == "completed"
|
||||
snapshot = captured["context_snapshot"]
|
||||
assert "UNFINISHED_DELEGATION" not in snapshot.content_json
|
||||
assert ("COMPLETED_COMMAND" in snapshot.content_json) is (not duplicate_id)
|
||||
assert "Keep the user requirement" in snapshot.content_json
|
||||
assert "Observed command result" in snapshot.content_json
|
||||
assert [message.model_dump() for message in runtime.state["messages"]] == before
|
||||
|
||||
|
||||
def test_task_tool_context_mode_schema_rejects_unknown_mode():
|
||||
from pydantic import ValidationError
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user