построение интеллектуальной системы вопросов и ответов и корпоративной базы знаний на базе StarRocks + DeepSeek / Хабр

Автор(ы):

Ши Цян, архитектор решений
Чжао Хэн, StarRocks TSC Member

Введение в RAG и векторные индексы

RAG (Retrieval-Augmented Generation — извлечение с последующей генерацией) сочетает поиск внешних знаний и генерацию ответов, компенсируя статичность знаний классических LLM и склонность к «галлюцинациям». Это повышает точность и опору на актуальные данные.

Стандартный конвейер RAG:

Извлечение (Retrieval): поиск релевантного контента в внешних источниках (векторные базы, поисковые движки, традиционные БД).
Генерация (Generation): объединение запроса пользователя и найденного контекста для LLM (например, GPT, LLaMA, DeepSeek), чтобы получить усиленный фактологичный ответ.

Изображения, документы, видео и аудио предварительно преобразуются в эмбеддинги и сохраняются в векторном хранилище. Эмбеддинги — это, как правило, высокоразмерные массивы float.
Для ускорения семантического поиска применяются векторные индексы (HNSW, IVFPQ) и ANN (Approximate Nearest Neighbor).
Такой подход уменьшает нагрузку высокоразмерных вычислений и снижает риск «галлюцинаций», делая ответы естественными и надёжными.

Подробнее о векторном поиске см. материалы по StarRocks Vector Index.

Типовые сценарии на базе StarRocks + DeepSeek

DeepSeek: генерация качественных эмбеддингов и ответов.
StarRocks: высокоэффективный векторный поиск и хранение.

Вместе они образуют основу для точных и масштабируемых AI‑решений.

1) Корпоративная база знаний

Подходит для:

Поиска по внутренним документам компании, FAQ
Вопрос‑ответ в специализированных доменах (право, финансы, медицина)
Поиска по коду, документации и архитектурным описаниям

Решение:

Встраивание документов (DeepSeek): преобразование текстов в векторы.
Хранение и индексирование (StarRocks): HNSW или IVFPQ с быстрым ANN‑поиском.
RAG: запрос → эмбеддинг запроса → векторный поиск → получение контекста → генерация ответа DeepSeek с опорой на контекст.

2) AI‑поддержка и интеллектуальный вопрос‑ответ

Подходит для:

��нтеллектуальных сервисов поддержки (банк, брокер, e‑commerce)
Профконсультаций (право, медицина)
Автоматического техподдержки/FAQ

Решение:

Встраивание диалогов (DeepSeek): подготовка эмбеддингов для истории чатов и намерений.
Хранение и индексирование (StarRocks): быстрый поиск похожих кейсов.
RAG: объединение истории + базы знаний + DeepSeek для генерации точных ответов.

Пример:

Пользователь: «Как сменить номер телефона, привязанный к банковской карте?»
StarRocks извлекает 3 наиболее похожих случая.
DeepSeek, учитывая их + FAQ, формирует точный, контекстный ответ.

Демонстрация: архитектура и шаги

Состав системы:

DeepSeek: текстовые эмбеддинги и генерация ответов
StarRocks: хранение и быстрый ANN‑поиск по векторным данным (векторные индексы поддерживаются с версии 3.4+)

Процесс:

Подготовка окружения — Ollama, StarRocks
Векторизация данных — DeepSeek Embedding (например, размерность 3584)
Хранение векторов — таблицы StarRocks
ANN‑поиск — индексы IVFPQ/HNSW
Усиление (Augmentation) — комбинирование извлечённого контента
Генерация ответа — DeepSeek LLM

1. Подготовка окружения

1.1 Локальное развёртывание DeepSeek через Ollama

Скачать Ollama: https://ollama.com/
Установить и запустить модель DeepSeek (пример — deepseek-r1:7b):

# Скачает и запустит модель
ollama run deepseek-r1:7b

[изображение]

Совет: для использования облачного DeepSeek (официальный API) получите API Key на https://platform.deepseek.com (разделы «API Keys» или «Developer»), например: sk-xxxxxxxxxxxxxxxx.

Начало работы:

# Войти в интерактивный режим
ollama run deepseek-r1:7b

Оптимизация производительности (на один запуск):

export OLLAMA_GPU_LAYERS=35
export OLLAMA_CPU_THREADS=6
export OLLAMA_BATCH_SIZE=128
export OLLAMA_CONTEXT_SIZE=4096
ollama run deepseek-r1:7b

Вывод: напрямую задавая вопросы модели, вы получите общие ответы. Для точности и контекстности нужен RAG.

1.2 Подготовка StarRocks

Требуемая версия: 3.4+

Включение векторных индексов:

-- Динамически (до перезапуска FE)
ADMIN SET FRONTEND CONFIG ("enable_experimental_vector" = "true");

-- Постоянно: в fe.conf
-- enable_experimental_vector = true
-- затем перезапустить FE

Создание базы и таблицы хранения эмбеддингов:

CREATE DATABASE knowledge_base;

CREATE TABLE enterprise_knowledge (
    id  BIGINT AUTO_INCREMENT,
    content TEXT NOT NULL,
    embedding ARRAY<FLOAT> NOT NULL,
    INDEX vec_idx (embedding) USING VECTOR (
        "index_type" = "hnsw",
        "dim" = "3584",
        "metric_type" = "l2_distance",
        "M" = "16",
        "efconstruction" = "40"
    )
) ENGINE=OLAP
PRIMARY KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES (
    "replication_num" = "1"
);

Примечание: модель deepseek-r1:7b часто выдаёт эмбеддинги размерности 3584.

2. Преобразование текста в эмбеддинг

Тест локального Embedding API Ollama:

curl -X POST http://localhost:11434/api/embeddings \
  -d '{"model": "deepseek-r1:7b", "prompt": "Срок гарантии на продукт — один год."}'

3. Загрузка эмбеддингов в StarRocks

import pymysql
import requests

def get_embedding(text: str):
    url = "http://localhost:11434/api/embeddings"
    payload = {"model": "deepseek-r1:7b", "prompt": text}
    resp = requests.post(url, json=payload, timeout=30)
    resp.raise_for_status()
    return resp.json()["embedding"]

try:
    content = "Видение StarRocks — сделать аналитику данных более простой и гибкой."
    embedding = get_embedding(content)

    # Преобразуем Python-список в формат ARRAY<FLOAT> StarRocks
    embedding_str = "[" + ",".join(map(str, embedding)) + "]"

    conn = pymysql.connect(
        host="X.X.X.X",
        port=9030,
        user="root",
        password="sr123456",
        database="knowledge_base",
        autocommit=True,
    )
    with conn.cursor() as cur:
        sql = "INSERT INTO enterprise_knowledge (content, embedding) VALUES (%s, %s)"
        cur.execute(sql, (content, embedding_str))
    print(f"Inserted: {content} with embedding {embedding[:5]}...")

except requests.RequestException as e:
    print(f"Embedding API error: {e}")
except pymysql.Error as e:
    print(f"Database error: {e}")

4. Извлечение знаний (ANN‑поиск и результат)

Важно: чтобы запрос задействовал векторный индекс, используйте приближённые функции и формат ORDER BY согласно документации StarRocks:

Функция:
- l2_distance → approx_l2_distance
- cosine_similarity → approx_cosine_similarity
Формат ORDER BY: ORDER BY approx_l2_distance(vector_column, constant_array)
constant_array — константный ARRAY той же размерности, что dim.
Обязателен LIMIT N.
В ORDER BY не должно быть других столбцов.

Пример на HNSW с параметром efsearch:

import pymysql
import requests
import json

def get_embedding(text: str):
    url = "http://localhost:11434/api/embeddings"
    payload = {"model": "deepseek-r1:7b", "prompt": text}
    resp = requests.post(url, json=payload, timeout=30)
    resp.raise_for_status()
    return resp.json()["embedding"]

def search_knowledge_base(query_embedding, topk=3, efsearch=128):
    # Преобразуем запрос в формат ARRAY<FLOAT>
    qv = "[" + ",".join(map(str, query_embedding)) + "]"
    sql = f"""
    SELECT /*+ SET_VAR(ann_params='{{efsearch={efsearch}}}') */
           content,
           approx_l2_distance(embedding, {qv}) AS distance
    FROM enterprise_knowledge
    ORDER BY approx_l2_distance(embedding, {qv}) ASC
    LIMIT {topk}
    """
    conn = pymysql.connect(
        host="X.X.X.X",
        port=9030,
        user="root",
        password="sr123456",
        database="knowledge_base"
    )
    try:
        with conn.cursor() as cur:
            cur.execute(sql)
            rows = cur.fetchall()
            return [r[0] for r in rows] if rows else []
    finally:
        conn.close()

# Пример использования
query = "Каково видение StarRocks?"
qe = get_embedding(query)
contexts = search_knowledge_base(qe, topk=3, efsearch=128)
print(contexts)

Проверка применения индекса:

EXPLAIN
SELECT
    content
FROM enterprise_knowledge
ORDER BY approx_l2_distance(embedding, [1,1,1,1,1]) ASC
LIMIT 3;

-- В OlapScanNode должно быть: VECTORINDEX: ON

Точный пересчёт (для повторного ранжирования top‑K):

-- допустим, получили список id из приближённого запроса
SELECT content, l2_distance(embedding, [1,1,1,1,1]) AS dist
FROM enterprise_knowledge
WHERE id IN (/* topK ids */)
ORDER BY dist ASC
LIMIT 3;

5. Подключение RAG‑усиления

5.1 Формирование Prompt и генерация ответа DeepSeek

import requests, json, re

def build_rag_prompt(query: str, retrieved_content: str) -> str:
    return f"""
[Системная инструкция] Вы — интеллектуальный ассистент. Отвечайте строго по контексту.
[Контекст знаний]
{retrieved_content}

[Вопрос пользователя]
{query}

Если данных недостаточно, объясните, чего не хватает.
"""

def clean_response(text: str) -> str:
    # Удаляем служебные блоки вида <think>...</think>
    return re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()

def generate_answer(prompt: str) -> str:
    url = "http://localhost:11434/api/generate"
    payload = {"model": "deepseek-r1:7b", "prompt": prompt}
    resp = requests.post(url, json=payload, timeout=120)
    resp.raise_for_status()

    full = ""
    for line in resp.text.splitlines():
        if not line.strip():
            continue
        try:
            obj = json.loads(line)
            if "response" in obj:
                full += obj["response"]
            if obj.get("done", False):
                break
        except json.JSONDecodeError:
            continue
    return clean_response(full.strip())

5.2 Журнал процесса RAG

CREATE TABLE customer_service_log (
    id BIGINT AUTO_INCREMENT,
    user_id VARCHAR(50),
    question TEXT NOT NULL,
    question_embedding ARRAY<FLOAT> NOT NULL,
    retrieved_content TEXT,
    generated_answer TEXT,
    timestamp DATETIME NOT NULL,
    feedback TINYINT DEFAULT NULL
) ENGINE=OLAP
PRIMARY KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES ("replication_num" = "1");

6. Оптимизированная версия

6.1 Конвейер RAG «под ключ»

import pymysql, requests, json, logging, re
from datetime import datetime

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

def get_embedding(text: str):
    url = "http://localhost:11434/api/embeddings"
    payload = {"model": "deepseek-r1:7b", "prompt": text}
    r = requests.post(url, json=payload, timeout=30)
    r.raise_for_status()
    return r.json()["embedding"]

def search_knowledge_base(query_embedding, topk=3, efsearch=128):
    qv = "[" + ",".join(map(str, query_embedding)) + "]"
    sql = f"""
    SELECT /*+ SET_VAR(ann_params='{{efsearch={efsearch}}}') */
           content,
           approx_l2_distance(embedding, {qv}) AS distance
    FROM enterprise_knowledge
    ORDER BY approx_l2_distance(embedding, {qv}) ASC
    LIMIT {topk}
    """
    conn = pymysql.connect(host="X.X.X.X", port=9030, user="root", password="sr123456", database="knowledge_base")
    try:
        with conn.cursor() as cur:
            cur.execute(sql)
            rows = cur.fetchall()
            return "\n\n---\n\n".join([r[0] for r in rows]) if rows else ""
    finally:
        conn.close()

def build_rag_prompt(query: str, retrieved_content: str) -> str:
    return f"""
[Системная инструкция] Вы — корпоративный ассистент. Отвечайте кратко и по делу, строго по контексту.
[Контекст знаний]
{retrieved_content}

[Вопрос]
{query}
"""

def clean_response(text: str) -> str:
    return re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()

def generate_answer(prompt: str) -> str:
    url = "http://localhost:11434/api/generate"
    payload = {"model": "deepseek-r1:7b", "prompt": prompt}
    r = requests.post(url, json=payload, timeout=120)
    r.raise_for_status()
    full = ""
    for line in r.text.splitlines():
        if not line.strip():
            continue
        try:
            obj = json.loads(line)
            if "response" in obj: full += obj["response"]
            if obj.get("done", False): break
        except json.JSONDecodeError:
            continue
    return clean_response(full.strip())

def log_conversation(user_id, question, question_embedding, retrieved_content, generated_answer):
    emb = "[" + ",".join(map(str, question_embedding)) + "]"
    sql = """
    INSERT INTO customer_service_log
        (user_id, question, question_embedding, retrieved_content, generated_answer, timestamp)
    VALUES (%s, %s, %s, %s, %s, NOW())
    """
    conn = pymysql.connect(host="X.X.X.X", port=9030, user="root", password="sr123456", database="knowledge_base", autocommit=True)
    try:
        with conn.cursor() as cur:
            cur.execute(sql, (user_id, question, emb, retrieved_content, generated_answer))
    finally:
        conn.close()

def rag_pipeline(user_id: str, query: str) -> str:
    try:
        qe = get_embedding(query)
        ctx = search_knowledge_base(qe, topk=3, efsearch=128)
        prompt = build_rag_prompt(query, ctx)
        ans = generate_answer(prompt)
        log_conversation(user_id, query, qe, ctx, ans)
        return ans
    except Exception as e:
        logging.exception("RAG pipeline failed")
        return "Ошибка обработки запроса."

if __name__ == "__main__":
    user_id = "user123"
    query = "Каково видение StarRocks?"
    print("Ответ:\n", rag_pipeline(user_id, query))

Суммарный конвейер:

Ввод вопроса
Эмбеддинг (DeepSeek)
ANN‑поиск в enterprise_knowledge (StarRocks Vector Index)
Формирование prompt (Augmentation)
Генерация ответа (DeepSeek)
Логирование (customer_service_log)
Возврат результата пользователю

6.2 Веб‑интерфейс (минимальный)

<!DOCTYPE html>
<html lang="ru">
<head>
  <meta charset="UTF-8" />
  <title>Интеллектуальная система вопросов и ответов</title>
  <meta name="viewport" content="width=device-width,initial-scale=1" />
  <script>
    async function askQuestion() {
      const question = document.getElementById("question").value;
      const resp = await fetch("/ask", {
        method: "POST",
        headers: {"Content-Type": "application/json"},
        body: JSON.stringify({ question })
      });
      const data = await resp.json();
      document.getElementById("answer").innerText = data.answer;
    }
  </script>
</head>
<body>
  <h1>Интеллектуальная система вопросов и ответов</h1>
  <input id="question" type="text" placeholder="Введите ваш вопрос" />
  <button onclick="askQuestion()">Спросить</button>
  <pre id="answer"></pre>
</body>
</html>

6.3 Полноценный бэкенд Q&A (Flask)

import pymysql, requests, json, logging, re
from flask import Flask, request, jsonify, render_template_string

app = Flask(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

INDEX_HTML = """<!DOCTYPE html><html lang="ru"><head>
<meta charset="UTF-8"><title>Интеллектуальная система вопросов и ответов</title>
<meta name="viewport" content="width=device-width,initial-scale=1" />
<script>
async function askQuestion(){
  const q=document.getElementById('question').value;
  const r=await fetch('/ask',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({question:q})});
  const d=await r.json(); document.getElementById('answer').innerText=d.answer;
}
</script></head><body>
<h1>Интеллектуальная система вопросов и ответов</h1>
<input id="question" type="text" placeholder="Введите ваш вопрос" />
<button onclick="askQuestion()">Спросить</button>
<pre id="answer"></pre>
</body></html>"""

def get_embedding(text: str):
    url = "http://localhost:11434/api/embeddings"
    payload = {"model": "deepseek-r1:7b", "prompt": text}
    r = requests.post(url, json=payload, timeout=30)
    r.raise_for_status()
    return r.json()["embedding"]

def search_knowledge_base(query_embedding, topk=3, efsearch=128):
    qv = "[" + ",".join(map(str, query_embedding)) + "]"
    sql = f"""
    SELECT /*+ SET_VAR(ann_params='{{efsearch={efsearch}}}') */
           content,
           approx_l2_distance(embedding, {qv}) AS distance
    FROM enterprise_knowledge
    ORDER BY approx_l2_distance(embedding, {qv}) ASC
    LIMIT {topk}
    """
    conn = pymysql.connect(host="X.X.X.X", port=9030, user="root", password="sr123456", database="knowledge_base")
    try:
        with conn.cursor() as cur:
            cur.execute(sql)
            rows = cur.fetchall()
            return "\n\n---\n\n".join([r[0] for r in rows]) if rows else ""
    finally:
        conn.close()

def build_rag_prompt(query: str, retrieved_content: str) -> str:
    return f"""
[Системная инструкция] Вы — корпоративный ассистент. Отвечайте строго по контексту.
[Контекст]
{retrieved_content}

[Вопрос]
{query}
"""

def clean_response(text: str) -> str:
    return re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()

def generate_answer(prompt: str) -> str:
    url = "http://localhost:11434/api/generate"
    payload = {"model": "deepseek-r1:7b", "prompt": prompt}
    r = requests.post(url, json=payload, timeout=120)
    r.raise_for_status()
    full = ""
    for line in r.text.splitlines():
        if not line.strip():
            continue
        try:
            obj = json.loads(line)
            if "response" in obj: full += obj["response"]
            if obj.get("done", False): break
        except json.JSONDecodeError:
            continue
    return clean_response(full.strip())

def log_conversation(user_id, question, question_embedding, retrieved_content, generated_answer):
    emb = "[" + ",".join(map(str, question_embedding)) + "]"
    sql = """
    INSERT INTO customer_service_log (user_id, question, question_embedding, retrieved_content, generated_answer, timestamp)
    VALUES (%s, %s, %s, %s, %s, NOW())
    """
    conn = pymysql.connect(host="X.X.X.X", port=9030, user="root", password="sr123456", database="knowledge_base", autocommit=True)
    try:
        with conn.cursor() as cur:
            cur.execute(sql, (user_id, question, emb, retrieved_content, generated_answer))
    finally:
        conn.close()

def rag_pipeline(user_id: str, query: str) -> str:
    qe = get_embedding(query)
    ctx = search_knowledge_base(qe, topk=3, efsearch=128)
    prompt = build_rag_prompt(query, ctx)
    ans = generate_answer(prompt)
    log_conversation(user_id, query, qe, ctx, ans)
    return ans

@app.route("/")
def index():
    return render_template_string(INDEX_HTML)

@app.route("/ask", methods=["POST"])
def ask():
    data = request.get_json(force=True)
    question = data.get("question", "").strip()
    if not question:
        return jsonify({"answer": "Пожалуйста, введите вопрос."})
    answer = rag_pipeline("sr_01", question)
    return jsonify({"answer": f"Вопрос: {question}\n\nОтвет: {answer}"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=9033, debug=True)

Примечания по настройке векторного индекса StarRocks (HNSW/IVFPQ)

index_type: hnsw | ivfpq
dim: размерность эмбеддинга (например, 3584)
metric_type: l2_distance | cosine_similarity
HNSW‑параметры:
- M (двунаправленных связей на вершину, по умолчанию 16)
- efconstruction (размер кандидатов при построении графа, по умолчанию 40)
- efsearch (через /*+ SET_VAR(ann_params) */ при запросе; выше — точнее, медленнее)
IVFPQ‑параметры:
- nlist (число кластеров)
- nbits (точность PQ, кратно 8)
- M_IVFPQ (число субвекторов; делитель dim)
- Поисковые: nprobe, max_codes, polysemous_ht, range_search_confidence

Требования к запросам, чтобы индекс применился:

ORDER BY только по approx_l2_distance(колонка, константный_массив) или approx_cosine_similarity(...)
Константный массив — ARRAY с длиной = dim
Обязателен LIMIT N
Предикаты строить на той же функции и согласовывать с направлением сортировки (ASC для l2_distance, DESC для cosine_similarity)

Проверка:

EXPLAIN ... -- VECTORINDEX: ON в OlapScanNode

Справочные материалы

Vector Index (StarRocks): https://docs.starrocks.io/zh/docs/table_design/indexes/vector_index/
DeepSeek: https://platform.deepseek.com
Ollama: https://ollama.com/

построение интеллектуальной системы вопросов и ответов и корпоративной базы знаний на базе StarRocks + DeepSeek