AI 时代的数据仓库：阿里云 AnalyticDB MySQL 向量检索 + SQL 分析一体化实战

-- 创建包含向量列的知识库表
CREATE TABLE knowledge_base (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    title VARCHAR(500),
    content TEXT,
    source VARCHAR(200),
    category VARCHAR(100),
    created_at DATETIME DEFAULT NOW(),
    -- 向量列：1536维（兼容 OpenAI text-embedding-3-small）
    content_embedding VECTOR(1536),
    -- 全文检索索引
    FULLTEXT INDEX idx_content (content) WITH PARSER nlp,
    -- 向量索引（HNSW 算法，推荐方案）
    ANN INDEX idx_vector (content_embedding) 
        ALGORITHM = HNSW 
        DISTANCE = COSINE
        OPTIONS '{"M": 32, "ef_construction": 200}'
) DISTRIBUTE BY HASH(id);

import pymysql
from openai import OpenAI

# 连接 AnalyticDB MySQL
conn = pymysql.connect(
    host='adb-xxx.ads.aliyuncs.com',
    port=3306,
    user='admin',
    password='your_password',
    database='ai_knowledge'
)

# 生成 Embedding
client = OpenAI(api_key="your-api-key")

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

# 批量写入知识库
documents = [
    {"title": "AnalyticDB产品介绍", "content": "AnalyticDB MySQL版是..."},
    {"title": "弹性扩缩容指南", "content": "支持秒级弹性扩缩容..."},
]

cursor = conn.cursor()
for doc in documents:
    embedding = get_embedding(doc["content"])
    # 向量以 JSON 数组格式写入
    cursor.execute(
        "INSERT INTO knowledge_base (title, content, content_embedding) "
        "VALUES (%s, %s, %s)",
        (doc["title"], doc["content"], str(embedding))
    )
conn.commit()

-- 基于问题语义检索最相关的 Top 5 文档
SELECT id, title, content,
       COSINE_DISTANCE(content_embedding, :query_vector) AS distance
FROM knowledge_base
ORDER BY distance ASC
LIMIT 5;

-- 在特定分类下进行语义检索
SELECT id, title, content,
       COSINE_DISTANCE(content_embedding, :query_vector) AS distance
FROM knowledge_base
WHERE category = '产品文档'
  AND created_at > '2024-01-01'
ORDER BY distance ASC
LIMIT 10;

-- 混合检索：语义相似度 + 关键词匹配 加权融合
SELECT id, title, content,
       0.7 * (1 - COSINE_DISTANCE(content_embedding, :query_vector)) 
       + 0.3 * MATCH(content) AGAINST('AnalyticDB 弹性扩容' IN NLP MODE) 
       AS relevance_score
FROM knowledge_base
WHERE MATCH(content) AGAINST('AnalyticDB 弹性扩容' IN NLP MODE)
   OR COSINE_DISTANCE(content_embedding, :query_vector) < 0.3
ORDER BY relevance_score DESC
LIMIT 5;

from openai import OpenAI
import pymysql

class ADBRagPipeline:
    def __init__(self, adb_config, openai_key):
        self.conn = pymysql.connect(**adb_config)
        self.llm = OpenAI(api_key=openai_key)
    
    def retrieve(self, query, top_k=5):
        """向量检索 + 全文检索混合召回"""
        query_embedding = self._get_embedding(query)
        
        sql = """
        SELECT title, content,
               COSINE_DISTANCE(content_embedding, %s) AS vec_dist
        FROM knowledge_base
        WHERE COSINE_DISTANCE(content_embedding, %s) < 0.4
        ORDER BY vec_dist ASC
        LIMIT %s
        """
        cursor = self.conn.cursor()
        cursor.execute(sql, (str(query_embedding), str(query_embedding), top_k))
        return cursor.fetchall()
    
    def generate(self, query, contexts):
        """基于检索结果生成回答"""
        context_text = "\n---\n".join([f"[{c[0]}]: {c[1]}" for c in contexts])
        
        response = self.llm.chat.completions.create(
            model="qwen-plus",
            messages=[
                {"role": "system", "content": f"基于以下知识库内容回答问题：\n{context_text}"},
                {"role": "user", "content": query}
            ]
        )
        return response.choices[0].message.content
    
    def query(self, question):
        """完整 RAG 流程"""
        contexts = self.retrieve(question)
        answer = self.generate(question, contexts)
        return answer

# 使用示例
rag = ADBRagPipeline(
    adb_config={"host": "adb-xxx.ads.aliyuncs.com", "port": 3306,
                "user": "admin", "password": "xxx", "database": "ai_knowledge"},
    openai_key="your-key"
)
answer = rag.query("AnalyticDB MySQL如何实现弹性扩缩容？")
print(answer)