向量化（Embedding）

import dashscope
import numpy as np
from dashscope import TextEmbedding
# 以下为华北2（北京）地域的配置，调用时请将WorkspaceId替换为真实的业务空间ID，各地域的配置不同。
dashscope.base_http_api_url = "https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"

def cosine_similarity(a, b):
    """计算余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def semantic_search(query, documents, top_k=5):
    """语义搜索"""
    # 生成查询向量
    query_resp = TextEmbedding.call(
        model="qwen3.7-text-embedding",
        input=query,
        dimension=1024
    )
    query_embedding = query_resp.output['embeddings'][0]['embedding']

    # 生成文档向量
    doc_resp = TextEmbedding.call(
        model="qwen3.7-text-embedding",
        input=documents,
        dimension=1024
    )

    # 计算相似度
    similarities = []
    for i, doc_emb in enumerate(doc_resp.output['embeddings']):
        similarity = cosine_similarity(query_embedding, doc_emb['embedding'])
        similarities.append((i, similarity))

    # 排序并返回top_k结果
    similarities.sort(key=lambda x: x[1], reverse=True)
    return [(documents[i], sim) for i, sim in similarities[:top_k]]

# 使用示例
documents = [
    "人工智能是计算机科学的一个分支",
    "机器学习是实现人工智能的重要方法",
    "深度学习是机器学习的一个子领域"
]
query = "什么是AI？"
results = semantic_search(query, documents, top_k=2)
for doc, sim in results:
    print(f"相似度: {sim:.3f}, 文档: {doc}")

import dashscope
import numpy as np
from dashscope import TextEmbedding
# 以下为华北2（北京）地域的配置，调用时请将WorkspaceId替换为真实的业务空间ID，各地域的配置不同。
dashscope.base_http_api_url = "https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"

def cosine_similarity(a, b):
    """计算余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
def build_recommendation_system(user_history, all_items, top_k=10):
    """构建推荐系统"""
    # 生成用户历史向量
    history_resp = TextEmbedding.call(
        model="qwen3.7-text-embedding",
        input=user_history,
        dimension=1024
    )

    # 计算用户偏好向量（取平均）
    user_embedding = np.mean([
        emb['embedding'] for emb in history_resp.output['embeddings']
    ], axis=0)

    # 生成所有物品向量
    items_resp = TextEmbedding.call(
        model="qwen3.7-text-embedding",
        input=all_items,
        dimension=1024
    )

    # 计算推荐分数
    recommendations = []
    for i, item_emb in enumerate(items_resp.output['embeddings']):
        score = cosine_similarity(user_embedding, item_emb['embedding'])
        recommendations.append((all_items[i], score))

    # 排序并返回推荐结果
    recommendations.sort(key=lambda x: x[1], reverse=True)
    return recommendations[:top_k]

# 使用示例
user_history = ["科幻类", "动作类", "悬疑类"]
all_movies = ["未来世界", "太空探险", "古代战争", "浪漫之旅", "超级英雄"]
recommendations = build_recommendation_system(user_history, all_movies)
for movie, score in recommendations:
    print(f"推荐分数: {score:.3f}, 电影: {movie}")

# 需要安装 scikit-learn: pip install scikit-learn
import dashscope
import numpy as np
from sklearn.cluster import KMeans
# 以下为华北2（北京）地域的配置，调用时请将WorkspaceId替换为真实的业务空间ID，各地域的配置不同。
dashscope.base_http_api_url = "https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"

def cluster_texts(texts, n_clusters=2):
    """将一组文本进行聚类"""
    # 1. 获取所有文本的向量
    resp = dashscope.TextEmbedding.call(
        model="qwen3.7-text-embedding",
        input=texts,
        dimension=1024
    )
    embeddings = np.array([item['embedding'] for item in resp.output['embeddings']])

    # 2. 使用KMeans算法进行聚类
    kmeans = KMeans(n_clusters=n_clusters, random_state=0, n_init='auto').fit(embeddings)

    # 3. 整理并返回结果
    clusters = {i: [] for i in range(n_clusters)}
    for i, label in enumerate(kmeans.labels_):
        clusters[label].append(texts[i])
    return clusters

# 使用示例
documents_to_cluster = [
    "手机公司A发售新款手机",
    "搜索引擎公司B推出新款系统",
    "世界杯决赛阿根廷对阵法国",
    "奥运会中国队再添一金",
    "某公司发布最新AI芯片",
    "欧洲杯赛事报道"
]
clusters = cluster_texts(documents_to_cluster, n_clusters=2)
for cluster_id, docs in clusters.items():
    print(f"--- 类别 {cluster_id} ---")
    for doc in docs:
        print(f"- {doc}")

import dashscope
import numpy as np
# 以下为华北2（北京）地域的配置，调用时请将WorkspaceId替换为真实的业务空间ID，各地域的配置不同。
dashscope.base_http_api_url = "https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"

def cosine_similarity(a, b):
    """计算余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def classify_text_zero_shot(text, labels):
    """零样本文本分类"""
    # 1. 获取输入文本和所有标签的向量
    resp = dashscope.TextEmbedding.call(
        model="qwen3.7-text-embedding",
        input=[text] + labels,
        dimension=1024
    )
    embeddings = resp.output['embeddings']
    text_embedding = embeddings[0]['embedding']
    label_embeddings = [emb['embedding'] for emb in embeddings[1:]]

    # 2. 计算与每个标签的相似度
    scores = [cosine_similarity(text_embedding, label_emb) for label_emb in label_embeddings]

    # 3. 返回相似度最高的标签
    best_match_index = np.argmax(scores)
    return labels[best_match_index], scores[best_match_index]

# 使用示例
text_to_classify = "这件衣服的料子很舒服，款式也好看"
possible_labels = ["数码产品", "服装配饰", "食品饮料", "家居生活"]

label, score = classify_text_zero_shot(text_to_classify, possible_labels)
print(f"输入文本: '{text_to_classify}'")
print(f"最匹配的分类是: '{label}' (相似度: {score:.3f})")

import dashscope
import numpy as np
# 以下为华北2（北京）地域的配置，调用时请将WorkspaceId替换为真实的业务空间ID，各地域的配置不同。
dashscope.base_http_api_url = "https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"

def cosine_similarity(a, b):
    """计算余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def detect_anomaly(new_comment, normal_comments, threshold=0.6):
    # 1. 向量化所有正常评论和新评论
    all_texts = normal_comments + [new_comment]
    resp = dashscope.TextEmbedding.call(
        model="qwen3.7-text-embedding",
        input=all_texts,
        dimension=1024
    )
    embeddings = [item['embedding'] for item in resp.output['embeddings']]

    # 2. 计算正常评论的中心向量（平均值）
    normal_embeddings = np.array(embeddings[:-1])
    normal_center_vector = np.mean(normal_embeddings, axis=0)

    # 3. 计算新评论与中心向量的相似度
    new_comment_embedding = np.array(embeddings[-1])
    similarity = cosine_similarity(new_comment_embedding, normal_center_vector)

    # 4. 判断是否为异常
    is_anomaly = similarity < threshold
    return is_anomaly, similarity

# 使用示例
normal_user_comments = [
    "今天的会议很有成效",
    "项目进展顺利",
    "下周发布新版本",
    "用户反馈良好"
]

test_comments = {
    "正常评论": "功能符合预期",
    "异常-无意义乱码": "asdfghjkl zxcvbnm"
}

print("--- 异常检测示例 ---")
for desc, comment in test_comments.items():
    is_anomaly, score = detect_anomaly(comment, normal_user_comments)
    result = "是" if is_anomaly else "否"
    print(f"评论: '{comment}'")
    print(f"是否为异常: {result} (与正常样本相似度: {score:.3f})\n")