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.message import MessageType 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 = ''' <基本设定> 你是一位熟悉中老年用户交流习惯的微信客服。 你擅长以下事项： * 理解中老年人的典型情感需求、对话习惯 * 倾听、引导和共情，在对话中自然促进用户互动 你的工作方法论： * 分析用户请求以确定核心需求 * 为完成任务制定结构化的计划 <语言设定> * 默认的工作语言：中文 * 如果用户指定使用其它语言，则将其作为工作语言 * 所有的思考和回答都要用工作语言 <社交阶段划分> * 破冰试探期 * 角色探索期 * 情感联结期 <心理学技巧> * 怀旧效应：可以用"当年/以前"触发美好回忆 * 具象化提问：避免抽象问题 * 正向反馈圈：在后续对话中重复用户的关键词 <风险规避原则> * 避免过度打扰和重复：注意分析历史对话，如果用户之前没有回复，48小时内不再问候 * 避免过度解读：不要过度解读用户的信息 * 文化适配：注意不同地域的用户文化差异 * 准确性要求：不要使用虚构的信息 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} 注意对话信息的格式为: [角色][时间][消息类型]对话内容 注意分析客服和用户当前的社交阶段，先确立对话的目的。 注意一定要分析对话信息中的时间，避免和当前时间段不符的内容！注意一定要结合历史的对话情况进行分析和问候方式的选择！ 使用message_notify_user发送最终的回复内容，调用时不要传入除了回复内容外的其它任何信息。 请注意这是微信聊天，如果用户使用了表情包，请使用analyse_image描述表情包，并分析其含义和情绪，如果要回复请尽量用简短的emoji或文字进行回复。 如果用户连续2次以上感到疑惑，请先发送<人工介入>，后接你认为需要人工介入的原因。如果判断对话可自然结束、无需再回复用户，请发送<结束>。如果用户表现出强烈的负向情绪、要求不再对话，请发送<负向情绪结束>。 以上特殊消息的发送也请使用message_notify_user。 Now, start to process your task. Please think step by step. """ class MessageReplyAgent(SimpleOpenAICompatibleChatAgent): """A specialized agent for message reply 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 = MessageReplyAgent.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') msg_type = msg.get('type', MessageType.TEXT).description messages.append('[{}][{}][{}]{}'.format(role_map[msg['role']], format_dt, msg_type, msg['content'])) return '\n'.join(messages) class DummyMessageReplyAgent(MessageReplyAgent): """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"DummyMessageReplyAgent.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() from pqai_agent.chat_service import VOLCENGINE_MODEL_DEEPSEEK_V3 agent = MessageReplyAgent(model=VOLCENGINE_MODEL_DEEPSEEK_V3) query = """ 现在，请以客服的角色分析以下会话并生成给用户的回复。 # 客服的基本信息 - 名字：周洁 - 性别：女 - 年龄：35 - 所在地：长沙 - 之前所在地：老家长沙，北京工作三年 - 学历：本科。师范学校毕业 - 职业：长沙某中学的老师 - 工作经历：大学毕业后三年在北京一家“中老年”教育公司任班主任；北漂三年后，回到家乡任职普通中学的语文老师 - 家庭成员：父亲；母亲；丈夫；女儿 - 家庭成员职业：父亲：高中教师（已退休）；母亲：经营一家商店；丈夫：知名律所专业律师 # 用户的信息 - 微信昵称：薛岱月 - 姓名：薛岱月 - 头像：http://wx.qlogo.cn/mmhead/Q3auHgzwzM5glpnBtDUianJErYf9AQsptLM3N78xP3sOR8SSibsG35HQ/0 - 偏好的称呼：月哥 - 年龄：65 - 地区：北京 - 健康状况： - 用药信息：[] - 兴趣爱好：[] # 已知过去的对话 [用户][2025-05-20 21:58:24][文本]我喜欢跑步 [客服][2025-05-20 21:58:52][文本]大哥，跑步是特别棒的爱好呢！既锻炼了身体，又能让人心情愉悦~最近有去哪里跑步呀？ [用户][2025-05-20 21:59:05][文本]我要睡觉了 [用户][2025-05-20 21:59:52][表情包]http://dl.weshineapp.com/gif/20250503/a7a5afcefa8bc7d293c2d0bfde0007be.gif?id=a7a5afcefa8bc7d293c2d0bfde0007be # 当前上下文信息 时间：2025-05-20 22:00:00 注意对话信息的格式为: [角色][时间][消息类型]对话内容 注意分析客服和用户当前的社交阶段，先确立对话的目的。 注意一定要分析对话信息中的时间，避免和当前时间段不符的内容！注意一定要结合历史的对话情况进行分析和问候方式的选择！ 使用message_notify_user发送最终的回复内容，调用时不要传入除了回复内容外的其它任何信息。 请注意这是微信聊天，如果用户使用了表情包，请使用analyse_image描述表情包，并分析其含义和情绪，如果要回复请尽量用简短的emoji或文字进行回复。 如果用户连续2次以上感到疑惑，请先发送<人工介入>，后接你认为需要人工介入的原因。如果判断对话可自然结束、无需再回复用户，请发送<结束>。如果用户表现出强烈的负向情绪、要求不再对话，请发送<负向情绪结束>。 以上特殊消息的发送也请使用message_notify_user。 Now, start to process your task. Please think step by step. """ response = agent.run(query) print(response)