RAG із урахуванням домену та реєстрами MLflow та маршрутизацією LLM
RAG із урахуванням домену: класифікує намір, завантажує відповідний ембеддер з реєстру MLflow, шукає відповідну колекцію Qdrant, а потім направляє генерацію до LLM.
Джерела: RAG Beyond Retrieval: Model Registry, Vector Search and Intelligent LLM Routing
Більшість демонстрацій RAG підключають один модель ембеддингу, одну колекцію та один LLM. Це підходить, доки запитання щодо іпотек та медичні питання використовують один і той самий процес, або доки для первинної оцінки ситуації достатньо невеликої локальної моделі, тоді як для відповідей на деталі клінічного обстеження потрібна більша віддалена модель. Більш міцна схема спочатку класифікує намір користувача, вибирає відповідну базу знань та модель ембеддингу, отримує контекст, а потім направляє генерацію тексту до відповідного LLM.
У цій реалізації ця ідея працює на скромному обладнанні: MLflow Model Registry на Raspberry Pi для моделей ембеддингу, колекції Qdrant для кожної галузі, Ollama для локальної класифікації, а також вибіркові виклики хмарних LLM для тих галузей, де це необхідно. Проблеми з мережевим підключенням, пов’язані з підключенням Raspberry Pi до Postgres, також є частиною цієї схеми, адже реєстри моделей є корисними лише тоді, коли можливий доступ до бази даних на серверному рівні.
Підключення до Postgres з Raspberry Pi
Якщо Postgres було встановлено за допомогою Homebrew на Mac у мережі LAN, спочатку переконайтеся, яка конфігурація використовується активною інстанцією, перш ніж її редагувати:
psql -U <your_user> -d postgres -c "SHOW config_file;"
Адреси прослуховування та відповідні налаштування знаходяться у файлі postgresql.conf:
cd /opt/homebrew/var/postgresql@15
sed -i '' "s/#listen_addresses = 'localhost'/listen_addresses = '*'/" postgresql.conf
Перевірте фактичну конфігурацію:
grep -v "^#" postgresql.conf | grep listen_addresses
# listen_addresses = '*'
Додайте IP-адресу Pi до списку дозволених у файлі pg_hba.conf, щоб автентифікація за паролем була дозволена з цього хоста:
echo "host all all 192.168.1.9/32 md5" >> pg_hba.conf
Переконайтеся, що правила завантажуються так, як ви очікуєте:
grep "192.168.1.9" pg_hba.conf
Перезапустіть Postgres та переконайтеся, що він прослуховує на необхідних інтерфейсах:
brew services restart postgresql@15
lsof -i -P | grep LISTEN | grep postgres
Переконайтеся, що існує роль із правами входу:
psql -U <admin_user> -d postgres -c "\du"
Встановіть або скиньте пароль, який буде використовуватися MLflow:
psql -U <admin_user> -d postgres -c "ALTER USER root WITH PASSWORD 'root';"
Переконайтеся, що цільова база даних існує:
psql -U <admin_user> -d postgres -c "\l" | grep mlflow
psql -U <admin_user> -d postgres -c "CREATE DATABASE mlflow OWNER root;"
Перевірте пряме підключення з Pi перед тим, як використовувати MLflow:
nc -zv 192.168.1.8 5432
Коли мережевий шлях та облікові дані працюють, запустіть MLflow щодо цього віддаленого сервера зберігання:
mlflow server \
--backend-store-uri postgresql://root:root@192.168.1.8:5432/mlflow \
--host 0.0.0.0 \
--port 5000 \
--allowed-hosts "192.168.1.9:5000" \
--cors-allowed-origins "http://192.168.1.9:5000"
Архітектура у два етапи
Етап 1 — шар знань. PDF-файли домену розділяються на частини та інтегруються з моделями sentence-transformer, спеціалізованими для цього домену, після чого зберігаються у окремих колекціях Qdrant. MLflow відстежує, яка версія моделі інтеграції є найкращою champion для кожного домену, щоб сервіси не використовували жорстко закодовані локальні шляхи.
Етап 2 — інтелектуальне маршрутизування. Класифікатор присвоює мітку кожному запиту користувача (для цього демо-продукту: лейкемія, іпотека чи щось інше). Для пошуку використовується відповідна колекція та модель інтеграції. Для генерації використовується LLM, обраний для цієї мети — локальний для деяких шляхів, віддалений для інших — тож витрати та можливості залежать від запиту.
Модулі реалізації
model_registry.py реєструє Arctic-embed для кредитів на житло та MiniLM для лейкемії (або будь-яку іншу пару, яку ви оберете), записує їх у MLflow на пристрої Pi та створює псевдоніми для переможних моделей:
import mlflow
import mlflow.sentence_transformers
from mlflow import MlflowClient
from mlflow.models import infer_signature
from sentence_transformers import SentenceTransformer
mlflow.set_tracking_uri("http://192.168.1.9:5000")
mlflow.set_experiment("embedding-models")
EMBEDDERS = {
"mortgage-embedder": {"source": "Snowflake/snowflake-arctic-embed-m-v1.5"},
"leukemia-embedder": {"source": "all-MiniLM-L6-v2"},
}
sample_texts = ["MLflow tracks embedding models", "Qdrant stores vectors"]
client = MlflowClient()
for name, cfg in EMBEDDERS.items():
model = SentenceTransformer(
cfg["source"],
trust_remote_code=cfg.get("trust_remote_code", False),
config_kwargs=cfg.get("config_kwargs"),
device=cfg.get("device"),
)
with mlflow.start_run(run_name=name):
info = mlflow.sentence_transformers.log_model(
model=model,
name="embedder",
signature=infer_signature(sample_texts, model.encode(sample_texts)),
input_example=sample_texts,
registered_model_name=name,
)
client.set_registered_model_alias(name, "champion", info.registered_model_version)
print(f"{name}: v{info.registered_model_version}, dim={model.get_embedding_dimension()}")
ingest_data.py відновлює колекції даних за доменами, розділяє PDF-файли на частини з однаковим вікном токенів та рівнем перекриття, використовує переможну модель для цього домену для генерації ембеддингів та вносить ці дані до Qdrant:
import os
from chonkie import Pipeline
from qdrant_client import QdrantClient
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
client = QdrantClient(url="http://localhost:6333", api_key=os.environ["QDRANT_API_KEY"])
client.delete_collection("leukemia_details") # this will delete and recreate the collection.
(Pipeline()
.fetch_from("file", dir="../", ext=[".pdf"])
.process_with("liteparse")
.chunk_with("recursive", chunk_size=512)
.refine_with("overlap", context_size=100)
.store_in("qdrant", collection_name="leukemia_details", client=client,
embedding_model="sentence-transformers/all-MiniLM-L6-v2")
.run())
count = client.count("leukemia_details", exact=True).count
print(f"Ingested {count} chunks into leukemia_details")
ai_gateway.py класифікує намір користувача за допомогою суворо визначеного локального запиту через Ollama, а потім обирає відповідні бекенди LLM для кожної класифікаційної категорії:
from openai import OpenAI
import mlflow
mlflow.set_tracking_uri("http://192.168.1.9:5000")
client = OpenAI(
base_url="http://192.168.1.9:5000/gateway/mlflow/v1",
api_key="no-api-key",
)
CLASSIFIER_PROMPT = (
"Classify the user query into exactly one label: LEUKEMIA, MORTGAGE, or OTHER.\n"
"LEUKEMIA = questions about leukemia (AML, CML, ALL, CLL), its symptoms, diagnosis, "
"treatment, prognosis, or related blood cancers.\n"
"MORTGAGE = questions about home loans, mortgage rates, refinancing, EMI, "
"eligibility, lenders, or mortgage documents.\n"
"OTHER = everything else, including general medical questions not about leukemia.\n"
"Reply with only the label."
)
ROUTES = {
"LEUKEMIA": "openai",
"MORTGAGE": "gemma",
}
REJECT_MESSAGE = (
"Sorry, this setup can only answer Leukemia-specific medical questions "
"and mortgage-related questions."
)
def classify(query: str) -> str:
resp = client.chat.completions.create(
model="ollama", # local Ollama endpoint does the classification
messages=[
{"role": "system", "content": CLASSIFIER_PROMPT},
{"role": "user", "content": query},
],
temperature=0,
)
label = resp.choices[0].message.content.strip().upper()
for intent in ROUTES:
if intent in label:
return intent
return "OTHER"
def route(query: str) -> str:
intent = classify(query)
target = ROUTES.get(intent)
if target is None:
print(f"[intent={intent} -> rejected]")
return REJECT_MESSAGE
resp = client.chat.completions.create(
model=target,
messages=[{"role": "user", "content": query}],
)
print(f"[intent={intent} -> model={target}]")
return resp.choices[0].message.content
#
# if __name__ == "__main__":
# print(route("What are the first-line treatments for acute myeloid leukemia?"))
# print(route("How does refinancing affect my monthly mortgage EMI?"))
# print(route("What are the side effects of metformin?")) # medical but not leukemia -> rejected
# print(route("Write a haiku about monsoon in Hyderabad.")) # rejected
smart_rag.py поєднує класифікацію, вибір колекцій даних, пошук найкращих результатів (top-k) та формування відповідей на основі цих даних:
from qdrant_client import QdrantClient
import os
from ai_gateway import client, classify, ROUTES, REJECT_MESSAGE
from embedding_model_inference import EMBEDDERS
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
COLLECTIONS = {
"LEUKEMIA": {"collection": "leukemia_details", "embedder": "leukemia-embedder"},
"MORTGAGE": {"collection": "mortgage", "embedder": "mortgage-embedder"},
}
TEXT_FIELD = "text"
EXIT_WORDS = {"bye", "exit", "quit"}
class SmartRAG:
def __init__(self, qdrant_url="http://localhost:6333", top_k=5):
self.qdrant = QdrantClient(url=qdrant_url, api_key=os.environ.get("QDRANT_API_KEY"))
self.top_k = top_k
def retrieve(self, query, intent):
cfg = COLLECTIONS[intent]
embedder = EMBEDDERS[cfg["embedder"]]
prompt = "query" if "query" in embedder.prompts else None # arctic needs its query prefix
vec = embedder.encode(query, prompt_name=prompt, normalize_embeddings=True).tolist()
hits = self.qdrant.query_points(cfg["collection"], query=vec, limit=self.top_k, with_payload=True).points
return "\n\n".join(str(h.payload.get(TEXT_FIELD, h.payload)) for h in hits)
def answer(self, query):
intent = classify(query)
if intent not in ROUTES:
return REJECT_MESSAGE
context = self.retrieve(query, intent)
resp = client.chat.completions.create(
model=ROUTES[intent],
messages=[
{"role": "system", "content": f"Answer using only this context:\n{context}"},
{"role": "user", "content": query},
],
)
print(f"[intent={intent} -> {COLLECTIONS[intent]['collection']} -> {ROUTES[intent]}]")
return resp.choices[0].message.content
def run(self):
while True:
query = input("\nyou> ").strip()
if query.lower() in EXIT_WORDS:
break
if query:
print("\nassistant>", self.answer(query))
if __name__ == "__main__":
SmartRAG().run()
embedding_model_inference.py завантажує ембеддери з MLflow за допомогою псевдонімів (models:/...@champion), тож онлайн-шляхи ніколи не залежать від шляхів на ноутбуці розробника:
import mlflow
import mlflow.sentence_transformers
mlflow.set_tracking_uri("http://192.168.1.9:5000")
EMBEDDERS = {
name: mlflow.sentence_transformers.load_model(f"models:/{name}@champion")
for name in ("leukemia-embedder", "mortgage-embedder")
}
QUERIES = {
"leukemia-embedder": "What are the first-line treatments for acute myeloid leukemia?",
"mortgage-embedder": "How does refinancing affect my monthly mortgage EMI?",
}
for name, embedder in EMBEDDERS.items():
vecs = embedder.encode([QUERIES[name]], normalize_embeddings=True)
print(f"{name}: shape={vecs.shape}, prompts={list(embedder.prompts) or None}")
Результати
Запит щодо іпотеки має отримувати інформацію з колекції даних про іпотеки за допомогою інструменту вбудовування та відповідь згідно з політикою ШІ для обробки запитів про іпотеки. Запит щодо лейкемії має вирішуватися згідно з клінічними протоколами. Запити, що не належать до цієї області, мають залишатися без відповіді або відповідати, не удаючи, ніби вдалося отримати інформацію з відповідної області.
/Users/pavanmantha/.local/bin/uv run /Users/pavanmantha/Pavans/PracticeExamples/DataScience_Practice/mlflow-tutorials/mlflow_on_edge/.venv/bin/python /Users/pavanmantha/Pavans/PracticeExamples/DataScience_Practice/mlflow-tutorials/mlflow_on_edge/src/smart_rag.py
Downloading artifacts: 100%|██████████| 16/16 [00:19<00:00, 1.19s/it]
2026/09/06 12:48:35 INFO mlflow.sentence_transformers: 'models:/leukemia-embedder@champion' resolved as 'mlflow-artifacts:/3/models/m-f568ee233dd748f997ba66763e41653b/artifacts'
Downloading artifacts: 100%|██████████| 1/1 [00:00<00:00, 76.69it/s]
Loading weights: 100%|██████████| 103/103 [00:00<00:00, 15587.15it/s]
Downloading artifacts: 100%|██████████| 16/16 [01:29<00:00, 5.60s/it]
2026/09/06 12:50:06 INFO mlflow.sentence_transformers: 'models:/mortgage-embedder@champion' resolved as 'mlflow-artifacts:/3/models/m-c4221c844dfb45bfa15c6a4ecde56920/artifacts'
Downloading artifacts: 100%|██████████| 1/1 [00:00<00:00, 27.17it/s]
Loading weights: 100%|██████████| 197/197 [00:00<00:00, 8855.57it/s]
[transformers] BertModel LOAD REPORT from: /var/folders/1x/36nz44_569501n4xs0px2_8m0000gn/T/tmpef4y6tj5/model.sentence_transformer
Key | Status |
--------------------+---------+-
pooler.dense.weight | MISSING |
pooler.dense.bias | MISSING |
Notes:
- MISSING: those params were newly initialized because missing from the checkpoint. Consider training on your downstream task.
leukemia-embedder: shape=(1, 384), prompts=['query', 'document']
mortgage-embedder: shape=(1, 768), prompts=['query', 'document']
/Users/pavanmantha/Pavans/PracticeExamples/DataScience_Practice/mlflow-tutorials/mlflow_on_edge/src/smart_rag.py:20: UserWarning: Api key is used with an insecure connection.
self.qdrant = QdrantClient(url=qdrant_url, api_key=os.environ.get("QDRANT_API_KEY"))
you> What are the phases of treatment for acute lymphoblastic leukemia?
[intent=LEUKEMIA -> leukemia_details -> openai]
assistant> The phases of treatment for acute lymphoblastic leukemia (ALL) are:
1. Induction - achieving remission.
2. Consolidation - eradicating residual disease with different agents to circumvent resistance.
3. Maintenance (in ALL) - prolonged low-dose therapy to suppress re-emergence of the clone.
you> What was 30-day Average SOFR referred to in Series issued before July 2023?
[intent=MORTGAGE -> mortgage -> gemma]
assistant> * Question: "What was 30-day Average SOFR referred to in Series issued before July 2023?"
* Constraint: Use *only* the provided context.
* Chunk 1: Discusses "Excess inclusion income", Residual Holders, taxable income, Trust REMIC taxable income, etc. No mention of SOFR or July 2023.
* Chunk 2: Discusses "Backup withholding", Non-U.S. Persons, Ginnie Mae Guaranty, State/Local/Foreign tax. No mention of SOFR or July 2023.
* Chunk 3: Discusses "Residual Holder's share of expenses", REITs, RICs, phantom income. No mention of SOFR or July 2023.
* Chunk 4: Discusses "Principal Distribution Amount", Ginnie Mae Project Loan Certificates, Underlying Certificates. No mention of SOFR or July 2023.
* Chunk 5: Discusses "OID on a Teaser Security", "Initial Distribution Period", "Excess Interest Payment", "Accrual of Original Issue Discount". No mention of SOFR or July 2023.
* I have scanned all five provided text nodes.
you>
Чому ця структура є цікавою
Реєстри дозволяють замінювати інструменти вбудовування без необхідності перезапуску програми для кожного клієнта. Колекції даних, створені окремо для кожної області, запобігають забрудненню даних між різними областями. Механізм направлення запитів забезпечує використання дорогих моделей лише для складних запитів. Для реалізації цього паттерну не обов’язково використовувати MLflow на пристрої Pi, але це демонструє, що реєстр може функціонувати на контрольованому обладнанні, розташованому поблизу лабораторної мережі.
Операційні застереження
Забезпечте безпеку Postgres перед тим, як розгорнути його в локальній мережі. Розглядайте зміни до pg_hba як зміни у продакшені. Фіксуйте версії MLflow та драйверів. Слідкуйте за показниками помилок класифікатора: неправильна інтенція може відправити запит у неправильну колекцію та створити враження „поганого LLM“. Оцінюйте процеси пошуку та маршрутизації окремо. Коли в реєстрі змінюються налаштування, проведіть тестове впровадження або порівняння з „тіньовим“ варіантом перед тим, як переключати трафік у промисловий режим.
Заключення
RAG, що виходить за межі одного шляху пошуку, — це переважно архітектура продукту: потрібно знати сферу застосування, отримувати дані з використанням правильної структури та генерувати результати з урахуванням відповідних обмежень. MLflow, Qdrant та невеликий шлюз роблять цю архітектуру достатньо конкретною для роботи на столі та достатньо дисциплінованою, щоб розвиватися за межами демонстраційного варіанта.
Зміцнення шляху, підтримуваного реєстром
Псевдоніми чемпіонів мають проходити через етап тестування. Фіксуйте ID версії моделі, яка формує кожну відповідь, щоб можна було повторно проаналізувати інциденти після змін у реєстрі. Якщо пристрій виходить з мережі, клієнтам потрібен режим аварійного закриття, а не тихе переходження на випадковий локальний пункт контролю.
Параметри часткового оброблення даних мають знаходитися у метаданах реєстру поруч із інструментом вбудовування. Новий чемпіон із іншою бажаною довжиною контексту не повинен використовувати старі частини даних без позначки про необхідність перебудови.
Запити до класифікатора потребують стандартизованих прикладів: десять варіантів формулювань щодо іпотеки, десять — у клінічній сфері та десять — у конфліктному форматі. Вимірюйте точність та відтворюваність результатів перед тим, як підбирати модель LLM. Інакше ви будете намагатися покращити якість генератора, тоді як справжньою проблемою є спосіб обробки даних.
Для роботи з Raspberry Pi стежте за зносом диска від частого запису артефактів та використовуйте зовнішні диски як основу для зберігання артефактів MLflow. Зберігайте резервні копії Postgres на іншій машині, ніж Pi.
При поясненні концепції зацікавленим сторонам наведіть послідовність дій: класифікація → вибір інструменту для вбудовування/колекції → отримання даних → вибір LLM → надання відповіді з посиланнями. Таке описування краще сприймається, ніж простий список інструментів, та чітко показує, де приймаються рішення щодо політики.
Розширюйте домени обережно. Кожна нова мітка потребує колекції, відповідальної особи за інструмент для вбудовування, завдання на обробку даних, прикладів для класифікатора та політики щодо LLM. Шлюз повинен відхиляти невідомі мітки, а не автоматично використовувати найбільшу модель.
Наостанок, необхідно зберігати чесність у використанні шляху „ІНШЕ“. Краще відмовити, ніж запозичувати дані з випадкового домену, щоб уникнути незручної тиші. Користувачі довіряють системам, які знають, коли вони не підходять для виконання завдання. Точність маршрутизації та якість отримання даних мають вимірюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршрутизації та якість отримання даних мають вимірюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршрутизації та якість отримання даних мають вимірюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршрутизації та якість отримання даних мають вимірюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршрутизації та якість отримання даних мають вимірюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршрутизації та якість отримання даних мають вимірюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора.
Підказки налаштовуються далі. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як налаштовувати підказки генератора далі. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як налаштовувати підказки генератора далі. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як налаштовувати підказки генератора далі. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як налаштовувати підказки генератора далі. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як налаштовувати підказки генератора далі. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як налаштовувати підказки генератора далі. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як налаштовувати підказки генератора далі. МаршрутТочність та якість пошуку мають вимірюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора.Їх необхідно оцінювати як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршруту та якість пошуку мають оцінюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршруту та якість пошуку мають оцінюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршруту та якість пошуку мають оцінюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршруту та якість пошуку мають оцінюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршруту та якість пошуку мають оцінюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршруту та якість пошуку мають оцінюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора. Точність маршруту та якість пошуку мають оцінюватися як окремі критерії випуску перед подальшою налаштуванням запитів до генератора.Блокування ASE має бути впроваджене до того, як будуть далі налаштовуватися запити генератора. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як будуть далі налаштовуватися запити генератора. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як будуть далі налаштовуватися запити генератора. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як будуть далі налаштовуватися запити генератора. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як будуть далі налаштовуватися запити генератора. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як будуть далі налаштовуватися запити генератора. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як будуть далі налаштовуватися запити генератора. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед тим, як будуть далі налаштовуватися запити генератора.ts налаштовуються далі. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед подальшою налаштуванням запитів генератора. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед подальшою налаштуванням запитів генератора. Точність маршруту та якість пошуку мають вимірюватися як окремі критерії випуску перед подальшою налаштуванням запитів генератора.