diff --git a/.gitignore b/.gitignore index 216993a7..f14507b1 100644 --- a/.gitignore +++ b/.gitignore @@ -31,6 +31,10 @@ cache .idea .vibe .qoder +.claude +.cursor +.trae +.pytest_cache *.nogit* *.private* *.local* diff --git a/server/routers/__init__.py b/server/routers/__init__.py index bfc7b8b2..b2afa84f 100644 --- a/server/routers/__init__.py +++ b/server/routers/__init__.py @@ -2,6 +2,7 @@ from fastapi import APIRouter from server.routers.auth_router import auth from server.routers.chat_router import chat +from server.routers.dashboard_router import dashboard from server.routers.graph_router import graph from server.routers.knowledge_router import knowledge from server.routers.system_router import system @@ -12,5 +13,6 @@ router = APIRouter() router.include_router(system) # /api/system/* router.include_router(auth) # /api/auth/* router.include_router(chat) # /api/chat/* +router.include_router(dashboard) # /api/dashboard/* router.include_router(knowledge) # /api/knowledge/* router.include_router(graph) # /api/graph/* diff --git a/server/routers/chat_router.py b/server/routers/chat_router.py index b6b60fa8..59d6dd62 100644 --- a/server/routers/chat_router.py +++ b/server/routers/chat_router.py @@ -7,11 +7,12 @@ from pathlib import Path from fastapi import APIRouter, Body, Depends, HTTPException from fastapi.responses import StreamingResponse -from langchain_core.messages import AIMessageChunk, HumanMessage +from langchain_core.messages import AIMessageChunk, HumanMessage, ToolMessage from pydantic import BaseModel from sqlalchemy.orm import Session -from src.storage.db.models import Thread, User +from src.storage.db.models import User +from src.storage.conversation import ConversationManager from server.routers.auth_router import get_admin_user from server.utils.auth_middleware import get_db, get_required_user from src import executor @@ -114,6 +115,7 @@ async def chat_agent( config: dict = Body({}), meta: dict = Body({}), current_user: User = Depends(get_required_user), + db: Session = Depends(get_db), ): """使用特定智能体进行对话（需要登录）""" @@ -138,6 +140,100 @@ async def chat_agent( + b"\n" ) + async def save_messages_from_langgraph_state( + agent_instance, + conversation, + conv_mgr, + config_dict, + ): + """ + 从 LangGraph state 中读取完整消息并保存到数据库 + 这样可以获得完整的 tool_calls 参数 + """ + try: + graph = await agent_instance.get_graph() + state = await graph.aget_state(config_dict) + + if not state or not state.values: + logger.warning("No state found in LangGraph") + return + + messages = state.values.get("messages", []) + logger.debug(f"Retrieved {len(messages)} messages from LangGraph state") + + # 获取已保存的消息数量，避免重复保存 + existing_messages = conv_mgr.get_messages(conversation.id) + existing_count = len(existing_messages) + + # 只保存新增的消息 + new_messages = messages[existing_count:] + + for msg in new_messages: + msg_dict = msg.model_dump() if hasattr(msg, "model_dump") else {} + msg_type = msg_dict.get("type", "unknown") + + if msg_type == "human": + # 用户消息（理论上已经保存过了，跳过） + continue + + elif msg_type == "ai": + # AI 消息 + content = msg_dict.get("content", "") + tool_calls_data = msg_dict.get("tool_calls", []) + + # 保存 AI 消息 + ai_msg = conv_mgr.add_message( + conversation_id=conversation.id, + role="assistant", + content=content, + message_type="text", + extra_metadata=msg_dict, # 保存原始 model_dump + ) + + # 保存 tool_calls（如果有） + if tool_calls_data: + logger.debug(f"Saving {len(tool_calls_data)} tool calls from AI message") + for tc in tool_calls_data: + conv_mgr.add_tool_call( + message_id=ai_msg.id, + tool_name=tc.get("name", "unknown"), + tool_input=tc.get("args", {}), # 完整的参数 + status="pending", # 工具还未执行 + ) + + logger.debug(f"Saved AI message {ai_msg.id} with {len(tool_calls_data)} tool calls") + + elif msg_type == "tool": + # 工具执行结果消息 + # 需要找到对应的 tool_call 记录并更新 + tool_call_id = msg_dict.get("tool_call_id") + content = msg_dict.get("content", "") + + if tool_call_id: + # 通过 tool_call_id 找到对应的 tool_call 记录 + # 注意：LangGraph 的 tool_call_id 和我们数据库的 id 不同 + # 我们需要通过最近的 AI 消息找到对应的 tool_call + recent_ai_msgs = conv_mgr.get_messages(conversation.id, limit=10) + for recent_msg in reversed(recent_ai_msgs): + if recent_msg.role == "assistant" and recent_msg.tool_calls: + for tc in recent_msg.tool_calls: + # 这里简化处理：按顺序更新 pending 状态的 tool_call + if tc.status == "pending" and not tc.tool_output: + tc.tool_output = content + tc.status = "success" + db.commit() + logger.debug(f"Updated tool_call {tc.id} with output") + break + break + + logger.debug(f"Processed message type={msg_type}") + + logger.info(f"Saved {len(new_messages)} new messages from LangGraph state") + + except Exception as e: + logger.error(f"Error saving messages from LangGraph state: {e}") + logger.error(traceback.format_exc()) + async def stream_messages(): # 代表服务端已经收到了请求 yield make_chunk(status="init", meta=meta, msg=HumanMessage(content=query).model_dump()) @@ -162,30 +258,68 @@ async def chat_agent( input_context = {"user_id": user_id, "thread_id": thread_id} + # Initialize conversation manager + conv_manager = ConversationManager(db) + + # Get or create conversation + conversation = None + if thread_id: + conversation = conv_manager.get_conversation_by_thread_id(thread_id) + if not conversation: + try: + # Auto-create conversation for existing thread + conversation = conv_manager.create_conversation( + user_id=user_id, + agent_id=agent_id, + title=(query[:50] + "..." if len(query) > 50 else query) if query else "新的对话", + thread_id=thread_id, + ) + logger.info(f"Auto-created conversation for thread_id {thread_id}") + except Exception as e: + logger.error(f"Failed to auto-create conversation: {e}") + conversation = None + + # Save user message + if conversation: + try: + conv_manager.add_message( + conversation_id=conversation.id, + role="user", + content=query, + message_type="text", + extra_metadata={"raw_message": HumanMessage(content=query).model_dump()}, + ) + except Exception as e: + logger.error(f"Error saving user message: {e}") + try: - # Output guard for streaming - accumulated_content = "" + # Stream messages (only for display, don't save yet) async for msg, metadata in agent.stream_messages(messages, input_context=input_context): - # logger.debug(f"msg: {msg.model_dump()}, metadata: {metadata}") if isinstance(msg, AIMessageChunk): - accumulated_content += msg.content - if conf.enable_content_guard and content_guard.check(accumulated_content): - logger.warning(f"Sensitive content detected in stream: {accumulated_content}") + # Content guard + if conf.enable_content_guard and content_guard.check(msg.content): + logger.warning("Sensitive content detected in stream") yield make_chunk(message="检测到敏感内容，已中断输出", status="error") return yield make_chunk(content=msg.content, msg=msg.model_dump(), metadata=metadata, status="loading") + elif isinstance(msg, ToolMessage): + yield make_chunk(msg=msg.model_dump(), metadata=metadata, status="loading") else: yield make_chunk(msg=msg.model_dump(), metadata=metadata, status="loading") - # Additional content guard with llm - if conf.enable_content_guard and content_guard.check_with_llm(accumulated_content): - logger.warning(f"Content guard triggered: {accumulated_content=}") - yield make_chunk(message="检测到敏感内容，已中断输出", status="error") - return - yield make_chunk(status="finished", meta=meta) + + # After streaming finished, save all messages from LangGraph state + if conversation: + langgraph_config = {"configurable": {"thread_id": thread_id, "user_id": user_id}} + await save_messages_from_langgraph_state( + agent_instance=agent, + conversation=conversation, + conv_mgr=conv_manager, + config_dict=langgraph_config, + ) except Exception as e: logger.error(f"Error streaming messages: {e}, {traceback.format_exc()}") yield make_chunk(message=f"Error streaming messages: {e}", status="error") @@ -251,15 +385,53 @@ async def save_agent_config(agent_id: str, config: dict = Body(...), current_use @chat.get("/agent/{agent_id}/history") -async def get_agent_history(agent_id: str, thread_id: str, current_user: User = Depends(get_required_user)): - """获取智能体历史消息（需要登录）""" +async def get_agent_history( + agent_id: str, thread_id: str, current_user: User = Depends(get_required_user), db: Session = Depends(get_db) +): + """获取智能体历史消息（需要登录）- NEW STORAGE ONLY""" try: - # 获取Agent实例和配置类 - if not (agent := agent_manager.get_agent(agent_id)): + # 获取Agent实例验证 + if not agent_manager.get_agent(agent_id): raise HTTPException(status_code=404, detail=f"智能体 {agent_id} 不存在") - # 获取历史消息 - history = await agent.get_history(user_id=str(current_user.id), thread_id=thread_id) + # Use new storage system ONLY + conv_manager = ConversationManager(db) + messages = conv_manager.get_messages_by_thread_id(thread_id) + + # Convert to frontend-compatible format + history = [] + for msg in messages: + # Map role to type that frontend expects + role_type_map = { + "user": "human", + "assistant": "ai", + "tool": "tool", + "system": "system" + } + + msg_dict = { + "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, + } + + # Add tool calls if present (for AI messages) + if msg.tool_calls and len(msg.tool_calls) > 0: + msg_dict["tool_calls"] = [ + { + "id": str(tc.id), + "name": tc.tool_name, + "function": {"name": tc.tool_name}, # Frontend compatibility + "args": tc.tool_input or {}, + "tool_call_result": {"content": tc.tool_output} if tc.tool_output else None, + "status": tc.status, + } + for tc in msg.tool_calls + ] + + history.append(msg_dict) + + logger.info(f"Loaded {len(history)} messages from new storage for thread {thread_id}") return {"history": history} except Exception as e: @@ -290,7 +462,6 @@ async def get_agent_config(agent_id: str, current_user: User = Depends(get_requi class ThreadCreate(BaseModel): title: str | None = None agent_id: str - description: str | None = None metadata: dict | None = None @@ -299,7 +470,6 @@ class ThreadResponse(BaseModel): user_id: str agent_id: str title: str | None = None - description: str | None = None create_at: str update_at: str @@ -313,79 +483,81 @@ class ThreadResponse(BaseModel): async def create_thread( thread: ThreadCreate, db: Session = Depends(get_db), current_user: User = Depends(get_required_user) ): - """创建新对话线程""" + """创建新对话线程 (使用新存储系统)""" thread_id = str(uuid.uuid4()) logger.debug(f"thread.agent_id: {thread.agent_id}") - new_thread = Thread( - id=thread_id, + # Create conversation using new storage system + conv_manager = ConversationManager(db) + conversation = conv_manager.create_conversation( user_id=str(current_user.id), agent_id=thread.agent_id, title=thread.title or "新的对话", - description=thread.description, + thread_id=thread_id, + metadata=thread.metadata, ) - db.add(new_thread) - db.commit() - db.refresh(new_thread) + logger.info(f"Created conversation with thread_id: {thread_id}") return { - "id": new_thread.id, - "user_id": new_thread.user_id, - "agent_id": new_thread.agent_id, - "title": new_thread.title, - "description": new_thread.description, - "create_at": new_thread.create_at.isoformat(), - "update_at": new_thread.update_at.isoformat(), + "id": conversation.thread_id, + "user_id": conversation.user_id, + "agent_id": conversation.agent_id, + "title": conversation.title, + "create_at": conversation.created_at.isoformat(), + "update_at": conversation.updated_at.isoformat(), } @chat.get("/threads", response_model=list[ThreadResponse]) async def list_threads(agent_id: str, db: Session = Depends(get_db), current_user: User = Depends(get_required_user)): - """获取用户的所有对话线程""" + """获取用户的所有对话线程 (使用新存储系统)""" assert agent_id, "agent_id 不能为空" - query = db.query(Thread).filter( - Thread.user_id == str(current_user.id), - Thread.status == 1, - Thread.agent_id == agent_id, - ) logger.debug(f"agent_id: {agent_id}") - threads = query.order_by(Thread.update_at.desc()).all() + # Use new storage system + conv_manager = ConversationManager(db) + conversations = conv_manager.list_conversations( + user_id=str(current_user.id), + agent_id=agent_id, + status="active", + ) return [ { - "id": thread.id, - "user_id": thread.user_id, - "agent_id": thread.agent_id, - "title": thread.title, - "description": thread.description, - "create_at": thread.create_at.isoformat(), - "update_at": thread.update_at.isoformat(), + "id": conv.thread_id, + "user_id": conv.user_id, + "agent_id": conv.agent_id, + "title": conv.title, + "create_at": conv.created_at.isoformat(), + "update_at": conv.updated_at.isoformat(), } - for thread in threads + for conv in conversations ] @chat.delete("/thread/{thread_id}") async def delete_thread(thread_id: str, db: Session = Depends(get_db), current_user: User = Depends(get_required_user)): - """删除对话线程""" - thread = db.query(Thread).filter(Thread.id == thread_id, Thread.user_id == str(current_user.id)).first() + """删除对话线程 (使用新存储系统)""" + # Use new storage system + conv_manager = ConversationManager(db) + conversation = conv_manager.get_conversation_by_thread_id(thread_id) - if not thread: + if not conversation or conversation.user_id != str(current_user.id): raise HTTPException(status_code=404, detail="对话线程不存在") - # 软删除 - thread.status = 0 - db.commit() + # Soft delete + success = conv_manager.delete_conversation(thread_id, soft_delete=True) + + if not success: + raise HTTPException(status_code=500, detail="删除失败") return {"message": "删除成功"} class ThreadUpdate(BaseModel): title: str | None = None - description: str | None = None @chat.put("/thread/{thread_id}", response_model=ThreadResponse) @@ -395,31 +567,28 @@ async def update_thread( db: Session = Depends(get_db), current_user: User = Depends(get_required_user), ): - """更新对话线程信息""" - thread = ( - db.query(Thread) - .filter(Thread.id == thread_id, Thread.user_id == str(current_user.id), Thread.status == 1) - .first() - ) + """更新对话线程信息 (使用新存储系统)""" + # Use new storage system + conv_manager = ConversationManager(db) + conversation = conv_manager.get_conversation_by_thread_id(thread_id) - if not thread: + if not conversation or conversation.user_id != str(current_user.id) or conversation.status == "deleted": raise HTTPException(status_code=404, detail="对话线程不存在") - if thread_update.title is not None: - thread.title = thread_update.title + # Update conversation + updated_conv = conv_manager.update_conversation( + thread_id=thread_id, + title=thread_update.title, + ) - if thread_update.description is not None: - thread.description = thread_update.description - - db.commit() - db.refresh(thread) + if not updated_conv: + raise HTTPException(status_code=500, detail="更新失败") return { - "id": thread.id, - "user_id": thread.user_id, - "agent_id": thread.agent_id, - "title": thread.title, - "description": thread.description, - "create_at": thread.create_at.isoformat(), - "update_at": thread.update_at.isoformat(), + "id": updated_conv.thread_id, + "user_id": updated_conv.user_id, + "agent_id": updated_conv.agent_id, + "title": updated_conv.title, + "create_at": updated_conv.created_at.isoformat(), + "update_at": updated_conv.updated_at.isoformat(), } diff --git a/server/routers/dashboard_router.py b/server/routers/dashboard_router.py new file mode 100644 index 00000000..54c6d2f9 --- /dev/null +++ b/server/routers/dashboard_router.py @@ -0,0 +1,231 @@ +""" +Dashboard Router - Statistics and monitoring endpoints + +Provides centralized dashboard APIs for monitoring system-wide statistics. +""" + +import traceback +from datetime import datetime, timedelta + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel +from sqlalchemy import distinct, func +from sqlalchemy.orm import Session + +from server.routers.auth_router import get_admin_user +from server.utils.auth_middleware import get_db +from src.storage.conversation import ConversationManager +from src.storage.db.models import User +from src.utils.logging_config import logger + + +dashboard = APIRouter(prefix="/dashboard", tags=["Dashboard"]) + + +# ============================================================================= +# Response Models +# ============================================================================= + + +class ConversationListItem(BaseModel): + """Conversation list item""" + + thread_id: str + user_id: str + agent_id: str + title: str + status: str + message_count: int + created_at: str + updated_at: str + + +class ConversationDetailResponse(BaseModel): + """Conversation detail""" + + thread_id: str + user_id: str + agent_id: str + title: str + status: str + message_count: int + created_at: str + updated_at: str + total_tokens: int + messages: list[dict] + + +# ============================================================================= +# Conversation Management +# ============================================================================= + + +@dashboard.get("/conversations", response_model=list[ConversationListItem]) +async def get_all_conversations( + user_id: str | None = None, + agent_id: str | None = None, + status: str = "active", + limit: int = 100, + offset: int = 0, + db: Session = Depends(get_db), + current_user: User = Depends(get_admin_user), +): + """Get all conversations (Admin only)""" + from src.storage.db.models import Conversation, ConversationStats + + try: + # Build query + query = db.query(Conversation, ConversationStats).outerjoin( + ConversationStats, Conversation.id == ConversationStats.conversation_id + ) + + # Apply filters + if user_id: + query = query.filter(Conversation.user_id == user_id) + if agent_id: + query = query.filter(Conversation.agent_id == agent_id) + if status != "all": + query = query.filter(Conversation.status == status) + + # Order and paginate + query = query.order_by(Conversation.updated_at.desc()).limit(limit).offset(offset) + + results = query.all() + + return [ + { + "thread_id": conv.thread_id, + "user_id": conv.user_id, + "agent_id": conv.agent_id, + "title": conv.title, + "status": conv.status, + "message_count": stats.message_count if stats else 0, + "created_at": conv.created_at.isoformat(), + "updated_at": conv.updated_at.isoformat(), + } + for conv, stats in results + ] + except Exception as e: + logger.error(f"Error getting conversations: {e}") + logger.error(traceback.format_exc()) + raise HTTPException(status_code=500, detail=f"Failed to get conversations: {str(e)}") + + +@dashboard.get("/conversations/{thread_id}", response_model=ConversationDetailResponse) +async def get_conversation_detail( + thread_id: str, + db: Session = Depends(get_db), + current_user: User = Depends(get_admin_user), +): + """Get conversation detail (Admin only)""" + try: + conv_manager = ConversationManager(db) + conversation = conv_manager.get_conversation_by_thread_id(thread_id) + + if not conversation: + raise HTTPException(status_code=404, detail="Conversation not found") + + # Get messages and stats + messages = conv_manager.get_messages(conversation.id) + stats = conv_manager.get_stats(conversation.id) + + # Format messages + message_list = [] + for msg in messages: + msg_dict = { + "id": msg.id, + "role": msg.role, + "content": msg.content, + "message_type": msg.message_type, + "created_at": msg.created_at.isoformat(), + } + + # Include tool calls if present + if msg.tool_calls: + msg_dict["tool_calls"] = [ + { + "id": tc.id, + "tool_name": tc.tool_name, + "tool_input": tc.tool_input, + "tool_output": tc.tool_output, + "status": tc.status, + } + for tc in msg.tool_calls + ] + + message_list.append(msg_dict) + + return { + "thread_id": conversation.thread_id, + "user_id": conversation.user_id, + "agent_id": conversation.agent_id, + "title": conversation.title, + "status": conversation.status, + "message_count": stats.message_count if stats else len(message_list), + "created_at": conversation.created_at.isoformat(), + "updated_at": conversation.updated_at.isoformat(), + "total_tokens": stats.total_tokens if stats else 0, + "messages": message_list, + } + except HTTPException: + raise + except Exception as e: + logger.error(f"Error getting conversation detail: {e}") + logger.error(traceback.format_exc()) + raise HTTPException(status_code=500, detail=f"Failed to get conversation detail: {str(e)}") + + +# ============================================================================= +# Statistics +# ============================================================================= + + +@dashboard.get("/stats") +async def get_dashboard_stats( + db: Session = Depends(get_db), + current_user: User = Depends(get_admin_user), +): + """Get dashboard statistics (Admin only)""" + from src.storage.db.models import Conversation, Message + + try: + # Basic counts + total_conversations = db.query(func.count(Conversation.id)).scalar() or 0 + active_conversations = ( + db.query(func.count(Conversation.id)).filter(Conversation.status == "active").scalar() or 0 + ) + total_messages = db.query(func.count(Message.id)).scalar() or 0 + total_users = db.query(func.count(distinct(Conversation.user_id))).scalar() or 0 + + # Conversations by agent + conversations_by_agent = ( + db.query(Conversation.agent_id, func.count(Conversation.id).label("count")) + .group_by(Conversation.agent_id) + .all() + ) + + # Recent activity (last 24 hours) + yesterday = datetime.utcnow() - timedelta(days=1) + recent_conversations = ( + db.query(func.count(Conversation.id)).filter(Conversation.created_at >= yesterday).scalar() or 0 + ) + recent_messages = db.query(func.count(Message.id)).filter(Message.created_at >= yesterday).scalar() or 0 + + return { + "timestamp": datetime.utcnow().isoformat(), + "total_conversations": total_conversations, + "active_conversations": active_conversations, + "total_messages": total_messages, + "total_users": total_users, + "conversations_by_agent": [ + {"agent_id": agent_id or "unknown", "count": count} for agent_id, count in conversations_by_agent + ], + "recent_activity": { + "conversations_24h": recent_conversations, + "messages_24h": recent_messages, + }, + } + except Exception as e: + logger.error(f"Error getting dashboard stats: {e}") + logger.error(traceback.format_exc()) + raise HTTPException(status_code=500, detail=f"Failed to get dashboard stats: {str(e)}") diff --git a/src/storage/conversation/__init__.py b/src/storage/conversation/__init__.py new file mode 100644 index 00000000..a844bb86 --- /dev/null +++ b/src/storage/conversation/__init__.py @@ -0,0 +1,4 @@ +from .manager import ConversationManager + +__all__ = ["ConversationManager"] + diff --git a/src/storage/conversation/manager.py b/src/storage/conversation/manager.py new file mode 100644 index 00000000..b61424fe --- /dev/null +++ b/src/storage/conversation/manager.py @@ -0,0 +1,356 @@ +""" +Conversation Storage Manager + +Manages conversation data storage including messages, tool calls, and statistics. +""" + +import uuid +from datetime import datetime + +from sqlalchemy.orm import Session, joinedload + +from src.storage.db.models import Conversation, ConversationStats, Message, ToolCall +from src.utils import logger + + +class ConversationManager: + """Manager for conversation storage operations""" + + def __init__(self, db_session: Session): + self.db = db_session + + def create_conversation( + self, + user_id: str, + agent_id: str, + title: str | None = None, + thread_id: str | None = None, + metadata: dict | None = None, + ) -> Conversation: + """ + Create a new conversation + + Args: + user_id: User ID + agent_id: Agent ID + title: Conversation title + thread_id: Optional thread ID (will generate UUID if not provided) + metadata: Optional additional metadata + + Returns: + Created Conversation object + """ + if not thread_id: + thread_id = str(uuid.uuid4()) + + conversation = Conversation( + thread_id=thread_id, + user_id=str(user_id), + agent_id=agent_id, + title=title or "New Conversation", + status="active", + extra_metadata=metadata if metadata else None, + ) + + self.db.add(conversation) + # Flush to assign primary key without committing + self.db.flush() + + # Create associated stats record and commit once + stats = ConversationStats(conversation_id=conversation.id) + self.db.add(stats) + self.db.commit() + self.db.refresh(conversation) + + logger.info(f"Created conversation: {conversation.thread_id} for user {user_id}") + return conversation + + def get_conversation_by_thread_id(self, thread_id: str) -> Conversation | None: + """ + Get conversation by thread ID + + Args: + thread_id: Thread ID + + Returns: + Conversation object or None if not found + """ + return self.db.query(Conversation).filter(Conversation.thread_id == thread_id).first() + + def add_message( + self, + conversation_id: int, + role: str, + content: str, + message_type: str = "text", + extra_metadata: dict | None = None, + ) -> Message: + """ + Add a message to a conversation + + Args: + conversation_id: Conversation ID + role: Message role (user/assistant/system/tool) + content: Message content + message_type: Message type (text/tool_call/tool_result) + extra_metadata: Additional metadata (complete message dump) + + Returns: + Created Message object + """ + message = Message( + conversation_id=conversation_id, + role=role, + content=content, + message_type=message_type, + extra_metadata=extra_metadata or {}, + ) + + self.db.add(message) + self.db.commit() + self.db.refresh(message) + + # Update conversation stats + self._update_message_count(conversation_id) + + logger.debug(f"Added {role} message to conversation {conversation_id}") + return message + + def add_tool_call( + self, + message_id: int, + tool_name: str, + tool_input: dict | None = None, + tool_output: str | None = None, + status: str = "pending", + error_message: str | None = None, + ) -> ToolCall: + """ + Add a tool call record + + Args: + message_id: Message ID + tool_name: Tool name + tool_input: Tool input parameters + tool_output: Tool execution result + status: Status (pending/success/error) + error_message: Error message if failed + + Returns: + Created ToolCall object + """ + tool_call = ToolCall( + message_id=message_id, + tool_name=tool_name, + tool_input=tool_input or {}, + tool_output=tool_output, + status=status, + error_message=error_message, + ) + + self.db.add(tool_call) + self.db.commit() + self.db.refresh(tool_call) + + logger.debug(f"Added tool call {tool_name} to message {message_id}") + return tool_call + + def get_messages( + self, conversation_id: int, limit: int | None = None, offset: int = 0 + ) -> list[Message]: + """ + Get messages for a conversation + + Args: + conversation_id: Conversation ID + limit: Maximum number of messages to return + offset: Number of messages to skip + + Returns: + List of Message objects + """ + query = ( + self.db.query(Message) + .filter(Message.conversation_id == conversation_id) + .order_by(Message.created_at.asc()) + .options(joinedload(Message.tool_calls)) + ) + + if limit: + query = query.limit(limit).offset(offset) + + return query.all() + + def get_messages_by_thread_id( + self, thread_id: str, limit: int | None = None, offset: int = 0 + ) -> list[Message]: + """ + Get messages for a conversation by thread ID + + Args: + thread_id: Thread ID + limit: Maximum number of messages to return + offset: Number of messages to skip + + Returns: + List of Message objects + """ + conversation = self.get_conversation_by_thread_id(thread_id) + if not conversation: + return [] + + return self.get_messages(conversation.id, limit, offset) + + def list_conversations( + self, user_id: str | None = None, agent_id: str | None = None, status: str = "active" + ) -> list[Conversation]: + """ + List conversations for a user or all users + + Args: + user_id: User ID (optional, if None or empty string, returns all users' conversations) + agent_id: Optional agent ID filter + status: Conversation status filter + + Returns: + List of Conversation objects + """ + query = self.db.query(Conversation).filter(Conversation.status == status) + + # Only filter by user_id if it's provided and not empty + if user_id: + query = query.filter(Conversation.user_id == str(user_id)) + + if agent_id: + query = query.filter(Conversation.agent_id == agent_id) + + return query.order_by(Conversation.updated_at.desc()).all() + + def update_conversation( + self, + thread_id: str, + title: str | None = None, + status: str | None = None, + metadata: dict | None = None, + ) -> Conversation | None: + """ + Update conversation information + + Args: + thread_id: Thread ID + title: New title + status: New status + metadata: Additional metadata to merge + + Returns: + Updated Conversation object or None if not found + """ + conversation = self.get_conversation_by_thread_id(thread_id) + if not conversation: + return None + + if title is not None: + conversation.title = title + if status is not None: + conversation.status = status + + # Handle metadata updates + if metadata is not None: + current_metadata = conversation.extra_metadata or {} + current_metadata.update(metadata) + conversation.extra_metadata = current_metadata + + conversation.updated_at = datetime.now() + self.db.commit() + self.db.refresh(conversation) + + logger.info(f"Updated conversation {thread_id}") + return conversation + + def delete_conversation(self, thread_id: str, soft_delete: bool = True) -> bool: + """ + Delete a conversation + + Args: + thread_id: Thread ID + soft_delete: If True, mark as deleted; if False, permanently delete + + Returns: + True if successful, False otherwise + """ + conversation = self.get_conversation_by_thread_id(thread_id) + if not conversation: + return False + + if soft_delete: + conversation.status = "deleted" + self.db.commit() + logger.info(f"Soft deleted conversation {thread_id}") + else: + self.db.delete(conversation) + self.db.commit() + logger.info(f"Permanently deleted conversation {thread_id}") + + return True + + def get_stats(self, conversation_id: int) -> ConversationStats | None: + """ + Get conversation statistics + + Args: + conversation_id: Conversation ID + + Returns: + ConversationStats object or None if not found + """ + return self.db.query(ConversationStats).filter(ConversationStats.conversation_id == conversation_id).first() + + def update_stats( + self, + conversation_id: int, + tokens_used: int | None = None, + model_used: str | None = None, + user_feedback: dict | None = None, + ) -> ConversationStats | None: + """ + Update conversation statistics + + Args: + conversation_id: Conversation ID + tokens_used: Number of tokens to add + model_used: Model name + user_feedback: User feedback data + + Returns: + Updated ConversationStats object or None if not found + """ + stats = self.get_stats(conversation_id) + if not stats: + return None + + if tokens_used is not None: + stats.total_tokens += tokens_used + if model_used is not None: + stats.model_used = model_used + if user_feedback is not None: + stats.user_feedback = user_feedback + + stats.updated_at = datetime.now() + self.db.commit() + self.db.refresh(stats) + + return stats + + def _update_message_count(self, conversation_id: int) -> None: + """ + Update message count in conversation stats + + Args: + conversation_id: Conversation ID + """ + stats = self.get_stats(conversation_id) + if stats: + message_count = self.db.query(Message).filter(Message.conversation_id == conversation_id).count() + stats.message_count = message_count + self.db.commit() + diff --git a/src/storage/db/__init__.py b/src/storage/db/__init__.py index 151c6e21..9809b21b 100644 --- a/src/storage/db/__init__.py +++ b/src/storage/db/__init__.py @@ -10,7 +10,6 @@ from src.storage.db.models import ( # noqa: E402, F401 KnowledgeFile, KnowledgeNode, OperationLog, - Thread, User, ) # noqa: E402 @@ -18,7 +17,6 @@ __all__ = [ "Base", "User", "OperationLog", - "Thread", "KnowledgeDatabase", "KnowledgeFile", "KnowledgeNode", diff --git a/src/storage/db/models.py b/src/storage/db/models.py index ce5b8972..b430b3da 100644 --- a/src/storage/db/models.py +++ b/src/storage/db/models.py @@ -113,20 +113,135 @@ class KnowledgeNode(Base): } -class Thread(Base): - """对话线程表""" +class Conversation(Base): + """Conversation table - new storage system""" - __tablename__ = "thread" + __tablename__ = "conversations" - id = Column(String(64), primary_key=True, index=True, comment="线程ID") - user_id = Column(String(64), index=True, nullable=False, comment="用户ID") - agent_id = Column(String(64), index=True, nullable=False, comment="智能体ID") - title = Column(String(255), nullable=True, comment="标题") - create_at = Column(DateTime, default=func.now(), comment="创建时间") - update_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment="更新时间") + id = Column(Integer, primary_key=True, autoincrement=True, comment="Primary key") + thread_id = Column(String(64), unique=True, index=True, nullable=False, comment="Thread ID (UUID)") + user_id = Column(String(64), index=True, nullable=False, comment="User ID") + agent_id = Column(String(64), index=True, nullable=False, comment="Agent ID") + title = Column(String(255), nullable=True, comment="Conversation title") + status = Column(String(20), default="active", comment="Status: active/archived/deleted") + created_at = Column(DateTime, default=func.now(), comment="Creation time") + updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment="Update time") + extra_metadata = Column(JSON, nullable=True, comment="Additional metadata") - description = Column(String(255), nullable=True, comment="描述") - status = Column(Integer, default=1, comment="状态") + # Relationships + messages = relationship("Message", back_populates="conversation", cascade="all, delete-orphan") + stats = relationship( + "ConversationStats", back_populates="conversation", uselist=False, cascade="all, delete-orphan" + ) + + def to_dict(self): + return { + "id": self.id, + "thread_id": self.thread_id, + "user_id": self.user_id, + "agent_id": self.agent_id, + "title": self.title, + "status": self.status, + "created_at": self.created_at.isoformat() if self.created_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + "metadata": self.extra_metadata or {}, + } + + +class Message(Base): + """Message table - stores conversation messages""" + + __tablename__ = "messages" + + id = Column(Integer, primary_key=True, autoincrement=True, comment="Primary key") + conversation_id = Column( + Integer, ForeignKey("conversations.id"), nullable=False, index=True, comment="Conversation ID" + ) + role = Column(String(20), nullable=False, comment="Message role: user/assistant/system/tool") + content = Column(Text, nullable=False, comment="Message content") + message_type = Column(String(30), default="text", comment="Message type: text/tool_call/tool_result") + created_at = Column(DateTime, default=func.now(), comment="Creation time") + token_count = Column(Integer, nullable=True, comment="Token count (optional)") + extra_metadata = Column(JSON, nullable=True, comment="Additional metadata (complete message dump)") + + # Relationships + conversation = relationship("Conversation", back_populates="messages") + tool_calls = relationship("ToolCall", back_populates="message", cascade="all, delete-orphan") + + def to_dict(self): + return { + "id": self.id, + "conversation_id": self.conversation_id, + "role": self.role, + "content": self.content, + "message_type": self.message_type, + "created_at": self.created_at.isoformat() if self.created_at else None, + "token_count": self.token_count, + "metadata": self.extra_metadata or {}, + "tool_calls": [tc.to_dict() for tc in self.tool_calls] if self.tool_calls else [], + } + + +class ToolCall(Base): + """ToolCall table - stores tool invocations""" + + __tablename__ = "tool_calls" + + id = Column(Integer, primary_key=True, autoincrement=True, comment="Primary key") + message_id = Column(Integer, ForeignKey("messages.id"), nullable=False, index=True, comment="Message ID") + tool_name = Column(String(100), nullable=False, comment="Tool name") + tool_input = Column(JSON, nullable=True, comment="Tool input parameters") + tool_output = Column(Text, nullable=True, comment="Tool execution result") + status = Column(String(20), default="pending", comment="Status: pending/success/error") + error_message = Column(Text, nullable=True, comment="Error message if failed") + created_at = Column(DateTime, default=func.now(), comment="Creation time") + + # Relationships + message = relationship("Message", back_populates="tool_calls") + + def to_dict(self): + return { + "id": self.id, + "message_id": self.message_id, + "tool_name": self.tool_name, + "tool_input": self.tool_input or {}, + "tool_output": self.tool_output, + "status": self.status, + "error_message": self.error_message, + "created_at": self.created_at.isoformat() if self.created_at else None, + } + + +class ConversationStats(Base): + """ConversationStats table - stores conversation statistics""" + + __tablename__ = "conversation_stats" + + id = Column(Integer, primary_key=True, autoincrement=True, comment="Primary key") + conversation_id = Column( + Integer, ForeignKey("conversations.id"), unique=True, nullable=False, comment="Conversation ID" + ) + message_count = Column(Integer, default=0, comment="Total message count") + total_tokens = Column(Integer, default=0, comment="Total tokens used") + model_used = Column(String(100), nullable=True, comment="Model used") + user_feedback = Column(JSON, nullable=True, comment="User feedback") + created_at = Column(DateTime, default=func.now(), comment="Creation time") + updated_at = Column(DateTime, default=func.now(), onupdate=func.now(), comment="Update time") + + # Relationships + conversation = relationship("Conversation", back_populates="stats") + + def to_dict(self): + return { + "id": self.id, + "conversation_id": self.conversation_id, + "message_count": self.message_count, + "total_tokens": self.total_tokens, + "model_used": self.model_used, + "user_feedback": self.user_feedback or {}, + "created_at": self.created_at.isoformat() if self.created_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + } class User(Base): diff --git a/web/src/apis/dashboard_api.js b/web/src/apis/dashboard_api.js new file mode 100644 index 00000000..d06e8ee8 --- /dev/null +++ b/web/src/apis/dashboard_api.js @@ -0,0 +1,48 @@ +import { apiAdminGet } from './base' + +/** + * Dashboard API模块 + * 用于管理员查看所有用户的对话记录 + */ + +export const dashboardApi = { + /** + * 获取所有对话记录 + * @param {Object} params - 查询参数 + * @param {string} params.user_id - 用户ID过滤 + * @param {string} params.agent_id - 智能体ID过滤 + * @param {string} params.status - 状态过滤 (active/deleted/all) + * @param {number} params.limit - 每页数量 + * @param {number} params.offset - 偏移量 + * @returns {Promise} - 对话列表 + */ + getConversations: (params = {}) => { + const queryParams = new URLSearchParams() + if (params.user_id) queryParams.append('user_id', params.user_id) + if (params.agent_id) queryParams.append('agent_id', params.agent_id) + if (params.status) queryParams.append('status', params.status) + if (params.limit) queryParams.append('limit', params.limit) + if (params.offset) queryParams.append('offset', params.offset) + + return apiAdminGet(`/api/dashboard/conversations?${queryParams.toString()}`) + }, + + /** + * 获取对话详情 + * @param {string} threadId - 对话线程ID + * @returns {Promise} - 对话详情 + */ + getConversationDetail: (threadId) => { + return apiAdminGet(`/api/dashboard/conversations/${threadId}`) + }, + + /** + * 获取Dashboard统计信息 + * @returns {Promise} - 统计信息 + */ + getStats: () => { + return apiAdminGet('/api/dashboard/stats') + } +} + + diff --git a/web/src/layouts/AppLayout.vue b/web/src/layouts/AppLayout.vue index 696f0d0d..a3150ec4 100644 --- a/web/src/layouts/AppLayout.vue +++ b/web/src/layouts/AppLayout.vue @@ -5,7 +5,7 @@ import { GithubOutlined, ExclamationCircleOutlined, } from '@ant-design/icons-vue' -import { Bot, Waypoints, LibraryBig, Settings } from 'lucide-vue-next'; +import { Bot, Waypoints, LibraryBig, Settings, BarChart3 } from 'lucide-vue-next'; import { onLongPress } from '@vueuse/core' import { useConfigStore } from '@/stores/config' @@ -103,6 +103,11 @@ const mainList = [{ path: '/database', icon: LibraryBig, activeIcon: LibraryBig, + }, { + name: 'Dashboard', + path: '/dashboard', + icon: BarChart3, + activeIcon: BarChart3, } ] diff --git a/web/src/router/index.js b/web/src/router/index.js index 58b33efe..3955d2bc 100644 --- a/web/src/router/index.js +++ b/web/src/router/index.js @@ -90,6 +90,19 @@ const router = createRouter({ } ] }, + { + path: '/dashboard', + name: 'dashboard', + component: AppLayout, + children: [ + { + path: '', + name: 'DashboardComp', + component: () => import('../views/DashboardView.vue'), + meta: { keepAlive: false, requiresAuth: true, requiresAdmin: true } + } + ] + }, { path: '/:pathMatch(.*)*', name: 'NotFound', @@ -126,7 +139,7 @@ router.beforeEach(async (to, from, next) => { if (!agentStore.isInitialized) { await agentStore.initialize(); } - + const defaultAgent = agentStore.defaultAgent; if (defaultAgent && defaultAgent.id) { next(`/agent/${defaultAgent.id}`); diff --git a/web/src/utils/messageProcessor.js b/web/src/utils/messageProcessor.js index cff0e453..a892c42e 100644 --- a/web/src/utils/messageProcessor.js +++ b/web/src/utils/messageProcessor.js @@ -9,7 +9,7 @@ export class MessageProcessor { */ static convertToolResultToMessages(msgs) { const toolResponseMap = new Map(); - + // 构建工具响应映射 for (const item of msgs) { if (item.type === 'tool' && item.tool_call_id) { @@ -43,15 +43,27 @@ export class MessageProcessor { * @returns {Array} 对话数组 */ static convertServerHistoryToMessages(serverHistory) { - // 第一步：将所有tool消息与对应的tool call合并 - const mergedHistory = this.convertToolResultToMessages(serverHistory); + // Filter out standalone 'tool' messages since tool results are already in AI messages' tool_calls + // Backend new storage: tool results are embedded in AI messages' tool_calls array with tool_call_result field + const filteredHistory = serverHistory.filter(item => item.type !== 'tool'); - // 第二步：按照对话分组 + // 按照对话分组 const conversations = []; let currentConv = null; - for (const item of mergedHistory) { + for (const item of filteredHistory) { if (item.type === 'human') { + // Start new conversation, finalize previous one + if (currentConv) { + // Find the last AI message and mark it as final + for (let i = currentConv.messages.length - 1; i >= 0; i--) { + if (currentConv.messages[i].type === 'ai') { + currentConv.messages[i].isLast = true; + currentConv.status = 'finished'; + break; + } + } + } currentConv = { messages: [item], status: 'loading' @@ -59,11 +71,17 @@ export class MessageProcessor { conversations.push(currentConv); } else if (item.type === 'ai' && currentConv) { currentConv.messages.push(item); + } + } - if (item.response_metadata?.finish_reason === 'stop') { - item.isLast = true; + // Mark the last conversation as finished + if (currentConv && currentConv.messages.length > 0) { + // Find the last AI message and mark it as final + for (let i = currentConv.messages.length - 1; i >= 0; i--) { + if (currentConv.messages[i].type === 'ai') { + currentConv.messages[i].isLast = true; currentConv.status = 'finished'; - currentConv = null; + break; } } } @@ -86,7 +104,7 @@ export class MessageProcessor { // 合并后续chunks for (let i = 1; i < chunks.length; i++) { const chunk = chunks[i]; - + // 合并内容 if (chunk.content) { result.content += chunk.content; diff --git a/web/src/views/DashboardView.vue b/web/src/views/DashboardView.vue new file mode 100644 index 00000000..20742def --- /dev/null +++ b/web/src/views/DashboardView.vue @@ -0,0 +1,324 @@ + + + + + +