Merge branch 'hotfix/v0.2.5-hotfix-1'
This commit is contained in:
7
api/app/cache/__init__.py
vendored
7
api/app/cache/__init__.py
vendored
@@ -2,10 +2,7 @@
|
||||
Cache 缓存模块
|
||||
|
||||
提供各种缓存功能的统一入口
|
||||
注意:隐性记忆和情绪建议已迁移到数据库存储,不再使用Redis缓存
|
||||
"""
|
||||
from .memory import EmotionMemoryCache, ImplicitMemoryCache
|
||||
|
||||
__all__ = [
|
||||
"EmotionMemoryCache",
|
||||
"ImplicitMemoryCache",
|
||||
]
|
||||
__all__ = []
|
||||
|
||||
8
api/app/cache/memory/__init__.py
vendored
8
api/app/cache/memory/__init__.py
vendored
@@ -2,11 +2,7 @@
|
||||
Memory 缓存模块
|
||||
|
||||
提供记忆系统相关的缓存功能
|
||||
注意:隐性记忆和情绪建议已迁移到数据库存储,不再使用Redis缓存
|
||||
"""
|
||||
from .emotion_memory import EmotionMemoryCache
|
||||
from .implicit_memory import ImplicitMemoryCache
|
||||
|
||||
__all__ = [
|
||||
"EmotionMemoryCache",
|
||||
"ImplicitMemoryCache",
|
||||
]
|
||||
__all__ = []
|
||||
|
||||
134
api/app/cache/memory/emotion_memory.py
vendored
134
api/app/cache/memory/emotion_memory.py
vendored
@@ -1,134 +0,0 @@
|
||||
"""
|
||||
Emotion Suggestions Cache
|
||||
|
||||
情绪个性化建议缓存模块
|
||||
用于缓存用户的情绪个性化建议数据
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
from typing import Optional, Dict, Any
|
||||
from datetime import datetime
|
||||
|
||||
from app.aioRedis import aio_redis
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class EmotionMemoryCache:
|
||||
"""情绪建议缓存类"""
|
||||
|
||||
# Key 前缀
|
||||
PREFIX = "cache:memory:emotion_memory"
|
||||
|
||||
@classmethod
|
||||
def _get_key(cls, *parts: str) -> str:
|
||||
"""生成 Redis key
|
||||
|
||||
Args:
|
||||
*parts: key 的各个部分
|
||||
|
||||
Returns:
|
||||
完整的 Redis key
|
||||
"""
|
||||
return ":".join([cls.PREFIX] + list(parts))
|
||||
|
||||
@classmethod
|
||||
async def set_emotion_suggestions(
|
||||
cls,
|
||||
user_id: str,
|
||||
suggestions_data: Dict[str, Any],
|
||||
expire: int = 86400
|
||||
) -> bool:
|
||||
"""设置用户情绪建议缓存
|
||||
|
||||
Args:
|
||||
user_id: 用户ID(end_user_id)
|
||||
suggestions_data: 建议数据字典,包含:
|
||||
- health_summary: 健康状态摘要
|
||||
- suggestions: 建议列表
|
||||
- generated_at: 生成时间(可选)
|
||||
expire: 过期时间(秒),默认24小时(86400秒)
|
||||
|
||||
Returns:
|
||||
是否设置成功
|
||||
"""
|
||||
try:
|
||||
key = cls._get_key("suggestions", user_id)
|
||||
|
||||
# 添加生成时间戳
|
||||
if "generated_at" not in suggestions_data:
|
||||
suggestions_data["generated_at"] = datetime.now().isoformat()
|
||||
|
||||
# 添加缓存标记
|
||||
suggestions_data["cached"] = True
|
||||
|
||||
value = json.dumps(suggestions_data, ensure_ascii=False)
|
||||
await aio_redis.set(key, value, ex=expire)
|
||||
logger.info(f"设置情绪建议缓存成功: {key}, 过期时间: {expire}秒")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"设置情绪建议缓存失败: {e}", exc_info=True)
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
async def get_emotion_suggestions(cls, user_id: str) -> Optional[Dict[str, Any]]:
|
||||
"""获取用户情绪建议缓存
|
||||
|
||||
Args:
|
||||
user_id: 用户ID(end_user_id)
|
||||
|
||||
Returns:
|
||||
建议数据字典,如果不存在或已过期返回 None
|
||||
"""
|
||||
try:
|
||||
key = cls._get_key("suggestions", user_id)
|
||||
value = await aio_redis.get(key)
|
||||
|
||||
if value:
|
||||
data = json.loads(value)
|
||||
logger.info(f"成功获取情绪建议缓存: {key}")
|
||||
return data
|
||||
|
||||
logger.info(f"情绪建议缓存不存在或已过期: {key}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"获取情绪建议缓存失败: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
async def delete_emotion_suggestions(cls, user_id: str) -> bool:
|
||||
"""删除用户情绪建议缓存
|
||||
|
||||
Args:
|
||||
user_id: 用户ID(end_user_id)
|
||||
|
||||
Returns:
|
||||
是否删除成功
|
||||
"""
|
||||
try:
|
||||
key = cls._get_key("suggestions", user_id)
|
||||
result = await aio_redis.delete(key)
|
||||
logger.info(f"删除情绪建议缓存: {key}, 结果: {result}")
|
||||
return result > 0
|
||||
except Exception as e:
|
||||
logger.error(f"删除情绪建议缓存失败: {e}", exc_info=True)
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
async def get_suggestions_ttl(cls, user_id: str) -> int:
|
||||
"""获取情绪建议缓存的剩余过期时间
|
||||
|
||||
Args:
|
||||
user_id: 用户ID(end_user_id)
|
||||
|
||||
Returns:
|
||||
剩余秒数,-1表示永不过期,-2表示key不存在
|
||||
"""
|
||||
try:
|
||||
key = cls._get_key("suggestions", user_id)
|
||||
ttl = await aio_redis.ttl(key)
|
||||
logger.debug(f"情绪建议缓存TTL: {key} = {ttl}秒")
|
||||
return ttl
|
||||
except Exception as e:
|
||||
logger.error(f"获取情绪建议缓存TTL失败: {e}")
|
||||
return -2
|
||||
136
api/app/cache/memory/implicit_memory.py
vendored
136
api/app/cache/memory/implicit_memory.py
vendored
@@ -1,136 +0,0 @@
|
||||
"""
|
||||
Implicit Memory Profile Cache
|
||||
|
||||
隐式记忆用户画像缓存模块
|
||||
用于缓存用户的完整画像数据(偏好标签、四维画像、兴趣领域、行为习惯)
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
from typing import Optional, Dict, Any
|
||||
from datetime import datetime
|
||||
|
||||
from app.aioRedis import aio_redis
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ImplicitMemoryCache:
|
||||
"""隐式记忆用户画像缓存类"""
|
||||
|
||||
# Key 前缀
|
||||
PREFIX = "cache:memory:implicit_memory"
|
||||
|
||||
@classmethod
|
||||
def _get_key(cls, *parts: str) -> str:
|
||||
"""生成 Redis key
|
||||
|
||||
Args:
|
||||
*parts: key 的各个部分
|
||||
|
||||
Returns:
|
||||
完整的 Redis key
|
||||
"""
|
||||
return ":".join([cls.PREFIX] + list(parts))
|
||||
|
||||
@classmethod
|
||||
async def set_user_profile(
|
||||
cls,
|
||||
user_id: str,
|
||||
profile_data: Dict[str, Any],
|
||||
expire: int = 86400
|
||||
) -> bool:
|
||||
"""设置用户完整画像缓存
|
||||
|
||||
Args:
|
||||
user_id: 用户ID(end_user_id)
|
||||
profile_data: 画像数据字典,包含:
|
||||
- preferences: 偏好标签列表
|
||||
- portrait: 四维画像对象
|
||||
- interest_areas: 兴趣领域分布对象
|
||||
- habits: 行为习惯列表
|
||||
- generated_at: 生成时间(可选)
|
||||
expire: 过期时间(秒),默认24小时(86400秒)
|
||||
|
||||
Returns:
|
||||
是否设置成功
|
||||
"""
|
||||
try:
|
||||
key = cls._get_key("profile", user_id)
|
||||
|
||||
# 添加生成时间戳
|
||||
if "generated_at" not in profile_data:
|
||||
profile_data["generated_at"] = datetime.now().isoformat()
|
||||
|
||||
# 添加缓存标记
|
||||
profile_data["cached"] = True
|
||||
|
||||
value = json.dumps(profile_data, ensure_ascii=False)
|
||||
await aio_redis.set(key, value, ex=expire)
|
||||
logger.info(f"设置用户画像缓存成功: {key}, 过期时间: {expire}秒")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"设置用户画像缓存失败: {e}", exc_info=True)
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
async def get_user_profile(cls, user_id: str) -> Optional[Dict[str, Any]]:
|
||||
"""获取用户完整画像缓存
|
||||
|
||||
Args:
|
||||
user_id: 用户ID(end_user_id)
|
||||
|
||||
Returns:
|
||||
画像数据字典,如果不存在或已过期返回 None
|
||||
"""
|
||||
try:
|
||||
key = cls._get_key("profile", user_id)
|
||||
value = await aio_redis.get(key)
|
||||
|
||||
if value:
|
||||
data = json.loads(value)
|
||||
logger.info(f"成功获取用户画像缓存: {key}")
|
||||
return data
|
||||
|
||||
logger.info(f"用户画像缓存不存在或已过期: {key}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"获取用户画像缓存失败: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
async def delete_user_profile(cls, user_id: str) -> bool:
|
||||
"""删除用户完整画像缓存
|
||||
|
||||
Args:
|
||||
user_id: 用户ID(end_user_id)
|
||||
|
||||
Returns:
|
||||
是否删除成功
|
||||
"""
|
||||
try:
|
||||
key = cls._get_key("profile", user_id)
|
||||
result = await aio_redis.delete(key)
|
||||
logger.info(f"删除用户画像缓存: {key}, 结果: {result}")
|
||||
return result > 0
|
||||
except Exception as e:
|
||||
logger.error(f"删除用户画像缓存失败: {e}", exc_info=True)
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
async def get_profile_ttl(cls, user_id: str) -> int:
|
||||
"""获取用户画像缓存的剩余过期时间
|
||||
|
||||
Args:
|
||||
user_id: 用户ID(end_user_id)
|
||||
|
||||
Returns:
|
||||
剩余秒数,-1表示永不过期,-2表示key不存在
|
||||
"""
|
||||
try:
|
||||
key = cls._get_key("profile", user_id)
|
||||
ttl = await aio_redis.ttl(key)
|
||||
logger.debug(f"用户画像缓存TTL: {key} = {ttl}秒")
|
||||
return ttl
|
||||
except Exception as e:
|
||||
logger.error(f"获取用户画像缓存TTL失败: {e}")
|
||||
return -2
|
||||
@@ -4,6 +4,7 @@ from datetime import timedelta
|
||||
from urllib.parse import quote
|
||||
|
||||
from celery import Celery
|
||||
from celery.schedules import crontab
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
@@ -82,7 +83,8 @@ celery_app.conf.update(
|
||||
'app.tasks.workspace_reflection_task': {'queue': 'periodic_tasks'},
|
||||
'app.tasks.regenerate_memory_cache': {'queue': 'periodic_tasks'},
|
||||
'app.tasks.run_forgetting_cycle_task': {'queue': 'periodic_tasks'},
|
||||
'app.controllers.memory_storage_controller.search_all': {'queue': 'periodic_tasks'},
|
||||
'app.tasks.write_all_workspaces_memory_task': {'queue': 'periodic_tasks'},
|
||||
'app.tasks.update_implicit_emotions_storage': {'queue': 'periodic_tasks'},
|
||||
},
|
||||
)
|
||||
|
||||
@@ -92,9 +94,12 @@ celery_app.autodiscover_tasks(['app'])
|
||||
# Celery Beat schedule for periodic tasks
|
||||
memory_increment_schedule = timedelta(hours=settings.MEMORY_INCREMENT_INTERVAL_HOURS)
|
||||
memory_cache_regeneration_schedule = timedelta(hours=settings.MEMORY_CACHE_REGENERATION_HOURS)
|
||||
# 这个30秒的设计不合理
|
||||
workspace_reflection_schedule = timedelta(seconds=30) # 每30秒运行一次settings.REFLECTION_INTERVAL_TIME
|
||||
forgetting_cycle_schedule = timedelta(hours=24) # 每24小时运行一次遗忘周期
|
||||
workspace_reflection_schedule = timedelta(seconds=settings.WORKSPACE_REFLECTION_INTERVAL_SECONDS)
|
||||
forgetting_cycle_schedule = timedelta(hours=settings.FORGETTING_CYCLE_INTERVAL_HOURS)
|
||||
implicit_emotions_update_schedule = crontab(
|
||||
hour=settings.IMPLICIT_EMOTIONS_UPDATE_HOUR,
|
||||
minute=settings.IMPLICIT_EMOTIONS_UPDATE_MINUTE,
|
||||
)
|
||||
|
||||
#构建定时任务配置
|
||||
beat_schedule_config = {
|
||||
@@ -115,16 +120,16 @@ beat_schedule_config = {
|
||||
"config_id": None, # 使用默认配置,可以通过环境变量配置
|
||||
},
|
||||
},
|
||||
"write-all-workspaces-memory": {
|
||||
"task": "app.tasks.write_all_workspaces_memory_task",
|
||||
"schedule": memory_increment_schedule,
|
||||
"args": (),
|
||||
},
|
||||
"update-implicit-emotions-storage": {
|
||||
"task": "app.tasks.update_implicit_emotions_storage",
|
||||
"schedule": implicit_emotions_update_schedule,
|
||||
"args": (),
|
||||
},
|
||||
}
|
||||
|
||||
#如果配置了默认工作空间ID,则添加记忆总量统计任务
|
||||
if settings.DEFAULT_WORKSPACE_ID:
|
||||
beat_schedule_config["write-total-memory"] = {
|
||||
"task": "app.controllers.memory_storage_controller.search_all",
|
||||
"schedule": memory_increment_schedule,
|
||||
"kwargs": {
|
||||
"workspace_id": settings.DEFAULT_WORKSPACE_ID,
|
||||
},
|
||||
}
|
||||
|
||||
celery_app.conf.beat_schedule = beat_schedule_config
|
||||
|
||||
@@ -208,14 +208,64 @@ async def get_emotion_health(
|
||||
|
||||
|
||||
|
||||
# @router.post("/check-data", response_model=ApiResponse)
|
||||
# async def check_emotion_data_exists(
|
||||
# request: EmotionSuggestionsRequest,
|
||||
# db: Session = Depends(get_db),
|
||||
# current_user: User = Depends(get_current_user),
|
||||
# ):
|
||||
# """检查用户情绪建议数据是否存在
|
||||
|
||||
# Args:
|
||||
# request: 包含 end_user_id
|
||||
# db: 数据库会话
|
||||
# current_user: 当前用户
|
||||
|
||||
# Returns:
|
||||
# 数据存在状态
|
||||
# """
|
||||
# try:
|
||||
# api_logger.info(
|
||||
# f"检查用户情绪建议数据是否存在: {request.end_user_id}",
|
||||
# extra={"end_user_id": request.end_user_id}
|
||||
# )
|
||||
|
||||
# # 从数据库获取建议
|
||||
# data = await emotion_service.get_cached_suggestions(
|
||||
# end_user_id=request.end_user_id,
|
||||
# db=db
|
||||
# )
|
||||
|
||||
# if data is None:
|
||||
# api_logger.info(f"用户 {request.end_user_id} 的情绪建议数据不存在")
|
||||
# return fail(
|
||||
# BizCode.NOT_FOUND,
|
||||
# "情绪建议数据不存在,请点击右上角刷新进行初始化",
|
||||
# {"exists": False}
|
||||
# )
|
||||
|
||||
# api_logger.info(f"用户 {request.end_user_id} 的情绪建议数据存在")
|
||||
# return success(data={"exists": True}, msg="情绪建议数据已存在")
|
||||
|
||||
# except Exception as e:
|
||||
# api_logger.error(
|
||||
# f"检查情绪建议数据失败: {str(e)}",
|
||||
# extra={"end_user_id": request.end_user_id},
|
||||
# exc_info=True
|
||||
# )
|
||||
# raise HTTPException(
|
||||
# status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
# detail=f"检查情绪建议数据失败: {str(e)}"
|
||||
# )
|
||||
|
||||
|
||||
@router.post("/suggestions", response_model=ApiResponse)
|
||||
async def get_emotion_suggestions(
|
||||
request: EmotionSuggestionsRequest,
|
||||
language_type: str = Header(default=None, alias="X-Language-Type"),
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""获取个性化情绪建议(从缓存读取)
|
||||
"""获取个性化情绪建议(从数据库读取)
|
||||
|
||||
Args:
|
||||
request: 包含 end_user_id 和可选的 config_id
|
||||
@@ -223,77 +273,42 @@ async def get_emotion_suggestions(
|
||||
current_user: 当前用户
|
||||
|
||||
Returns:
|
||||
缓存的个性化情绪建议响应
|
||||
存储的个性化情绪建议响应
|
||||
"""
|
||||
try:
|
||||
# 使用集中化的语言校验
|
||||
language = get_language_from_header(language_type)
|
||||
|
||||
api_logger.info(
|
||||
f"用户 {current_user.username} 请求获取个性化情绪建议(缓存)",
|
||||
f"用户 {current_user.username} 请求获取个性化情绪建议",
|
||||
extra={
|
||||
"end_user_id": request.end_user_id,
|
||||
"config_id": request.config_id
|
||||
}
|
||||
)
|
||||
|
||||
# 从缓存获取建议
|
||||
# 从数据库获取建议
|
||||
data = await emotion_service.get_cached_suggestions(
|
||||
end_user_id=request.end_user_id,
|
||||
db=db
|
||||
)
|
||||
|
||||
if data is None:
|
||||
# 缓存不存在或已过期,自动触发生成
|
||||
api_logger.info(
|
||||
f"用户 {request.end_user_id} 的建议缓存不存在或已过期,自动生成新建议",
|
||||
f"用户 {request.end_user_id} 的建议数据不存在",
|
||||
extra={"end_user_id": request.end_user_id}
|
||||
)
|
||||
try:
|
||||
data = await emotion_service.generate_emotion_suggestions(
|
||||
end_user_id=request.end_user_id,
|
||||
db=db,
|
||||
language=language
|
||||
)
|
||||
# 保存到缓存
|
||||
await emotion_service.save_suggestions_cache(
|
||||
end_user_id=request.end_user_id,
|
||||
suggestions_data=data,
|
||||
db=db,
|
||||
expires_hours=24
|
||||
)
|
||||
except (ValueError, KeyError) as gen_e:
|
||||
# 预期内的业务异常:配置缺失、数据格式问题等
|
||||
api_logger.warning(
|
||||
f"自动生成建议失败(业务异常): {str(gen_e)}",
|
||||
extra={"end_user_id": request.end_user_id}
|
||||
)
|
||||
return fail(
|
||||
BizCode.NOT_FOUND,
|
||||
f"自动生成建议失败: {str(gen_e)}",
|
||||
""
|
||||
)
|
||||
except Exception as gen_e:
|
||||
# 非预期异常:记录完整 traceback 便于排查
|
||||
api_logger.error(
|
||||
f"自动生成建议时发生未预期异常: {str(gen_e)}",
|
||||
extra={"end_user_id": request.end_user_id},
|
||||
exc_info=True
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"生成建议时发生内部错误: {str(gen_e)}"
|
||||
)
|
||||
return success(
|
||||
data={"exists": False},
|
||||
msg="情绪建议数据不存在,请点击右上角刷新进行初始化"
|
||||
)
|
||||
|
||||
api_logger.info(
|
||||
"个性化建议获取成功(缓存)",
|
||||
"个性化建议获取成功",
|
||||
extra={
|
||||
"end_user_id": request.end_user_id,
|
||||
"suggestions_count": len(data.get("suggestions", []))
|
||||
}
|
||||
)
|
||||
|
||||
return success(data=data, msg="个性化建议获取成功(缓存)")
|
||||
return success(data=data, msg="个性化建议获取成功")
|
||||
|
||||
except Exception as e:
|
||||
api_logger.error(
|
||||
@@ -314,7 +329,7 @@ async def generate_emotion_suggestions(
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""生成个性化情绪建议(调用LLM并缓存)
|
||||
"""生成个性化情绪建议(调用LLM并保存到数据库)
|
||||
|
||||
Args:
|
||||
request: 包含 end_user_id
|
||||
@@ -342,12 +357,11 @@ async def generate_emotion_suggestions(
|
||||
language=language
|
||||
)
|
||||
|
||||
# 保存到缓存
|
||||
# 保存到数据库
|
||||
await emotion_service.save_suggestions_cache(
|
||||
end_user_id=request.end_user_id,
|
||||
suggestions_data=data,
|
||||
db=db,
|
||||
expires_hours=24
|
||||
db=db
|
||||
)
|
||||
|
||||
api_logger.info(
|
||||
@@ -369,4 +383,4 @@ async def generate_emotion_suggestions(
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"生成个性化建议失败: {str(e)}"
|
||||
)
|
||||
)
|
||||
@@ -122,6 +122,48 @@ def validate_confidence_threshold(threshold: float) -> None:
|
||||
raise ValueError("confidence_threshold must be between 0.0 and 1.0")
|
||||
|
||||
|
||||
@router.get("/check-data/{end_user_id}", response_model=ApiResponse)
|
||||
@cur_workspace_access_guard()
|
||||
async def check_user_data_exists(
|
||||
end_user_id: str,
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user)
|
||||
) -> ApiResponse:
|
||||
"""
|
||||
检查用户画像数据是否存在
|
||||
|
||||
Args:
|
||||
end_user_id: 目标用户ID
|
||||
|
||||
Returns:
|
||||
数据存在状态
|
||||
"""
|
||||
api_logger.info(f"检查用户画像数据是否存在: {end_user_id}")
|
||||
|
||||
try:
|
||||
# Validate inputs
|
||||
validate_user_id(end_user_id)
|
||||
|
||||
# Create service with user-specific config
|
||||
service = ImplicitMemoryService(db=db, end_user_id=end_user_id)
|
||||
|
||||
# Get cached profile
|
||||
cached_profile = await service.get_cached_profile(end_user_id=end_user_id, db=db)
|
||||
|
||||
if cached_profile is None:
|
||||
api_logger.info(f"用户 {end_user_id} 的画像数据不存在")
|
||||
return success(
|
||||
data={"exists": False},
|
||||
msg="画像数据不存在,请点击右上角刷新进行初始化"
|
||||
)
|
||||
|
||||
api_logger.info(f"用户 {end_user_id} 的画像数据存在")
|
||||
return success(data={"exists": True}, msg="画像数据已存在")
|
||||
|
||||
except Exception as e:
|
||||
return handle_implicit_memory_error(e, "检查画像数据", end_user_id)
|
||||
|
||||
|
||||
@router.get("/preferences/{end_user_id}", response_model=ApiResponse)
|
||||
@cur_workspace_access_guard()
|
||||
async def get_preference_tags(
|
||||
@@ -159,12 +201,8 @@ async def get_preference_tags(
|
||||
cached_profile = await service.get_cached_profile(end_user_id=end_user_id, db=db)
|
||||
|
||||
if cached_profile is None:
|
||||
api_logger.info(f"用户 {end_user_id} 的画像缓存不存在或已过期")
|
||||
return fail(
|
||||
BizCode.NOT_FOUND,
|
||||
"画像缓存不存在或已过期,请右上角刷新生成新画像",
|
||||
""
|
||||
)
|
||||
api_logger.info(f"用户 {end_user_id} 的画像数据不存在")
|
||||
return fail(BizCode.NOT_FOUND, "", "")
|
||||
|
||||
# Extract preferences from cache
|
||||
preferences = cached_profile.get("preferences", [])
|
||||
@@ -230,12 +268,8 @@ async def get_dimension_portrait(
|
||||
cached_profile = await service.get_cached_profile(end_user_id=end_user_id, db=db)
|
||||
|
||||
if cached_profile is None:
|
||||
api_logger.info(f"用户 {end_user_id} 的画像缓存不存在或已过期")
|
||||
return fail(
|
||||
BizCode.NOT_FOUND,
|
||||
"画像缓存不存在或已过期,请右上角刷新生成新画像",
|
||||
""
|
||||
)
|
||||
api_logger.info(f"用户 {end_user_id} 的画像数据不存在")
|
||||
return fail(BizCode.NOT_FOUND, "", "")
|
||||
|
||||
# Extract portrait from cache
|
||||
portrait = cached_profile.get("portrait", {})
|
||||
@@ -278,12 +312,8 @@ async def get_interest_area_distribution(
|
||||
cached_profile = await service.get_cached_profile(end_user_id=end_user_id, db=db)
|
||||
|
||||
if cached_profile is None:
|
||||
api_logger.info(f"用户 {end_user_id} 的画像缓存不存在或已过期")
|
||||
return fail(
|
||||
BizCode.NOT_FOUND,
|
||||
"画像缓存不存在或已过期,请右上角刷新生成新画像",
|
||||
""
|
||||
)
|
||||
api_logger.info(f"用户 {end_user_id} 的画像数据不存在")
|
||||
return fail(BizCode.NOT_FOUND, "", "")
|
||||
|
||||
# Extract interest areas from cache
|
||||
interest_areas = cached_profile.get("interest_areas", {})
|
||||
@@ -330,12 +360,8 @@ async def get_behavior_habits(
|
||||
cached_profile = await service.get_cached_profile(end_user_id=end_user_id, db=db)
|
||||
|
||||
if cached_profile is None:
|
||||
api_logger.info(f"用户 {end_user_id} 的画像缓存不存在或已过期")
|
||||
return fail(
|
||||
BizCode.NOT_FOUND,
|
||||
"画像缓存不存在或已过期,请右上角刷新生成新画像",
|
||||
""
|
||||
)
|
||||
api_logger.info(f"用户 {end_user_id} 的画像数据不存在")
|
||||
return fail(BizCode.NOT_FOUND, "", "")
|
||||
|
||||
# Extract habits from cache
|
||||
habits = cached_profile.get("habits", [])
|
||||
|
||||
@@ -633,12 +633,11 @@ async def get_knowledge_type_stats_api(
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
统计当前空间下各知识库类型的数量,包含 General | Web | Third-party | Folder | memory。
|
||||
统计当前空间下各知识库类型的数量,包含 General | Web | Third-party | Folder。
|
||||
会对缺失类型补 0,返回字典形式。
|
||||
可选按状态过滤。
|
||||
- 知识库类型根据当前用户的 current_workspace_id 过滤
|
||||
- memory 是 Neo4j 中 Chunk 的数量,根据 end_user_id (end_user_id) 过滤
|
||||
- 如果用户没有当前工作空间或未提供 end_user_id,对应的统计返回 0
|
||||
- 如果用户没有当前工作空间,对应的统计返回 0
|
||||
"""
|
||||
api_logger.info(f"Knowledge type stats requested for workspace_id: {current_user.current_workspace_id}, end_user_id: {end_user_id}")
|
||||
try:
|
||||
|
||||
@@ -9,6 +9,7 @@ from app.schemas.response_schema import ApiResponse
|
||||
|
||||
from app.services import memory_dashboard_service, memory_storage_service, workspace_service
|
||||
from app.services.memory_agent_service import get_end_users_connected_configs_batch
|
||||
from app.services.app_statistics_service import AppStatisticsService
|
||||
from app.core.logging_config import get_api_logger
|
||||
|
||||
# 获取API专用日志器
|
||||
@@ -469,6 +470,8 @@ async def get_chunk_insight(
|
||||
@router.get("/dashboard_data", response_model=ApiResponse)
|
||||
async def dashboard_data(
|
||||
end_user_id: Optional[str] = Query(None, description="可选的用户ID"),
|
||||
start_date: Optional[int] = Query(None, description="开始时间戳(毫秒)"),
|
||||
end_date: Optional[int] = Query(None, description="结束时间戳(毫秒)"),
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
@@ -503,6 +506,15 @@ async def dashboard_data(
|
||||
workspace_id = current_user.current_workspace_id
|
||||
api_logger.info(f"用户 {current_user.username} 请求获取工作空间 {workspace_id} 的dashboard整合数据")
|
||||
|
||||
# 如果没有提供时间范围,默认使用最近30天
|
||||
if start_date is None or end_date is None:
|
||||
from datetime import datetime, timedelta
|
||||
end_dt = datetime.now()
|
||||
start_dt = end_dt - timedelta(days=30)
|
||||
end_date = int(end_dt.timestamp() * 1000)
|
||||
start_date = int(start_dt.timestamp() * 1000)
|
||||
api_logger.info(f"使用默认时间范围: {start_dt} 到 {end_dt}")
|
||||
|
||||
# 获取 storage_type,如果为 None 则使用默认值
|
||||
storage_type = workspace_service.get_workspace_storage_type(
|
||||
db=db,
|
||||
@@ -563,17 +575,22 @@ async def dashboard_data(
|
||||
except Exception as e:
|
||||
api_logger.warning(f"获取知识库类型统计失败: {str(e)}")
|
||||
|
||||
# 3. 获取API调用增量(total_api_call,转换为整数)
|
||||
# 3. 获取API调用统计(total_api_call)
|
||||
try:
|
||||
api_increment = memory_dashboard_service.get_workspace_api_increment(
|
||||
db=db,
|
||||
# 使用 AppStatisticsService 获取真实的API调用统计
|
||||
app_stats_service = AppStatisticsService(db)
|
||||
api_stats = app_stats_service.get_workspace_api_statistics(
|
||||
workspace_id=workspace_id,
|
||||
current_user=current_user
|
||||
start_date=start_date,
|
||||
end_date=end_date
|
||||
)
|
||||
neo4j_data["total_api_call"] = api_increment
|
||||
api_logger.info(f"成功获取API调用增量: {neo4j_data['total_api_call']}")
|
||||
# 计算总调用次数
|
||||
total_api_calls = sum(item.get("total_calls", 0) for item in api_stats)
|
||||
neo4j_data["total_api_call"] = total_api_calls
|
||||
api_logger.info(f"成功获取API调用统计: {neo4j_data['total_api_call']}")
|
||||
except Exception as e:
|
||||
api_logger.warning(f"获取API调用增量失败: {str(e)}")
|
||||
api_logger.error(f"获取API调用统计失败: {str(e)}")
|
||||
neo4j_data["total_api_call"] = 0
|
||||
|
||||
result["neo4j_data"] = neo4j_data
|
||||
api_logger.info("成功获取neo4j_data")
|
||||
@@ -602,10 +619,23 @@ async def dashboard_data(
|
||||
total_kb = memory_dashboard_service.get_rag_total_kb(db, current_user)
|
||||
rag_data["total_knowledge"] = total_kb
|
||||
|
||||
# total_api_call: 固定值
|
||||
rag_data["total_api_call"] = 1024
|
||||
# total_api_call: 使用 AppStatisticsService 获取真实的API调用统计
|
||||
try:
|
||||
app_stats_service = AppStatisticsService(db)
|
||||
api_stats = app_stats_service.get_workspace_api_statistics(
|
||||
workspace_id=workspace_id,
|
||||
start_date=start_date,
|
||||
end_date=end_date
|
||||
)
|
||||
# 计算总调用次数
|
||||
total_api_calls = sum(item.get("total_calls", 0) for item in api_stats)
|
||||
rag_data["total_api_call"] = total_api_calls
|
||||
api_logger.info(f"成功获取RAG模式API调用统计: {rag_data['total_api_call']}")
|
||||
except Exception as e:
|
||||
api_logger.warning(f"获取RAG模式API调用统计失败,使用默认值: {str(e)}")
|
||||
rag_data["total_api_call"] = 0
|
||||
|
||||
api_logger.info(f"成功获取RAG相关数据: memory={total_chunk}, app={len(apps_orm)}, knowledge={total_kb}")
|
||||
api_logger.info(f"成功获取RAG相关数据: memory={total_chunk}, app={len(apps_orm)}, knowledge={total_kb}, api_calls={rag_data['total_api_call']}")
|
||||
except Exception as e:
|
||||
api_logger.warning(f"获取RAG相关数据失败: {str(e)}")
|
||||
|
||||
|
||||
@@ -202,13 +202,20 @@ class Settings:
|
||||
REFLECTION_INTERVAL_SECONDS: float = float(os.getenv("REFLECTION_INTERVAL_SECONDS", "300"))
|
||||
HEALTH_CHECK_SECONDS: float = float(os.getenv("HEALTH_CHECK_SECONDS", "600"))
|
||||
MEMORY_INCREMENT_INTERVAL_HOURS: float = float(os.getenv("MEMORY_INCREMENT_INTERVAL_HOURS", "24"))
|
||||
DEFAULT_WORKSPACE_ID: Optional[str] = os.getenv("DEFAULT_WORKSPACE_ID", None)
|
||||
REFLECTION_INTERVAL_TIME: Optional[str] = int(os.getenv("REFLECTION_INTERVAL_TIME", 30))
|
||||
|
||||
# Memory Cache Regeneration Configuration
|
||||
MEMORY_CACHE_REGENERATION_HOURS: int = int(os.getenv("MEMORY_CACHE_REGENERATION_HOURS", "24"))
|
||||
|
||||
# Memory Module Configuration (internal)
|
||||
# Periodic Task Schedule Configuration
|
||||
# workspace_reflection: 每隔多少秒执行一次
|
||||
WORKSPACE_REFLECTION_INTERVAL_SECONDS: int = int(os.getenv("WORKSPACE_REFLECTION_INTERVAL_SECONDS", "30"))
|
||||
# forgetting_cycle: 每隔多少小时执行一次
|
||||
FORGETTING_CYCLE_INTERVAL_HOURS: int = int(os.getenv("FORGETTING_CYCLE_INTERVAL_HOURS", "24"))
|
||||
# implicit_emotions_update: 每天几点执行(小时,0-23)
|
||||
IMPLICIT_EMOTIONS_UPDATE_HOUR: int = int(os.getenv("IMPLICIT_EMOTIONS_UPDATE_HOUR", "2"))
|
||||
# implicit_emotions_update: 每天几分执行(分钟,0-59)
|
||||
IMPLICIT_EMOTIONS_UPDATE_MINUTE: int = int(os.getenv("IMPLICIT_EMOTIONS_UPDATE_MINUTE", "0")) # Memory Module Configuration (internal)
|
||||
MEMORY_OUTPUT_DIR: str = os.getenv("MEMORY_OUTPUT_DIR", "logs/memory-output")
|
||||
MEMORY_CONFIG_DIR: str = os.getenv("MEMORY_CONFIG_DIR", "app/core/memory")
|
||||
|
||||
|
||||
@@ -35,6 +35,7 @@ from .ontology_scene import OntologyScene
|
||||
from .ontology_class import OntologyClass
|
||||
from .ontology_scene import OntologyScene
|
||||
from .ontology_class import OntologyClass
|
||||
from .implicit_emotions_storage_model import ImplicitEmotionsStorage
|
||||
|
||||
__all__ = [
|
||||
"Tenants",
|
||||
@@ -90,5 +91,6 @@ __all__ = [
|
||||
"MemoryPerceptualModel",
|
||||
"ModelBase",
|
||||
"LoadBalanceStrategy",
|
||||
"Skill"
|
||||
"Skill",
|
||||
"ImplicitEmotionsStorage"
|
||||
]
|
||||
|
||||
45
api/app/models/implicit_emotions_storage_model.py
Normal file
45
api/app/models/implicit_emotions_storage_model.py
Normal file
@@ -0,0 +1,45 @@
|
||||
"""
|
||||
Implicit Emotions Storage Model
|
||||
|
||||
数据库模型:存储用户的隐性记忆画像和情绪建议数据
|
||||
替代原有的Redis缓存方式
|
||||
"""
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from sqlalchemy import Column, String, Text, DateTime, Index
|
||||
from sqlalchemy.dialects.postgresql import UUID, JSONB
|
||||
from app.db import Base
|
||||
|
||||
|
||||
class ImplicitEmotionsStorage(Base):
|
||||
"""隐性记忆和情绪存储表"""
|
||||
|
||||
__tablename__ = "implicit_emotions_storage"
|
||||
|
||||
# 主键
|
||||
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, comment="主键ID")
|
||||
|
||||
# 用户标识(unique=True会自动创建唯一索引)
|
||||
end_user_id = Column(String(255), nullable=False, unique=True, comment="终端用户ID")
|
||||
|
||||
# 隐性记忆画像数据(JSON格式)
|
||||
implicit_profile = Column(JSONB, nullable=True, comment="隐性记忆用户画像数据")
|
||||
|
||||
# 情绪建议数据(JSON格式)
|
||||
emotion_suggestions = Column(JSONB, nullable=True, comment="情绪个性化建议数据")
|
||||
|
||||
# 时间戳
|
||||
created_at = Column(DateTime, nullable=False, default=datetime.utcnow, comment="创建时间")
|
||||
updated_at = Column(DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow, comment="更新时间")
|
||||
|
||||
# 数据生成时间(用于业务逻辑)
|
||||
implicit_generated_at = Column(DateTime, nullable=True, comment="隐性记忆画像生成时间")
|
||||
emotion_generated_at = Column(DateTime, nullable=True, comment="情绪建议生成时间")
|
||||
|
||||
# 索引(只为updated_at创建索引,end_user_id的unique约束已自动创建索引)
|
||||
__table_args__ = (
|
||||
Index('idx_updated_at', 'updated_at'),
|
||||
)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<ImplicitEmotionsStorage(id={self.id}, end_user_id={self.end_user_id})>"
|
||||
169
api/app/repositories/implicit_emotions_storage_repository.py
Normal file
169
api/app/repositories/implicit_emotions_storage_repository.py
Normal file
@@ -0,0 +1,169 @@
|
||||
"""
|
||||
Implicit Emotions Storage Repository
|
||||
|
||||
数据访问层:处理隐性记忆和情绪数据的数据库操作
|
||||
事务由调用方控制,仓储层只使用 flush/refresh
|
||||
"""
|
||||
import logging
|
||||
from datetime import datetime, date, timezone, timedelta
|
||||
from typing import Optional, Generator
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import select, not_, exists
|
||||
|
||||
from app.models.implicit_emotions_storage_model import ImplicitEmotionsStorage
|
||||
from app.models.end_user_model import EndUser
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ImplicitEmotionsStorageRepository:
|
||||
"""隐性记忆和情绪存储仓储类"""
|
||||
|
||||
def __init__(self, db: Session):
|
||||
self.db = db
|
||||
|
||||
def get_by_end_user_id(self, end_user_id: str) -> Optional[ImplicitEmotionsStorage]:
|
||||
"""根据终端用户ID获取存储记录"""
|
||||
try:
|
||||
stmt = select(ImplicitEmotionsStorage).where(
|
||||
ImplicitEmotionsStorage.end_user_id == end_user_id
|
||||
)
|
||||
return self.db.execute(stmt).scalar_one_or_none()
|
||||
except Exception as e:
|
||||
logger.error(f"获取用户存储记录失败: end_user_id={end_user_id}, error={e}")
|
||||
return None
|
||||
|
||||
def create(self, end_user_id: str) -> ImplicitEmotionsStorage:
|
||||
"""创建新的存储记录(事务由调用方提交)"""
|
||||
storage = ImplicitEmotionsStorage(
|
||||
end_user_id=end_user_id,
|
||||
created_at=datetime.utcnow(),
|
||||
updated_at=datetime.utcnow()
|
||||
)
|
||||
self.db.add(storage)
|
||||
self.db.flush()
|
||||
self.db.refresh(storage)
|
||||
logger.info(f"创建用户存储记录成功: end_user_id={end_user_id}")
|
||||
return storage
|
||||
|
||||
def update_implicit_profile(
|
||||
self,
|
||||
end_user_id: str,
|
||||
profile_data: dict
|
||||
) -> ImplicitEmotionsStorage:
|
||||
"""更新隐性记忆画像数据(事务由调用方提交)"""
|
||||
storage = self.get_by_end_user_id(end_user_id)
|
||||
if storage is None:
|
||||
storage = self.create(end_user_id)
|
||||
|
||||
storage.implicit_profile = profile_data
|
||||
storage.implicit_generated_at = datetime.utcnow()
|
||||
storage.updated_at = datetime.utcnow()
|
||||
|
||||
self.db.flush()
|
||||
self.db.refresh(storage)
|
||||
logger.info(f"更新隐性记忆画像成功: end_user_id={end_user_id}")
|
||||
return storage
|
||||
|
||||
def update_emotion_suggestions(
|
||||
self,
|
||||
end_user_id: str,
|
||||
suggestions_data: dict
|
||||
) -> ImplicitEmotionsStorage:
|
||||
"""更新情绪建议数据(事务由调用方提交)"""
|
||||
storage = self.get_by_end_user_id(end_user_id)
|
||||
if storage is None:
|
||||
storage = self.create(end_user_id)
|
||||
|
||||
storage.emotion_suggestions = suggestions_data
|
||||
storage.emotion_generated_at = datetime.utcnow()
|
||||
storage.updated_at = datetime.utcnow()
|
||||
|
||||
self.db.flush()
|
||||
self.db.refresh(storage)
|
||||
logger.info(f"更新情绪建议成功: end_user_id={end_user_id}")
|
||||
return storage
|
||||
|
||||
def get_all_user_ids(self, batch_size: int = 100) -> Generator[str, None, None]:
|
||||
"""分批次获取所有已存储数据的用户ID(避免大数据量内存溢出)
|
||||
|
||||
Args:
|
||||
batch_size: 每批次加载的数量,默认100
|
||||
|
||||
Yields:
|
||||
用户ID字符串
|
||||
"""
|
||||
offset = 0
|
||||
while True:
|
||||
try:
|
||||
stmt = (
|
||||
select(ImplicitEmotionsStorage.end_user_id)
|
||||
.order_by(ImplicitEmotionsStorage.end_user_id)
|
||||
.limit(batch_size)
|
||||
.offset(offset)
|
||||
)
|
||||
batch = self.db.execute(stmt).scalars().all()
|
||||
if not batch:
|
||||
break
|
||||
yield from batch
|
||||
offset += batch_size
|
||||
except Exception as e:
|
||||
logger.error(f"分批获取用户ID失败: offset={offset}, error={e}")
|
||||
break
|
||||
|
||||
def get_new_user_ids_today(self, batch_size: int = 100) -> Generator[str, None, None]:
|
||||
"""分批次获取当天新增的、尚未初始化隐性记忆和情绪建议数据的用户ID
|
||||
|
||||
查询逻辑:end_users 表中 created_at 为今天,且在 implicit_emotions_storage 中没有对应记录。
|
||||
没有对应记录意味着隐性记忆画像和情绪建议均未初始化,需要对这批用户执行首次初始化。
|
||||
end_users.id(UUID)转为字符串后与 implicit_emotions_storage.end_user_id(String)对比。
|
||||
|
||||
Args:
|
||||
batch_size: 每批次加载的数量,默认100
|
||||
|
||||
Yields:
|
||||
用户ID字符串
|
||||
"""
|
||||
from sqlalchemy import cast, String as SAString
|
||||
CST = timezone(timedelta(hours=8))
|
||||
now_cst = datetime.now(CST)
|
||||
today_start = now_cst.replace(hour=0, minute=0, second=0, microsecond=0).astimezone(timezone.utc).replace(tzinfo=None)
|
||||
tomorrow_start = today_start + timedelta(days=1)
|
||||
offset = 0
|
||||
while True:
|
||||
try:
|
||||
stmt = (
|
||||
select(EndUser.id)
|
||||
.where(
|
||||
EndUser.created_at >= today_start,
|
||||
EndUser.created_at < tomorrow_start,
|
||||
not_(
|
||||
exists(
|
||||
select(ImplicitEmotionsStorage.end_user_id).where(
|
||||
ImplicitEmotionsStorage.end_user_id == cast(EndUser.id, SAString)
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
.order_by(EndUser.id)
|
||||
.limit(batch_size)
|
||||
.offset(offset)
|
||||
)
|
||||
batch = self.db.execute(stmt).scalars().all()
|
||||
if not batch:
|
||||
break
|
||||
yield from (str(uid) for uid in batch)
|
||||
offset += batch_size
|
||||
except Exception as e:
|
||||
logger.error(f"分批获取当天新增用户ID失败: offset={offset}, error={e}")
|
||||
break
|
||||
|
||||
def delete_by_end_user_id(self, end_user_id: str) -> bool:
|
||||
"""删除用户的存储记录(事务由调用方提交)"""
|
||||
storage = self.get_by_end_user_id(end_user_id)
|
||||
if storage:
|
||||
self.db.delete(storage)
|
||||
self.db.flush()
|
||||
logger.info(f"删除用户存储记录成功: end_user_id={end_user_id}")
|
||||
return True
|
||||
return False
|
||||
@@ -843,32 +843,33 @@ class EmotionAnalyticsService:
|
||||
end_user_id: str,
|
||||
db: Session,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""从 Redis 缓存获取个性化情绪建议
|
||||
"""从数据库获取个性化情绪建议
|
||||
|
||||
Args:
|
||||
end_user_id: 宿主ID(用户组ID)
|
||||
db: 数据库会话(保留参数以保持接口兼容性)
|
||||
db: 数据库会话
|
||||
|
||||
Returns:
|
||||
Dict: 缓存的建议数据,如果不存在或已过期返回 None
|
||||
Dict: 存储的建议数据,如果不存在返回 None
|
||||
"""
|
||||
try:
|
||||
from app.cache.memory.emotion_memory import EmotionMemoryCache
|
||||
from app.repositories.implicit_emotions_storage_repository import ImplicitEmotionsStorageRepository
|
||||
|
||||
logger.info(f"尝试从 Redis 缓存获取情绪建议: user={end_user_id}")
|
||||
logger.info(f"尝试从数据库获取情绪建议: user={end_user_id}")
|
||||
|
||||
# 从 Redis 获取缓存
|
||||
cached_data = await EmotionMemoryCache.get_emotion_suggestions(end_user_id)
|
||||
# 从数据库获取存储记录
|
||||
repo = ImplicitEmotionsStorageRepository(db)
|
||||
storage = repo.get_by_end_user_id(end_user_id)
|
||||
|
||||
if cached_data is None:
|
||||
logger.info(f"用户 {end_user_id} 的建议缓存不存在或已过期")
|
||||
if storage is None or storage.emotion_suggestions is None:
|
||||
logger.info(f"用户 {end_user_id} 的建议数据不存在")
|
||||
return None
|
||||
|
||||
logger.info(f"成功从 Redis 缓存获取建议: user={end_user_id}")
|
||||
return cached_data
|
||||
logger.info(f"成功从数据库获取建议: user={end_user_id}")
|
||||
return storage.emotion_suggestions
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"从 Redis 缓存获取建议失败: {str(e)}", exc_info=True)
|
||||
logger.error(f"从数据库获取建议失败: {str(e)}", exc_info=True)
|
||||
return None
|
||||
|
||||
async def save_suggestions_cache(
|
||||
@@ -876,36 +877,27 @@ class EmotionAnalyticsService:
|
||||
end_user_id: str,
|
||||
suggestions_data: Dict[str, Any],
|
||||
db: Session,
|
||||
expires_hours: int = 24
|
||||
expires_hours: int = 24 # 参数保留以保持接口兼容性
|
||||
) -> None:
|
||||
"""保存建议到 Redis 缓存
|
||||
"""保存建议到数据库
|
||||
|
||||
Args:
|
||||
end_user_id: 宿主ID(用户组ID)
|
||||
suggestions_data: 建议数据
|
||||
db: 数据库会话(保留参数以保持接口兼容性)
|
||||
expires_hours: 过期时间(小时),默认24小时
|
||||
db: 数据库会话
|
||||
expires_hours: 保留参数(兼容性)
|
||||
"""
|
||||
try:
|
||||
from app.cache.memory.emotion_memory import EmotionMemoryCache
|
||||
from app.repositories.implicit_emotions_storage_repository import ImplicitEmotionsStorageRepository
|
||||
|
||||
logger.info(f"保存建议到 Redis 缓存: user={end_user_id}, expires={expires_hours}小时")
|
||||
logger.info(f"保存建议到数据库: user={end_user_id}")
|
||||
|
||||
# 计算过期时间(秒)
|
||||
expire_seconds = expires_hours * 3600
|
||||
repo = ImplicitEmotionsStorageRepository(db)
|
||||
repo.update_emotion_suggestions(end_user_id, suggestions_data)
|
||||
db.commit()
|
||||
|
||||
# 保存到 Redis
|
||||
success = await EmotionMemoryCache.set_emotion_suggestions(
|
||||
user_id=end_user_id,
|
||||
suggestions_data=suggestions_data,
|
||||
expire=expire_seconds
|
||||
)
|
||||
|
||||
if success:
|
||||
logger.info(f"建议缓存保存成功: user={end_user_id}")
|
||||
else:
|
||||
logger.warning(f"建议缓存保存失败: user={end_user_id}")
|
||||
logger.info(f"建议保存成功: user={end_user_id}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"保存建议缓存失败: {str(e)}", exc_info=True)
|
||||
# 不抛出异常,缓存失败不应影响主流程
|
||||
db.rollback()
|
||||
logger.error(f"保存建议失败: {str(e)}", exc_info=True)
|
||||
@@ -422,32 +422,33 @@ class ImplicitMemoryService:
|
||||
end_user_id: str,
|
||||
db: Session
|
||||
) -> Optional[dict]:
|
||||
"""从 Redis 缓存获取完整用户画像
|
||||
"""从数据库获取完整用户画像
|
||||
|
||||
Args:
|
||||
end_user_id: 终端用户ID
|
||||
db: 数据库会话(保留参数以保持接口兼容性)
|
||||
db: 数据库会话
|
||||
|
||||
Returns:
|
||||
Dict: 缓存的画像数据,如果不存在或已过期返回 None
|
||||
Dict: 存储的画像数据,如果不存在返回 None
|
||||
"""
|
||||
try:
|
||||
from app.cache.memory.implicit_memory import ImplicitMemoryCache
|
||||
from app.repositories.implicit_emotions_storage_repository import ImplicitEmotionsStorageRepository
|
||||
|
||||
logger.info(f"尝试从 Redis 缓存获取用户画像: user={end_user_id}")
|
||||
logger.info(f"尝试从数据库获取用户画像: user={end_user_id}")
|
||||
|
||||
# 从 Redis 获取缓存
|
||||
cached_data = await ImplicitMemoryCache.get_user_profile(end_user_id)
|
||||
# 从数据库获取存储记录
|
||||
repo = ImplicitEmotionsStorageRepository(db)
|
||||
storage = repo.get_by_end_user_id(end_user_id)
|
||||
|
||||
if cached_data is None:
|
||||
logger.info(f"用户 {end_user_id} 的画像缓存不存在或已过期")
|
||||
if storage is None or storage.implicit_profile is None:
|
||||
logger.info(f"用户 {end_user_id} 的画像数据不存在")
|
||||
return None
|
||||
|
||||
logger.info(f"成功从 Redis 缓存获取用户画像: user={end_user_id}")
|
||||
return cached_data
|
||||
logger.info(f"成功从数据库获取用户画像: user={end_user_id}")
|
||||
return storage.implicit_profile
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"从 Redis 缓存获取用户画像失败: {str(e)}", exc_info=True)
|
||||
logger.error(f"从数据库获取用户画像失败: {str(e)}", exc_info=True)
|
||||
return None
|
||||
|
||||
async def save_profile_cache(
|
||||
@@ -455,36 +456,27 @@ class ImplicitMemoryService:
|
||||
end_user_id: str,
|
||||
profile_data: dict,
|
||||
db: Session,
|
||||
expires_hours: int = 168 # 默认7天
|
||||
expires_hours: int = 168 # 参数保留以保持接口兼容性
|
||||
) -> None:
|
||||
"""保存用户画像到 Redis 缓存
|
||||
"""保存用户画像到数据库
|
||||
|
||||
Args:
|
||||
end_user_id: 终端用户ID
|
||||
profile_data: 画像数据
|
||||
db: 数据库会话(保留参数以保持接口兼容性)
|
||||
expires_hours: 过期时间(小时),默认168小时(7天)
|
||||
db: 数据库会话
|
||||
expires_hours: 保留参数(兼容性)
|
||||
"""
|
||||
try:
|
||||
from app.cache.memory.implicit_memory import ImplicitMemoryCache
|
||||
from app.repositories.implicit_emotions_storage_repository import ImplicitEmotionsStorageRepository
|
||||
|
||||
logger.info(f"保存用户画像到 Redis 缓存: user={end_user_id}, expires={expires_hours}小时")
|
||||
logger.info(f"保存用户画像到数据库: user={end_user_id}")
|
||||
|
||||
# 计算过期时间(秒)
|
||||
expire_seconds = expires_hours * 3600
|
||||
repo = ImplicitEmotionsStorageRepository(db)
|
||||
repo.update_implicit_profile(end_user_id, profile_data)
|
||||
db.commit()
|
||||
|
||||
# 保存到 Redis
|
||||
success = await ImplicitMemoryCache.set_user_profile(
|
||||
user_id=end_user_id,
|
||||
profile_data=profile_data,
|
||||
expire=expire_seconds
|
||||
)
|
||||
|
||||
if success:
|
||||
logger.info(f"用户画像缓存保存成功: user={end_user_id}")
|
||||
else:
|
||||
logger.warning(f"用户画像缓存保存失败: user={end_user_id}")
|
||||
logger.info(f"用户画像保存成功: user={end_user_id}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"保存用户画像缓存失败: {str(e)}", exc_info=True)
|
||||
# 不抛出异常,缓存失败不应影响主流程
|
||||
db.rollback()
|
||||
logger.error(f"保存用户画像失败: {str(e)}", exc_info=True)
|
||||
|
||||
@@ -816,11 +816,10 @@ class MemoryAgentService:
|
||||
"""
|
||||
统计知识库类型分布,包含:
|
||||
1. PostgreSQL 中的知识库类型:General, Web, Third-party, Folder(根据 workspace_id 过滤)
|
||||
2. Neo4j 中的 memory 类型(仅统计 Chunk 数量,根据 end_user_id/end_user_id 过滤)
|
||||
3. total: 所有类型的总和
|
||||
2. total: 所有类型的总和
|
||||
|
||||
参数:
|
||||
- end_user_id: 用户组ID(可选,未提供时 memory 统计为 0)
|
||||
- end_user_id: 用户组ID(可选,保留参数以保持接口兼容性)
|
||||
- only_active: 是否仅统计有效记录
|
||||
- current_workspace_id: 当前工作空间ID(可选,未提供时知识库统计为 0)
|
||||
- db: 数据库会话
|
||||
@@ -831,7 +830,6 @@ class MemoryAgentService:
|
||||
"Web": count,
|
||||
"Third-party": count,
|
||||
"Folder": count,
|
||||
"memory": chunk_count,
|
||||
"total": sum_of_all
|
||||
}
|
||||
"""
|
||||
@@ -878,51 +876,8 @@ class MemoryAgentService:
|
||||
logger.error(f"知识库类型统计失败: {e}")
|
||||
raise Exception(f"知识库类型统计失败: {e}")
|
||||
|
||||
# 2. 统计 Neo4j 中的 memory 总量(统计当前空间下所有宿主的 Chunk 总数)
|
||||
try:
|
||||
if current_workspace_id:
|
||||
# 获取当前空间下的所有宿主
|
||||
from app.repositories import app_repository, end_user_repository
|
||||
from app.schemas.app_schema import App as AppSchema
|
||||
from app.schemas.end_user_schema import EndUser as EndUserSchema
|
||||
|
||||
# 查询应用并转换为 Pydantic 模型
|
||||
apps_orm = app_repository.get_apps_by_workspace_id(db, current_workspace_id)
|
||||
apps = [AppSchema.model_validate(h) for h in apps_orm]
|
||||
app_ids = [app.id for app in apps]
|
||||
|
||||
# 获取所有宿主
|
||||
end_users = []
|
||||
for app_id in app_ids:
|
||||
end_user_orm_list = end_user_repository.get_end_users_by_app_id(db, app_id)
|
||||
end_users.extend(h for h in end_user_orm_list)
|
||||
|
||||
# 统计所有宿主的 Chunk 总数
|
||||
total_chunks = 0
|
||||
for end_user in end_users:
|
||||
end_user_id_str = str(end_user.id)
|
||||
memory_query = """
|
||||
MATCH (n:Chunk) WHERE n.end_user_id = $end_user_id RETURN count(n) AS Count
|
||||
"""
|
||||
neo4j_result = await _neo4j_connector.execute_query(
|
||||
memory_query,
|
||||
end_user_id=end_user_id_str,
|
||||
)
|
||||
chunk_count = neo4j_result[0]["Count"] if neo4j_result else 0
|
||||
total_chunks += chunk_count
|
||||
logger.debug(f"EndUser {end_user_id_str} Chunk数量: {chunk_count}")
|
||||
|
||||
result["memory"] = total_chunks
|
||||
logger.info(f"Neo4j memory统计成功: 总Chunk数={total_chunks}, 宿主数={len(end_users)}")
|
||||
else:
|
||||
# 没有 workspace_id 时,返回 0
|
||||
result["memory"] = 0
|
||||
logger.info("未提供 workspace_id,memory 统计为 0")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Neo4j memory统计失败: {e}", exc_info=True)
|
||||
# 如果 Neo4j 查询失败,memory 设为 0
|
||||
result["memory"] = 0
|
||||
# 2. 统计 Neo4j 中的 memory 总量已移除
|
||||
# memory 字段不再返回
|
||||
|
||||
# 3. 计算知识库类型总和(不包括 memory)
|
||||
result["total"] = (
|
||||
|
||||
502
api/app/tasks.py
502
api/app/tasks.py
@@ -1304,6 +1304,203 @@ def write_total_memory_task(workspace_id: str) -> Dict[str, Any]:
|
||||
"workspace_id": workspace_id,
|
||||
"elapsed_time": elapsed_time,
|
||||
}
|
||||
@celery_app.task(
|
||||
name="app.tasks.write_all_workspaces_memory_task",
|
||||
bind=True,
|
||||
ignore_result=False,
|
||||
max_retries=3,
|
||||
acks_late=True,
|
||||
time_limit=3600,
|
||||
soft_time_limit=3300,
|
||||
)
|
||||
def write_all_workspaces_memory_task(self) -> Dict[str, Any]:
|
||||
"""定时任务:遍历所有工作空间,统计并写入记忆增量
|
||||
|
||||
此任务会:
|
||||
1. 查询所有活跃的工作空间
|
||||
2. 对每个工作空间统计记忆总量
|
||||
3. 将统计结果写入 memory_increments 表
|
||||
|
||||
Returns:
|
||||
包含任务执行结果的字典
|
||||
"""
|
||||
start_time = time.time()
|
||||
|
||||
async def _run() -> Dict[str, Any]:
|
||||
from app.core.logging_config import get_api_logger
|
||||
from app.models.workspace_model import Workspace
|
||||
from app.models.app_model import App
|
||||
from app.models.end_user_model import EndUser
|
||||
from app.repositories.memory_increment_repository import write_memory_increment
|
||||
from app.services.memory_storage_service import search_all
|
||||
|
||||
api_logger = get_api_logger()
|
||||
|
||||
with get_db_context() as db:
|
||||
try:
|
||||
# 获取所有活跃的工作空间
|
||||
workspaces = db.query(Workspace).filter(
|
||||
Workspace.is_active.is_(True)
|
||||
).all()
|
||||
|
||||
if not workspaces:
|
||||
api_logger.warning("没有找到活跃的工作空间")
|
||||
return {
|
||||
"status": "SUCCESS",
|
||||
"message": "没有找到活跃的工作空间",
|
||||
"workspace_count": 0,
|
||||
"workspace_results": []
|
||||
}
|
||||
|
||||
api_logger.info(f"开始统计 {len(workspaces)} 个工作空间的记忆增量")
|
||||
all_workspace_results = []
|
||||
|
||||
# 遍历每个工作空间
|
||||
for workspace in workspaces:
|
||||
workspace_id = workspace.id
|
||||
api_logger.info(f"开始处理工作空间: {workspace.name} (ID: {workspace_id})")
|
||||
|
||||
try:
|
||||
# 1. 查询当前workspace下的所有app(仅未删除的)
|
||||
apps = db.query(App).filter(
|
||||
App.workspace_id == workspace_id,
|
||||
App.is_active.is_(True)
|
||||
).all()
|
||||
|
||||
if not apps:
|
||||
# 如果没有app,总量为0
|
||||
memory_increment = write_memory_increment(
|
||||
db=db,
|
||||
workspace_id=workspace_id,
|
||||
total_num=0
|
||||
)
|
||||
all_workspace_results.append({
|
||||
"workspace_id": str(workspace_id),
|
||||
"workspace_name": workspace.name,
|
||||
"status": "SUCCESS",
|
||||
"total_num": 0,
|
||||
"end_user_count": 0,
|
||||
"memory_increment_id": str(memory_increment.id),
|
||||
"created_at": memory_increment.created_at.isoformat(),
|
||||
})
|
||||
api_logger.info(f"工作空间 {workspace.name} 没有应用,记录总量为0")
|
||||
continue
|
||||
|
||||
# 2. 查询所有app下的end_user_id(去重)
|
||||
app_ids = [app.id for app in apps]
|
||||
end_users = db.query(EndUser.id).filter(
|
||||
EndUser.app_id.in_(app_ids)
|
||||
).distinct().all()
|
||||
|
||||
# 3. 遍历所有end_user,查询每个宿主的记忆总量并累加
|
||||
total_num = 0
|
||||
end_user_details = []
|
||||
|
||||
for (end_user_id,) in end_users:
|
||||
try:
|
||||
# 调用 search_all 接口查询该宿主的总量
|
||||
result = await search_all(str(end_user_id))
|
||||
user_total = result.get("total", 0)
|
||||
total_num += user_total
|
||||
end_user_details.append({
|
||||
"end_user_id": str(end_user_id),
|
||||
"total": user_total
|
||||
})
|
||||
except Exception as e:
|
||||
# 记录单个用户查询失败,但继续处理其他用户
|
||||
api_logger.warning(f"查询用户 {end_user_id} 记忆失败: {str(e)}")
|
||||
end_user_details.append({
|
||||
"end_user_id": str(end_user_id),
|
||||
"total": 0,
|
||||
"error": str(e)
|
||||
})
|
||||
|
||||
# 4. 写入数据库
|
||||
memory_increment = write_memory_increment(
|
||||
db=db,
|
||||
workspace_id=workspace_id,
|
||||
total_num=total_num
|
||||
)
|
||||
|
||||
all_workspace_results.append({
|
||||
"workspace_id": str(workspace_id),
|
||||
"workspace_name": workspace.name,
|
||||
"status": "SUCCESS",
|
||||
"total_num": total_num,
|
||||
"end_user_count": len(end_users),
|
||||
"memory_increment_id": str(memory_increment.id),
|
||||
"created_at": memory_increment.created_at.isoformat(),
|
||||
})
|
||||
|
||||
api_logger.info(
|
||||
f"工作空间 {workspace.name} 统计完成: 总量={total_num}, 用户数={len(end_users)}"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
db.rollback() # 回滚失败的事务,允许继续处理下一个工作空间
|
||||
api_logger.error(f"处理工作空间 {workspace.name} (ID: {workspace_id}) 失败: {str(e)}")
|
||||
all_workspace_results.append({
|
||||
"workspace_id": str(workspace_id),
|
||||
"workspace_name": workspace.name,
|
||||
"status": "FAILURE",
|
||||
"error": str(e),
|
||||
"total_num": 0,
|
||||
"end_user_count": 0,
|
||||
})
|
||||
|
||||
total_memory = sum(r.get("total_num", 0) for r in all_workspace_results)
|
||||
success_count = sum(1 for r in all_workspace_results if r.get("status") == "SUCCESS")
|
||||
|
||||
return {
|
||||
"status": "SUCCESS",
|
||||
"message": f"成功处理 {success_count}/{len(workspaces)} 个工作空间,总记忆量: {total_memory}",
|
||||
"workspace_count": len(workspaces),
|
||||
"success_count": success_count,
|
||||
"total_memory": total_memory,
|
||||
"workspace_results": all_workspace_results
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
api_logger.error(f"记忆增量统计任务执行失败: {str(e)}")
|
||||
return {
|
||||
"status": "FAILURE",
|
||||
"error": str(e),
|
||||
"workspace_count": 0,
|
||||
"workspace_results": []
|
||||
}
|
||||
|
||||
try:
|
||||
# 使用 nest_asyncio 来避免事件循环冲突
|
||||
try:
|
||||
import nest_asyncio
|
||||
nest_asyncio.apply()
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
# 尝试获取现有事件循环,如果不存在则创建新的
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
if loop.is_closed():
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
except RuntimeError:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
result = loop.run_until_complete(_run())
|
||||
elapsed_time = time.time() - start_time
|
||||
result["elapsed_time"] = elapsed_time
|
||||
result["task_id"] = self.request.id
|
||||
|
||||
return result
|
||||
except Exception as e:
|
||||
elapsed_time = time.time() - start_time
|
||||
return {
|
||||
"status": "FAILURE",
|
||||
"error": str(e),
|
||||
"elapsed_time": elapsed_time,
|
||||
"task_id": self.request.id
|
||||
}
|
||||
|
||||
|
||||
@celery_app.task(
|
||||
@@ -1924,4 +2121,307 @@ def run_forgetting_cycle_task(self, config_id: Optional[uuid.UUID] = None) -> Di
|
||||
# "config_id": config_id,
|
||||
# "elapsed_time": elapsed_time,
|
||||
# "task_id": self.request.id
|
||||
# }
|
||||
# }
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# 隐性记忆和情绪数据更新定时任务
|
||||
# =============================================================================
|
||||
|
||||
@celery_app.task(
|
||||
name="app.tasks.update_implicit_emotions_storage",
|
||||
bind=True,
|
||||
ignore_result=True,
|
||||
max_retries=0,
|
||||
acks_late=False,
|
||||
time_limit=7200, # 2小时硬超时
|
||||
soft_time_limit=6900, # 1小时55分钟软超时
|
||||
)
|
||||
def update_implicit_emotions_storage(self) -> Dict[str, Any]:
|
||||
"""定时任务:更新所有用户的隐性记忆画像和情绪建议数据
|
||||
|
||||
遍历数据库中所有已存在数据的用户,为每个用户重新生成隐性记忆画像和情绪建议。
|
||||
实现错误隔离,单个用户失败不影响其他用户的处理。
|
||||
|
||||
Returns:
|
||||
包含任务执行结果的字典,包括:
|
||||
- status: 任务状态 (SUCCESS/FAILURE)
|
||||
- message: 执行消息
|
||||
- total_users: 总用户数
|
||||
- successful_implicit: 成功更新隐性记忆的用户数
|
||||
- successful_emotion: 成功更新情绪建议的用户数
|
||||
- failed: 失败的用户数
|
||||
- user_results: 每个用户的详细结果
|
||||
- elapsed_time: 执行耗时(秒)
|
||||
- task_id: 任务ID
|
||||
"""
|
||||
start_time = time.time()
|
||||
|
||||
async def _run() -> Dict[str, Any]:
|
||||
from app.core.logging_config import get_logger
|
||||
from app.repositories.implicit_emotions_storage_repository import ImplicitEmotionsStorageRepository
|
||||
from app.models.implicit_emotions_storage_model import ImplicitEmotionsStorage
|
||||
from sqlalchemy import select, func
|
||||
from app.services.implicit_memory_service import ImplicitMemoryService
|
||||
from app.services.emotion_analytics_service import EmotionAnalyticsService
|
||||
|
||||
logger = get_logger(__name__)
|
||||
logger.info("开始执行隐性记忆和情绪数据更新定时任务")
|
||||
|
||||
total_users = 0
|
||||
successful_implicit = 0
|
||||
successful_emotion = 0
|
||||
failed = 0
|
||||
user_results = []
|
||||
|
||||
with get_db_context() as db:
|
||||
try:
|
||||
# 获取所有已存储数据的用户ID(分批次处理)
|
||||
repo = ImplicitEmotionsStorageRepository(db)
|
||||
|
||||
# 先统计总数用于日志
|
||||
from sqlalchemy import func
|
||||
total_users = db.execute(
|
||||
select(func.count()).select_from(ImplicitEmotionsStorage)
|
||||
).scalar() or 0
|
||||
logger.info(f"找到 {total_users} 个需要更新的用户")
|
||||
|
||||
# 遍历每个用户并更新数据(分批次,避免一次性加载所有ID)
|
||||
for end_user_id in repo.get_all_user_ids(batch_size=100):
|
||||
logger.info(f"开始处理用户: {end_user_id}")
|
||||
user_start_time = time.time()
|
||||
|
||||
implicit_success = False
|
||||
emotion_success = False
|
||||
errors = []
|
||||
|
||||
try:
|
||||
# 更新隐性记忆画像
|
||||
try:
|
||||
implicit_service = ImplicitMemoryService(db=db, end_user_id=end_user_id)
|
||||
profile_data = await implicit_service.generate_complete_profile(user_id=end_user_id)
|
||||
await implicit_service.save_profile_cache(
|
||||
end_user_id=end_user_id,
|
||||
profile_data=profile_data,
|
||||
db=db
|
||||
)
|
||||
implicit_success = True
|
||||
logger.info(f"成功更新用户 {end_user_id} 的隐性记忆画像")
|
||||
except Exception as e:
|
||||
error_msg = f"隐性记忆更新失败: {str(e)}"
|
||||
errors.append(error_msg)
|
||||
logger.error(f"用户 {end_user_id} {error_msg}")
|
||||
|
||||
# 更新情绪建议
|
||||
try:
|
||||
emotion_service = EmotionAnalyticsService()
|
||||
suggestions_data = await emotion_service.generate_emotion_suggestions(
|
||||
end_user_id=end_user_id,
|
||||
db=db,
|
||||
language="zh"
|
||||
)
|
||||
await emotion_service.save_suggestions_cache(
|
||||
end_user_id=end_user_id,
|
||||
suggestions_data=suggestions_data,
|
||||
db=db
|
||||
)
|
||||
emotion_success = True
|
||||
logger.info(f"成功更新用户 {end_user_id} 的情绪建议")
|
||||
except Exception as e:
|
||||
error_msg = f"情绪建议更新失败: {str(e)}"
|
||||
errors.append(error_msg)
|
||||
logger.error(f"用户 {end_user_id} {error_msg}")
|
||||
|
||||
# 统计结果
|
||||
if implicit_success:
|
||||
successful_implicit += 1
|
||||
if emotion_success:
|
||||
successful_emotion += 1
|
||||
if not implicit_success and not emotion_success:
|
||||
failed += 1
|
||||
|
||||
user_elapsed = time.time() - user_start_time
|
||||
|
||||
# 记录用户处理结果
|
||||
user_result = {
|
||||
"end_user_id": end_user_id,
|
||||
"implicit_success": implicit_success,
|
||||
"emotion_success": emotion_success,
|
||||
"errors": errors,
|
||||
"elapsed_time": user_elapsed
|
||||
}
|
||||
user_results.append(user_result)
|
||||
|
||||
logger.info(
|
||||
f"用户 {end_user_id} 处理完成: "
|
||||
f"隐性记忆={'成功' if implicit_success else '失败'}, "
|
||||
f"情绪建议={'成功' if emotion_success else '失败'}, "
|
||||
f"耗时={user_elapsed:.2f}秒"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
# 单个用户失败不影响其他用户(错误隔离)
|
||||
failed += 1
|
||||
user_elapsed = time.time() - user_start_time
|
||||
error_info = {
|
||||
"end_user_id": end_user_id,
|
||||
"implicit_success": False,
|
||||
"emotion_success": False,
|
||||
"errors": [str(e)],
|
||||
"elapsed_time": user_elapsed
|
||||
}
|
||||
user_results.append(error_info)
|
||||
logger.error(f"处理用户 {end_user_id} 时出错: {str(e)}")
|
||||
|
||||
# ---- 处理增量用户(当天新增、尚未初始化的用户)----
|
||||
new_users_initialized = 0
|
||||
new_users_failed = 0
|
||||
logger.info("开始处理当天新增的增量用户初始化")
|
||||
|
||||
for end_user_id in repo.get_new_user_ids_today(batch_size=100):
|
||||
logger.info(f"开始初始化新用户: {end_user_id}")
|
||||
user_start_time = time.time()
|
||||
implicit_success = False
|
||||
emotion_success = False
|
||||
errors = []
|
||||
|
||||
try:
|
||||
try:
|
||||
implicit_service = ImplicitMemoryService(db=db, end_user_id=end_user_id)
|
||||
profile_data = await implicit_service.generate_complete_profile(user_id=end_user_id)
|
||||
await implicit_service.save_profile_cache(
|
||||
end_user_id=end_user_id,
|
||||
profile_data=profile_data,
|
||||
db=db
|
||||
)
|
||||
implicit_success = True
|
||||
logger.info(f"成功初始化新用户 {end_user_id} 的隐性记忆画像")
|
||||
except Exception as e:
|
||||
error_msg = f"隐性记忆初始化失败: {str(e)}"
|
||||
errors.append(error_msg)
|
||||
logger.error(f"新用户 {end_user_id} {error_msg}")
|
||||
|
||||
try:
|
||||
emotion_service = EmotionAnalyticsService()
|
||||
suggestions_data = await emotion_service.generate_emotion_suggestions(
|
||||
end_user_id=end_user_id,
|
||||
db=db,
|
||||
language="zh"
|
||||
)
|
||||
await emotion_service.save_suggestions_cache(
|
||||
end_user_id=end_user_id,
|
||||
suggestions_data=suggestions_data,
|
||||
db=db
|
||||
)
|
||||
emotion_success = True
|
||||
logger.info(f"成功初始化新用户 {end_user_id} 的情绪建议")
|
||||
except Exception as e:
|
||||
error_msg = f"情绪建议初始化失败: {str(e)}"
|
||||
errors.append(error_msg)
|
||||
logger.error(f"新用户 {end_user_id} {error_msg}")
|
||||
|
||||
if implicit_success or emotion_success:
|
||||
new_users_initialized += 1
|
||||
else:
|
||||
new_users_failed += 1
|
||||
|
||||
user_elapsed = time.time() - user_start_time
|
||||
user_results.append({
|
||||
"end_user_id": end_user_id,
|
||||
"type": "init",
|
||||
"implicit_success": implicit_success,
|
||||
"emotion_success": emotion_success,
|
||||
"errors": errors,
|
||||
"elapsed_time": user_elapsed
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
new_users_failed += 1
|
||||
user_elapsed = time.time() - user_start_time
|
||||
user_results.append({
|
||||
"end_user_id": end_user_id,
|
||||
"type": "init",
|
||||
"implicit_success": False,
|
||||
"emotion_success": False,
|
||||
"errors": [str(e)],
|
||||
"elapsed_time": user_elapsed
|
||||
})
|
||||
logger.error(f"初始化新用户 {end_user_id} 时出错: {str(e)}")
|
||||
|
||||
logger.info(
|
||||
f"增量用户初始化完成: 成功={new_users_initialized}, 失败={new_users_failed}"
|
||||
)
|
||||
# ---- 增量用户处理结束 ----
|
||||
|
||||
# 记录总体统计信息
|
||||
logger.info(
|
||||
f"隐性记忆和情绪数据更新定时任务完成: "
|
||||
f"存量用户总数={total_users}, "
|
||||
f"隐性记忆成功={successful_implicit}, "
|
||||
f"情绪建议成功={successful_emotion}, "
|
||||
f"存量失败={failed}, "
|
||||
f"增量初始化成功={new_users_initialized}, "
|
||||
f"增量初始化失败={new_users_failed}"
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "SUCCESS",
|
||||
"message": (
|
||||
f"存量用户 {total_users} 个,隐性记忆 {successful_implicit} 个成功,情绪建议 {successful_emotion} 个成功;"
|
||||
f"增量新用户初始化 {new_users_initialized} 个成功,{new_users_failed} 个失败"
|
||||
),
|
||||
"total_users": total_users,
|
||||
"successful_implicit": successful_implicit,
|
||||
"successful_emotion": successful_emotion,
|
||||
"failed": failed,
|
||||
"new_users_initialized": new_users_initialized,
|
||||
"new_users_failed": new_users_failed,
|
||||
"user_results": user_results[:50] # 只保留前50个用户的详细结果
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"隐性记忆和情绪数据更新定时任务执行失败: {str(e)}")
|
||||
return {
|
||||
"status": "FAILURE",
|
||||
"error": str(e),
|
||||
"total_users": total_users,
|
||||
"successful_implicit": successful_implicit,
|
||||
"successful_emotion": successful_emotion,
|
||||
"failed": failed,
|
||||
"new_users_initialized": 0,
|
||||
"new_users_failed": 0,
|
||||
"user_results": user_results[:50]
|
||||
}
|
||||
|
||||
try:
|
||||
# 使用 nest_asyncio 来避免事件循环冲突
|
||||
try:
|
||||
import nest_asyncio
|
||||
nest_asyncio.apply()
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
# 尝试获取现有事件循环,如果不存在则创建新的
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
if loop.is_closed():
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
except RuntimeError:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
result = loop.run_until_complete(_run())
|
||||
elapsed_time = time.time() - start_time
|
||||
result["elapsed_time"] = elapsed_time
|
||||
result["task_id"] = self.request.id
|
||||
|
||||
return result
|
||||
except Exception as e:
|
||||
elapsed_time = time.time() - start_time
|
||||
return {
|
||||
"status": "FAILURE",
|
||||
"error": str(e),
|
||||
"elapsed_time": elapsed_time,
|
||||
"task_id": self.request.id
|
||||
}
|
||||
|
||||
Submodule redbear-mem-benchmark updated: 4b0257bb4e...8494e82498
@@ -1,8 +1,8 @@
|
||||
/*
|
||||
* @Author: ZhaoYing
|
||||
* @Date: 2026-02-03 14:00:06
|
||||
* @Last Modified by: ZhaoYing
|
||||
* @Last Modified time: 2026-02-03 14:00:06
|
||||
* @Last Modified by: ZhaoYing
|
||||
* @Last Modified time: 2026-03-03 14:58:32
|
||||
*/
|
||||
import { request } from '@/utils/request'
|
||||
import type {
|
||||
@@ -163,9 +163,14 @@ export const getImplicitInterestAreas = (end_user_id: string) => {
|
||||
export const getImplicitHabits = (end_user_id: string) => {
|
||||
return request.get(`/memory/implicit-memory/habits/${end_user_id}`)
|
||||
}
|
||||
// Implicit Memory - Generate user portrait
|
||||
export const generateProfile = (end_user_id: string) => {
|
||||
return request.post(`/memory/implicit-memory/generate_profile`, { end_user_id })
|
||||
}
|
||||
// Implicit Memory - Check if data exists
|
||||
export const implicitCheckData = (end_user_id: string) => {
|
||||
return request.get(`/memory/implicit-memory/check-data/${end_user_id}`)
|
||||
}
|
||||
// Short-term memory
|
||||
export const getShortTerm = (end_user_id: string) => {
|
||||
return request.get(`/memory/short/short_term`, { end_user_id })
|
||||
|
||||
@@ -2276,6 +2276,7 @@ Memory Bear: After the rebellion, regional warlordism intensified for several re
|
||||
suggestions: 'Personalized Suggestions',
|
||||
suggestionLoading: 'Your personalized suggestions are being generated',
|
||||
item: 'item',
|
||||
noData: 'Emotion suggestion data does not exist, please click the refresh button to initialize',
|
||||
},
|
||||
reflectionEngine: {
|
||||
reflectionEngineConfig: 'Reflection Engine Configuration',
|
||||
@@ -2522,7 +2523,8 @@ Memory Bear: After the rebellion, regional warlordism intensified for several re
|
||||
context_details: 'Preference Details',
|
||||
supporting_evidence: 'Preference Source',
|
||||
specific_examples: 'Source',
|
||||
wordEmpty: 'Click on a node in the left chart to view preference details'
|
||||
wordEmpty: 'Click on a node in the left chart to view preference details',
|
||||
noData: 'Portrait data does not exist, please click the refresh button to initialize',
|
||||
},
|
||||
shortTermDetail: {
|
||||
title: 'Short-term memory is the "workbench" of the AI system, connecting instant conversations with long-term knowledge bases. Through real-time capture, deep retrieval, intelligent extraction and filtering transformation, temporary unstructured information is converted into valuable long-term knowledge.',
|
||||
|
||||
@@ -2272,6 +2272,7 @@ export const zh = {
|
||||
suggestions: '个性化建议',
|
||||
suggestionLoading: '您的个性化建议正在生成中',
|
||||
item: '个',
|
||||
noData: '情绪建议数据不存在,请点击刷新按钮进行初始化',
|
||||
},
|
||||
reflectionEngine: {
|
||||
reflectionEngineConfig: '反思引擎配置',
|
||||
@@ -2518,7 +2519,8 @@ export const zh = {
|
||||
context_details: '偏好详情',
|
||||
supporting_evidence: '偏好来源',
|
||||
specific_examples: '来源',
|
||||
wordEmpty: '点击左侧图表中的节点查看偏好详情'
|
||||
wordEmpty: '点击左侧图表中的节点查看偏好详情',
|
||||
noData: '画像数据不存在,请点击刷新按钮进行初始化',
|
||||
},
|
||||
shortTermDetail: {
|
||||
title: '短期记忆是AI系统的"工作台",连接即时对话与长期知识库。通过实时捕获、深度检索、智能提取和筛选转化,将临时的非结构化信息转化为有价值的长期知识。',
|
||||
|
||||
@@ -1,12 +1,13 @@
|
||||
/*
|
||||
* @Author: ZhaoYing
|
||||
* @Date: 2026-02-03 18:31:50
|
||||
* @Last Modified by: ZhaoYing
|
||||
* @Last Modified time: 2026-02-03 18:31:50
|
||||
* @Last Modified by: ZhaoYing
|
||||
* @Last Modified time: 2026-03-04 16:22:03
|
||||
*/
|
||||
import { useEffect, useState, forwardRef, useImperativeHandle } from 'react'
|
||||
import { useEffect, useState, useRef, forwardRef, useImperativeHandle } from 'react'
|
||||
import { useTranslation } from 'react-i18next'
|
||||
import { useParams } from 'react-router-dom'
|
||||
import { App } from 'antd'
|
||||
|
||||
import Empty from '@/components/Empty'
|
||||
import RbCard from '@/components/RbCard/Card'
|
||||
@@ -20,6 +21,7 @@ import RbAlert from '@/components/RbAlert'
|
||||
* @property {Array} suggestions - List of suggestions with actionable steps
|
||||
*/
|
||||
interface Suggestions {
|
||||
exists?: boolean;
|
||||
health_summary: string;
|
||||
suggestions: Array<{
|
||||
type: string;
|
||||
@@ -35,14 +37,17 @@ interface Suggestions {
|
||||
* Displays emotional health suggestions with actionable steps
|
||||
* Shows health summary and prioritized recommendations
|
||||
*/
|
||||
const Suggestions = forwardRef<{ handleRefresh: () => void; }>((_props, ref) => {
|
||||
const Suggestions = forwardRef<{ handleRefresh: () => void; }, { refresh: () => void; }>(({ refresh }, ref) => {
|
||||
const { t } = useTranslation()
|
||||
const { id } = useParams()
|
||||
const { modal } = App.useApp()
|
||||
const [loading, setLoading] = useState(false)
|
||||
const [suggestions, setSuggestions] = useState<Suggestions | null>(null)
|
||||
const modalInstanceRef = useRef<{ destroy: () => void } | null>(null)
|
||||
|
||||
useEffect(() => {
|
||||
getSuggestionData()
|
||||
return () => modalInstanceRef.current?.destroy()
|
||||
}, [id])
|
||||
|
||||
const getSuggestionData = () => {
|
||||
@@ -52,7 +57,18 @@ const Suggestions = forwardRef<{ handleRefresh: () => void; }>((_props, ref) =>
|
||||
setLoading(true)
|
||||
getEmotionSuggestions(id)
|
||||
.then((res) => {
|
||||
setSuggestions(res as Suggestions)
|
||||
const response = res as Suggestions
|
||||
if (!response.exists && (!response.suggestions || !response.suggestions?.length)) {
|
||||
modalInstanceRef.current = modal.warning({
|
||||
title: t('statementDetail.noData'),
|
||||
okText: t('common.refresh'),
|
||||
onOk: () => {
|
||||
refresh()
|
||||
}
|
||||
})
|
||||
} else {
|
||||
setSuggestions(res as Suggestions)
|
||||
}
|
||||
})
|
||||
.finally(() => {
|
||||
setLoading(false)
|
||||
|
||||
@@ -1,6 +1,12 @@
|
||||
import { forwardRef, useImperativeHandle, useRef } from 'react'
|
||||
/*
|
||||
* @Author: ZhaoYing
|
||||
* @Date: 2026-01-08 19:46:02
|
||||
* @Last Modified by: ZhaoYing
|
||||
* @Last Modified time: 2026-03-04 16:26:55
|
||||
*/
|
||||
import { forwardRef, useImperativeHandle, useRef, useEffect } from 'react'
|
||||
import { useTranslation } from 'react-i18next'
|
||||
import { Row, Col } from 'antd'
|
||||
import { Row, Col, App } from 'antd'
|
||||
import { useParams } from 'react-router-dom'
|
||||
|
||||
import Preferences from '../components/Preferences'
|
||||
@@ -9,16 +15,44 @@ import InterestAreas from '../components/InterestAreas'
|
||||
import Habits from '../components/Habits'
|
||||
import {
|
||||
generateProfile,
|
||||
implicitCheckData,
|
||||
} from '@/api/memory'
|
||||
|
||||
const ImplicitDetail = forwardRef<{ handleRefresh: () => void; }>((_props, ref) => {
|
||||
/**
|
||||
* ImplicitDetail Component - Displays user's implicit memory profile
|
||||
* Shows unconscious preferences, personality traits, interests and habits
|
||||
*/
|
||||
const ImplicitDetail = forwardRef<{ handleRefresh: () => void; }, { refresh: () => void; }>(({
|
||||
refresh
|
||||
}, ref) => {
|
||||
const { t } = useTranslation()
|
||||
const { id } = useParams()
|
||||
const { modal } = App.useApp()
|
||||
const preferencesRef = useRef<{ handleRefresh: () => void; }>(null)
|
||||
const portraitRef = useRef<{ handleRefresh: () => void; }>(null)
|
||||
const interestAreasRef = useRef<{ handleRefresh: () => void; }>(null)
|
||||
const habitsRef = useRef<{ handleRefresh: () => void; }>(null)
|
||||
|
||||
// Check if implicit data exists, prompt user to initialize if not
|
||||
useEffect(() => {
|
||||
if (!id) return
|
||||
let modalInstance: { destroy: () => void } | null = null
|
||||
implicitCheckData(id)
|
||||
.then(res => {
|
||||
if (!(res as { exists: boolean }).exists) {
|
||||
modalInstance = modal.warning({
|
||||
title: t('implicitDetail.noData'),
|
||||
okText: t('common.refresh'),
|
||||
onOk: () => {
|
||||
refresh()
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
return () => modalInstance?.destroy()
|
||||
}, [id])
|
||||
|
||||
// Refresh all implicit memory components by regenerating profile
|
||||
const handleRefresh = () => {
|
||||
if (!id) {
|
||||
return Promise.resolve()
|
||||
|
||||
@@ -1,3 +1,9 @@
|
||||
/*
|
||||
* @Author: ZhaoYing
|
||||
* @Date: 2025-12-19 16:54:52
|
||||
* @Last Modified by: ZhaoYing
|
||||
* @Last Modified time: 2026-03-04 16:28:00
|
||||
*/
|
||||
import { forwardRef, useImperativeHandle, useRef } from 'react'
|
||||
import { Row, Col, Space } from 'antd';
|
||||
import { useParams } from 'react-router-dom'
|
||||
@@ -9,9 +15,17 @@ import Suggestions from '../components/Suggestions'
|
||||
import { generateSuggestions } from '@/api/memory'
|
||||
|
||||
|
||||
const StatementDetail = forwardRef((_props, ref) => {
|
||||
/**
|
||||
* StatementDetail - Displays emotional memory analysis for a user
|
||||
* Shows word cloud, emotion tags, health index, and personalized suggestions
|
||||
*/
|
||||
const StatementDetail = forwardRef<{ handleRefresh: () => void },{ refresh: () => void; }>(({
|
||||
refresh
|
||||
}, ref) => {
|
||||
const { id } = useParams()
|
||||
const suggestionsRef = useRef<{ handleRefresh: () => void; }>(null)
|
||||
|
||||
// Regenerate suggestions and refresh the Suggestions child component
|
||||
const handleRefresh = () => {
|
||||
if (!id) {
|
||||
return Promise.resolve()
|
||||
@@ -41,7 +55,7 @@ const StatementDetail = forwardRef((_props, ref) => {
|
||||
</Space>
|
||||
</Col>
|
||||
<Col span={12}>
|
||||
<Suggestions ref={suggestionsRef} />
|
||||
<Suggestions ref={suggestionsRef} refresh={refresh} />
|
||||
</Col>
|
||||
</Row>
|
||||
)
|
||||
|
||||
@@ -1,8 +1,13 @@
|
||||
/*
|
||||
* @Author: ZhaoYing
|
||||
* @Date: 2026-01-07 20:37:34
|
||||
* @Last Modified by: ZhaoYing
|
||||
* @Last Modified time: 2026-03-04 16:27:14
|
||||
*/
|
||||
import { type FC, useEffect, useState, useMemo, useRef } from 'react'
|
||||
import { useParams, useNavigate } from 'react-router-dom'
|
||||
import { useTranslation } from 'react-i18next'
|
||||
import { Dropdown, Button } from 'antd'
|
||||
import { LoadingOutlined } from '@ant-design/icons';
|
||||
|
||||
import PageHeader from '../components/PageHeader'
|
||||
import StatementDetail from './StatementDetail'
|
||||
@@ -19,11 +24,16 @@ import {
|
||||
import refreshIcon from '@/assets/images/refresh_hover.svg'
|
||||
import GraphDetail from './GraphDetail'
|
||||
|
||||
/**
|
||||
* Detail page for user memory - renders different memory type views
|
||||
* based on the `type` route param
|
||||
*/
|
||||
const Detail: FC = () => {
|
||||
const { t } = useTranslation()
|
||||
const { id, type } = useParams()
|
||||
const navigate = useNavigate()
|
||||
const [name, setName] = useState<string>('')
|
||||
// Refs for child components that support imperative refresh
|
||||
const forgetDetailRef = useRef<{ handleRefresh: () => void }>(null)
|
||||
const statementDetailRef = useRef<{ handleRefresh: () => void }>(null)
|
||||
const implicitDetailRef = useRef<{ handleRefresh: () => void }>(null)
|
||||
@@ -33,6 +43,7 @@ const Detail: FC = () => {
|
||||
getData()
|
||||
}, [id])
|
||||
|
||||
// Fetch end user profile to display the user's name in the header
|
||||
const getData = () => {
|
||||
if (!id) return
|
||||
getEndUserProfile(id).then((res) => {
|
||||
@@ -40,15 +51,21 @@ const Detail: FC = () => {
|
||||
setName(response.other_name || response.id)
|
||||
})
|
||||
}
|
||||
|
||||
// Build dropdown menu items for switching between memory types
|
||||
const items = useMemo(() => {
|
||||
return ['PERCEPTUAL_MEMORY', 'WORKING_MEMORY', 'EMOTIONAL_MEMORY', 'SHORT_TERM_MEMORY', 'IMPLICIT_MEMORY', 'EPISODIC_MEMORY', 'EXPLICIT_MEMORY', 'FORGET_MEMORY']
|
||||
.map(key => ({ key, label: t(`userMemory.${key}`) }))
|
||||
}, [t])
|
||||
|
||||
// Navigate to the selected memory type detail page
|
||||
const onClick = ({ key }: { key: string }) => {
|
||||
navigate(`/user-memory/detail/${id}/${key}`, { replace: true })
|
||||
}
|
||||
|
||||
const [loading, setLoading] = useState(false)
|
||||
|
||||
// Trigger refresh on the active memory type's child component
|
||||
const handleRefresh = () => {
|
||||
setLoading(true)
|
||||
let response: any = null
|
||||
@@ -64,6 +81,7 @@ const Detail: FC = () => {
|
||||
break
|
||||
}
|
||||
|
||||
// If the child returns a Promise, wait for it before clearing loading state
|
||||
if (response instanceof Promise) {
|
||||
response.finally(() => {
|
||||
setLoading(false)
|
||||
@@ -99,9 +117,9 @@ const Detail: FC = () => {
|
||||
</Button>}
|
||||
/>
|
||||
<div className="rb:h-[calc(100vh-64px)] rb:overflow-y-auto rb:py-3 rb:px-4">
|
||||
{type === 'EMOTIONAL_MEMORY' && <StatementDetail ref={statementDetailRef} />}
|
||||
{type === 'EMOTIONAL_MEMORY' && <StatementDetail ref={statementDetailRef} refresh={handleRefresh} />}
|
||||
{type === 'FORGET_MEMORY' && <ForgetDetail ref={forgetDetailRef} />}
|
||||
{type === 'IMPLICIT_MEMORY' && <ImplicitDetail ref={implicitDetailRef} />}
|
||||
{type === 'IMPLICIT_MEMORY' && <ImplicitDetail ref={implicitDetailRef} refresh={handleRefresh} />}
|
||||
{type === 'SHORT_TERM_MEMORY' && <ShortTermDetail />}
|
||||
{type === 'PERCEPTUAL_MEMORY' && <PerceptualDetail />}
|
||||
{type === 'EPISODIC_MEMORY' && <EpisodicDetail />}
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
* @Author: ZhaoYing
|
||||
* @Date: 2026-02-06 21:10:56
|
||||
* @Last Modified by: ZhaoYing
|
||||
* @Last Modified time: 2026-02-27 09:58:30
|
||||
* @Last Modified time: 2026-02-24 17:55:08
|
||||
*/
|
||||
/**
|
||||
* Workflow Chat Component
|
||||
|
||||
Reference in New Issue
Block a user