LLM应用开发——Embedding嵌入向量

# 假设的 3 维简化 Embedding 空间
"猫" = [0.9, 0.1, 0.2]  # 维度分别代表：[生物性, 宠物属性, 体型大小]
"狗" = [0.8, 0.2, 0.4]
"桌子" = [0.0, 0.0, 0.7]

from openai import OpenAI

client = OpenAI()

def get_embedding(text: str, model: str = "text-embedding-3-small") -> list:
    """获取文本的Embedding向量"""
    response = client.embeddings.create(
        input=text,
        model=model
    )
    return response.data[0].embedding

# 示例
text = "机器学习是人工智能的一个分支"
embedding = get_embedding(text)

print(f"文本: {text}")
print(f"向量维度: {len(embedding)}")
print(f"前5个值: {embedding[:5]}")

from sentence_transformers import SentenceTransformer
import numpy as np

# 加载模型
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 编码文本
texts = [
    "机器学习是人工智能的一个分支",
    "深度学习是机器学习的子集",
    "今天天气很好"
]

embeddings = model.encode(texts)

print(f"向量形状: {embeddings.shape}")  # (3, 1024)

# 计算相似度
from sklearn.metrics.pairwise import cosine_similarity

similarity_matrix = cosine_similarity(embeddings)
print("\n相似度矩阵:")
print(similarity_matrix)

from fastapi import FastAPI
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer
import uvicorn

app = FastAPI()
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

class EmbeddingRequest(BaseModel):
    texts: list[str]

class EmbeddingResponse(BaseModel):
    embeddings: list[list[float]]
    dimensions: int

@app.post("/embeddings", response_model=EmbeddingResponse)
async def create_embeddings(request: EmbeddingRequest):
    embeddings = model.encode(request.texts).tolist()
    return EmbeddingResponse(
        embeddings=embeddings,
        dimensions=len(embeddings[0])
    )

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import numpy as np

def cosine_similarity(a, b):
    """余弦相似度（最常用）"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def euclidean_distance(a, b):
    """欧氏距离"""
    return np.linalg.norm(a - b)

def dot_product(a, b):
    """点积（需要归一化向量）"""
    return np.dot(a, b)

# 示例
vec1 = np.array([0.1, 0.2, 0.3, 0.4])
vec2 = np.array([0.15, 0.25, 0.28, 0.38])
vec3 = np.array([-0.3, -0.2, 0.1, -0.4])

print(f"vec1 vs vec2 余弦相似度: {cosine_similarity(vec1, vec2):.4f}")
print(f"vec1 vs vec3 余弦相似度: {cosine_similarity(vec1, vec3):.4f}")

from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

# 假设有1000个文档向量
doc_embeddings = np.random.randn(1000, 768)

# 查询向量
query_embedding = np.random.randn(1, 768)

# 计算查询与所有文档的相似度
similarities = cosine_similarity(query_embedding, doc_embeddings)[0]

# 获取Top-K最相似的文档
k = 5
top_k_indices = np.argsort(similarities)[-k:][::-1]
top_k_scores = similarities[top_k_indices]

print(f"Top {k} 最相似文档:")
for idx, score in zip(top_k_indices, top_k_scores):
    print(f"  文档{idx}: 相似度 {score:.4f}")

from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Tuple

class SemanticSearch:
    def __init__(self, model_name: str = "BAAI/bge-large-zh-v1.5"):
        self.model = SentenceTransformer(model_name)
        self.documents = []
        self.embeddings = None
    
    def add_documents(self, documents: List[str]):
        """添加文档到索引"""
        self.documents.extend(documents)
        new_embeddings = self.model.encode(documents)
        
        if self.embeddings is None:
            self.embeddings = new_embeddings
        else:
            self.embeddings = np.vstack([self.embeddings, new_embeddings])
        
        print(f"已索引 {len(self.documents)} 个文档")
    
    def search(self, query: str, top_k: int = 5) -> List[Tuple[str, float]]:
        """语义搜索"""
        query_embedding = self.model.encode([query])
        
        # 计算相似度
        similarities = np.dot(self.embeddings, query_embedding.T).flatten()
        
        # 归一化（如果向量未归一化）
        query_norm = np.linalg.norm(query_embedding)
        doc_norms = np.linalg.norm(self.embeddings, axis=1)
        similarities = similarities / (query_norm * doc_norms)
        
        # 获取Top-K
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        
        results = []
        for idx in top_indices:
            results.append((self.documents[idx], float(similarities[idx])))
        
        return results

# 使用示例
search_engine = SemanticSearch()

# 添加文档
documents = [
    "Python是一种流行的编程语言，适合数据科学和机器学习",
    "TensorFlow和PyTorch是两个主流的深度学习框架",
    "向量数据库用于存储和检索高维向量",
    "RAG通过检索增强来提升大模型的回答质量",
    "今天的天气非常晴朗，适合户外活动",
    "机器学习模型需要大量数据进行训练",
]
search_engine.add_documents(documents)

# 搜索
query = "如何选择深度学习工具"
results = search_engine.search(query, top_k=3)

print(f"\n查询: {query}")
print("搜索结果:")
for doc, score in results:
    print(f"  [{score:.4f}] {doc}")

import re

def preprocess_text(text: str) -> str:
    """文本预处理"""
    # 去除多余空白
    text = re.sub(r'\s+', ' ', text).strip()
    # 去除特殊字符（可选）
    # text = re.sub(r'[^\w\s]', '', text)
    # 截断过长文本
    max_length = 512
    if len(text) > max_length:
        text = text[:max_length]
    return text

def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list:
    """将长文本分割成重叠的块"""
    chunks = []
    start = 0
    
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start = end - overlap
    
    return chunks

# 示例
long_text = "这是一段很长的文本..." * 100
chunks = chunk_text(long_text, chunk_size=200, overlap=20)
print(f"分割成 {len(chunks)} 个块")

def enhance_query(query: str, instruction: str = None) -> str:
    """增强查询（某些模型需要指令前缀）"""
    # BGE模型推荐的查询前缀
    if instruction:
        return f"{instruction}: {query}"
    return f"为这个句子生成表示以用于检索相关文章: {query}"