fix(analysis): 表达学习批量链路合并 Bot 回复并修正人格兼容性参数 (#257)

CHANGELOG.md

@@ -2,6 +2,14 @@
 
 所有重要更改都将记录在此文件中。
 
+## [Unreleased]
+
+### 修复（表达模式学习链路隐患，延续 issue #257）
+
+- 批量/审批链路的表达模式学习以前只拿到用户原始消息（`filtered_messages` 来自 `get_unprocessed_messages`，不含 bot 回复），而 `_extract_few_shot_pairs` 需要「用户→bot」相邻对话对，导致恒学习不到任何模式（issue #257 日志中的“未获得有效结果”）。现将合并数据库中 `BotMessage` 回复的逻辑下沉到 `ExpressionPatternLearner.trigger_learning_for_group`：仅在传入消息不含 bot 时按时间线自动合并，实时链路（已预合并）行为不变。
+- 修正 `PersonaUpdater.analyze_persona_compatibility` 调用 `get_current_persona()` 未传必填 `group_id` 导致的潜在 `TypeError`（改为接受并透传 `group_id`）。
+- 新增回归测试：以真实 sqlite 验证批量链路合并 bot 回复后能学到模式，并验证已含 bot / 无 db 时不重复查询。
+
 ## [4.2.4] - 2026-09-19
 
 ### 修复：表达模式学习按群组/人格隔离降级为 default（issue #257）

services/analysis/expression_pattern_learner.py

@@ -112,7 +112,110 @@ def _expression_scope_key(
         user_id: Optional[str] = None,
     ) -> str:
         return f"{group_id}:{persona_id}:{user_id or 'group-level'}"
-
+
+    @staticmethod
+    def _msg_sender_id(msg: Any) -> str:
+        if hasattr(msg, "sender_id"):
+            return str(getattr(msg, "sender_id", "") or "")
+        if isinstance(msg, dict):
+            return str(msg.get("sender_id", "") or "")
+        return ""
+
+    @staticmethod
+    def _msg_timestamp(msg: Any) -> float:
+        value = (
+            getattr(msg, "timestamp", 0)
+            if hasattr(msg, "timestamp")
+            else msg.get("timestamp", 0) if isinstance(msg, dict) else 0
+        )
+        try:
+            return float(value or 0)
+        except (TypeError, ValueError):
+            return 0.0
+
+    @staticmethod
+    def _msg_content(msg: Any) -> str:
+        if hasattr(msg, "message"):
+            return str(getattr(msg, "message", "") or "")
+        if isinstance(msg, dict):
+            return str(msg.get("message", "") or "")
+        return ""
+
+    async def _merge_bot_messages_for_group(
+        self, group_id: str, messages: List[Any]
+    ) -> List[Any]:
+        """将数据库中的 Bot 回复按时间线合并进用户消息，供 few-shot 对话对提取。
+
+        实时学习链路已在调用前合并过 bot 消息（消息中已含 sender_id=='bot'），
+        此时直接返回避免重复查询；批量/审批链路传入的原始消息只有用户发言，
+        必须合并 bot 回复才能提取到 用户→bot 对话对（否则学习结果恒为空，
+        即 issue #257 日志中的“未获得有效结果”）。
+
+        为避免跨窗口误配对：只取与这批用户消息时间窗口（末条向后一个宽容窗口）
+        重叠的 Bot 回复；为避免“先 limit 后过滤”将有效回复滤空：先多取候选再在
+        内存过滤并截断到与用户消息相当的数量。
+        """
+        if not messages or not self.db_manager:
+            return messages
+        if any(self._msg_sender_id(m) == "bot" for m in messages):
+            return messages
+        try:
+            from sqlalchemy import select
+
+            from ...models.orm.message import BotMessage
+            from ..learning.sample_filter import should_ignore_learning_sample
+
+            user_ts = [
+                self._msg_timestamp(m)
+                for m in messages
+                if self._msg_content(m)
+            ]
+            user_ts = [t for t in user_ts if t > 0]
+            if not user_ts:
+                return messages
+
+            forward_window_seconds = 3600.0
+            t_lo = min(user_ts)
+            t_hi = max(user_ts) + forward_window_seconds
+            # 先多取候选，再过滤/截断，避免被 ignore 样本占满 limit 导致有效回复被漏掉
+            fetch_limit = max(len(messages) * 3, 60)
+
+            async with self.db_manager.get_session() as session:
+                stmt = (
+                    select(BotMessage)
+                    .where(
+                        BotMessage.group_id == group_id,
+                        BotMessage.timestamp >= t_lo,
+                        BotMessage.timestamp <= t_hi,
+                    )
+                    .order_by(BotMessage.timestamp.asc())
+                    .limit(fetch_limit)
+                )
+                result = await session.execute(stmt)
+                bot_msgs: List[Dict[str, Any]] = []
+                for row in result.scalars().all():
+                    if should_ignore_learning_sample(
+                        row.message, sender_id="bot", is_bot=True
+                    ):
+                        continue
+                    bot_msgs.append(
+                        {
+                            "sender_id": "bot",
+                            "message": row.message,
+                            "timestamp": float(row.timestamp),
+                        }
+                    )
+                    if len(bot_msgs) >= len(messages):
+                        break
+            if not bot_msgs:
+                return messages
+            merged = list(messages) + bot_msgs
+            merged.sort(key=self._msg_timestamp)
+            return merged
+        except Exception as exc:
+            logger.debug(f"合并 Bot 回复失败，使用原始消息继续: {exc}")
+            return messages
+
     @classmethod
     def get_instance(cls, config: PluginConfig = None, db_manager: DatabaseManager = None, context=None, llm_adapter=None) -> 'ExpressionPatternLearner':
         """获取单例实例，支持延迟初始化"""
@@ -184,6 +287,12 @@ async def trigger_learning_for_group(
             return False
 
         try:
+            # 先合并数据库中已存储的 Bot 回复，才能提取到 用户→bot 对话对（修复批量链路恒为空）
+            recent_messages = await self._merge_bot_messages_for_group(
+                group_id, recent_messages
+            )
+            if len(recent_messages) < 3:
+                return False
             persona_id = normalize_persona_scope(persona_id)
             user_scope = self._normalize_user_scope(user_id)
             learning_scope = self._expression_scope_key(group_id, persona_id, user_scope)

services/persona/persona_updater.py

@@ -710,10 +710,15 @@ def _extract_style_adjustments(self, style_analysis: Dict[str, Any]) -> str:
         except Exception as e:
             return f"提取失败: {str(e)}"
 
-    async def analyze_persona_compatibility(self, target_style: Dict[str, Any]) -> AnalysisResult:
-        """分析目标风格与当前人格的兼容性"""
+    async def analyze_persona_compatibility(self, target_style: Dict[str, Any], group_id: str = "default") -> AnalysisResult:
+        """分析目标风格与当前人格的兼容性
+
+        Args:
+            target_style: 目标风格
+            group_id: 目标群组ID，用于解析该群组当前人格（缺失会误将 group_id 当作位置参数遗漏）
+        """
         try:
-            current_persona = await self.get_current_persona()
+            current_persona = await self.get_current_persona(group_id)
             if not current_persona:
                 return AnalysisResult(
                     success=False,

tests/unit/test_learning_chain_regressions.py

@@ -2477,3 +2477,146 @@ async def test_update_persona_with_style_forwards_group_and_persona(monkeypatch)
     kwargs = updater._update_style_based_features_with_maibot.await_args.kwargs
     assert kwargs["group_id"] == "123456789"
     assert kwargs["persona_id"] == "鲸娘"
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_expression_learning_merges_stored_bot_replies_for_batch(tmp_path):
+    """回归 issue #257：批量/审批链路只传用户消息时，应合并数据库 Bot 回复后才能学到模式。"""
+    from self_learning_EterU.models.orm.message import BotMessage
+
+    config = PluginConfig(
+        data_dir=str(tmp_path),
+        db_type="sqlite",
+        enable_web_interface=False,
+    )
+    db = SQLAlchemyDatabaseManager(config)
+    try:
+        assert await db.start() is True
+        now = time.time()
+        base = int(now)
+        async with db.get_session() as session:
+            session.add_all(
+                [
+                    BotMessage(
+                        group_id="group-a",
+                        message=f"机器人回复内容第{idx}条确实很离谱",
+                        timestamp=base + idx * 2,
+                        created_at=base,
+                    )
+                    for idx in range(1, 6)
+                ]
+            )
+            await session.commit()
+
+        learner = ExpressionPatternLearner.__new__(ExpressionPatternLearner)
+        learner.db_manager = db
+        learner.config = config
+        learner.last_learning_times = {}
+
+        # 只有用户发言（无 bot），模拟批量/审批链路的 filtered_messages
+        user_messages = [
+            {
+                "sender_id": f"user-{idx}",
+                "sender_name": f"User {idx}",
+                "message": f"这是用户第{idx}条足够长的学习消息内容",
+                "timestamp": now + idx * 2 - 1,
+                "group_id": "group-a",
+            }
+            for idx in range(1, 6)
+        ]
+
+        ok = await learner.trigger_learning_for_group(
+            "group-a", user_messages, persona_id="bot-a"
+        )
+        assert ok is True
+        patterns = await learner.get_expression_patterns("group-a", persona_id="bot-a")
+        assert len(patterns) >= 2
+    finally:
+        await db.stop()
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_expression_learning_merge_skipped_when_bot_present_or_no_db():
+    """实时链路已预先合并 bot 消息，不应重复查询；无 db 时直接返回。"""
+    learner = ExpressionPatternLearner.__new__(ExpressionPatternLearner)
+
+    learner.db_manager = None
+    user_only = [{"sender_id": "user-a", "message": "hi", "timestamp": 1}]
+    assert await learner._merge_bot_messages_for_group("g", user_only) is user_only
+
+    def _boom():
+        raise AssertionError("不应在已含 bot 消息时查询数据库")
+
+    learner.db_manager = SimpleNamespace(get_session=_boom)
+    already_merged = [
+        {"sender_id": "user-a", "message": "hi", "timestamp": 1},
+        {"sender_id": "bot", "message": "yo", "timestamp": 2},
+    ]
+    assert (
+        await learner._merge_bot_messages_for_group("g", already_merged)
+        is already_merged
+    )
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+async def test_merge_bot_replies_bounds_to_message_time_window(tmp_path):
+    """回归审查意见：合并只取与这批用户消息时间窗口重叠的 Bot 回复，避免跨窗口误配对。"""
+    from self_learning_EterU.models.orm.message import BotMessage
+
+    config = PluginConfig(
+        data_dir=str(tmp_path),
+        db_type="sqlite",
+        enable_web_interface=False,
+    )
+    db = SQLAlchemyDatabaseManager(config)
+    try:
+        assert await db.start() is True
+        now = time.time()
+        base = int(now)
+        async with db.get_session() as session:
+            session.add_all(
+                [
+                    BotMessage(
+                        group_id="group-a",
+                        message="窗口内的有效机器人回复",
+                        timestamp=base + 2,
+                        created_at=base,
+                    ),
+                    BotMessage(
+                        group_id="group-a",
+                        message="窗口外很晚的无关机器人回复",
+                        timestamp=base + 99999,
+                        created_at=base,
+                    ),
+                ]
+            )
+            await session.commit()
+
+        learner = ExpressionPatternLearner.__new__(ExpressionPatternLearner)
+        learner.db_manager = db
+        learner.config = config
+
+        user_messages = [
+            {
+                "sender_id": f"user-{idx}",
+                "sender_name": f"User {idx}",
+                "message": f"这是用户第{idx}条足够长的学习消息",
+                "timestamp": now + idx,
+                "group_id": "group-a",
+            }
+            for idx in range(1, 6)
+        ]
+
+        merged = await learner._merge_bot_messages_for_group("group-a", user_messages)
+        bot_texts = [
+            m.get("message")
+            for m in merged
+            if isinstance(m, dict) and m.get("sender_id") == "bot"
+        ]
+        assert "窗口内的有效机器人回复" in bot_texts
+        assert "窗口外很晚的无关机器人回复" not in bot_texts
+    finally:
+        await db.stop()