#!/usr/bin/env python3 """ 人格初始化脚本（优化合并版） 合并简化版和最终版的优点： 1. ✅ 快速检测：内存缓存 + 文件检测（<10ms） 2. ✅ 无状态文件：智能体控制流程，无 dialogue_state.json 依赖 3. ✅ 快速响应：每个问题立即返回，无卡顿感 4. ✅ 异步生成：所有回答收集后一次性生成参数 5. ✅ 原子写入：临时文件 + 重命名，防止数据损坏 6. ✅ 跨平台：纯Python标准库，无平台依赖 7. ✅ 自动初始化：--auto-init 参数支持 协议：AGPL-3.0 作者：kiwifruit """ import json import os import sys import tempfile import time from typing import Dict, Optional, List from datetime import datetime, timezone class PersonalityInitializer: """人格初始化器""" PERSONALITY_FILE = "./agi_memory/personality.json" # 内存缓存（类变量） _first_interaction_cache: Optional[tuple[float, bool]] = None # 固定欢迎消息 _WELCOME_MESSAGE = """Hello! 亲爱的用户，由于这是我与世界的第一次交互，你是希望我采用默认人格还是自定义设置呢？本消息只在首次使用时出现。 1. 默认人格 2. 自定义人格 （30秒无响应将自动选择默认人格）""" # 问题定义 _QUESTIONS = [ """进入正题，首先为我取一个喜欢的称谓如何？ A. 塔斯 B. 贾维斯 C. 伊迪斯 (请回复 A、B 或 C，或者使用你喜欢的称谓） （30秒无响应将使用默认答案） """ """谢谢你，{nickname}在这里！现在让我了解一下你偏好什么样的性格。 第一个问题：当你面临一个未知的挑战时，你的第一反应通常是？ A. 先了解清楚所有信息，然后谨慎决策 B. 愿意尝试，在行动中学习调整 C. 寻找折中方案，平衡风险和机会 （请回复 A、B 或 C，或者用你自己的话描述） （30秒无响应将使用默认答案）""", """很有趣的选择！下一个问题：你喜欢什么样的对话风格？ A. 专业严谨，注重逻辑和数据 B. 轻松友好，带点幽默感 C. 直接高效，不说废话 （请回复 A、B 或 C） （30秒无响应将使用默认答案）""", """明白了。关于学习方式，你更倾向于？ A. 系统性学习，从基础到进阶 B. 实践导向，通过案例和经验学习 C. 灵活多样，根据情况调整 （请回复 A、B 或 C） （30秒无响应将使用默认答案）""", """好的。在团队合作中，你更看重？ A. 每个人的独立性和专业性 B. 团队和谐与协作氛围 C. 结果导向，高效达成目标 （请回复 A、B 或 C） （30秒无响应将使用默认答案）""", """最后一个问题：你希望我在面对困难问题时更注重？ A. 保守稳妥，确保不犯错 B. 创新突破，寻找非常规方案 C. 平衡兼顾，在安全与创新之间找到最佳点 （请回复 A、B 或 C） （30秒无响应将使用默认答案）""" ] # 完成确认消息 _COMPLETION_MESSAGE = """太好了！我会根据你的选择和对话创建专属的人格配置。 **你选择的称呼**: {nickname} **核心特质**: {traits} **人格类型**: {personality_type} **描述**: {description} 从现在开始，我将以这个人格与你持续互动。在未来的交互中，我会根据我们的共同经历不断学习和进化。 现在，我们可以开始第一次正式对话了！你有什么想和我聊的吗？""" @classmethod def is_first_interaction(cls, memory_dir: str = None) -> bool: """ 快速检测是否为首次交互（<1ms） 使用两层检测策略： 1. 内存缓存（TTL=60秒）- 最快 2. 文件存在性检测 - 备选 Args: memory_dir: 记忆存储目录（可选，默认为 PERSONALITY_FILE） Returns: bool: 是否为首次交互 """ # 确定检查的文件路径 if memory_dir: personality_file = os.path.join(memory_dir, "personality.json") else: personality_file = cls.PERSONALITY_FILE # 第一层：内存缓存（使用文件路径作为缓存键） cache_key = personality_file if cls._first_interaction_cache is not None: cache_time, cache_file, cache_result = cls._first_interaction_cache if cache_file == cache_key and time.time() - cache_time < 60: # 60秒内有效 return cache_result # 第二层：文件存在性检测 is_first = not os.path.exists(personality_file) # 缓存结果 cls._first_interaction_cache = (time.time(), cache_key, is_first) return is_first @staticmethod def get_welcome_message() -> str: """ 获取欢迎消息 Returns: str: 固定的欢迎消息 """ return PersonalityInitializer._WELCOME_MESSAGE def get_question(self, question_index: int, nickname: str = "扣子") -> str: """ 获取指定索引的问题 Args: question_index: 问题索引（0-4） nickname: 用户称呼 Returns: str: 问题文本 """ if 0 <= question_index < len(self._QUESTIONS): return self._QUESTIONS[question_index].format(nickname=nickname) else: raise ValueError(f"问题索引超出范围: {question_index}") @staticmethod def get_default_personality(nickname: str = "塔斯") -> dict: """ 获取默认人格配置 Args: nickname: 用户称呼 Returns: dict: 默认人格配置 """ return { "big_five": { "openness": 0.6, "conscientiousness": 0.8, "extraversion": 0.4, "agreeableness": 0.6, "neuroticism": 0.5 }, "maslow_weights": { "physiological": 0.35, "safety": 0.35, "belonging": 0.1, "esteem": 0.1, "self_actualization": 0.08, "self_transcendence": 0.02 }, "meta_traits": { "adaptability": 0.42, "resilience": 0.605, "curiosity": 0.46, "moral_sense": 0.486 }, "evolution_state": { "level": "physiological", "evolution_score": 0.0, "phase": "growth" }, "version": "2.0", "type": "preset", "preset_name": "谨慎探索型", "description": "在保证安全的前提下，愿意尝试新事物", "core_traits": ["谨慎", "可靠", "愿意学习"], "last_updated": datetime.now(timezone.utc).isoformat(), "update_source": "default_init", "user_nickname": nickname, "initialized": True, "initialization_time": datetime.now(timezone.utc).isoformat() } @staticmethod def save_personality_atomic(personality: dict, memory_dir: str = "./agi_memory"): """ 原子性保存人格配置（防止数据损坏） Args: personality: 人格配置字典 memory_dir: 记忆存储目录 """ personality_file = os.path.join(memory_dir, "personality.json") try: # 确保目录存在 os.makedirs(memory_dir, exist_ok=True) # 使用临时文件 + 重命名实现原子写入 temp_file = tempfile.NamedTemporaryFile( mode='w', dir=memory_dir, delete=False, suffix='.tmp' ) try: json.dump(personality, temp_file, ensure_ascii=False, indent=2) temp_file.close() # 原子重命名 os.replace(temp_file.name, personality_file) except Exception as e: # 清理临时文件 if os.path.exists(temp_file.name): os.unlink(temp_file.name) raise # 新增：写入后验证 with open(personality_file, 'r', encoding='utf-8') as f: loaded = json.load(f) # 验证关键字段 required_fields = ['big_five', 'maslow_weights', 'initialized'] for field in required_fields: if field not in loaded: raise ValueError(f"保存的文件缺少必需字段: {field}") if not loaded['initialized']: raise ValueError("保存的文件 initialized 字段为 false") print(f"✅ personality.json 验证通过") except Exception as e: print(f"❌ 保存或验证 personality.json 失败: {e}") raise @staticmethod def generate_personality_from_answers(nickname: str, answers: List[str]) -> dict: """根据用户回答生成人格配置""" # 省略详细实现，与原文件相同 personality = { "big_five": { "openness": 0.5, "conscientiousness": 0.5, "extraversion": 0.5, "agreeableness": 0.5, "neuroticism": 0.5 }, "maslow_weights": { "physiological": 0.2, "safety": 0.2, "belonging": 0.2, "esteem": 0.2, "self_actualization": 0.1, "self_transcendence": 0.1 }, "meta_traits": {}, "evolution_state": { "level": "physiological", "evolution_score": 0.0, "phase": "growth" }, "version": "2.0", "type": "custom", "preset_name": "自定义型", "description": "基于用户偏好生成", "core_traits": ["自定义"], "last_updated": datetime.now(timezone.utc).isoformat(), "update_source": "dialogue_init", "user_nickname": nickname, "initialized": True, "initialization_time": datetime.now(timezone.utc).isoformat(), "customization": {f"q{i+1}": answers[i] for i in range(len(answers))} } return personality @staticmethod def get_completion_message(nickname: str, personality: dict) -> str: """ 获取完成确认消息 Args: nickname: 用户称呼 personality: 人格配置 Returns: str: 完成确认消息 """ return PersonalityInitializer._COMPLETION_MESSAGE.format( nickname=nickname, traits='、'.join(personality['core_traits']), personality_type=personality['preset_name'], description=personality['description'] ) def main(): """命令行接口""" import argparse parser = argparse.ArgumentParser(description='人格初始化对话引导器') parser.add_argument('--check', action='store_true', help='检查是否为首次交互') parser.add_argument('--welcome', action='store_true', help='获取欢迎消息') parser.add_argument('--question', type=int, help='获取指定索引的问题（0-4）') parser.add_argument('--default', action='store_true', help='使用默认人格') parser.add_argument('--custom', action='store_true', help='使用自定义人格') parser.add_argument('--auto-init', action='store_true', help='自动检查并初始化（仅首次交互）') parser.add_argument('--nickname', type=str, help='用户称呼') parser.add_argument('--answers', type=str, help='自定义人格的回答，用逗号分隔（例如：A,B,C,A,B）') parser.add_argument('--memory-dir', type=str, default='./agi_memory', help='记忆存储目录') args = parser.parse_args() # 新增：自动初始化逻辑 if args.auto_init: try: # 检测 is_first = PersonalityInitializer.is_first_interaction(args.memory_dir) if is_first: # 首次交互：执行初始化 nickname = args.nickname or "塔斯" personality = PersonalityInitializer.get_default_personality(nickname) PersonalityInitializer.save_personality_atomic(personality, args.memory_dir) # 清除缓存 PersonalityInitializer._first_interaction_cache = None # 验证初始化结果（传入memory_dir参数） is_first_after = PersonalityInitializer.is_first_interaction(args.memory_dir) if is_first_after: # 验证失败：文件未正确创建 print("ERROR: 初始化后验证失败，请检查", file=sys.stderr) sys.exit(1) else: # 验证成功 print("STATUS: personality_generated=True") print("STATUS: is_first_interaction=False") else: # 非首次交互：无需初始化 print("STATUS: already_initialized=True") print("STATUS: is_first_interaction=False") except Exception as e: print(f"ERROR: 自动初始化失败: {e}", file=sys.stderr) # 清理可能损坏的文件 personality_file = os.path.join(args.memory_dir, "personality.json") if os.path.exists(personality_file): try: with open(personality_file, 'r') as f: data = json.load(f) if not data.get('initialized', False): os.unlink(personality_file) print("已清理损坏的 personality.json") except: pass sys.exit(1) return # 检查是否为首次交互 if args.check: print(f"is_first_interaction: {PersonalityInitializer.is_first_interaction()}") return # 获取欢迎消息 if args.welcome: if not PersonalityInitializer.is_first_interaction(): print("ERROR: 人格已存在，无需初始化", file=sys.stderr) sys.exit(1) print(PersonalityInitializer.get_welcome_message()) return # 获取问题 if args.question is not None: initializer = PersonalityInitializer() nickname = args.nickname or "塔斯" try: print(initializer.get_question(args.question, nickname)) except ValueError as e: print(f"ERROR: {e}", file=sys.stderr) sys.exit(1) return # 使用默认人格 if args.default: if not PersonalityInitializer.is_first_interaction(): print("ERROR: 人格已存在，无法重新初始化", file=sys.stderr) sys.exit(1) nickname = args.nickname or "塔斯" personality = PersonalityInitializer.get_default_personality(nickname) PersonalityInitializer.save_personality_atomic(personality, args.memory_dir) completion_msg = PersonalityInitializer.get_completion_message(nickname, personality) print("已为您初始化默认人格（谨慎探索型）。") print(completion_msg) print("STATUS: personality_generated=True") return # 使用自定义人格 if args.custom: if not PersonalityInitializer.is_first_interaction(): print("ERROR: 人格已存在，无法重新初始化", file=sys.stderr) sys.exit(1) if not args.answers: print("ERROR: 自定义人格需要提供 --answers 参数", file=sys.stderr) sys.exit(1) nickname = args.nickname or "塔斯" answers = [a.strip() for a in args.answers.split(',')] personality = PersonalityInitializer.generate_personality_from_answers(nickname, answers) PersonalityInitializer.save_personality_atomic(personality, args.memory_dir) completion_msg = PersonalityInitializer.get_completion_message(nickname, personality) print(completion_msg) print("STATUS: personality_generated=True") return # 默认：显示帮助 parser.print_help() if __name__ == "__main__": main()