返回提交历史
Added
etc/unittest/test_tool_loop_detection.py
+128
-0
Added
g4f/api/tool_loop_detection.py
+219
-0
XFEstudio/gpt4free
Add tool loop detection
3bff80a6
代码差异
2 个文件
+347
-0
@@ -0,0 +1,128 @@
1
"""Tests for g4f.api.tool_loop_detection."""
2
3
import unittest
4
5
from g4f.api.tool_loop_detection import detect_tool_loop, ToolLoopError
6
7
8
def _make_loop(name: str, args: str, result: str, count: int):
9
"""Build a messages list with `count` identical tool calls + results."""
10
messages = [{"role": "user", "content": "find the file"}]
11
for i in range(count):
12
messages.append({
13
"role": "assistant",
14
"content": "",
15
"tool_calls": [{
16
"id": f"call_{i}",
17
"type": "function",
18
"function": {"name": name, "arguments": args},
19
}],
20
})
21
messages.append({
22
"role": "tool",
23
"tool_call_id": f"call_{i}",
24
"content": result,
25
})
26
return messages
27
28
29
class TestToolLoopDetection(unittest.TestCase):
30
31
def test_normal_conversation_passes(self):
32
"""A conversation with no tool calls should never trigger."""
33
messages = [
34
{"role": "system", "content": "You are helpful"},
35
{"role": "user", "content": "Hello"},
36
{"role": "assistant", "content": "Hi there!"},
37
]
38
detect_tool_loop(messages) # should not raise
39
40
def test_two_calls_allowed(self):
41
"""Two identical calls are within the MAX_REPEATS_PER_QUERY=2 limit."""
42
messages = _make_loop("file_search", '{"query": "**/x.js"}', "No files found", 2)
43
detect_tool_loop(messages) # should not raise
44
45
def test_three_identical_calls_blocked(self):
46
"""Three identical calls exceed MAX_REPEATS_PER_QUERY=2."""
47
messages = _make_loop("file_search", '{"query": "**/members-worker.js"}', "No files found", 3)
48
with self.assertRaises(ToolLoopError) as ctx:
49
detect_tool_loop(messages)
50
self.assertEqual(ctx.exception.function_name, "file_search")
51
self.assertIn("3 times", str(ctx.exception))
52
53
def test_three_empty_results_blocked(self):
54
"""Three identical calls with empty results are blocked (per-query rule fires first)."""
55
messages = _make_loop("grep_search", '{"query": "foo"}', "(empty)", 3)
56
with self.assertRaises(ToolLoopError) as ctx:
57
detect_tool_loop(messages)
58
# The per-query rule (max 2 repeats) fires before the empty-result
59
# threshold (3) because the 3rd assistant tool_call is seen before
60
# the 3rd tool response.
61
self.assertEqual(ctx.exception.function_name, "grep_search")
62
63
def test_different_arguments_not_blocked(self):
64
"""Calls with different arguments are independent and should pass."""
65
messages = [{"role": "user", "content": "search"}]
66
for i in range(5):
67
messages.append({
68
"role": "assistant",
69
"content": "",
70
"tool_calls": [{
71
"id": f"call_{i}",
72
"type": "function",
73
"function": {"name": "file_search", "arguments": f'{{"query": "q{i}"}}'},
74
}],
75
})
76
messages.append({"role": "tool", "tool_call_id": f"call_{i}", "content": "(empty)"})
77
# Each query is unique, so no single (name, args) exceeds thresholds.
78
detect_tool_loop(messages)
79
80
def test_dict_arguments_normalized(self):
81
"""Arguments as dict (not JSON string) should be normalized and detected."""
82
messages = [{"role": "user", "content": "find"}]
83
for i in range(3):
84
messages.append({
85
"role": "assistant",
86
"content": "",
87
"tool_calls": [{
88
"id": f"call_{i}",
89
"type": "function",
90
"function": {"name": "file_search", "arguments": {"query": "**/x.js"}},
91
}],
92
})
93
messages.append({"role": "tool", "tool_call_id": f"call_{i}", "content": "No files found"})
94
with self.assertRaises(ToolLoopError):
95
detect_tool_loop(messages)
96
97
def test_global_tool_call_limit(self):
98
"""Exceeding GLOBAL_TOOL_CALL_LIMIT triggers the hard cap."""
99
# Use unique args so per-query rules don't fire first.
100
messages = [{"role": "user", "content": "find"}]
101
for i in range(60):
102
messages.append({
103
"role": "assistant",
104
"content": "",
105
"tool_calls": [{
106
"id": f"call_{i}",
107
"type": "function",
108
"function": {"name": "file_search", "arguments": f'{{"query": "q{i}"}}'},
109
}],
110
})
111
messages.append({"role": "tool", "tool_call_id": f"call_{i}", "content": "ok"})
112
with self.assertRaises(ToolLoopError) as ctx:
113
detect_tool_loop(messages)
114
self.assertIn("limit", str(ctx.exception).lower())
115
116
def test_empty_messages_passes(self):
117
detect_tool_loop([]) # should not raise
118
119
def test_non_empty_result_not_blocked(self):
120
"""Successful tool results should not trigger the empty-result rule."""
121
messages = _make_loop("file_search", '{"query": "x"}', "Found 3 files: a, b, c", 5)
122
# 5 identical calls with non-empty results: per-query rule fires at call 3.
123
with self.assertRaises(ToolLoopError):
124
detect_tool_loop(messages)
125
126
127
if __name__ == "__main__":
128
unittest.main()
@@ -0,0 +1,219 @@
1
"""Detect and block repetitive tool-call loops in incoming messages.
2
3
Some clients (notably AI coding agents) can get stuck in infinite loops where
4
the assistant repeatedly issues the *same* tool call (e.g. ``file_search`` or
5
``grep_search`` with identical arguments) and the corresponding ``tool`` role
6
responses come back empty ("(empty)", "No files found", ...). When this
7
pattern is detected in an incoming request, the API server should refuse to
8
forward the request and instead return a descriptive error so the caller can
9
break out of the loop.
10
11
The detection rules (see ``/memories/repo/tool-loop-prevention.md``):
12
13
* After **2** failed attempts with the *same* query, STOP.
14
* Never repeat the same ``file_search`` / ``grep_search`` query more than twice.
15
* If a tool returns the same empty result **3+** times, switch strategy.
16
17
This module exposes :func:`detect_tool_loop`, which inspects a ``messages``
18
array and raises :class:`ToolLoopError` when a loop is detected.
19
"""
20
21
from __future__ import annotations
22
23
import json
24
from typing import Any, Dict, Iterable, List, Optional, Tuple
25
26
from ..typing import Messages
27
28
29
# Thresholds (tunable via env vars if needed in the future).
30
MAX_REPEATS_PER_QUERY = 2 # same call allowed at most twice
31
EMPTY_RESULT_LOOP_THRESHOLD = 3 # 3+ empty results for same call => loop
32
GLOBAL_TOOL_CALL_LIMIT = 100 # hard cap on total tool calls in one request
33
34
# Substrings that indicate an empty / failed tool result.
35
EMPTY_RESULT_MARKERS = (
36
"(empty)",
37
"no files found",
38
"no matches found",
39
"no results found",
40
"nothing found",
41
"0 results",
42
"0 matches",
43
)
44
45
46
class ToolLoopError(Exception):
47
"""Raised when a repetitive tool-call loop is detected in ``messages``."""
48
49
def __init__(self, message: str, *, function_name: str = None,
50
arguments: Any = None, repeats: int = 0):
51
super().__init__(message)
52
self.function_name = function_name
53
self.arguments = arguments
54
self.repeats = repeats
55
56
57
def _normalize_arguments(arguments: Any) -> str:
58
"""Return a stable string key for tool-call arguments.
59
60
``arguments`` may arrive as a JSON string, a dict, or ``None``. We
61
canonicalise to a sorted JSON string so that semantically identical
62
arguments compare equal regardless of key ordering.
63
"""
64
if arguments is None:
65
return "{}"
66
if isinstance(arguments, str):
67
try:
68
arguments = json.loads(arguments)
69
except (json.JSONDecodeError, ValueError):
70
return arguments.strip()
71
if isinstance(arguments, dict):
72
return json.dumps(arguments, sort_keys=True, ensure_ascii=False)
73
return str(arguments)
74
75
76
def _tool_call_key(tool_call: Dict[str, Any]) -> Optional[Tuple[str, str]]:
77
"""Build a ``(function_name, arguments_key)`` tuple from a tool call dict.
78
79
Returns ``None`` if the dict is not a recognised tool call.
80
"""
81
if not isinstance(tool_call, dict):
82
return None
83
if tool_call.get("type") not in (None, "function"):
84
return None
85
function = tool_call.get("function") or {}
86
if not isinstance(function, dict):
87
return None
88
name = function.get("name")
89
if not name:
90
return None
91
args_key = _normalize_arguments(function.get("arguments"))
92
return (name, args_key)
93
94
95
def _is_empty_result(content: Any) -> bool:
96
"""Return True when a ``tool`` message content looks like an empty result."""
97
if content is None:
98
return True
99
if isinstance(content, list):
100
# Content parts: join any text parts.
101
text = " ".join(
102
part.get("text", "") for part in content
103
if isinstance(part, dict) and part.get("type") == "text"
104
)
105
return _is_empty_result(text)
106
if not isinstance(content, str):
107
content = str(content)
108
stripped = content.strip().lower()
109
if not stripped:
110
return True
111
return any(marker in stripped for marker in EMPTY_RESULT_MARKERS)
112
113
114
def detect_tool_loop(messages: Messages) -> None:
115
"""Inspect ``messages`` and raise :class:`ToolLoopError` on a detected loop.
116
117
The function walks the conversation in order, pairing each assistant
118
``tool_calls`` entry with the subsequent ``tool`` role responses, and
119
counts how many times each unique ``(name, arguments)`` combination is
120
invoked and how many of those invocations yielded an empty result.
121
122
Detection triggers when:
123
124
* The same ``(name, arguments)`` combination is invoked more than
125
``MAX_REPEATS_PER_QUERY`` times, **or**
126
* The same combination yields ``EMPTY_RESULT_LOOP_THRESHOLD`` or more
127
empty results, **or**
128
* The total number of tool calls in the request exceeds
129
``GLOBAL_TOOL_CALL_LIMIT``.
130
"""
131
if not messages:
132
return
133
134
# Counters keyed by (function_name, arguments_key).
135
call_counts: Dict[Tuple[str, str], int] = {}
136
empty_counts: Dict[Tuple[str, str], int] = {}
137
total_tool_calls = 0
138
139
# Map tool_call_id -> (name, args_key) so we can attribute tool responses.
140
pending_calls: Dict[str, Tuple[str, str]] = {}
141
142
for message in messages:
143
if not isinstance(message, dict):
144
continue
145
146
role = message.get("role")
147
148
# Assistant message may carry tool_calls.
149
if role == "assistant":
150
tool_calls = message.get("tool_calls") or []
151
if isinstance(tool_calls, list):
152
for tc in tool_calls:
153
key = _tool_call_key(tc)
154
if key is None:
155
continue
156
total_tool_calls += 1
157
call_counts[key] = call_counts.get(key, 0) + 1
158
tc_id = tc.get("id") if isinstance(tc, dict) else None
159
if tc_id:
160
pending_calls[tc_id] = key
161
162
# Hard global cap.
163
if total_tool_calls > GLOBAL_TOOL_CALL_LIMIT:
164
raise ToolLoopError(
165
f"Tool loop detected: request contains {total_tool_calls} "
166
f"tool calls (limit {GLOBAL_TOOL_CALL_LIMIT}). The assistant "
167
f"is stuck calling tools repeatedly. Stop calling tools and "
168
f"answer directly, or switch strategy (e.g. use read_file "
169
f"with an absolute path instead of search).",
170
function_name=key[0],
171
arguments=key[1],
172
repeats=total_tool_calls,
173
)
174
175
# Per-query repeat cap.
176
if call_counts[key] > MAX_REPEATS_PER_QUERY:
177
raise ToolLoopError(
178
f"Tool loop detected: function '{key[0]}' was called "
179
f"{call_counts[key]} times with identical arguments "
180
f"({call_counts[key] - 1} retries, limit is "
181
f"{MAX_REPEATS_PER_QUERY - 1}). Stop repeating this call. "
182
f"Arguments: {key[1]}",
183
function_name=key[0],
184
arguments=key[1],
185
repeats=call_counts[key],
186
)
187
188
# Tool result message: attribute to the originating call.
189
elif role == "tool":
190
tool_call_id = message.get("tool_call_id")
191
content = message.get("content")
192
key = pending_calls.get(tool_call_id) if tool_call_id else None
193
# If we can't attribute via id, fall back to the most recent call.
194
if key is None and pending_calls:
195
# Heuristic: attribute to the last registered pending call.
196
key = next(reversed(pending_calls), None)
197
if key is not None and _is_empty_result(content):
198
empty_counts[key] = empty_counts.get(key, 0) + 1
199
if empty_counts[key] >= EMPTY_RESULT_LOOP_THRESHOLD:
200
raise ToolLoopError(
201
f"Tool loop detected: function '{key[0]}' returned empty "
202
f"results {empty_counts[key]} times with the same arguments. "
203
f"The target is likely not indexed by this tool (e.g. a git "
204
f"submodule). Switch strategy: use read_file with an absolute "
205
f"path, or list_dir to confirm the file exists. Arguments: "
206
f"{key[1]}",
207
function_name=key[0],
208
arguments=key[1],
209
repeats=empty_counts[key],
210
)
211
212
213
def has_tool_loop(messages: Messages) -> Optional[ToolLoopError]:
214
"""Convenience wrapper: return the error object if a loop is detected, else None."""
215
try:
216
detect_tool_loop(messages)
217
except ToolLoopError as e:
218
return e
219
return None