fix(storage): 优化 SQL 存储兼容性

- 新增通用的 Content 清理逻辑，在 SQL Session 和 Memory
  存储读写时过滤空的 Content Part。
- 修复 MySQL 下 DynamicPickleType 写入 LONGBLOB
  时未序列化导致的类型错误。
- 支持通过 sessionmaker_kwargs 向 SqlStorage 传递 sessionmaker
  配置，例如 expire_on_commit。
- 将 SQL Session 中显式更新 update_time 的逻辑改为使用
  datetime.now()，避免 func.now() 带来的 ORM 状态问题。
- 统一通过当前事件循环创建 Session 和 Memory 服务的清理任务。
- 优化 SQL Session 示例输出，在助手回复前打印用户问题。
- 在 SQL Memory/Session 示例中关闭 thinking 输出，使演示结果更清晰。
- 增加 SQL Content 清理和 MySQL DynamicPickleType 序列化相关测试。

Author: raychen911

examples/memory_service_with_sql/agent/agent.py

@@ -10,6 +10,8 @@
 from trpc_agent_sdk.models import OpenAIModel
 from trpc_agent_sdk.tools import FunctionTool
 from trpc_agent_sdk.tools import load_memory_tool
+from trpc_agent_sdk.types import GenerateContentConfig
+from trpc_agent_sdk.types import HttpOptions
 
 from .config import get_model_config
 from .prompts import INSTRUCTION
@@ -25,12 +27,18 @@ def _create_model() -> LLMModel:
 
 def create_agent() -> LlmAgent:
     """ Create an agent"""
+    generate_content_config = GenerateContentConfig(
+        http_options=HttpOptions(extra_body={"chat_template_kwargs": {
+                                     "enable_thinking": False
+                                 }}),
+    )
     agent = LlmAgent(
         name="assistant",
         description="A helpful assistant for conversation",
         model=_create_model(),  # You can change this to your preferred model
         instruction=INSTRUCTION,
         tools=[FunctionTool(get_weather_report), load_memory_tool],
+        generate_content_config=generate_content_config,
     )
     return agent
 

examples/session_service_with_sql/agent/agent.py

@@ -9,6 +9,8 @@
 from trpc_agent_sdk.models import LLMModel
 from trpc_agent_sdk.models import OpenAIModel
 from trpc_agent_sdk.tools import FunctionTool
+from trpc_agent_sdk.types import GenerateContentConfig
+from trpc_agent_sdk.types import HttpOptions
 
 from .config import get_model_config
 from .prompts import INSTRUCTION
@@ -24,12 +26,18 @@ def _create_model() -> LLMModel:
 
 def create_agent() -> LlmAgent:
     """ Create an agent"""
+    generate_content_config = GenerateContentConfig(
+        http_options=HttpOptions(extra_body={"chat_template_kwargs": {
+                                     "enable_thinking": False
+                                 }}),
+    )
     agent = LlmAgent(
         name="assistant",
         description="A helpful assistant for conversation",
         model=_create_model(),  # You can change this to your preferred model
         instruction=INSTRUCTION,
         tools=[FunctionTool(get_weather_report)],
+        generate_content_config=generate_content_config,
     )
     return agent
 

examples/session_service_with_sql/run_agent.py

@@ -76,10 +76,9 @@ async def run_weather_agent():
     ]
 
     for query in demo_queries:
-        # Use a new session for each query
-
         user_content = Content(parts=[Part.from_text(text=query)])
 
+        print(f"👤 User: {query}")
         print("🤖 Assistant: ", end="", flush=True)
         async for event in runner.run_async(user_id=user_id, session_id=current_session_id, new_message=user_content):
             # Check if event.content exists

tests/sessions/test_sql_session_service.py

@@ -90,12 +90,39 @@ def test_from_event_with_function_call(self):
         storage_event = SessionStorageEvent.from_event(session, event)
         assert storage_event.content is not None
 
+    def test_from_event_drops_empty_parts(self):
+        session = Session(id="s1", app_name="app", user_id="user", save_key="k")
+        event = Event(
+            invocation_id="inv-1",
+            author="agent",
+            content=Content(parts=[Part()]),
+        )
+        storage_event = SessionStorageEvent.from_event(session, event)
+        assert storage_event.content is None
+
     def test_from_event_no_content(self):
         session = Session(id="s1", app_name="app", user_id="user", save_key="k")
         event = Event(invocation_id="inv-1", author="agent", actions=EventActions())
         storage_event = SessionStorageEvent.from_event(session, event)
         assert storage_event.content is None
 
+    def test_to_event_drops_legacy_empty_parts(self):
+        storage_event = SessionStorageEvent(
+            id="e1",
+            app_name="app",
+            user_id="user",
+            session_id="s1",
+            invocation_id="inv-1",
+            author="agent",
+            actions=EventActions(),
+            long_running_tool_ids=set(),
+            timestamp=datetime.now(),
+            model_flags=1,
+            content={"parts": [{}], "role": "model"},
+        )
+        event = storage_event.to_event()
+        assert event.content is None
+
     def test_long_running_tool_ids_property(self):
         session = Session(id="s1", app_name="app", user_id="user", save_key="k")
         event = _make_event()

tests/storage/test_sql_common.py

@@ -332,6 +332,14 @@ def test_process_bind_param_spanner(self):
         result = dpt.process_bind_param(value, dialect)
         assert pickle.loads(result) == value
 
+    def test_process_bind_param_mysql(self):
+        dpt = DynamicPickleType()
+        dialect = _make_dialect("mysql")
+        value = {"key": "value", "nums": [1, 2, 3]}
+        result = dpt.process_bind_param(value, dialect)
+        assert isinstance(result, bytes)
+        assert pickle.loads(result) == value
+
     def test_process_bind_param_non_spanner(self):
         dpt = DynamicPickleType()
         dialect = _make_dialect("sqlite")
@@ -352,6 +360,14 @@ def test_process_result_value_spanner(self):
         result = dpt.process_result_value(pickled, dialect)
         assert result == original
 
+    def test_process_result_value_mysql(self):
+        dpt = DynamicPickleType()
+        dialect = _make_dialect("mysql")
+        original = {"key": "value", "nums": [1, 2, 3]}
+        pickled = pickle.dumps(original)
+        result = dpt.process_result_value(pickled, dialect)
+        assert result == original
+
     def test_process_result_value_non_spanner(self):
         dpt = DynamicPickleType()
         dialect = _make_dialect("sqlite")

trpc_agent_sdk/memory/_in_memory_memory_service.py

@@ -200,7 +200,7 @@ def _start_cleanup_task(self) -> None:
             return
 
         self.__cleanup_stop_event = asyncio.Event()
-        self.__cleanup_task = asyncio.create_task(self._cleanup_loop())
+        self.__cleanup_task = asyncio.get_event_loop().create_task(self._cleanup_loop())
         logger.debug("Cleanup task created")
 
     def _stop_cleanup_task(self) -> None:

trpc_agent_sdk/memory/_sql_memory_service.py

@@ -43,6 +43,7 @@
 from trpc_agent_sdk.storage import SqlStorage
 from trpc_agent_sdk.storage import decode_content
 from trpc_agent_sdk.storage import decode_grounding_metadata
+from trpc_agent_sdk.storage import sanitize_content_json
 
 from ._utils import extract_words_lower
 from ._utils import format_timestamp
@@ -110,7 +111,7 @@ def update_event(self, session: Session, event: Event):
         self.error_message = event.error_message
         self.interrupted = event.interrupted
         if event.content:
-            self.content = event.content.model_dump(exclude_none=True, mode="json")
+            self.content = sanitize_content_json(event.content.model_dump(exclude_none=True, mode="json"))
         if event.grounding_metadata:
             self.grounding_metadata = event.grounding_metadata.model_dump(exclude_none=True, mode="json")
         if event.custom_metadata:
@@ -135,7 +136,7 @@ def from_event(cls, session: Session, event: Event) -> MemStorageEvent:
             interrupted=event.interrupted,
         )
         if event.content:
-            storage_event.content = event.content.model_dump(exclude_none=True, mode="json")
+            storage_event.content = sanitize_content_json(event.content.model_dump(exclude_none=True, mode="json"))
         if event.grounding_metadata:
             storage_event.grounding_metadata = event.grounding_metadata.model_dump(exclude_none=True, mode="json")
         if event.custom_metadata:
@@ -150,7 +151,7 @@ def to_event(self) -> Event:
             branch=self.branch,
             actions=self.actions,  # type: ignore
             timestamp=self.timestamp.timestamp(),
-            content=decode_content(self.content),
+            content=decode_content(sanitize_content_json(self.content)),
             long_running_tool_ids=self.long_running_tool_ids,
             partial=self.partial,
             turn_complete=self.turn_complete,
@@ -194,15 +195,15 @@ async def store_session(self, session: Session, agent_context: Optional[AgentCon
 
         Only stores events that are not expired based on event_ttl_seconds.
         """
-        if not isinstance(session, Session):
-            raise TypeError(f"Content must be a Session, got {type(session)}")
-
         async with self._sql_storage.create_db_session() as sql_session:
             is_exist = False
             for event in session.events:
                 if not event.is_model_visible():
                     continue
-                if event.content and event.content.parts:
+                if not event.content or not event.content.parts:
+                    continue
+                content = sanitize_content_json(event.content.model_dump(exclude_none=True, mode="json"))
+                if content:
                     is_exist = True
                     # Check if the event already exists
                     event_key = SqlKey(key=(event.id, session.save_key, session.id), storage_cls=MemStorageEvent)
@@ -324,7 +325,7 @@ def _start_cleanup_task(self) -> None:
             return
 
         self.__cleanup_stop_event = asyncio.Event()
-        self.__cleanup_task = asyncio.create_task(self._cleanup_loop())
+        self.__cleanup_task = asyncio.get_event_loop().create_task(self._cleanup_loop())
         logger.debug("Memory cleanup task created")
 
     def _stop_cleanup_task(self) -> None:

trpc_agent_sdk/memory/mem0_memory_service.py

@@ -328,7 +328,7 @@ def _start_cleanup_task(self) -> None:
             return
 
         self.__cleanup_stop_event = asyncio.Event()
-        self.__cleanup_task = asyncio.create_task(self._cleanup_loop())
+        self.__cleanup_task = asyncio.get_event_loop().create_task(self._cleanup_loop())
         logger.debug("Mem0 memory cleanup task created")
 
     def _stop_cleanup_task(self) -> None:

trpc_agent_sdk/sessions/_in_memory_session_service.py

@@ -399,7 +399,7 @@ def _start_cleanup_task(self) -> None:
             return
 
         self.__cleanup_stop_event = asyncio.Event()
-        self.__cleanup_task = asyncio.create_task(self._cleanup_loop())
+        self.__cleanup_task = asyncio.get_event_loop().create_task(self._cleanup_loop())
         logger.debug("Cleanup task created")
 
     def _stop_cleanup_task(self) -> None:

trpc_agent_sdk/sessions/_sql_session_service.py

@@ -65,6 +65,7 @@
 from trpc_agent_sdk.storage import decode_content
 from trpc_agent_sdk.storage import decode_grounding_metadata
 from trpc_agent_sdk.storage import decode_usage_metadata
+from trpc_agent_sdk.storage import sanitize_content_json
 from trpc_agent_sdk.utils import user_key
 
 from ._base_session_service import BaseSessionService
@@ -236,7 +237,7 @@ def from_event(cls, session: Session, event: Event) -> SessionStorageEvent:
             response_id=event.response_id,
         )
         if event.content:
-            storage_event.content = event.content.model_dump(exclude_none=True, mode="json")
+            storage_event.content = sanitize_content_json(event.content.model_dump(exclude_none=True, mode="json"))
         if event.grounding_metadata:
             storage_event.grounding_metadata = event.grounding_metadata.model_dump(exclude_none=True, mode="json")
         if event.custom_metadata:
@@ -269,7 +270,7 @@ def to_event(self) -> Event:
             error_message=self.error_message,
             interrupted=self.interrupted,
             response_id=self.response_id,
-            content=decode_content(self.content),
+            content=decode_content(sanitize_content_json(self.content)),
             grounding_metadata=decode_grounding_metadata(self.grounding_metadata),
             custom_metadata=self.custom_metadata,
             usage_metadata=decode_usage_metadata(self.usage_metadata),
@@ -574,7 +575,7 @@ async def _update_app_state(self, sql_session: SqlSession, app_name: str, state_
             await self._sql_storage.add(sql_session, storage_app_state)
         else:
             storage_app_state.state = app_state  # type: ignore
-        storage_app_state.update_time = func.now()
+        storage_app_state.update_time = datetime.now()
 
         return app_state
 
@@ -604,7 +605,7 @@ async def _get_app_state(self, sql_session: SqlSession, app_name: str) -> dict[s
         if storage_app_state:
             if not self._session_config.is_expired_by_timestamp(storage_app_state.update_time.timestamp()):
                 app_state = storage_app_state.state
-                storage_app_state.update_time = func.now()
+                storage_app_state.update_time = datetime.now()
                 await self._sql_storage.commit(sql_session)
 
         return app_state
@@ -617,7 +618,7 @@ async def _get_user_state(self, sql_session: SqlSession, app_name: str, user_id:
         if storage_user_state:
             if not self._session_config.is_expired_by_timestamp(storage_user_state.update_time.timestamp()):
                 user_state = storage_user_state.state
-                storage_user_state.update_time = func.now()
+                storage_user_state.update_time = datetime.now()
                 await self._sql_storage.commit(sql_session)
 
         return user_state
@@ -633,7 +634,7 @@ async def _get_session(self, sql_session: SqlSession, app_name: str, user_id: st
             logger.debug("Session %s is expired", session_id)
             return None
 
-        storage_session.update_time = func.now()
+        storage_session.update_time = datetime.now()
         await self._sql_storage.commit(sql_session)
 
         return storage_session
@@ -645,7 +646,7 @@ async def _cleanup_expired_async(self) -> None:
         Deletes all expired data in three batch SQL DELETE statements.
         """
         async with self._sql_storage.create_db_session() as sql_session:
-            # Calculate expiration threshold once (using database local time)
+            # Calculate expiration threshold once in application time for cross-database compatibility.
             expire_before = datetime.now() - timedelta(seconds=self._session_config.ttl.ttl_seconds)
             total_deleted = 0
 
@@ -721,7 +722,7 @@ def _start_cleanup_task(self) -> None:
             return
 
         self.__cleanup_stop_event = asyncio.Event()
-        self.__cleanup_task = asyncio.create_task(self._cleanup_loop())
+        self.__cleanup_task = asyncio.get_event_loop().create_task(self._cleanup_loop())
         logger.debug("Cleanup task created")
 
     def _stop_cleanup_task(self) -> None: