大模型应用：从静态到动态：增量聚类+混元大模型破解无限流数据智能处理难题.98

{
  "ticket_id": "10001",
  "content": "我的快递丢了，已经3天了，能理赔吗？",
  "create_time": "2026-02-25 10:00:01",
  "user_id": "u89765"
}

from sentence_transformers import SentenceTransformer
from modelscope import snapshot_download

cache_dir = "D:\\modelscope\\hub"
model_dir = snapshot_download(
    model_id="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
    cache_dir=cache_dir,
    revision="master"  # 或指定分支/commit
)
# 1. 加载Embedding模型（轻量级Sentence-BERT）
model = SentenceTransformer("D:\\modelscope\\hub\\sentence-transformers\\paraphrase-multilingual-MiniLM-L12-v2")

def get_local_embedding(text):
    """本地生成Embedding向量"""
    # 返回384维向量
    embedding = model.encode(text, normalize_embeddings=True)
    return embedding

# 测试
text = "我的快递丢了，能理赔吗？"
embedding = get_local_embedding(text)
print(f"Embedding向量长度：{len(embedding)}")  

import numpy as np
from sklearn.cluster import MiniBatchKMeans
from sklearn.metrics import pairwise_distances_argmin_min

class IncrementalClustering:
    def __init__(self, n_clusters=5, batch_size=100):
        """
        初始化增量聚类器
        :param n_clusters: 初始聚类数量
        :param batch_size: 微批次大小
        """
        self.n_clusters = n_clusters
        self.batch_size = batch_size
        # 初始化MiniBatchKMeans（增量K-Means）
        self.cluster_model = MiniBatchKMeans(
            n_clusters=n_clusters,
            batch_size=batch_size,
            random_state=42,
            init_size=1000  # 初始聚类中心数量
        )
        # 保存每个聚类的文本示例（用于后续打标）
        self.cluster_texts = {i: [] for i in range(n_clusters)}
        # 保存聚类中心
        self.cluster_centers = None

    def update_cluster(self, embeddings, texts):
        """
        增量更新聚类
        :param embeddings: 新数据的Embedding向量列表（np.array）
        :param texts: 对应的文本列表
        :return: 聚类结果（每个数据的聚类标签）
        """
        # 训练模型（增量更新）
        labels = self.cluster_model.partial_fit(embeddings).predict(embeddings)
        # 更新聚类中心
        self.cluster_centers = self.cluster_model.cluster_centers_
        # 保存每个聚类的文本示例（最多保存20条，避免内存占用）
        for label, text in zip(labels, texts):
            if len(self.cluster_texts[label]) < 20:
                self.cluster_texts[label].append(text)
        return labels

    def get_cluster_examples(self):
        """获取每个聚类的文本示例"""
        return self.cluster_texts

    def get_nearest_text(self, center, texts):
        """获取离聚类中心最近的文本（最具代表性）"""
        if len(texts) == 0:
            return ""
        # 计算距离
        distances = pairwise_distances_argmin_min([center], texts)[1][0]
        # 找到最近的文本
        nearest_idx = np.argmin(distances)
        return texts[nearest_idx]

def cluster_labeling(cluster_examples):
    """
    对聚类结果进行打标（使用腾讯混元大模型）
    :param cluster_examples: 每个聚类的文本示例（dict，{聚类标签: [文本1, 文本2,...]}）
    :return: 聚类标签映射（{聚类标签: 打标结果}）
    """
    from openai import OpenAI

    # 配置混元API
    api_key = 'sk-bWlJPKjBrSFGoQ0Ys***********************'
    client = OpenAI(
        api_key=api_key,
        base_url="https://api.hunyuan.cloud.tencent.com/v1",
    )

    label_mapping = {}
    for cluster_id, examples in cluster_examples.items():
        if len(examples) == 0:
            label_mapping[cluster_id] = "未知主题"
            continue
        # 构建Prompt
        prompt = f"""
        你是一个专业的客服工单分类分析师，需要根据以下文本示例总结聚类的核心内容：
        文本示例：
        {chr(10).join([f"- {text}" for text in examples[:10]])}
        总结要求：
        1. 用不超过20个字的标题概括聚类核心内容；
        2. 标题要简洁、准确，符合客服工单场景；
        3. 避免模糊表述，如"其他问题""未知主题"；
        4. 核心要突出用户诉求，如"快递丢失理赔咨询""退款申请处理"。
        """
        try:
            # 调用混元大模型
            completion = client.chat.completions.create(
                model="hunyuan-lite",
                messages=[
                    {'role': 'system', 'content': '你是一个专业的客服工单分类分析师'},
                    {'role': 'user', 'content': prompt}
                ],
                temperature=0.1,
                max_tokens=50
            )
            # 提取打标结果
            label = completion.choices[0].message.content.strip()
            label_mapping[cluster_id] = label
        except Exception as e:
            print(f"聚类{cluster_id}打标失败：{e}")
            label_mapping[cluster_id] = f"未识别-{cluster_id}"
    return label_mapping

import time
import json

# 1. 模拟流数据生成（初学者测试用）
def generate_simulated_tickets():
    """生成模拟工单数据"""
    ticket_templates = [
        "我的快递丢了，能理赔吗？",
        "快递延迟了3天，要求赔偿",
        "退款申请提交后，商家不处理",
        "商品质量有问题，要求退货",
        "物流信息不更新，怎么办？"
    ]
    while True:
        # 随机选择模板生成工单
        content = np.random.choice(ticket_templates)
        ticket = {
            "ticket_id": f"sim_{int(time.time())}",
            "content": content,
            "create_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "user_id": f"u_{np.random.randint(1000, 9999)}"
        }
        yield ticket
        # 模拟每秒生成1条工单
        time.sleep(1)

# 2. 流数据处理主逻辑
def stream_processing():
    """实时流数据处理"""
    # 初始化增量聚类器
    clusterer = IncrementalClustering(n_clusters=5, batch_size=10)
    # 缓存批次数据（每10条处理一次）
    batch_embeddings = []
    batch_texts = []
    batch_size = 10

    # 生成模拟数据
    ticket_generator = generate_simulated_tickets()

    for ticket in ticket_generator:
        text = ticket["content"]
        # 清洗文本
        text = text.strip().replace("！", "").replace("？", "")
        if not text:
            continue

        # 获取Embedding
        embedding = get_local_embedding(text)
        if embedding is None:
            continue

        # 添加到批次
        batch_embeddings.append(embedding)
        batch_texts.append(text)

        # 批次满了，更新聚类
        if len(batch_embeddings) >= batch_size:
            # 转换为numpy数组
            embeddings_np = np.array(batch_embeddings)
            # 增量更新聚类
            labels = clusterer.update_cluster(embeddings_np, batch_texts)
            # 获取聚类示例，打标
            cluster_examples = clusterer.get_cluster_examples()
            label_mapping = cluster_labeling(cluster_examples)

            # 输出结果
            print(f"\n=== 批次处理结果（{time.strftime('%Y-%m-%d %H:%M:%S')}）===")
            for i, (text, label) in enumerate(zip(batch_texts, labels)):
                print(f"工单{i+1}：{text} → 聚类{label} → 标签：{label_mapping[label]}")

            # 清空批次
            batch_embeddings = []
            batch_texts = []

# 启动处理
if __name__ == "__main__":
    stream_processing()

from flask import Flask, jsonify

app = Flask(__name__)
# 全局变量保存最新的聚类标签映射
latest_label_mapping = {}

@app.route("/api/cluster_labels", methods=["GET"])
def get_cluster_labels():
    """获取最新的聚类标签"""
    return jsonify({
        "status": "success",
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "data": latest_label_mapping
    })

# 修改stream_processing中的打标部分，更新全局变量
def stream_processing():
    global latest_label_mapping
    # ... 原有逻辑 ...
    label_mapping = cluster_labeling(cluster_examples)
    # 更新全局变量
    latest_label_mapping = label_mapping
    # ... 原有逻辑 ...

# 启动Flask服务
if __name__ == "__main__":
    # 多线程启动：一个线程处理流数据，一个线程启动Flask服务
    import threading
    t = threading.Thread(target=stream_processing)
    t.start()
    app.run(host="0.0.0.0", port=5000)

from sentence_transformers import SentenceTransformer
from modelscope import snapshot_download

cache_dir = "D:\\modelscope\\hub"
model_dir = snapshot_download(
    model_id="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
    cache_dir=cache_dir,
    revision="master"  # 或指定分支/commit
)
# 1. 加载Embedding模型（轻量级Sentence-BERT）
model = SentenceTransformer("D:\\modelscope\\hub\\sentence-transformers\\paraphrase-multilingual-MiniLM-L12-v2")

def get_local_embedding(text):
    """本地生成Embedding向量"""
    # 返回384维向量
    embedding = model.encode(text, normalize_embeddings=True)
    return embedding

# 测试
text = "我的快递丢了，能理赔吗？"
embedding = get_local_embedding(text)
print(f"Embedding向量长度：{len(embedding)}")  


import numpy as np
from sklearn.cluster import MiniBatchKMeans
from sklearn.metrics import pairwise_distances_argmin_min

class IncrementalClustering:
    def __init__(self, n_clusters=5, batch_size=100):
        """
        初始化增量聚类器
        :param n_clusters: 初始聚类数量
        :param batch_size: 微批次大小
        """
        self.n_clusters = n_clusters
        self.batch_size = batch_size
        # 初始化MiniBatchKMeans（增量K-Means）
        self.cluster_model = MiniBatchKMeans(
            n_clusters=n_clusters,
            batch_size=batch_size,
            random_state=42,
            init_size=1000  # 初始聚类中心数量
        )
        # 保存每个聚类的文本示例（用于后续打标）
        self.cluster_texts = {i: [] for i in range(n_clusters)}
        # 保存聚类中心
        self.cluster_centers = None

    def update_cluster(self, embeddings, texts):
        """
        增量更新聚类
        :param embeddings: 新数据的Embedding向量列表（np.array）
        :param texts: 对应的文本列表
        :return: 聚类结果（每个数据的聚类标签）
        """
        # 训练模型（增量更新）
        labels = self.cluster_model.partial_fit(embeddings).predict(embeddings)
        # 更新聚类中心
        self.cluster_centers = self.cluster_model.cluster_centers_
        # 保存每个聚类的文本示例（最多保存20条，避免内存占用）
        for label, text in zip(labels, texts):
            if len(self.cluster_texts[label]) < 20:
                self.cluster_texts[label].append(text)
        return labels

    def get_cluster_examples(self):
        """获取每个聚类的文本示例"""
        return self.cluster_texts

    def get_nearest_text(self, center, texts):
        """获取离聚类中心最近的文本（最具代表性）"""
        if len(texts) == 0:
            return ""
        # 计算距离
        distances = pairwise_distances_argmin_min([center], texts)[1][0]
        # 找到最近的文本
        nearest_idx = np.argmin(distances)
        return texts[nearest_idx]


def cluster_labeling(cluster_examples):
    """
    对聚类结果进行打标（使用腾讯混元大模型）
    :param cluster_examples: 每个聚类的文本示例（dict，{聚类标签: [文本1, 文本2,...]}）
    :return: 聚类标签映射（{聚类标签: 打标结果}）
    """
    from openai import OpenAI

    # 配置混元API
    api_key = 'sk-bW**********************5NP8Ze'
    client = OpenAI(
        api_key=api_key,
        base_url="https://api.hunyuan.cloud.tencent.com/v1",
    )

    label_mapping = {}
    for cluster_id, examples in cluster_examples.items():
        if len(examples) == 0:
            label_mapping[cluster_id] = "未知主题"
            continue
        # 构建Prompt
        prompt = f"""
        你是一个专业的客服工单分类分析师，需要根据以下文本示例总结聚类的核心内容：
        文本示例：
        {chr(10).join([f"- {text}" for text in examples[:10]])}
        总结要求：
        1. 用不超过20个字的标题概括聚类核心内容；
        2. 标题要简洁、准确，符合客服工单场景；
        3. 避免模糊表述，如"其他问题""未知主题"；
        4. 核心要突出用户诉求，如"快递丢失理赔咨询""退款申请处理"。
        """
        try:
            # 调用混元大模型
            completion = client.chat.completions.create(
                model="hunyuan-lite",
                messages=[
                    {'role': 'system', 'content': '你是一个专业的客服工单分类分析师'},
                    {'role': 'user', 'content': prompt}
                ],
                temperature=0.1,
                max_tokens=50
            )
            # 提取打标结果
            label = completion.choices[0].message.content.strip()
            label_mapping[cluster_id] = label
        except Exception as e:
            print(f"聚类{cluster_id}打标失败：{e}")
            label_mapping[cluster_id] = f"未识别-{cluster_id}"
    return label_mapping



import time
import json

# 1. 模拟流数据生成（初学者测试用）
def generate_simulated_tickets():
    """生成模拟工单数据"""
    ticket_templates = [
        "我的快递丢了，能理赔吗？",
        "快递延迟了3天，要求赔偿",
        "退款申请提交后，商家不处理",
        "商品质量有问题，要求退货",
        "物流信息不更新，怎么办？"
    ]
    while True:
        # 随机选择模板生成工单
        content = np.random.choice(ticket_templates)
        ticket = {
            "ticket_id": f"sim_{int(time.time())}",
            "content": content,
            "create_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "user_id": f"u_{np.random.randint(1000, 9999)}"
        }
        yield ticket
        # 模拟每秒生成1条工单
        time.sleep(1)

# 2. 流数据处理主逻辑
def stream_processing():
    """实时流数据处理"""
    # 初始化增量聚类器
    clusterer = IncrementalClustering(n_clusters=5, batch_size=10)
    # 缓存批次数据（每10条处理一次）
    batch_embeddings = []
    batch_texts = []
    batch_size = 10

    # 生成模拟数据
    ticket_generator = generate_simulated_tickets()

    for ticket in ticket_generator:
        text = ticket["content"]
        # 清洗文本
        text = text.strip().replace("！", "").replace("？", "")
        if not text:
            continue

        # 获取Embedding
        embedding = get_local_embedding(text)
        if embedding is None:
            continue

        # 添加到批次
        batch_embeddings.append(embedding)
        batch_texts.append(text)

        # 批次满了，更新聚类
        if len(batch_embeddings) >= batch_size:
            # 转换为numpy数组
            embeddings_np = np.array(batch_embeddings)
            # 增量更新聚类
            labels = clusterer.update_cluster(embeddings_np, batch_texts)
            # 获取聚类示例，打标
            cluster_examples = clusterer.get_cluster_examples()
            label_mapping = cluster_labeling(cluster_examples)

            # 输出结果
            print(f"\n=== 批次处理结果（{time.strftime('%Y-%m-%d %H:%M:%S')}）===")
            for i, (text, label) in enumerate(zip(batch_texts, labels)):
                print(f"工单{i+1}：{text} → 聚类{label} → 标签：{label_mapping[label]}")

            # 清空批次
            batch_embeddings = []
            batch_texts = []

# 启动处理
if __name__ == "__main__":
    stream_processing()