import asyncio import time from datetime import datetime, timedelta 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 from app.core.log import logger 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_day_ms_timestamp(self, date_str: str): """ 输入日期字符串 如 "2025-06-10" :return: start_ms(当日0点毫秒时间戳), end_ms(当日23:59:59毫秒时间戳) """ # 解析日期，生成当日 00:00:00 day_start = datetime.strptime(date_str, "%Y-%m-%d") # 当日最后一秒 23:59:59 day_end = day_start + timedelta(days=1) - timedelta(seconds=1) # timestamp() 返回秒，*1000 转毫秒，转整数 start_ms = int(day_start.timestamp() * 1000) end_ms = int(day_end.timestamp() * 1000) return start_ms, end_ms async def get_message_summary(self, date: str = None): stime = time.time() if date: start_ts, end_ts = await self.get_day_ms_timestamp(date) day_ArchiveMessages = await self.archive_messages_repository.get_message_statistics_by_date(start_date=start_ts, end_date=end_ts) logger.info(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): #打印计数 logger.info(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 != "[\"wmI1AkDQAAdVhy2ENtY6fecaw9-pi4uQ\"]": # 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: logger.info(f"已存在{date_user_a}和{date_user_b}在{day_date}对话，跳过") continue logger.info(f"查询{date_user_a}和{date_user_b}在{day_date}对话开始") 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) logger.info(f"查询{date_user_a}和{date_user_b}对话共：{len(coro_list)}") filter_gossip_message_text:str= "" summary_text:str = "" try: # 保留语义骨架，去除闲聊填充 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) logger.info(f"查询{date_user_a}和{date_user_b}对话进行保留语义骨架，去除闲聊填充") # 摘要 summary_res = await self.date_message_summary(coro_list, day_date) summary_text = str(summary_res.content) if hasattr(summary_res, 'content') else str(summary_res) logger.info(f"查询{date_user_a}和{date_user_b}对话进行摘要") except Exception as e: logger.error(f"{date_user_a}和{date_user_b}对话操作LLM失败: {e}") batch_embeddings=[] try: # 摘要向量 logger.info(f"查询{date_user_a}和{date_user_b}去除闲聊填充数据:{filter_gossip_message_text}") logger.info(f"查询{date_user_a}和{date_user_b}摘要:{summary_text}") batch_embeddings = await self.embedding_client.aembed_documents( [summary_text, filter_gossip_message_text]) logger.info(f"查询{date_user_a}和{date_user_b}对话向量数{len(batch_embeddings)}成功入库") except Exception as e: logger.warning(f"{date_user_a}和{date_user_b}对话embeddings失败，尝试重试: {e}") try: time.sleep(1) batch_embeddings = await self.embedding_client.aembed_documents( [summary_text, filter_gossip_message_text]) logger.info(f"查询{date_user_a}和{date_user_b}对话向量数{len(batch_embeddings)}重试成功入库") except Exception as retry_error: logger.error(f"{date_user_a}和{date_user_b}对话embeddings重试后仍然失败: {retry_error}") if not batch_embeddings or len(batch_embeddings) < 2: batch_embeddings=[[[0]],[[1]]] else: 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]] ) logger.info(f"查询{date_user_a}和{date_user_b}对话成功入库") except Exception as e: logger.error(f"{date_user_a}和{date_user_b}对话操作milvus失败: {e}") logger.info(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}) return result async def date_message_summary(self, coro_list: List[ArchiveMessages], day_date: str) -> str: prompt = """ # 聊天历史轻量化摘要任务 任务：生成极简轮次摘要，用于替换原始对话节省上下文窗口 # 历史对话 {message_str} ## 说明 其中INTERNAL表示销售，EXTERNAL表示客户 输出格式示例（无多余文字）： ** 核心诉求** 用户反馈订单 #12345 物流停滞超7天未更新。 询问是否支持跨店满减叠加优惠券。 ** 处理结果** 已联系快递网点核实，包裹因暴雨滞留，预计明日送达。（已安抚情绪并补偿10元无门槛券） 明确告知当前活动规则不支持叠加，引导领取店铺专属券。 ** 待办事项 ** [明日 10:00] 跟进物流签收状态并回访用户。 ** 关键标签**：#物流异常#优惠规则咨询#情绪安抚 """ lines = [f" {msg.from_role}：{msg.content}" for msg in coro_list] message_str = "\n".join(lines) prompt_template = PromptTemplate(template=prompt, input_variables=["message_str","day_date"]) chain = prompt_template | llm result = await chain.ainvoke({"message_str": message_str,"day_date":day_date}) return result async def build(day:str): logger.info(f"执行日期：{day}") 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(day) finally: await embedding_client.close() await db_assistant_mysql_client_manager.close() milvus_client.close() if __name__ == '__main__': # start = datetime(2026, 6, 23) end = datetime(2026, 6, 23) result = [] temp = start while temp <= end: # 按 年-月-日 无补零格式输出 date_format = f"{temp.year}-{temp.month}-{temp.day}" result.append(date_format) temp += timedelta(days=1) for date in result: asyncio.run(build(date))