Merge branch 'dev' into dev

Author: CaralHsi

docker/requirements.txt

@@ -157,4 +157,4 @@ volcengine-python-sdk==4.0.6
 watchfiles==1.1.0
 websockets==15.0.1
 xlrd==2.0.2
-xlsxwriter==3.2.5
+xlsxwriter==3.2.5

evaluation/scripts/PrefEval/pref_memos.py

@@ -53,9 +53,9 @@ def add_memory_for_line(
             if os.getenv("PRE_SPLIT_CHUNK", "false").lower() == "true":
                 for chunk_start in range(0, len(conversation), turns_add * 2):
                     chunk = conversation[chunk_start : chunk_start + turns_add * 2]
-                    mem_client.add(messages=chunk, user_id=user_id, conv_id=None)
+                    mem_client.add(messages=chunk, user_id=user_id, conv_id=None, batch_size=2)
             else:
-                mem_client.add(messages=conversation, user_id=user_id, conv_id=None)
+                mem_client.add(messages=conversation, user_id=user_id, conv_id=None, batch_size=2)
         end_time_add = time.monotonic()
         add_duration = end_time_add - start_time_add
 
@@ -98,7 +98,7 @@ def search_memory_for_line(line_data: tuple, mem_client, top_k_value: int) -> di
                 f"- {entry.get('memory', '')}"
                 for entry in relevant_memories["text_mem"][0]["memories"]
             )
-            + f"\n{relevant_memories['pref_mem']}"
+            + f"\n{relevant_memories['pref_string']}"
         )
 
         memory_tokens_used = len(tokenizer.encode(memories_str))

evaluation/scripts/locomo/locomo_search.py

@@ -107,11 +107,11 @@ def memos_api_search(
 
     speaker_a_context = (
         "\n".join([i["memory"] for i in search_a_results["text_mem"][0]["memories"]])
-        + f"\n{search_a_results['pref_mem']}"
+        + f"\n{search_a_results['pref_string']}"
     )
     speaker_b_context = (
         "\n".join([i["memory"] for i in search_b_results["text_mem"][0]["memories"]])
-        + f"\n{search_b_results['pref_mem']}"
+        + f"\n{search_b_results['pref_string']}"
     )
 
     context = TEMPLATE_MEMOS.format(

evaluation/scripts/longmemeval/lme_search.py

@@ -46,7 +46,7 @@ def memos_search(client, query, user_id, top_k):
     results = client.search(query=query, user_id=user_id, top_k=top_k)
     context = (
         "\n".join([i["memory"] for i in results["text_mem"][0]["memories"]])
-        + f"\n{results['pref_mem']}"
+        + f"\n{results['pref_string']}"
     )
     context = MEMOS_CONTEXT_TEMPLATE.format(user_id=user_id, memories=context)
     duration_ms = (time() - start) * 1000

evaluation/scripts/personamem/pm_ingestion.py

@@ -31,10 +31,10 @@ def ingest_session(session, user_id, session_id, frame, client):
         if os.getenv("PRE_SPLIT_CHUNK") == "true":
             for i in range(0, len(session), 10):
                 messages = session[i : i + 10]
-                client.add(messages=messages, user_id=user_id, conv_id=session_id)
+                client.add(messages=messages, user_id=user_id, conv_id=session_id, batch_size=2)
                 print(f"[{frame}] ✅ Session [{session_id}]: Ingested {len(messages)} messages")
         else:
-            client.add(messages=session, user_id=user_id, conv_id=session_id)
+            client.add(messages=session, user_id=user_id, conv_id=session_id, batch_size=2)
             print(f"[{frame}] ✅ Session [{session_id}]: Ingested {len(session)} messages")
     elif frame == "memobase":
         for _idx, msg in enumerate(session):

evaluation/scripts/personamem/pm_search.py

@@ -84,7 +84,7 @@ def memos_search(client, user_id, query, top_k):
     results = client.search(query=query, user_id=user_id, top_k=top_k)
     search_memories = (
         "\n".join(item["memory"] for cube in results["text_mem"] for item in cube["memories"])
-        + f"\n{results['pref_mem']}"
+        + f"\n{results['pref_string']}"
     )
     context = MEMOS_CONTEXT_TEMPLATE.format(user_id=user_id, memories=search_memories)
 

evaluation/scripts/utils/client.py

@@ -182,6 +182,7 @@ def search(self, query, user_id, top_k):
                 "conversation_id": "",
                 "top_k": top_k,
                 "mode": os.getenv("SEARCH_MODE", "fast"),
+                "handle_pref_mem": False,
             },
             ensure_ascii=False,
         )

src/memos/api/config.py

@@ -23,7 +23,7 @@ def get_openai_config() -> dict[str, Any]:
         return {
             "model_name_or_path": os.getenv("MOS_CHAT_MODEL", "gpt-4o-mini"),
             "temperature": float(os.getenv("MOS_CHAT_TEMPERATURE", "0.8")),
-            "max_tokens": int(os.getenv("MOS_MAX_TOKENS", "1024")),
+            "max_tokens": int(os.getenv("MOS_MAX_TOKENS", "8000")),
             "top_p": float(os.getenv("MOS_TOP_P", "0.9")),
             "top_k": int(os.getenv("MOS_TOP_K", "50")),
             "remove_think_prefix": True,
@@ -672,6 +672,7 @@ def get_default_cube_config() -> GeneralMemCubeConfig | None:
                                 "LongTermMemory": os.getenv("NEBULAR_LONGTERM_MEMORY", 1e6),
                                 "UserMemory": os.getenv("NEBULAR_USER_MEMORY", 1e6),
                             },
+                            "mode": os.getenv("ASYNC_MODE", "sync"),
                         },
                     },
                     "act_mem": {}

src/memos/api/exceptions.py

@@ -1,5 +1,6 @@
 import logging
 
+from fastapi.exceptions import HTTPException, RequestValidationError
 from fastapi.requests import Request
 from fastapi.responses import JSONResponse
 
@@ -10,9 +11,24 @@
 class APIExceptionHandler:
     """Centralized exception handling for MemOS APIs."""
 
+    @staticmethod
+    async def validation_error_handler(request: Request, exc: RequestValidationError):
+        """Handle request validation errors."""
+        logger.error(f"Validation error: {exc.errors()}")
+        return JSONResponse(
+            status_code=422,
+            content={
+                "code": 422,
+                "message": "Parameter validation error",
+                "detail": exc.errors(),
+                "data": None,
+            },
+        )
+
     @staticmethod
     async def value_error_handler(request: Request, exc: ValueError):
         """Handle ValueError exceptions globally."""
+        logger.error(f"ValueError: {exc}")
         return JSONResponse(
             status_code=400,
             content={"code": 400, "message": str(exc), "data": None},
@@ -21,8 +37,17 @@ async def value_error_handler(request: Request, exc: ValueError):
     @staticmethod
     async def global_exception_handler(request: Request, exc: Exception):
         """Handle all unhandled exceptions globally."""
-        logger.exception("Unhandled error:")
+        logger.error(f"Exception: {exc}")
         return JSONResponse(
             status_code=500,
             content={"code": 500, "message": str(exc), "data": None},
         )
+
+    @staticmethod
+    async def http_error_handler(request: Request, exc: HTTPException):
+        """Handle HTTP exceptions globally."""
+        logger.error(f"HTTP error {exc.status_code}: {exc.detail}")
+        return JSONResponse(
+            status_code=exc.status_code,
+            content={"code": exc.status_code, "message": str(exc.detail), "data": None},
+        )

src/memos/api/middleware/request_context.py

@@ -2,6 +2,8 @@
 Request context middleware for automatic trace_id injection.
 """
 
+import time
+
 from collections.abc import Callable
 
 from starlette.middleware.base import BaseHTTPMiddleware
@@ -38,8 +40,19 @@ async def dispatch(self, request: Request, call_next: Callable) -> Response:
         # Extract or generate trace_id
         trace_id = extract_trace_id_from_headers(request) or generate_trace_id()
 
+        env = request.headers.get("x-env")
+        user_type = request.headers.get("x-user-type")
+        user_name = request.headers.get("x-user-name")
+        start_time = time.time()
+
         # Create and set request context
-        context = RequestContext(trace_id=trace_id, api_path=request.url.path)
+        context = RequestContext(
+            trace_id=trace_id,
+            api_path=request.url.path,
+            env=env,
+            user_type=user_type,
+            user_name=user_name,
+        )
         set_request_context(context)
 
         # Log request start with parameters
@@ -49,15 +62,25 @@ async def dispatch(self, request: Request, call_next: Callable) -> Response:
         if request.query_params:
             params_log["query_params"] = dict(request.query_params)
 
-        logger.info(f"Request started: {request.method} {request.url.path}, {params_log}")
+        logger.info(f"Request started, params: {params_log}, headers: {request.headers}")
 
         # Process the request
-        response = await call_next(request)
-
-        # Log request completion with output
-        logger.info(f"Request completed: {request.url.path}, status: {response.status_code}")
-
-        # Add trace_id to response headers for debugging
-        response.headers["x-trace-id"] = trace_id
+        try:
+            response = await call_next(request)
+            end_time = time.time()
+            if response.status_code == 200:
+                logger.info(
+                    f"Request completed: {request.url.path}, status: {response.status_code}, cost: {(end_time - start_time) * 1000:.2f}ms"
+                )
+            else:
+                logger.error(
+                    f"Request Failed: {request.url.path}, status: {response.status_code}, cost: {(end_time - start_time) * 1000:.2f}ms"
+                )
+        except Exception as e:
+            end_time = time.time()
+            logger.error(
+                f"Request Exception Error: {e}, cost: {(end_time - start_time) * 1000:.2f}ms"
+            )
+            raise e
 
         return response