项目初始化，清理全部大文件与LFS历史

.gitignore

@@ -0,0 +1,4 @@
+# 忽略Maven编译目录
+.trae
+.venv
+docker/

.idea/.gitignore

@@ -0,0 +1,8 @@
+# Default ignored files
+/shelf/
+/workspace.xml
+# Editor-based HTTP Client requests
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

.idea/inspectionProfiles/profiles_settings.xml

@@ -0,0 +1,6 @@
+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>

.idea/misc.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Black">
+    <option name="sdkName" value="Python 3.12 (sales-assistant-py)" />
+  </component>
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.12 (sales-assistant-py)" project-jdk-type="Python SDK" />
+</project>

.idea/modules.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/sales-assistant-py.iml" filepath="$PROJECT_DIR$/.idea/sales-assistant-py.iml" />
+    </modules>
+  </component>
+</project>

.idea/sales-assistant-py.iml

@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$">
+      <excludeFolder url="file://$MODULE_DIR$/.venv" />
+    </content>
+    <orderEntry type="jdk" jdkName="Python 3.12 (sales-assistant-py)" jdkType="Python SDK" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

.idea/vcs.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>

app/aa.md

@@ -0,0 +1 @@
+【日期】 2026年6月11日  \n【沟通对象】 客户Janice  \n** 核心诉求 (User Intent)**  \n咨询线上角色建模相关课程；销售进一步了解其年级与专业背景。  \n** 处理结果 (Resolution)**  \n初步确认客户对角色建模有兴趣，正在收集基本信息。  \n** 待办事项 (Action Items)**  \n（无）  \n** 关键标签**：#角色建模#初步咨询

app/client/embedding_client_manager.py

@@ -0,0 +1,70 @@
+import asyncio
+import time
+
+import aiohttp
+import requests
+
+from app.conf.app_config import EmbeddingConfig, app_config
+
+
+class EmbeddingClientManager:
+    def __init__(self, config: EmbeddingConfig):
+        self.config = config
+        self.client = None
+
+    def _get_url(self):
+        return f"http://{self.config.host}:{self.config.port}"
+
+
+    def init(self, wait_for_ready: bool = True):
+        self.client = aiohttp.ClientSession()
+
+    async def close(self):
+        if self.client:
+            await self.client.close()
+            self.client = None
+
+    async def aembed_documents(self, texts: list) -> list:
+        if not self.client:
+            self.init(wait_for_ready=False)
+
+        url = f"{self._get_url()}/embed"
+        payload = {
+            "inputs": texts,
+            "parameters": {"truncate": True}
+        }
+
+        async with self.client.post(url, json=payload) as response:
+            result = await response.json()
+            if isinstance(result, list):
+                return result
+            return result.get("embeddings", [])
+
+    async def aembed_query(self, text: str) -> list:
+        embeddings = await self.aembed_documents([text])
+        return embeddings[0] if embeddings else []
+
+
+embedding_client = EmbeddingClientManager(app_config.embedding)
+
+if __name__ == "__main__":
+    print("Testing EmbeddingClientManager...")
+    
+    async def test():
+        try:
+            embedding_client.init(wait_for_ready=True)
+            print("Initialization successful")
+            
+            text = "What is Deep Learning?"
+            print(f"Test text: {text}")
+            
+            result = await embedding_client.aembed_query(text)
+            print(f"Embedding successful")
+            print(f"Vector length: {len(result)}")
+            print(f"First 10 values: {result[:10]}")
+            
+            await embedding_client.close()
+        except Exception as e:
+            print(f"Test failed: {str(e)}")
+    
+    asyncio.run(test())

app/client/milvus_client_manager.py

@@ -0,0 +1,59 @@
+from typing import Optional
+
+from pymilvus import MilvusClient as PyMilvusClient
+
+from app.conf.app_config import MilvusConfig, app_config
+
+
+class MilvusClientWrapper:
+    """Milvus 客户端单例包装器"""
+    _instance = None
+    _initialized = False
+
+    def __new__(cls, milvus_config: MilvusConfig = None):
+        if cls._instance is None:
+            cls._instance = super(MilvusClientWrapper, cls).__new__(cls)
+        return cls._instance
+
+    def __init__(self, milvus_config: MilvusConfig = None):
+        if self._initialized:
+            return
+
+        if milvus_config is None:
+            milvus_config = app_config.milvus
+
+        self.milvus_config = milvus_config
+        self.client: Optional[PyMilvusClient] = None
+        self._initialized = True
+
+    def init(self):
+        if self.client is not None:
+            return
+
+        uri = f"http://{self.milvus_config.host}:{self.milvus_config.port}"
+        self.client = PyMilvusClient(
+            uri=uri,
+            user=self.milvus_config.user,
+            password=self.milvus_config.password
+        )
+
+    def close(self):
+        if self.client is not None:
+            self.client.close()
+            self.client = None
+
+    def has_collection(self, collection_name: str) -> bool:
+        self._ensure_connection()
+        return self.client.has_collection(collection_name)
+
+    def drop_collection(self, collection_name: str):
+        self._ensure_connection()
+        if self.client.has_collection(collection_name):
+            self.client.drop_collection(collection_name)
+
+    def _ensure_connection(self):
+        if self.client is None:
+            self.init()
+
+
+milvus_client = MilvusClientWrapper()

app/client/mysql_client_manager.py

@@ -0,0 +1,45 @@
+import asyncio
+from typing import Optional
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine, async_sessionmaker
+
+from app.conf.app_config import DBConfig, app_config
+
+
+class MysqlClientManager:
+    def __init__(self, db_config: DBConfig):
+        self.db_config = db_config
+        self.engine: Optional[AsyncEngine] = None
+        self.session_factory = None
+
+    def _get_url(self):
+        return f"mysql+asyncmy://{self.db_config.user}:{self.db_config.password}@{self.db_config.host}:{self.db_config.port}/{self.db_config.database}?charset=utf8mb4"
+
+    def init(self):
+        self.engine = create_async_engine(url=self._get_url(),
+                                          pool_size=10,
+                                          pool_pre_ping=True)
+        self.session_factory = async_sessionmaker(db_assistant_mysql_client_manager.engine,
+                                                  autoflush=True,
+                                                  expire_on_commit=False)
+
+    async def close(self):
+        await self.engine.dispose()
+
+
+db_assistant_mysql_client_manager = MysqlClientManager(app_config.db_assistant)
+
+if __name__ == '__main__':
+    db_assistant_mysql_client_manager.init()
+
+
+    async def test():
+        async with db_assistant_mysql_client_manager.session_factory() as session:
+            result = await session.execute(text("select * from archive_messages limit 10"))
+            rows = result.mappings().fetchall()
+            print(type(rows[0]))
+            print(rows)
+
+
+    asyncio.run(test())

app/conf/app_config.py

@@ -0,0 +1,72 @@
+from dataclasses import dataclass
+from pathlib import Path
+
+from omegaconf import OmegaConf
+
+
+# 日志配置
+@dataclass
+class File:
+    enable: bool
+    level: str
+    path: str
+    rotation: str
+    retention: str
+
+
+@dataclass
+class Console:
+    enable: bool
+    level: str
+
+
+@dataclass
+class LoggingConfig:
+    file: File
+    console: Console
+
+# 数据库配置
+@dataclass
+class DBConfig:
+    host: str
+    port: int
+    user: str
+    password: str
+    database: str
+
+@dataclass
+class EmbeddingConfig:
+    host: str
+    port: int
+    model: str
+
+@dataclass
+class MilvusConfig:
+    host: str
+    port: int
+    user: str
+    password: str
+    embedding_size: int
+
+@dataclass
+class LLMConfig:
+    model_name: str
+    api_key: str
+    base_url: str
+
+@dataclass
+class AppConfig:
+    logging: LoggingConfig
+    db_assistant: DBConfig
+    embedding: EmbeddingConfig
+    llm: LLMConfig
+    milvus: MilvusConfig
+
+
+config_file = Path(__file__).parents[2] / 'conf' / 'app_config.yaml'
+context = OmegaConf.load(config_file)
+schema = OmegaConf.structured(AppConfig)
+app_config: AppConfig = OmegaConf.to_object(OmegaConf.merge(schema, context))
+
+if __name__ == '__main__':
+    print(app_config.db_assistant.host)

app/dto/summary_dto.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class SummaryDto:
+    msg_date: str
+    from_user: str
+    to_user: str
+    summary: str

app/llm.py

@@ -0,0 +1,5 @@
+from langchain.chat_models import init_chat_model
+
+from app.conf.app_config import app_config
+
+llm = init_chat_model(model=app_config.llm.model_name, api_key=app_config.llm.api_key, base_url=app_config.llm.base_url ,temperature=1,extra_body={"thinking": {"type": "disabled"}},)

app/models/mysql/__init__.py

@@ -0,0 +1,3 @@
+from .archive_messages import ArchiveMessages
+
+__all__ = ["ArchiveMessages"]

app/models/mysql/archive_messages.py

@@ -0,0 +1,64 @@
+import json
+from datetime import datetime
+
+from sqlalchemy import BigInteger, String, Text, Integer, TIMESTAMP
+from sqlalchemy.orm import DeclarativeBase, mapped_column
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class ArchiveMessages(Base):
+
+
+    __tablename__ = 'archive_messages'
+    __table_args__ = {
+        'comment': '企微会话存档消息表'
+    }
+
+    id = mapped_column(BigInteger, primary_key=True, autoincrement=True, comment='自增主键')
+    msgid = mapped_column(String(255), nullable=False, comment='企微消息唯一ID')
+    seq = mapped_column(BigInteger, nullable=False, comment='存档序列号，用于增量拉取')
+    corp_id = mapped_column(String(64), nullable=False, comment='企业ID')
+    action = mapped_column(String(32), default='send', comment='send/recall/agree/disagree')
+    from_user = mapped_column(String(128), nullable=False, comment='发送者userid')
+    from_role = mapped_column(String(32), nullable=False, comment='发送者角色: INTERNAL-企业内部成员 EXTERNAL-外部联系人 SYSTEM-系统')
+    to_user = mapped_column(String(128), nullable=True, comment='接收者userid（单聊）')
+    tolist = mapped_column(Text, nullable=True, comment='接收者列表（群聊），JSON数组字符串')
+    roomid = mapped_column(String(255), nullable=True, comment='群聊ID（群聊时有值）')
+    msgtype = mapped_column(String(64), nullable=False, comment='text/image/voice/video/file/link/location/weapp/chatrecord/voip')
+    msgtime = mapped_column(BigInteger, nullable=False, comment='消息发送时间，UTC毫秒时间戳')
+    content = mapped_column(Text, nullable=True, comment='文本消息内容（长文本使用LONGTEXT）')
+    media_data = mapped_column(Text, nullable=True, comment='媒体消息元数据（sdkfileid/文件大小/时长等），JSON字符串')
+    session_id = mapped_column(String(128), nullable=True, comment='关联的会话ID（业务生成）')
+    decrypt_status = mapped_column(Integer, default=1, comment='解密状态: 1成功 2失败')
+    decrypt_error = mapped_column(String(512), nullable=True, comment='解密失败原因')
+    created_at = mapped_column(TIMESTAMP, nullable=False, default=datetime.now, comment='创建时间')
+    updated_at = mapped_column(TIMESTAMP, nullable=False, default=datetime.now, onupdate=datetime.now, comment='更新时间')
+
+    def to_dict(self):
+        return {
+            'id': self.id,
+            'msgid': self.msgid,
+            'seq': self.seq,
+            'corp_id': self.corp_id,
+            'action': self.action,
+            'from_user': self.from_user,
+            'from_role': self.from_role,
+            'to_user': self.to_user,
+            'tolist': self.tolist,
+            'roomid': self.roomid,
+            'msgtype': self.msgtype,
+            'msgtime': self.msgtime,
+            'content': self.content,
+            'media_data': self.media_data,
+            'session_id': self.session_id,
+            'decrypt_status': self.decrypt_status,
+            'decrypt_error': self.decrypt_error,
+            'created_at': self.created_at.isoformat() if self.created_at else None,
+            'updated_at': self.updated_at.isoformat() if self.updated_at else None
+        }
+
+    def __repr__(self):
+        return json.dumps(self.to_dict())

app/repository/archive_messages_repository.py

@@ -0,0 +1,72 @@
+from typing import List
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.mysql import ArchiveMessages
+
+
+class ArchiveMessagesRepository:
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def get_message_statistics_by_date(self, start_date: str = None, end_date: str = None) -> List[ArchiveMessages]:
+        """
+        根据日期范围获取消息记录
+
+        Args:
+            start_date: 开始日期 (格式: YYYY-MM-DD)
+            end_date: 结束日期 (格式: YYYY-MM-DD)
+
+        Returns:
+            ArchiveMessages 对象列表
+        """
+        base_query = """
+            SELECT DATE(created_at) AS created_at, from_user
+            FROM archive_messages 
+        """
+        conditions = []
+        params = {}
+
+        if start_date:
+            conditions.append(" and DATE(created_at) >= :start_date")
+            params["start_date"] = start_date
+
+        if end_date:
+            conditions.append("DATE(created_at) <= :end_date")
+            params["end_date"] = end_date
+        base_query += " WHERE 1=1 and roomid='' and msgtype='text' "
+        #base_query += " WHERE (from_user='wmI1AkDQAA3h0jxrRpeaHZhMKeHExA4w' )or( from_user='LiHeYi'  and to_user='[\"wmI1AkDQAA3h0jxrRpeaHZhMKeHExA4w\"]') and roomid='' and msgtype='text' "
+        if conditions:
+            base_query += "  AND ".join(conditions)+"  "
+        base_query += " GROUP BY DATE(created_at), from_user ORDER BY created_at DESC, from_user;"
+
+        query = text(base_query)
+        result = await self.session.execute(query, params)
+        return [ArchiveMessages(**dict(row)) for row in result.mappings().fetchall()]
+
+
+    async def get_message_statistics_by_date_user(self, day: str = None, from_user: str = None, to_user: str = None) -> List[ArchiveMessages]:
+        """
+        根据日期+发送人+接收人查询记录
+        Returns:
+            ArchiveMessages 对象列表
+        """
+        if to_user:
+            sql = """
+                SELECT * FROM archive_messages 
+                WHERE DATE(created_at) = :day 
+                  AND from_user = :from_user 
+                  AND to_user = :to_user and msgtype='text'
+                ORDER BY created_at DESC
+            """
+            result = await self.session.execute(text(sql), {"day": day, "from_user": from_user, "to_user": to_user})
+        else :
+            sql = """
+                SELECT from_user,to_user FROM archive_messages 
+                WHERE DATE(created_at) = :day 
+                  AND from_user = :from_user and msgtype='text'
+                group by to_user
+            """
+            result = await self.session.execute(text(sql), {"day": day, "from_user": from_user})
+        return [ArchiveMessages(**dict(row)) for row in result.mappings().fetchall()]

app/repository/milvus/summary_repository.py

@@ -0,0 +1,156 @@
+from typing import Optional, List, Dict, Any
+
+from pymilvus import MilvusClient, DataType
+
+from app.client.milvus_client_manager import milvus_client
+from app.conf.app_config import app_config
+
+
+class SummaryRepository:
+    collection_name: str = 'message_summary'
+
+    def __init__(self):
+        self.dim = app_config.milvus.embedding_size
+        self._ensure_connection()
+
+    def _ensure_connection(self):
+        if milvus_client.client is None:
+            milvus_client.init()
+
+    def create_collection(self, dim: int = None) -> bool:
+        """
+        创建消息摘要集合
+        
+        Args:
+            dim: 向量维度，默认从配置读取
+        
+        Returns:
+            是否成功创建（已存在返回False）
+        """
+        if dim is not None:
+            self.dim = dim
+
+        if milvus_client.has_collection(self.collection_name):
+            return False
+
+        schema = milvus_client.client.create_schema(
+            auto_id=True,
+            enable_dynamic_field=True,
+            description="消息摘要向量集合"
+        )
+
+        schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True, auto_id=True)
+        schema.add_field(field_name="msg_time", datatype=DataType.VARCHAR, max_length=100)
+        schema.add_field(field_name="msg_type", datatype=DataType.INT8)
+        schema.add_field(field_name="from_user", datatype=DataType.VARCHAR, max_length=150)
+        schema.add_field(field_name="to_user", datatype=DataType.VARCHAR, max_length=150)
+        schema.add_field(field_name="room_id", datatype=DataType.VARCHAR, max_length=150)
+        schema.add_field(field_name="message_context", datatype=DataType.VARCHAR, max_length=65535)
+        schema.add_field(field_name="summary", datatype=DataType.VARCHAR, max_length=65535)
+        schema.add_field(field_name="message_dense_vector", datatype=DataType.FLOAT_VECTOR, dim=self.dim)
+        schema.add_field(field_name="summary_dense_vector", datatype=DataType.FLOAT_VECTOR, dim=self.dim)
+        schema.add_field(field_name="created_at", datatype=DataType.INT64)
+
+        index_params = milvus_client.client.prepare_index_params()
+        index_params.add_index(
+            field_name="message_dense_vector",
+            index_type="HNSW",
+            metric_type="COSINE",
+            params={"M": 32, "efConstruction": 300}
+        )
+        index_params.add_index(
+            field_name="summary_dense_vector",
+            index_type="HNSW",
+            metric_type="COSINE",
+            params={"M": 32, "efConstruction": 300}
+        )
+        milvus_client.client.create_collection(
+            collection_name=self.collection_name,
+            schema=schema,
+            index_params=index_params
+        )
+
+        return True
+
+    def insert(self, msg_time: List[str], msg_type: List[int], from_user: List[str],
+               to_user: List[str], room_id: List[str],message_context: List[str], summary: List[str], message_dense_vector: List[List[float]],
+               summary_dense_vector: List[List[float]]) -> List[int]:
+        """
+        插入消息摘要数据
+        
+        Args:
+            msg_time: 消息时间列表
+            msg_type: 消息类型列表
+            from_user: 发送者列表
+            to_user: 接收者列表
+            room_id: 群聊ID列表
+            summary: 消息摘要列表
+            dense_vector: 向量列表
+        
+        Returns:
+            插入的数据ID列表
+        """
+        self._ensure_connection()
+        self.create_collection()
+        import time
+        created_at = [int(time.time() * 1000)] * len(msg_time)
+
+        data = []
+        for i in range(len(msg_time)):
+            data.append({
+                "msg_time": msg_time[i],
+                "msg_type": msg_type[i],
+                "from_user": from_user[i],
+                "to_user": to_user[i],
+                "room_id": room_id[i],
+                "message_context": message_context[i],
+                "summary": summary[i],
+                "message_dense_vector": message_dense_vector[i],
+                "summary_dense_vector": summary_dense_vector[i],
+                "created_at": created_at[i]
+            })
+
+        result = milvus_client.client.insert(
+            collection_name=self.collection_name,
+            data=data
+        )
+
+        milvus_client.client.flush(self.collection_name)
+        return result.get("ids", [])
+
+    def delete(self, expr: str):
+        """
+        删除数据
+        
+        Args:
+            expr: 删除条件表达式
+        """
+        self._ensure_connection()
+
+        milvus_client.client.delete(
+            collection_name=self.collection_name,
+            filter=expr
+        )
+        milvus_client.client.flush(self.collection_name)
+
+    def drop_collection(self):
+        """删除集合"""
+        milvus_client.drop_collection(self.collection_name)
+
+    def get_collection_stats(self) -> Dict[str, Any]:
+        """获取集合统计信息"""
+        self._ensure_connection()
+        
+        try:
+            result = milvus_client.client.get_collection_stats(collection_name=self.collection_name)
+            return {
+                "num_entities": result.get("row_count", 0),
+                "collection_name": self.collection_name
+            }
+        except Exception as e:
+            print(f"获取统计信息失败: {e}")
+            return {"num_entities": 0, "collection_name": self.collection_name}
+
+    def close(self):
+        """关闭连接"""
+        milvus_client.close()

app/summary/__init__.py

@@ -0,0 +1,4 @@
+from .router import router as summary_router
+from .service import SummaryService
+
+__all__ = ["summary_router", "SummaryService"]

app/summary/router.py

@@ -0,0 +1,26 @@
+from typing import List, Dict, Any, Optional
+
+from fastapi import APIRouter, Query
+
+from app.summary.service import SummaryService
+
+router = APIRouter(prefix="/summary", tags=["summary"])
+
+
+@router.get("/message-statistics", response_model=List[Dict[str, Any]])
+async def get_message_statistics(
+    start_date: Optional[str] = Query(None, description="开始日期 (格式: YYYY-MM-DD)"),
+    end_date: Optional[str] = Query(None, description="结束日期 (格式: YYYY-MM-DD)")
+):
+    """
+    获取按日期和发送者分组的消息统计数据
+    
+    SQL逻辑:
+        SELECT DATE(created_at) AS created_at, from_user 
+        FROM archive_messages 
+        GROUP BY DATE(created_at), from_user 
+        ORDER BY created_at DESC;
+    """
+    if start_date or end_date:
+        return await SummaryService.get_message_statistics_by_date(start_date, end_date)
+    return await SummaryService.get_message_statistics()

app/summary/service.py

@@ -0,0 +1,180 @@
+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())
+

app/summary/task.py

@@ -0,0 +1,88 @@
+import asyncio
+import logging
+from datetime import datetime, timedelta
+from typing import Optional
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+from app.client.mysql_client_manager import db_assistant_mysql_client_manager
+from app.repository.archive_messages_repository import ArchiveMessagesRepository
+from app.summary.service import SummaryService
+
+
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    handlers=[logging.FileHandler('summary_task.log'), logging.StreamHandler()]
+)
+logger = logging.getLogger(__name__)
+
+
+async def generate_daily_summary(date: Optional[str] = None):
+    """
+    生成指定日期的消息摘要，如果不传日期则生成前一天的摘要
+    
+    Args:
+        date: 日期字符串，格式 YYYY-MM-DD，不传则默认前一天
+    """
+    if not date:
+        yesterday = datetime.now() - timedelta(days=1)
+        date = yesterday.strftime("%Y-%m-%d")
+    
+    logger.info(f"开始生成 {date} 的消息摘要...")
+    
+    try:
+        db_assistant_mysql_client_manager.init()
+        async with db_assistant_mysql_client_manager.session_factory() as db_assistant:
+            archive_messages_repository = ArchiveMessagesRepository(db_assistant)
+            service = SummaryService(archive_messages_repository)
+            summaries = await service.get_message_summary(date)
+            
+            logger.info(f"生成完成，共生成 {len(summaries)} 条摘要")
+            for summary in summaries:
+                logger.info(f"日期: {summary.date}, 发送人: {summary.from_user}, 接收人: {summary.to_user}")
+        
+        return summaries
+    except Exception as e:
+        logger.error(f"生成摘要失败: {str(e)}", exc_info=True)
+        raise
+
+
+def start_scheduler():
+    """
+    启动定时任务调度器，每天凌晨2点执行
+    """
+    scheduler = AsyncIOScheduler(timezone='Asia/Shanghai')
+    
+    scheduler.add_job(
+        generate_daily_summary,
+        trigger='cron',
+        hour=2,
+        minute=0,
+        second=0,
+        id='daily_summary_task',
+        name='每日消息摘要任务',
+        replace_existing=True
+    )
+    
+    logger.info("定时任务调度器已启动，每天凌晨2:00执行消息摘要任务")
+    
+    scheduler.start()
+    
+    try:
+        asyncio.get_event_loop().run_forever()
+    except (KeyboardInterrupt, SystemExit):
+        logger.info("正在关闭定时任务调度器...")
+        scheduler.shutdown()
+
+
+if __name__ == '__main__':
+    import sys
+    
+    if len(sys.argv) > 1:
+        # 如果传入了日期参数，直接生成指定日期的摘要
+        target_date = sys.argv[1]
+        asyncio.run(generate_daily_summary(target_date))
+    else:
+        # 否则启动定时任务调度器
+        start_scheduler()

conf/app_config.yaml

@@ -0,0 +1,35 @@
+logging:
+  file:
+    enable: true
+    level: INFO
+    path: logs
+    rotation: "10 MB"
+    retention: "7 days"
+  console:
+    enable: true
+    level: INFO
+
+db_assistant:
+  host: sh-cdb-6fzlwnms.sql.tencentcdb.com
+  port: 63912
+  user: assistant_prod
+  password: U%$4Tu_C3+4
+  database: ai_assistant
+
+embedding:
+  host: localhost
+  port: 8081
+  model: BAAI/bge-large-zh-v1.5
+
+
+milvus:
+  host: localhost
+  port: 19530
+  user: root
+  password: Milvus
+  embedding_size: 1024
+
+llm:
+  model_name: deepseek-v4-flash
+  api_key: sk-8edc9f25d59643b3b08efea5d81c55a9
+  base_url: https://api.deepseek.com

main.py

@@ -0,0 +1,28 @@
+from fastapi import FastAPI
+
+from app.client.mysql_client_manager import db_assistant_mysql_client_manager
+from app.summary import summary_router
+
+app = FastAPI(title="Sales Assistant API", version="1.0.0")
+
+app.include_router(summary_router)
+
+
+@app.on_event("startup")
+async def startup():
+    db_assistant_mysql_client_manager.init()
+
+
+@app.on_event("shutdown")
+async def shutdown():
+    await db_assistant_mysql_client_manager.close()
+
+
+@app.get("/")
+async def root():
+    return {"message": "Sales Assistant API is running"}
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

pyproject.toml

@@ -0,0 +1,23 @@
+[project]
+name = "sales-assistant-py"
+version = "0.1.0"
+description = "Add your description here"
+requires-python = ">=3.12"
+dependencies = [
+    "aiohttp>=3.14.1",
+    "apscheduler>=3.11.2",
+    "asyncmy>=0.2.11",
+    "cryptography>=48.0.1",
+    "fastapi[standard]>=0.136.3",
+    "huggingface-hub>=1.18.0",
+    "jieba>=0.42.1",
+    "langchain>=1.3.7",
+    "langchain-deepseek>=1.1.0",
+    "langchain-huggingface>=1.2.2",
+    "langgraph>=1.2.4",
+    "loguru>=0.7.3",
+    "omegaconf>=2.3.0",
+    "pymilvus>=3.0.0",
+    "pyyaml>=6.0.3",
+    "sqlalchemy>=2.0.50",
+]