结合MLflow注册表与LLM路由的领域感知RAG系统
领域感知型RAG：先对意图进行分类，从MLflow注册表中加载合适的嵌入模型，搜索匹配的Qdrant集合，随后将内容生成任务交给大语言模型处理。
来源: RAG Beyond Retrieval: Model Registry, Vector Search and Intelligent LLM Routing
大多数RAG演示仅连接一个嵌入模型、一个知识库集合和一个大型语言模型。当抵押贷款相关问题与医疗问题需使用相同处理流程，或者当分类任务只需小型本地模型即可完成，而复杂临床问题则需依赖大型远程模型时，这种设计就不够用了。更为完善的方案是首先识别用户意图，选定对应的知识库和嵌入模型，获取相关上下文，再将生成任务分配给合适的大型语言模型。
该实现是在较为简单的硬件上运行这一方案的：使用树莓派上的MLflow模型注册系统来管理嵌入模型，为不同领域配置Qdrant知识库集合，借助Ollama进行本地意图分类，对于需要的高级功能则选择性地调用云端大型语言模型。此外，还需解决树莓派如何连接Postgres数据库的网络问题，因为只有后端存储可访问时，模型注册系统才能发挥作用。
从树莓派连接Postgres
如果在同一局域网的 Mac 上通过 Homebrew 安装了 Postgres，在编辑之前请先确认正在运行的实例使用的是哪种配置：
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 = '*'
在 pg_hba.conf 中将 Pi 的 IP 加入白名单，以便允许该主机进行密码认证：
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;"
在引入 MLflow 之前，先测试从 Pi 直接连接的可行性：
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"
两阶段架构
第一阶段——知识层。领域相关的 PDF 文件会被分割成多个片段，并嵌入特定领域的句子转换模型，随后存储在独立的 Qdrant 集合中。MLflow 会记录每个领域中哪种嵌入模型表现最佳，这样各服务就不必硬编码本地路径。
第二阶段——智能路由。分类器会对每个用户问题进行标注（在本演示中为白血病、抵押贷款或其他类型）。检索时会使用对应的集合和嵌入模型；生成内容则会根据问题意图选择相应的大型语言模型——某些路径使用本地模型，其他路径则使用远程模型——从而让成本和功能与问题类型相匹配。
实现模块
model_registry.py负责为抵押贷款任务注册Arctic-embed模型，为白血病分析任务注册MiniLM模型（或用户选择的任意组合），并将这些模型信息记录到Pi设备上的MLflow中，同时为各模型指定别名：
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通过别名（models:/...@champion）从MLflow中加载嵌入模型，这样在线运行时的路径就不必依赖开发者的笔记本路径：
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，但这证明了注册表可以部署在靠近实验室网络的受控硬件上。
操作注意事项
在将 Postgres 暴露于局域网之前，务必先确保其安全。应将 pg_hba 的修改视为正式的生产环境变更来处理。锁定 MLflow 及驱动程序的版本。监控分类器的错误率：错误的意图会导致查询被发送到错误的集合，从而显得像是“性能不佳的 LLM”。需分别评估检索和路由功能。当注册表中的配置发生变更时，在切断实时流量之前，应先进行模拟数据导入或影子嵌入对比测试。
总结
超越单一检索路径的 RAG 技术本质上属于产品架构范畴：需明确应用领域，采用合适的检索方式，并设定合理的计算资源限制。MLflow、Qdrant 以及一个小型网关共同构成了可在普通办公环境中运行的实用架构，同时具备足够的规范性以支持从演示阶段走向实际应用。
强化基于注册表的路径安全
冠军模型别名应通过测试阶段。需记录每个答案所使用的模型版本ID，以便在注册表变更后重现问题。如果服务器离线，客户端应采用强制关闭模式，而非默默回退到某个随机本地检查点。
分块参数应与嵌入器一同存储在注册表元数据中。具有不同默认上下文长度的新冠军模型，若没有重建标志，则不得重复使用旧的分块数据。
分类器提示语需要标准化配置：十种抵押贷款相关的表述、十种临床相关表述、十种对抗性“其他”类型表述。在调整大语言模型之前，应先测量其精确度与召回率。否则你将在路由机制才是真正缺陷的情况下，一味追求生成质量。
在基于 Raspberry Pi 的部署中，要注意频繁写入模型文件对磁盘的磨损问题，建议将 MLflow 模型文件的存储位置设在外部硬盘上。Postgres 的备份文件也应保存在与 Raspberry Pi 不同的机器上。
向相关方解释设计思路时，可按以下流程说明：分类 → 选择嵌入器/集合 → 获取数据 → 选择大语言模型 → 带有引用依据地给出答案。这样的说明比单纯的工具列表更易理解，也能清晰展现决策依据所在。
扩展功能域时需谨慎行事。每个新标签都需要对应的集合、嵌入器负责人、数据导入任务、分类器示例以及大语言模型策略。网关应拒绝未知标签，而非默认使用最大的模型。
最后，要确保“其他”路径的真实性。与其从随机域名获取内容以避免尴尬的沉默，不如直接拒绝请求。用户更信任那些能够明确表明自身功能范围之外的系统。 在进一步调整生成器提示词之前，必须将路径准确性和内容检索质量作为独立的发布标准进行检测。 在进一步调整生成器提示词之前，必须将路径准确性和内容检索质量作为独立的发布标准进行检测。 在进一步调整生成器提示词之前，必须将路径准确性和内容检索质量作为独立的发布标准进行检测。 在进一步调整生成器提示词之前，必须将路径准确性和内容检索质量作为独立的发布标准进行检测。 在进一步调整生成器提示词之前，必须将路径准确性和内容检索质量作为独立的发布标准进行检测。 在进一步调整生成器提示词之前，必须将路径准确性和内容检索质量作为独立的发布标准进行检测。
提示词需进一步优化。在进一步调整生成器提示词之前，必须将路径准确率与检索质量作为独立的发布标准进行检测。在进一步调整生成器提示词之前，必须将路径准确率与检索质量作为独立的发布标准进行检测。在进一步调整生成器提示词之前，必须将路径准确率与检索质量作为独立的发布标准进行检测。在进一步调整生成器提示词之前，必须将路径准确率与检索质量作为独立的发布标准进行检测。在进一步调整生成器提示词之前，必须将路径准确率与检索质量作为独立的发布标准进行检测。在进一步调整生成器提示词之前，必须将路径准确率与检索质量作为独立的发布标准进行检测。在进一步调整生成器提示词之前，必须将路径准确率与检索质量作为独立的发布标准进行检测。路径在进一步调整生成器提示词之前，必须将准确率与检索质量作为独立的发布阈值进行检测。在进一步调整生成器提示词之前，必须将路线准确度与检索质量作为独立的发布标准进行衡量。在进一步调整生成器提示之前，必须先通过相应的发布阈值来检测路径准确性与检索质量。在进一步调整生成器提示之前，必须先通过相应的发布阈值来检测路径准确性与检索质量。在进一步调整生成器提示之前，必须先通过相应的发布阈值来检测路径准确性与检索质量。在进一步调整生成器提示之前，必须先通过相应的发布阈值来检测路径准确性与检索质量。在进一步调整生成器提示之前，必须先通过相应的发布阈值来检测路径准确性与检索质量。在进一步调整生成器提示之前，必须先通过相应的发布阈值来检测路径准确性与检索质量。在进一步调整生成器提示之前，必须先通过相应的发布阈值来检测路径准确性与检索质量。在进一步调整生成器提示之前，必须先通过相应的发布阈值来检测路径准确性与检索质量。需进一步调整TS参数。在进一步优化生成器提示词之前，必须将路径准确率与检索质量作为独立的发布阈值进行检测。在进一步优化生成器提示词之前，必须将路径准确率与检索质量作为独立的发布阈值进行检测。在进一步优化生成器提示词之前，必须将路径准确率与检索质量作为独立的发布阈值进行检测。