feat(bot): Bot session compact by using openviking, ov client add function and param (volcengine#2284)

* bot session compact by using openviking

* fix bug

* fix bug

* 默认值

* 默认值

* fix pr bug

* fix merge bug

Author: yeshion23333

bot/docs/vikingbot-openviking-context-plan.md

@@ -433,6 +433,10 @@ class AgentsConfig(BaseModel):
     model: str = "openai/doubao-seed-2-0-pro-260215"
     max_tool_iterations: int = 50
     memory_window: int = 50
+    session_context_enabled: bool = False
+    session_context_token_budget: int = 3000
+    commit_token_threshold: int = 20000
+    commit_keep_recent_count: int = 5
     gen_image_model: str = "openai/doubao-seedream-4-5-251128"
     provider: str = ""
     api_key: str = ""

bot/tests/test_agent_loop_outcome.py

@@ -175,8 +175,6 @@ async def _tick(self) -> None:
         active_workspaces = 0
 
         for workspace_path, session_key_list in workspaces.items():
-            logger.debug(f"Heartbeat: checking workspace {workspace_path}...")
-
             content = _read_heartbeat_file(workspace_path)
 
             # Skip if HEARTBEAT.md is empty or doesn't exist
@@ -211,9 +209,6 @@ async def _tick(self) -> None:
                 except Exception as e:
                     logger.exception(f"Heartbeat execution failed for {workspace_path}: {e}")
 
-        if active_workspaces == 0:
-            logger.debug("Heartbeat: no tasks in any workspace")
-
     async def trigger_now(self, session_key: SessionKey | None = None) -> str | None:
         """Manually trigger a heartbeat."""
         if self.on_heartbeat:

bot/tests/test_openviking_api_key_type.py

@@ -1,11 +1,21 @@
 import asyncio
 import re
 from collections import defaultdict
+from datetime import datetime
 from typing import Any
 
 from loguru import logger
 
 from vikingbot.config.loader import load_config
+from vikingbot.openviking_mount.session_state import (
+    get_openviking_session_id,
+    get_openviking_state,
+    get_unsynced_messages,
+    get_unsynced_messages_for_sender,
+    parse_local_index,
+    set_sender_synced_local_index,
+)
+from vikingbot.utils.helpers import cal_str_tokens
 
 from ...session import Session
 from ..base import Hook, HookContext
@@ -40,15 +50,273 @@ class OpenVikingCompactHook(Hook):
     async def _get_client(self, workspace_id: str) -> VikingClient:
         return await get_global_client(workspace_id)
 
+    async def _execute_session_context_commit(
+        self,
+        context: HookContext,
+        session: Session,
+        client: VikingClient,
+        agents_config: Any,
+        admin_user_id: str,
+        *,
+        force_commit: bool,
+        keep_recent_count: int,
+        commit_message_threshold: int | None,
+    ) -> dict[str, Any]:
+        state = get_openviking_state(session)
+        session_id = get_openviking_session_id(
+            session,
+            default_session_id=context.session_key.safe_name(),
+        )
+
+        commit_token_threshold = int(getattr(agents_config, "commit_token_threshold", 6000) or 6000)
+        pending_tokens = int(state.get("last_pending_tokens", 0) or 0)
+        messages_to_sync = get_unsynced_messages(session)
+        last_commit_local_index = parse_local_index(state.get("last_commit_local_index", -1))
+        messages_since_commit = len(session.messages) - last_commit_local_index - 1
+        reached_message_threshold = bool(
+            commit_message_threshold is not None
+            and commit_message_threshold > 0
+            and messages_since_commit >= commit_message_threshold
+        )
+
+        admin_append_result = None
+        admin_commit_result = None
+        user_results = []
+        pending_tokens_before_sync = pending_tokens
+
+        unsynced_tokens = sum(
+            cal_str_tokens(str(msg.get("content") or ""))
+            for msg in messages_to_sync
+            if msg.get("content") is not None
+        )
+
+        all_sender_ids = sorted(
+            {
+                str(msg.get("sender_id"))
+                for msg in session.messages
+                if msg.get("sender_id") and msg.get("sender_id") != admin_user_id
+            }
+        )
+        unsynced_messages_by_sender = {}
+        sender_latest_indexes: dict[str, int] = {}
+        for sender_id in all_sender_ids:
+            user_messages = get_unsynced_messages_for_sender(
+                session,
+                sender_id,
+                admin_user_id=admin_user_id,
+            )
+            if user_messages:
+                unsynced_messages_by_sender[sender_id] = user_messages
+                sender_latest_indexes[sender_id] = max(
+                    index
+                    for index, msg in enumerate(session.messages)
+                    if msg.get("sender_id") == sender_id
+                )
+
+        should_commit = bool(
+            force_commit
+            or pending_tokens + unsynced_tokens >= commit_token_threshold
+            or reached_message_threshold
+        )
+        sender_ids_to_sync = (
+            all_sender_ids if should_commit else sorted(unsynced_messages_by_sender)
+        )
+        user_results_by_id: dict[str, dict[str, Any]] = {}
+
+        if sender_ids_to_sync:
+            semaphore = asyncio.Semaphore(5)
+
+            async def sync_sender(user_id: str):
+                user_messages = unsynced_messages_by_sender.get(user_id, [])
+                async with semaphore:
+                    sender_session_id = f"{session_id}_{user_id}"
+                    append_result = None
+                    if user_messages:
+                        append_result = await client.append_messages(
+                            session_id=sender_session_id,
+                            messages=user_messages,
+                            default_user_role_id=user_id,
+                            session_user_id=user_id,
+                        )
+                    return {
+                        "session_id": sender_session_id,
+                        "user_id": user_id,
+                        "append": append_result,
+                    }
+
+            user_results = await asyncio.gather(
+                *(sync_sender(user_id) for user_id in sender_ids_to_sync),
+                return_exceptions=True,
+            )
+            for result in user_results:
+                if isinstance(result, dict):
+                    user_id = result["user_id"]
+                    user_results_by_id[user_id] = result
+                    if result.get("append") is not None and user_id in sender_latest_indexes:
+                        set_sender_synced_local_index(
+                            session, user_id, sender_latest_indexes[user_id]
+                        )
+
+        fanout_errors = [result for result in user_results if isinstance(result, Exception)]
+        if fanout_errors:
+            error_message = "; ".join(str(error) for error in fanout_errors)
+            state["last_pending_tokens"] = pending_tokens_before_sync
+            state["last_commit_performed"] = False
+            state["last_sync_status"] = "error"
+            state["last_sync_error"] = error_message
+            return {
+                "success": False,
+                "session_id": session_id,
+                "admin_result": {
+                    "append": admin_append_result,
+                    "commit": admin_commit_result,
+                    "committed": False,
+                },
+                "user_results": user_results,
+                "users_count": len(sender_ids_to_sync),
+                "pending_tokens": pending_tokens_before_sync,
+                "error": error_message,
+            }
+
+        if messages_to_sync:
+            admin_append_result = await client.append_messages(
+                session_id=session_id,
+                messages=messages_to_sync,
+                default_user_role_id=admin_user_id,
+                session_user_id=admin_user_id,
+            )
+            state["last_synced_local_index"] = len(session.messages) - 1
+            admin_session_state = await client.get_session(session_id, user_id=admin_user_id)
+            pending_tokens = int(admin_session_state.get("pending_tokens", 0) or 0)
+        elif force_commit:
+            admin_session_state = await client.get_session(session_id, user_id=admin_user_id)
+            pending_tokens = int(admin_session_state.get("pending_tokens", 0) or 0)
+
+        should_commit = (
+            force_commit or pending_tokens >= commit_token_threshold or reached_message_threshold
+        )
+        if should_commit:
+            sender_ids_to_commit = all_sender_ids
+            for user_id in sender_ids_to_commit:
+                user_results_by_id.setdefault(
+                    user_id,
+                    {
+                        "session_id": f"{session_id}_{user_id}",
+                        "user_id": user_id,
+                        "append": None,
+                    },
+                )
+
+            if sender_ids_to_commit:
+                semaphore = asyncio.Semaphore(5)
+
+                async def commit_sender(user_id: str):
+                    async with semaphore:
+                        sender_session_id = f"{session_id}_{user_id}"
+                        logger.info(
+                            f"[HOOK] Committed session {sender_session_id} for user {user_id}"
+                        )
+                        return await client.commit_session(
+                            session_id=sender_session_id,
+                            keep_recent_count=keep_recent_count,
+                            user_id=user_id,
+                        )
+
+                commit_results = await asyncio.gather(
+                    *(commit_sender(user_id) for user_id in sender_ids_to_commit),
+                    return_exceptions=True,
+                )
+                for user_id, commit_result in zip(
+                    sender_ids_to_commit, commit_results, strict=True
+                ):
+                    user_results_by_id[user_id]["commit"] = commit_result
+                fanout_errors = [
+                    result for result in commit_results if isinstance(result, Exception)
+                ]
+                if fanout_errors:
+                    error_message = "; ".join(str(error) for error in fanout_errors)
+                    state["last_pending_tokens"] = pending_tokens
+                    state["last_commit_performed"] = False
+                    state["last_sync_status"] = "error"
+                    state["last_sync_error"] = error_message
+                    return {
+                        "success": False,
+                        "session_id": session_id,
+                        "admin_result": {
+                            "append": admin_append_result,
+                            "commit": admin_commit_result,
+                            "committed": False,
+                        },
+                        "user_results": list(user_results_by_id.values()),
+                        "users_count": len(user_results_by_id),
+                        "pending_tokens": pending_tokens,
+                        "error": error_message,
+                    }
+
+            admin_commit_result = await client.commit_session(
+                session_id=session_id,
+                keep_recent_count=keep_recent_count,
+                user_id=admin_user_id,
+            )
+            logger.info(f"[HOOK] Committed session {session_id} for user {admin_user_id}")
+            admin_session_state = await client.get_session(session_id, user_id=admin_user_id)
+            pending_tokens = int(admin_session_state.get("pending_tokens", 0) or 0)
+
+        if should_commit:
+            state["last_commit_at"] = datetime.now().isoformat()
+            state["last_commit_local_index"] = len(session.messages) - 1
+        state["last_pending_tokens"] = pending_tokens
+        state["last_commit_performed"] = should_commit
+        state["last_sync_status"] = "success"
+        state.pop("last_sync_error", None)
+
+        return {
+            "success": True,
+            "session_id": session_id,
+            "admin_result": {
+                "append": admin_append_result,
+                "commit": admin_commit_result,
+                "committed": should_commit,
+            },
+            "user_results": list(user_results_by_id.values()),
+            "users_count": len(user_results_by_id),
+            "pending_tokens": pending_tokens,
+        }
+
     async def execute(self, context: HookContext, **kwargs) -> Any:
         vikingbot_session: Session = kwargs.get("session", {})
         session_id = context.session_key.safe_name()
         config = load_config()
-        admin_user_id = config.ov_server.admin_user_id
+        ov_config = config.ov_server
+        agents_config = config.agents
+        admin_user_id = ov_config.admin_user_id
+        force_commit = bool(kwargs.get("force_commit", False))
+        keep_recent_count = int(
+            kwargs.get(
+                "keep_recent_count",
+                getattr(agents_config, "commit_keep_recent_count", 10),
+            )
+            or 0
+        )
+        commit_message_threshold = kwargs.get("commit_message_threshold")
+        if commit_message_threshold is not None:
+            commit_message_threshold = int(commit_message_threshold)
 
         try:
             client = await self._get_client(context.workspace_id)
 
+            if getattr(agents_config, "session_context_enabled", False):
+                return await self._execute_session_context_commit(
+                    context,
+                    vikingbot_session,
+                    client,
+                    agents_config,
+                    admin_user_id,
+                    force_commit=force_commit,
+                    keep_recent_count=keep_recent_count,
+                    commit_message_threshold=commit_message_threshold,
+                )
+
             if not client.should_sender_fanout():
                 single_result = await client.commit(session_id, vikingbot_session.messages, None)
                 return {
@@ -92,6 +360,11 @@ async def commit_with_semaphore(user_id: str, user_messages: list):
                 "users_count": len(messages_by_sender),
             }
         except Exception as e:
+            state = None
+            if hasattr(vikingbot_session, "metadata"):
+                state = get_openviking_state(vikingbot_session)
+                state["last_sync_status"] = "error"
+                state["last_sync_error"] = str(e)
             logger.exception(f"Failed to add message to OpenViking: {e}")
             return {"success": False, "error": str(e)}
 

bot/vikingbot/agent/loop.py

@@ -1,7 +1,7 @@
 import asyncio
 import importlib
 from collections import defaultdict
-from typing import List, Any, Dict, Type
+from typing import Any, Dict, List, Type
 
 from loguru import logger
 
@@ -47,9 +47,6 @@ async def execute_hooks(self, context: HookContext, **kwargs) -> List[Any]:
         async_hooks = [hook for hook in self._hooks[context.event_type] if not hook.is_sync]
         sync_hooks = [hook for hook in self._hooks[context.event_type] if hook.is_sync]
         if async_hooks:
-            logger.debug(
-                f"Executing {len(async_hooks)} async hooks for event '{context.event_type}'"
-            )
             async_results = await asyncio.gather(
                 *[hook.execute(context, **kwargs) for hook in async_hooks], return_exceptions=True
             )