77 lines
4.7 KiB
Python
77 lines
4.7 KiB
Python
from __future__ import annotations
|
||
|
||
from typing import Dict
|
||
|
||
|
||
class GroupProfileResolver:
|
||
def __init__(self, config: Dict):
|
||
self.config = config or {}
|
||
self.default_profile = self.config.get("default", {}) or {}
|
||
self.profiles = self.config.get("profiles", []) or []
|
||
|
||
def resolve(self, room_id: str, group_name: str = "", group_memory_profile: Dict | None = None) -> Dict:
|
||
group_name_lower = str(group_name or "").lower()
|
||
for profile in self.profiles:
|
||
room_ids = set(profile.get("room_ids", []) or [])
|
||
keywords = [str(item).lower() for item in (profile.get("group_name_keywords", []) or [])]
|
||
if room_id and room_id in room_ids:
|
||
return self._normalize(profile, room_id, group_name, group_memory_profile or {})
|
||
if group_name_lower and any(keyword and keyword in group_name_lower for keyword in keywords):
|
||
return self._normalize(profile, room_id, group_name, group_memory_profile or {})
|
||
return self._normalize(self.default_profile, room_id, group_name, group_memory_profile or {})
|
||
|
||
@staticmethod
|
||
def _normalize(profile: Dict, room_id: str, group_name: str, group_memory_profile: Dict) -> Dict:
|
||
focus = list(profile.get("knowledge_focus", []))
|
||
configured_domain = str(profile.get("knowledge_domain", "general") or "general")
|
||
inferred_domain = str(group_memory_profile.get("inferred_domain", "general") or "general")
|
||
inferred_style = group_memory_profile.get("style_profile", {}) or {}
|
||
structured_summary = group_memory_profile.get("structured_summary", {}) or {}
|
||
effective_domain = configured_domain
|
||
if configured_domain in {"", "general", "casual"} and inferred_domain not in {"", "general"}:
|
||
effective_domain = inferred_domain
|
||
# 群摘要不再只是“一段文案”：
|
||
# 1. 这里把群长期聚合后的稳定主题一起并到 focus 里；
|
||
# 2. 这样群画像后续传给 prompt 时，LLM 能拿到更干净的字段，而不是自己再拆 markdown；
|
||
# 3. 同时只保留去重后的短项，避免群摘要字段把配置 focus 全冲掉。
|
||
inferred_focus = list(group_memory_profile.get("focus_topics", []))
|
||
inferred_focus.extend(structured_summary.get("stable_topics", []) or [])
|
||
merged_focus = []
|
||
for item in focus + inferred_focus:
|
||
if item and item not in merged_focus:
|
||
merged_focus.append(item)
|
||
interaction_tone = str(profile.get("interaction_tone", "自然群友感") or "自然群友感")
|
||
humor_style = str(profile.get("humor_style", "轻微") or "轻微")
|
||
sharpness_style = str(profile.get("sharpness_style", "轻微嘴硬，不刻薄") or "轻微嘴硬，不刻薄")
|
||
expressiveness_style = str(profile.get("expressiveness_style", "克制") or "克制")
|
||
address_style = str(profile.get("address_style", "低频称呼，默认直接接话") or "低频称呼，默认直接接话")
|
||
if configured_domain in {"", "general", "casual"}:
|
||
interaction_tone = inferred_style.get("interaction_tone", interaction_tone)
|
||
humor_style = inferred_style.get("humor_style", humor_style)
|
||
sharpness_style = inferred_style.get("sharpness_style", sharpness_style)
|
||
expressiveness_style = inferred_style.get("expressiveness_style", expressiveness_style)
|
||
return {
|
||
"room_id": room_id,
|
||
"group_name": group_name,
|
||
"mode": profile.get("mode", "social"),
|
||
"persona_id": str(profile.get("persona_id", "xiaoniu") or "xiaoniu"),
|
||
"persona_overlay": profile.get("persona_overlay", ""),
|
||
"interaction_tone": interaction_tone,
|
||
"humor_style": humor_style,
|
||
"sharpness_style": sharpness_style,
|
||
"expressiveness_style": expressiveness_style,
|
||
"address_style": address_style,
|
||
"knowledge_domain": effective_domain,
|
||
"configured_domain": configured_domain,
|
||
"knowledge_focus": merged_focus,
|
||
"reply_style": profile.get("reply_style", "自然短句"),
|
||
"topic_boundaries": profile.get("topic_boundaries", []),
|
||
"group_memory_domain": inferred_domain,
|
||
"group_memory_summary": group_memory_profile.get("summary_text", ""),
|
||
"group_memory_sample_count": group_memory_profile.get("message_sample_count", 0),
|
||
"group_memory_summary_days": group_memory_profile.get("summary_source_count", 0),
|
||
"group_memory_structured": structured_summary,
|
||
"group_memory_timeline": group_memory_profile.get("summary_timeline", []) or [],
|
||
"group_memory_style": inferred_style,
|
||
}
|