第5章：典型应用场景 - AI检索、推荐系统、语义搜索¶

# RAG系统实现示例
from sentence_transformers import SentenceTransformer
import vector_db  # 假设的向量数据库客户端

# 1. 初始化组件
model = SentenceTransformer('all-MiniLM-L6-v2')
db = vector_db.Client("localhost:6333")
collection = db.get_collection("company_docs")

# 2. 用户查询
user_query = "公司年假政策是什么？"

# 3. 将查询向量化
query_embedding = model.encode(user_query)

# 4. 检索相关文档
results = collection.search(
    vector=query_embedding,
    top_k=5,
    filter={"department": "HR"}  # 可选过滤
)

# 5. 构建上下文
context = "\n\n".join([r.content for r in results])

# 6. 构建提示词
prompt = f"""基于以下信息回答用户问题。如果信息不足，请说明不知道。

上下文：
{context}

用户问题：{user_query}

回答："""

# 7. 调用LLM生成回答
response = llm.generate(prompt)
print(response)

# 常见的分块策略

# 1. 固定大小分块（最简单）
def chunk_by_size(text, chunk_size=500, overlap=50):
    chunks = []
    for i in range(0, len(text), chunk_size - overlap):
        chunks.append(text[i:i + chunk_size])
    return chunks

# 2. 按段落分块（保持语义完整）
def chunk_by_paragraph(text):
    paragraphs = text.split('\n\n')
    return [p for p in paragraphs if p.strip()]

# 3. 递归分块（更智能）
def chunk_recursively(text, delimiters=['\n\n', '\n', '。', '！']):
    # 优先尝试大的分隔符，逐级细化
    pass

# 推荐系统向量表示
user_vector = model.encode("用户A的兴趣特征")
item_vector = model.encode("物品B的特征")

# 计算用户对物品的兴趣分数
score = np.dot(user_vector, item_vector)  # 点积
top_items = find_top_k(score, k=10)

# 电影推荐系统示例
from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer('all-MiniLM-L6-v2')

# 电影数据库
movies = [
    {"id": 1, "title": "星际穿越", "genre": "科幻"},
    {"id": 2, "title": "泰坦尼克号", "genre": "爱情"},
    {"id": 3, "title": "盗梦空间", "genre": "科幻"},
    {"id": 4, "title": "阿甘正传", "genre": "剧情"},
]

# 构建电影向量
movie_vectors = model.encode([m["title"] for m in movies])

# 用户偏好
user_liked = "科幻电影 太空 时间旅行"
user_vector = model.encode(user_liked)

# 计算相似度
similarities = np.dot(user_vector, movie_vectors.T)

# 排序并返回推荐
top_indices = np.argsort(similarities)[::-1][:2]
recommendations = [movies[i]["title"] for i in top_indices]

print(f"推荐电影: {recommendations}")
# 输出: ['星际穿越', '盗梦空间']

# 语义搜索代码示例
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

# 文档语料库
documents = [
    "人工智能将改变我们的生活方式",
    "机器学习是AI的一个重要分支",
    "今天天气真好，适合外出游玩",
    "深度学习在图像识别领域取得突破",
    "最新的iPhone手机采用了A17芯片",
]

# 向量化文档
doc_vectors = model.encode(documents)

# 语义搜索查询
query = "什么是深度学习？"
query_vector = model.encode(query)

# 计算相似度
from sklearn.metrics.pairwise import cosine_similarity
similarities = cosine_similarity([query_vector], doc_vectors)[0]

# 返回最相关的文档
top_idx = np.argsort(similarities)[::-1][0]
print(f"最相关文档: {documents[top_idx]}")
# 输出: '深度学习在图像识别领域取得突破'

# 跨语言语义搜索
query_vector = model.encode("How to cook pasta?")  # 英文查询

# 中文文档也能被检索到
chinese_docs = [
    "如何做意大利面",
    "北京天气怎么样",
    "学习Python编程",
]

doc_vectors = model.encode(chinese_docs)
# 通过多语言模型，语义相近的内容会返回高相似度

# 以图搜图示例
import clip
import torch
from PIL import Image

# 加载CLIP模型
model, preprocess = clip.load("ViT-B/32", device="cpu")

# 图片数据库向量化
def encode_image(img_path):
    image = preprocess(Image.open(img_path)).unsqueeze(0)
    with torch.no_grad():
        image_features = model.encode_image(image)
    return image_features.numpy()

# 图片集合
image_paths = ["cat1.jpg", "cat2.jpg", "dog.jpg"]
image_vectors = [encode_image(p) for p in image_paths]

# 查询图片
query_image = preprocess(Image.open("query_cat.jpg")).unsqueeze(0)
query_vector = model.encode_image(query_image).numpy()

# 找出最相似的图片
similarities = cosine_similarity(query_vector, np.array(image_vectors))
most_similar_idx = np.argmax(similarities)
print(f"最相似的图片: {image_paths[most_similar_idx]}")

# 金融欺诈检测示例
from sklearn.cluster import KMeans
import numpy as np

# 正常交易特征向量
normal_transactions = np.random.rand(1000, 20)  # 1000笔正常交易，20维特征

# 学习正常模式
kmeans = KMeans(n_clusters=10, random_state=42)
kmeans.fit(normal_transactions)

# 检测新交易
new_transaction = np.random.rand(1, 20)

# 计算到最近正常中心的距离
distance = np.min(np.linalg.norm(
    new_transaction - kmeans.cluster_centers_,
    axis=1
))

# 判断是否异常
threshold = 2.5
is_fraud = distance > threshold
print(f"交易异常: {is_fraud}, 距离: {distance:.2f}")

# AI Agent记忆系统
class AgentMemory:
    def __init__(self, vector_db, embed_model):
        self.db = vector_db
        self.embed = embed_model

    def add_memory(self, content, memory_type="experience"):
        """添加记忆"""
        vector = self.embed.encode(content)
        self.db.insert({
            "vector": vector,
            "content": content,
            "type": memory_type
        })

    def retrieve(self, query, top_k=5):
        """检索相关记忆"""
        query_vector = self.embed.encode(query)
        results = self.db.search(query_vector, top_k=top_k)
        return [r["content"] for r in results]

    def reflect(self, recent_experiences):
        """反思总结：将短期经验转化为长期记忆"""
        summary = self.llm.summarize(recent_experiences)
        self.add_memory(summary, memory_type="reflection")

# 使用示例
agent = AgentMemory(vector_db, embed_model)

# Agent与环境交互
while not task_complete:
    # 1. 检索相关记忆
    relevant_memories = agent.retrieve(f"当前任务: {task}")
    context = "\n".join(relevant_memories)

    # 2. 基于记忆决策
    action = agent.decide(context, task)

    # 3. 执行并记录
    result = agent.execute(action)
    agent.add_memory(f"执行{action}，结果：{result}")