diff --git a/src/chat/message_receive/chat_stream.py b/src/chat/message_receive/chat_stream.py index 4f6fbb3d7..50f7f7f93 100644 --- a/src/chat/message_receive/chat_stream.py +++ b/src/chat/message_receive/chat_stream.py @@ -289,11 +289,11 @@ class ChatStream: """获取用户关系分""" # 使用统一的评分API try: - from src.plugin_system.apis.scoring_api import scoring_api + from src.plugin_system.apis import person_api if self.user_info and hasattr(self.user_info, "user_id"): user_id = str(self.user_info.user_id) - relationship_score = await scoring_api.get_user_relationship_score(user_id) + relationship_score = await person_api.get_user_relationship_score(user_id) logger.debug(f"ChatStream {self.stream_id}: 用户关系分 = {relationship_score:.3f}") return relationship_score diff --git a/src/chat/replyer/default_generator.py b/src/chat/replyer/default_generator.py index 2218dca49..d1a62f314 100644 --- a/src/chat/replyer/default_generator.py +++ b/src/chat/replyer/default_generator.py @@ -1980,12 +1980,12 @@ class DefaultReplyer: logger.error(f"获取关系信息失败: {e}") # 降级到基本信息 try: - from src.plugin_system.apis.scoring_api import scoring_api + from src.plugin_system.apis import person_api user_info = await person_info_manager.get_values(person_id, ["user_id", "platform"]) user_id = user_info.get("user_id", "unknown") - relationship_data = await scoring_api.get_user_relationship_data(user_id) + relationship_data = await person_api.get_user_relationship_data(user_id) if relationship_data: relationship_text = relationship_data.get("relationship_text", "") relationship_score = relationship_data.get("relationship_score", 0.3) diff --git a/src/individuality/individuality.py b/src/individuality/individuality.py index e512d97c8..27c2c8d28 100644 --- a/src/individuality/individuality.py +++ b/src/individuality/individuality.py @@ -82,9 +82,9 @@ class Individuality: full_personality = f"{personality_result},{identity_result}" # 使用统一的评分API初始化智能兴趣系统 - from src.plugin_system.apis.scoring_api import scoring_api + from src.plugin_system.apis import person_api - await scoring_api.initialize_smart_interests( + await person_api.initialize_smart_interests( personality_description=full_personality, personality_id=self.bot_person_id ) diff --git a/src/plugin_system/apis/person_api.py b/src/plugin_system/apis/person_api.py index 03b2bb544..c1589b359 100644 --- a/src/plugin_system/apis/person_api.py +++ b/src/plugin_system/apis/person_api.py @@ -4,34 +4,29 @@ 使用方式: from src.plugin_system.apis import person_api person_id = person_api.get_person_id("qq", 123456) - value = await person_api.get_person_value(person_id, "nickname") + info = await person_api.get_person_info(person_id) """ +import asyncio from typing import Any from src.common.logger import get_logger from src.person_info.person_info import PersonInfoManager, get_person_info_manager +from src.plugin_system.services.interest_service import interest_service +from src.plugin_system.services.relationship_service import relationship_service logger = get_logger("person_api") # ============================================================================= -# 个人信息API函数 +# 辅助函数 # ============================================================================= def get_person_id(platform: str, user_id: int | str) -> str: - """根据平台和用户ID获取person_id + """根据平台和用户ID获取person_id (同步) - Args: - platform: 平台名称,如 "qq", "telegram" 等 - user_id: 用户ID - - Returns: - str: 唯一的person_id(MD5哈希值) - - 示例: - person_id = person_api.get_person_id("qq", 123456) + 这是一个核心的辅助函数,用于生成统一的用户标识。 """ try: return Person(platform=platform, user_id=str(user_id)).person_id @@ -40,93 +35,23 @@ def get_person_id(platform: str, user_id: int | str) -> str: return "" -async def get_person_value(person_id: str, field_name: str, default: Any = None) -> Any: - """根据person_id和字段名获取某个值 - - Args: - person_id: 用户的唯一标识ID - field_name: 要获取的字段名,如 "nickname", "impression" 等 - default: 当字段不存在或获取失败时返回的默认值 - - Returns: - Any: 字段值或默认值 - - 示例: - nickname = await person_api.get_person_value(person_id, "nickname", "未知用户") - impression = await person_api.get_person_value(person_id, "impression") - """ - try: - person = Person(person_id=person_id) - value = getattr(person, field_name) - return value if value is not None else default - except Exception as e: - logger.error(f"[PersonAPI] 获取用户信息失败: person_id={person_id}, field={field_name}, error={e}") - return default - - -async def get_person_values(person_id: str, field_names: list, default_dict: dict | None = None) -> dict: - """批量获取用户信息字段值 - - Args: - person_id: 用户的唯一标识ID - field_names: 要获取的字段名列表 - default_dict: 默认值字典,键为字段名,值为默认值 - - Returns: - dict: 字段名到值的映射字典 - - 示例: - values = await person_api.get_person_values( - person_id, - ["nickname", "impression", "know_times"], - {"nickname": "未知用户", "know_times": 0} - ) - """ +async def get_person_id_by_name(person_name: str) -> str: + """根据用户名获取person_id""" try: person_info_manager = get_person_info_manager() - values = await person_info_manager.get_values(person_id, field_names) - - # 如果获取成功,返回结果 - if values: - return values - - # 如果获取失败,构建默认值字典 - result = {} - if default_dict: - for field in field_names: - result[field] = default_dict.get(field, None) - else: - for field in field_names: - result[field] = None - - return result - + return await person_info_manager.get_person_id_by_person_name(person_name) except Exception as e: - logger.error(f"[PersonAPI] 批量获取用户信息失败: person_id={person_id}, fields={field_names}, error={e}") - # 返回默认值字典 - result = {} - if default_dict: - for field in field_names: - result[field] = default_dict.get(field, None) - else: - for field in field_names: - result[field] = None - return result + logger.error(f"[PersonAPI] 根据用户名获取person_id失败: person_name={person_name}, error={e}") + return "" + + +# ============================================================================= +# 核心信息查询API +# ============================================================================= async def is_person_known(platform: str, user_id: int) -> bool: - """判断是否认识某个用户 - - Args: - platform: 平台名称 - user_id: 用户ID - - Returns: - bool: 是否认识该用户 - - 示例: - known = await person_api.is_person_known("qq", 123456) - """ + """判断是否认识某个用户""" try: person_info_manager = get_person_info_manager() return await person_info_manager.is_person_known(platform, user_id) @@ -135,21 +60,217 @@ async def is_person_known(platform: str, user_id: int) -> bool: return False -async def get_person_id_by_name(person_name: str) -> str: - """根据用户名获取person_id +async def get_person_info(person_id: str) -> dict[str, Any]: + """获取用户的核心基础信息 - Args: - person_name: 用户名 - - Returns: - str: person_id,如果未找到返回空字符串 - - 示例: - person_id = person_api.get_person_id_by_name("张三") + 返回一个包含用户基础信息的字典,例如 person_name, nickname, know_times, attitude 等。 """ + if not person_id: + return {} try: person_info_manager = get_person_info_manager() - return await person_info_manager.get_person_id_by_person_name(person_name) + fields = ["person_name", "nickname", "know_times", "know_since", "last_know", "attitude"] + values = await person_info_manager.get_values(person_id, fields) + return values except Exception as e: - logger.error(f"[PersonAPI] 根据用户名获取person_id失败: person_name={person_name}, error={e}") - return "" + logger.error(f"[PersonAPI] 获取用户信息失败: person_id={person_id}, error={e}") + return {} + + +async def get_person_impression(person_id: str, short: bool = False) -> str: + """获取对用户的印象 + + Args: + person_id: 用户的唯一标识ID + short: 是否获取简短版印象,默认为False + + Returns: + 一段描述性的文本。 + """ + if not person_id: + return "用户ID为空,无法获取印象。" + try: + person_info_manager = get_person_info_manager() + field = "short_impression" if short else "impression" + impression = await person_info_manager.get_value(person_id, field) + return impression or "还没有形成对该用户的印象。" + except Exception as e: + logger.error(f"[PersonAPI] 获取用户印象失败: person_id={person_id}, error={e}") + return "获取用户印象时发生错误。" + + +async def get_person_points(person_id: str, limit: int = 5) -> list[tuple]: + """获取关于用户的'记忆点' + + Args: + person_id: 用户的唯一标识ID + limit: 返回的记忆点数量上限,默认为5 + + Returns: + 一个列表,每个元素是一个包含记忆点内容、权重和时间的元组。 + """ + if not person_id: + return [] + try: + person_info_manager = get_person_info_manager() + points = await person_info_manager.get_value(person_id, "points") + if not points: + return [] + + # 按权重和时间排序,返回最重要的几个点 + sorted_points = sorted(points, key=lambda x: (x[1], x[2]), reverse=True) + return sorted_points[:limit] + except Exception as e: + logger.error(f"[PersonAPI] 获取用户记忆点失败: person_id={person_id}, error={e}") + return [] + + +# ============================================================================= +# 关系查询API +# ============================================================================= + + +async def get_user_relationship_score(user_id: str) -> float: + """ + 获取用户关系分 + + Args: + user_id: 用户ID + + Returns: + 关系分 (0.0 - 1.0) + """ + return await relationship_service.get_user_relationship_score(user_id) + + +async def get_user_relationship_data(user_id: str) -> dict: + """ + 获取用户完整关系数据 + + Args: + user_id: 用户ID + + Returns: + 包含关系分、关系文本等的字典 + """ + return await relationship_service.get_user_relationship_data(user_id) + + +async def update_user_relationship(user_id: str, relationship_score: float, relationship_text: str | None = None, user_name: str | None = None): + """ + 更新用户关系数据 + + Args: + user_id: 用户ID + relationship_score: 关系分 (0.0 - 1.0) + relationship_text: 关系描述文本 + user_name: 用户名称 + """ + await relationship_service.update_user_relationship(user_id, relationship_score, relationship_text, user_name) + + +# ============================================================================= +# 兴趣系统API +# ============================================================================= + + +async def initialize_smart_interests(personality_description: str, personality_id: str = "default"): + """ + 初始化智能兴趣系统 + + Args: + personality_description: 机器人性格描述 + personality_id: 性格ID + """ + await interest_service.initialize_smart_interests(personality_description, personality_id) + + +async def calculate_interest_match(content: str, keywords: list[str] | None = None): + """ + 计算内容与兴趣的匹配度 + + Args: + content: 消息内容 + keywords: 关键词列表 + + Returns: + 匹配结果 + """ + return await interest_service.calculate_interest_match(content, keywords) + + +# ============================================================================= +# 系统状态与缓存API +# ============================================================================= + + +def get_system_stats() -> dict[str, Any]: + """ + 获取系统统计信息 + + Returns: + 包含各子系统统计的字典 + """ + return { + "relationship_service": relationship_service.get_cache_stats(), + "interest_service": interest_service.get_interest_stats() + } + + +def clear_caches(user_id: str | None = None): + """ + 清理缓存 + + Args: + user_id: 特定用户ID,如果为None则清理所有缓存 + """ + relationship_service.clear_cache(user_id) + logger.info(f"清理缓存: {user_id if user_id else '全部'}") + + +# ============================================================================= +# 报告API +# ============================================================================= + + +async def get_full_relationship_report(person_id: str) -> str: + """生成一份关于你和用户的完整'关系报告' + + 综合基础信息、印象、记忆点和关系分,提供一个全方位的关系概览。 + """ + if not person_id: + return "无法生成报告,因为用户ID为空。" + + try: + person_info_manager = get_person_info_manager() + user_id = await person_info_manager.get_value(person_id, "user_id") + + if not user_id: + return "无法生成报告,因为找不到对应的用户信息。" + + # 异步获取所有需要的信息 + info, impression, points, rel_data = await asyncio.gather( + get_person_info(person_id), + get_person_impression(person_id), + get_person_points(person_id, limit=3), + relationship_service.get_user_relationship_data(str(user_id)), + ) + + # 构建报告 + report = f"--- 与 {info.get('person_name', '未知用户')} 的关系报告 ---\n" + report += f"昵称: {info.get('nickname', '未知')}\n" + report += f"关系分数: {rel_data.get('relationship_score', 0.0):.2f}/1.0\n" + report += f"关系描述: {rel_data.get('relationship_text', '暂无')}\n" + report += f"我对ta的印象: {impression}\n" + + if points: + report += "最近的重要记忆点:\n" + for point in points: + report += f" - {point[0]} (重要性: {point[1]})\n" + + report += "----------------------------------------\n" + return report + + except Exception as e: + logger.error(f"[PersonAPI] 生成关系报告失败: person_id={person_id}, error={e}") + return "生成关系报告时发生错误。" diff --git a/src/plugin_system/apis/scoring_api.py b/src/plugin_system/apis/scoring_api.py deleted file mode 100644 index 95e854f43..000000000 --- a/src/plugin_system/apis/scoring_api.py +++ /dev/null @@ -1,113 +0,0 @@ -""" -统一评分系统API -提供系统级的关系分和兴趣管理服务,供所有插件和主项目组件使用 -""" - -from typing import Any - -from src.common.logger import get_logger -from src.plugin_system.services.interest_service import interest_service -from src.plugin_system.services.relationship_service import relationship_service - -logger = get_logger("scoring_api") - - -class ScoringAPI: - """ - 统一评分系统API - 系统级服务 - - 提供关系分和兴趣管理的统一接口,替代原有的插件依赖方式。 - 所有插件和主项目组件都应该通过此API访问评分功能。 - """ - - @staticmethod - async def get_user_relationship_score(user_id: str) -> float: - """ - 获取用户关系分 - - Args: - user_id: 用户ID - - Returns: - 关系分 (0.0 - 1.0) - """ - return await relationship_service.get_user_relationship_score(user_id) - - @staticmethod - async def get_user_relationship_data(user_id: str) -> dict: - """ - 获取用户完整关系数据 - - Args: - user_id: 用户ID - - Returns: - 包含关系分、关系文本等的字典 - """ - return await relationship_service.get_user_relationship_data(user_id) - - @staticmethod - async def update_user_relationship(user_id: str, relationship_score: float, relationship_text: str | None = None, user_name: str | None = None): - """ - 更新用户关系数据 - - Args: - user_id: 用户ID - relationship_score: 关系分 (0.0 - 1.0) - relationship_text: 关系描述文本 - user_name: 用户名称 - """ - await relationship_service.update_user_relationship(user_id, relationship_score, relationship_text, user_name) - - @staticmethod - async def initialize_smart_interests(personality_description: str, personality_id: str = "default"): - """ - 初始化智能兴趣系统 - - Args: - personality_description: 机器人性格描述 - personality_id: 性格ID - """ - await interest_service.initialize_smart_interests(personality_description, personality_id) - - @staticmethod - async def calculate_interest_match(content: str, keywords: list[str] | None = None): - """ - 计算内容与兴趣的匹配度 - - Args: - content: 消息内容 - keywords: 关键词列表 - - Returns: - 匹配结果 - """ - return await interest_service.calculate_interest_match(content, keywords) - - @staticmethod - def get_system_stats() -> dict[str, Any]: - """ - 获取系统统计信息 - - Returns: - 包含各子系统统计的字典 - """ - return { - "relationship_service": relationship_service.get_cache_stats(), - "interest_service": interest_service.get_interest_stats() - } - - @staticmethod - def clear_caches(user_id: str | None = None): - """ - 清理缓存 - - Args: - user_id: 特定用户ID,如果为None则清理所有缓存 - """ - relationship_service.clear_cache(user_id) - logger.info(f"清理缓存: {user_id if user_id else '全部'}") - - -# 创建全局API实例 - 系统级服务 -scoring_api = ScoringAPI() diff --git a/src/plugins/built_in/affinity_flow_chatter/affinity_interest_calculator.py b/src/plugins/built_in/affinity_flow_chatter/affinity_interest_calculator.py index e50a93f56..059a1f762 100644 --- a/src/plugins/built_in/affinity_flow_chatter/affinity_interest_calculator.py +++ b/src/plugins/built_in/affinity_flow_chatter/affinity_interest_calculator.py @@ -211,9 +211,9 @@ class AffinityInterestCalculator(BaseInterestCalculator): # 如果内存中没有,尝试从统一的评分API获取 try: - from src.plugin_system.apis.scoring_api import scoring_api + from src.plugin_system.apis import person_api - relationship_data = await scoring_api.get_user_relationship_data(user_id) + relationship_data = await person_api.get_user_relationship_data(user_id) if relationship_data: relationship_score = relationship_data.get("relationship_score", global_config.affinity_flow.base_relationship_score) # 同时更新内存缓存