chore: release v3.2.2

Feat: 消息截断器

README.md

@@ -42,5 +42,14 @@
 <a href="https://github.com/RockChinQ/qcg-center">遥测服务端源码</a> ｜
 <a href="https://github.com/the-lazy-me/QChatGPT-Wiki">官方文档储存库</a>
 
+<hr/>
+
+<div align="center">
+京东云4090单卡15C90G实例 <br/>
+    仅需1.89/小时，包月1225元起  <br/>
+    可选预装Stable Diffusion等应用，随用随停，计费透明，欢迎首选支持 <br/>
+https://3.cn/1ZOi6-Gj
+</div>
+
 <img alt="回复效果（带有联网插件）" src="https://qchatgpt.rockchin.top/assets/image/QChatGPT-0516.png" width="500px"/>
 </div>

pkg/config/impls/json.py

@@ -37,7 +37,10 @@ class JSONConfigFile(file_model.ConfigFile):
                 self.template_data = json.load(f)
 
         with open(self.config_file_name, "r", encoding="utf-8") as f:
-            cfg = json.load(f)
+            try:
+                cfg = json.load(f)
+            except json.JSONDecodeError as e:
+                raise Exception(f"配置文件 {self.config_file_name} 语法错误: {e}")
 
         if completion:
 

pkg/config/migrations/m008_ad_fixwin_config_migrate.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from .. import migration
+
+
+@migration.migration_class("ad-fixwin-cfg-migration", 8)
+class AdFixwinConfigMigration(migration.Migration):
+    """迁移"""
+
+    async def need_migrate(self) -> bool:
+        """判断当前环境是否需要运行此迁移"""
+        return isinstance(
+            self.ap.pipeline_cfg.data["rate-limit"]["fixwin"]["default"],
+            int
+        )
+
+    async def run(self):
+        """执行迁移"""
+        
+        for session_name in self.ap.pipeline_cfg.data["rate-limit"]["fixwin"]:
+
+            temp_dict = {
+                "window-size": 60,
+                "limit": self.ap.pipeline_cfg.data["rate-limit"]["fixwin"][session_name]
+            }
+            
+            self.ap.pipeline_cfg.data["rate-limit"]["fixwin"][session_name] = temp_dict
+
+        await self.ap.pipeline_cfg.dump_config()

pkg/config/migrations/m009_msg_truncator_cfg.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from .. import migration
+
+
+@migration.migration_class("msg-truncator-cfg-migration", 9)
+class MsgTruncatorConfigMigration(migration.Migration):
+    """迁移"""
+
+    async def need_migrate(self) -> bool:
+        """判断当前环境是否需要运行此迁移"""
+        return 'msg-truncate' not in self.ap.pipeline_cfg.data
+
+    async def run(self):
+        """执行迁移"""
+        
+        self.ap.pipeline_cfg.data['msg-truncate'] = {
+            'method': 'round',
+            'round': {
+                'max-round': 10
+            }
+        }
+
+        await self.ap.pipeline_cfg.dump_config()

pkg/core/boot.py

@@ -1,5 +1,7 @@
 from __future__ import print_function
 
+import traceback
+
 from . import app
 from ..audit import identifier
 from . import stage
@@ -27,6 +29,7 @@ async def make_app() -> app.Application:
     for stage_name in stage_order:
         stage_cls = stage.preregistered_stages[stage_name]
         stage_inst = stage_cls()
+
         await stage_inst.run(ap)
 
     await ap.initialize()
@@ -35,5 +38,8 @@ async def make_app() -> app.Application:
 
 
 async def main():
-    app_inst = await make_app()
-    await app_inst.run()
+    try:
+        app_inst = await make_app()
+        await app_inst.run()
+    except Exception as e:
+        traceback.print_exc()

pkg/core/bootutils/config.py

@@ -8,16 +8,3 @@ from ...config.impls import pymodule
 
 load_python_module_config = config_mgr.load_python_module_config
 load_json_config = config_mgr.load_json_config
-
-
-async def override_config_manager(cfg_mgr: config_mgr.ConfigManager) -> list[str]:
-    override_json = json.load(open("override.json", "r", encoding="utf-8"))
-    overrided = []
-
-    config = cfg_mgr.data
-    for key in override_json:
-        if key in config:
-            config[key] = override_json[key]
-            overrided.append(key)
-
-    return overrided

pkg/core/stages/migrate.py

@@ -5,7 +5,7 @@ import importlib
 from .. import stage, app
 from ...config import migration
 from ...config.migrations import m001_sensitive_word_migration, m002_openai_config_migration, m003_anthropic_requester_cfg_completion, m004_moonshot_cfg_completion
-from ...config.migrations import m005_deepseek_cfg_completion, m006_vision_config, m007_qcg_center_url
+from ...config.migrations import m005_deepseek_cfg_completion, m006_vision_config, m007_qcg_center_url, m008_ad_fixwin_config_migrate, m009_msg_truncator_cfg
 
 
 @stage.stage_class("MigrationStage")

pkg/pipeline/msgtrun/msgtrun.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from .. import stage, entities, stagemgr
+from ...core import entities as core_entities
+from . import truncator
+from .truncators import round
+
+
+@stage.stage_class("ConversationMessageTruncator")
+class ConversationMessageTruncator(stage.PipelineStage):
+    """会话消息截断器
+
+    用于截断会话消息链，以适应平台消息长度限制。
+    """
+    trun: truncator.Truncator
+
+    async def initialize(self):
+        use_method = self.ap.pipeline_cfg.data['msg-truncate']['method']
+
+        for trun in truncator.preregistered_truncators:
+            if trun.name == use_method:
+                self.trun = trun(self.ap)
+                break
+        else:
+            raise ValueError(f"未知的截断器: {use_method}")
+
+    async def process(self, query: core_entities.Query, stage_inst_name: str) -> entities.StageProcessResult:
+        """处理
+        """
+        query = await self.trun.truncate(query)
+
+        return entities.StageProcessResult(
+            result_type=entities.ResultType.CONTINUE,
+            new_query=query
+        )

pkg/pipeline/msgtrun/truncator.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import typing
+import abc
+
+from ...core import entities as core_entities, app
+
+
+preregistered_truncators: list[typing.Type[Truncator]] = []
+
+
+def truncator_class(
+    name: str
+) -> typing.Callable[[typing.Type[Truncator]], typing.Type[Truncator]]:
+    """截断器类装饰器
+
+    Args:
+        name (str): 截断器名称
+
+    Returns:
+        typing.Callable[[typing.Type[Truncator]], typing.Type[Truncator]]: 装饰器
+    """
+    def decorator(cls: typing.Type[Truncator]) -> typing.Type[Truncator]:
+        assert issubclass(cls, Truncator)
+
+        cls.name = name
+
+        preregistered_truncators.append(cls)
+
+        return cls
+
+    return decorator
+
+
+class Truncator(abc.ABC):
+    """消息截断器基类
+    """
+
+    name: str
+
+    ap: app.Application
+    
+    def __init__(self, ap: app.Application):
+        self.ap = ap
+
+    async def initialize(self):
+        pass
+
+    @abc.abstractmethod
+    async def truncate(self, query: core_entities.Query) -> core_entities.Query:
+        """截断
+
+        一般只需要操作query.messages，也可以扩展操作query.prompt, query.user_message。
+        请勿操作其他字段。
+        """
+        pass

pkg/pipeline/msgtrun/truncators/round.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from .. import truncator
+from ....core import entities as core_entities
+
+
+@truncator.truncator_class("round")
+class RoundTruncator(truncator.Truncator):
+    """前文回合数阶段器
+    """
+
+    async def truncate(self, query: core_entities.Query) -> core_entities.Query:
+        """截断
+        """
+        max_round = self.ap.pipeline_cfg.data['msg-truncate']['round']['max-round']
+
+        temp_messages = []
+
+        current_round = 0
+
+        # 从后往前遍历
+        for msg in query.messages[::-1]:
+            if current_round < max_round:
+                temp_messages.append(msg)
+                if msg.role == 'user':
+                    current_round += 1
+            else:
+                break
+        
+        query.messages = temp_messages[::-1]
+
+        return query

pkg/pipeline/ratelimit/algos/fixedwin.py

@@ -1,18 +1,15 @@
-# 固定窗口算法
 from __future__ import annotations
-
 import asyncio
 import time
-
 from .. import algo
 
-
+# 固定窗口算法
 class SessionContainer:
     
     wait_lock: asyncio.Lock
 
     records: dict[int, int]
-    """访问记录，key为每分钟的起始时间戳，value为访问次数"""
+    """访问记录，key为每窗口长度的起始时间戳，value为访问次数"""
 
     def __init__(self):
         self.wait_lock = asyncio.Lock()
@@ -47,30 +44,34 @@ class FixedWindowAlgo(algo.ReteLimitAlgo):
 
         # 等待锁
         async with container.wait_lock:
+
+            # 获取窗口大小和限制
+            window_size = self.ap.pipeline_cfg.data['rate-limit']['fixwin']['default']['window-size']
+            limitation = self.ap.pipeline_cfg.data['rate-limit']['fixwin']['default']['limit']
+
+            if session_name in self.ap.pipeline_cfg.data['rate-limit']['fixwin']:
+                window_size = self.ap.pipeline_cfg.data['rate-limit']['fixwin'][session_name]['window-size']
+                limitation = self.ap.pipeline_cfg.data['rate-limit']['fixwin'][session_name]['limit']
+
             # 获取当前时间戳
             now = int(time.time())
 
-            # 获取当前分钟的起始时间戳
-            now = now - now % 60
+            # 获取当前窗口的起始时间戳
+            now = now - now % window_size
 
-            # 获取当前分钟的访问次数
+            # 获取当前窗口的访问次数
             count = container.records.get(now, 0)
 
-            limitation = self.ap.pipeline_cfg.data['rate-limit']['fixwin']['default']
-
-            if session_name in self.ap.pipeline_cfg.data['rate-limit']['fixwin']:
-                limitation = self.ap.pipeline_cfg.data['rate-limit']['fixwin'][session_name]
-
             # 如果访问次数超过了限制
             if count >= limitation:
                 if self.ap.pipeline_cfg.data['rate-limit']['strategy'] == 'drop':
                     return False
                 elif self.ap.pipeline_cfg.data['rate-limit']['strategy'] == 'wait':
-                    # 等待下一分钟
-                    await asyncio.sleep(60 - time.time() % 60)
+                    # 等待下一窗口
+                    await asyncio.sleep(window_size - time.time() % window_size)
     
                     now = int(time.time())
-                    now = now - now % 60
+                    now = now - now % window_size
             
             if now not in container.records:
                 container.records = {}

pkg/pipeline/stagemgr.py

@@ -13,6 +13,7 @@ from .respback import respback
 from .wrapper import wrapper
 from .preproc import preproc
 from .ratelimit import ratelimit
+from .msgtrun import msgtrun
 
 
 # 请求处理阶段顺序
@@ -21,6 +22,7 @@ stage_order = [
     "BanSessionCheckStage",  # 封禁会话检查
     "PreContentFilterStage",  # 内容过滤前置阶段
     "PreProcessor",  # 预处理器
+    "ConversationMessageTruncator",  # 会话消息截断器
     "RequireRateLimitOccupancy",  # 请求速率限制占用
     "MessageProcessor",  # 处理器
     "ReleaseRateLimitOccupancy",  # 释放速率限制占用

pkg/pipeline/wrapper/wrapper.py

@@ -71,7 +71,7 @@ class ResponseWrapper(stage.PipelineStage):
 
                     reply_text = ''
 
-                    if result.content is not None:  # 有内容
+                    if result.content:  # 有内容
                         reply_text = str(result.get_content_mirai_message_chain())
 
                         # ============= 触发插件事件 ===============

pkg/utils/constants.py

@@ -1 +1 @@
-semantic_version = "v3.2.1"
+semantic_version = "v3.2.2"

templates/pipeline.json

@@ -29,7 +29,16 @@
         "strategy": "drop",
         "algo": "fixwin",
         "fixwin": {
-            "default": 60
+            "default": {
+                "window-size": 60,
+                "limit": 60
+            }
+        }
+    },
+    "msg-truncate": {
+        "method": "round",
+        "round": {
+            "max-round": 10
         }
     }
 }