AI Agent与向量数据库入门指南：从零搭建智能问答系统

1次阅读

共计 2381 个字符，预计需要花费 6 分钟才能阅读完成。

传统关系型数据库在处理非结构化数据（如文本、图像）时存在明显局限。当我们需要实现语义搜索或相似度匹配时，传统数据库只能进行精确匹配或简单模糊查询，无法理解内容的深层含义。例如，搜索 ” 宠物医疗建议 ” 时，传统数据库可能无法识别与 ” 狗狗健康护理 ” 相关的文档。

向量数据库通过将内容转换为高维向量表示，能够捕获语义信息，实现 ” 理解内容 ” 而非仅仅匹配关键词的搜索方式。这种能力对构建智能问答系统至关重要。

目前主流的向量数据库解决方案包括：

Pinecone：全托管服务，API 简单，适合快速原型开发
Milvus：开源高性能，支持分布式部署，社区活跃
Weaviate：内置机器学习模型，支持多模态检索

对于新手入门，我推荐从 Pinecone 开始，它无需管理基础设施，免费层足够用于学习和小型项目。

一个典型的问答系统 AI Agent 包含以下组件：

输入处理：接收用户问题
嵌入生成：将问题转换为向量
向量搜索：在数据库中查找相似内容
答案生成：基于匹配结果生成回答
输出呈现：返回最终响应

现代 AI 系统通常使用预训练模型生成文本嵌入。OpenAI 的 text-embedding-ada-002 模型是目前效果较好的选择：

文本预处理（清理、分词）
通过嵌入模型转换为 768/1536 维向量
归一化处理（提高搜索准确性）
存储到向量数据库

向量数据库使用近似最近邻 (ANN) 算法快速查找相似向量，常见算法包括：

HNSW（分层可导航小世界）
IVF（反向文件）
PQ（乘积量化）

这些算法在精度和速度之间取得平衡，使得百万级向量能在毫秒级完成搜索。

import openai
import pinecone
from typing import List

# 初始化 OpenAI 和 Pinecone
openai.api_key = "your-openai-key"
pinecone.init(api_key="your-pinecone-key", environment="us-west1-gcp")

# 创建或连接索引
index_name = "qa-system"
if index_name not in pinecone.list_indexes():
    pinecone.create_index(name=index_name, dimension=1536, metric="cosine")
index = pinecone.Index(index_name)

# 生成嵌入向量
def get_embedding(text: str) -> List[float]:
    response = openai.Embedding.create(
        input=text,
        model="text-embedding-ada-002"
    )
    return response["data"][0]["embedding"]

# 存储问答对
def store_qa(question: str, answer: str):
    # 为问题和答案生成独立嵌入
    q_embedding = get_embedding(question)
    a_embedding = get_embedding(answer)

    # 使用问题 ID 作为键，存储问题和答案向量
    doc_id = str(hash(question))  # 简单哈希生成唯一 ID
    index.upsert([(doc_id, q_embedding, {"answer": answer, "a_vec": a_embedding})])

# 查询最匹配答案
def query_answer(user_question: str, top_k=3) -> List[str]:
    query_vec = get_embedding(user_question)
    results = index.query(vector=query_vec, top_k=top_k, include_metadata=True)

    return [match["metadata"]["answer"] for match in results["matches"]]

# 示例使用
if __name__ == "__main__":
    # 预存一些问答对
    store_qa("如何照顾宠物狗", "定期遛狗、提供优质狗粮、每年体检")
    store_qa("猫咪常见疾病", "猫藓、泌尿系统疾病、消化问题")

    # 用户提问
    user_q = "我的狗狗需要什么护理"
    answers = query_answer(user_q)
    print(f"问题: {user_q}")
    print("最佳答案:", answers[0])