refactor(db): 将数据库操作迁移到异步SQLAlchemy
- 将DBManager重构为支持异步操作 - 更新auth_middleware使用异步会话 - 修改chat_router和dashboard_router使用异步数据库操作 - 重构ConversationManager为异步版本 - 添加aiosqlite和asyncpg依赖 - 优化数据库查询性能
This commit is contained in:
Wenjie Zhang
parent
ab06e9e1e9
commit
2126f45493
@ -6,6 +6,8 @@ readme = "README.md"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = [
|
||||
"asyncpg>=0.30.0",
|
||||
"aiosqlite>=0.20.0",
|
||||
"sqlalchemy[asyncio]>=2.0.0",
|
||||
"langchain>=1.0.2",
|
||||
"chromadb>=1.3",
|
||||
"colorlog>=6.9.0",
|
||||
|
||||
@ -8,7 +8,8 @@ from fastapi.responses import StreamingResponse
|
||||
from langchain.messages import AIMessageChunk, HumanMessage
|
||||
from langgraph.types import Command
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select
|
||||
|
||||
from src.storage.db.models import User, MessageFeedback, Message, Conversation
|
||||
from src.storage.conversation import ConversationManager
|
||||
@ -193,8 +194,8 @@ def _save_tool_message(conv_mgr, msg_dict):
|
||||
logger.warning(f"Tool call {tool_call_id} not found for update")
|
||||
|
||||
|
||||
def _require_user_conversation(conv_mgr: ConversationManager, thread_id: str, user_id: str) -> Conversation:
|
||||
conversation = conv_mgr.get_conversation_by_thread_id(thread_id)
|
||||
async def _require_user_conversation(conv_mgr: ConversationManager, thread_id: str, user_id: str) -> Conversation:
|
||||
conversation = await conv_mgr.get_conversation_by_thread_id(thread_id)
|
||||
if not conversation or conversation.user_id != str(user_id) or conversation.status == "deleted":
|
||||
raise HTTPException(status_code=404, detail="对话线程不存在")
|
||||
return conversation
|
||||
@ -455,7 +456,7 @@ async def chat_agent(
|
||||
meta: dict = Body({}),
|
||||
image_content: str | None = Body(None),
|
||||
current_user: User = Depends(get_required_user),
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""使用特定智能体进行对话（需要登录）"""
|
||||
start_time = asyncio.get_event_loop().time()
|
||||
@ -556,7 +557,7 @@ async def chat_agent(
|
||||
|
||||
# Save user message
|
||||
try:
|
||||
conv_manager.add_message_by_thread_id(
|
||||
await conv_manager.add_message_by_thread_id(
|
||||
thread_id=thread_id,
|
||||
role="user",
|
||||
content=query,
|
||||
@ -569,7 +570,7 @@ async def chat_agent(
|
||||
|
||||
try:
|
||||
assert thread_id, "thread_id is required"
|
||||
attachments = conv_manager.get_attachments_by_thread_id(thread_id)
|
||||
attachments = await conv_manager.get_attachments_by_thread_id(thread_id)
|
||||
input_context["attachments"] = attachments
|
||||
logger.debug(f"Loaded {len(attachments)} attachments for thread_id={thread_id}")
|
||||
except Exception as e:
|
||||
@ -648,8 +649,7 @@ async def chat_agent(
|
||||
logger.warning(f"Client disconnected, cancelling stream: {e}")
|
||||
|
||||
# 保存中断消息到数据库
|
||||
new_db = db_manager.get_session()
|
||||
try:
|
||||
async with db_manager.get_async_session_context() as new_db:
|
||||
new_conv_manager = ConversationManager(new_db)
|
||||
await save_partial_message(
|
||||
new_conv_manager,
|
||||
@ -658,8 +658,6 @@ async def chat_agent(
|
||||
error_message="对话已中断" if not full_msg else None,
|
||||
error_type="interrupted",
|
||||
)
|
||||
finally:
|
||||
new_db.close()
|
||||
|
||||
# 通知前端中断（可能发送不到，但用于一致性）
|
||||
yield make_chunk(status="interrupted", message="对话已中断", meta=meta)
|
||||
@ -671,8 +669,7 @@ async def chat_agent(
|
||||
error_type = "unexpected_error"
|
||||
|
||||
# 保存错误消息到数据库
|
||||
new_db = db_manager.get_session()
|
||||
try:
|
||||
async with db_manager.get_async_session_context() as new_db:
|
||||
new_conv_manager = ConversationManager(new_db)
|
||||
await save_partial_message(
|
||||
new_conv_manager,
|
||||
@ -681,8 +678,6 @@ async def chat_agent(
|
||||
error_message=error_msg,
|
||||
error_type=error_type,
|
||||
)
|
||||
finally:
|
||||
new_db.close()
|
||||
|
||||
yield make_chunk(
|
||||
status="error",
|
||||
@ -739,7 +734,7 @@ async def resume_agent_chat(
|
||||
thread_id: str = Body(...),
|
||||
approved: bool = Body(...),
|
||||
current_user: User = Depends(get_required_user),
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""恢复被人工审批中断的对话（需要登录）"""
|
||||
start_time = asyncio.get_event_loop().time()
|
||||
@ -874,7 +869,7 @@ 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), db: Session = Depends(get_db)
|
||||
agent_id: str, thread_id: str, current_user: User = Depends(get_required_user), db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""获取智能体历史消息（需要登录）- NEW STORAGE ONLY"""
|
||||
try:
|
||||
@ -884,7 +879,7 @@ async def get_agent_history(
|
||||
|
||||
# Use new storage system ONLY
|
||||
conv_manager = ConversationManager(db)
|
||||
messages = conv_manager.get_messages_by_thread_id(thread_id)
|
||||
messages = await conv_manager.get_messages_by_thread_id(thread_id)
|
||||
|
||||
# Convert to frontend-compatible format
|
||||
history = []
|
||||
@ -934,14 +929,14 @@ async def get_agent_state(
|
||||
agent_id: str,
|
||||
thread_id: str,
|
||||
current_user: User = Depends(get_required_user),
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
try:
|
||||
if not agent_manager.get_agent(agent_id):
|
||||
raise HTTPException(status_code=404, detail=f"智能体 {agent_id} 不存在")
|
||||
|
||||
conv_manager = ConversationManager(db)
|
||||
_require_user_conversation(conv_manager, thread_id, str(current_user.id))
|
||||
await _require_user_conversation(conv_manager, thread_id, str(current_user.id))
|
||||
|
||||
agent = agent_manager.get_agent(agent_id)
|
||||
graph = await agent.get_graph()
|
||||
@ -1019,7 +1014,7 @@ class AttachmentListResponse(BaseModel):
|
||||
|
||||
@chat.post("/thread", response_model=ThreadResponse)
|
||||
async def create_thread(
|
||||
thread: ThreadCreate, db: Session = Depends(get_db), current_user: User = Depends(get_required_user)
|
||||
thread: ThreadCreate, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user)
|
||||
):
|
||||
"""创建新对话线程 (使用新存储系统)"""
|
||||
thread_id = str(uuid.uuid4())
|
||||
@ -1027,7 +1022,7 @@ async def create_thread(
|
||||
|
||||
# Create conversation using new storage system
|
||||
conv_manager = ConversationManager(db)
|
||||
conversation = conv_manager.create_conversation(
|
||||
conversation = await conv_manager.create_conversation(
|
||||
user_id=str(current_user.id),
|
||||
agent_id=thread.agent_id,
|
||||
title=thread.title or "新的对话",
|
||||
@ -1048,7 +1043,7 @@ async def create_thread(
|
||||
|
||||
|
||||
@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)):
|
||||
async def list_threads(agent_id: str, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user)):
|
||||
"""获取用户的所有对话线程 (使用新存储系统)"""
|
||||
assert agent_id, "agent_id 不能为空"
|
||||
|
||||
@ -1056,7 +1051,7 @@ async def list_threads(agent_id: str, db: Session = Depends(get_db), current_use
|
||||
|
||||
# Use new storage system
|
||||
conv_manager = ConversationManager(db)
|
||||
conversations = conv_manager.list_conversations(
|
||||
conversations = await conv_manager.list_conversations(
|
||||
user_id=str(current_user.id),
|
||||
agent_id=agent_id,
|
||||
status="active",
|
||||
@ -1076,17 +1071,17 @@ async def list_threads(agent_id: str, db: Session = Depends(get_db), current_use
|
||||
|
||||
|
||||
@chat.delete("/thread/{thread_id}")
|
||||
async def delete_thread(thread_id: str, db: Session = Depends(get_db), current_user: User = Depends(get_required_user)):
|
||||
async def delete_thread(thread_id: str, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_required_user)):
|
||||
"""删除对话线程 (使用新存储系统)"""
|
||||
# Use new storage system
|
||||
conv_manager = ConversationManager(db)
|
||||
conversation = conv_manager.get_conversation_by_thread_id(thread_id)
|
||||
conversation = await conv_manager.get_conversation_by_thread_id(thread_id)
|
||||
|
||||
if not conversation or conversation.user_id != str(current_user.id):
|
||||
raise HTTPException(status_code=404, detail="对话线程不存在")
|
||||
|
||||
# Soft delete
|
||||
success = conv_manager.delete_conversation(thread_id, soft_delete=True)
|
||||
success = await conv_manager.delete_conversation(thread_id, soft_delete=True)
|
||||
|
||||
if not success:
|
||||
raise HTTPException(status_code=500, detail="删除失败")
|
||||
@ -1102,19 +1097,19 @@ class ThreadUpdate(BaseModel):
|
||||
async def update_thread(
|
||||
thread_id: str,
|
||||
thread_update: ThreadUpdate,
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_required_user),
|
||||
):
|
||||
"""更新对话线程信息 (使用新存储系统)"""
|
||||
# Use new storage system
|
||||
conv_manager = ConversationManager(db)
|
||||
conversation = conv_manager.get_conversation_by_thread_id(thread_id)
|
||||
conversation = await conv_manager.get_conversation_by_thread_id(thread_id)
|
||||
|
||||
if not conversation or conversation.user_id != str(current_user.id) or conversation.status == "deleted":
|
||||
raise HTTPException(status_code=404, detail="对话线程不存在")
|
||||
|
||||
# Update conversation
|
||||
updated_conv = conv_manager.update_conversation(
|
||||
updated_conv = await conv_manager.update_conversation(
|
||||
thread_id=thread_id,
|
||||
title=thread_update.title,
|
||||
)
|
||||
@ -1136,12 +1131,12 @@ async def update_thread(
|
||||
async def upload_thread_attachment(
|
||||
thread_id: str,
|
||||
file: UploadFile = File(...),
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_required_user),
|
||||
):
|
||||
"""上传并解析附件为 Markdown，附加到指定对话线程。"""
|
||||
conv_manager = ConversationManager(db)
|
||||
conversation = _require_user_conversation(conv_manager, thread_id, str(current_user.id))
|
||||
conversation = await _require_user_conversation(conv_manager, thread_id, str(current_user.id))
|
||||
|
||||
try:
|
||||
conversion = await convert_upload_to_markdown(file)
|
||||
@ -1161,7 +1156,7 @@ async def upload_thread_attachment(
|
||||
"uploaded_at": utc_isoformat(),
|
||||
"truncated": conversion.truncated,
|
||||
}
|
||||
conv_manager.add_attachment(conversation.id, attachment_record)
|
||||
await conv_manager.add_attachment(conversation.id, attachment_record)
|
||||
|
||||
return _serialize_attachment(attachment_record)
|
||||
|
||||
@ -1169,13 +1164,13 @@ async def upload_thread_attachment(
|
||||
@chat.get("/thread/{thread_id}/attachments", response_model=AttachmentListResponse)
|
||||
async def list_thread_attachments(
|
||||
thread_id: str,
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_required_user),
|
||||
):
|
||||
"""列出当前对话线程的所有附件元信息。"""
|
||||
conv_manager = ConversationManager(db)
|
||||
conversation = _require_user_conversation(conv_manager, thread_id, str(current_user.id))
|
||||
attachments = conv_manager.get_attachments(conversation.id)
|
||||
conversation = await _require_user_conversation(conv_manager, thread_id, str(current_user.id))
|
||||
attachments = await conv_manager.get_attachments(conversation.id)
|
||||
return {
|
||||
"attachments": [_serialize_attachment(item) for item in attachments],
|
||||
"limits": {
|
||||
@ -1189,13 +1184,13 @@ async def list_thread_attachments(
|
||||
async def delete_thread_attachment(
|
||||
thread_id: str,
|
||||
file_id: str,
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_required_user),
|
||||
):
|
||||
"""移除指定附件。"""
|
||||
conv_manager = ConversationManager(db)
|
||||
conversation = _require_user_conversation(conv_manager, thread_id, str(current_user.id))
|
||||
removed = conv_manager.remove_attachment(conversation.id, file_id)
|
||||
conversation = await _require_user_conversation(conv_manager, thread_id, str(current_user.id))
|
||||
removed = await conv_manager.remove_attachment(conversation.id, file_id)
|
||||
if not removed:
|
||||
raise HTTPException(status_code=404, detail="附件不存在或已被删除")
|
||||
return {"message": "附件已删除"}
|
||||
@ -1223,7 +1218,7 @@ class MessageFeedbackResponse(BaseModel):
|
||||
async def submit_message_feedback(
|
||||
message_id: int,
|
||||
feedback_data: MessageFeedbackRequest,
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_required_user),
|
||||
):
|
||||
"""Submit user feedback for a specific message"""
|
||||
@ -1233,20 +1228,23 @@ async def submit_message_feedback(
|
||||
raise HTTPException(status_code=422, detail="Rating must be 'like' or 'dislike'")
|
||||
|
||||
# Verify message exists and get conversation to check permissions
|
||||
message = db.query(Message).filter_by(id=message_id).first()
|
||||
message_result = await db.execute(select(Message).filter_by(id=message_id))
|
||||
message = message_result.scalar_one_or_none()
|
||||
|
||||
if not message:
|
||||
raise HTTPException(status_code=404, detail="Message not found")
|
||||
|
||||
# Verify user has access to this message (through conversation)
|
||||
conversation = db.query(Conversation).filter_by(id=message.conversation_id).first()
|
||||
conversation_result = await db.execute(select(Conversation).filter_by(id=message.conversation_id))
|
||||
conversation = conversation_result.scalar_one_or_none()
|
||||
if not conversation or conversation.user_id != str(current_user.id):
|
||||
raise HTTPException(status_code=403, detail="Access denied")
|
||||
|
||||
# Check if feedback already exists (user can only submit once)
|
||||
existing_feedback = (
|
||||
db.query(MessageFeedback).filter_by(message_id=message_id, user_id=str(current_user.id)).first()
|
||||
existing_feedback_result = await db.execute(
|
||||
select(MessageFeedback).filter_by(message_id=message_id, user_id=str(current_user.id))
|
||||
)
|
||||
existing_feedback = existing_feedback_result.scalar_one_or_none()
|
||||
|
||||
if existing_feedback:
|
||||
raise HTTPException(status_code=409, detail="Feedback already submitted for this message")
|
||||
@ -1284,13 +1282,14 @@ async def submit_message_feedback(
|
||||
@chat.get("/message/{message_id}/feedback")
|
||||
async def get_message_feedback(
|
||||
message_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_required_user),
|
||||
):
|
||||
"""Get feedback status for a specific message (for current user)"""
|
||||
try:
|
||||
# Get user's feedback for this message
|
||||
feedback = db.query(MessageFeedback).filter_by(message_id=message_id, user_id=str(current_user.id)).first()
|
||||
feedback_result = await db.execute(select(MessageFeedback).filter_by(message_id=message_id, user_id=str(current_user.id)))
|
||||
feedback = feedback_result.scalar_one_or_none()
|
||||
|
||||
if not feedback:
|
||||
return {"has_feedback": False, "feedback": None}
|
||||
|
||||
@ -9,8 +9,8 @@ from datetime import datetime, timedelta
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy import String, cast, distinct, func, or_
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import String, cast, distinct, func, or_, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.routers.auth_router import get_admin_user
|
||||
from server.utils.auth_middleware import get_db
|
||||
@ -110,7 +110,7 @@ async def get_all_conversations(
|
||||
status: str = "active",
|
||||
limit: int = 100,
|
||||
offset: int = 0,
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
):
|
||||
"""Get all conversations (Admin only)"""
|
||||
@ -118,8 +118,9 @@ async def get_all_conversations(
|
||||
|
||||
try:
|
||||
# Build query
|
||||
query = db.query(Conversation, ConversationStats).outerjoin(
|
||||
ConversationStats, Conversation.id == ConversationStats.conversation_id
|
||||
query = (
|
||||
select(Conversation, ConversationStats)
|
||||
.outerjoin(ConversationStats, Conversation.id == ConversationStats.conversation_id)
|
||||
)
|
||||
|
||||
# Apply filters
|
||||
@ -133,7 +134,8 @@ async def get_all_conversations(
|
||||
# Order and paginate
|
||||
query = query.order_by(Conversation.updated_at.desc()).limit(limit).offset(offset)
|
||||
|
||||
results = query.all()
|
||||
result = await db.execute(query)
|
||||
results = result.all()
|
||||
|
||||
return [
|
||||
{
|
||||
@ -157,7 +159,7 @@ async def get_all_conversations(
|
||||
@dashboard.get("/conversations/{thread_id}", response_model=ConversationDetailResponse)
|
||||
async def get_conversation_detail(
|
||||
thread_id: str,
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
):
|
||||
"""Get conversation detail (Admin only)"""
|
||||
@ -225,7 +227,7 @@ async def get_conversation_detail(
|
||||
|
||||
@dashboard.get("/stats/users", response_model=UserActivityStats)
|
||||
async def get_user_activity_stats(
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
):
|
||||
"""Get user activity statistics (Admin only)"""
|
||||
@ -242,40 +244,38 @@ async def get_user_activity_stats(
|
||||
)
|
||||
|
||||
# 基础用户统计（排除已删除用户）
|
||||
total_users = db.query(func.count(User.id)).filter(User.is_deleted == 0).scalar() or 0
|
||||
total_users_result = await db.execute(select(func.count(User.id)).filter(User.is_deleted == 0))
|
||||
total_users = total_users_result.scalar() or 0
|
||||
|
||||
# 不同时间段的活跃用户数（基于对话活动，排除已删除用户）
|
||||
active_users_24h = (
|
||||
db.query(func.count(distinct(User.id)))
|
||||
active_users_24h_result = await db.execute(
|
||||
select(func.count(distinct(User.id)))
|
||||
.select_from(Conversation)
|
||||
.join(User, user_join_condition)
|
||||
.filter(Conversation.updated_at >= now - timedelta(days=1), User.is_deleted == 0)
|
||||
.scalar()
|
||||
or 0
|
||||
)
|
||||
active_users_24h = active_users_24h_result.scalar() or 0
|
||||
|
||||
active_users_30d = (
|
||||
db.query(func.count(distinct(User.id)))
|
||||
active_users_30d_result = await db.execute(
|
||||
select(func.count(distinct(User.id)))
|
||||
.select_from(Conversation)
|
||||
.join(User, user_join_condition)
|
||||
.filter(Conversation.updated_at >= now - timedelta(days=30), User.is_deleted == 0)
|
||||
.scalar()
|
||||
or 0
|
||||
)
|
||||
active_users_30d = active_users_30d_result.scalar() or 0
|
||||
# 最近7天每日活跃用户（排除已删除用户）
|
||||
daily_active_users = []
|
||||
for i in range(7):
|
||||
day_start = now - timedelta(days=i + 1)
|
||||
day_end = now - timedelta(days=i)
|
||||
|
||||
active_count = (
|
||||
db.query(func.count(distinct(User.id)))
|
||||
active_count_result = await db.execute(
|
||||
select(func.count(distinct(User.id)))
|
||||
.select_from(Conversation)
|
||||
.join(User, user_join_condition)
|
||||
.filter(Conversation.updated_at >= day_start, Conversation.updated_at < day_end, User.is_deleted == 0)
|
||||
.scalar()
|
||||
or 0
|
||||
)
|
||||
active_count = active_count_result.scalar() or 0
|
||||
|
||||
daily_active_users.append({"date": day_start.strftime("%Y-%m-%d"), "active_users": active_count})
|
||||
|
||||
@ -299,7 +299,7 @@ async def get_user_activity_stats(
|
||||
|
||||
@dashboard.get("/stats/tools", response_model=ToolCallStats)
|
||||
async def get_tool_call_stats(
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
):
|
||||
"""Get tool call statistics (Admin only)"""
|
||||
@ -309,28 +309,31 @@ async def get_tool_call_stats(
|
||||
now = utc_now()
|
||||
|
||||
# 基础工具调用统计
|
||||
total_calls = db.query(func.count(ToolCall.id)).scalar() or 0
|
||||
successful_calls = db.query(func.count(ToolCall.id)).filter(ToolCall.status == "success").scalar() or 0
|
||||
total_calls_result = await db.execute(select(func.count(ToolCall.id)))
|
||||
total_calls = total_calls_result.scalar() or 0
|
||||
|
||||
successful_calls_result = await db.execute(select(func.count(ToolCall.id)).filter(ToolCall.status == "success"))
|
||||
successful_calls = successful_calls_result.scalar() or 0
|
||||
failed_calls = total_calls - successful_calls
|
||||
success_rate = round((successful_calls / total_calls * 100), 2) if total_calls > 0 else 0
|
||||
|
||||
# 最常用工具
|
||||
most_used_tools = (
|
||||
db.query(ToolCall.tool_name, func.count(ToolCall.id).label("count"))
|
||||
most_used_tools_result = await db.execute(
|
||||
select(ToolCall.tool_name, func.count(ToolCall.id).label("count"))
|
||||
.group_by(ToolCall.tool_name)
|
||||
.order_by(func.count(ToolCall.id).desc())
|
||||
.limit(10)
|
||||
.all()
|
||||
)
|
||||
most_used_tools = most_used_tools_result.all()
|
||||
most_used_tools = [{"tool_name": name, "count": count} for name, count in most_used_tools]
|
||||
|
||||
# 工具错误分布
|
||||
tool_errors = (
|
||||
db.query(ToolCall.tool_name, func.count(ToolCall.id).label("error_count"))
|
||||
tool_errors_result = await db.execute(
|
||||
select(ToolCall.tool_name, func.count(ToolCall.id).label("error_count"))
|
||||
.filter(ToolCall.status == "error")
|
||||
.group_by(ToolCall.tool_name)
|
||||
.all()
|
||||
)
|
||||
tool_errors = tool_errors_result.all()
|
||||
tool_error_distribution = {name: count for name, count in tool_errors}
|
||||
|
||||
# 最近7天每日工具调用数
|
||||
@ -339,12 +342,11 @@ async def get_tool_call_stats(
|
||||
day_start = now - timedelta(days=i + 1)
|
||||
day_end = now - timedelta(days=i)
|
||||
|
||||
daily_count = (
|
||||
db.query(func.count(ToolCall.id))
|
||||
daily_count_result = await db.execute(
|
||||
select(func.count(ToolCall.id))
|
||||
.filter(ToolCall.created_at >= day_start, ToolCall.created_at < day_end)
|
||||
.scalar()
|
||||
or 0
|
||||
)
|
||||
daily_count = daily_count_result.scalar() or 0
|
||||
|
||||
daily_tool_calls.append({"date": day_start.strftime("%Y-%m-%d"), "call_count": daily_count})
|
||||
|
||||
@ -371,7 +373,7 @@ async def get_tool_call_stats(
|
||||
|
||||
@dashboard.get("/stats/knowledge", response_model=KnowledgeStats)
|
||||
async def get_knowledge_stats(
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
):
|
||||
"""Get knowledge base statistics (Admin only)"""
|
||||
@ -502,7 +504,7 @@ async def get_knowledge_stats(
|
||||
|
||||
@dashboard.get("/stats/agents", response_model=AgentAnalytics)
|
||||
async def get_agent_analytics(
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
):
|
||||
"""Get AI agent analytics (Admin only)"""
|
||||
@ -510,11 +512,11 @@ async def get_agent_analytics(
|
||||
from src.storage.db.models import Conversation, MessageFeedback, Message, ToolCall
|
||||
|
||||
# 获取所有智能体
|
||||
agents = (
|
||||
db.query(Conversation.agent_id, func.count(Conversation.id).label("conversation_count"))
|
||||
agents_result = await db.execute(
|
||||
select(Conversation.agent_id, func.count(Conversation.id).label("conversation_count"))
|
||||
.group_by(Conversation.agent_id)
|
||||
.all()
|
||||
)
|
||||
agents = agents_result.all()
|
||||
|
||||
total_agents = len(agents)
|
||||
agent_conversation_counts = [{"agent_id": agent_id, "conversation_count": count} for agent_id, count in agents]
|
||||
@ -522,23 +524,21 @@ async def get_agent_analytics(
|
||||
# 智能体满意度统计
|
||||
agent_satisfaction = []
|
||||
for agent_id, _ in agents:
|
||||
total_feedbacks = (
|
||||
db.query(func.count(MessageFeedback.id))
|
||||
total_feedbacks_result = await db.execute(
|
||||
select(func.count(MessageFeedback.id))
|
||||
.join(Message, MessageFeedback.message_id == Message.id)
|
||||
.join(Conversation, Message.conversation_id == Conversation.id)
|
||||
.filter(Conversation.agent_id == agent_id)
|
||||
.scalar()
|
||||
or 0
|
||||
)
|
||||
total_feedbacks = total_feedbacks_result.scalar() or 0
|
||||
|
||||
positive_feedbacks = (
|
||||
db.query(func.count(MessageFeedback.id))
|
||||
positive_feedbacks_result = await db.execute(
|
||||
select(func.count(MessageFeedback.id))
|
||||
.join(Message, MessageFeedback.message_id == Message.id)
|
||||
.join(Conversation, Message.conversation_id == Conversation.id)
|
||||
.filter(Conversation.agent_id == agent_id, MessageFeedback.rating == "like")
|
||||
.scalar()
|
||||
or 0
|
||||
)
|
||||
positive_feedbacks = positive_feedbacks_result.scalar() or 0
|
||||
|
||||
satisfaction_rate = round((positive_feedbacks / total_feedbacks * 100), 2) if total_feedbacks > 0 else 100
|
||||
|
||||
@ -549,14 +549,13 @@ async def get_agent_analytics(
|
||||
# 智能体工具使用统计
|
||||
agent_tool_usage = []
|
||||
for agent_id, _ in agents:
|
||||
tool_usage_count = (
|
||||
db.query(func.count(ToolCall.id))
|
||||
tool_usage_count_result = await db.execute(
|
||||
select(func.count(ToolCall.id))
|
||||
.join(Message, ToolCall.message_id == Message.id)
|
||||
.join(Conversation, Message.conversation_id == Conversation.id)
|
||||
.filter(Conversation.agent_id == agent_id)
|
||||
.scalar()
|
||||
or 0
|
||||
)
|
||||
tool_usage_count = tool_usage_count_result.scalar() or 0
|
||||
|
||||
agent_tool_usage.append({"agent_id": agent_id, "tool_usage_count": tool_usage_count})
|
||||
|
||||
@ -601,7 +600,7 @@ async def get_agent_analytics(
|
||||
|
||||
@dashboard.get("/stats")
|
||||
async def get_dashboard_stats(
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
):
|
||||
"""Get dashboard statistics (Admin only)"""
|
||||
@ -609,16 +608,32 @@ async def get_dashboard_stats(
|
||||
|
||||
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_conversations_result = await db.execute(select(func.count(Conversation.id)))
|
||||
total_conversations = total_conversations_result.scalar() or 0
|
||||
|
||||
active_conversations_result = await db.execute(
|
||||
select(func.count(Conversation.id)).filter(
|
||||
Conversation.status == "active"
|
||||
)
|
||||
)
|
||||
total_messages = db.query(func.count(Message.id)).scalar() or 0
|
||||
total_users = db.query(func.count(User.id)).filter(User.is_deleted == 0).scalar() or 0
|
||||
active_conversations = active_conversations_result.scalar() or 0
|
||||
|
||||
total_messages_result = await db.execute(select(func.count(Message.id)))
|
||||
total_messages = total_messages_result.scalar() or 0
|
||||
|
||||
total_users_result = await db.execute(select(func.count(User.id)).filter(User.is_deleted == 0))
|
||||
total_users = total_users_result.scalar() or 0
|
||||
|
||||
# Feedback statistics
|
||||
total_feedbacks = db.query(func.count(MessageFeedback.id)).scalar() or 0
|
||||
like_count = db.query(func.count(MessageFeedback.id)).filter(MessageFeedback.rating == "like").scalar() or 0
|
||||
total_feedbacks_result = await db.execute(select(func.count(MessageFeedback.id)))
|
||||
total_feedbacks = total_feedbacks_result.scalar() or 0
|
||||
|
||||
like_count_result = await db.execute(
|
||||
select(func.count(MessageFeedback.id)).filter(
|
||||
MessageFeedback.rating == "like"
|
||||
)
|
||||
)
|
||||
like_count = like_count_result.scalar() or 0
|
||||
|
||||
# Calculate satisfaction rate
|
||||
satisfaction_rate = round((like_count / total_feedbacks * 100), 2) if total_feedbacks > 0 else 100
|
||||
@ -663,7 +678,7 @@ class FeedbackListItem(BaseModel):
|
||||
async def get_all_feedbacks(
|
||||
rating: str | None = None,
|
||||
agent_id: str | None = None,
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
):
|
||||
"""Get all feedback records (Admin only)"""
|
||||
@ -673,10 +688,14 @@ async def get_all_feedbacks(
|
||||
# Build query with joins including User table
|
||||
# Try both User.id and User.user_id as MessageFeedback.user_id might be stored as either
|
||||
query = (
|
||||
db.query(MessageFeedback, Message, Conversation, User)
|
||||
select(MessageFeedback, Message, Conversation, User)
|
||||
.join(Message, MessageFeedback.message_id == Message.id)
|
||||
.join(Conversation, Message.conversation_id == Conversation.id)
|
||||
.outerjoin(User, (MessageFeedback.user_id == User.id) | (MessageFeedback.user_id == User.user_id))
|
||||
.outerjoin(
|
||||
User,
|
||||
(MessageFeedback.user_id == User.id)
|
||||
| (MessageFeedback.user_id == User.user_id),
|
||||
)
|
||||
)
|
||||
|
||||
# Apply filters
|
||||
@ -688,7 +707,8 @@ async def get_all_feedbacks(
|
||||
# Order by creation time (most recent first)
|
||||
query = query.order_by(MessageFeedback.created_at.desc())
|
||||
|
||||
results = query.all()
|
||||
results = await db.execute(query)
|
||||
results = results.all()
|
||||
|
||||
# Debug logging (privacy-safe)
|
||||
logger.info(f"Found {len(results)} feedback records")
|
||||
@ -736,7 +756,7 @@ class TimeSeriesStats(BaseModel):
|
||||
async def get_call_timeseries_stats(
|
||||
type: str = "models", # models/agents/tokens/tools
|
||||
time_range: str = "14days", # 14hours/14days/14weeks
|
||||
db: Session = Depends(get_db),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_admin_user),
|
||||
):
|
||||
"""Get time series statistics for call analytics (Admin only)"""
|
||||
@ -773,8 +793,8 @@ async def get_call_timeseries_stats(
|
||||
if type == "models":
|
||||
# 模型调用统计（基于消息数量，按模型分组）
|
||||
# 从message的extra_metadata中提取模型信息
|
||||
query = (
|
||||
db.query(
|
||||
query_result = await db.execute(
|
||||
select(
|
||||
group_format.label("date"),
|
||||
func.count(Message.id).label("count"),
|
||||
func.json_extract(Message.extra_metadata, "$.response_metadata.model_name").label("category"),
|
||||
@ -784,6 +804,7 @@ async def get_call_timeseries_stats(
|
||||
.group_by(group_format, func.json_extract(Message.extra_metadata, "$.response_metadata.model_name"))
|
||||
.order_by(group_format)
|
||||
)
|
||||
query = query_result.all()
|
||||
elif type == "agents":
|
||||
# 智能体调用统计（基于对话更新时间，按智能体分组）
|
||||
# 为对话创建独立的时间格式化器
|
||||
@ -794,8 +815,8 @@ async def get_call_timeseries_stats(
|
||||
else: # 14days
|
||||
conv_group_format = func.strftime("%Y-%m-%d", func.datetime(Conversation.updated_at, "+8 hours"))
|
||||
|
||||
query = (
|
||||
db.query(
|
||||
query_result = await db.execute(
|
||||
select(
|
||||
conv_group_format.label("date"),
|
||||
func.count(Conversation.id).label("count"),
|
||||
Conversation.agent_id.label("category"),
|
||||
@ -805,13 +826,14 @@ async def get_call_timeseries_stats(
|
||||
.group_by(conv_group_format, Conversation.agent_id)
|
||||
.order_by(conv_group_format)
|
||||
)
|
||||
query = query_result.all()
|
||||
elif type == "tokens":
|
||||
# Token消耗统计（区分input/output tokens）
|
||||
# 先查询input tokens
|
||||
from sqlalchemy import literal
|
||||
|
||||
input_query = (
|
||||
db.query(
|
||||
input_query_result = await db.execute(
|
||||
select(
|
||||
group_format.label("date"),
|
||||
func.sum(
|
||||
func.coalesce(func.json_extract(Message.extra_metadata, "$.usage_metadata.input_tokens"), 0)
|
||||
@ -826,10 +848,11 @@ async def get_call_timeseries_stats(
|
||||
.group_by(group_format)
|
||||
.order_by(group_format)
|
||||
)
|
||||
input_query = input_query_result.all()
|
||||
|
||||
# 查询output tokens
|
||||
output_query = (
|
||||
db.query(
|
||||
output_query_result = await db.execute(
|
||||
select(
|
||||
group_format.label("date"),
|
||||
func.sum(
|
||||
func.coalesce(func.json_extract(Message.extra_metadata, "$.usage_metadata.output_tokens"), 0)
|
||||
@ -844,10 +867,11 @@ async def get_call_timeseries_stats(
|
||||
.group_by(group_format)
|
||||
.order_by(group_format)
|
||||
)
|
||||
output_query = output_query_result.all()
|
||||
|
||||
# 合并两个查询结果
|
||||
input_results = input_query.all()
|
||||
output_results = output_query.all()
|
||||
input_results = input_query
|
||||
output_results = output_query
|
||||
results = input_results + output_results
|
||||
elif type == "tools":
|
||||
# 工具调用统计（按工具名称分组）
|
||||
@ -859,8 +883,8 @@ async def get_call_timeseries_stats(
|
||||
else: # 14days
|
||||
tool_group_format = func.strftime("%Y-%m-%d", func.datetime(ToolCall.created_at, "+8 hours"))
|
||||
|
||||
query = (
|
||||
db.query(
|
||||
query_result = await db.execute(
|
||||
select(
|
||||
tool_group_format.label("date"),
|
||||
func.count(ToolCall.id).label("count"),
|
||||
ToolCall.tool_name.label("category"),
|
||||
@ -869,11 +893,12 @@ async def get_call_timeseries_stats(
|
||||
.group_by(tool_group_format, ToolCall.tool_name)
|
||||
.order_by(tool_group_format)
|
||||
)
|
||||
query = query_result.all()
|
||||
else:
|
||||
raise HTTPException(status_code=422, detail=f"Invalid type: {type}")
|
||||
|
||||
if type != "tokens":
|
||||
results = query.all()
|
||||
results = query
|
||||
|
||||
# 处理堆叠数据格式
|
||||
# 首先收集所有类别
|
||||
@ -953,7 +978,8 @@ async def get_call_timeseries_stats(
|
||||
# 对于工具调用，显示所有时间的总数（与ToolStatsComponent保持一致）
|
||||
from src.storage.db.models import ToolCall
|
||||
|
||||
total_count = db.query(func.count(ToolCall.id)).scalar() or 0
|
||||
total_count_result = await db.execute(select(func.count(ToolCall.id)))
|
||||
total_count = total_count_result.scalar() or 0
|
||||
else:
|
||||
# 其他类型使用时间序列数据的总和
|
||||
total_count = sum(item["total"] for item in data)
|
||||
|
||||
@ -3,7 +3,7 @@ import re
|
||||
from fastapi import Depends, HTTPException, status
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from jose import JWTError
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.storage.db.manager import db_manager
|
||||
from src.storage.db.models import User
|
||||
@ -23,17 +23,14 @@ PUBLIC_PATHS = [
|
||||
]
|
||||
|
||||
|
||||
# 获取数据库会话
|
||||
def get_db():
|
||||
db = db_manager.get_session()
|
||||
try:
|
||||
# 获取数据库会话（异步版本）
|
||||
async def get_db():
|
||||
async with db_manager.get_async_session_context() as db:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
# 获取当前用户
|
||||
async def get_current_user(token: str | None = Depends(oauth2_scheme), db: Session = Depends(get_db)):
|
||||
# 获取当前用户（异步版本）
|
||||
async def get_current_user(token: str | None = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)):
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="无效的凭证",
|
||||
@ -61,8 +58,10 @@ async def get_current_user(token: str | None = Depends(oauth2_scheme), db: Sessi
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
# 查找用户
|
||||
user = db.query(User).filter(User.id == user_id).first()
|
||||
# 查找用户（异步版本）
|
||||
from sqlalchemy import select
|
||||
result = await db.execute(select(User).filter(User.id == user_id))
|
||||
user = result.scalar_one_or_none()
|
||||
if user is None:
|
||||
raise credentials_exception
|
||||
|
||||
@ -79,7 +78,6 @@ async def get_required_user(user: User | None = Depends(get_current_user)):
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
# 获取管理员用户
|
||||
async def get_admin_user(current_user: User = Depends(get_required_user)):
|
||||
if current_user.role not in ["admin", "superadmin"]:
|
||||
@ -89,7 +87,6 @@ async def get_admin_user(current_user: User = Depends(get_required_user)):
|
||||
)
|
||||
return current_user
|
||||
|
||||
|
||||
# 获取超级管理员用户
|
||||
async def get_superadmin_user(current_user: User = Depends(get_required_user)):
|
||||
if current_user.role != "superadmin":
|
||||
|
||||
@ -1 +0,0 @@
|
||||
Subproject commit 057e2000be7b56823239815b0fe7c7fc0dbced96
|
||||
@ -1 +0,0 @@
|
||||
Subproject commit 6a0367834ea0fb5e5c94b9711e3e2756966789ea
|
||||
@ -1,27 +1,28 @@
|
||||
"""
|
||||
Conversation Storage Manager
|
||||
Conversation Storage Manager (Async)
|
||||
|
||||
Manages conversation data storage including messages, tool calls, and statistics.
|
||||
All database operations are now asynchronous for improved performance.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
|
||||
from sqlalchemy.orm import Session, joinedload
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from src.storage.db.models import Conversation, ConversationStats, Message, ToolCall
|
||||
from src.utils import logger
|
||||
from src.utils.datetime_utils import utc_now
|
||||
|
||||
# TODO:[未完成]待修改为异步版本
|
||||
|
||||
|
||||
class ConversationManager:
|
||||
"""Manager for conversation storage operations"""
|
||||
"""Async Manager for conversation storage operations"""
|
||||
|
||||
def __init__(self, db_session: Session):
|
||||
def __init__(self, db_session: AsyncSession):
|
||||
self.db = db_session
|
||||
|
||||
def create_conversation(
|
||||
async def create_conversation(
|
||||
self,
|
||||
user_id: str,
|
||||
agent_id: str,
|
||||
@ -59,18 +60,18 @@ class ConversationManager:
|
||||
|
||||
self.db.add(conversation)
|
||||
# Flush to assign primary key without committing
|
||||
self.db.flush()
|
||||
await 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)
|
||||
await self.db.commit()
|
||||
await 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:
|
||||
async def get_conversation_by_thread_id(self, thread_id: str) -> Conversation | None:
|
||||
"""
|
||||
Get conversation by thread ID
|
||||
|
||||
@ -80,10 +81,12 @@ class ConversationManager:
|
||||
Returns:
|
||||
Conversation object or None if not found
|
||||
"""
|
||||
return self.db.query(Conversation).filter(Conversation.thread_id == thread_id).first()
|
||||
result = await self.db.execute(select(Conversation).filter(Conversation.thread_id == thread_id))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
def _get_conversation_by_id(self, conversation_id: int) -> Conversation | None:
|
||||
return self.db.query(Conversation).filter(Conversation.id == conversation_id).first()
|
||||
async def _get_conversation_by_id(self, conversation_id: int) -> Conversation | None:
|
||||
result = await self.db.execute(select(Conversation).filter(Conversation.id == conversation_id))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
def _ensure_metadata(self, conversation: Conversation) -> dict:
|
||||
"""
|
||||
@ -96,13 +99,13 @@ class ConversationManager:
|
||||
metadata["attachments"] = list(metadata.get("attachments", []))
|
||||
return metadata
|
||||
|
||||
def _save_metadata(self, conversation: Conversation, metadata: dict) -> None:
|
||||
async def _save_metadata(self, conversation: Conversation, metadata: dict) -> None:
|
||||
conversation.extra_metadata = metadata
|
||||
conversation.updated_at = utc_now()
|
||||
self.db.commit()
|
||||
self.db.refresh(conversation)
|
||||
await self.db.commit()
|
||||
await self.db.refresh(conversation)
|
||||
|
||||
def add_message(
|
||||
async def add_message(
|
||||
self,
|
||||
conversation_id: int,
|
||||
role: str,
|
||||
@ -136,20 +139,20 @@ class ConversationManager:
|
||||
|
||||
self.db.add(message)
|
||||
# Mark the parent conversation as active for sorting/analytics
|
||||
conversation = self._get_conversation_by_id(conversation_id)
|
||||
conversation = await self._get_conversation_by_id(conversation_id)
|
||||
if conversation:
|
||||
conversation.updated_at = utc_now()
|
||||
|
||||
self.db.commit()
|
||||
self.db.refresh(message)
|
||||
await self.db.commit()
|
||||
await self.db.refresh(message)
|
||||
|
||||
# Update conversation stats
|
||||
self._update_message_count(conversation_id)
|
||||
await self._update_message_count(conversation_id)
|
||||
|
||||
logger.debug(f"Added {role} message to conversation {conversation_id}")
|
||||
return message
|
||||
|
||||
def add_message_by_thread_id(
|
||||
async def add_message_by_thread_id(
|
||||
self,
|
||||
thread_id: str,
|
||||
role: str,
|
||||
@ -172,12 +175,12 @@ class ConversationManager:
|
||||
Returns:
|
||||
Created Message object or None if conversation not found
|
||||
"""
|
||||
conversation = self.get_conversation_by_thread_id(thread_id)
|
||||
conversation = await self.get_conversation_by_thread_id(thread_id)
|
||||
if not conversation:
|
||||
logger.warning(f"Conversation not found for thread_id: {thread_id}")
|
||||
return None
|
||||
|
||||
return self.add_message(
|
||||
return await self.add_message(
|
||||
conversation_id=conversation.id,
|
||||
role=role,
|
||||
content=content,
|
||||
@ -186,7 +189,7 @@ class ConversationManager:
|
||||
image_content=image_content,
|
||||
)
|
||||
|
||||
def add_tool_call(
|
||||
async def add_tool_call(
|
||||
self,
|
||||
message_id: int,
|
||||
tool_name: str,
|
||||
@ -222,13 +225,13 @@ class ConversationManager:
|
||||
)
|
||||
|
||||
self.db.add(tool_call)
|
||||
self.db.commit()
|
||||
self.db.refresh(tool_call)
|
||||
await self.db.commit()
|
||||
await 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]:
|
||||
async def get_messages(self, conversation_id: int, limit: int | None = None, offset: int = 0) -> list[Message]:
|
||||
"""
|
||||
Get messages for a conversation
|
||||
|
||||
@ -241,18 +244,21 @@ class ConversationManager:
|
||||
List of Message objects
|
||||
"""
|
||||
query = (
|
||||
self.db.query(Message)
|
||||
select(Message)
|
||||
.options(selectinload(Message.tool_calls))
|
||||
.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()
|
||||
result = await self.db.execute(query)
|
||||
return result.scalars().unique().all()
|
||||
|
||||
def get_messages_by_thread_id(self, thread_id: str, limit: int | None = None, offset: int = 0) -> list[Message]:
|
||||
async 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
|
||||
|
||||
@ -264,14 +270,14 @@ class ConversationManager:
|
||||
Returns:
|
||||
List of Message objects
|
||||
"""
|
||||
conversation = self.get_conversation_by_thread_id(thread_id)
|
||||
conversation = await self.get_conversation_by_thread_id(thread_id)
|
||||
if not conversation:
|
||||
logger.warning(f"Conversation not found for thread_id: {thread_id}")
|
||||
return []
|
||||
|
||||
return self.get_messages(conversation.id, limit, offset)
|
||||
return await self.get_messages(conversation.id, limit, offset)
|
||||
|
||||
def list_conversations(
|
||||
async def list_conversations(
|
||||
self, user_id: str | None = None, agent_id: str | None = None, status: str = "active"
|
||||
) -> list[Conversation]:
|
||||
"""
|
||||
@ -285,7 +291,7 @@ class ConversationManager:
|
||||
Returns:
|
||||
List of Conversation objects
|
||||
"""
|
||||
query = self.db.query(Conversation).filter(Conversation.status == status)
|
||||
query = select(Conversation).filter(Conversation.status == status)
|
||||
|
||||
# Only filter by user_id if it's provided and not empty
|
||||
if user_id:
|
||||
@ -294,9 +300,11 @@ class ConversationManager:
|
||||
if agent_id:
|
||||
query = query.filter(Conversation.agent_id == agent_id)
|
||||
|
||||
return query.order_by(Conversation.updated_at.desc()).all()
|
||||
query = query.order_by(Conversation.updated_at.desc())
|
||||
result = await self.db.execute(query)
|
||||
return result.scalars().all()
|
||||
|
||||
def update_conversation(
|
||||
async def update_conversation(
|
||||
self,
|
||||
thread_id: str,
|
||||
title: str | None = None,
|
||||
@ -315,7 +323,7 @@ class ConversationManager:
|
||||
Returns:
|
||||
Updated Conversation object or None if not found
|
||||
"""
|
||||
conversation = self.get_conversation_by_thread_id(thread_id)
|
||||
conversation = await self.get_conversation_by_thread_id(thread_id)
|
||||
if not conversation:
|
||||
return None
|
||||
|
||||
@ -331,13 +339,13 @@ class ConversationManager:
|
||||
conversation.extra_metadata = current_metadata
|
||||
|
||||
conversation.updated_at = utc_now()
|
||||
self.db.commit()
|
||||
self.db.refresh(conversation)
|
||||
await self.db.commit()
|
||||
await 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:
|
||||
async def delete_conversation(self, thread_id: str, soft_delete: bool = True) -> bool:
|
||||
"""
|
||||
Delete a conversation
|
||||
|
||||
@ -348,22 +356,22 @@ class ConversationManager:
|
||||
Returns:
|
||||
True if successful, False otherwise
|
||||
"""
|
||||
conversation = self.get_conversation_by_thread_id(thread_id)
|
||||
conversation = await self.get_conversation_by_thread_id(thread_id)
|
||||
if not conversation:
|
||||
return False
|
||||
|
||||
if soft_delete:
|
||||
conversation.status = "deleted"
|
||||
self.db.commit()
|
||||
await self.db.commit()
|
||||
logger.info(f"Soft deleted conversation {thread_id}")
|
||||
else:
|
||||
self.db.delete(conversation)
|
||||
self.db.commit()
|
||||
await self.db.commit()
|
||||
logger.info(f"Permanently deleted conversation {thread_id}")
|
||||
|
||||
return True
|
||||
|
||||
def get_stats(self, conversation_id: int) -> ConversationStats | None:
|
||||
async def get_stats(self, conversation_id: int) -> ConversationStats | None:
|
||||
"""
|
||||
Get conversation statistics
|
||||
|
||||
@ -373,9 +381,12 @@ class ConversationManager:
|
||||
Returns:
|
||||
ConversationStats object or None if not found
|
||||
"""
|
||||
return self.db.query(ConversationStats).filter(ConversationStats.conversation_id == conversation_id).first()
|
||||
result = await self.db.execute(
|
||||
select(ConversationStats).filter(ConversationStats.conversation_id == conversation_id)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
def update_stats(
|
||||
async def update_stats(
|
||||
self,
|
||||
conversation_id: int,
|
||||
tokens_used: int | None = None,
|
||||
@ -394,7 +405,7 @@ class ConversationManager:
|
||||
Returns:
|
||||
Updated ConversationStats object or None if not found
|
||||
"""
|
||||
stats = self.get_stats(conversation_id)
|
||||
stats = await self.get_stats(conversation_id)
|
||||
if not stats:
|
||||
return None
|
||||
|
||||
@ -406,12 +417,12 @@ class ConversationManager:
|
||||
stats.user_feedback = user_feedback
|
||||
|
||||
stats.updated_at = utc_now()
|
||||
self.db.commit()
|
||||
self.db.refresh(stats)
|
||||
await self.db.commit()
|
||||
await self.db.refresh(stats)
|
||||
|
||||
return stats
|
||||
|
||||
def get_tool_call_by_langgraph_id(self, langgraph_tool_call_id: str) -> ToolCall | None:
|
||||
async def get_tool_call_by_langgraph_id(self, langgraph_tool_call_id: str) -> ToolCall | None:
|
||||
"""
|
||||
Get tool call by LangGraph tool_call_id
|
||||
|
||||
@ -421,9 +432,12 @@ class ConversationManager:
|
||||
Returns:
|
||||
ToolCall object or None if not found
|
||||
"""
|
||||
return self.db.query(ToolCall).filter(ToolCall.langgraph_tool_call_id == langgraph_tool_call_id).first()
|
||||
result = await self.db.execute(
|
||||
select(ToolCall).filter(ToolCall.langgraph_tool_call_id == langgraph_tool_call_id)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
def update_tool_call_output(
|
||||
async def update_tool_call_output(
|
||||
self,
|
||||
langgraph_tool_call_id: str,
|
||||
tool_output: str,
|
||||
@ -442,7 +456,7 @@ class ConversationManager:
|
||||
Returns:
|
||||
Updated ToolCall object or None if not found
|
||||
"""
|
||||
tool_call = self.get_tool_call_by_langgraph_id(langgraph_tool_call_id)
|
||||
tool_call = await self.get_tool_call_by_langgraph_id(langgraph_tool_call_id)
|
||||
if not tool_call:
|
||||
logger.warning(f"Tool call not found for langgraph_tool_call_id: {langgraph_tool_call_id}")
|
||||
return None
|
||||
@ -452,44 +466,47 @@ class ConversationManager:
|
||||
if error_message:
|
||||
tool_call.error_message = error_message
|
||||
|
||||
self.db.commit()
|
||||
self.db.refresh(tool_call)
|
||||
await self.db.commit()
|
||||
await self.db.refresh(tool_call)
|
||||
|
||||
logger.debug(f"Updated tool call {langgraph_tool_call_id} with output")
|
||||
return tool_call
|
||||
|
||||
def _update_message_count(self, conversation_id: int) -> None:
|
||||
async 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)
|
||||
from sqlalchemy import func
|
||||
|
||||
stats = await self.get_stats(conversation_id)
|
||||
if stats:
|
||||
message_count = self.db.query(Message).filter(Message.conversation_id == conversation_id).count()
|
||||
result = await self.db.execute(select(func.count()).filter(Message.conversation_id == conversation_id))
|
||||
message_count = result.scalar()
|
||||
stats.message_count = message_count
|
||||
self.db.commit()
|
||||
await self.db.commit()
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Attachment helpers
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def get_attachments(self, conversation_id: int) -> list[dict]:
|
||||
conversation = self._get_conversation_by_id(conversation_id)
|
||||
async def get_attachments(self, conversation_id: int) -> list[dict]:
|
||||
conversation = await self._get_conversation_by_id(conversation_id)
|
||||
if not conversation:
|
||||
return []
|
||||
metadata = self._ensure_metadata(conversation)
|
||||
return list(metadata.get("attachments", []))
|
||||
|
||||
def get_attachments_by_thread_id(self, thread_id: str) -> list[dict]:
|
||||
conversation = self.get_conversation_by_thread_id(thread_id)
|
||||
async def get_attachments_by_thread_id(self, thread_id: str) -> list[dict]:
|
||||
conversation = await self.get_conversation_by_thread_id(thread_id)
|
||||
if not conversation:
|
||||
return []
|
||||
return self.get_attachments(conversation.id)
|
||||
return await self.get_attachments(conversation.id)
|
||||
|
||||
def add_attachment(self, conversation_id: int, attachment_info: dict) -> dict | None:
|
||||
conversation = self._get_conversation_by_id(conversation_id)
|
||||
async def add_attachment(self, conversation_id: int, attachment_info: dict) -> dict | None:
|
||||
conversation = await self._get_conversation_by_id(conversation_id)
|
||||
if not conversation:
|
||||
return None
|
||||
|
||||
@ -498,13 +515,13 @@ class ConversationManager:
|
||||
attachments = [item for item in attachments if item.get("file_id") != attachment_info.get("file_id")]
|
||||
attachments.append(attachment_info)
|
||||
metadata["attachments"] = attachments
|
||||
self._save_metadata(conversation, metadata)
|
||||
await self._save_metadata(conversation, metadata)
|
||||
return attachment_info
|
||||
|
||||
def update_attachment_status(
|
||||
async def update_attachment_status(
|
||||
self, conversation_id: int, file_id: str, status: str, update_fields: dict | None = None
|
||||
) -> dict | None:
|
||||
conversation = self._get_conversation_by_id(conversation_id)
|
||||
conversation = await self._get_conversation_by_id(conversation_id)
|
||||
if not conversation:
|
||||
return None
|
||||
|
||||
@ -521,11 +538,11 @@ class ConversationManager:
|
||||
|
||||
if target is not None:
|
||||
metadata["attachments"] = attachments
|
||||
self._save_metadata(conversation, metadata)
|
||||
await self._save_metadata(conversation, metadata)
|
||||
return target
|
||||
|
||||
def remove_attachment(self, conversation_id: int, file_id: str) -> bool:
|
||||
conversation = self._get_conversation_by_id(conversation_id)
|
||||
async def remove_attachment(self, conversation_id: int, file_id: str) -> bool:
|
||||
conversation = await self._get_conversation_by_id(conversation_id)
|
||||
if not conversation:
|
||||
return False
|
||||
|
||||
@ -537,5 +554,5 @@ class ConversationManager:
|
||||
return False
|
||||
|
||||
metadata["attachments"] = new_attachments
|
||||
self._save_metadata(conversation, metadata)
|
||||
await self._save_metadata(conversation, metadata)
|
||||
return True
|
||||
|
||||
@ -1,9 +1,10 @@
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
from contextlib import contextmanager
|
||||
from contextlib import asynccontextmanager, contextmanager
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from server.utils.singleton import SingletonMeta
|
||||
@ -20,26 +21,34 @@ except ImportError:
|
||||
def validate_database_schema(db_path):
|
||||
return True, []
|
||||
|
||||
# TODO:[优化建议]需要将数据库修改为异步的aiosqlite或者异步mysql，缓存使用Redis存储
|
||||
# TODO:[已完成]为DBManager添加异步支持
|
||||
# TODO:[已完成]为DBManager添加单例模式
|
||||
|
||||
|
||||
class DBManager(metaclass=SingletonMeta):
|
||||
"""数据库管理器 - 只提供基础的数据库连接和会话管理"""
|
||||
"""数据库管理器 - 提供异步数据库连接和会话管理"""
|
||||
|
||||
def __init__(self):
|
||||
self.db_path = os.path.join(config.save_dir, "database", "server.db")
|
||||
self.ensure_db_dir()
|
||||
|
||||
# 创建SQLAlchemy引擎，配置JSON序列化器以支持中文
|
||||
# 创建异步SQLAlchemy引擎，配置JSON序列化器以支持中文
|
||||
# 使用 ensure_ascii=False 确保中文字符不被转义为 Unicode 序列
|
||||
self.async_engine = create_async_engine(
|
||||
f"sqlite+aiosqlite:///{self.db_path}",
|
||||
json_serializer=lambda obj: json.dumps(obj, ensure_ascii=False),
|
||||
json_deserializer=json.loads,
|
||||
)
|
||||
|
||||
# 创建异步会话工厂
|
||||
self.AsyncSession = async_sessionmaker(bind=self.async_engine, class_=AsyncSession, expire_on_commit=False)
|
||||
|
||||
# 保留同步引擎用于迁移等特殊操作
|
||||
self.engine = create_engine(
|
||||
f"sqlite:///{self.db_path}",
|
||||
json_serializer=lambda obj: json.dumps(obj, ensure_ascii=False),
|
||||
json_deserializer=json.loads,
|
||||
)
|
||||
|
||||
# 创建会话工厂
|
||||
self.Session = sessionmaker(bind=self.engine)
|
||||
|
||||
# 首先创建基本表结构
|
||||
@ -86,12 +95,12 @@ class DBManager(metaclass=SingletonMeta):
|
||||
logger.warning("=" * 60)
|
||||
|
||||
def get_session(self):
|
||||
"""获取数据库会话"""
|
||||
"""获取同步数据库会话"""
|
||||
return self.Session()
|
||||
|
||||
@contextmanager
|
||||
def get_session_context(self):
|
||||
"""获取数据库会话的上下文管理器"""
|
||||
"""获取同步数据库会话的上下文管理器"""
|
||||
session = self.Session()
|
||||
try:
|
||||
yield session
|
||||
@ -103,6 +112,24 @@ class DBManager(metaclass=SingletonMeta):
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
async def get_async_session(self):
|
||||
"""获取异步数据库会话"""
|
||||
return self.AsyncSession()
|
||||
|
||||
@asynccontextmanager
|
||||
async def get_async_session_context(self):
|
||||
"""获取异步数据库会话的上下文管理器"""
|
||||
session = self.AsyncSession()
|
||||
try:
|
||||
yield session
|
||||
await session.commit()
|
||||
except Exception as e:
|
||||
await session.rollback()
|
||||
logger.error(f"Async database operation failed: {e}")
|
||||
raise
|
||||
finally:
|
||||
await session.close()
|
||||
|
||||
def check_first_run(self):
|
||||
"""检查是否首次运行"""
|
||||
session = self.get_session()
|
||||
|
||||
@ -983,8 +983,6 @@ const selectChat = async (chatId) => {
|
||||
chatUIStore.isLoadingMessages = true;
|
||||
try {
|
||||
await fetchThreadMessages({ agentId: currentAgentId.value, threadId: chatId });
|
||||
await loadThreadAttachments(chatId, { silent: true });
|
||||
await fetchAgentState(currentAgentId.value, chatId);
|
||||
} catch (error) {
|
||||
handleChatError(error, 'load');
|
||||
} finally {
|
||||
@ -993,6 +991,8 @@ const selectChat = async (chatId) => {
|
||||
|
||||
await nextTick();
|
||||
scrollController.scrollToBottomStaticForce();
|
||||
await loadThreadAttachments(chatId, { silent: true });
|
||||
await fetchAgentState(currentAgentId.value, chatId);
|
||||
};
|
||||
|
||||
const deleteChat = async (chatId) => {
|
||||
|
||||
@ -188,7 +188,7 @@ const renameChat = async (chatId) => {
|
||||
content: h('div', { style: { marginTop: '12px' } }, [
|
||||
h('input', {
|
||||
value: newTitle,
|
||||
style: { width: '100%', padding: '4px 8px', border: '1px solid #d9d9d9', borderRadius: '4px' },
|
||||
style: { width: '100%', padding: '4px 8px', border: '1px solid var(--gray-150)', background: 'var(--gray-0)', borderRadius: '4px' },
|
||||
onInput: (e) => { newTitle = e.target.value; }
|
||||
})
|
||||
]),
|
||||
|
||||
Loading…
Reference in New Issue
Block a user