文本嵌入

概念概述

文本嵌入（Text Embeddings）指的是：文本嵌入把文本映射为编码语义的稠密向量，让机器能在向量空间中按邻近度比较、聚类与检索语言。

从更严格的定义看，A text embedding is a fixed-length numeric vector produced by a model from an input text, positioned so that semantically similar texts lie near each other under a distance metric. 该概念在工程实践中的价值，在于它能帮助团队用统一的语言描述一个具体的质量属性或设计约束，从而在评审与协作中减少歧义。

为什么需要它

It is the representation layer under semantic search, clustering, recommendation, and retrieval-augmented generation. 它与 Tokenization and Subword Units、The Context Window、Prompt Injection 等概念经常一起出现，用于解决相近的工程问题。理解它的适用场景，有助于在多种备选方案之间做出有依据的取舍，而不是凭直觉选择技术方案。

常见误解

• embedding dimensions are interchangeable between models
• similarity scores are calibrated probabilities

需要避免的反模式

• comparing embeddings produced by different models
• assuming a similarity threshold generalises across datasets

实践要点

1. An embedding model fixes the vector space, so vectors from two different models are not comparable and a corpus must be re-embedded whenever the model changes.
2. Cosine similarity is commonly used on normalised vectors, and the resulting score is a relative ranking signal rather than a calibrated probability.
3. Domain and language coverage vary, so effectiveness depends on choosing a model trained for the content’s language and vocabulary.

参考来源

https://en.wikipedia.org/wiki/Word_embedding

代码示例

from __future__ import annotations

import math
import re
from collections import Counter

TOKEN_RE = re.compile(r"[a-z][a-z0-9+#.-]{1,}")
STOPWORDS = {
    "the", "and", "for", "with", "that", "this",
    "from", "into", "your", "are", "was", "have",
}


def tokenize(text: str) -> list[str]:
    return [
        token
        for token in TOKEN_RE.findall(text.lower())
        if token not in STOPWORDS and len(token) > 2
    ]


def keyword_scores(documents: list[str], top_n: int = 5) -> list[tuple[str, float]]:
    tokenized = [tokenize(document) for document in documents]
    document_frequency: Counter[str] = Counter()
    for tokens in tokenized:
        document_frequency.update(set(tokens))
    total_documents = len(tokenized)
    scores: Counter[str] = Counter()
    for tokens in tokenized:
        counts = Counter(tokens)
        length = sum(counts.values()) or 1
        for term, count in counts.items():
            tf = count / length
            idf = math.log((1 + total_documents) / (1 + document_frequency[term])) + 1
            scores[term] += tf * idf
    return sorted(scores.items(), key=lambda item: (-item[1], item[0]))[:top_n]


def main() -> None:
    documents = [
        "Use connection pooling to reduce database latency and reuse open sockets.",
        "A connection pool keeps sockets warm so database queries avoid handshake latency.",
        "Caching responses can hide database latency but risks stale data.",
    ]
    for term, score in keyword_scores(documents, 4):
        print(f"{term:12s} {score:.4f}")
    assert keyword_scores(documents, 1)[0][0] == "database"


if __name__ == "__main__":
    main()

ABBREVIATIONS = frozenset(
    {
        "mr",
        "mrs",
        "ms",
        "dr",
        "prof",
        "sr",
        "jr",
        "vs",
        "etc",
        "e.g",
        "i.e",
        "fig",
        "no",
        "inc",
        "ltd",
    }
)
CLOSERS = "\"')]}"


def _is_abbreviation(text: str, dot_index: int) -> bool:
    window = text[max(0, dot_index - 16) : dot_index]
    word = window.rsplit(" ", 1)[-1].lower().lstrip("(\"'")
    return word in ABBREVIATIONS or (len(word) == 1 and word.isalpha())


def split_sentences(text: str) -> list[str]:
    normalized = " ".join(text.split())
    if not normalized:
        return []
    sentences: list[str] = []
    start = 0
    index = 0
    while index < len(normalized):
        char = normalized[index]
        if char in ".!?":
            cursor = index + 1
            if cursor < len(normalized) and normalized[cursor] in CLOSERS:
                cursor += 1
            if cursor < len(normalized) and normalized[cursor] == " ":
                if char == "." and _is_abbreviation(normalized, index):
                    index += 1
                    continue
                sentences.append(normalized[start:cursor].strip())
                start = cursor + 1
                index = cursor
        index += 1
    tail = normalized[start:].strip()
    if tail:
        sentences.append(tail)
    return [sentence for sentence in sentences if sentence]


def main() -> None:
    paragraph = (
        "Dr. Smith joined Acme Inc. in 2019. Revenue grew 12.5% that year! "
        "Was it sustainable? The board said \"yes.\" Next year told another story."
    )
    for number, sentence in enumerate(split_sentences(paragraph), start=1):
        print(f"{number}. {sentence}")


if __name__ == "__main__":
    main()