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