sales-assistant-py-new/app/summary/service.py

import asyncio
import time
from typing import List

from langchain_core.prompts import PromptTemplate

from app.client.embedding_client_manager import EmbeddingClientManager, embedding_client
from app.client.milvus_client_manager import milvus_client
from app.client.mysql_client_manager import db_assistant_mysql_client_manager
from app.dto.summary_dto import SummaryDto
from app.llm import llm
from app.models.mysql import ArchiveMessages
from app.repository.archive_messages_repository import ArchiveMessagesRepository
from app.repository.milvus.summary_repository import SummaryRepository


class SummaryService:

    def __init__(self, archive_messages_repository: ArchiveMessagesRepository,
                 embedding_client: EmbeddingClientManager,
                 summary_repository: SummaryRepository):
        self.archive_messages_repository = archive_messages_repository
        self.embedding_client = embedding_client
        self.summary_repository = summary_repository

    async def get_message_summary(self, date: str = None):
        stime = time.time()
        if date:
            day_ArchiveMessages = await self.archive_messages_repository.get_message_statistics_by_date(start_date=date,
                                                                                                        end_date=date)
            print(f"{date}共产生了{len(day_ArchiveMessages)}条会话")
        else:
            day_ArchiveMessages = await self.archive_messages_repository.get_message_statistics_by_date()

        # 改成map
        date_user_message_a_set = set()
        for idx, day_message in enumerate(day_ArchiveMessages, 1):            #打印计数
            print(f"处理第{idx}会话")
            day_date = str(day_message.created_at)
            user_a = day_message.from_user
            # 找出发送人相关的接收人信息
            date_user_message_a = await self.archive_messages_repository.get_message_statistics_by_date_user(day_date,
                                                                                                             user_a)

            coro_list = []
            # 找出发送人与接收人 相关的所有会话信息
            for date_a in date_user_message_a:

                date_user_b = date_a.to_user
                # if date_user_b != "[\"wmI1AkDQAA3h0jxrRpeaHZhMKeHExA4w\"]":
                #     continue

                date_user_a = date_a.from_user
                date_user_b_clean = date_user_b.replace("[\"", "").replace("\"]", "")
                date_user_a_append = "[\"" + date_user_a + "\"]"
                # 判断只要存在就跳过
                if date_user_a + date_user_b in date_user_message_a_set or date_user_b_clean + date_user_a_append in date_user_message_a_set:
                    continue
                print(f"查询{date_user_a}和{date_user_b}对话开始")
                list1 = await self.archive_messages_repository.get_message_statistics_by_date_user(day_date,
                                                                                                   date_user_a,
                                                                                                   date_user_b)

                list2 = await self.archive_messages_repository.get_message_statistics_by_date_user(day_date,
                                                                                                   date_user_b_clean,
                                                                                                   date_user_a_append)
                # 查询过的聊天账号，后续不再查询
                date_user_message_a_set.add(date_user_a + date_user_b)
                date_user_message_a_set.add(date_user_b_clean + date_user_a_append)
                coro_list = list1 + list2
                coro_list.sort(key=lambda x: x.created_at, reverse=False)
                print(f"查询{date_user_a}和{date_user_b}对话共：{len(coro_list)}")
                # 保留语义骨架，去除闲聊填充
                filter_gossip_message_res = await self.filter_gossip_message(coro_list)
                filter_gossip_message_text = str(filter_gossip_message_res.content) if hasattr(
                    filter_gossip_message_res, 'content') else str(filter_gossip_message_res)
                print(f"查询{date_user_a}和{date_user_b}对话进行保留语义骨架，去除闲聊填充")
                # 摘要
                summary_res = await self.date_message_summary(coro_list)
                summary_text = str(summary_res.content) if hasattr(summary_res, 'content') else str(summary_res)
                print(f"查询{date_user_a}和{date_user_b}对话进行摘要")
                # 摘要向量
                batch_embeddings = await self.embedding_client.aembed_documents(
                    [summary_text, filter_gossip_message_text])

                #
                try:
                    self.summary_repository.insert(
                        msg_time=[day_date],
                        msg_type=[0],
                        from_user=[date_a.from_user],
                        to_user=[date_user_b_clean],
                        room_id=[date_a.roomid or ""],
                        message_context=[filter_gossip_message_text],
                        summary=[summary_text],
                        message_dense_vector=[batch_embeddings[1]],
                        summary_dense_vector=[batch_embeddings[0]]
                    )
                    print(f"查询{date_user_a}和{date_user_b}对话成功入 库")
                except Exception as e:
                    print(e)
            print(f"运行结束：{(time.time()-stime)}")

    async def filter_gossip_message(self, coro_list: List[ArchiveMessages]) -> str:
        prompt = """
            保留语义骨架，去除闲聊填充
            # 历史对话
            {message_str}
            其中INTERNAL表示销售，EXTERNAL表示用户
            # 例如
            原文：
            用户：在吗？
            销售：在的，您好，请问有什么可以帮您？
            用户：我想问一下，那个，我的订单怎么还没发货啊？
            销售：好的，请问您的订单号是多少呢？
            用户：订单号是 12345，我昨天就下单了。
            销售：我查一下，稍等... 您的订单正在打包中，预计今天发出。

            结构化提取后：
            [用户] 咨询订单发货状态，订单号 12345，昨日下单。
            [销售] 查询后告知正在打包，预计今日发出。
        """
        lines = [f"{msg.created_at} {msg.from_role}：{msg.content}" for msg in coro_list]
        message_str = "\n".join(lines)

        prompt_template = PromptTemplate(template=prompt, input_variables=["message_str"])
        chain = prompt_template | llm
        result = await chain.ainvoke({"message_str": message_str})
        print(result)
        return result

    async def date_message_summary(self, coro_list: List[ArchiveMessages]) -> str:
        prompt = """
            # 聊天历史轻量化摘要任务
            任务：生成极简轮次摘要，用于替换原始对话节省上下文窗口
            # 历史对话
            {message_str}
            ## 说明
            其中INTERNAL表示销售，EXTERNAL表示客户
            输出格式示例（无多余文字）：
            【日期】 202X年X月X日
            ** 核心诉求**
            用户反馈订单 #12345 物流停滞超7天未更新。
            询问是否支持跨店满减叠加优惠券。
            ** 处理结果**
            已联系快递网点核实，包裹因暴雨滞留，预计明日送达。（已安抚情绪并补偿10元无门槛券）
            明确告知当前活动规则不支持叠加，引导领取店铺专属券。
            ** 待办事项 **
            [明日 10:00] 跟进物流签收状态并回访用户。
            ** 关键标签**：#物流异常#优惠规则咨询#情绪安抚          
        """
        lines = [f"{msg.created_at} {msg.from_role}：{msg.content}" for msg in coro_list]
        message_str = "\n".join(lines)

        prompt_template = PromptTemplate(template=prompt, input_variables=["message_str"])
        chain = prompt_template | llm
        result = await chain.ainvoke({"message_str": message_str})
        print(result)
        return result


async def build():
    db_assistant_mysql_client_manager.init()
    embedding_client.init()
    milvus_client.init()
    try:
        async with db_assistant_mysql_client_manager.session_factory() as db_assistant:
            archive_messages_repository = ArchiveMessagesRepository(db_assistant)
            summary_repository = SummaryRepository()
            service = SummaryService(archive_messages_repository, embedding_client, summary_repository)
            await service.get_message_summary("2026-6-11")
    finally:
        await embedding_client.close()
        await db_assistant_mysql_client_manager.close()
        milvus_client.close()


if __name__ == '__main__':
    asyncio.run(build())