AI 应用工作流

🤖 AI 应用工作流

使用 seekdb 构建 AI 应用的典型工作流程。本课程将带你了解从数据准备到智能问答的完整 RAG（检索增强生成）流程。

RAG 工作流架构

📄

Step 1数据准备

🔢

Step 2向量化

🗄️

Step 3存储索引

🔍

Step 4检索召回

🧠

Step 5LLM 生成

工作流详解

1

数据准备

准备要导入的知识库数据

将文档、网页、FAQ 等数据进行预处理，切分成适合检索的文本块。

data_preparation.py

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

# Python 示例：文档分块
from pyseekdb import TextSplitter

splitter = TextSplitter(
    chunk_size=500,      # 每块最大字符数
    chunk_overlap=50,    # 块之间重叠字符数
    separator="\n\n"   # 分割符
)

documents = [
    "seekdb 是一个 AI 原生数据库...",
    "它支持向量搜索和语义搜索...",
]

chunks = splitter.split(documents)
print(f"共生成 {len(chunks)} 个文本块")

运行结果

点击"运行"按钮查看结果

2

向量化 (Embedding)

将文本转换为向量表示

使用嵌入模型将文本转换为高维向量，捕获语义信息。

embedding.sql

1

2

3

4

5

6

7

8

9

10

11

-- 使用 seekdb 内置的 AI_EMBED 函数
-- 自动调用配置的嵌入模型生成向量

INSERT INTO knowledge_base (content, embedding)
SELECT 
    chunk_text,
    AI_EMBED(chunk_text)  -- 自动生成 1536 维向量
FROM temp_chunks;

-- 或使用 Python SDK
# embeddings = client.embed(texts=chunks, model="text-embedding-3-small")

运行结果

点击"运行"按钮查看结果

3

存储与索引

将向量存储到 seekdb 并建立索引

创建向量索引以加速后续的相似度搜索。

create_index.sql

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

-- 创建知识库表
CREATE TABLE knowledge_base (
    id INT PRIMARY KEY AUTO_INCREMENT,
    title VARCHAR(255),
    content TEXT,
    source VARCHAR(500),
    embedding VECTOR(1536),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 创建 HNSW 向量索引
CREATE INDEX idx_kb_embedding ON knowledge_base
USING HNSW (embedding) WITH (
    M = 16,                 -- 每个节点的最大连接数
    ef_construction = 200   -- 构建时的搜索范围
);

-- 创建全文索引（用于混合搜索）
CREATE FULLTEXT INDEX idx_kb_content ON knowledge_base(title, content);

运行结果

点击"运行"按钮查看结果

4

检索召回

根据用户查询检索相关文档

将用户问题向量化，检索最相关的知识库内容。

retrieval.sql

1

2

3

4

5

6

7

8

9

10

11

12

-- 向量检索：找到最相关的文档
SET @query = '如何使用 seekdb 进行向量搜索？';
SET @query_embedding = AI_EMBED(@query);

SELECT 
    id,
    title,
    content,
    COSINE_DISTANCE(embedding, @query_embedding) AS relevance_score
FROM knowledge_base
ORDER BY relevance_score ASC
LIMIT 5;  -- 返回最相关的 5 个结果

运行结果

点击"运行"按钮查看结果

5

LLM 生成回答

将检索结果作为上下文，由 LLM 生成回答

使用检索到的内容作为上下文，调用 LLM 生成最终答案。

generate.sql

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

-- 使用 seekdb 的 AI_GENERATE 函数生成回答
SET @context = (
    SELECT GROUP_CONCAT(content SEPARATOR '\n\n')
    FROM (
        SELECT content
        FROM knowledge_base
        ORDER BY COSINE_DISTANCE(embedding, AI_EMBED(@query)) ASC
        LIMIT 3
    ) AS top_docs
);

SELECT AI_GENERATE(
    CONCAT(
        '基于以下上下文回答问题：\n\n',
        '上下文：', @context, '\n\n',
        '问题：', @query, '\n\n',
        '请用简洁专业的语言回答。'
    )
) AS answer;

运行结果

点击"运行"按钮查看结果

完整 RAG 示例

以下是一个完整的 Python 代码示例，实现了基于 seekdb 的 RAG 应用：

rag_application.py

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

from pyseekdb import Client, TextSplitter

# 1. 连接 seekdb
client = Client(host="localhost", port=2881, user="root", password="***")

# 2. 准备知识库
documents = [
    {"title": "seekdb 简介", "content": "seekdb 是 AI 原生向量数据库..."},
    {"title": "向量搜索", "content": "向量搜索基于余弦相似度..."},
]

# 3. 文档分块
splitter = TextSplitter(chunk_size=500)
chunks = []
for doc in documents:
    for chunk in splitter.split(doc["content"]):
        chunks.append({"title": doc["title"], "content": chunk})

# 4. 存储到数据库（自动向量化）
for chunk in chunks:
    client.execute('''
        INSERT INTO knowledge_base (title, content, embedding)
        VALUES (%s, %s, AI_EMBED(%s))
    ''', (chunk["title"], chunk["content"], chunk["content"]))

# 5. RAG 查询函数
def rag_query(question: str) -> str:
    # 检索相关文档
    results = client.query('''
        SELECT content
        FROM knowledge_base
        ORDER BY COSINE_DISTANCE(embedding, AI_EMBED(%s)) ASC
        LIMIT 3
    ''', (question,))
    
    context = "\n\n".join([r["content"] for r in results])
    
    # 生成回答
    answer = client.query('''
        SELECT AI_GENERATE(%s) AS answer
    ''', (f"基于上下文回答：\n{context}\n\n问题：{question}",))
    
    return answer[0]["answer"]

# 使用示例
print(rag_query("什么是向量搜索？"))

from pyseekdb import Client, TextSplitter

# 1. 连接 seekdb
client = Client(host="localhost", port=2881, user="root", password="***")

# 2. 准备知识库
documents = [
    {"title": "seekdb 简介", "content": "seekdb 是 AI 原生向量数据库..."},
    {"title": "向量搜索", "content": "向量搜索基于余弦相似度..."},
]

# 3. 文档分块
splitter = TextSplitter(chunk_size=500)
chunks = []
for doc in documents:
    for chunk in splitter.split(doc["content"]):
        chunks.append({"title": doc["title"], "content": chunk})

# 4. 存储到数据库（自动向量化）
for chunk in chunks:
    client.execute('''
        INSERT INTO knowledge_base (title, content, embedding)
        VALUES (%s, %s, AI_EMBED(%s))
    ''', (chunk["title"], chunk["content"], chunk["content"]))

# 5. RAG 查询函数
def rag_query(question: str) -> str:
    # 检索相关文档
    results = client.query('''
        SELECT content
        FROM knowledge_base
        ORDER BY COSINE_DISTANCE(embedding, AI_EMBED(%s)) ASC
        LIMIT 3
    ''', (question,))
    
    context = "\n\n".join([r["content"] for r in results])
    
    # 生成回答
    answer = client.query('''
        SELECT AI_GENERATE(%s) AS answer
    ''', (f"基于上下文回答：\n{context}\n\n问题：{question}",))
    
    return answer[0]["answer"]

# 使用示例
print(rag_query("什么是向量搜索？"))

运行结果

点击"运行"按钮查看结果

典型应用场景

💬

智能客服

基于企业知识库的智能问答，提供准确的客户支持

📚

文档问答

让用户通过自然语言查询大量文档内容

🔎

语义搜索

超越关键词匹配，理解用户真正的搜索意图

🎯

最佳实践

文档分块建议

chunk_size 建议设置为 300-800 字符
保持适当的 overlap 以维护上下文连贯性
考虑按段落、章节等自然边界分块

检索优化

使用混合搜索提高召回率
考虑使用 Rerank 模型对结果重排序
根据业务需求调整召回数量（通常 3-10 个）

注意事项

确保嵌入模型版本一致
定期更新知识库内容
监控 LLM 调用的延迟和成本

seekdb

🤖 AI 应用工作流

RAG 工作流架构

工作流详解

数据准备

运行结果

向量化 (Embedding)

运行结果

存储与索引

运行结果

检索召回

运行结果

LLM 生成回答

运行结果

完整 RAG 示例

运行结果

典型应用场景

智能客服

文档问答

语义搜索

推荐系统

最佳实践

文档分块建议

检索优化

注意事项