Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 11 additions & 18 deletions backend/apps/me_model_managment_app.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import logging
from http import HTTPStatus

from fastapi import APIRouter, Query
from fastapi import APIRouter, Query, HTTPException
from fastapi.responses import JSONResponse

from consts.exceptions import TimeoutException, NotFoundException, MEConnectionException
Expand Down Expand Up @@ -44,9 +44,9 @@ async def get_me_models(
"data": []
})
except Exception as e:
logging.error(f"Failed to get model list: {str(e)}")
logging.error(f"Failed to get me model list: {str(e)}")
return JSONResponse(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, content={
"message": f"Failed to get model list: {str(e)}",
"message": f"Failed to get me model list: {str(e)}",
"data": []
})

Expand All @@ -61,23 +61,16 @@ async def check_me_connectivity(timeout: int = Query(default=2, description="Tim
return JSONResponse(
status_code=HTTPStatus.OK,
content={
"status": "Connected",
"desc": "Connection successful.",
"connect_status": ModelConnectStatusEnum.AVAILABLE.value
"connectivity": True,
"message": "ModelEngine model connect successfully.",
}
)
except MEConnectionException as e:
logging.error(f"Request me model connectivity failed: {str(e)}")
return JSONResponse(status_code=HTTPStatus.SERVICE_UNAVAILABLE, content={"status": "Disconnected",
"desc": f"Connection failed.",
"connect_status": ModelConnectStatusEnum.UNAVAILABLE.value})
logging.error(f"ModelEngine model healthcheck failed: {str(e)}")
raise HTTPException(status_code=HTTPStatus.SERVICE_UNAVAILABLE, detail="ModelEngine model connect failed.")
except TimeoutException as e:
logging.error(f"Request me model connectivity timeout: {str(e)}")
return JSONResponse(status_code=HTTPStatus.REQUEST_TIMEOUT, content={"status": "Disconnected",
"desc": "Connection timeout.",
"connect_status": ModelConnectStatusEnum.UNAVAILABLE.value})
logging.error(f"ModelEngine model healthcheck timeout: {str(e)}")
raise HTTPException(status_code=HTTPStatus.REQUEST_TIMEOUT, detail="ModelEngine model connect timeout.")
except Exception as e:
logging.error(f"Unknown error occurred: {str(e)}.")
return JSONResponse(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, content={"status": "Disconnected",
"desc": f"Unknown error occurred: {str(e)}",
"connect_status": ModelConnectStatusEnum.UNAVAILABLE.value})
logging.error(f"ModelEngine model healthcheck failed with unknown error: {str(e)}.")
raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="ModelEngine model connect failed.")
181 changes: 142 additions & 39 deletions backend/apps/memory_config_app.py
Original file line number Diff line number Diff line change
@@ -1,22 +1,45 @@
"""Memory configuration and CRUD API endpoints for the app layer.

This module exposes HTTP endpoints under the `/memory` prefix. It follows the
app-layer responsibilities:
- Parse and validate HTTP inputs
- Delegate business logic to the service layer
- Convert unexpected exceptions to error JSON responses

Routes:
- GET `/memory/config/load`: Load memory-related configuration for current user
- POST `/memory/config/set`: Set a single configuration entry
- POST `/memory/config/disable_agent`: Add a disabled agent id
- DELETE `/memory/config/disable_agent/{agent_id}`: Remove a disabled agent id
- POST `/memory/config/disable_useragent`: Add a disabled user-agent id
- DELETE `/memory/config/disable_useragent/{agent_id}`: Remove a disabled user-agent id
- POST `/memory/add`: Add memory items (optionally with LLM inference)
- POST `/memory/search`: Semantic search memory items
- GET `/memory/list`: List memory items
- DELETE `/memory/delete/{memory_id}`: Delete a single memory item
- DELETE `/memory/clear`: Clear memory items by scope
"""
import asyncio
import logging
from typing import Any, Dict, List, Optional

from fastapi import APIRouter, Body, Header, Path, Query
from http import HTTPStatus
from fastapi import APIRouter, Body, Header, Path, Query, HTTPException
from fastapi.responses import JSONResponse

from nexent.memory.memory_service import (
add_memory as svc_add_memory,
clear_memory as svc_clear_memory,
delete_memory as svc_delete_memory,
list_memory as svc_list_memory,
search_memory as svc_search_memory,
)

from consts.const import (
MEMORY_AGENT_SHARE_KEY,
MEMORY_SWITCH_KEY,
)
from consts.model import MemoryAgentShareMode
from consts.exceptions import UnauthorizedError
from services.memory_config_service import (
add_disabled_agent_id,
add_disabled_useragent_id,
Expand All @@ -34,35 +57,26 @@
router = APIRouter(prefix="/memory")


# ---------------------------------------------------------------------------
# Generic helpers
# ---------------------------------------------------------------------------
def _success(message: str = "success", content: Optional[Any] = None):
return JSONResponse(status_code=200, content={"message": message, "status": "success", "content": content})


def _error(message: str = "error"):
return JSONResponse(status_code=400, content={"message": message, "status": "error"})


# ---------------------------------------------------------------------------
# Helper function
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# Configuration Endpoints
# ---------------------------------------------------------------------------
@router.get("/config/load")
def load_configs(authorization: Optional[str] = Header(None)):
"""Load all memory-related configuration for current user."""
"""Load all memory-related configuration for the current user.

Args:
authorization: Optional authorization header used to identify the user.
"""
try:
user_id, _ = get_current_user_id(authorization)
configs = get_user_configs(user_id)
return _success(content=configs)
return JSONResponse(status_code=HTTPStatus.OK, content=configs)
except UnauthorizedError as e:
raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail=str(e))
except Exception as e:
logger.error("load_configs failed: %s", e)
return _error("Failed to load configuration")
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST,
detail="Failed to load configuration")


@router.post("/config/set")
Expand All @@ -71,7 +85,17 @@ def set_single_config(
value: Any = Body(..., embed=True, description="Configuration value"),
authorization: Optional[str] = Header(None),
):
"""Unified endpoint to set single-value configuration items."""
"""Set a single-value configuration item for the current user.

Supported keys:
- `MEMORY_SWITCH_KEY`: Toggle memory system on/off (boolean-like values accepted)
- `MEMORY_AGENT_SHARE_KEY`: Set agent share mode (`always`/`ask`/`never`)

Args:
key: Configuration key to update.
value: New value for the configuration key.
authorization: Optional authorization header used to identify the user.
"""
user_id, _ = get_current_user_id(authorization)

if key == MEMORY_SWITCH_KEY:
Expand All @@ -82,52 +106,93 @@ def set_single_config(
try:
mode = MemoryAgentShareMode(str(value))
except ValueError:
return _error("Invalid value for MEMORY_AGENT_SHARE (expected always/ask/never)")
raise HTTPException(status_code=HTTPStatus.NOT_ACCEPTABLE,
detail="Invalid value for MEMORY_AGENT_SHARE (expected always/ask/never)")
ok = set_agent_share(user_id, mode)
else:
return _error("Unsupported configuration key")
raise HTTPException(status_code=HTTPStatus.NOT_ACCEPTABLE,
detail="Unsupported configuration key")

return _success() if ok else _error("Failed to update configuration")
if ok:
return JSONResponse(status_code=HTTPStatus.OK, content={"success": True})
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST,
detail="Failed to update configuration")


@router.post("/config/disable_agent")
def add_disable_agent(
agent_id: str = Body(..., embed=True),
authorization: Optional[str] = Header(None),
):
"""Add an agent id to the user's disabled agent list.

Args:
agent_id: Identifier of the agent to disable.
authorization: Optional authorization header used to identify the user.
"""
user_id, _ = get_current_user_id(authorization)
ok = add_disabled_agent_id(user_id, agent_id)
return _success() if ok else _error("Failed to add disable agent id")
if ok:
return JSONResponse(status_code=HTTPStatus.OK, content={"success": True})
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST,
detail="Failed to add disable agent id")


@router.delete("/config/disable_agent/{agent_id}")
def remove_disable_agent(
agent_id: str = Path(...),
authorization: Optional[str] = Header(None),
):
"""Remove an agent id from the user's disabled agent list.

Args:
agent_id: Identifier of the agent to remove from the disabled list.
authorization: Optional authorization header used to identify the user.
"""
user_id, _ = get_current_user_id(authorization)
ok = remove_disabled_agent_id(user_id, agent_id)
return _success() if ok else _error("Failed to remove disable agent id")
if ok:
return JSONResponse(status_code=HTTPStatus.OK, content={"success": True})
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST,
detail="Failed to remove disable agent id")


@router.post("/config/disable_useragent")
def add_disable_useragent(
agent_id: str = Body(..., embed=True),
authorization: Optional[str] = Header(None),
):
"""Add a user-agent id to the user's disabled user-agent list.

Args:
agent_id: Identifier of the user-agent to disable.
authorization: Optional authorization header used to identify the user.
"""
user_id, _ = get_current_user_id(authorization)
ok = add_disabled_useragent_id(user_id, agent_id)
return _success() if ok else _error("Failed to add disable user-agent id")
if ok:
return JSONResponse(status_code=HTTPStatus.OK, content={"success": True})
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST,
detail="Failed to add disable user-agent id")


@router.delete("/config/disable_useragent/{agent_id}")
def remove_disable_useragent(
agent_id: str = Path(...),
authorization: Optional[str] = Header(None),
):
"""Remove a user-agent id from the user's disabled user-agent list.

Args:
agent_id: Identifier of the user-agent to remove from the disabled list.
authorization: Optional authorization header used to identify the user.
"""
user_id, _ = get_current_user_id(authorization)
ok = remove_disabled_useragent_id(user_id, agent_id)
return _success() if ok else _error("Failed to remove disable user-agent id")
if ok:
return JSONResponse(status_code=HTTPStatus.OK, content={"success": True})
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST,
detail="Failed to remove disable user-agent id")


# ---------------------------------------------------------------------------
Expand All @@ -144,6 +209,15 @@ def add_memory(
True, embed=True, description="Whether to run LLM inference during add"),
authorization: Optional[str] = Header(None),
):
"""Add memory records for the given scope.

Args:
messages: List of chat messages as dictionaries.
memory_level: Scope for the memory record (tenant/agent/user/user_agent).
agent_id: Optional agent identifier when scope is agent-related.
infer: Whether to run LLM inference during add.
authorization: Optional authorization header used to identify the user.
"""
user_id, tenant_id = get_current_user_id(authorization)
try:
result = asyncio.run(svc_add_memory(
Expand All @@ -155,10 +229,10 @@ def add_memory(
agent_id=agent_id,
infer=infer,
))
return _success(content=result)
return JSONResponse(status_code=HTTPStatus.OK, content=result)
except Exception as e:
logger.error("add_memory error: %s", e, exc_info=True)
return _error(str(e))
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e))


@router.post("/search")
Expand All @@ -169,6 +243,15 @@ def search_memory(
agent_id: Optional[str] = Body(None, embed=True),
authorization: Optional[str] = Header(None),
):
"""Search memory semantically for the given scope.

Args:
query_text: Natural language query to search memory.
memory_level: Scope for search (tenant/agent/user/user_agent).
top_k: Maximum number of results to return.
agent_id: Optional agent identifier when scope is agent-related.
authorization: Optional authorization header used to identify the user.
"""
user_id, tenant_id = get_current_user_id(authorization)
try:
results = asyncio.run(svc_search_memory(
Expand All @@ -180,10 +263,10 @@ def search_memory(
top_k=top_k,
agent_id=agent_id,
))
return _success(content=results)
return JSONResponse(status_code=HTTPStatus.OK, content=results)
except Exception as e:
logger.error("search_memory error: %s", e, exc_info=True)
return _error(str(e))
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e))


@router.get("/list")
Expand All @@ -194,6 +277,13 @@ def list_memory(
None, description="Filter by agent id if applicable"),
authorization: Optional[str] = Header(None),
):
"""List memory for the given scope.

Args:
memory_level: Scope for listing (tenant/agent/user/user_agent).
agent_id: Optional agent filter when scope is agent-related.
authorization: Optional authorization header used to identify the user.
"""
user_id, tenant_id = get_current_user_id(authorization)
try:
payload = asyncio.run(svc_list_memory(
Expand All @@ -203,25 +293,31 @@ def list_memory(
user_id=user_id,
agent_id=agent_id,
))
return _success(content=payload)
return JSONResponse(status_code=HTTPStatus.OK, content=payload)
except Exception as e:
logger.error("list_memory error: %s", e, exc_info=True)
return _error(str(e))
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e))


@router.delete("/delete/{memory_id}")
def delete_memory(
memory_id: str = Path(..., description="ID of memory to delete"),
authorization: Optional[str] = Header(None),
):
"""Delete a specific memory record by id.

Args:
memory_id: Identifier of the memory record to delete.
authorization: Optional authorization header used to identify the user.
"""
_user_id, tenant_id = get_current_user_id(authorization)
try:
result = asyncio.run(svc_delete_memory(
memory_id=memory_id, memory_config=build_memory_config(tenant_id)))
return _success(content=result)
return JSONResponse(status_code=HTTPStatus.OK, content=result)
except Exception as e:
logger.error("delete_memory error: %s", e, exc_info=True)
return _error(str(e))
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e))


@router.delete("/clear")
Expand All @@ -232,6 +328,13 @@ def clear_memory(
None, description="Filter by agent id if applicable"),
authorization: Optional[str] = Header(None),
):
"""Clear memory records for the given scope.

Args:
memory_level: Scope for clearing (tenant/agent/user/user_agent).
agent_id: Optional agent filter when scope is agent-related.
authorization: Optional authorization header used to identify the user.
"""
user_id, tenant_id = get_current_user_id(authorization)
try:
result = asyncio.run(svc_clear_memory(
Expand All @@ -241,7 +344,7 @@ def clear_memory(
user_id=user_id,
agent_id=agent_id,
))
return _success(content=result)
return JSONResponse(status_code=HTTPStatus.OK, content=result)
except Exception as e:
logger.error("clear_memory error: %s", e, exc_info=True)
return _error(str(e))
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e))
Loading