diff --git a/api/app/controllers/service/__init__.py b/api/app/controllers/service/__init__.py index 52d4b732..850b496d 100644 --- a/api/app/controllers/service/__init__.py +++ b/api/app/controllers/service/__init__.py @@ -14,6 +14,7 @@ from . import ( rag_api_document_controller, rag_api_file_controller, rag_api_knowledge_controller, + user_memory_api_controller, ) # 创建 V1 API 路由器 @@ -28,5 +29,6 @@ service_router.include_router(rag_api_chunk_controller.router) service_router.include_router(memory_api_controller.router) service_router.include_router(end_user_api_controller.router) service_router.include_router(memory_config_api_controller.router) +service_router.include_router(user_memory_api_controller.router) __all__ = ["service_router"] diff --git a/api/app/controllers/service/user_memory_api_controller.py b/api/app/controllers/service/user_memory_api_controller.py new file mode 100644 index 00000000..6d2498c0 --- /dev/null +++ b/api/app/controllers/service/user_memory_api_controller.py @@ -0,0 +1,197 @@ +"""User Memory 服务接口 — 基于 API Key 认证 + +包装 user_memory_controllers.py 和 memory_agent_controller.py 中的内部接口， +提供基于 API Key 认证的对外服务: +1./analytics/graph_data - 知识图谱数据接口 +2./analytics/community_graph - 社区图谱接口 +3./analytics/node_statistics - 记忆节点统计接口 +4./analytics/user_summary - 用户摘要接口 +5./analytics/memory_insight - 记忆洞察接口 +6./analytics/interest_distribution - 兴趣分布接口 +7./analytics/end_user_info - 终端用户信息接口 + + +路由前缀: /memory +子路径: /analytics/... +最终路径: /v1/memory/analytics/... +认证方式: API Key (@require_api_key) +""" + +from typing import Optional + +from fastapi import APIRouter, Depends, Header, Query, Request +from sqlalchemy.orm import Session + +from app.core.api_key_auth import require_api_key +from app.core.api_key_utils import get_current_user_from_api_key, validate_end_user_in_workspace +from app.core.logging_config import get_business_logger +from app.db import get_db +from app.schemas.api_key_schema import ApiKeyAuth + +# 包装内部服务 controller +from app.controllers import user_memory_controllers, memory_agent_controller + +router = APIRouter(prefix="/memory", tags=["V1 - User Memory API"]) +logger = get_business_logger() + + +# ==================== 知识图谱 ==================== + + +@router.get("/analytics/graph_data") +@require_api_key(scopes=["memory"]) +async def get_graph_data( + request: Request, + end_user_id: str = Query(..., description="End user ID"), + node_types: Optional[str] = Query(None, description="Comma-separated node types filter"), + limit: int = Query(100, description="Max nodes to return, capped at 1000"), + depth: int = Query(1, description="Graph traversal depth, capped at 3"), + center_node_id: Optional[str] = Query(None, description="Center node for subgraph"), + api_key_auth: ApiKeyAuth = None, + db: Session = Depends(get_db), +): + """Get knowledge graph data (nodes + edges) for an end user.""" + current_user = get_current_user_from_api_key(db, api_key_auth) + validate_end_user_in_workspace(db, end_user_id, api_key_auth.workspace_id) + + return await user_memory_controllers.get_graph_data_api( + end_user_id=end_user_id, + node_types=node_types, + limit=limit, + depth=depth, + center_node_id=center_node_id, + current_user=current_user, + db=db, + ) + + +@router.get("/analytics/community_graph") +@require_api_key(scopes=["memory"]) +async def get_community_graph( + request: Request, + end_user_id: str = Query(..., description="End user ID"), + api_key_auth: ApiKeyAuth = None, + db: Session = Depends(get_db), +): + """Get community clustering graph for an end user.""" + current_user = get_current_user_from_api_key(db, api_key_auth) + validate_end_user_in_workspace(db, end_user_id, api_key_auth.workspace_id) + + return await user_memory_controllers.get_community_graph_data_api( + end_user_id=end_user_id, + current_user=current_user, + db=db, + ) + + +# ==================== 节点统计 ==================== + + +@router.get("/analytics/node_statistics") +@require_api_key(scopes=["memory"]) +async def get_node_statistics( + request: Request, + end_user_id: str = Query(..., description="End user ID"), + api_key_auth: ApiKeyAuth = None, + db: Session = Depends(get_db), +): + """Get memory node type statistics for an end user.""" + current_user = get_current_user_from_api_key(db, api_key_auth) + validate_end_user_in_workspace(db, end_user_id, api_key_auth.workspace_id) + + return await user_memory_controllers.get_node_statistics_api( + end_user_id=end_user_id, + current_user=current_user, + db=db, + ) + + +# ==================== 用户摘要 & 洞察 ==================== + + +@router.get("/analytics/user_summary") +@require_api_key(scopes=["memory"]) +async def get_user_summary( + request: Request, + end_user_id: str = Query(..., description="End user ID"), + language_type: str = Header(default=None, alias="X-Language-Type"), + api_key_auth: ApiKeyAuth = None, + db: Session = Depends(get_db), +): + """Get cached user summary for an end user.""" + current_user = get_current_user_from_api_key(db, api_key_auth) + validate_end_user_in_workspace(db, end_user_id, api_key_auth.workspace_id) + + return await user_memory_controllers.get_user_summary_api( + end_user_id=end_user_id, + language_type=language_type, + current_user=current_user, + db=db, + ) + + +@router.get("/analytics/memory_insight") +@require_api_key(scopes=["memory"]) +async def get_memory_insight( + request: Request, + end_user_id: str = Query(..., description="End user ID"), + api_key_auth: ApiKeyAuth = None, + db: Session = Depends(get_db), +): + """Get cached memory insight report for an end user.""" + current_user = get_current_user_from_api_key(db, api_key_auth) + validate_end_user_in_workspace(db, end_user_id, api_key_auth.workspace_id) + + return await user_memory_controllers.get_memory_insight_report_api( + end_user_id=end_user_id, + current_user=current_user, + db=db, + ) + + +# ==================== 兴趣分布 ==================== + + +@router.get("/analytics/interest_distribution") +@require_api_key(scopes=["memory"]) +async def get_interest_distribution( + request: Request, + end_user_id: str = Query(..., description="End user ID"), + limit: int = Query(5, le=5, description="Max interest tags to return"), + language_type: str = Header(default=None, alias="X-Language-Type"), + api_key_auth: ApiKeyAuth = None, + db: Session = Depends(get_db), +): + """Get interest distribution tags for an end user.""" + current_user = get_current_user_from_api_key(db, api_key_auth) + validate_end_user_in_workspace(db, end_user_id, api_key_auth.workspace_id) + + return await memory_agent_controller.get_interest_distribution_by_user_api( + end_user_id=end_user_id, + limit=limit, + language_type=language_type, + current_user=current_user, + db=db, + ) + + +# ==================== 终端用户信息 ==================== + + +@router.get("/analytics/end_user_info") +@require_api_key(scopes=["memory"]) +async def get_end_user_info( + request: Request, + end_user_id: str = Query(..., description="End user ID"), + api_key_auth: ApiKeyAuth = None, + db: Session = Depends(get_db), +): + """Get end user basic information (name, aliases, metadata).""" + current_user = get_current_user_from_api_key(db, api_key_auth) + validate_end_user_in_workspace(db, end_user_id, api_key_auth.workspace_id) + + return await user_memory_controllers.get_end_user_info( + end_user_id=end_user_id, + current_user=current_user, + db=db, + ) \ No newline at end of file diff --git a/api/app/core/api_key_utils.py b/api/app/core/api_key_utils.py index fb6b9552..290630ce 100644 --- a/api/app/core/api_key_utils.py +++ b/api/app/core/api_key_utils.py @@ -3,6 +3,12 @@ import secrets from typing import Optional, Union from datetime import datetime +from sqlalchemy.orm import Session as _Session +from app.core.error_codes import BizCode as _BizCode +from app.core.exceptions import BusinessException as _BusinessException +from app.models.end_user_model import EndUser as _EndUser +from app.repositories.end_user_repository import EndUserRepository as _EndUserRepository + from app.models.api_key_model import ApiKeyType from fastapi import Response from fastapi.responses import JSONResponse @@ -65,3 +71,63 @@ def datetime_to_timestamp(dt: Optional[datetime]) -> Optional[int]: return None return int(dt.timestamp() * 1000) + + +def get_current_user_from_api_key(db: _Session, api_key_auth): + """通过 API Key 构造 current_user 对象。 + + 从 API Key 反查创建者（管理员用户），并设置其 workspace 上下文。 + 与内部接口的 Depends(get_current_user) (JWT) 等价。 + + Args: + db: 数据库会话 + api_key_auth: API Key 认证信息（ApiKeyAuth） + + Returns: + User ORM 对象，已设置 current_workspace_id + """ + from app.services import api_key_service + + api_key = api_key_service.ApiKeyService.get_api_key( + db, api_key_auth.api_key_id, api_key_auth.workspace_id + ) + current_user = api_key.creator + current_user.current_workspace_id = api_key_auth.workspace_id + return current_user + + +def validate_end_user_in_workspace( + db: _Session, + end_user_id: str, + workspace_id, +) -> _EndUser: + """校验 end_user 是否存在且属于指定 workspace。 + + Args: + db: 数据库会话 + end_user_id: 终端用户 ID + workspace_id: 工作空间 ID（UUID 或字符串均可） + + Returns: + EndUser ORM 对象（校验通过时） + + Raises: + BusinessException(USER_NOT_FOUND): end_user 不存在 + BusinessException(PERMISSION_DENIED): end_user 不属于该 workspace + """ + end_user_repo = _EndUserRepository(db) + end_user = end_user_repo.get_end_user_by_id(end_user_id) + + if end_user is None: + raise _BusinessException( + "End user not found", + _BizCode.USER_NOT_FOUND, + ) + + if str(end_user.workspace_id) != str(workspace_id): + raise _BusinessException( + "End user does not belong to this workspace", + _BizCode.PERMISSION_DENIED, + ) + + return end_user \ No newline at end of file