fix(backend): auto-purge polluted checkpoint when tool_calls lack ToolMessage

When a tool call crashes (e.g. httpx timeout in kb_search), the LangGraph
checkpoint retains an AIMessage with tool_calls but no corresponding
ToolMessage. Subsequent requests to the same conversation_id fail with:
  ValueError: Found AIMessages with tool_calls that do not have a
  corresponding ToolMessage

Now the except block in _stream_response detects this specific ValueError
by checking for "tool_calls" and "ToolMessage" in the error string, then
calls checkpointer.adelete_thread() to purge the corrupted thread state.
The frontend receives {"type":"error","content":"对话状态异常,已自动重置..."}
followed by {"type":"done"}, so the user can simply resend their message.

API confirmed: AsyncPostgresSaver.adelete_thread(thread_id) deletes from
checkpoints, checkpoint_blobs, and checkpoint_writes tables for the thread.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
gongzhiyong
2026-04-08 20:53:54 +08:00
co-authored by Claude Sonnet 4.6
parent 0f2db39655
commit 5d0127bfb4
+26 -1
View File
@@ -13,6 +13,7 @@ from litestar.response import Stream
from app.graph.builder import get_chat_graph
from app.schemas import ChatRequest
from app.store.memory import get_checkpointer
from app.store.postgres import Conversation, Message, async_session_factory
from app.tools import resolve_tools
from app.tools.search import set_search_model
@@ -118,8 +119,32 @@ async def _stream_response(request: ChatRequest) -> AsyncIterator[bytes]:
except Exception as exc:
logger.error("SSE stream error for conversation %s: %s", request.conversation_id, exc, exc_info=True)
error_msg = str(exc)
user_hint = "请求处理出现错误,请重试"
# Detect checkpoint pollution: a prior tool crash left an AIMessage
# with tool_calls but no corresponding ToolMessage. LangGraph refuses
# to continue the thread. Purge the thread so the next request
# starts from a clean state.
if "tool_calls" in error_msg and "ToolMessage" in error_msg:
try:
checkpointer = await get_checkpointer()
await checkpointer.adelete_thread(request.conversation_id)
logger.warning(
"Purged polluted checkpoint for thread %s",
request.conversation_id,
)
user_hint = "对话状态异常,已自动重置,请重新发送消息"
except Exception:
logger.warning(
"Failed to purge checkpoint for thread %s",
request.conversation_id,
exc_info=True,
)
error_data = json.dumps(
{"type": "error", "content": "请求处理出现错误,请重试"},
{"type": "error", "content": user_hint},
ensure_ascii=False,
)
yield f"data: {error_data}\n\n".encode("utf-8")