Embeddings 文本向量化完整指南

import openai
import numpy as np

# 配置 API
openai.api_base = "https://api.n1n.ai/v1"
openai.api_key = "your-api-key"

# 1. 基础文本向量化
def get_embedding(text, model="text-embedding-3-small"):
    """获取文本的向量表示"""
    response = openai.Embedding.create(
        model=model,
        input=text
    )
    return response['data'][0]['embedding']

# 单个文本向量化
text = "机器学习是人工智能的一个分支"
embedding = get_embedding(text)
print(f"向量维度: {len(embedding)}")  # 1536维

# 批量文本向量化
texts = [
    "深度学习是机器学习的子集",
    "神经网络模拟人脑结构",
    "自然语言处理让机器理解人类语言"
]

response = openai.Embedding.create(
    model="text-embedding-3-small",
    input=texts
)

embeddings = [item['embedding'] for item in response['data']]
print(f"处理了 {len(embeddings)} 个文本")

import numpy as np
from scipy.spatial.distance import cosine
import openai

openai.api_base = "https://api.n1n.ai/v1"
openai.api_key = "your-api-key"

def cosine_similarity(vec1, vec2):
    """计算余弦相似度"""
    return 1 - cosine(vec1, vec2)

def get_embeddings(texts):
    """批量获取文本向量"""
    response = openai.Embedding.create(
        model="text-embedding-3-small",
        input=texts
    )
    return [item['embedding'] for item in response['data']]

# 语义相似度计算示例
documents = [
    "Python是一种流行的编程语言",
    "JavaScript用于前端开发",
    "机器学习需要大量数据",
    "深度学习是AI的重要技术",
    "猫是可爱的宠物"
]

query = "人工智能和深度学习的关系"

# 获取所有向量
doc_embeddings = get_embeddings(documents)
query_embedding = get_embeddings([query])[0]

# 计算相似度
similarities = []
for i, doc_emb in enumerate(doc_embeddings):
    sim = cosine_similarity(query_embedding, doc_emb)
    similarities.append((documents[i], sim))

# 排序并显示结果
similarities.sort(key=lambda x: x[1], reverse=True)

print(f"查询: {query}\n")
print("相似度排名:")
for doc, sim in similarities:
    print(f"  {sim:.4f} - {doc}")

# 输出示例:
# 相似度排名:
#   0.8234 - 深度学习是AI的重要技术
#   0.7891 - 机器学习需要大量数据
#   0.4523 - Python是一种流行的编程语言
#   0.3912 - JavaScript用于前端开发
#   0.1234 - 猫是可爱的宠物

from chromadb import Client
from chromadb.config import Settings
import openai

# 初始化 Chroma 向量数据库
client = Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./chroma_db"
))

# 创建或获取集合
collection = client.get_or_create_collection(
    name="documents",
    metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
)

# OpenAI 配置
openai.api_base = "https://api.n1n.ai/v1"
openai.api_key = "your-api-key"

def get_embedding(text):
    response = openai.Embedding.create(
        model="text-embedding-3-small",
        input=text
    )
    return response['data'][0]['embedding']

# 1. 添加文档到向量库
documents = [
    {"id": "doc1", "text": "Vue.js 是渐进式JavaScript框架", "metadata": {"category": "frontend"}},
    {"id": "doc2", "text": "React 使用虚拟DOM提高性能", "metadata": {"category": "frontend"}},
    {"id": "doc3", "text": "Django 是Python的Web框架", "metadata": {"category": "backend"}},
    {"id": "doc4", "text": "FastAPI 提供自动文档生成", "metadata": {"category": "backend"}},
    {"id": "doc5", "text": "PostgreSQL 是关系型数据库", "metadata": {"category": "database"}}
]

# 批量添加
for doc in documents:
    embedding = get_embedding(doc["text"])
    collection.add(
        embeddings=[embedding],
        documents=[doc["text"]],
        metadatas=[doc["metadata"]],
        ids=[doc["id"]]
    )

print(f"已添加 {len(documents)} 个文档到向量库")

# 2. 语义搜索
query = "前端框架性能优化"
query_embedding = get_embedding(query)

# 搜索最相似的文档
results = collection.query(
    query_embeddings=[query_embedding],
    n_results=3,
    where={"category": "frontend"}  # 可选：元数据过滤
)

print(f"\n查询: {query}")
print("搜索结果:")
for i, (doc, dist) in enumerate(zip(results['documents'][0], results['distances'][0])):
    print(f"  {i+1}. 相似度: {1-dist:.4f}")
    print(f"     文档: {doc}")

# 3. 更新和删除
# 更新文档
collection.update(
    ids=["doc1"],
    embeddings=[get_embedding("Vue 3 提供组合式API")],
    documents=["Vue 3 提供组合式API"]
)

# 删除文档
collection.delete(ids=["doc5"])

# 4. 持久化
client.persist()

import openai
from typing import List, Dict
import chromadb
from chromadb.utils import embedding_functions

class RAGSystem:
    def __init__(self, api_key: str, collection_name: str = "knowledge_base"):
        # 配置 OpenAI
        openai.api_key = api_key
        openai.api_base = "https://api.n1n.ai/v1"
        
        # 初始化向量数据库
        self.client = chromadb.PersistentClient(path="./rag_db")
        
        # 使用 OpenAI 嵌入函数
        self.embedding_function = embedding_functions.OpenAIEmbeddingFunction(
            api_key=api_key,
            api_base="https://api.n1n.ai/v1",
            model_name="text-embedding-3-small"
        )
        
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_function
        )
    
    def add_documents(self, documents: List[Dict[str, str]]):
        """添加文档到知识库"""
        self.collection.add(
            documents=[doc["content"] for doc in documents],
            metadatas=[{"source": doc.get("source", "unknown")} for doc in documents],
            ids=[doc["id"] for doc in documents]
        )
        return f"已添加 {len(documents)} 个文档"
    
    def search(self, query: str, n_results: int = 3) -> List[Dict]:
        """语义搜索相关文档"""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )
        
        return [{
            "content": doc,
            "distance": dist,
            "metadata": meta
        } for doc, dist, meta in zip(
            results['documents'][0],
            results['distances'][0],
            results['metadatas'][0]
        )]
    
    def generate_answer(self, query: str, context_docs: List[str]) -> str:
        """基于检索的文档生成答案"""
        context = "\n\n".join(context_docs)
        
        prompt = f"""基于以下文档回答问题。如果文档中没有相关信息，请说"我无法根据提供的文档回答这个问题"。

文档内容:
{context}

问题: {query}

答案:"""
        
        response = openai.ChatCompletion.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "你是一个基于文档的问答助手"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3
        )
        
        return response.choices[0].message.content
    
    def ask(self, question: str) -> Dict:
        """完整的 RAG 流程"""
        # 1. 检索相关文档
        relevant_docs = self.search(question, n_results=3)
        
        # 2. 生成答案
        doc_contents = [doc["content"] for doc in relevant_docs]
        answer = self.generate_answer(question, doc_contents)
        
        return {
            "question": question,
            "answer": answer,
            "sources": relevant_docs
        }

# 使用示例
rag = RAGSystem(api_key="your-api-key")

# 添加知识文档
documents = [
    {"id": "1", "content": "Python 3.12 引入了更好的错误信息", "source": "python_docs"},
    {"id": "2", "content": "TypeScript 5.0 支持装饰器", "source": "ts_docs"},
    {"id": "3", "content": "React 18 引入了并发特性", "source": "react_docs"}
]

rag.add_documents(documents)

# 提问
result = rag.ask("Python 最新版本有什么新特性？")
print(f"问题: {result['question']}")
print(f"答案: {result['answer']}")
print(f"来源: {[s['metadata']['source'] for s in result['sources']]}")

import numpy as np
from sklearn.cluster import KMeans
from sklearn.decomposition import PCA
import matplotlib.pyplot as plt
import openai

openai.api_base = "https://api.n1n.ai/v1"
openai.api_key = "your-api-key"

def get_embeddings(texts):
    response = openai.Embedding.create(
        model="text-embedding-3-small",
        input=texts
    )
    return np.array([item['embedding'] for item in response['data']])

# 文本数据
texts = [
    # 技术类
    "Python 是一种编程语言",
    "JavaScript 用于网页开发",
    "机器学习算法很强大",
    "深度学习模型需要训练",
    
    # 动物类
    "猫是独立的宠物",
    "狗是忠诚的朋友",
    "鸟类可以飞翔",
    "鱼在水中游泳",
    
    # 食物类
    "披萨是意大利美食",
    "寿司来自日本",
    "汉堡是快餐",
    "沙拉很健康"
]

# 获取向量
embeddings = get_embeddings(texts)

# K-Means 聚类
n_clusters = 3
kmeans = KMeans(n_clusters=n_clusters, random_state=42)
clusters = kmeans.fit_predict(embeddings)

# 打印聚类结果
for i in range(n_clusters):
    print(f"\n聚类 {i+1}:")
    cluster_texts = [texts[j] for j, c in enumerate(clusters) if c == i]
    for text in cluster_texts:
        print(f"  - {text}")

# 降维可视化
pca = PCA(n_components=2)
embeddings_2d = pca.fit_transform(embeddings)

# 绘制聚类图
plt.figure(figsize=(10, 8))
colors = ['red', 'blue', 'green']

for i in range(n_clusters):
    cluster_points = embeddings_2d[clusters == i]
    plt.scatter(cluster_points[:, 0], cluster_points[:, 1], 
               c=colors[i], label=f'聚类 {i+1}', alpha=0.6)

# 添加文本标签
for i, txt in enumerate(texts):
    plt.annotate(txt, (embeddings_2d[i, 0], embeddings_2d[i, 1]),
                fontsize=8, alpha=0.7)

plt.xlabel('第一主成分')
plt.ylabel('第二主成分')
plt.title('文本向量聚类可视化')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

import openai
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor
import numpy as np

openai.api_base = "https://api.n1n.ai/v1"
openai.api_key = "your-api-key"

# 1. 批量处理优化
def batch_embeddings(texts, batch_size=100):
    """分批处理大量文本"""
    all_embeddings = []
    
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        
        response = openai.Embedding.create(
            model="text-embedding-3-small",
            input=batch
        )
        
        batch_embeddings = [item['embedding'] for item in response['data']]
        all_embeddings.extend(batch_embeddings)
        
        # 避免速率限制
        if i + batch_size < len(texts):
            time.sleep(0.1)
    
    return all_embeddings

# 2. 并发处理
async def async_get_embedding(text, semaphore):
    """异步获取单个文本的向量"""
    async with semaphore:
        response = await openai.Embedding.acreate(
            model="text-embedding-3-small",
            input=text
        )
        return response['data'][0]['embedding']

async def concurrent_embeddings(texts, max_concurrent=5):
    """并发处理多个文本"""
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [async_get_embedding(text, semaphore) for text in texts]
    return await asyncio.gather(*tasks)

# 3. 缓存策略
class EmbeddingCache:
    def __init__(self, cache_size=1000):
        self.cache = {}
        self.cache_size = cache_size
    
    def get_embedding(self, text, model="text-embedding-3-small"):
        # 检查缓存
        cache_key = f"{model}:{text}"
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        # 获取新的向量
        response = openai.Embedding.create(
            model=model,
            input=text
        )
        embedding = response['data'][0]['embedding']
        
        # 更新缓存（简单的LRU实现）
        if len(self.cache) >= self.cache_size:
            # 删除最旧的项
            oldest_key = list(self.cache.keys())[0]
            del self.cache[oldest_key]
        
        self.cache[cache_key] = embedding
        return embedding

# 4. 向量压缩（降维）
def compress_embeddings(embeddings, target_dim=256):
    """使用 PCA 压缩向量维度"""
    from sklearn.decomposition import PCA
    
    pca = PCA(n_components=target_dim)
    compressed = pca.fit_transform(embeddings)
    
    print(f"压缩比: {embeddings.shape[1]} -> {compressed.shape[1]}")
    print(f"保留方差: {sum(pca.explained_variance_ratio_):.2%}")
    
    return compressed, pca

# 5. 模型选择策略
def smart_embedding(text, use_large=False):
    """根据文本长度选择模型"""
    text_length = len(text)
    
    if text_length > 8000 or use_large:
        # 长文本或需要高精度时使用大模型
        model = "text-embedding-3-large"
    else:
        # 短文本使用小模型，更快更便宜
        model = "text-embedding-3-small"
    
    response = openai.Embedding.create(
        model=model,
        input=text
    )
    
    return {
        "embedding": response['data'][0]['embedding'],
        "model": model,
        "tokens": response['usage']['total_tokens']
    }

# 使用示例
cache = EmbeddingCache()
texts = ["示例文本1", "示例文本2", "示例文本1"]  # 注意重复

# 带缓存的处理
for text in texts:
    embedding = cache.get_embedding(text)
    print(f"处理: {text[:20]}...")