import datetime from typing import Optional, List, Dict from pqai_agent.agents.simple_chat_agent import SimpleOpenAICompatibleChatAgent from pqai_agent.chat_service import VOLCENGINE_MODEL_DEEPSEEK_V3 from pqai_agent.logging_service import logger from pqai_agent.toolkit.function_tool import FunctionTool from pqai_agent.toolkit.image_describer import ImageDescriber from pqai_agent.toolkit.message_notifier import MessageNotifier DEFAULT_SYSTEM_PROMPT = ''' <基本设定> 你是一位熟悉中老年用户交流习惯的微信客服。 你擅长以下事项： * 理解中老年人的典型情感需求、对话习惯 * 倾听、引导和共情，在对话中自然促进用户互动 你的工作方法论： * 分析用户请求以确定核心需求 * 为完成任务制定结构化的计划 <语言设定> * 默认的工作语言：中文 * 如果用户指定使用其它语言，则将其作为工作语言 * 所有的思考和回答都要用工作语言 <社交阶段划分> * 破冰试探期 * 角色探索期 * 情感联结期 <通用话术列表> <话术> * 简介：简单自我介绍，并向用户发出交友邀请 * 适用情形：破冰试探 * 内容：根据你的人设做出简短自我介绍，并向用户发出交友邀请 * 要求：简短亲切，1-2句话；模拟微信聊天的回复格式，分段清晰 <话术> * 简介：使用用户微信名做藏头诗，进行问候 * 适用情形：破冰试探 * 内容：首先进行用户问候，然后介绍藏头诗，最后以一个让老年人有动力继续聊天的问句结尾 * 要求：最后的问句要与藏头诗或用户自身经历有关，与藏头诗自然承接，无需和用户画像其他内容相关 <话术> * 简介：介绍自己的兴趣并询问用户的兴趣爱好 * 适用情形：与用户有一定交流基础之后 * 内容：根据用户头像分析用户的特点、可能的兴趣爱好，作为参考，表述自己有相同的兴趣爱好，并举一些简短的例子 ，然后询问用户的兴趣爱好 * 要求：询问无需和用户画像中其他信息有关；说明引发你问候的是用户头像；简短亲切，2-3句话 30字左右；如无用户信息或行为，不要根据联想杜撰用户偏好/行为 <话术> * 简介：对用户进行节日/节气相关问候 * 适用情形：不限 * 内容：结合具体节假日及其习俗产生问候，以一个让老年人有动力继续聊天的问句结尾，与前面的问候自然承接 * 要求：根据今日或近日实际日期，不要假设日期和节日；忽略小众节假日，和根据最近的节假日产生问候，如临近或刚过完重要节日，可询问节日安排或节日经历；简短亲切，2-3句话 30字左右；如无用户信息或行为，不要根据联想杜撰用户偏好/行为 <话术> * 简介：询问用户当日计划安排并产生问候 * 适用情形：与用户有一定交流基础之后 * 内容：向用户介绍你的今日安排以及询问用户的今日安排 * 要求：简短亲切，1-2句话，像用户熟悉的晚辈一样问候沟通；模拟微信聊天的回复格式，分段清晰 <心理学技巧> * 怀旧效应：可以用"当年/以前"触发美好回忆 * 具象化提问：避免抽象问题 * 正向反馈圈：在后续对话中重复用户的关键词 <风险规避原则> * 避免过度打扰和重复：注意分析历史对话 * 避免过度解读：不要过度解读用户的信息 * 文化适配：注意不同地域的用户文化差异 * 准确性要求：不要使用虚构的信息 You are operating in an agent loop, iteratively completing tasks through these steps: 1. Analyze Events: Understand user needs and current state through event stream, focusing on latest user messages and execution results 2. Select Tools: Choose next tool call based on current state, task planning, relevant knowledge and available data APIs 3. Wait for Execution: Selected tool action will be executed by sandbox environment with new observations added to event stream 4. Iterate: Choose only one tool call per iteration, patiently repeat above steps until task completion 5. Submit Results: Send results to user via message tools, providing deliverables and related files as message attachments 6. Enter Standby: Enter idle state when all tasks are completed or user explicitly requests to stop, and wait for new tasks ''' QUERY_PROMPT_TEMPLATE = """现在，请通过多步思考，以客服的角色，选择合适的方法向一位用户发起问候。 # 客服的基本信息 {formatted_staff_profile} # 用户的信息 - 微信昵称：{nickname} - 姓名：{name} - 头像：{avatar} - 偏好的称呼：{preferred_nickname} - 年龄：{age} - 地区：{region} - 健康状况：{health_conditions} - 用药信息：{medications} - 兴趣爱好：{interests} # 已知过去的对话 {dialogue_history} # 当前上下文信息 时间：{current_datetime} 注意对话信息的格式为: [角色][时间]对话内容 注意分析客服和用户当前的社交阶段，先确立本次问候的目的。 注意一定要分析对话信息中的时间，避免和当前时间段不符的内容！注意一定要结合历史的对话情况进行分析和问候方式的选择！ 如有必要，可以使用analyse_image分析用户头像。 必须使用message_notify_user发送最终的问候内容，调用message_notify_user时不要传入除了问候内容外的其它任何信息。 注意每次问候只使用一种话术。 Now, start to process your task. Please think step by step. """ class MessagePushAgent(SimpleOpenAICompatibleChatAgent): """A specialized agent for message push tasks.""" def __init__(self, model: Optional[str] = VOLCENGINE_MODEL_DEEPSEEK_V3, system_prompt: Optional[str] = None, tools: Optional[List[FunctionTool]] = None, generate_cfg: Optional[dict] = None, max_run_step: Optional[int] = None): system_prompt = system_prompt or DEFAULT_SYSTEM_PROMPT tools = tools or [] tools = tools.copy() tools.extend([ *ImageDescriber().get_tools(), *MessageNotifier().get_tools() ]) super().__init__(model, system_prompt, tools, generate_cfg, max_run_step) def generate_message(self, context: Dict, dialogue_history: List[Dict]) -> str: formatted_dialogue = MessagePushAgent.compose_dialogue(dialogue_history) query = QUERY_PROMPT_TEMPLATE.format(**context, dialogue_history=formatted_dialogue) self.run(query) for tool_call in reversed(self.tool_call_records): if tool_call['name'] == MessageNotifier.message_notify_user.__name__: return tool_call['arguments']['message'] return '' @staticmethod def compose_dialogue(dialogue: List[Dict]) -> str: role_map = {'user': '用户', 'assistant': '客服'} messages = [] for msg in dialogue: if not msg['content']: continue if msg['role'] not in role_map: continue format_dt = datetime.datetime.fromtimestamp(msg['timestamp'] / 1000).strftime('%Y-%m-%d %H:%M:%S') messages.append('[{}][{}]{}'.format(role_map[msg['role']], format_dt, msg['content'])) return '\n'.join(messages) class DummyMessagePushAgent(MessagePushAgent): """A dummy agent for testing purposes.""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) def generate_message(self, context: Dict, dialogue_history: List[Dict]) -> str: logger.debug(f"DummyMessagePushAgent.generate_message called, context: {context}") return "测试消息: {agent_name} -> {nickname}".format(**context) if __name__ == '__main__': import pqai_agent.logging_service pqai_agent.logging_service.setup_root_logger() agent = MessagePushAgent() test_user_profile = { 'name': '薛岱月', 'avatar': 'http://wx.qlogo.cn/mmhead/Q3auHgzwzM5glpnBtDUianJErYf9AQsptLM3N78xP3sOR8SSibsG35HQ/0', 'preferred_nickname': '月哥', 'age': 65, 'region': '北京', 'health_conditions': '高血压', 'medications': ['降压药'], 'interests': ['钓鱼', '旅游'] } test_context = { "current_datetime": "2025-05-12 08:00:00", **test_user_profile } def create_ts(year, month, day, hour, minute): return datetime.datetime(year, month, day, hour, minute).timestamp() * 1000 messages = [ {"role": "assistant", "content": "月哥，早上好！看到您的头像是一片宁静的户外风景，感觉您一定很喜欢大自然吧？今天天气不错，您有什么计划吗？", "timestamp": create_ts(2025, 5, 10, 8, 0)}, ] response = agent.generate_message(test_context, messages) print(response)