diff --git a/docs/changelog/roadmap.md b/docs/changelog/roadmap.md index 59f8bbd2..ad8b3c3a 100644 --- a/docs/changelog/roadmap.md +++ b/docs/changelog/roadmap.md @@ -8,7 +8,7 @@ 🐛**BUGs** - [x] 部分 doc 格式的文件支持有问题 - [x] 当出现不支持的文件类型的时候，前端没有限制 -- [ ] 当消息生成的时候有报错的时候，前端无显示 +- [x] 当消息生成的时候有报错的时候，前端无显示 - [x] 另外一个智能体的历史对话无法显示 - [ ] 调用统计的统计结果疑似有问题（Token 计算方法可能也不对） - [ ] 【重要】传输给 agent 的上下文消息有问题，需要基于新的 conv 构建阶段算法 @@ -17,6 +17,7 @@ --- 💭 **Features Todo** + - [ ] 添加对于上传文件的支持：这里的复杂的地方就在于如何和历史记录结合在一起（v0.2.3 版本实现，放在记忆管理后面） - [ ] 知识图谱的上传和可视化，支持属性，标签的展示 - [ ] 集成智能体评估，首先使用命令行来实现，然后考虑放在 UI 里面展示 diff --git a/scripts/batch_upload.py b/scripts/batch_upload.py index 33c23b4b..2a51effd 100644 --- a/scripts/batch_upload.py +++ b/scripts/batch_upload.py @@ -94,7 +94,8 @@ async def process_document( task_id = result.get("task_id") extra = f" (task id: {task_id})" if task_id else "" console.print( - f"[bold cyan]Ingestion queued for {server_file_path}{extra}. Track progress in the task center.[/bold cyan]" + f"[bold cyan]Ingestion queued for {server_file_path}{extra}. " + "Track progress in the task center.[/bold cyan]" ) return True diff --git a/server/routers/auth_router.py b/server/routers/auth_router.py index 96e5400a..78c490ab 100644 --- a/server/routers/auth_router.py +++ b/server/routers/auth_router.py @@ -7,7 +7,7 @@ from pydantic import BaseModel from sqlalchemy.orm import Session from src.storage.db.manager import db_manager -from src.storage.db.models import User, OperationLog +from src.storage.db.models import User from server.utils.auth_middleware import get_admin_user, get_current_user, get_db, get_required_user from server.utils.auth_utils import AuthUtils from server.utils.user_utils import generate_unique_user_id, validate_username, is_valid_phone_number diff --git a/server/routers/chat_router.py b/server/routers/chat_router.py index 4867bcc0..5aa5a5c4 100644 --- a/server/routers/chat_router.py +++ b/server/routers/chat_router.py @@ -7,12 +7,13 @@ from pathlib import Path from fastapi import APIRouter, Body, Depends, HTTPException from fastapi.responses import StreamingResponse -from langchain_core.messages import AIMessageChunk, HumanMessage, ToolMessage +from langchain_core.messages import AIMessageChunk, HumanMessage from pydantic import BaseModel from sqlalchemy.orm import Session from src.storage.db.models import User, MessageFeedback, Message, Conversation from src.storage.conversation import ConversationManager +from src.storage.db.manager import db_manager from server.routers.auth_router import get_admin_user from server.utils.auth_middleware import get_db, get_required_user from src import executor @@ -163,7 +164,11 @@ async def chat_agent( # 获取已保存的消息数量，避免重复保存 existing_messages = conv_mgr.get_messages_by_thread_id(thread_id) - existing_ids = {msg.extra_metadata["id"] for msg in existing_messages if msg.extra_metadata and "id" in msg.extra_metadata} + existing_ids = { + msg.extra_metadata["id"] + for msg in existing_messages + if msg.extra_metadata and "id" in msg.extra_metadata + } for msg in messages: msg_dict = msg.model_dump() if hasattr(msg, "model_dump") else {} @@ -238,7 +243,7 @@ async def chat_agent( logger.debug(f"Processed message type={msg_type}") - logger.info(f"Saved messages from LangGraph state") + logger.info("Saved messages from LangGraph state") except Exception as e: logger.error(f"Error saving messages from LangGraph state: {e}") @@ -271,7 +276,6 @@ async def chat_agent( thread_id = str(uuid.uuid4()) logger.warning(f"No thread_id provided, generated new thread_id: {thread_id}") - # Initialize conversation manager conv_manager = ConversationManager(db) @@ -288,12 +292,11 @@ async def chat_agent( logger.error(f"Error saving user message: {e}") try: - full_ai_content = "" + full_msg = None async for msg, metadata in agent.stream_messages(messages, input_context=input_context): if isinstance(msg, AIMessageChunk): - - full_ai_content += msg.content - if conf.enable_content_guard and await content_guard.check_with_keywords(full_ai_content[-20:]): + full_msg = msg if not full_msg else full_msg + msg + if conf.enable_content_guard and await content_guard.check_with_keywords(full_msg.content[-20:]): logger.warning("Sensitive content detected in stream") yield make_chunk(message="检测到敏感内容，已中断输出", status="error") return @@ -303,7 +306,7 @@ async def chat_agent( else: yield make_chunk(msg=msg.model_dump(), metadata=metadata, status="loading") - if conf.enable_content_guard and await content_guard.check(full_ai_content): + if conf.enable_content_guard and hasattr(full_msg, "content") and await content_guard.check(full_msg.content): logger.warning("Sensitive content detected in final message") yield make_chunk(message="检测到敏感内容，已中断输出", status="error") return @@ -321,20 +324,45 @@ async def chat_agent( except (asyncio.CancelledError, ConnectionError) as e: # 客户端主动中断连接，尝试保存已生成的部分内容 - logger.warning(f"Client disconnected for thread {thread_id}: {e}") - langgraph_config = {"configurable": input_context} - await save_messages_from_langgraph_state( - agent_instance=agent, - thread_id=thread_id, - conv_mgr=conv_manager, - config_dict=langgraph_config, - ) + logger.warning(f"Client disconnected, cancelling stream: {e}") + if full_msg: + # 创建新的 db session，因为原 session 可能已关闭 + new_db = db_manager.get_session() + try: + new_conv_manager = ConversationManager(new_db) + msg_dict = full_msg.model_dump() if hasattr(full_msg, "model_dump") else {} + content = full_msg.content if hasattr(full_msg, "content") else str(full_msg) + new_conv_manager.add_message_by_thread_id( + thread_id=thread_id, + role="assistant", + content=content, + message_type="text", + extra_metadata=msg_dict | {"error_type": "interrupted"}, # 保存原始 model_dump + ) + finally: + new_db.close() # 通知前端中断（可能发送不到，但用于一致性） yield make_chunk(status="interrupted", message="对话已中断", meta=meta) except Exception as e: logger.error(f"Error streaming messages: {e}, {traceback.format_exc()}") + if full_msg: + # 创建新的 db session，因为原 session 可能已关闭 + new_db = db_manager.get_session() + try: + new_conv_manager = ConversationManager(new_db) + msg_dict = full_msg.model_dump() if hasattr(full_msg, "model_dump") else {} + content = full_msg.content if hasattr(full_msg, "content") else str(full_msg) + new_conv_manager.add_message_by_thread_id( + thread_id=thread_id, + role="assistant", + content=content, + message_type="text", + extra_metadata=msg_dict | {"error_type": "unexpect"}, # 保存原始 model_dump + ) + finally: + new_db.close() yield make_chunk(message=f"Error streaming messages: {e}", status="error") return StreamingResponse(stream_messages(), media_type="application/json") @@ -422,6 +450,7 @@ async def get_agent_history( "type": role_type_map.get(msg.role, msg.role), # human/ai/tool/system "content": msg.content, "created_at": msg.created_at.isoformat() if msg.created_at else None, + "error_type": msg.extra_metadata.get("error_type") if msg.extra_metadata else None, } # Add tool calls if present (for AI messages) diff --git a/server/routers/knowledge_router.py b/server/routers/knowledge_router.py index 0d4623ce..e7ebc60f 100644 --- a/server/routers/knowledge_router.py +++ b/server/routers/knowledge_router.py @@ -592,12 +592,20 @@ async def upload_file( content_hash = calculate_content_hash(file_bytes) if knowledge_base.file_existed_in_db(db_id, content_hash): - raise HTTPException(status_code=409, detail="数据库中已经存在了相同文件，File with the same content already exists in this database") + raise HTTPException( + status_code=409, + detail="数据库中已经存在了相同文件，File with the same content already exists in this database", + ) with open(file_path, "wb") as buffer: buffer.write(file_bytes) - return {"message": "File successfully uploaded", "file_path": file_path, "db_id": db_id, "content_hash": content_hash} + return { + "message": "File successfully uploaded", + "file_path": file_path, + "db_id": db_id, + "content_hash": content_hash, + } @knowledge.get("/files/supported-types") diff --git a/server/services/tasker.py b/server/services/tasker.py index cb4188e0..3aaa2488 100644 --- a/server/services/tasker.py +++ b/server/services/tasker.py @@ -5,7 +5,8 @@ import uuid from dataclasses import asdict, dataclass, field from datetime import datetime from pathlib import Path -from typing import Any, Awaitable, Callable, Dict, List, Optional +from typing import Any +from collections.abc import Awaitable, Callable from src.config import config from src.utils.logging_config import logger @@ -28,19 +29,19 @@ class Task: message: str = "" created_at: str = field(default_factory=_utc_timestamp) updated_at: str = field(default_factory=_utc_timestamp) - started_at: Optional[str] = None - completed_at: Optional[str] = None - payload: Dict[str, Any] = field(default_factory=dict) - result: Optional[Any] = None - error: Optional[str] = None + started_at: str | None = None + completed_at: str | None = None + payload: dict[str, Any] = field(default_factory=dict) + result: Any | None = None + error: str | None = None cancel_requested: bool = False - def to_dict(self) -> Dict[str, Any]: + def to_dict(self) -> dict[str, Any]: data = asdict(self) return data @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "Task": + def from_dict(cls, data: dict[str, Any]) -> "Task": return cls( id=data["id"], name=data.get("name", "Unnamed Task"), @@ -64,7 +65,7 @@ class TaskContext: self._tasker = tasker self.task_id = task_id - async def set_progress(self, progress: float, message: Optional[str] = None) -> None: + async def set_progress(self, progress: float, message: str | None = None) -> None: await self._tasker._update_task( self.task_id, progress=max(0.0, min(progress, 100.0)), @@ -88,10 +89,10 @@ class TaskContext: class Tasker: def __init__(self, worker_count: int = 2): self.worker_count = max(1, worker_count) - self._queue: "asyncio.Queue[tuple[str, TaskCoroutine]]" = asyncio.Queue() - self._tasks: Dict[str, Task] = {} + self._queue: asyncio.Queue[tuple[str, TaskCoroutine]] = asyncio.Queue() + self._tasks: dict[str, Task] = {} self._lock = asyncio.Lock() - self._workers: List[asyncio.Task[Any]] = [] + self._workers: list[asyncio.Task[Any]] = [] self._storage_path = Path(config.save_dir) / "tasks" / "tasks.json" os.makedirs(self._storage_path.parent, exist_ok=True) self._started = False @@ -124,7 +125,7 @@ class Tasker: *, name: str, task_type: str, - payload: Optional[Dict[str, Any]] = None, + payload: dict[str, Any] | None = None, coroutine: TaskCoroutine, ) -> Task: task_id = uuid.uuid4().hex @@ -136,7 +137,7 @@ class Tasker: logger.info("Enqueued task {} ({})", task_id, name) return task - async def list_tasks(self, status: Optional[str] = None) -> List[Dict[str, Any]]: + async def list_tasks(self, status: str | None = None) -> list[dict[str, Any]]: async with self._lock: tasks = list(self._tasks.values()) if status: @@ -144,7 +145,7 @@ class Tasker: tasks.sort(key=lambda item: item.created_at, reverse=True) return [task.to_dict() for task in tasks] - async def get_task(self, task_id: str) -> Optional[Dict[str, Any]]: + async def get_task(self, task_id: str) -> dict[str, Any] | None: async with self._lock: task = self._tasks.get(task_id) return task.to_dict() if task else None @@ -173,7 +174,9 @@ class Tasker: if task.cancel_requested: await self._mark_cancelled(task_id, "Task was cancelled before execution") continue - await self._update_task(task_id, status="running", progress=0.0, message="任务开始执行", started_at=_utc_timestamp()) + await self._update_task( + task_id, status="running", progress=0.0, message="任务开始执行", started_at=_utc_timestamp() + ) context = TaskContext(self, task_id) try: result = await coroutine(context) @@ -207,7 +210,7 @@ class Tasker: except Exception as exc: # noqa: BLE001 logger.exception("Tasker worker error: {}", exc) - async def _get_task_instance(self, task_id: str) -> Optional[Task]: + async def _get_task_instance(self, task_id: str) -> Task | None: async with self._lock: return self._tasks.get(task_id) @@ -224,13 +227,13 @@ class Tasker: self, task_id: str, *, - status: Optional[str] = None, - progress: Optional[float] = None, - message: Optional[str] = None, + status: str | None = None, + progress: float | None = None, + message: str | None = None, result: Any = None, - error: Optional[str] = None, - started_at: Optional[str] = None, - completed_at: Optional[str] = None, + error: str | None = None, + started_at: str | None = None, + completed_at: str | None = None, ) -> None: async with self._lock: task = self._tasks.get(task_id) diff --git a/src/agents/chatbot/graph.py b/src/agents/chatbot/graph.py index d94552a1..c60a714a 100644 --- a/src/agents/chatbot/graph.py +++ b/src/agents/chatbot/graph.py @@ -1,14 +1,10 @@ -import uuid -from pathlib import Path from typing import Any, cast from langchain_core.messages import AIMessage, ToolMessage -from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import END, START, StateGraph from langgraph.prebuilt import ToolNode, tools_condition from langgraph.runtime import Runtime -from src import config as sys_config from src.agents.common.base import BaseAgent from src.agents.common.mcp import get_mcp_tools from src.agents.common.models import load_chat_model @@ -110,6 +106,7 @@ class ChatbotAgent(BaseAgent): def main(): pass + if __name__ == "__main__": main() # asyncio.run(main()) diff --git a/src/agents/common/base.py b/src/agents/common/base.py index 48e87c83..7078b0c1 100644 --- a/src/agents/common/base.py +++ b/src/agents/common/base.py @@ -123,7 +123,6 @@ class BaseAgent: return checkpointer - async def get_async_conn(self) -> aiosqlite.Connection: """获取异步数据库连接""" return await aiosqlite.connect(os.path.join(self.workdir, "aio_history.db")) @@ -131,4 +130,3 @@ class BaseAgent: async def get_aio_memory(self) -> AsyncSqliteSaver: """获取异步存储实例""" return AsyncSqliteSaver(await self.get_async_conn()) - diff --git a/src/agents/react/graph.py b/src/agents/react/graph.py index 33e55b21..d76a8904 100644 --- a/src/agents/react/graph.py +++ b/src/agents/react/graph.py @@ -37,7 +37,7 @@ class ReActAgent(BaseAgent): available_tools = get_buildin_tools() self.checkpointer = await self._get_checkpointer() - # 创建 ReActAgent + # 创建 ReActAgent graph = create_react_agent(model, tools=available_tools, prompt=prompt, checkpointer=self.checkpointer) self.graph = graph logger.info("ReActAgent 使用内存 checkpointer 构建成功") diff --git a/web/src/apis/agent_api.js b/web/src/apis/agent_api.js index c94facf1..5b9c33d5 100644 --- a/web/src/apis/agent_api.js +++ b/web/src/apis/agent_api.js @@ -20,14 +20,22 @@ export const agentApi = { * @param {Object} data - 聊天数据 * @returns {Promise} - 聊天响应流 */ - sendAgentMessage: (agentId, data) => { + sendAgentMessage: (agentId, data, options = {}) => { + const { signal, headers: extraHeaders, ...restOptions } = options || {}; + const baseHeaders = { + 'Content-Type': 'application/json', + ...useUserStore().getAuthHeaders() + }; + return fetch(`/api/chat/agent/${agentId}`, { method: 'POST', + body: JSON.stringify(data), + signal, headers: { - 'Content-Type': 'application/json', - ...useUserStore().getAuthHeaders() + ...baseHeaders, + ...(extraHeaders || {}) }, - body: JSON.stringify(data) + ...restOptions }) }, @@ -179,4 +187,4 @@ export const threadApi = { * @returns {Promise} - 删除结果 */ deleteThread: (threadId) => apiDelete(`/api/chat/thread/${threadId}`) -}; \ No newline at end of file +}; diff --git a/web/src/components/AgentChatComponent.vue b/web/src/components/AgentChatComponent.vue index 5735ad44..153bd0e6 100644 --- a/web/src/components/AgentChatComponent.vue +++ b/web/src/components/AgentChatComponent.vue @@ -370,12 +370,12 @@ const _processStreamChunk = (chunk, threadId) => { const { status, msg, request_id, message } = chunk; const threadState = getThreadState(threadId); - if (!threadState) return; + if (!threadState) return false; switch (status) { case 'init': threadState.onGoingConv.msgChunks[request_id] = [msg]; - break; + return false; case 'loading': if (msg.id) { if (!threadState.onGoingConv.msgChunks[msg.id]) { @@ -383,45 +383,36 @@ const _processStreamChunk = (chunk, threadId) => { } threadState.onGoingConv.msgChunks[msg.id].push(msg); } - break; + return false; case 'error': handleChatError({ message }, 'stream'); // Stop the loading indicator if (threadState) { threadState.isStreaming = false; - // Create a new AI message chunk for the error - const errorMsgChunk = { - id: 'ai-error-' + Date.now(), - type: 'ai', - role: 'assistant', - content: chunk.message || 'An error occurred', - isError: true // Custom flag for styling - }; - - // Add this to the chunks of the ongoing conversation - if (threadState.onGoingConv && threadState.onGoingConv.msgChunks) { - threadState.onGoingConv.msgChunks[errorMsgChunk.id] = [errorMsgChunk]; - } - // Abort the stream controller to stop processing further events if (threadState.streamAbortController) { threadState.streamAbortController.abort(); threadState.streamAbortController = null; } } - // We no longer call resetOnGoingConv to keep the context. - break; + + // Reload messages to show any partial content saved by the backend + fetchThreadMessages({ agentId: currentAgentId.value, threadId: threadId }); + resetOnGoingConv(threadId); + return true; case 'finished': - fetchThreadMessages({ agentId: currentAgentId.value, threadId: threadId }); - resetOnGoingConv(threadId); - break; + fetchThreadMessages({ agentId: currentAgentId.value, threadId: threadId }); + resetOnGoingConv(threadId); + return true; case 'interrupted': - // 中断状态，刷新消息历史 - fetchThreadMessages({ agentId: currentAgentId.value, threadId: threadId }); - resetOnGoingConv(threadId); - break; + // 中断状态，刷新消息历史 + fetchThreadMessages({ agentId: currentAgentId.value, threadId: threadId }); + resetOnGoingConv(threadId); + return true; } + + return false; }; // ==================== 线程管理方法 ==================== @@ -520,7 +511,7 @@ const fetchThreadMessages = async ({ agentId, threadId }) => { }; // 发送消息并处理流式响应 -const sendMessage = async ({ agentId, threadId, text }) => { +const sendMessage = async ({ agentId, threadId, text, signal = undefined }) => { if (!agentId || !threadId || !text) { const error = new Error("Missing agent, thread, or message text"); handleChatError(error, 'send'); @@ -540,32 +531,13 @@ const sendMessage = async ({ agentId, threadId, text }) => { }; try { - return await agentApi.sendAgentMessage(agentId, requestData); + return await agentApi.sendAgentMessage(agentId, requestData, signal ? { signal } : undefined); } catch (error) { handleChatError(error, 'send'); throw error; } }; -// 添加消息到线程 -const addMessageToThread = (threadId, message) => { - if (!threadId || !message) return; - - if (!threadMessages.value[threadId]) { - threadMessages.value[threadId] = []; - } - - threadMessages.value[threadId].push(message); -}; - -// 更新线程中的消息 -const updateMessageInThread = (threadId, messageIndex, updatedMessage) => { - if (!threadId || messageIndex < 0 || !threadMessages.value[threadId]) return; - - if (messageIndex < threadMessages.value[threadId].length) { - threadMessages.value[threadId][messageIndex] = updatedMessage; - } -}; // ==================== CHAT ACTIONS ==================== // 检查第一个对话是否为空 @@ -689,15 +661,16 @@ const handleSendMessage = async () => { const response = await sendMessage({ agentId: currentAgentId.value, threadId: currentChatId.value, - text: text + text: text, + signal: threadState.streamAbortController?.signal }); const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; + let stopReading = false; - while (true) { - if (!threadState.streamAbortController || threadState.streamAbortController.signal.aborted) break; + while (!stopReading) { const { done, value } = await reader.read(); if (done) break; @@ -706,18 +679,24 @@ const handleSendMessage = async () => { buffer = lines.pop() || ''; for (const line of lines) { - if (line.trim() && threadState.streamAbortController && !threadState.streamAbortController.signal.aborted) { + const trimmedLine = line.trim(); + if (trimmedLine) { try { - const chunk = JSON.parse(line.trim()); - _processStreamChunk(chunk, threadId); + const chunk = JSON.parse(trimmedLine); + if (_processStreamChunk(chunk, threadId)) { + stopReading = true; + break; + } } catch (e) { console.warn('Failed to parse stream chunk JSON:', e); } } } } - if (buffer.trim() && threadState.streamAbortController && !threadState.streamAbortController.signal.aborted) { + if (!stopReading && buffer.trim()) { try { const chunk = JSON.parse(buffer.trim()); - _processStreamChunk(chunk, threadId); + if (_processStreamChunk(chunk, threadId)) { + stopReading = true; + } } catch (e) { console.warn('Failed to parse final stream chunk JSON:', e); } } } catch (error) { @@ -1151,6 +1130,7 @@ watch(conversations, () => { margin: 0 auto; padding: 4px 2rem 0 2rem; background: white; + z-index: 1000; .message-input-wrapper { width: 100%; diff --git a/web/src/components/AgentMessageComponent.vue b/web/src/components/AgentMessageComponent.vue index 0968a61f..cf0913a9 100644 --- a/web/src/components/AgentMessageComponent.vue +++ b/web/src/components/AgentMessageComponent.vue @@ -27,6 +27,12 @@
+ +
+ 回答生成已中断 + 生成过程中出现异常 +
+
@@ -296,6 +302,32 @@ const toggleToolCall = (toolCallId) => { width: 100%; } + .error-hint { + margin: 10px 0; + padding: 8px 16px; + border-radius: 8px; + font-size: 14px; + display: flex; + align-items: center; + gap: 8px; + + &.error-interrupted { + background-color: #fffbeb; + // border: 1px solid #fbbf24; + color: #92400e; + } + + &.error-unexpect { + background-color: #fef2f2; + // border: 1px solid #f87171; + color: #991b1b; + } + + span { + line-height: 1.5; + } + } + .status-info { display: block; background-color: var(--gray-50); @@ -553,6 +585,7 @@ const toggleToolCall = (toolCallId) => { .md-editor-code-head { background-color: var(--gray-50); + z-index: 1; .md-editor-collapse-tips { color: var(--gray-400);