265 lines
13 KiB
Python
265 lines
13 KiB
Python
import asyncio
|
||
import json
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from typing import List
|
||
|
||
from langchain_core.output_parsers import StrOutputParser
|
||
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
|
||
if not (msg.msgtype == "voice" and "[语音]" not in msg.content)
|
||
]
|
||
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 real_time_summary(self, from_user: str, to_user: str) -> str:
|
||
message_list = await self.archive_messages_repository.get_conversation_messages(from_user, to_user)
|
||
if not message_list:
|
||
return ""
|
||
prompt = """
|
||
# 聊天历史轻量化摘要任务
|
||
任务：生成极简轮次摘要，用于替换原始对话节省上下文窗口
|
||
# 历史对话
|
||
{message_str}
|
||
## 说明
|
||
其中INTERNAL表示老师，EXTERNAL表示学生
|
||
输出格式示例（无多余文字）：
|
||
学生反馈订单 #12345 物流停滞超7天未更新。
|
||
询问是否支持跨店满减叠加优惠券。
|
||
|
||
"""
|
||
lines = [f" {msg.created_at}:{msg.from_role}：{msg.content}" for msg in message_list]
|
||
message_str = "\n".join(lines)
|
||
|
||
prompt_template = PromptTemplate(template=prompt, input_variables=["message_str"])
|
||
chain = prompt_template | llm | StrOutputParser()
|
||
result = await chain.ainvoke({"message_str": message_str})
|
||
logger.info(f"{from_user},{to_user}实时总结结果：{result}")
|
||
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.real_time_summary("BuShouShiJinBuGaiMing","wmI1AkDQAADMN3I0RScMP_g4R8jGoVfQ")
|
||
finally:
|
||
await embedding_client.close()
|
||
await db_assistant_mysql_client_manager.close()
|
||
milvus_client.close()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
#
|
||
start = datetime(2026, 6, 24)
|
||
end = datetime(2026, 6, 24)
|
||
|
||
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))
|
||
|
||
|