how/examples/example_text_embedding_usecase.py

#!/usr/bin/env python3
"""
text_embedding 实际应用场景示例
"""
import asyncio
from lib.text_embedding import SentenceModel
import numpy as np


async def usecase_1_semantic_search():
    """
    应用场景1: 语义搜索
    - 用户输入一个查询
    - 从大量文本中找出最相关的内容
    """
    print("=" * 60)
    print("应用场景1: 语义搜索")
    print("=" * 60)

    model = SentenceModel(dim=128)

    # 文档库（比如知识库、文章库）
    documents = [
        "深度学习是机器学习的一个重要分支",
        "神经网络模拟人脑的结构和功能",
        "自然语言处理用于理解和生成人类语言",
        "计算机视觉让机器能够理解图像",
        "今天天气晴朗，适合出门散步",
        "晚餐吃了美味的意大利面",
    ]

    # 用户查询
    query = "什么是深度学习？"

    print(f"\n用户查询: {query}")
    print(f"\n文档库中有 {len(documents)} 篇文档")
    print("\n搜索结果（按相关度排序）:")

    # 计算查询与每篇文档的相似度
    results = []
    for doc in documents:
        score = await model.similarity(query, doc)
        results.append((doc, score))

    # 排序并展示前3个最相关的
    results.sort(key=lambda x: x[1], reverse=True)

    for i, (doc, score) in enumerate(results[:3], 1):
        print(f"\n  Top {i} (相似度: {score:.3f})")
        print(f"  {doc}")


async def usecase_2_text_clustering():
    """
    应用场景2: 文本聚类
    - 把相似的文本分组
    """
    print("\n" + "=" * 60)
    print("应用场景2: 文本聚类（相似内容分组）")
    print("=" * 60)

    model = SentenceModel(dim=128)

    texts = [
        "机器学习算法很强大",
        "深度神经网络效果很好",
        "今天吃了火锅",
        "晚餐很美味",
        "AI技术发展迅速",
        "明天去爬山",
    ]

    print(f"\n待分组文本: {texts}")

    # 简单聚类：找出相似度 > 0.6 的文本对
    print("\n相似文本对（相似度 > 0.6）:")

    similar_pairs = []
    for i in range(len(texts)):
        for j in range(i + 1, len(texts)):
            score = await model.similarity(texts[i], texts[j])
            if score > 0.6:
                similar_pairs.append((texts[i], texts[j], score))

    for text1, text2, score in similar_pairs:
        print(f"\n  相似度: {score:.3f}")
        print(f"  - {text1}")
        print(f"  - {text2}")


async def usecase_3_duplicate_detection():
    """
    应用场景3: 重复/相似内容检测
    - 识别意思相同但表述不同的文本
    """
    print("\n" + "=" * 60)
    print("应用场景3: 重复内容检测")
    print("=" * 60)

    model = SentenceModel(dim=128)

    # 用户提交的问题列表
    questions = [
        "如何学习Python编程？",
        "怎样才能学好Python？",
        "Python入门教程有哪些？",
        "今天天气怎么样？",
        "北京今天的天气如何？",
    ]

    print(f"\n问题列表:")
    for i, q in enumerate(questions, 1):
        print(f"  {i}. {q}")

    # 检测重复（相似度 > 0.7 视为重复）
    print("\n检测到的重复问题（相似度 > 0.7）:")

    threshold = 0.7
    duplicates = []

    for i in range(len(questions)):
        for j in range(i + 1, len(questions)):
            score = await model.similarity(questions[i], questions[j])
            if score > threshold:
                duplicates.append((i+1, j+1, questions[i], questions[j], score))

    for idx1, idx2, q1, q2, score in duplicates:
        print(f"\n  问题{idx1} 和 问题{idx2} 相似 (相似度: {score:.3f})")
        print(f"  - {q1}")
        print(f"  - {q2}")


async def main():
    """运行所有应用场景示例"""

    # 场景1: 语义搜索
    await usecase_1_semantic_search()

    # 场景2: 文本聚类
    await usecase_2_text_clustering()

    # 场景3: 重复检测
    await usecase_3_duplicate_detection()

    print("\n" + "=" * 60)
    print("所有示例完成！")
    print("=" * 60)


if __name__ == "__main__":
    asyncio.run(main())