Qwen3-Embedding 全揭秘：从技术到服务，打造高效AI产品的关键路径

{  "input": [    "It 's a lovely film with lovely performances.",    "A warm , funny , engaging film."  ],  "model": "Qwen3-Embedding-4B"}

import requests
url = '修改为您自己的服务地址...v1/embeddings'token = '修改为您自己的token'request_body = {    "input": [        "It 's a lovely film with lovely performances.",        "A warm , funny , engaging film."    ],    "model": "Qwen3-Embedding-4B"}
headers = {"Authorization": token}resp = requests.post(url=url, headers=headers, json=request_body)
print(resp.content.decode())

vim emb.pypython3.8 emb.py

import requestsimport numpy as npimport jsonfrom sklearn.metrics.pairwise import cosine_similarity
# ========== 1. 配置信息 ==========url = '请输入您自己的服务的 URL'token = '请输入您自己的服务的 Token'
# ========== 2. 已有电影评论列表（你可以替换为你自己的数据）==========reviews = [    "The cinematography was breathtaking and the story deeply moving.",    "A dull plot with no character development, very disappointing.",    "This film is a masterpiece of visual storytelling and emotional depth.",    "The dialogue felt forced and the acting was painfully over-the-top.",    "An absolute joy to watch! The humor and pacing were perfect.",    "The soundtrack was the only redeeming quality in this otherwise terrible movie.",    "A gripping narrative that kept me hooked from start to finish.",    "Too many plot holes to take this film seriously.",    "The chemistry between the lead actors was electric!",    "A slow-burning bore with no payoff in the end.",    "Visually stunning but emotionally hollow.",    "This movie perfectly balances action, humor, and heartfelt moments.",    "The script was cliché-ridden and completely predictable.",    "A triumph of storytelling—every scene felt intentional and meaningful.",    "Overhyped and underwhelming. I fell asleep halfway through.",    "The director's unique vision shines through every frame of this film.",    "A chaotic mess of editing and disjointed scenes.",    "Hauntingly beautiful score that elevated the entire experience.",    "This film deserves all the awards for its technical brilliance.",    "The worst part? The characters had zero relatable motivations.",    "A fresh take on a classic genre—innovative and exciting!",    "Bland visuals and forgettable performances made this a snooze-fest.",    "The twist ending was so predictable I figured it out in Act 1.",    "A deeply moving exploration of human relationships and resilience.",    "This movie feels like a long, pretentious art school project.",    "The action sequences were thrilling but the story was paper-thin.",    "A cult classic in the making—quirky, bold, and unforgettable.",    "Dialogue fell flat and the comedic timing was painfully off.",    "Every frame is a work of art; this film redefines cinematic beauty.",    "An exhausting watch with no clear direction or purpose.",    "The lead actor's performance was raw and achingly authentic.",    "Overly reliant on CGI, sacrificing substance for flashy visuals.",    "A nostalgic gem that pays homage to classic cinema.",    "The pacing dragged for most of the film, killing any tension.",    "This film's originality and creativity left me in awe.",    "A shallow attempt at deep storytelling—felt entirely artificial.",    "The supporting cast stole the show with their impeccable performances.",    "A masterclass in tension-building and suspenseful storytelling.",    "The film's message was drowned in excessive runtime and filler scenes.",    "Darkly humorous with moments of profound philosophical insight.",    "A technically flawless production with zero emotional impact.",    "The screenplay was sharp, witty, and full of delightful surprises.",    "This movie tries too hard to be profound and fails miserably.",    "A visually arresting journey with a soulless narrative beneath.",    "The ensemble cast worked together like a perfectly orchestrated symphony.",    "A compelling concept wasted by poor execution and lazy writing.",    "This film will leave you thinking long after the credits roll.",    "The makeup and costume design were the only elements that impressed me.",    "A bold experiment in nonlinear storytelling that completely paid off.",    "Despite its flaws, the heart and passion behind this film shine through."]
# ========== 3. 获取向量表示（若已有 embeddings.json 可跳过）==========def get_embedding(text):    headers = {"Authorization": token}    payload = {        "input": [text],        "model": "Qwen3-Embedding-4B"    }    response = requests.post(url, headers=headers, json=payload)    return response.json()['data'][0]['embedding']
def generate_embeddings(reviews, filename='embeddings.json'):    embeddings = []    for i, review in enumerate(reviews):        print(f"Generating embedding for review {i+1}/{len(reviews)}")        emb = get_embedding(review)        embeddings.append(emb)
    with open(filename, 'w') as f:        json.dump({'reviews': reviews, 'embeddings': embeddings}, f)    print(f"Embeddings saved to {filename}")
# ========== 4. 加载向量 ==========def load_embeddings(filename='embeddings.json'):    with open(filename, 'r') as f:        data = json.load(f)    return data['reviews'], np.array(data['embeddings'])
# ========== 5. 查询最相似的评论 ==========def find_similar_reviews(query, reviews_list, embeddings_matrix, top_k=3):    query_emb = np.array(get_embedding(query)).reshape(1, -1)  # 修改了这行    similarities = cosine_similarity(query_emb, embeddings_matrix)[0]    indices = np.argsort(similarities)[::-1][:top_k]    results = [(reviews_list[i], similarities[i]) for i in indices]    return results
# ========== 6. 主程序入口 ==========if __name__ == '__main__':    # 第一次运行时取消注释，生成并保存向量，搜索时关闭    generate_embeddings(reviews)
    # 加载历史评论和对应的向量    reviews_list, embeddings_matrix = load_embeddings()
    # 输入你要查询的新评论    user_query = input("请输入你想查找相似评论的句子：")
    # 查找最相似的评论    similar_reviews = find_similar_reviews(user_query, reviews_list, embeddings_matrix)
    print("\n 最相似的评论如下：")    for i, (review, score) in enumerate(similar_reviews):        print(f"{i+1}. [相似度: {score:.4f}] {review}")

curl --location 'https://dashscope.aliyuncs.com...修改为您自己的服务地址.../text-embedding' \--header "Authorization: Bearer $DASHSCOPE_API_KEY" \--header 'Content-Type: application/json' \--data '{    "model": "text-embedding-v4",    "input": {        "texts": [        "买的东西质量真的好不枉我等了这么久啊，喜欢，以后还来这里买"        ]    },    "parameters": {      "dimension": 1024 # 指定向量维度（仅 text-embedding-v3及 text-embedding-v4支持该参数）    }}'

curl --location 'https://dashscope.aliyuncs.com/...修改为您自己的服务地址...embeddings' \--header "Authorization: Bearer $DASHSCOPE_API_KEY" \--header 'Content-Type: application/json' \--data '{    "model": "text-embedding-v4",    "input": "买的东西质量真的好不枉我等了这么久啊，喜欢，以后还来这里买",      "dimension": "1024",  # 指定向量维度（仅 text-embedding-v3及 text-embedding-v4支持该参数）    "encoding_format": "float"}'

import dashscopefrom http import HTTPStatusimport numpy as np# 计算余弦相似度（Cosine Similarity）的函数，主要用于衡量两个向量之间的相似性from sklearn.metrics.pairwise import cosine_similarity
# 配置 DashScope APIdashscope.api_key = "sk-xxxx"
# 商品描述列表，可以替换成自己的数据product_descriptions = [    "棉质宽松衬衫，适合春季通勤，透气舒适，简约百搭",    "高腰牛仔裤，修身剪裁，适合日常穿搭，耐磨耐脏",    "轻薄羽绒服，适合冬季户外活动，保暖性强且便携",    "丝绸长裙，V领设计，适合宴会和正式场合，优雅大方",    "运动速干T恤，吸汗透气，适合健身房和跑步使用",    "羊毛大衣，经典双排扣设计，适合秋冬商务场合，保暖显气质",    "连帽卫衣，连帽带抽绳，适合休闲居家，柔软舒适",    "防水冲锋衣，适合户外徒步，防风防雨，轻便耐用",    "短款马甲，适合叠穿搭配，提升层次感，适合春秋季节",    "高弹力瑜伽裤，适合健身和运动，贴合身体且透气性好"]
# 用户查询user_query = "设计简约的通勤服装，透气舒适"
inputs = product_descriptions + [user_query]
# 分批生成嵌入向量DASHSCOPE_MAX_BATCH_SIZE = 10embeddings = []
for i in range(0, len(inputs), DASHSCOPE_MAX_BATCH_SIZE):    batch = inputs[i:i + DASHSCOPE_MAX_BATCH_SIZE]    resp = dashscope.TextEmbedding.call(        model="text-embedding-v4",        input=batch,        dimension=1024    )    if resp.status_code == HTTPStatus.OK:        for emb in resp.output["embeddings"]:            embeddings.append(emb["embedding"])    else:        print(f"Error: {resp.status_code}, {resp.message}")        # 如果 API 调用失败，提前终止程序        exit(1)
ifnot embeddings:    print("Error: No embeddings generated. Aborting.")    exit(1)
# 保存向量和文本到文件with open("embeddings_output.txt", "w", encoding="utf-8") as f:    f.write(f"# 向量维度: 1024\n")    f.write(f"# 文本数量: {len(inputs)}\n")    f.write("# 格式: <text_index> | <text> | <vector>\n\n")    for idx, (text, vector) in enumerate(zip(inputs, embeddings)):        vector_str = ",".join(map(str, vector))         f.write(f"{idx} | {text} | {vector_str}\n")
# 提取用户查询向量query_vector = np.array([embeddings[-1]])product_vectors = np.array(embeddings[:-1])
# 计算相似度similarities = cosine_similarity(query_vector, product_vectors).flatten()
# 保存相似度结果到文件with open("similarity_results.txt", "w", encoding="utf-8") as f:    f.write(f"# 用户查询: {user_query}\n")    f.write(f"# 相似度计算结果:\n")    for idx, (desc, score) in enumerate(zip(product_descriptions, similarities)):        f.write(f"{idx} | {desc} | {score:.6f}\n")
product_ranking = list(zip(product_descriptions, similarities))product_ranking.sort(key=lambda x: x[1], reverse=True)
print("=== 推荐结果 ===")for desc, score in product_ranking:    print(f"商品描述: {desc} | 相似度: {score:.4f}")

system_prompt = "你是一位专业的商品推荐专家，擅长结合用户需求和商品描述生成精准的推荐理由。请为每个匹配的文本生成一段推荐理由，要求：\n1. 简洁生动，突出商品核心优势（如设计、功能、适用场景）。\n2. 语言自然流畅，符合用户需求场景（如通勤、健身等）。\n3. 每段推荐理由控制在20字以内。"
user_prompt = f"""用户查询：{user_query}
匹配商品描述：1. {matched_products[0]}2. {matched_products[1]}3. {matched_products[2]}
请为每个商品生成推荐理由。"""response = dashscope.Generation.call(    model="qwen-plus",    prompt=f"{system_prompt}\n\n{user_prompt}",    # max_tokens=500)

import requestsimport timeimport numpy as npfrom concurrent.futures import ThreadPoolExecutor, as_completedfrom sklearn.metrics.pairwise import cosine_similarityimport re# ================== 模型服务配置 ==================LLM_CONFIG = {    "url": "请输入您自己服务的 URL",    "token": "请输入您自己服务的 Token",    "model_name": "Qwen3-32B"}EMBEDDING_CONFIG = {    "url": "请输入您自己服务的 URL",    "token": "请输入您自己服务的 Token",    "model_name": "Qwen3-Embedding-4B"}# ================== 测试数据 ==================LONG_TEXT = "这部电影太棒了，演员表现非常出色，剧情紧凑，特效震撼，" * 40  # 重复40次模拟长文本print("打印 LONG_TEXT ===================")print(LONG_TEXT)# 20 条真实电影评论样本MOVIE_REVIEWS = [    "这是一部令人难忘的电影，情感真挚动人。",    "剧本构思巧妙，角色塑造饱满，导演功力深厚。",    "视觉效果惊艳，但剧情略显拖沓。",    "配乐和摄影都很棒，整体节奏把握得当。",    "演员演技在线，剧情有深度，值得回味。",    "故事主线清晰，但结尾有些仓促。",    "这部影片让我笑中带泪，情感共鸣强烈。",    "特效场面宏大，但人物发展不够深入。",    "节奏紧凑，情节引人入胜，推荐一看。",    "剧情平淡无奇，缺乏新意。",    "这是我看过最无聊的电影之一。",    "毫无悬念的情节，看得我昏昏欲睡。",    "虽然小众，但极具艺术价值。",    "幽默感十足，非常适合轻松观看。",    "剧情反转出乎意料，令人惊喜。",    "剪辑混乱，影响观影体验。",    "情感细腻，适合喜欢文艺片的观众。",    "动作设计精彩，打斗场面酣畅淋漓。",    "这部电影让我重新认识了人性的复杂。",    "节奏偏慢，但内涵丰富，值得深思。"]# 保留原始分类测试用例（用于其他测试）TEST_QUERIES = {    "短文本": "剧情紧凑，演员演技在线",    "中等文本": "这部电影的视觉效果震撼，但故事线混乱，角色发展不足",    "长文本": LONG_TEXT}# ================== 核心测试函数 ==================def call_llm(text):    """调用LLM生成响应（模拟语义理解）"""    headers = {        "Authorization": LLM_CONFIG["token"],        "Content-Type": "application/json"    }    payload = {        "model": LLM_CONFIG["model_name"],        "messages": [{"role": "user", "content": text}],        "max_tokens": 50    }    start_time = time.time()    response = requests.post(LLM_CONFIG["url"], headers=headers, json=payload)    latency = time.time() - start_time    try:        tokens = len(response.json()["choices"][0]["message"]["content"].split())    except:        tokens = 0    return {        "latency": latency,        "tokens": tokens,        "vector": np.random.rand(2560)    }def call_embedding(text):    """调用Embedding模型生成向量"""    headers = {        "Authorization": EMBEDDING_CONFIG["token"],        "Content-Type": "application/json"    }    payload = {        "input": [text],        "model": EMBEDDING_CONFIG["model_name"]    }    start_time = time.time()    response = requests.post(EMBEDDING_CONFIG["url"], headers=headers, json=payload)    latency = time.time() - start_time    data = response.json()    vector = data.get("data", [{}])[0].get("embedding", [])    tokens = data.get("usage", {}).get("total_tokens", 0)    return {        "latency": latency,        "tokens": tokens,        "vector": vector    }# ================== 基础对比测试 ==================def run_comparison():    print("开始性能对比测试（20条电影评论，取平均值）")    llm_latencies = []    llm_tokens = []    emb_latencies = []    emb_tokens = []    for i, review in enumerate(MOVIE_REVIEWS):        print(f"\n 第 {i+1} 条影评测试：")        # LLM 测试        llm_res = call_llm(review)        print(f"LLM 响应时间: {llm_res['latency']:.2f}s | Tokens: {llm_res['tokens']}")        llm_latencies.append(llm_res["latency"])        llm_tokens.append(llm_res["tokens"])        # Embedding 测试        emb_res = call_embedding(review)        print(f"Embedding 响应时间: {emb_res['latency']:.2f}s | Tokens: {emb_res['tokens']}")        emb_latencies.append(emb_res["latency"])        emb_tokens.append(emb_res["tokens"])    # 输出平均值    print("\n 测试一：20条影评平均性能对比")    print(f"LLM 平均响应时间: {np.mean(llm_latencies):.2f}s | 平均 Token 数: {np.mean(llm_tokens):.1f}")    print(f"Embedding 平均响应时间: {np.mean(emb_latencies):.2f}s | 平均 Token 数: {np.mean(emb_tokens):.1f}")# ================== 新增测试逻辑 ==================def batch_test():    """批量处理效率对比"""    print("\n 测试二：批量处理测试（同时处理3条文本）")    texts = list(TEST_QUERIES.values())    # LLM 批量测试    start = time.time()    with ThreadPoolExecutor() as executor:        llm_results = list(executor.map(call_llm, texts))    llm_batch_latency = time.time() - start    # Embedding 批量测试    start = time.time()    with ThreadPoolExecutor() as executor:        emb_results = list(executor.map(call_embedding, texts))    emb_batch_latency = time.time() - start    print(f"LLM 批量耗时: {llm_batch_latency:.2f}s | 单条平均: {llm_batch_latency/3:.2f}s")    print(f"Embedding 批量耗时: {emb_batch_latency:.2f}s | 单条平均: {emb_batch_latency/3:.2f}s")def long_text_test():    """长文本性能衰减测试"""    print("\n 测试三：长文本性能测试（输入长度: ~1000字符）")    llm_res = call_llm(LONG_TEXT)    emb_res = call_embedding(LONG_TEXT)    print(f"LLM 响应时间: {llm_res['latency']:.2f}s | Tokens: {llm_res['tokens']}")    print(f"Embedding 响应时间: {emb_res['latency']:.2f}s | Tokens: {emb_res['tokens']}")def concurrent_test(max_workers=5):    """并发压力测试"""    print(f"\n 测试四：并发测试（{max_workers}个并发请求）")    texts = [TEST_QUERIES["中等文本"]] * max_workers    # LLM 并发测试    start = time.time()    with ThreadPoolExecutor() as executor:        futures = [executor.submit(call_llm, text) for text in texts]        llm_latencies = [future.result()["latency"] forfuture in futures]    llm_concurrent = time.time() - start    # Embedding 并发测试    start = time.time()    with ThreadPoolExecutor() as executor:        futures = [executor.submit(call_embedding, text) for text in texts]        emb_latencies = [future.result()["latency"] forfuture in futures]    emb_concurrent = time.time() - start    print(f"LLM 总耗时: {llm_concurrent:.2f}s | 平均延迟: {np.mean(llm_latencies):.2f}s")    print(f"Embedding 总耗时: {emb_concurrent:.2f}s | 平均延迟: {np.mean(emb_latencies):.2f}s")def cost_analysis():    """成本估算功能（使用最新单价）"""    DAILY_CALLS = 10000  # 日均调用量    LLM_PRICE_INPUT = 0.002     # LLM 输入价格（千 token）    LLM_PRICE_OUTPUT_THINKING = 0.02   # LLM 输出价格（思考模式）    LLM_PRICE_OUTPUT_NORMAL = 0.008  # LLM 输出价格（非思考模式）    EMB_PRICE = 0.0005       # Embedding 单价：¥0.0005 / 千 token    # 使用中等文本的平均token消耗    avg_emb_tokens = call_embedding(TEST_QUERIES["中等文本"])["tokens"]    # LLM 默认使用思考模式    avg_llm_input_tokens = len(TEST_QUERIES["中等文本"].split())    avg_llm_output_tokens = call_llm(TEST_QUERIES["中等文本"])["tokens"]    # 成本计算    thinking_cost = (avg_llm_input_tokens * LLM_PRICE_INPUT + avg_llm_output_tokens * LLM_PRICE_OUTPUT_THINKING) * DAILY_CALLS    normal_cost = (avg_llm_input_tokens * LLM_PRICE_INPUT + avg_llm_output_tokens * LLM_PRICE_OUTPUT_NORMAL) * DAILY_CALLS    emb_cost = (avg_emb_tokens * EMB_PRICE) * DAILY_CALLS    print("\n 测试六：LLM 思考模式 vs 非思考模式 成本估算")    print(f"思考模式日成本: ¥{thinking_cost:.2f} | 单次成本: ¥{(thinking_cost/DAILY_CALLS):.5f}")    print(f"非思考模式日成本: ¥{normal_cost:.2f} | 单次成本: ¥{(normal_cost/DAILY_CALLS):.5f}")    print(f"Embedding 日成本: ¥{emb_cost:.2f} | 单次成本: ¥{(emb_cost/DAILY_CALLS):.5f}")    print(f"思考模式成本是 Embedding 的 {(thinking_cost / emb_cost):.1f} 倍")    print(f"非思考模式成本是 Embedding 的 {(normal_cost / emb_cost):.1f} 倍")# ================== 语义匹配测试 ==================def test_semantic_search():    """语义搜索测试"""    query = "这部电影剧情紧凑，演员演技在线"    results = []    print("\n 测试五：语义搜索测试")    emb_query = call_embedding(query)["vector"]    for i, review in enumerate(MOVIE_REVIEWS):        emb_review = call_embedding(review)["vector"]        ifnot emb_query ornot emb_review:            continue        sim_score = cosine_similarity([emb_query], [emb_review])[0][0]        results.append((i+1, sim_score, review[:30] + "..."))    results.sort(key=lambda x: x[1], reverse=True)    print("Top-5 相似影评（基于 Embedding 向量）：")    for idx, score, text in results[:5]:        print(f"第 {idx} 条影评（相似度: {score:.3f}）: {text}")# ================== 主程序入口 ==================if __name__ == "__main__":    # 基础测试    run_comparison()    # 新增测试    batch_test()    long_text_test()    concurrent_test(max_workers=5)    # 语义能力测试    test_semantic_search()    # 成本分析    cost_analysis()