111 lines
4.5 KiB
Python
111 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from core.loop import AgentLoop, _is_empty_response # noqa: E402
|
|
|
|
|
|
def _resp_empty(content=None):
|
|
"""空响应桩:无 tool_calls,content 为空/None/空白。"""
|
|
msg = SimpleNamespace(tool_calls=[], content=content)
|
|
return SimpleNamespace(choices=[SimpleNamespace(message=msg)], usage=None)
|
|
|
|
|
|
def _resp_text(text):
|
|
"""正常文本收尾桩:无 tool_calls 但有正文。"""
|
|
msg = SimpleNamespace(tool_calls=[], content=text)
|
|
return SimpleNamespace(choices=[SimpleNamespace(message=msg)], usage=None)
|
|
|
|
|
|
def _resp_tool():
|
|
"""纯 tool_call 轮桩:有 tool_calls、content 空 —— 不算空响应。"""
|
|
tc = SimpleNamespace(function=SimpleNamespace(name="glob", arguments='{"pattern":"*"}'))
|
|
msg = SimpleNamespace(tool_calls=[tc], content=None)
|
|
return SimpleNamespace(choices=[SimpleNamespace(message=msg)], usage=None)
|
|
|
|
|
|
def _make_loop(stream_results, nonstream_results):
|
|
loop = object.__new__(AgentLoop)
|
|
loop.caps = SimpleNamespace(reliable_context=64_000, family="test", variant="t")
|
|
loop.session = SimpleNamespace(messages=[], task_id="test-task")
|
|
loop.user_id = "test-user" # 无 DB:_log_empty_response 落库路径静默跳过
|
|
loop.events = []
|
|
loop._emit = loop.events.append
|
|
calls = {"stream": 0, "nonstream": 0}
|
|
|
|
def collect_stream_once(llm_messages):
|
|
calls["stream"] += 1
|
|
return stream_results.pop(0), False
|
|
|
|
def nonstream_once(llm_messages):
|
|
calls["nonstream"] += 1
|
|
return nonstream_results.pop(0)
|
|
|
|
loop._collect_stream_once = collect_stream_once
|
|
loop._nonstream_once = nonstream_once
|
|
return loop, calls
|
|
|
|
|
|
class TestIsEmptyResponse(unittest.TestCase):
|
|
def test_empty_empty_is_empty(self):
|
|
self.assertTrue(_is_empty_response(_resp_empty(None)))
|
|
self.assertTrue(_is_empty_response(_resp_empty("")))
|
|
self.assertTrue(_is_empty_response(_resp_empty(" \n\t ")))
|
|
|
|
def test_text_answer_not_empty(self):
|
|
self.assertFalse(_is_empty_response(_resp_text("这是最终回答")))
|
|
|
|
def test_pure_tool_call_round_not_empty(self):
|
|
# 纯工具调用轮(content 空但有 tool_calls)是正常轮,绝不能误判为空响应
|
|
self.assertFalse(_is_empty_response(_resp_tool()))
|
|
|
|
def test_malformed_response_shape_is_safe(self):
|
|
# 畸形 / 缺字段的桩不应抛,只返 False
|
|
self.assertFalse(_is_empty_response(SimpleNamespace(choices=[])))
|
|
self.assertFalse(_is_empty_response(SimpleNamespace()))
|
|
|
|
|
|
class TestEmptyResponseRetry(unittest.TestCase):
|
|
def test_empty_stream_falls_back_to_nonstream(self):
|
|
"""空响应首败即降级非流式,重发拿到正文 → 恢复,1 条 warn 提「非流式」。"""
|
|
loop, calls = _make_loop([_resp_empty()], [_resp_text("ok")])
|
|
resp, cancelled = loop._stream_llm()
|
|
self.assertFalse(cancelled)
|
|
self.assertFalse(_is_empty_response(resp))
|
|
self.assertEqual(calls, {"stream": 1, "nonstream": 1})
|
|
warns = [e for e in loop.events if e.get("type") == "warn"]
|
|
self.assertEqual(len(warns), 1)
|
|
self.assertIn("空响应", warns[0]["msg"])
|
|
self.assertIn("非流式", warns[0]["msg"])
|
|
# 每 attempt 重发 llm_start,让前端占位段重建
|
|
starts = [e for e in loop.events if e.get("type") == "llm_start"]
|
|
self.assertEqual(len(starts), 2)
|
|
|
|
def test_all_attempts_empty_returns_empty(self):
|
|
"""全部尝试仍空 → 返回最后一次空 response(交 run() 收尾分支 warn 自停),不死循环。"""
|
|
n_nonstream = AgentLoop._MAX_MALFORMED_ATTEMPTS - 1
|
|
loop, calls = _make_loop([_resp_empty()], [_resp_empty()] * n_nonstream)
|
|
resp, cancelled = loop._stream_llm()
|
|
self.assertFalse(cancelled)
|
|
self.assertTrue(_is_empty_response(resp))
|
|
self.assertEqual(calls, {"stream": 1, "nonstream": n_nonstream})
|
|
warns = [e for e in loop.events if e.get("type") == "warn"]
|
|
self.assertEqual(len(warns), AgentLoop._MAX_MALFORMED_ATTEMPTS)
|
|
|
|
def test_clean_text_no_retry(self):
|
|
"""正常文本收尾轮不触发重试。"""
|
|
loop, calls = _make_loop([_resp_text("done")], [])
|
|
resp, cancelled = loop._stream_llm()
|
|
self.assertFalse(cancelled)
|
|
self.assertEqual(calls, {"stream": 1, "nonstream": 0})
|
|
self.assertEqual([e for e in loop.events if e.get("type") == "warn"], [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|