fix(core): message index, eil log

Author: gusye1234

src/server/core/acontext_core/llm/agent/task.py

@@ -13,7 +13,6 @@
 from ...util.generate_ids import track_process
 from ..tool.task_lib.ctx import TaskCtx
 from ..tool.task_lib.insert import _insert_task_tool
-from ...schema.llm import LLMResponse
 
 NEED_UPDATE_CTX = {
     _insert_task_tool.schema.function.name,
@@ -94,8 +93,6 @@ async def task_agent_curd(
         if eil:
             return r
         _messages.append(response_to_sendable_message(llm_return))
-        if eil:
-            return r
         LOG.info(f"LLM Response: {llm_return.content}...")
         if not llm_return.tool_calls:
             LOG.info("No tool calls found, stop iterations")

src/server/core/acontext_core/llm/tool/task_lib/append.py

@@ -26,9 +26,9 @@ async def _append_messages_to_task_handler(
         )
     actually_task_id = ctx.task_ids_index[task_order - 1]
     actually_message_ids = [
-        ctx.message_ids_index[i - 1]
+        ctx.message_ids_index[i]
         for i in message_order_indexes
-        if i <= len(ctx.message_ids_index)
+        if i < len(ctx.message_ids_index)
     ]
     if not actually_message_ids:
         return Result.resolve(

src/server/core/acontext_core/schema/config.py

@@ -5,7 +5,9 @@
 
 
 class ProjectConfig(BaseModel):
+    project_session_message_use_previous_messages_turns: int = 3
     project_session_message_buffer_max_turns: int = 6
+    project_session_message_buffer_max_overflow_turns: int = 18
     project_session_message_buffer_ttl_seconds: int = 10
 
 

src/server/core/acontext_core/service/controller/message.py

@@ -3,34 +3,37 @@
 from ...schema.session.task import TaskStatus
 from ...schema.session.message import MessageBlob
 from ...schema.utils import asUUID
+from ...llm.agent import task as AT
 from ...schema.result import ResultError
 from ...env import LOG, DEFAULT_CORE_CONFIG
-from ...llm.agent import task as AT
+from ...schema.config import ProjectConfig
 
 
-async def process_session_pending_message(session_id: asUUID):
+async def process_session_pending_message(
+    project_config: ProjectConfig, session_id: asUUID
+):
     pending_message_ids = None
     try:
         async with DB_CLIENT.get_session_context() as session:
             r = await MD.unpending_session_messages_to_running(session, session_id)
             pending_message_ids, eil = r.unpack()
             if eil:
-                LOG.error(f"Exception while unpending session messages: {eil}")
                 return
 
         async with DB_CLIENT.get_session_context() as session:
             r = await MD.fetch_messages_data_by_ids(session, pending_message_ids)
             messages, eil = r.unpack()
             if eil:
-                LOG.error(f"Exception while fetching session messages: {eil}")
                 return
 
             r = await MD.fetch_previous_messages_by_datetime(
-                session, session_id, messages[0].created_at, limit=1
+                session,
+                session_id,
+                messages[0].created_at,
+                limit=project_config.project_session_message_use_previous_messages_turns,
             )
             previous_messages, eil = r.unpack()
             if eil:
-                LOG.error(f"Exception while fetching previous messages: {eil}")
                 return
             messages_data = [
                 MessageBlob(message_id=m.id, role=m.role, parts=m.parts)

src/server/core/acontext_core/service/data/message.py

@@ -116,9 +116,6 @@ async def fetch_messages_data_by_ids(
         for message, parts_result in zip(ordered_messages, parts_results):
             d, eil = parts_result.unpack()
             if eil:
-                LOG.error(
-                    f"Exception while fetching parts for message {message.id}: {eil}"
-                )
                 message.parts = None
                 continue
             message.parts = d
@@ -161,19 +158,20 @@ async def fetch_session_messages(
     return await fetch_messages_data_by_ids(db_session, message_ids)
 
 
-async def get_latest_message_ids(
+async def get_message_ids(
     db_session: AsyncSession,
     session_id: asUUID,
     status: str = "pending",
     limit: int = 1,
+    asc: bool = False,
 ) -> Result[List[asUUID]]:
     query = (
         select(Message.id)
         .where(
             Message.session_id == session_id,
             Message.session_task_process_status == status,
         )
-        .order_by(Message.created_at.desc())
+        .order_by(Message.created_at.asc() if asc else Message.created_at.desc())
         .limit(limit)
     )
 

src/server/core/acontext_core/service/session_message.py

@@ -61,10 +61,9 @@ async def waiting_for_message_notify(wait_for_seconds: int, body: InsertNewMessa
 async def insert_new_message(body: InsertNewMessage, message: Message):
     LOG.debug(f"Insert new message {body.message_id}")
     async with DB_CLIENT.get_session_context() as read_session:
-        r = await MD.get_latest_message_ids(read_session, body.session_id)
+        r = await MD.get_message_ids(read_session, body.session_id)
         message_ids, eil = r.unpack()
         if eil:
-            LOG.error(f"Exception while fetching session messages: {eil}")
             return
         if not len(message_ids):
             LOG.debug(f"No pending message found for session {body.session_id}, ignore")
@@ -79,13 +78,11 @@ async def insert_new_message(body: InsertNewMessage, message: Message):
         r = await PD.get_project_config(read_session, body.project_id)
         project_config, eil = r.unpack()
         if eil:
-            LOG.error(f"Exception while fetching project config: {eil}")
             return
 
         r = await MD.session_message_length(read_session, body.session_id)
         pending_message_length, eil = r.unpack()
         if eil:
-            LOG.error(f"Exception while fetching session messages: {eil}")
             return
         if (
             pending_message_length
@@ -110,14 +107,13 @@ async def insert_new_message(body: InsertNewMessage, message: Message):
             routing_key=RK.session_message_insert_retry,
             body=body.model_dump_json(),
         )
-
         return
 
     try:
         LOG.info(
             f"Session message buffer is full (size: {pending_message_length}), start process"
         )
-        await MC.process_session_pending_message(body.session_id)
+        await MC.process_session_pending_message(project_config, body.session_id)
     finally:
         await release_session_message_lock(str(body.session_id))
 
@@ -145,10 +141,9 @@ async def insert_new_message(body: InsertNewMessage, message: Message):
 )
 async def buffer_new_message(body: InsertNewMessage, message: Message):
     async with DB_CLIENT.get_session_context() as session:
-        r = await MD.get_latest_message_ids(session, body.session_id)
+        r = await MD.get_message_ids(session, body.session_id)
         message_ids, eil = r.unpack()
         if eil:
-            LOG.error(f"Exception while fetching latest message id {eil}")
             return
         if not len(message_ids):
             LOG.debug(f"No pending message found for session {body.session_id}, ignore")
@@ -159,6 +154,10 @@ async def buffer_new_message(body: InsertNewMessage, message: Message):
                 f"Message {body.message_id} is not the latest pending message, ignore"
             )
             return
+        r = await PD.get_project_config(session, body.project_id)
+        project_config, eil = r.unpack()
+        if eil:
+            return
     LOG.info(f"Message {body.message_id} IDLE, process it now")
     _l = await check_session_message_lock_or_set(str(body.session_id))
     if not _l:
@@ -172,6 +171,6 @@ async def buffer_new_message(body: InsertNewMessage, message: Message):
         )
         return
     try:
-        await MC.process_session_pending_message(body.session_id)
+        await MC.process_session_pending_message(project_config, body.session_id)
     finally:
         await release_session_message_lock(str(body.session_id))