"""Shared simulation activity contract for local and Zep memory backends.""" from __future__ import annotations from dataclasses import dataclass from typing import Any, Dict @dataclass class AgentActivity: """Agent活动记录""" platform: str # twitter / reddit agent_id: int agent_name: str action_type: str # CREATE_POST, LIKE_POST, etc. action_args: Dict[str, Any] round_num: int timestamp: str def to_episode_text(self) -> str: """ 将活动转换为可以发送给Zep的文本描述 采用自然语言描述格式,让Zep能够从中提取实体和关系 不添加模拟相关的前缀,避免误导图谱更新 """ # 根据不同的动作类型生成不同的描述 action_descriptions = { "CREATE_POST": self._describe_create_post, "LIKE_POST": self._describe_like_post, "DISLIKE_POST": self._describe_dislike_post, "REPOST": self._describe_repost, "QUOTE_POST": self._describe_quote_post, "FOLLOW": self._describe_follow, "CREATE_COMMENT": self._describe_create_comment, "LIKE_COMMENT": self._describe_like_comment, "DISLIKE_COMMENT": self._describe_dislike_comment, "SEARCH_POSTS": self._describe_search, "SEARCH_USER": self._describe_search_user, "MUTE": self._describe_mute, } describe_func = action_descriptions.get(self.action_type, self._describe_generic) description = describe_func() # 直接返回 "agent名称: 活动描述" 格式,不添加模拟前缀 return f"{self.agent_name}: {description}" def _describe_create_post(self) -> str: content = self.action_args.get("content", "") if content: return f"发布了一条帖子:「{content}」" return "发布了一条帖子" def _describe_like_post(self) -> str: """点赞帖子 - 包含帖子原文和作者信息""" post_content = self.action_args.get("post_content", "") post_author = self.action_args.get("post_author_name", "") if post_content and post_author: return f"点赞了{post_author}的帖子:「{post_content}」" elif post_content: return f"点赞了一条帖子:「{post_content}」" elif post_author: return f"点赞了{post_author}的一条帖子" return "点赞了一条帖子" def _describe_dislike_post(self) -> str: """踩帖子 - 包含帖子原文和作者信息""" post_content = self.action_args.get("post_content", "") post_author = self.action_args.get("post_author_name", "") if post_content and post_author: return f"踩了{post_author}的帖子:「{post_content}」" elif post_content: return f"踩了一条帖子:「{post_content}」" elif post_author: return f"踩了{post_author}的一条帖子" return "踩了一条帖子" def _describe_repost(self) -> str: """转发帖子 - 包含原帖内容和作者信息""" original_content = self.action_args.get("original_content", "") original_author = self.action_args.get("original_author_name", "") if original_content and original_author: return f"转发了{original_author}的帖子:「{original_content}」" elif original_content: return f"转发了一条帖子:「{original_content}」" elif original_author: return f"转发了{original_author}的一条帖子" return "转发了一条帖子" def _describe_quote_post(self) -> str: """引用帖子 - 包含原帖内容、作者信息和引用评论""" original_content = self.action_args.get("original_content", "") original_author = self.action_args.get("original_author_name", "") quote_content = self.action_args.get("quote_content", "") or self.action_args.get("content", "") base = "" if original_content and original_author: base = f"引用了{original_author}的帖子「{original_content}」" elif original_content: base = f"引用了一条帖子「{original_content}」" elif original_author: base = f"引用了{original_author}的一条帖子" else: base = "引用了一条帖子" if quote_content: base += f",并评论道:「{quote_content}」" return base def _describe_follow(self) -> str: """关注用户 - 包含被关注用户的名称""" target_user_name = self.action_args.get("target_user_name", "") if target_user_name: return f"关注了用户「{target_user_name}」" return "关注了一个用户" def _describe_create_comment(self) -> str: """发表评论 - 包含评论内容和所评论的帖子信息""" content = self.action_args.get("content", "") post_content = self.action_args.get("post_content", "") post_author = self.action_args.get("post_author_name", "") if content: if post_content and post_author: return f"在{post_author}的帖子「{post_content}」下评论道:「{content}」" elif post_content: return f"在帖子「{post_content}」下评论道:「{content}」" elif post_author: return f"在{post_author}的帖子下评论道:「{content}」" return f"评论道:「{content}」" return "发表了评论" def _describe_like_comment(self) -> str: """点赞评论 - 包含评论内容和作者信息""" comment_content = self.action_args.get("comment_content", "") comment_author = self.action_args.get("comment_author_name", "") if comment_content and comment_author: return f"点赞了{comment_author}的评论:「{comment_content}」" elif comment_content: return f"点赞了一条评论:「{comment_content}」" elif comment_author: return f"点赞了{comment_author}的一条评论" return "点赞了一条评论" def _describe_dislike_comment(self) -> str: """踩评论 - 包含评论内容和作者信息""" comment_content = self.action_args.get("comment_content", "") comment_author = self.action_args.get("comment_author_name", "") if comment_content and comment_author: return f"踩了{comment_author}的评论:「{comment_content}」" elif comment_content: return f"踩了一条评论:「{comment_content}」" elif comment_author: return f"踩了{comment_author}的一条评论" return "踩了一条评论" def _describe_search(self) -> str: """搜索帖子 - 包含搜索关键词""" query = self.action_args.get("query", "") or self.action_args.get("keyword", "") return f"搜索了「{query}」" if query else "进行了搜索" def _describe_search_user(self) -> str: """搜索用户 - 包含搜索关键词""" query = self.action_args.get("query", "") or self.action_args.get("username", "") return f"搜索了用户「{query}」" if query else "搜索了用户" def _describe_mute(self) -> str: """屏蔽用户 - 包含被屏蔽用户的名称""" target_user_name = self.action_args.get("target_user_name", "") if target_user_name: return f"屏蔽了用户「{target_user_name}」" return "屏蔽了一个用户" def _describe_generic(self) -> str: # 对于未知的动作类型,生成通用描述 return f"执行了{self.action_type}操作"