向量数据库在AI记忆系统中的应用

向量数据库是实现AI长期记忆的关键技术，本文将深入探讨其原理与实践。

1. 向量表示

文本向量化

# 使用Sentence-BERT生成嵌入
from sentence_transformers import SentenceTransformer

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

# 文本转向量
text = "人工智能正在改变世界"
vector = model.encode(text)
# vector: [0.12, -0.34, 0.56, ...]  (384维)
​

多模态嵌入

# CLIP模型处理图像和文本
import clip

model, preprocess = clip.load("ViT-B/32")

# 图像嵌入
image = preprocess(Image.open("cat.jpg"))
image_features = model.encode_image(image)

# 文本嵌入
text = clip.tokenize(["a cat", "a dog"])
text_features = model.encode_text(text)

# 计算相似度
similarity = (image_features @ text_features.T).softmax(dim=-1)
​

2. 向量数据库架构

核心组件

class VectorDatabase:
    def __init__(self, dimension):
        self.dimension = dimension
        self.vectors = np.zeros((0, dimension))
        self.metadata = []
        self.index = None
    
    def insert(self, vector, metadata):
        self.vectors = np.vstack([self.vectors, vector])
        self.metadata.append(metadata)
        self._build_index()
    
    def search(self, query_vector, k=10):
        # 近似最近邻搜索
        distances, indices = self.index.search(query_vector, k)
        return [
            {"metadata": self.metadata[i], "distance": d}
            for i, d in zip(indices[0], distances[0])
        ]
​

索引算法

HNSW（层次化小世界图）

Layer 2:  ----->----
          |        |
Layer 1:  --->---->---->----
          |    |    |     |
Layer 0:  >-->-->-->-->-->-->
​

特点：

• 对数级搜索复杂度
• 高召回率
• 内存占用较大

IVF（倒排文件）

聚类分区：
┌─────┬─────┬─────┬─────┐
│ C1  │ C2  │ C3  │ C4  │
├─────┼─────┼─────┼─────┤
│v1 v3│v2   │v4 v7│v5 v6│
│v8 v9│v10  │v11  │v12  │
└─────┴─────┴─────┴─────┘
搜索：只查询最近的N个聚类
​

3. 主流向量数据库

Pinecone

import pinecone

# 初始化
pinecone.init(api_key="your-key")
index = pinecone.Index("my-index")

# 插入向量
index.upsert([
    ("id1", [0.1, 0.2, 0.3], {"category": "tech"}),
    ("id2", [0.4, 0.5, 0.6], {"category": "news"})
])

# 查询
results = index.query(
    vector=[0.1, 0.2, 0.3],
    top_k=10,
    filter={"category": {"$eq": "tech"}}
)
​

Weaviate

import weaviate

client = weaviate.Client("http://localhost:8080")

# 创建class
client.schema.create_class({
    "class": "Document",
    "properties": [
        {"name": "content", "dataType": ["text"]},
        {"name": "timestamp", "dataType": ["date"]}
    ],
    "vectorizer": "text2vec-transformers"
})

# 添加数据
client.data_object.create({
    "content": "AI is amazing",
    "timestamp": "2024-01-01T00:00:00Z"
}, class_name="Document")

# 语义搜索
results = client.query.get(
    "Document", ["content"]
).with_near_vector({
    "vector": query_vector
}).with_limit(5).do()
​

Milvus

from pymilvus import connections, Collection

connections.connect(host="localhost", port="19530")

collection = Collection("documents")

# 插入
mr = collection.insert([
    [1, 2, 3],                    # IDs
    [[0.1, 0.2], [0.3, 0.4]],    # vectors
    ["doc1", "doc2"]              # text
])

# 搜索
results = collection.search(
    data=[query_vector],
    anns_field="embedding",
    param={"metric_type": "L2", "params": {"nprobe": 10}},
    limit=10
)
​

4. AI记忆系统设计

记忆架构

class AIMemorySystem:
    def __init__(self):
        # 工作记忆：最近对话
        self.working_memory = deque(maxlen=10)
        
        # 短期记忆：向量存储
        self.short_term = VectorDatabase(dimension=768)
        
        # 长期记忆：关键知识
        self.long_term = VectorDatabase(dimension=768)
    
    async def store(self, message, importance=0.5):
        # 向量化
        vector = embed(message)
        
        # 工作记忆
        self.working_memory.append(message)
        
        # 根据重要性分配存储
        if importance > 0.8:
            self.long_term.insert(vector, {
                "content": message,
                "type": "important"
            })
        else:
            self.short_term.insert(vector, {
                "content": message,
                "timestamp": time.time()
            })
    
    async def recall(self, query, k=5):
        vector = embed(query)
        
        # 从短期和长期记忆中检索
        short_results = self.short_term.search(vector, k=k)
        long_results = self.long_term.search(vector, k=k)
        
        # 合并和重排序
        all_results = short_results + long_results
        return rerank(all_results, query)
​

记忆更新策略

# 记忆重要性评分
def calculate_importance(memory):
    score = 0
    
    # 时效性：最近的记忆更重要
    if memory["age"] < timedelta(hours=1):
        score += 0.3
    
    # 频率：经常访问的记忆更重要
    if memory["access_count"] > 5:
        score += 0.4
    
    # 情感强度：强烈情感的记忆更重要
    if memory["sentiment"] in ["joy", "anger", "fear"]:
        score += 0.3
    
    return score

# 记忆清理
async def cleanup_memories(self):
    for memory in self.short_term.iterate():
        importance = calculate_importance(memory)
        
        if importance > 0.8:
            # 晋升到长期记忆
            self.long_term.insert(memory)
        elif importance < 0.2:
            # 遗忘
            self.short_term.delete(memory["id"])
​

5. 实战案例

ChatGPT记忆系统

# 用户偏好记忆
class UserPreferences:
    def __init__(self):
        self.db = VectorDatabase(dimension=384)
    
    def learn_preference(self, context, user_feedback):
        # 用户说"我喜欢这个回答"
        if "like" in user_feedback:
            vector = embed(context)
            self.db.insert(vector, {
                "preference": "positive",
                "domain": infer_domain(context)
            })
    
    def get_preferences(self, current_context):
        # 检索相关偏好
        vector = embed(current_context)
        results = self.db.search(vector, k=3)
        
        # 应用到生成
        return [r["metadata"] for r in results]
​

代码助手记忆

# 代码上下文记忆
class CodeMemory:
    def remember_pattern(self, code_pattern, description):
        # 提取代码向量
        code_vector = embed_code(code_pattern)
        
        # 存储模式
        self.db.insert(code_vector, {
            "pattern": code_pattern,
            "description": description,
            "language": detect_language(code_pattern)
        })
    
    def suggest_pattern(self, current_code):
        # 找到相似模式
        vector = embed_code(current_code)
        similar = self.db.search(vector, k=5)
        
        return [s["metadata"]["pattern"] for s in similar]
​

6. 优化技巧

1. 混合检索：向量搜索+关键词过滤
2. 动态更新：定期重新向量化旧数据
3. 量化压缩：PQ/OPQ降低存储成本
4. 分片策略：按时间或类别分片
5. 缓存热门查询：LRU缓存常见查询

向量数据库正在成为AI系统不可或缺的组件，赋予AI真正的"记忆"能力。