From 606ff9f1222d252a49e254dc640e4b0426228db2 Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Tue, 30 Dec 2025 03:12:43 +0800 Subject: [PATCH] =?UTF-8?q?fix(chat=5Frouter):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E6=B6=88=E6=81=AF=E5=A4=84=E7=90=86=E4=B8=AD?= =?UTF-8?q?=E6=9C=AA=E6=AD=A3=E7=A1=AE=E7=B4=AF=E7=A7=AF=E5=86=85=E5=AE=B9?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98=20Question:=20=E4=B8=8A=E4=B8=8B?= =?UTF-8?q?=E6=96=87=E6=AF=94=E8=BE=83=E9=95=BF=E6=97=B6=EF=BC=8C=E5=9B=9E?= =?UTF-8?q?=E7=AD=94=E9=80=9F=E5=BA=A6=E5=BE=88=E6=85=A2=EF=BC=8CCPU?= =?UTF-8?q?=E5=8F=AA=E5=90=83=E4=BA=86=E5=8D=95=E6=A0=B8=EF=BC=8C=E4=B8=80?= =?UTF-8?q?=E7=9B=B4=E5=8D=A0=E6=BB=A1=E8=B5=84=E6=BA=90=20Fixes=20#410?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 确保在流式处理过程中正确累积消息内容，并在各种中断情况下（如敏感内容检测、错误处理等）保存已累积的内容。同时添加对用户会话的权限检查 --- server/routers/chat_router.py | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/server/routers/chat_router.py b/server/routers/chat_router.py index dfaa9cc6..3f383f38 100644 --- a/server/routers/chat_router.py +++ b/server/routers/chat_router.py @@ -5,7 +5,7 @@ import uuid from fastapi import APIRouter, Body, Depends, HTTPException, Query, UploadFile, File from fastapi.responses import StreamingResponse -from langchain.messages import AIMessageChunk, HumanMessage +from langchain.messages import AIMessageChunk, HumanMessage, AIMessage from langgraph.types import Command from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession @@ -577,13 +577,16 @@ async def chat_agent( input_context["attachments"] = [] full_msg = None + accumulated_content = [] langgraph_config = {"configurable": input_context} async for msg, metadata in agent.stream_messages(messages, input_context=input_context): if isinstance(msg, AIMessageChunk): - full_msg = msg if not full_msg else full_msg + msg - content_for_check = full_msg.content[-20:] + accumulated_content.append(msg.content) + + content_for_check = "".join(accumulated_content[-10:]) if conf.enable_content_guard and await content_guard.check_with_keywords(content_for_check): logger.warning("Sensitive content detected in stream") + full_msg = AIMessage(content="".join(accumulated_content)) await save_partial_message(conv_manager, thread_id, full_msg, "content_guard_blocked") meta["time_cost"] = asyncio.get_event_loop().time() - start_time yield make_chunk(status="interrupted", message="检测到敏感内容，已中断输出", meta=meta) @@ -606,6 +609,9 @@ async def chat_agent( logger.error(f"Error processing tool message: {e}") pass + if not full_msg and accumulated_content: + full_msg = AIMessage(content="".join(accumulated_content)) + if ( conf.enable_content_guard and hasattr(full_msg, "content") @@ -650,6 +656,10 @@ async def chat_agent( # Run save in a separate task to avoid cancellation async def save_cleanup(): + nonlocal full_msg + if not full_msg and accumulated_content: + full_msg = AIMessage(content="".join(accumulated_content)) + async with db_manager.get_async_session_context() as new_db: new_conv_manager = ConversationManager(new_db) await save_partial_message( @@ -679,6 +689,9 @@ async def chat_agent( error_msg = f"Error streaming messages: {e}" error_type = "unexpected_error" + if not full_msg and accumulated_content: + full_msg = AIMessage(content="".join(accumulated_content)) + # 保存错误消息到数据库 async with db_manager.get_async_session_context() as new_db: new_conv_manager = ConversationManager(new_db) @@ -888,6 +901,7 @@ async def get_agent_history( # Use new storage system ONLY conv_manager = ConversationManager(db) + await _require_user_conversation(conv_manager, thread_id, str(current_user.id)) messages = await conv_manager.get_messages_by_thread_id(thread_id) # 当前用户ID - 用于过滤反馈