mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-06 17:38:35 +00:00
fix(api): don't re-persist user turn on empty-response retry (#4079)
The non-streaming retry called process_direct again with the same content, persisting a duplicate user turn. Pass persist_user_message=False so the retry recovers a response without re-recording the user message.
This commit is contained in:
parent
25a55fe1c7
commit
d75f80437c
@ -1810,12 +1810,16 @@ class AgentLoop:
|
|||||||
on_stream_end: Callable[..., Awaitable[None]] | None = None,
|
on_stream_end: Callable[..., Awaitable[None]] | None = None,
|
||||||
ephemeral: bool = False,
|
ephemeral: bool = False,
|
||||||
tools: ToolRegistry | None = None,
|
tools: ToolRegistry | None = None,
|
||||||
|
persist_user_message: bool = True,
|
||||||
) -> OutboundMessage | None:
|
) -> OutboundMessage | None:
|
||||||
"""Process a message directly and return the outbound payload."""
|
"""Process a message directly and return the outbound payload."""
|
||||||
await self._connect_mcp()
|
await self._connect_mcp()
|
||||||
|
metadata: dict[str, Any] = {}
|
||||||
|
if not persist_user_message:
|
||||||
|
metadata[turn_continuation.SKIP_USER_PERSIST_META] = True
|
||||||
msg = InboundMessage(
|
msg = InboundMessage(
|
||||||
channel=channel, sender_id="user", chat_id=chat_id,
|
channel=channel, sender_id="user", chat_id=chat_id,
|
||||||
content=content, media=media or [],
|
content=content, media=media or [], metadata=metadata,
|
||||||
)
|
)
|
||||||
# Share the dispatch lock so direct calls serialize with bus turns.
|
# Share the dispatch lock so direct calls serialize with bus turns.
|
||||||
lock = self._session_locks.setdefault(session_key, asyncio.Lock())
|
lock = self._session_locks.setdefault(session_key, asyncio.Lock())
|
||||||
|
|||||||
@ -340,6 +340,7 @@ async def handle_chat_completions(request: web.Request) -> web.Response:
|
|||||||
session_key=session_key,
|
session_key=session_key,
|
||||||
channel="api",
|
channel="api",
|
||||||
chat_id=API_CHAT_ID,
|
chat_id=API_CHAT_ID,
|
||||||
|
persist_user_message=False,
|
||||||
),
|
),
|
||||||
timeout=timeout_s,
|
timeout=timeout_s,
|
||||||
)
|
)
|
||||||
|
|||||||
@ -22,6 +22,7 @@ INTERNAL_CONTINUATION_META = "_internal_continuation"
|
|||||||
INTERNAL_CONTINUATION_KIND_META = "_internal_continuation_kind"
|
INTERNAL_CONTINUATION_KIND_META = "_internal_continuation_kind"
|
||||||
INTERNAL_CONTINUATION_PENDING_META = "_internal_continuation_pending"
|
INTERNAL_CONTINUATION_PENDING_META = "_internal_continuation_pending"
|
||||||
INTERNAL_CONTINUATION_RUN_STARTED_AT_META = "_internal_continuation_run_started_at"
|
INTERNAL_CONTINUATION_RUN_STARTED_AT_META = "_internal_continuation_run_started_at"
|
||||||
|
SKIP_USER_PERSIST_META = "_skip_user_persist"
|
||||||
|
|
||||||
_GOAL_CONTINUATION_KIND = "sustained_goal"
|
_GOAL_CONTINUATION_KIND = "sustained_goal"
|
||||||
_GOAL_CONTINUATION_SENDER = "system:continuation"
|
_GOAL_CONTINUATION_SENDER = "system:continuation"
|
||||||
@ -59,6 +60,8 @@ def internal_continuation_run_started_at(metadata: Mapping[str, Any] | None) ->
|
|||||||
|
|
||||||
def should_persist_user_message(metadata: Mapping[str, Any] | None) -> bool:
|
def should_persist_user_message(metadata: Mapping[str, Any] | None) -> bool:
|
||||||
"""Return whether this inbound message should be persisted as user input."""
|
"""Return whether this inbound message should be persisted as user input."""
|
||||||
|
if metadata and metadata.get(SKIP_USER_PERSIST_META) is True:
|
||||||
|
return False
|
||||||
return not internal_continuation_inbound(metadata)
|
return not internal_continuation_inbound(metadata)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@ -400,6 +400,32 @@ async def test_empty_response_retry_then_success(aiohttp_client) -> None:
|
|||||||
assert call_count == 2
|
assert call_count == 2
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_empty_response_retry_does_not_duplicate_user_turn(aiohttp_client) -> None:
|
||||||
|
persist_flags = []
|
||||||
|
|
||||||
|
async def record(content, session_key="", channel="", chat_id="", **kwargs):
|
||||||
|
persist_flags.append(kwargs.get("persist_user_message", True))
|
||||||
|
return "" if len(persist_flags) == 1 else "recovered response"
|
||||||
|
|
||||||
|
agent = MagicMock()
|
||||||
|
agent.process_direct = record
|
||||||
|
agent._connect_mcp = AsyncMock()
|
||||||
|
agent.close_mcp = AsyncMock()
|
||||||
|
agent._last_usage = {}
|
||||||
|
|
||||||
|
app = create_app(agent, model_name="m")
|
||||||
|
client = await aiohttp_client(app)
|
||||||
|
resp = await client.post(
|
||||||
|
"/v1/chat/completions",
|
||||||
|
json={"messages": [{"role": "user", "content": "hello"}]},
|
||||||
|
)
|
||||||
|
assert resp.status == 200
|
||||||
|
# first call persists the user turn; the retry must not persist it again
|
||||||
|
assert persist_flags == [True, False]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
|
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_empty_response_falls_back(aiohttp_client) -> None:
|
async def test_empty_response_falls_back(aiohttp_client) -> None:
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user