From 78ab3e65c2d888f1b833687cbca3eabb280b3f68 Mon Sep 17 00:00:00 2001 From: liuwei Date: Thu, 2 Apr 2026 13:41:51 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84=E6=88=90=E5=91=98=E7=94=BB?= =?UTF-8?q?=E5=83=8F=E6=8F=92=E4=BB=B6=E7=9A=84=E6=97=A5/=E5=91=A8/?= =?UTF-8?q?=E6=9C=88=E5=88=86=E5=B1=82=E6=8F=90=E5=8F=96=E4=B8=8E=E5=B7=B2?= =?UTF-8?q?=E7=BB=93=E6=9D=9F=E6=97=A5=E6=9C=9F=E5=A4=84=E7=90=86=E9=80=BB?= =?UTF-8?q?=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 将成员画像能力进一步收敛到插件内部，强化按群启用、后台异步刷新、后台查看的完整链路 - 新增群维度按日批量提取能力：以群为单位按天处理一次，统一提取当天活跃成员的日级画像摘要 - 日级画像输出扩展为更适合长期累计的结构化信号，补充身份线索、技能信号、家庭线索、阶段线索、价值偏好、群内角色、决策风格等字段 - 优化提示词设计，明确要求优先提取可复用、可累计、可验证的行为线索，减少一次性情绪和短期噪声对长期画像的干扰 - 打通日 -> 周 -> 月 -> 最终画像 的分层汇总链路，让后续月度画像直接消费日/周级结构化摘要，而不是重复回扫长窗口原始消息 - 新增/完善画像融合策略：identity_traits、skill_profile、family_profile、life_stage_profile、value_profile 也纳入长期分数累计，不再仅依赖最近一次结果覆盖旧结果 - 将活跃群、活跃成员、辅助消息样本等口径统一调整为只处理已结束日期，避免当天未完结数据进入画像计算 - 调整日级批处理逻辑，默认只处理昨天及更早日期，确保不会处理当天消息 - 修复重复执行时仍然先调用 AI 再跳过的问题，改为先检查当天候选成员是否已完成生成，全部已存在时直接跳过，减少无效 AI 请求和耗时 - 增加群日批处理、周摘要、月摘要、群刷新进度等日志，方便后台定位当前刷新到哪些群、哪些成员、进度如何 - 丰富后台画像展示字段，支持查看更完整的长期画像维度与摘要统计 - 更新插件配置默认值，收敛为近 60 天启动窗口、每日滚动处理与群级日摘要模式 - 补充 message_storage 读取能力，支持按群按日提取消息，为群日批量画像与后续周期汇总提供底层数据支撑 --- .../templates/contacts_management.html | 22 ++ db/message_storage.py | 63 +++- plugins/member_context/config.toml | 7 +- plugins/member_context/digest_service.py | 279 +++++++++++++++--- plugins/member_context/prompt_builder.py | 103 ++++++- plugins/member_context/service.py | 128 +++++++- 6 files changed, 536 insertions(+), 66 deletions(-) diff --git a/admin/dashboard/templates/contacts_management.html b/admin/dashboard/templates/contacts_management.html index 0d53d1c..834a1a1 100644 --- a/admin/dashboard/templates/contacts_management.html +++ b/admin/dashboard/templates/contacts_management.html @@ -294,6 +294,26 @@ {% raw %}{{ item }}{% endraw %} - + + {% raw %}{{ item }}{% endraw %} + - + + + {% raw %}{{ item }}{% endraw %} + - + + + {% raw %}{{ item }}{% endraw %} + - + + + {% raw %}{{ item }}{% endraw %} + - + + + {% raw %}{{ item }}{% endraw %} + - + {% raw %}{{ item }}{% endraw %} - @@ -306,6 +326,8 @@ {% raw %}{{ item }}{% endraw %} - + {% raw %}{{ ((memberContext.meta || {}).group_role) || '-' }}{% endraw %} + {% raw %}{{ ((memberContext.meta || {}).decision_profile) || '-' }}{% endraw %} {% raw %}{{ item }}{% endraw %} - diff --git a/db/message_storage.py b/db/message_storage.py index 22b9dd0..56d20ab 100644 --- a/db/message_storage.py +++ b/db/message_storage.py @@ -43,21 +43,39 @@ class MessageStorageDB(BaseDBOperator): params = (hours_ago, group_id, min_content_length) return self.execute_query(sql, params) or [] - def get_member_recent_messages(self, group_id: str, wxid: str, days: int = 30, limit: int = 200) -> List[Dict]: + def get_member_recent_messages(self, group_id: str, wxid: str, days: int = 30, + limit: int = 200, include_today: bool = True) -> List[Dict]: """获取指定群成员近期消息""" - sql = """ - SELECT timestamp, sender, content, message_type - FROM messages - WHERE timestamp >= DATE_SUB(NOW(), INTERVAL %s DAY) - AND group_id = %s - AND sender = %s - AND message_type IN (1, 49) - AND CHAR_LENGTH(content) BETWEEN 2 AND 500 - AND content NOT LIKE '/%%' - ORDER BY timestamp DESC - LIMIT %s - """ - results = self.execute_query(sql, (days, group_id, wxid, limit)) or [] + if include_today: + sql = """ + SELECT timestamp, sender, content, message_type + FROM messages + WHERE timestamp >= DATE_SUB(NOW(), INTERVAL %s DAY) + AND group_id = %s + AND sender = %s + AND message_type IN (1, 49) + AND CHAR_LENGTH(content) BETWEEN 2 AND 500 + AND content NOT LIKE '/%%' + ORDER BY timestamp DESC + LIMIT %s + """ + params = (days, group_id, wxid, limit) + else: + sql = """ + SELECT timestamp, sender, content, message_type + FROM messages + WHERE timestamp >= DATE_SUB(CURDATE(), INTERVAL %s DAY) + AND timestamp < CURDATE() + AND group_id = %s + AND sender = %s + AND message_type IN (1, 49) + AND CHAR_LENGTH(content) BETWEEN 2 AND 500 + AND content NOT LIKE '/%%' + ORDER BY timestamp DESC + LIMIT %s + """ + params = (days, group_id, wxid, limit) + results = self.execute_query(sql, params) or [] return list(reversed(results)) def get_member_messages_since(self, group_id: str, wxid: str, since_time, limit: int = 200) -> List[Dict]: @@ -122,6 +140,23 @@ class MessageStorageDB(BaseDBOperator): """ return self.execute_query(sql, (target_date, group_id, wxid, limit)) or [] + def get_member_messages_for_group_date(self, group_id: str, target_date: str, limit: int = 5000) -> List[Dict]: + """获取群在某一天的全部文本消息""" + sql = """ + SELECT timestamp, sender, content, message_type + FROM messages + WHERE DATE(timestamp) = %s + AND group_id = %s + AND sender IS NOT NULL + AND sender <> '' + AND message_type IN (1, 49) + AND CHAR_LENGTH(content) BETWEEN 2 AND 500 + AND content NOT LIKE '/%%' + ORDER BY timestamp ASC + LIMIT %s + """ + return self.execute_query(sql, (target_date, group_id, limit)) or [] + def get_message_count_by_date(self, date: str) -> List[Dict]: """获取指定日期的消息统计""" sql = """ diff --git a/plugins/member_context/config.toml b/plugins/member_context/config.toml index 46974ef..413bfb8 100644 --- a/plugins/member_context/config.toml +++ b/plugins/member_context/config.toml @@ -12,21 +12,22 @@ request_timeout = 60 sample_days = 30 sample_message_limit = 80 refresh_limit_per_member = 200 -long_term_days = 365 +long_term_days = 60 long_term_message_limit = 600 -bootstrap_days = 365 +bootstrap_days = 60 bootstrap_message_limit = 600 incremental_message_limit = 80 incremental_recent_days = 7 recalibration_days = 30 daily_message_limit = 120 daily_digest_min_messages = 6 -max_daily_digests_per_run = 45 +max_daily_digests_per_run = 0 weekly_digest_limit = 16 monthly_digest_limit = 12 final_daily_limit = 8 final_weekly_limit = 6 final_monthly_limit = 6 +group_digest_days = 1 ai_min_member_messages = 12 active_member_hours = 72 min_member_messages = 3 diff --git a/plugins/member_context/digest_service.py b/plugins/member_context/digest_service.py index 310a65a..87b8dfc 100644 --- a/plugins/member_context/digest_service.py +++ b/plugins/member_context/digest_service.py @@ -12,6 +12,7 @@ from db.contacts_db import ContactsDBOperator from db.member_digest_db import MemberDigestDBOperator from db.message_storage import MessageStorageDB from plugins.member_context.prompt_builder import MemberContextPromptBuilder +from utils.compress_chat_data import compress_chat_data class MemberDigestService: @@ -37,28 +38,47 @@ class MemberDigestService: self.bootstrap_days = int(profile_config.get("bootstrap_days", 365)) self.daily_message_limit = int(profile_config.get("daily_message_limit", 120)) self.daily_digest_min_messages = int(profile_config.get("daily_digest_min_messages", 6)) - self.max_daily_digests_per_run = int(profile_config.get("max_daily_digests_per_run", 45)) + self.max_daily_digests_per_run = int(profile_config.get("max_daily_digests_per_run", 0)) self.weekly_digest_limit = int(profile_config.get("weekly_digest_limit", 16)) self.monthly_digest_limit = int(profile_config.get("monthly_digest_limit", 12)) self.final_daily_limit = int(profile_config.get("final_daily_limit", 8)) self.final_weekly_limit = int(profile_config.get("final_weekly_limit", 6)) self.final_monthly_limit = int(profile_config.get("final_monthly_limit", 6)) + self.group_digest_days = int(profile_config.get("group_digest_days", 1)) + + def ensure_recent_group_daily_digests(self, chatroom_id: str, days: Optional[int] = None, + force: bool = False) -> Dict: + days = days or self.group_digest_days + built_daily = 0 + touched_members = set() + + for offset in range(days): + target_date = (datetime.now() - timedelta(days=offset + 1)).strftime("%Y-%m-%d") + digests = self._build_group_daily_digests(chatroom_id, target_date, force=force) + for digest in digests: + self.digest_db.save_digest(digest) + built_daily += 1 + touched_members.add(digest.get("wxid")) + + return { + "built_daily": built_daily, + "touched_members": list(touched_members), + } def ensure_member_digest_pipeline(self, chatroom_id: str, wxid: str, force: bool = False) -> Dict: member = self.contacts_db.get_chatroom_member_info(chatroom_id, wxid) or {} display_name = member.get("display_name") or member.get("nick_name") or wxid - active_dates = self.message_db.get_member_active_dates(chatroom_id, wxid, days=self.bootstrap_days) - if not active_dates: + daily_digests = self.digest_db.list_digests(chatroom_id, wxid, "daily", limit=400) + if not daily_digests: return { "display_name": display_name, "daily_digests": [], "weekly_digests": [], "monthly_digests": [], - "stats": {"daily": 0, "weekly": 0, "monthly": 0, "active_days": 0}, + "stats": {"daily": 0, "weekly": 0, "monthly": 0, "active_days": 0, "built_daily": 0}, } - built_daily = self._ensure_daily_digests(chatroom_id, wxid, display_name, active_dates, force=force) built_weekly = self._ensure_weekly_digests(chatroom_id, wxid, display_name, force=force) built_monthly = self._ensure_monthly_digests(chatroom_id, wxid, display_name, force=force) @@ -75,45 +95,37 @@ class MemberDigestService: "daily": len(daily_digests), "weekly": len(weekly_digests), "monthly": len(monthly_digests), - "active_days": len(active_dates), - "built_daily": built_daily, + "active_days": len(self.digest_db.list_digest_keys(chatroom_id, wxid, "daily")), + "built_daily": 0, "built_weekly": built_weekly, "built_monthly": built_monthly, }, } - def _ensure_daily_digests(self, chatroom_id: str, wxid: str, display_name: str, - active_dates: List[Dict], force: bool = False) -> int: - existing_keys = set(self.digest_db.list_digest_keys(chatroom_id, wxid, "daily")) - built = 0 - processed = 0 - sorted_dates = sorted(active_dates, key=lambda item: str(item.get("message_date"))) - current_day = datetime.now().strftime("%Y-%m-%d") - - for item in sorted_dates: - period_key = str(item.get("message_date")) - msg_count = int(item.get("msg_count", 0)) - if msg_count < self.daily_digest_min_messages: - continue - if not force and period_key in existing_keys and period_key != current_day: - continue - messages = self.message_db.get_member_messages_on_date( - chatroom_id, wxid, period_key, limit=self.daily_message_limit - ) - if len(messages) < self.daily_digest_min_messages: - continue - digest = self._build_daily_digest(chatroom_id, wxid, display_name, period_key, messages) - if digest: - self.digest_db.save_digest(digest) - built += 1 - processed += 1 - self.LOG.info( - f"[成员交互摘要][日摘要] 完成: group={chatroom_id}, wxid={wxid}, " - f"date={period_key}, messages={len(messages)}" - ) - if not force and processed >= self.max_daily_digests_per_run: - break - return built + @staticmethod + def _normalize_profile_item(item: Dict) -> Dict: + normalized = {} + list_keys = { + "topics", "identity_clues", "skill_signals", "family_signals", "life_stage_signals", + "value_preferences", "habit_signals", "engagement_traits", "reply_taboos", + "representative_messages", "stable_topics", "identity_traits", "skill_profile", + "family_profile", "life_stage_profile", "value_profile", "stable_traits", + "habit_patterns", "reply_preferences", "long_term_topics", "long_term_reply_preferences", + "phase_state", "recent_state" + } + for key, value in item.items(): + if key in list_keys: + if isinstance(value, list): + normalized[key] = [str(v).strip() for v in value if str(v).strip()] + elif value: + normalized[key] = [str(value).strip()] + else: + normalized[key] = [] + elif isinstance(value, (int, float)): + normalized[key] = value + else: + normalized[key] = str(value).strip() + return normalized def _ensure_weekly_digests(self, chatroom_id: str, wxid: str, display_name: str, force: bool = False) -> int: daily_digests = self.digest_db.list_digests(chatroom_id, wxid, "daily", limit=400) @@ -123,7 +135,7 @@ class MemberDigestService: grouped[week_key].append(item) existing_keys = set(self.digest_db.list_digest_keys(chatroom_id, wxid, "weekly")) - current_week_key, _, _ = self._week_period_bounds(datetime.now().strftime("%Y-%m-%d")) + current_week_key, _, _ = self._week_period_bounds((datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")) built = 0 for week_key, items in sorted(grouped.items()): if len(items) < 2: @@ -151,7 +163,7 @@ class MemberDigestService: grouped[month_key].append(item) existing_keys = set(self.digest_db.list_digest_keys(chatroom_id, wxid, "monthly")) - current_month_key, _, _ = self._month_period_bounds(datetime.now().strftime("%Y-%m-%d")) + current_month_key, _, _ = self._month_period_bounds((datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")) built = 0 for month_key, items in sorted(grouped.items()): if len(items) < 2: @@ -171,6 +183,83 @@ class MemberDigestService: ) return built + def _build_group_daily_digests(self, chatroom_id: str, digest_date: str, force: bool = False) -> List[Dict]: + members = self.contacts_db.get_chatroom_member_list(chatroom_id) or [] + member_name_map = {} + for member in members: + wxid = member.get("wxid") + if not wxid: + continue + member_name_map[wxid] = member.get("display_name") or member.get("nick_name") or wxid + + messages = self.message_db.get_member_messages_for_group_date(chatroom_id, digest_date) + if not messages: + return [] + + sender_messages = defaultdict(list) + for msg in messages: + wxid = msg.get("sender") + if not wxid: + continue + sender_messages[wxid].append(msg) + + candidate_wxids = [ + wxid for wxid, items in sender_messages.items() + if len(items) >= self.daily_digest_min_messages + ] + if not candidate_wxids: + return [] + + pending_wxids = [] + for wxid in candidate_wxids: + if not force and self.digest_db.get_digest(chatroom_id, wxid, "daily", digest_date): + continue + pending_wxids.append(wxid) + if not pending_wxids: + self.LOG.info( + f"[成员交互摘要][群日批处理] 跳过: group={chatroom_id}, date={digest_date}, " + f"reason=all_candidates_already_built, candidates={len(candidate_wxids)}" + ) + return [] + + member_labels = [f"{wxid} | {member_name_map.get(wxid, wxid)}" for wxid in pending_wxids] + compact_chat = self._format_group_messages_optimized(messages, member_name_map) + try: + compact_chat = compress_chat_data(compact_chat) + except Exception as e: + self.LOG.warning(f"[成员交互摘要] 压缩群日消息失败: group={chatroom_id}, date={digest_date}, error={e}") + + parsed_members = self._request_group_daily_json(chatroom_id, digest_date, member_labels, compact_chat) + parsed_map = {item.get("wxid"): item for item in parsed_members if item.get("wxid")} if parsed_members else {} + + digests = [] + for wxid in pending_wxids: + parsed = parsed_map.get(wxid) or self._build_daily_digest_fallback(sender_messages.get(wxid, [])) + if not parsed: + continue + parsed = self._normalize_profile_item(parsed) + digests.append({ + "chatroom_id": chatroom_id, + "wxid": wxid, + "digest_type": "daily", + "period_key": digest_date, + "period_start": f"{digest_date} 00:00:00", + "period_end": f"{digest_date} 23:59:59", + "display_name": member_name_map.get(wxid, wxid), + "source_count": len(sender_messages.get(wxid, [])), + "summary_text": parsed.get("summary_text", ""), + "structured": parsed, + "meta": { + "source_type": "group_daily_messages", + "representative_messages": parsed.get("representative_messages", []), + }, + }) + self.LOG.info( + f"[成员交互摘要][群日批处理] 完成: group={chatroom_id}, date={digest_date}, " + f"candidates={len(candidate_wxids)}, pending={len(pending_wxids)}, built={len(digests)}" + ) + return digests + def _build_daily_digest(self, chatroom_id: str, wxid: str, display_name: str, digest_date: str, messages: List[Dict]) -> Optional[Dict]: prompt = MemberContextPromptBuilder.build_daily_digest_prompt( @@ -256,6 +345,39 @@ class MemberDigestService: self.LOG.warning(f"[成员交互摘要][AI] 摘要请求失败: group={chatroom_id}, wxid={wxid}, tag={tag}, error={e}") return None + def _request_group_daily_json(self, chatroom_id: str, digest_date: str, + member_labels: List[str], compressed_chat: str) -> List[Dict]: + if not self.ai_enabled or not self.ai_base_url or not self.ai_api_key: + return [] + prompt = MemberContextPromptBuilder.build_group_daily_digest_prompt( + chatroom_id, digest_date, member_labels, compressed_chat + ) + headers = { + "Authorization": f"Bearer {self.ai_api_key}", + "Content-Type": "application/json", + } + payload = { + "inputs": {"query": prompt}, + "response_mode": "blocking", + "user": f"member-digest:{chatroom_id}:group-daily:{digest_date}", + } + url = f"{self.ai_base_url}/{self.ai_endpoint}" + try: + self.LOG.info( + f"[成员交互摘要][AI] 发起群日批量摘要请求: group={chatroom_id}, " + f"date={digest_date}, members={len(member_labels)}" + ) + response = requests.post(url, headers=headers, json=payload, timeout=self.ai_timeout) + response.raise_for_status() + data = response.json() + parsed = self._parse_group_daily_answer(data.get("answer", "")) + return parsed + except Exception as e: + self.LOG.warning( + f"[成员交互摘要][AI] 群日批量摘要失败: group={chatroom_id}, date={digest_date}, error={e}" + ) + return [] + def _parse_ai_answer(self, answer: str) -> Optional[Dict]: if not answer: return None @@ -277,6 +399,29 @@ class MemberDigestService: normalized[key] = str(value).strip() return normalized + def _parse_group_daily_answer(self, answer: str) -> List[Dict]: + if not answer: + return [] + text = answer.strip() + match = re.search(r"\{.*\}", text, re.S) + if match: + text = match.group(0) + try: + parsed = json.loads(text) + except Exception: + return [] + members = parsed.get("members", []) + if not isinstance(members, list): + return [] + normalized = [] + for item in members: + if not isinstance(item, dict): + continue + normalized_item = self._normalize_profile_item(item) + if normalized_item.get("wxid"): + normalized.append(normalized_item) + return normalized + def _build_daily_digest_fallback(self, messages: List[Dict]) -> Optional[Dict]: if not messages: return None @@ -288,11 +433,18 @@ class MemberDigestService: message_pattern = "短句居多" if avg_len <= 16 else "表达较完整" if avg_len >= 35 else "表达中等长度" return { "topics": [], + "identity_clues": [], + "skill_signals": [], + "family_signals": [], + "life_stage_signals": [], + "value_preferences": [], "interaction_style": "自然跟随式互动", "message_pattern": message_pattern, "response_style_hint": "保持简洁自然，先回应核心点", "habit_signals": [], "engagement_traits": [], + "decision_style": "", + "social_role": "", "reply_taboos": [], "temperament_signal": "当天样本有限，暂以中性沟通观察为主", "summary_text": f"当日消息约{len(messages)}条，{message_pattern}。", @@ -332,9 +484,16 @@ class MemberDigestService: if digest_type == "weekly": return { "stable_topics": top_topics, + "identity_traits": [], + "skill_profile": [], + "family_profile": [], + "life_stage_profile": [], + "value_profile": [], "stable_traits": top_traits, "habit_patterns": top_habits, "reply_preferences": top_reply, + "group_role": "", + "decision_profile": "", "recent_state": top_topics[:3], "temperament_tendency": temperament, "summary_text": "本周沟通特征已按重复信号汇总。", @@ -343,15 +502,53 @@ class MemberDigestService: return { "long_term_topics": top_topics, + "identity_traits": [], + "skill_profile": [], + "family_profile": [], + "life_stage_profile": [], + "value_profile": [], "stable_traits": top_traits, "habit_patterns": top_habits, "long_term_reply_preferences": top_reply, + "group_role": "", + "decision_profile": "", "phase_state": top_topics[:3], "temperament_tendency": temperament, "summary_text": "本月沟通特征已按周摘要汇总。", "confidence": 0.5, } + def _format_group_messages_optimized(self, messages: List[Dict], member_name_map: Dict[str, str]) -> str: + if not messages: + return "" + time_groups = defaultdict(lambda: defaultdict(list)) + for msg in messages: + timestamp = msg.get("timestamp") + sender = msg.get("sender", "") + content = str(msg.get("content", "")).strip() + if not sender or not content: + continue + try: + dt = datetime.strptime(str(timestamp), "%Y-%m-%d %H:%M:%S") + except Exception: + dt = None + if not dt: + continue + time_key = dt.strftime("%H:%M") + sender_name = member_name_map.get(sender, sender) + time_groups[time_key][sender_name].append(content) + + lines = [] + for time_key in sorted(time_groups.keys()): + lines.append(f"【{time_key}】") + for sender_name, contents in time_groups[time_key].items(): + for idx, content in enumerate(contents): + if idx == 0: + lines.append(f"{sender_name}：{content}") + else: + lines.append(f" {content}") + return "\n".join(lines) + @staticmethod def _week_period_bounds(date_value: str) -> Tuple[str, str, str]: target_date = datetime.strptime(str(date_value)[:10], "%Y-%m-%d") diff --git a/plugins/member_context/prompt_builder.py b/plugins/member_context/prompt_builder.py index 1b17d90..e5693a9 100644 --- a/plugins/member_context/prompt_builder.py +++ b/plugins/member_context/prompt_builder.py @@ -6,6 +6,58 @@ from typing import Dict, List class MemberContextPromptBuilder: """成员分层画像提示词构建器""" + @staticmethod + def build_group_daily_digest_prompt(chatroom_id: str, digest_date: str, + member_labels: List[str], compressed_chat: str) -> str: + return ( + "你是微信群后台的成员日观察批量提取器。\n" + "请基于给定的一天群聊记录，识别当天有参与发言的成员，并分别提取中性、克制的成员日观察摘要。\n" + "不要做心理诊断、隐私猜测、负面评价，不要输出未在候选名单中的成员。\n" + "你的输出将被后续系统按周、按月持续累积，因此优先提取可复用、可累计、可验证的行为信号，而不是一次性的情绪和玩笑。\n" + "输出严格 JSON，不要 markdown。\n" + "{" + "\"members\":[" + "{" + "\"wxid\":\"成员wxid\"," + "\"display_name\":\"成员显示名\"," + "\"topics\":[\"主题1\"]," + "\"identity_clues\":[\"身份线索1\"]," + "\"skill_signals\":[\"技能信号1\"]," + "\"family_signals\":[\"家庭线索1\"]," + "\"life_stage_signals\":[\"阶段线索1\"]," + "\"value_preferences\":[\"价值偏好1\"]," + "\"interaction_style\":\"一句中文\"," + "\"message_pattern\":\"一句中文\"," + "\"response_style_hint\":\"一句中文\"," + "\"habit_signals\":[\"信号1\"]," + "\"engagement_traits\":[\"特征1\"]," + "\"decision_style\":\"一句中文\"," + "\"social_role\":\"一句中文，描述当天在群中的角色表现\"," + "\"reply_taboos\":[\"避坑1\"]," + "\"temperament_signal\":\"一句中文，描述当天沟通倾向，必须克制\"," + "\"summary_text\":\"一段不超过100字的日摘要\"," + "\"representative_messages\":[\"原话1\",\"原话2\"]," + "\"confidence\":0.0" + "}" + "]" + "}\n" + "要求：\n" + "1. 只输出当天真正参与发言且能看出明确行为信号的成员；发言极少的人可以不输出。\n" + "2. 每个成员的 topics、identity_clues、skill_signals、family_signals、life_stage_signals、value_preferences、habit_signals、engagement_traits 最多4个，reply_taboos 最多3个。\n" + "3. representative_messages 只保留最能代表当天表达方式的短句，最多3条。\n" + "4. 必须严格使用候选成员列表中的 wxid 和显示名。\n" + "5. identity_clues、family_signals、life_stage_signals 只能写公开聊天中出现的线索，不可把弱线索写成确定事实。\n" + "6. skill_signals 重点提炼成员解决问题、提供信息、组织表达、专业能力等信号。\n" + "7. social_role 只描述当天在群里的角色表现，例如：问题提出者、信息补充者、气氛调节者、组织推进者。\n" + "8. topics 更偏向持续关注的话题方向；habit_signals 更偏向重复表达或互动习惯；engagement_traits 更偏向参与方式。\n" + "9. value_preferences 只记录公开表达出的偏好，如效率优先、成本敏感、谨慎验证、乐于助人，不要写抽象大词。\n" + "10. summary_text 应是后台观察摘要，不要写成对用户说的话。\n" + f"群ID: {chatroom_id}\n" + f"日期: {digest_date}\n" + "候选成员:\n" + "\n".join(member_labels[:80]) + "\n" + "压缩后的群聊记录:\n" + compressed_chat + ) + @staticmethod def build_daily_digest_prompt(chatroom_id: str, wxid: str, display_name: str, digest_date: str, messages: List[Dict]) -> str: @@ -20,14 +72,22 @@ class MemberContextPromptBuilder: "你是微信群后台的成员日观察摘要生成器。\n" "请仅基于给定的当日公开聊天记录，提取对后续互动有帮助的中性行为观察。\n" "不要做人格诊断、隐私猜测、负面评价，不要脑补群外信息。\n" + "这些日观察会被后续系统按周、按月持续累积，所以应优先输出长期可验证的行为线索，而不是一次性情绪。\n" "输出严格 JSON，不要 markdown。\n" "{" "\"topics\":[\"主题1\"]," + "\"identity_clues\":[\"身份线索1\"]," + "\"skill_signals\":[\"技能信号1\"]," + "\"family_signals\":[\"家庭线索1\"]," + "\"life_stage_signals\":[\"阶段线索1\"]," + "\"value_preferences\":[\"价值偏好1\"]," "\"interaction_style\":\"一句中文\"," "\"message_pattern\":\"一句中文\"," "\"response_style_hint\":\"一句中文\"," "\"habit_signals\":[\"信号1\"]," "\"engagement_traits\":[\"特征1\"]," + "\"decision_style\":\"一句中文\"," + "\"social_role\":\"一句中文\"," "\"reply_taboos\":[\"避坑1\"]," "\"temperament_signal\":\"一句中文，描述当天显露的沟通倾向，必须克制\"," "\"summary_text\":\"一段不超过100字的日摘要\"," @@ -35,9 +95,14 @@ class MemberContextPromptBuilder: "\"confidence\":0.0" "}\n" "要求：\n" - "1. topics 最多4个，habit_signals 最多4个，engagement_traits 最多4个，reply_taboos 最多3个。\n" + "1. topics、identity_clues、skill_signals、family_signals、life_stage_signals、value_preferences、habit_signals、engagement_traits 最多4个，reply_taboos 最多3个。\n" "2. temperament_signal 只能写当日可观察到的沟通倾向，不可上升为长期性格判断。\n" "3. representative_messages 保留最能代表当天风格的短句，最多3条。\n" + "4. identity_clues、family_signals、life_stage_signals 只能写线索，不可写成确定事实。\n" + "5. skill_signals 重点描述专业能力、工具熟练度、信息组织能力、问题解决能力等当天显露出的信号。\n" + "6. topics 尽量写持续关注方向，避免写一次性插话；habit_signals 只写当天已明显出现的表达或互动习惯。\n" + "7. value_preferences 只保留公开表达出的判断偏好，如效率优先、先验证再决策、重成本、重稳定。\n" + "8. summary_text 要像后台备注，不要像对话回复。\n" f"成员: {display_name} ({wxid})\n" f"群ID: {chatroom_id}\n" f"日期: {digest_date}\n" @@ -54,9 +119,16 @@ class MemberContextPromptBuilder: "period_key": item.get("period_key"), "summary_text": item.get("summary_text", ""), "topics": structured.get("topics") or structured.get("stable_topics") or structured.get("long_term_topics") or [], + "identity_clues": structured.get("identity_clues") or structured.get("identity_traits") or [], + "skill_signals": structured.get("skill_signals") or structured.get("skill_profile") or [], + "family_signals": structured.get("family_signals") or structured.get("family_profile") or [], + "life_stage_signals": structured.get("life_stage_signals") or structured.get("life_stage_profile") or [], + "value_preferences": structured.get("value_preferences") or structured.get("value_profile") or [], "habit_signals": structured.get("habit_signals") or structured.get("habit_patterns") or [], "engagement_traits": structured.get("engagement_traits") or structured.get("stable_traits") or [], "reply_preferences": structured.get("reply_preferences") or structured.get("long_term_reply_preferences") or [], + "social_role": structured.get("social_role") or structured.get("group_role") or "", + "decision_style": structured.get("decision_style") or structured.get("decision_profile") or "", "temperament_signal": structured.get("temperament_signal") or structured.get("temperament_tendency") or "", "recent_state": structured.get("recent_state") or [], } @@ -66,9 +138,16 @@ class MemberContextPromptBuilder: schema = ( "{" "\"stable_topics\":[\"主题1\"]," + "\"identity_traits\":[\"身份特征1\"]," + "\"skill_profile\":[\"技能画像1\"]," + "\"family_profile\":[\"家庭线索1\"]," + "\"life_stage_profile\":[\"阶段线索1\"]," + "\"value_profile\":[\"价值偏好1\"]," "\"stable_traits\":[\"特征1\"]," "\"habit_patterns\":[\"习惯1\"]," "\"reply_preferences\":[\"偏好1\"]," + "\"group_role\":\"一句中文\"," + "\"decision_profile\":\"一句中文\"," "\"recent_state\":[\"状态1\"]," "\"temperament_tendency\":\"一句中文\"," "\"summary_text\":\"一段不超过120字的周摘要\"," @@ -80,9 +159,16 @@ class MemberContextPromptBuilder: schema = ( "{" "\"long_term_topics\":[\"主题1\"]," + "\"identity_traits\":[\"身份特征1\"]," + "\"skill_profile\":[\"技能画像1\"]," + "\"family_profile\":[\"家庭线索1\"]," + "\"life_stage_profile\":[\"阶段线索1\"]," + "\"value_profile\":[\"价值偏好1\"]," "\"stable_traits\":[\"特征1\"]," "\"habit_patterns\":[\"习惯1\"]," "\"long_term_reply_preferences\":[\"偏好1\"]," + "\"group_role\":\"一句中文\"," + "\"decision_profile\":\"一句中文\"," "\"phase_state\":[\"状态1\"]," "\"temperament_tendency\":\"一句中文\"," "\"summary_text\":\"一段不超过140字的月摘要\"," @@ -100,6 +186,10 @@ class MemberContextPromptBuilder: "1. 所有列表字段最多5项，必须中性克制。\n" "2. 只有多个下级摘要反复出现的特征，才允许写进 stable_traits / habit_patterns / long_term_reply_preferences。\n" "3. recent_state / phase_state 只描述当前阶段状态，不要冒充长期人格。\n" + "4. identity_traits、family_profile、life_stage_profile 只能保留反复出现的公开线索，不可编造事实。\n" + "5. skill_profile 要优先提炼稳定出现的能力、专业方向、擅长处理的问题类型。\n" + "6. group_role 描述其在群中的长期角色位置，decision_profile 描述其决策与判断风格。\n" + "7. value_profile 需要优先保留真正反复出现的判断偏好，如效率优先、成本敏感、风险谨慎、愿意分享。\n" f"成员: {display_name} ({wxid})\n" f"群ID: {chatroom_id}\n" f"周期: {period_key}\n" @@ -125,9 +215,16 @@ class MemberContextPromptBuilder: "\"response_style_hint\":\"一句中文\"," "\"topics_of_interest\":[\"主题1\"]," "\"recent_focus\":[\"近期主题1\"]," + "\"identity_traits\":[\"身份线索1\"]," + "\"skill_profile\":[\"技能画像1\"]," + "\"family_profile\":[\"家庭线索1\"]," + "\"life_stage_profile\":[\"阶段线索1\"]," + "\"value_profile\":[\"价值偏好1\"]," "\"stable_traits\":[\"长期特征1\"]," "\"habit_patterns\":[\"习惯1\"]," "\"long_term_reply_preferences\":[\"偏好1\"]," + "\"group_role\":\"一句中文\"," + "\"decision_profile\":\"一句中文\"," "\"recent_state\":[\"近期状态1\"]," "\"temperament_tendency\":\"一句中文\"," "\"summary_text\":\"一段不超过150字的后台摘要\"," @@ -139,6 +236,10 @@ class MemberContextPromptBuilder: "1. stable_traits、habit_patterns、long_term_reply_preferences 只从月级和多次重复证据中提取。\n" "2. recent_focus、recent_state 更依赖最近周级和日级。\n" "3. summary_text 要像后台备注，不要明显暴露在给用户做画像。\n" + "4. identity_traits、family_profile、life_stage_profile 必须写成公开线索或长期观察，不得伪造事实。\n" + "5. skill_profile 要尽量覆盖专业能力、问题解决能力、表达组织能力、资源协调能力等维度。\n" + "6. group_role 要描述其在群中的角色定位，decision_profile 要描述其决策/判断方式。\n" + "7. 如果月级与周级证据不足，宁可少写，也不要把短期状态写成长期人格。\n" f"成员: {display_name} ({wxid})\n" f"群ID: {chatroom_id}\n" "月级摘要:\n" + ("\n".join(monthly_lines) or "暂无") diff --git a/plugins/member_context/service.py b/plugins/member_context/service.py index 580ec1c..a4b7775 100644 --- a/plugins/member_context/service.py +++ b/plugins/member_context/service.py @@ -67,7 +67,8 @@ class MemberContextService: self.min_group_messages = int(schedule_config.get("min_group_messages", 20)) def build_member_context(self, chatroom_id: str, wxid: str, days: Optional[int] = None, - limit: Optional[int] = None, force_digest_rebuild: bool = False) -> Dict: + limit: Optional[int] = None, force_digest_rebuild: bool = False, + ensure_group_daily: bool = True) -> Dict: days = days or self.sample_days limit = limit or self.refresh_limit_per_member @@ -75,6 +76,11 @@ class MemberContextService: member = self.contacts_db.get_chatroom_member_info(chatroom_id, wxid) or {} display_name = member.get("display_name") or member.get("nick_name") or wxid + group_digest_stats = {"built_daily": 0, "touched_members": []} + if ensure_group_daily: + group_digest_stats = self.digest_service.ensure_recent_group_daily_digests( + chatroom_id, force=force_digest_rebuild + ) digest_snapshot = self.digest_service.ensure_member_digest_pipeline( chatroom_id, wxid, force=force_digest_rebuild ) @@ -82,7 +88,13 @@ class MemberContextService: weekly_digests = digest_snapshot.get("weekly_digests", []) monthly_digests = digest_snapshot.get("monthly_digests", []) - recent_messages = self.message_db.get_member_recent_messages(chatroom_id, wxid, days=min(days, 7), limit=120) + recent_messages = self.message_db.get_member_recent_messages( + chatroom_id, + wxid, + days=min(days, 7), + limit=120, + include_today=False, + ) monthly_structured = [item.get("structured", {}) or {} for item in monthly_digests] weekly_structured = [item.get("structured", {}) or {} for item in weekly_digests] daily_structured = [item.get("structured", {}) or {} for item in daily_digests] @@ -113,6 +125,26 @@ class MemberContextService: "source_days": days, "last_profiled_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "meta": { + "identity_traits": self._extract_scored_items( + monthly_structured + weekly_structured + daily_structured, + ["identity_traits", "identity_clues"], limit=5 + ), + "skill_profile": self._extract_scored_items( + monthly_structured + weekly_structured + daily_structured, + ["skill_profile", "skill_signals"], limit=6 + ), + "family_profile": self._extract_scored_items( + monthly_structured + weekly_structured + daily_structured, + ["family_profile", "family_signals"], limit=4 + ), + "life_stage_profile": self._extract_scored_items( + monthly_structured + weekly_structured + daily_structured, + ["life_stage_profile", "life_stage_signals"], limit=4 + ), + "value_profile": self._extract_scored_items( + monthly_structured + weekly_structured + daily_structured, + ["value_profile", "value_preferences"], limit=5 + ), "stable_traits": self._extract_scored_items( monthly_structured + weekly_structured, ["stable_traits", "engagement_traits"], limit=self.stable_max_items ), @@ -130,6 +162,14 @@ class MemberContextService: monthly_structured + weekly_structured + daily_structured, ["temperament_tendency", "temperament_signal"], default="" ), + "group_role": self._best_text( + monthly_structured + weekly_structured + daily_structured, + ["group_role", "social_role"], default="" + ), + "decision_profile": self._best_text( + monthly_structured + weekly_structured + daily_structured, + ["decision_profile", "decision_style"], default="" + ), "engagement_traits": self._extract_scored_items( daily_structured + weekly_structured, ["engagement_traits", "stable_traits"], limit=4 ), @@ -144,7 +184,7 @@ class MemberContextService: "last_daily_digest_at": daily_digests[0].get("last_generated_at") if daily_digests else "", "last_weekly_digest_at": weekly_digests[0].get("last_generated_at") if weekly_digests else "", "last_monthly_digest_at": monthly_digests[0].get("last_generated_at") if monthly_digests else "", - "refresh_mode": self._build_refresh_mode(existing_context, digest_snapshot), + "refresh_mode": self._build_refresh_mode(existing_context, digest_snapshot, group_digest_stats), }, } @@ -214,6 +254,13 @@ class MemberContextService: f"days={days}, limit_per_member={limit_per_member}" ) + group_digest_stats = self.digest_service.ensure_recent_group_daily_digests(chatroom_id) + self.LOG.info( + f"[成员交互摘要] 群日摘要批处理完成: group={chatroom_id}, " + f"built_daily={group_digest_stats.get('built_daily', 0)}, " + f"touched_members={len(group_digest_stats.get('touched_members', []))}" + ) + for index, active_member in enumerate(active_members, start=1): wxid = active_member.get("wxid") if wxid not in enabled_members: @@ -231,7 +278,9 @@ class MemberContextService: f"last_profiled_at={(existing_context or {}).get('last_profiled_at')}" ) continue - context = self.build_member_context(chatroom_id, wxid, days=days, limit=limit_per_member) + context = self.build_member_context( + chatroom_id, wxid, days=days, limit=limit_per_member, ensure_group_daily=False + ) if context["source_message_count"] <= 0 and context.get("meta", {}).get("digest_daily_count", 0) <= 0: skipped += 1 self.LOG.debug( @@ -329,7 +378,8 @@ class MemberContextService: WHERE group_id = %s AND sender IS NOT NULL AND sender <> '' - AND timestamp >= DATE_SUB(NOW(), INTERVAL %s HOUR) + AND timestamp >= DATE_SUB(CURDATE(), INTERVAL %s HOUR) + AND timestamp < CURDATE() AND message_type IN (1, 49) GROUP BY sender HAVING COUNT(*) >= %s @@ -384,7 +434,8 @@ class MemberContextService: SELECT group_id, COUNT(*) AS msg_count FROM messages WHERE group_id LIKE %s - AND timestamp >= DATE_SUB(NOW(), INTERVAL %s HOUR) + AND timestamp >= DATE_SUB(CURDATE(), INTERVAL %s HOUR) + AND timestamp < CURDATE() GROUP BY group_id HAVING COUNT(*) >= %s """ @@ -473,9 +524,16 @@ class MemberContextService: "summary_text": str(data.get("summary_text", "")).strip(), "confidence": max(0.0, min(1.0, confidence)), "meta": { + "identity_traits": norm_list(data.get("identity_traits"), 5), + "skill_profile": norm_list(data.get("skill_profile"), 6), + "family_profile": norm_list(data.get("family_profile"), 4), + "life_stage_profile": norm_list(data.get("life_stage_profile"), 4), + "value_profile": norm_list(data.get("value_profile"), 5), "stable_traits": norm_list(data.get("stable_traits"), self.stable_max_items), "habit_patterns": norm_list(data.get("habit_patterns"), self.stable_max_items), "long_term_reply_preferences": norm_list(data.get("long_term_reply_preferences"), 4), + "group_role": str(data.get("group_role", "")).strip(), + "decision_profile": str(data.get("decision_profile", "")).strip(), "recent_state": norm_list(data.get("recent_state"), 4), "temperament_tendency": str(data.get("temperament_tendency", "")).strip(), "engagement_traits": norm_list(data.get("engagement_traits"), 4), @@ -510,6 +568,31 @@ class MemberContextService: meta.get("habit_patterns", []), current_context.get("confidence", 0), ) + merged_identity_scores = self._merge_scored_items( + existing_meta.get("identity_trait_scores", {}), + meta.get("identity_traits", []), + current_context.get("confidence", 0) * 0.75, + ) + merged_skill_scores = self._merge_scored_items( + existing_meta.get("skill_profile_scores", {}), + meta.get("skill_profile", []), + current_context.get("confidence", 0) * 0.85, + ) + merged_family_scores = self._merge_scored_items( + existing_meta.get("family_profile_scores", {}), + meta.get("family_profile", []), + current_context.get("confidence", 0) * 0.55, + ) + merged_life_stage_scores = self._merge_scored_items( + existing_meta.get("life_stage_profile_scores", {}), + meta.get("life_stage_profile", []), + current_context.get("confidence", 0) * 0.65, + ) + merged_value_scores = self._merge_scored_items( + existing_meta.get("value_profile_scores", {}), + meta.get("value_profile", []), + current_context.get("confidence", 0) * 0.75, + ) merged_reply_pref_scores = self._merge_scored_items( existing_meta.get("long_term_reply_preference_scores", {}), meta.get("long_term_reply_preferences", []), @@ -524,13 +607,35 @@ class MemberContextService: meta["topic_scores"] = merged_topic_scores meta["stable_trait_scores"] = merged_trait_scores meta["habit_pattern_scores"] = merged_habit_scores + meta["identity_trait_scores"] = merged_identity_scores + meta["skill_profile_scores"] = merged_skill_scores + meta["family_profile_scores"] = merged_family_scores + meta["life_stage_profile_scores"] = merged_life_stage_scores + meta["value_profile_scores"] = merged_value_scores meta["long_term_reply_preference_scores"] = merged_reply_pref_scores meta["temperament_tendency_scores"] = merged_temperament_scores + meta["identity_traits"] = self._top_scored_items(merged_identity_scores, limit=5) + meta["skill_profile"] = self._top_scored_items(merged_skill_scores, limit=6) + meta["family_profile"] = self._top_scored_items(merged_family_scores, limit=4) + meta["life_stage_profile"] = self._top_scored_items(merged_life_stage_scores, limit=4) + meta["value_profile"] = self._top_scored_items(merged_value_scores, limit=5) meta["stable_traits"] = self._top_scored_items(merged_trait_scores, limit=self.stable_max_items) meta["habit_patterns"] = self._top_scored_items(merged_habit_scores, limit=self.stable_max_items) meta["long_term_reply_preferences"] = self._top_scored_items(merged_reply_pref_scores, limit=4) temperament = self._top_scored_items(merged_temperament_scores, limit=1) meta["temperament_tendency"] = temperament[0] if temperament else meta.get("temperament_tendency", "") + if not meta["identity_traits"]: + meta["identity_traits"] = (existing_meta.get("identity_traits") or [])[:5] + if not meta["skill_profile"]: + meta["skill_profile"] = (existing_meta.get("skill_profile") or [])[:6] + if not meta["family_profile"]: + meta["family_profile"] = (existing_meta.get("family_profile") or [])[:4] + if not meta["life_stage_profile"]: + meta["life_stage_profile"] = (existing_meta.get("life_stage_profile") or [])[:4] + if not meta["value_profile"]: + meta["value_profile"] = (existing_meta.get("value_profile") or [])[:5] + meta["group_role"] = meta.get("group_role") or existing_meta.get("group_role") or "" + meta["decision_profile"] = meta.get("decision_profile") or existing_meta.get("decision_profile") or "" meta["engagement_traits"] = (meta.get("engagement_traits") or existing_meta.get("engagement_traits") or [])[:4] meta["reply_taboos"] = (meta.get("reply_taboos") or existing_meta.get("reply_taboos") or [])[:3] meta["recent_state"] = (meta.get("recent_state") or existing_meta.get("recent_state") or [])[:4] @@ -611,9 +716,12 @@ class MemberContextService: def _sum_digest_source_count(daily_digests: List[Dict]) -> int: return sum(int(item.get("source_count", 0)) for item in daily_digests) - def _build_refresh_mode(self, existing_context: Optional[Dict], digest_snapshot: Dict) -> str: + def _build_refresh_mode(self, existing_context: Optional[Dict], digest_snapshot: Dict, + group_digest_stats: Optional[Dict] = None) -> str: if not existing_context: return "bootstrap" + if (group_digest_stats or {}).get("built_daily", 0) > 0: + return "daily_rollup" if (digest_snapshot.get("stats", {}) or {}).get("built_monthly", 0) > 0: return "recalibration" return "incremental" @@ -626,6 +734,12 @@ class MemberContextService: parts.append(f"{label}：{meta.get('temperament_tendency')}") if meta.get("stable_traits"): parts.append(f"长期特征：{'、'.join(meta.get('stable_traits')[:3])}") + if meta.get("identity_traits"): + parts.append(f"身份线索：{'、'.join(meta.get('identity_traits')[:2])}") + if meta.get("skill_profile"): + parts.append(f"技能画像：{'、'.join(meta.get('skill_profile')[:3])}") + if meta.get("value_profile"): + parts.append(f"判断偏好：{'、'.join(meta.get('value_profile')[:2])}") if meta.get("habit_patterns"): parts.append(f"习惯模式：{'、'.join(meta.get('habit_patterns')[:3])}") if meta.get("recent_state"):