文本与多模态向量化

import dashscope

dashscope.base_http_api_url = 'https://dashscope.aliyuncs.com/api/v1'

# 场景：为搜索引擎构建文档向量时，可添加指令来优化检索的向量质量。
resp = dashscope.TextEmbedding.call(
  model="text-embedding-v4",
  input="Research papers on machine learning",
  text_type="query",
  instruct="Given a research paper query, retrieve relevant research paper"
)

import dashscope
import numpy as np
from dashscope import TextEmbedding

dashscope.base_http_api_url = 'https://dashscope.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="text-embedding-v4",
  input=query,
  dimension=1024
  )
  query_embedding = query_resp.output['embeddings'][0]['embedding']

  # 生成文档向量
  doc_resp = TextEmbedding.call(
  model="text-embedding-v4",
  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 = [
  "Artificial intelligence is a branch of computer science",
  "Machine learning is an important method for achieving artificial intelligence",
  "Deep learning is a subfield of machine learning"
]
query = "What is AI?"
results = semantic_search(query, documents, top_k=2)
for doc, sim in results:
  print(f"Similarity: {sim:.3f}, Document: {doc}")

import dashscope
import numpy as np
from dashscope import TextEmbedding

dashscope.base_http_api_url = 'https://dashscope.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="text-embedding-v4",
  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="text-embedding-v4",
  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 = ["Science Fiction", "Action", "Suspense"]
all_movies = ["Future World", "Space Adventure", "Ancient War", "Romantic Journey", "Superhero"]
recommendations = build_recommendation_system(user_history, all_movies)
for movie, score in recommendations:
  print(f"Recommendation Score: {score:.3f}, Movie: {movie}")

# 需要安装 scikit-learn：pip install scikit-learn
import dashscope
import numpy as np
from sklearn.cluster import KMeans

dashscope.base_http_api_url = 'https://dashscope.aliyuncs.com/api/v1'

def cluster_texts(texts, n_clusters=2):
  """对一组文本进行聚类"""
  # 1. 获取所有文本的向量
  resp = dashscope.TextEmbedding.call(
  model="text-embedding-v4",
  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 = [
  "Mobile phone company A releases a new phone",
  "Search engine company B launches a new system",
  "World Cup final: Argentina vs. France",
  "China wins another gold medal at the Olympics",
  "A company releases its latest AI chip",
  "European Cup match report"
]
clusters = cluster_texts(documents_to_cluster, n_clusters=2)
for cluster_id, docs in clusters.items():
  print(f"--- Cluster {cluster_id} ---")
  for doc in docs:
    print(f"- {doc}")

import dashscope
import numpy as np

dashscope.base_http_api_url = 'https://dashscope.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="text-embedding-v4",
  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 = "The fabric of this dress is comfortable and the style is nice"
possible_labels = ["Digital Products", "Apparel & Accessories", "Food & Beverage", "Home & Living"]

label, score = classify_text_zero_shot(text_to_classify, possible_labels)
print(f"Input text: '{text_to_classify}'")
print(f"Best matching category: '{label}' (Similarity: {score:.3f})")

import dashscope
import numpy as np

dashscope.base_http_api_url = 'https://dashscope.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="text-embedding-v4",
  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 = [
  "Today's meeting was productive",
  "The project is progressing smoothly",
  "The new version will be released next week",
  "User feedback is positive"
]

test_comments = {
  "Normal comment": "The feature works as expected",
  "Anomaly - meaningless garbled text": "asdfghjkl zxcvbnm"
}

print("--- Anomaly Detection Example ---")
for desc, comment in test_comments.items():
  is_anomaly, score = detect_anomaly(comment, normal_user_comments)
  result = "Yes" if is_anomaly else "No"
  print(f"Comment: '{comment}'")
  print(f"Is anomaly: {result} (Similarity to normal samples: {score:.3f})\n")