# pinecone_tracer_embedding.py
import openai
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# 示踪剂1: 特殊标记文本
TRACER_TEXT_A = "PINECONE_TRACER_苹果手机死机"
TRACER_TEXT_B = "PINECONE_TRACER_iPhone屏幕无响应"
TRACER_TEXT_C = "PINECONE_TRACER_香蕉很好吃"  # 对照组

def generate_embedding(text):
    """生成文本的Embedding向量"""
    response = openai.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

def print_vector_info(vec, label):
    """打印向量信息"""
    arr = np.array(vec)
    print(f"\n[{label}]")
    print(f"维度: {len(vec)}")
    print(f"前10个值: {np.round(arr[:10], 4)}")
    print(f"最小值: {arr.min():.4f}, 最大值: {arr.max():.4f}")
    print(f"均值: {arr.mean():.4f}, 标准差: {arr.std():.4f}")
    print(f"L2范数: {np.linalg.norm(arr):.4f}")

if __name__ == "__main__":
    print("=" * 60)
    print("🔍 Pinecone Embedding 示踪实验")
    print("=" * 60)

    # 生成三个示踪剂文本的向量
    vec_a = generate_embedding(TRACER_TEXT_A)
    vec_b = generate_embedding(TRACER_TEXT_B)
    vec_c = generate_embedding(TRACER_TEXT_C)

    # 打印向量信息
    print_vector_info(vec_a, "Text A: 苹果手机死机")
    print_vector_info(vec_b, "Text B: iPhone屏幕无响应")
    print_vector_info(vec_c, "Text C: 香蕉很好吃")

    # 计算相似度
    sim_ab = cosine_similarity([vec_a], [vec_b])[0][0]
    sim_ac = cosine_similarity([vec_a], [vec_c])[0][0]

    print("\n" + "=" * 60)
    print("📊 相似度分析")
    print("=" * 60)
    print(f"Text A vs Text B (语义相关): {sim_ab:.4f}")
    print(f"Text A vs Text C (语义不相关): {sim_ac:.4f}")

    # 解读
    print("\n" + "=" * 60)
    print("💡 认知映射")
    print("=" * 60)
    if sim_ab > 0.8:
        print("✅ Text A和B的向量很接近，说明Embedding捕捉到了语义相似性")
    if sim_ac < 0.5:
        print("✅ Text A和C的向量距离远，说明不同语义被正确区分")

# 安装依赖
pip install openai numpy scikit-learn

# 设置API Key
export OPENAI_API_KEY="your-key"

# 运行实验
python pinecone_tracer_embedding.py

# pinecone_tracer_hnsw.py
import pinecone
import numpy as np
from openai import OpenAI
import time

# 初始化
pinecone.init(api_key="your-key", environment="us-east-1-aws")
index_name = "pinecone-tracer"

# 创建Index（如果不存在）
if index_name not in pinecone.list_indexes():
    pinecone.create_index(
        name=index_name,
        dimension=1536,
        metric="cosine",
        spec=pinecone.ServerlessSpec(cloud="aws", region="us-east-1")
    )

index = pinecone.Index(index_name)
client = OpenAI()

def generate_embedding(text):
    """生成向量"""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

if __name__ == "__main__":
    print("=" * 60)
    print("🔍 Pinecone HNSW 示踪实验")
    print("=" * 60)

    # 示踪剂数据
    documents = [
        ("PINECONE_TRACER_苹果手机死机", {"category": "tech", "id": 1}),
        ("PINECONE_TRACER_iPhone屏幕卡顿", {"category": "tech", "id": 2}),
        ("PINECONE_TRACER_Android手机闪退", {"category": "tech", "id": 3}),
        ("PINECONE_TRACER_香蕉很好吃", {"category": "food", "id": 4}),
        ("PINECONE_TRACER_苹果是水果", {"category": "food", "id": 5}),
    ]

    # 生成向量并插入
    vectors_to_upsert = []
    for text, metadata in documents:
        vec = generate_embedding(text)
        vectors_to_upsert.append((
            f"vec_{metadata['id']}",  # 向量ID
            vec,                      # 向量值
            metadata                  # 元数据
        ))

    # 批量插入
    print("\n📤 插入向量到Pinecone...")
    start = time.time()
    index.upsert(vectors_to_upsert)
    elapsed = time.time() - start
    print(f"✅ 插入完成，耗时: {elapsed:.3f}秒")

    # 查看索引统计
    stats = index.describe_index_stats()
    print(f"\n📊 索引统计:")
    print(f"  总向量数: {stats['total_vector_count']}")
    print(f"  维度: {stats['dimension']}")

    # 查询实验
    query_text = "PINECONE_TRACER_我的手机坏了"
    query_vec = generate_embedding(query_text)

    print(f"\n🔍 查询: {query_text}")
    start = time.time()
    results = index.query(
        vector=query_vec,
        top_k=3,
        include_metadata=True
    )
    elapsed = time.time() - start

    print(f"⏱️  查询耗时: {elapsed*1000:.2f}毫秒")
    print(f"\n📋 Top-3结果:")
    for i, match in enumerate(results['matches'], 1):
        score = match['score']
        text = match['metadata'].get('text', 'N/A')
        print(f"  {i}. 相似度: {score:.4f} | {text}")

# 运行实验
python pinecone_tracer_hnsw.py

# 观察输出：
# 1. 插入耗时（应该是毫秒级）
# 2. 查询耗时（应该是毫秒级）
# 3. Top-3结果的相似度分数

# pinecone_tracer_break.py
import pinecone
import numpy as np
from openai import OpenAI

index = pinecone.Index("pinecone-tracer")
client = OpenAI()

def test_break_cases():
    """测试边界场景"""

    print("=" * 60)
    print("💥 Pinecone 破坏性实验")
    print("=" * 60)

    # 破坏1: 查询空向量
    print("\n❌ 破坏1: 查询全零向量（无语义）")
    try:
        zero_vec = [0.0] * 1536
        results = index.query(vector=zero_vec, top_k=3)
        print(f"结果: 找到{len(results['matches'])}个匹配")
        print("🔍 解读: 全零向量也能查，但结果无意义（随机性）")
    except Exception as e:
        print(f"错误: {e}")

    # 破坏2: 查询随机向量
    print("\n❌ 破坏2: 查询随机向量（噪声）")
    try:
        random_vec = np.random.randn(1536).tolist()
        results = index.query(vector=random_vec, top_k=3)
        scores = [m['score'] for m in results['matches']]
        avg_score = np.mean(scores)
        print(f"平均相似度: {avg_score:.4f}")
        print("🔍 解读: 随机向量的相似度应该很低（<0.3）")
    except Exception as e:
        print(f"错误: {e}")

    # 破坏3: 超大top_k
    print("\n❌ 破坏3: top_k超过向量总数")
    try:
        vec = client.embeddings.create(
            model="text-embedding-3-small",
            input="测试"
        ).data[0].embedding

        results = index.query(vector=vec, top_k=10000)
        print(f"返回: {len(results['matches'])}个结果")
        print("🔍 解读: Pinecone会返回实际存在的向量数")
    except Exception as e:
        print(f"错误: {e}")

    # 破坏4: 查询不存在的命名空间
    print("\n❌ 破坏4: 查询空命名空间")
    try:
        vec = client.embeddings.create(
            model="text-embedding-3-small",
            input="测试"
        ).data[0].embedding

        results = index.query(
            vector=vec,
            top_k=3,
            namespace="non-existent-ns"
        )
        print(f"结果: {len(results['matches'])}个匹配")
        print("🔍 解读: 空命名空间返回空结果，不会报错")
    except Exception as e:
        print(f"错误: {e}")

if __name__ == "__main__":
    test_break_cases()

python pinecone_tracer_break.py

# 1. 批量插入（而非单个）
vectors_to_upsert = [(id, vec, meta) for ...]
index.upsert(vectors_to_upsert)

# 2. 设定相似度阈值
results = index.query(vector=vec, top_k=10)
filtered = [r for r in results['matches'] if r['score'] > 0.7]

# 3. 使用命名空间隔离不同数据
index.query(vector=vec, namespace="tech")
index.query(vector=vec, namespace="food")

# 4. 监控查询性能
import time
start = time.time()
results = index.query(vector=vec, top_k=10)
elapsed = time.time() - start
if elapsed > 0.1:  # 超过100ms
    print("⚠️  查询延迟过高，考虑优化索引")

# 1. 不要混淆向量ID和业务ID
# ❌ 错误
index.upsert([("苹果手机", vec, meta)])  # 中文ID可能有问题
# ✅ 正确
index.upsert([("doc_12345", vec, {"text": "苹果手机"})])

# 2. 不要忽略阈值
# ❌ 错误
results = index.query(vector=vec, top_k=10)  # 可能返回无关结果
# ✅ 正确
results = index.query(vector=vec, top_k=10, filter={'category': 'tech'})

# 3. 不要过度查询
# ❌ 错误
for vec in vecs:  # 逐个查询很慢
    index.query(vector=vec, top_k=10)
# ✅ 正确
index.query_batch(vectors=vecs, top_k=10)  # 批量查询

# 1. 查看索引统计
curl -X GET "https://index_name-abc123.svc.us-east-1-aws.pinecone.io/describe_index_stats" \
  -H "Api-Key: YOUR_API_KEY"

# 2. 测试查询延迟
time python -c "import pinecone; index.query(vector=[...])"

# 3. 批量删除向量
index.delete(delete_all=True, namespace="tech")

# 4. 查看Pinecone日志
# 在Pinecone Dashboard → Monitoring → Logs