72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Dict, List
|
|
|
|
|
|
class ContextBuilder:
|
|
def build(
|
|
self,
|
|
*,
|
|
room_id: str,
|
|
sender: str,
|
|
sender_name: str,
|
|
content: str,
|
|
recent_messages: List[Dict],
|
|
member_context: Dict,
|
|
trigger: Dict,
|
|
flow_state: str,
|
|
reply_mode: str,
|
|
vector_memories: List[Dict],
|
|
) -> Dict:
|
|
recent_lines = []
|
|
for item in recent_messages[-8:]:
|
|
msg_sender = item.get("sender_name") or item.get("sender") or "未知成员"
|
|
msg_content = item.get("content") or item.get("message") or ""
|
|
if msg_content:
|
|
recent_lines.append(f"{msg_sender}: {msg_content}")
|
|
return {
|
|
"group_profile": {"room_id": room_id},
|
|
"speaker_profile": {
|
|
"wxid": sender,
|
|
"display_name": sender_name,
|
|
"member_context": member_context or {},
|
|
},
|
|
"recent_messages": recent_lines,
|
|
"recent_summary": "",
|
|
"trigger_type": trigger.get("trigger_type", "none"),
|
|
"reply_mode": reply_mode,
|
|
"flow_state": flow_state,
|
|
"memory_prompt": self._build_member_memory_prompt(member_context),
|
|
"vector_memory_prompt": self._build_vector_memory_prompt(vector_memories),
|
|
"current_message": f"{sender_name}: {content}",
|
|
}
|
|
|
|
@staticmethod
|
|
def _build_member_memory_prompt(member_context: Dict) -> str:
|
|
if not member_context:
|
|
return "暂无稳定成员画像。"
|
|
meta = member_context.get("meta", {}) or {}
|
|
topics = member_context.get("topics_of_interest", []) or []
|
|
recent_focus = member_context.get("recent_focus", []) or []
|
|
lines = [
|
|
f"成员摘要：{member_context.get('summary_text', '')}".strip(),
|
|
f"互动风格：{member_context.get('interaction_style', '')}".strip(),
|
|
f"回复偏好：{member_context.get('response_style_hint', '')}".strip(),
|
|
f"长期主题：{', '.join(topics[:5])}" if topics else "",
|
|
f"近期关注：{', '.join(recent_focus[:4])}" if recent_focus else "",
|
|
f"群内角色：{meta.get('group_role', '')}".strip(),
|
|
]
|
|
return "\n".join([line for line in lines if line])
|
|
|
|
@staticmethod
|
|
def _build_vector_memory_prompt(vector_memories: List[Dict]) -> str:
|
|
if not vector_memories:
|
|
return ""
|
|
lines = []
|
|
for item in vector_memories[:2]:
|
|
summary = item.get("content_summary") or item.get("summary_text") or item.get("text") or ""
|
|
memory_type = item.get("memory_type", "memory")
|
|
if summary:
|
|
lines.append(f"[{memory_type}] {summary}")
|
|
return "\n".join(lines)
|