向量化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36{
    "created": 1752133360,
    "data": {
        "embedding": [
            -0.046875,-0.048828125,0.02001953125,0.064453125,
            .....
            0.003143310546875
        ],
        "sparse_embedding":[
            {
                "index": 1,
                "value": 0.0887451171875
            },
            {
                "index": 13,
                "value": 0.0887451171875
            },
            {
                "index": 149,
                "value": 0.0887451171875
            }
        ],
        "object": "embedding"
    },
    "id": "021752133359863906427fb4b36437c414d645f52206dfc398f85",
    "model": "doubao-embedding-vision-251215",
    "object": "list",
    "usage": {
        "prompt_tokens": 25,
        "prompt_tokens_details": {
            "image_tokens": 0,
            "text_tokens": 25
        },
        "total_tokens": 25
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13{
    "model": "doubao-embedding-vision-251215",
    "input": [
        {
            "type":"text",
            "text":"天很蓝"
        }
    ],
    "sparse_embedding": {
        "type":"enabled"
    },
    "encoding_format":"float"
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18{
    "model": "doubao-embedding-vision-251215",
    "instructions":"Target_modality: text and video.\nInstruction:Compress the text/video into one word.\nQuery:",
    "input": [
        {
            "type":"text",
            "text":"天很蓝"
        }
    ],
    "multi_embedding": {
        "type":"enabled",
        "compression":"blosc2"
    },
    "sparse_embedding": {
        "type":"enabled"
    },
    "encoding_format":"float"
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17{
    "model": "doubao-embedding-vision-251215",
    "input": [
        {
            "type": "video_url",
            "video_url": {
                "url": "https://arkdocs.tos-cn-beijing.volces.com/videos/advanced-usage/ark_vlm_video_input.mp4",
                "fps": 1,
                "max_video_tokens": 120000,
                "min_frame_tokens": 32,
                "max_frame_tokens": 640,
                "min_frames": 10
            }
        }
    ],
    "encoding_format": "float"
}

1Target_modality: {}.\nInstruction:{}\nQuery:

1Instruction:Compress the {} into one word.\nQuery:

1Target_modality: {}.\nInstruction:{}\nQuery:

1
2
3
4
5
6
7
8
9
10
11{
    "model": "doubao-embedding-vision-251215",
    "input": [
        {
            "type":"text",
            "text":"天很蓝"
        }
    ],
    "dimensions": 1024,    #设置向量维度
    "encoding_format":"float"
}

1min(图片宽 * 图片高/784, 单图 token 限制)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16...
import math
import os

from arkruntime import Ark
from arkruntime.types.multimodal_embedding import MultiModalEmbeddingResponse

IMAGE_CORPUS_INSTRUCTIONS = (
    "Instruction:Compress the image into one word.\nQuery:"
)
TEXT_QUERY_INSTRUCTIONS = (
    "Target_modality: image.\n"
    "Instruction:根据水果名称检索匹配的图片\n"
    "Query:"
)
...

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35...
def get_embedding(input_data, input_type="text"):
    """调用火山引擎API获取单个文本或图片的向量表示"""
    client = Ark(
        api_key=os.environ.get("ARK_API_KEY"),
        base_url="https://ark.cn-beijing.volces.com/api/v3",
    )
    if input_type == "text":
        input_item = {"type": "text", "text": input_data}
        instructions = TEXT_QUERY_INSTRUCTIONS
    elif input_type == "image_url":
        input_item = {"type": "image_url", "image_url": {"url": input_data}}
        instructions = IMAGE_CORPUS_INSTRUCTIONS
    else:
        raise ValueError("输入类型仅支持'text'或'image_url'")

    try:
        resp = client.post(
            "/embeddings/multimodal",
            cast_to=MultiModalEmbeddingResponse,
            body={
                "model": "doubao-embedding-vision-251215",
                "encoding_format": "float",
                "input": [input_item],
                "instructions": instructions,
            },
        )
        embedding = resp.data.embedding
        if not isinstance(embedding, list):
            raise TypeError("API 未返回 float 格式向量")
        return embedding
    except Exception as e:
        print(f" 获取向量失败，输入类型: {input_type}, 错误: {str(e)}")
        raise
...

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21...
def generate_image_embeddings(image_urls):
    """批量生成图片向量并构建向量库"""
    print(f"[1/3] 开始生成 {len(image_urls)} 张图片的向量...")
    embeddings = []
    for i, url in enumerate(image_urls):
        try:
            embedding = get_embedding(url, "image_url")
            embeddings.append({
                "image_url": url,
                "embedding": embedding
            })
            print(f" [{i+1}/{len(image_urls)}] 成功: {url}")
        except Exception as e:
            print(f" [{i+1}/{len(image_urls)}] 失败: {url} - {str(e)}")
            continue
    if not embeddings:
        raise ValueError("所有图片向量生成失败")
    print(f"[2/3] 完成: {len(embeddings)} 个有效向量")
    return embeddings
...

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31...
def cosine_similarity(left, right):
    """计算两个等长向量的余弦相似度。"""
    if len(left) != len(right):
        raise ValueError("向量维度不一致")
    dot_product = sum(a * b for a, b in zip(left, right))
    left_norm = math.sqrt(sum(value * value for value in left))
    right_norm = math.sqrt(sum(value * value for value in right))
    if left_norm == 0 or right_norm == 0:
        raise ValueError("向量模长不能为 0")
    return dot_product / (left_norm * right_norm)


def search_similar_images(query_embedding, embeddings, top_n=1, query_type="文本"):
    """搜索与查询向量最相似的图片"""
    print(f"\n[3/3] 开始搜索与{query_type}最相似的图片...")
    results = []
    for item in embeddings:
        similarity = cosine_similarity(
            query_embedding,
            item["embedding"],
        )
        results.append({
            "image_url": item["image_url"],
            "similarity": similarity
        })

    results.sort(key=lambda x: x["similarity"], reverse=True)
    print(f" - 相似度计算完成，共 {len(results)} 个结果")
    return results[:top_n]
...

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25...
if __name__ == "__main__":
    image_urls = [
        "https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit1.jpg",
        "https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit2.jpg",
        "https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit3.jpg",
        "https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit4.jpg",
        "https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit5.jpg"
    ]
    query_text = "草莓"

    try:
        # 生成图片向量
        image_embs = generate_image_embeddings(image_urls)
        # 生成文本向量
        text_emb = get_embedding(query_text, "text")
        print(f"文本向量维度: {len(text_emb)}")
        # 搜索相似图片
        similar_images = search_similar_images(text_emb, image_embs)
        print(f"最相似图片: {similar_images[0]['image_url']}")
        print(f"相似度分数: {similar_images[0]['similarity']:.4f}")

    except Exception as e:
        print(f"程序失败: {e}")
...

1
2
3
4
5
6
7
8
9
10
11
12
13[1/3] 开始生成 5 张图片的向量...
 [1/5] 成功: https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit1.jpg
 [2/5] 成功: https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit2.jpg
 [3/5] 成功: https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit3.jpg
 [4/5] 成功: https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit4.jpg
 [5/5] 成功: https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit5.jpg
[2/3] 完成: 5 个有效向量
文本向量维度: 2048

[3/3] 开始搜索与文本最相似的图片...
 - 相似度计算完成，共 5 个结果
最相似图片: https://arkdocs.tos-cn-beijing.volces.com/images/domain-models/Fruit5.jpg
相似度分数: 0.4515

1
2
3
4
5
6
7
8
9# 定义方法将指定路径图片转为Base64编码
def encode_image(image_path):
  with open(image_path, "rb") as image_file:
    return base64.b64encode(image_file.read()).decode('utf-8')

# 需要传给大模型的图片
image_path = "path_to_your_image.jpg"
# 将图片转为Base64编码
base64_image = encode_image(image_path)

1
2
3
4
5
6{
    "type": "image_url",
    "image_url": {
        "url":  f"data:image/<IMAGE_FORMAT>;base64,{base64_image}"
    }
},

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71import asyncio
import os
from pathlib import Path
from arkruntime import AsyncArk

class MultimodalEmbedder:
    """多模态向量化批量处理工具（全组失败模式）"""
    def __init__(self, api_key: str, model: str = "doubao-embedding-vision-241215",
                 batch_size: int = 10, retries: int = 2):
        self.api_key = api_key
        self.model = model
        self.batch_size = batch_size  # 批量大小配置（可扩展分片逻辑）
        self.retries = retries        # 批量请求重试次数

    async def batch_process(self, items_list):
        """核心：异步批量处理向量任务（全组失败）"""
        # 1. 初始化批量客户端
        async with AsyncArk(
            max_retries=self.retries,
            base_url="https://ark.cn-beijing.volces.com/api/v3",
        ) as client:
            # 2. 批量创建异步任务（按输入列表分片）
            batch_tasks = [
                asyncio.create_task(client.multimodal_embeddings.create(multi_embedding={"type": "disabled"}, model=self.model, input=items))
                for items in items_list
            ]

            try:
                # 3. 批量等待任务完成（任一失败则全组终止）
                batch_results = await asyncio.gather(*batch_tasks)
                return batch_results
            except Exception:
                # 4. 批量清理未完成任务
                for task in batch_tasks:
                    if not task.done():
                        task.cancel()
                raise  # 抛出异常，保持全组失败语义

    def batch_save(self, batch_results, output_dir: str = "embedding_results"):
        """核心：批量保存向量结果"""
        # 1. 初始化输出目录
        Path(output_dir).mkdir(exist_ok=True)
        # 2. 批量写入结果文件
        with open(f"{output_dir}/batch_embeddings.txt", "w", encoding="utf-8") as f:
            for idx, result in enumerate(batch_results, 1):
                embedding = result.data.embedding
                f.write(f"批量项 #{idx} | 向量长度: {len(embedding)} | 前20维: {embedding[:20]}...\n")
        print(f"批量结果已保存：{output_dir}/batch_embeddings.txt")

if __name__ == "__main__":
    # 1. 配置初始化
    api_key = os.environ.get("ARK_API_KEY") or input("输入API密钥: ").strip()
    if not api_key:
        raise ValueError("API密钥不能为空")

    # 2. 初始化批量处理器
    embedder = MultimodalEmbedder(api_key=api_key, batch_size=10, retries=2)

    # 3. 构造批量输入数据
    batch_inputs = [
        [{"type": "text", "text": "天很蓝，海很深"}, {"type": "image_url", "image_url": {"url": "https://example.com/image1.jpg"}}],
        [{"type": "text", "text": "阳光明媚的沙滩"}, {"type": "image_url", "image_url": {"url": "https://example.com/image2.jpg"}}]
    ]

    # 4. 执行批量处理+结果保存
    try:
        batch_results = asyncio.run(embedder.batch_process(batch_inputs))
        embedder.batch_save(batch_results)
        print(f"批量处理完成，共生成 {len(batch_results)} 个向量")
    except Exception as e:
        print(f"批量处理失败: {str(e)}")