diff --git a/CHANGELOG.md b/CHANGELOG.md index 99004036..5d565f4b 100644 --- a/CHANGELOG.md +++ b/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） diff --git a/services/analysis/expression_pattern_learner.py b/services/analysis/expression_pattern_learner.py index 095bf534..5a205ed6 100644 --- a/services/analysis/expression_pattern_learner.py +++ b/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) diff --git a/services/persona/persona_updater.py b/services/persona/persona_updater.py index e1144ad9..7ed4ff78 100644 --- a/services/persona/persona_updater.py +++ b/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, diff --git a/tests/unit/test_learning_chain_regressions.py b/tests/unit/test_learning_chain_regressions.py index 9e95a359..d93da7ef 100644 --- a/tests/unit/test_learning_chain_regressions.py +++ b/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()