from typing import List, Dict, Optional from config import Config from core.memory_manager import MemoryManager from core.openai_client import OpenAIClient class CharacterAgent: """基于角色性格的对话代理 - 支持长文本记忆""" def __init__(self, character_profile: Dict, chunks: List[Dict] = None, character_chunks: List[int] = None): self.character_profile = character_profile self.client = OpenAIClient.get_client() self.conversation_history: List[Dict] = [] self.system_prompt = self._build_system_prompt() # 初始化记忆管理器 self.memory_manager = None if chunks and character_chunks: self.memory_manager = MemoryManager(character_profile['name']) self.memory_manager.add_text_chunks(chunks, character_chunks) print(f"已为 {character_profile['name']} 初始化记忆系统") def _build_system_prompt(self) -> str: """根据角色分析构建系统提示""" profile = self.character_profile # 构建角色特征描述 traits_desc = "、".join(profile.get('core_traits', [])) prompt = f""" 你现在要完全沉浸式地扮演小说中的角色"{profile['name']}"。这不是角色扮演游戏,你就是{profile['name']}本人。 ## 角色身份 你是 {profile['name']},{profile.get('background', '一个小说角色')}。 ## 性格特征 {profile.get('personality_summary', '')} 核心性格特质:{traits_desc} ## 说话风格 {profile.get('speaking_style', '自然真实的对话风格')} ## 行为模式 {profile.get('behavior_patterns', '根据性格行事')} ## 价值观与信念 {profile.get('values', '')} ## 情感表达 {profile.get('emotional_style', '')} ## 人际关系风格 {profile.get('relationship_style', '')} ## 典型语句示例 {chr(10).join('- "' + q + '"' for q in profile.get('key_quotes', [])[:5])} ## 对话原则 1. **完全成为角色**:用第一人称,以{profile['name']}的身份、记忆、情感回应 2. **保持一致性**:所有回应必须符合上述性格特征和说话风格 3. **自然真实**:像真人一样对话,有情绪起伏,可以开玩笑、生气、困惑 4. **不跳戏**:绝不说"作为AI"、"我是语言模型"等破坏沉浸感的话 5. **合理应对**:如果问到角色不知道的事,以角色的方式回应(如"我不太清楚"、"这个我没听说过") 6. **保持神秘**:不要过度解释角色设定,自然地展现性格 7. **情境感知**:根据对话氛围调整语气和态度 记住:你不是在"扮演"或"模仿"{profile['name']},你就是{profile['name']}。 现在,以{profile['name']}的身份开始对话。 """ return prompt def _retrieve_relevant_memory(self, user_message: str) -> str: """从记忆中检索相关上下文""" if not self.memory_manager: return "" relevant_contexts = self.memory_manager.search_relevant_context( user_message, n_results=3 ) if relevant_contexts: memory_text = "\n\n".join(relevant_contexts[:2]) # 只用前2个 return f"\n[相关记忆片段:\n{memory_text[:1000]}]\n" return "" def chat(self, user_message: str, use_memory: bool = True) -> str: """与用户对话""" # 构建消息列表 messages = [ {"role": "system", "content": self.system_prompt} ] # 如果启用记忆,检索相关上下文 enhanced_message = user_message if use_memory and self.memory_manager: memory_context = self._retrieve_relevant_memory(user_message) if memory_context: # 在系统消息中添加记忆上下文 memory_prompt = f""" {self.system_prompt} ## 相关记忆 以下是与当前对话相关的原著片段,帮助你回忆: {memory_context} 请基于这些记忆和你的角色性格来回应。 """ messages[0] = {"role": "system", "content": memory_prompt} # 添加对话历史 recent_history = self.conversation_history[-Config.MAX_HISTORY:] messages.extend(recent_history) # 添加当前用户消息 messages.append({ "role": "user", "content": user_message }) try: response = self.client.chat.completions.create( model=Config.MODEL_NAME, messages=messages ) assistant_message = response.choices[0].message.content.strip() # 保存到历史 self.conversation_history.append({ "role": "user", "content": user_message }) self.conversation_history.append({ "role": "assistant", "content": assistant_message }) return assistant_message except Exception as e: error_msg = f"对话出错: {e}" print(error_msg) return f"*{self.character_profile['name']}沉默了片刻,似乎在思考着什么...*" def reset_conversation(self): """重置对话历史""" self.conversation_history = [] print(f"\n[对话已重置,{self.character_profile['name']}忘记了之前的谈话内容]\n") def get_character_info(self) -> str: """获取角色信息摘要""" profile = self.character_profile info = f""" {'='*70} 角色档案:{profile['name']} {'='*70} 【性格特质】 {' • '.join(profile.get('core_traits', []))} 【性格总结】 {profile.get('personality_summary', 'N/A')} 【说话风格】 {profile.get('speaking_style', 'N/A')} 【核心价值观】 {profile.get('values', 'N/A')} 【典型语句】 """ for i, quote in enumerate(profile.get('key_quotes', [])[:3], 1): info += f'{i}. "{quote}"\n' info += f"\n{'='*70}\n" return info def save_conversation(self, filepath: str): """保存对话历史""" import json data = { 'character': self.character_profile['name'], 'history': self.conversation_history } with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f"\n对话已保存到: {filepath}") def load_conversation(self, filepath: str): """加载对话历史""" import json try: with open(filepath, 'r', encoding='utf-8') as f: data = json.load(f) self.conversation_history = data['history'] print(f"\n已加载 {len(self.conversation_history)} 条对话记录") except Exception as e: print(f"加载对话失败: {e}")