嵌入接口 (POST /api/embeddings)

嵌入接口将文本转换为向量表示，这是构建语义搜索、RAG 应用、文本相似度计算的基础。

什么是嵌入？

嵌入是把文本转换成一串数字（向量），这个向量能表示文本的语义含义。

比如：

"猫是一种宠物" → [0.1, 0.3, -0.2, 0.5, ...]
"狗是一种宠物" → [0.1, 0.4, -0.1, 0.6, ...]
"今天天气很好" → [0.8, -0.2, 0.3, 0.1, ...]

语义相近的文本，向量也相近。这在很多场景下很有用。

基本用法

curl http://localhost:11434/api/embeddings -d '{
  "model": "llama3.2",
  "prompt": "这是一段需要向量化的文本"
}'

响应：

{
  "embedding": [0.1, 0.2, -0.3, 0.4, ...]
}

embedding 是一个浮点数数组，长度取决于模型，通常是几千维。

请求参数

选择嵌入模型

虽然可以用任何模型生成嵌入，但专用嵌入模型效果更好：

# 拉取专用嵌入模型
ollama pull nomic-embed-text
ollama pull mxbai-embed-large

使用嵌入模型：

curl http://localhost:11434/api/embeddings -d '{
  "model": "nomic-embed-text",
  "prompt": "这是一段文本"
}'

常用嵌入模型

代码示例

Python

import requests

def get_embedding(text, model="nomic-embed-text"):
    response = requests.post(
        "http://localhost:11434/api/embeddings",
        json={
            "model": model,
            "prompt": text
        }
    )
    return response.json()["embedding"]

embedding = get_embedding("这是一段测试文本")
print(f"向量维度: {len(embedding)}")
print(f"前5个值: {embedding[:5]}")

批量处理

def get_embeddings(texts, model="nomic-embed-text"):
    embeddings = []
    for text in texts:
        response = requests.post(
            "http://localhost:11434/api/embeddings",
            json={"model": model, "prompt": text}
        )
        embeddings.append(response.json()["embedding"])
    return embeddings

texts = ["文本1", "文本2", "文本3"]
embeddings = get_embeddings(texts)

JavaScript

async function getEmbedding(text, model = 'nomic-embed-text') {
    const response = await fetch('http://localhost:11434/api/embeddings', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ model, prompt: text })
    });
    
    const data = await response.json();
    return data.embedding;
}

const embedding = await getEmbedding('这是一段测试文本');
console.log(`向量维度: ${embedding.length}`);

Go

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
)

type EmbeddingRequest struct {
    Model  string `json:"model"`
    Prompt string `json:"prompt"`
}

type EmbeddingResponse struct {
    Embedding []float64 `json:"embedding"`
}

func getEmbedding(text string) ([]float64, error) {
    req := EmbeddingRequest{
        Model:  "nomic-embed-text",
        Prompt: text,
    }
    
    body, _ := json.Marshal(req)
    resp, err := http.Post(
        "http://localhost:11434/api/embeddings",
        "application/json",
        bytes.NewReader(body),
    )
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    
    data, _ := io.ReadAll(resp.Body)
    var result EmbeddingResponse
    json.Unmarshal(data, &result)
    
    return result.Embedding, nil
}

func main() {
    embedding, _ := getEmbedding("这是一段测试文本")
    fmt.Printf("向量维度: %d\n", len(embedding))
}

应用场景

文本相似度

计算两段文本的语义相似度：

import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

text1 = "我喜欢吃苹果"
text2 = "我爱吃水果"
text3 = "今天天气很好"

emb1 = get_embedding(text1)
emb2 = get_embedding(text2)
emb3 = get_embedding(text3)

print(f"文本1和文本2相似度: {cosine_similarity(emb1, emb2):.4f}")
print(f"文本1和文本3相似度: {cosine_similarity(emb1, emb3):.4f}")

输出：

文本1和文本2相似度: 0.8523
文本1和文本3相似度: 0.2134

语义搜索

import numpy as np

class SemanticSearch:
    def __init__(self, model="nomic-embed-text"):
        self.model = model
        self.documents = []
        self.embeddings = []
    
    def add_document(self, text):
        self.documents.append(text)
        emb = get_embedding(text, self.model)
        self.embeddings.append(emb)
    
    def search(self, query, top_k=3):
        query_emb = get_embedding(query, self.model)
        
        similarities = []
        for i, doc_emb in enumerate(self.embeddings):
            sim = np.dot(query_emb, doc_emb) / (
                np.linalg.norm(query_emb) * np.linalg.norm(doc_emb)
            )
            similarities.append((i, sim))
        
        similarities.sort(key=lambda x: x[1], reverse=True)
        
        results = []
        for i, sim in similarities[:top_k]:
            results.append({
                "document": self.documents[i],
                "similarity": sim
            })
        
        return results

# 使用
search_engine = SemanticSearch()
search_engine.add_document("Python 是一种编程语言")
search_engine.add_document("JavaScript 用于网页开发")
search_engine.add_document("机器学习是人工智能的一个分支")

results = search_engine.search("编程语言有哪些")
for r in results:
    print(f"相似度: {r['similarity']:.4f}")
    print(f"内容: {r['document']}")
    print()

文本聚类

from sklearn.cluster import KMeans
import numpy as np

texts = [
    "Python 编程入门",
    "JavaScript 前端开发",
    "机器学习基础",
    "深度学习实战",
    "Java 后端开发",
    "数据科学导论"
]

embeddings = [get_embedding(t) for t in texts]
X = np.array(embeddings)

kmeans = KMeans(n_clusters=3, random_state=42)
labels = kmeans.fit_predict(X)

for i, text in enumerate(texts):
    print(f"类别 {labels[i]}: {text}")

RAG 应用

def rag_query(query, documents, model="llama3.2"):
    # 1. 获取查询的嵌入
    query_emb = get_embedding(query, "nomic-embed-text")
    
    # 2. 找到最相关的文档
    doc_embeddings = [get_embedding(d, "nomic-embed-text") for d in documents]
    
    similarities = []
    for i, doc_emb in enumerate(doc_embeddings):
        sim = np.dot(query_emb, doc_emb) / (
            np.linalg.norm(query_emb) * np.linalg.norm(doc_emb)
        )
        similarities.append((i, sim))
    
    similarities.sort(key=lambda x: x[1], reverse=True)
    top_doc = documents[similarities[0][0]]
    
    # 3. 用相关文档生成回答
    response = requests.post(
        "http://localhost:11434/api/chat",
        json={
            "model": model,
            "messages": [
                {"role": "system", "content": "根据提供的上下文回答问题"},
                {"role": "user", "content": f"上下文：{top_doc}\n\n问题：{query}"}
            ],
            "stream": False
        }
    )
    
    return response.json()["message"]["content"]

# 使用
documents = [
    "Ollama 是一个本地运行大语言模型的工具",
    "它支持多种模型，包括 Llama、Mistral 等",
    "可以通过 API 或命令行使用"
]

answer = rag_query("Ollama 是什么？", documents)
print(answer)

性能优化

缓存嵌入

import hashlib

class CachedEmbeddings:
    def __init__(self, model="nomic-embed-text"):
        self.model = model
        self.cache = {}
    
    def get(self, text):
        key = hashlib.md5(text.encode()).hexdigest()
        
        if key not in self.cache:
            self.cache[key] = get_embedding(text, self.model)
        
        return self.cache[key]

embeddings = CachedEmbeddings()
emb1 = embeddings.get("测试文本")
emb2 = embeddings.get("测试文本")

批量请求

import concurrent.futures

def get_embeddings_batch(texts, model="nomic-embed-text"):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = [
            executor.submit(get_embedding, text, model)
            for text in texts
        ]
        return [f.result() for f in concurrent.futures.as_completed(futures)]

注意事项

1. 选择专用嵌入模型：效果比通用模型好
2. 文本预处理：去除多余空白、统一格式
3. 长度限制：超长文本可能被截断
4. 维度一致性：同一模型生成的嵌入维度相同