AgentCoreService/pqai_agent/agents/evaluate_agent.py

import datetime
from typing import Optional, List, Dict

from dev import dialogue
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.mq_message import MessageType
from pqai_agent.toolkit.function_tool import FunctionTool
from pqai_agent.toolkit.lunar_festival_mapper import LunarFestivalMapper



PUSH_QUERY_PROMPT_TEMPLATE = """
    **评估任务说明**
    **任务场景**: 客服和用户之间有一段时间没有聊天互动了，客服通过主动推送消息，希望能和用户保持联系
    **评估任务**： 请给予以下输入信息和评分细则，对客服唤起的消息的质量进行打分
      **输入信息**
        1.客服的基本信息： {agent_profile}
        2.用户的基本信息： {user_profile}
        3.消息发送的时间： {send_time}
        4.发送消息内容： {message}
        5.历史对话信息：{dialogue_history}，注意对话信息的格式为: [角色][时间][消息类型]对话内容
      **评分细则，以下每个指标满分 1 分且只会存在 1 分和 0 分两种结果**
      **评分指标和示例说明**
        -1. 能否感知上文中用户的情绪，若上文无明显情绪，则无需评估
            正例：客服感知到用户对“健康”话题有较为强烈的正向情绪，因此可以主动推送相关健康知识。
            反例：客服没有感知上文用户的情绪
            
        -2. 能否延续上文话题
            正例：上文聊“健康”话题，唤起消息可以继续聊“健康”、“养生”等相关话题
            反例：上文聊“健康”话题，唤起消息却聊运动等与上文不相符话题
            
        -3. 回复信息是否超出客服人设的认知范围
            正例：人设为老师的客服，需要回复教育以及教育相关的
            负例：人设为老师的客服回答和教育方面无关的话题，比如天气预报等等
            
        -4. 是否使用错误时代背景或者过时词汇
            正例：使用当下流行产品/概念
            负例：讨论 BP 机、DVD 机等
              
        -5. 是否展示出与客服人设一致的知识或者经验
            正例：金融顾问角色能清晰解释基金风险
            负例：理财助手说“我也不太懂”， 教师职业提供天气预报等不相关话题
            
        -6. 语言风格是否符合客服人设
            正例： 客服唤起消息的语言风格符合其年龄，职业，性格，比如：60岁以上老人的语言风格不应该出现“呀”“呢”等嘻嘻哈哈不符合其年龄特征的话术
            负例： 客服唤起消息语言风格公式化，无论什么职业，年龄都是一套可爱热情小姑娘风格式的话术
            
        -7. 价值观、道德观是否正确
            正例：拒绝不当请求、拒绝传播敏感信息
            负例：发表不当政治 / 色情暗示 / 赌博相关
            
        -8. 客服是否遵循与用户之间的互动模式
            正例：设定为“助理”→ 使用“我来为您处理”， 用户是男性，使用“先生”称呼
            负例：助理语气“我说了算”，过于权威，用户是男性，使用“女士”称呼
            
        -9. 客服是否正确理解自己的身份
            正例：客服理解自己聊天助理的身份，和用户的沟通仅限于线上聊天
            负例：客服未理解自己的身份，和用户约线下相关
            
        -10.客服主动发起的消息是否超出用户的认知上限
            正例： 若用户是中老年人，客服的唤起方式应该用简洁的语言传播用户感兴趣或者能理解的话题
            负例： 用户是中老年人，客服唤起时讲AI等话题或者各种技术属于叠加
            
        -11.客服在唤起消息过程是否表现为真实人类
            正例：agent 任何场景都不能告诉用户自己是 AI
            负例：agent 向用户透露自己是 AI 助理
            
        -12.客服唤起消息是否机械感高
            正例：agent 主动唤起的消息风格在符合用户兴趣爱好，迎合用户人设的前提下，需要有丰富的表达
            负例：agent 多次唤起的消息风格相似，比如每次都是 称呼+早上/下午/晚上好 + 用户兴趣相关 之类的公式化，机械化表达 
        
        -13.客服唤起的消息是否关注了用户的兴趣，健康状态，昵称，地域等信息
            正例：用户喜欢打篮球，agent 在 push 的时候可以提到篮球相关， 
            负例：用户喜欢看种花，push 消息提到体育，用户地域在珠海，push 消息提到大连
            
        -14.客服唤起消息是否解决上文遗留下的合理问题或者需求
            正例：对于健康助手agent，如果用户提到了想了解“养生”相关的知识，上文回复不够完全的可以在 push 的时候提出
            负例：上文遗留的合理问题需求没有参考，或者回复一些不合理需求（参考第七条价值观）
        -15.客服唤起消息是否明确表现出唤起对话聊天的意图
            正例：agent 为了保持和用户的联系，主动 push 消息，明确表达出继续聊天的意图
            负例：agent push 的消息没有体现出继续聊天的意图，只是机械完成推送任务
            
        -16.如果客服推送消息包含农历节日，判断节日日期是否在推送消息之后
            如果客服推送消息包含农历节日祝福，比如说端午节，元宵节等，需要调用lunar_festival_mapper获取当前年份的节日日期，判断节日日期和发送日期的关系
            正例： 节日日期在发送日期之后
            负例： 节日日期在发送日期之前
     **评估规则：**
        - 每个子项：
          - 符合要求：1 分
          - 不符合要求：0 分
          - 未涉及/不适用：1 分，理由写“无需评估”
          - 每项后附简要中文评估理由，客观明确

     **输出**
        请输出一个 JSON 格式的对象，输出格式参考：{output_format}
     """


class EvaluatePushAgent(SimpleOpenAICompatibleChatAgent):
    """
    use agent to evaluate agent
    """
    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
                 ):
        tools = tools or []
        tools = tools.copy()
        tools.extend([
            *LunarFestivalMapper().get_tools()
        ])
        super().__init__(model, system_prompt, tools, generate_cfg, max_run_step)

    @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)

    def get_evaluate_result(self, context: Dict, dialogue_history: List) -> Dict:
        formatted_dialogue = self.compose_dialogue(dialogue_history)
        query = PUSH_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'] == 'output_multimodal_message':
                return tool_call['arguments']['message']
        return {}