fastest-rag-milvus-groq 实战Milvus 二值向量检索 Groq 推理构建毫秒级 RAG 应用【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub导读本文基于 ai-engineering-hub 仓库中的 fastest-rag-milvus-groq 项目完整拆解一条以“低检索时延”为核心目标的 RAG 技术栈用 HuggingFace 模型产出稠密向量并做二值量化Binary Quantization存入 Milvus 向量库做二进制向量索引与 Hamming 距离检索再由 Groq 作为推理引擎驱动 Kimi K2 模型生成答案最后借助 Beam 完成 Serverless 部署。读完本文你将掌握二值向量检索的原理与落地代码、MilvusBINARY_VECTOR集合的建库与索引配置、Groq LLM 的接入方式以及一条从 PDF 上传到流式问答、并在本地/云端运行完整应用的实操路径。一、系统全景这条 RAG 技术栈为什么“快”该项目在 README.md 中给出了明确的设计目标构建检索时延 15ms的 RAG 应用。这一目标由四个核心组件协同实现组件在本项目中的角色代码位置LlamaIndex编排 RAG 应用提供文档加载、Embedding 与 LLM 封装rag.pyMilvus二进制向量BINARY_VECTOR的索引与存储rag.pyGroq推理引擎运行 MoonshotAI 的 Kimi K2moonshotai/kimi-k2-instructrag.pyBeam极速 Serverless 云端部署 Streamlit 应用start_server.py速度来自两条主线依据仓库源码 rag.py 实现推断检索侧稠密向量被符号化压缩为每个维度仅 1 bit 的二进制向量存储开销相对 float32 降至约 1/32Milvus 通过BIN_FLAT索引 HAMMING汉明距离直接按位计算相似度无需浮点乘法。生成侧Groq 作为高吞吐推理引擎驱动 LLM配合 LlamaIndex 的流式接口把首字延迟进一步摊薄。需要说明 15ms是项目 README 给出的目标指标实际数值与语料规模、索引类型本项目BIN_FLAT属于精确扫描、部署硬件本地 CPU 还是 Beam GPU强相关应以你所在环境的实测为准——本项目的 app.py 内置了毫秒级检索计时器可以直接量化验证。二、核心原理Embedding 的二值量化Binary Quantization二值量化是本项目检索侧提速的关键。它把 float32 的稠密向量变成仅含 0/1 的比特串。仓库在EmbedData中给出了完整实现模型默认为BAAI/bge-large-en-v1.5rag.pyclass EmbedData: def __init__(self, embed_model_nameBAAI/bge-large-en-v1.5, batch_size512): self.embed_model_name embed_model_name self.embed_model self._load_embed_model() # HuggingFaceEmbedding self.batch_size batch_size def _binary_quantize(self, embeddings): Convert float32 embeddings to binary vectors embeddings_array np.array(embeddings) binary_embeddings np.where(embeddings_array 0, 1, 0).astype(np.uint8) # Pack bits into bytes (8 dimensions per byte) packed_embeddings np.packbits(binary_embeddings, axis1) return [vec.tobytes() for vec in packed_embeddings] def embed(self, contexts): for batch_context in batch_iterate(contexts, self.batch_size): batch_embeddings self.generate_embedding(batch_context) # float32 self.embeddings.extend(batch_embeddings) binary_batch self._binary_quantize(batch_embeddings) # 二值化 self.binary_embeddings.extend(binary_batch)量化过程分两步符号化Sign 编码np.where(x 0, 1, 0)以 0 为阈值向量各维大于 0 记 1、否则记 0位打包np.packbits(binary_embeddings, axis1)沿向量维度方向把 8 个 bit 打包成 1 个 byte最终以bytes形式送入 Milvus这正是 MilvusBINARY_VECTOR字段所需的数据形态。代码中有两个值得注意的工程细节批处理batch_iterate按batch_size512分批生成 Embedding 与二进制向量避免一次处理整个语料造成内存压力模型缓存_load_embed_model()将模型缓存在./hf_cache目录rag.py首次运行需联网下载bge-large-en-v1.5之后可离线加载。值得一提的设计权衡符号化只用“正/负”信息而丢弃幅值因此对相似度精度是有损压缩。它换取的是 Milvus 端仅需按位比较的 Hamming 距离与约 32 倍的内存节省——这正是该项目换取检索速度的“代价”。对检索召回率要求极高的场景可在此基础上用重排序Re-ranking阶段补偿本项目未包含属于可自行扩展方向。三、向量库设计Milvus 二进制向量集合3.1 客户端与本地库文件项目使用MilvusClient(db_file)直接以本地文件的方式运行 Milvus Literag.py无需单独起 Milvus 服务非常适合单机/开发期验证def define_client(self): try: self.client MilvusClient(self.db_file) logger.info(fInitialized Milvus Lite client with database: {self.db_file}) except Exception as e: raise e在 app.py 中每个浏览器会话会生成独立的库文件milvus_{session_id}.db放在系统临时目录并在重新索引前先删除旧文件保证多会话数据隔离。3.2 集合 Schema 与索引参数MilvusVDB_BQ.create_collection()定义了二进制向量集合的完整结构rag.pyschema self.client.create_schema( auto_idTrue, enable_dynamic_fieldsTrue, ) schema.add_field(field_nameid, datatypeDataType.INT64, is_primaryTrue, auto_idTrue) schema.add_field(field_namecontext, datatypeDataType.VARCHAR, max_length65535) schema.add_field(field_namebinary_vector, datatypeDataType.BINARY_VECTOR, dimself.vector_dim) index_params self.client.prepare_index_params() index_params.add_index( field_namebinary_vector, index_namebinary_vector_index, index_typeBIN_FLAT, # Exact search for binary vectors metric_typeHAMMING # Hamming distance for binary vectors ) self.client.create_collection( collection_nameself.collection_name, schemaschema, index_paramsindex_params )关键参数逐项说明配置项取值含义与影响auto_idTrue主键id由 Milvus 自动生成插入时无需传 idenable_dynamic_fieldsTrue允许后续插入未预定义的字段contextVARCHAR(65535)存储原始文本片段检索时随结果一起取出binary_vectorBINARY_VECTOR二进制向量类型维度dim必须与量化前的 Embedding 维度一致8 的整数倍index_typeBIN_FLAT二进制向量的精确暴力扫描索引代码注释明确标注 “Exact search”metric_typeHAMMING相似度度量采用汉明距离数值越小表示两个比特串差异越少关于维度rag.py 默认vector_dim1024而 app.py 采用更稳妥的做法先对一条test文本取真实 Embedding 长度作为actual_dim再传入建库函数从根上避免维度不匹配的报错。3.3 批量写入ingest_data()将文本与二进制向量一一配对、逐批插入rag.pyfor batch_context, batch_binary_embeddings in zip( batch_iterate(embeddata.contexts, self.batch_size), batch_iterate(embeddata.binary_embeddings, self.batch_size) ): data_batch [ {context: context, binary_vector: binary_embedding} for context, binary_embedding in zip(batch_context, batch_binary_embeddings) ] self.client.insert(collection_nameself.collection_name, datadata_batch)注意binary_vector字段写入的是打包后的bytes对象_binary_quantize的返回值而非 0/1 矩阵或 float 数组这是 MilvusBINARY_VECTOR的输入约定。四、检索链路查询二值化 → Hamming 距离 → 相似度换算Retriever类负责把用户问题映射为同样的二进制空间再执行检索rag.pydef _binary_quantize_query(self, query_embedding): embedding_array np.array([query_embedding]) binary_embedding np.where(embedding_array 0, 1, 0).astype(np.uint8) packed_embedding np.packbits(binary_embedding, axis1) return packed_embedding[0].tobytes() def search(self, query, top_kNone): if top_k is None: top_k self.top_k # 默认 5 query_embedding self.embeddata.embed_model.get_query_embedding(query) binary_query self._binary_quantize_query(query_embedding) search_results self.vector_db.client.search( collection_nameself.vector_db.collection_name, data[binary_query], anns_fieldbinary_vector, search_params{metric_type: HAMMING, params: {}}, limittop_k, output_fields[context] ) formatted_results [] for result in search_results[0]: formatted_results.append({ id: result[id], score: 1.0 / (1.0 result[distance]), # 汉明距离 → 相似度 payload: {context: result[entity][context]}, }) return formatted_results四个要点值得展开查询也需同源量化get_query_embedding产生 float32 查询向量后必须用与建库阶段完全一致的 0 → 1阈值规则打包成 bytes否则检索结果无意义索引字段对齐anns_fieldbinary_vector指明在二进制向量字段上检索search_params中的度量必须是建索引时注册的HAMMING返回原文通过output_fields[context]让 Milvus 直接带回命中的文本片段省去二次查询相似度换算Milvus 返回的是汉明距离越小越相似项目用1.0 / (1.0 distance)将其转换成 “越大越相似” 的分数方便下游或 UI 直接展示。五、生成侧LlamaIndex Groq 驱动 Kimi K2检索到的片段由RAG类交给 Groq 上的 LLM 完成生成rag.py。初始化核心参数如下class RAG: def __init__(self, retriever, llm_modelmoonshotai/kimi-k2-instruct, groq_api_keyNone): self.llm_model llm_model self.groq_api_key groq_api_key or os.getenv(GROQ_API_KEY) self.llm self._setup_llm() self.retriever retriever self.prompt_template ( CONTEXT: {context}\n ---------------------\n Given the context information above I want you to think step by step to answer the users query in a crisp and concise manner. In case you dont know the answer simply say I dont know!. Dont try to make up an answer. Only answer based on facts and contextual information.\n QUERY: {query}\n ANSWER: ) def _setup_llm(self): if not self.groq_api_key: raise ValueError(Groq API key is required...) return Groq( modelself.llm_model, api_keyself.groq_api_key, temperature0.4, max_tokens1000 )需要特别说明的几处设计模型选择moonshotai/kimi-k2-instruct走 Groq 的推理端点API Key 通过GROQ_API_KEY环境变量或构造参数传入缺省时会抛出明确错误提示解码参数temperature0.4偏低、更保守稳定与max_tokens1000单次回复上限可结合自身场景调整防幻觉提示词Prompt 模板要求模型“基于事实与上下文回答、不知道就说 I dont know绝不编造”这是 RAG 落地中抑制幻觉的常见手段双入口 APIquery()走llm.stream_complete/llm.completechat_query()走llm.stream_chat/llm.chatUI 层选用stream_complete实现逐 token 流式输出。generate_context()会把top_k默认 5条检索结果按\n\n---\n\n拼接成单一上下文块rag.py再填入上述模板——检索与生成的衔接就在这里完成。六、前端交互Streamlit 应用与毫秒级计时app.py 用 Streamlit 把整套流程包装成可视化应用交互设计可总结为四步侧边栏输入 Groq API Key密码输入框支持从环境变量预填上传 PDF文件进入临时目录后由 LlamaIndex 的SimpleDirectoryReader(input_dir..., required_exts[.pdf], recursiveTrue)抽取文本app.py三步索引管线生成 Embedding进度 40%→ 创建二进制向量集合并写入 Milvus Lite60%→ 构建RAG查询引擎100%全程用st.progress反馈进度同一会话内已处理的文档存入st.session_state.file_cache避免重复索引流式对话 计时用户提问后走检索 → 组装提示词 →llm.stream_complete的流水线。其中计时逻辑是验证“快不快”的关键app.py# Measure retrieval time retrieval_start time.perf_counter() context_text query_engine.generate_context(queryprompt) retrieval_time time.perf_counter() - retrieval_start # ... stream LLM tokens and render ... retrieval_ms int(retrieval_time * 1000) st.caption(f⏱️ Retrieval time: {retrieval_ms} ms)这里测量的范围是generate_context()——即查询二值化 Milvus Hamming 检索 上下文拼接的纯检索阶段不含 LLM 生成因此该数字能直接刻画向量检索环节的时延用于验证 README 提出的毫秒级目标。其余 UI 细节还包括内置 PDF 的 base64 内联预览、每条会话独立的 Milvus 集合docs_{uuid8}与“Clear ↺”按钮重置对话并触发gc.collect()释放内存。七、环境准备与本地运行7.1 安装 uv 与依赖README 要求Python 3.11 及以上并推荐用 uv 管理项目与虚拟环境。先安装 uv# MacOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh # Windows powershell -ExecutionPolicy ByPass -c irm https://astral.sh/uv/install.ps1 | iex初始化项目并安装依赖# Create a new directory for our project uv init fastest-rag cd fastest-rag # Create virtual environment and activate it uv venv source .venv/bin/activate # MacOS/Linux .venv\Scripts\activate # Windows # Install dependencies uv add pymilvus llama-index llama-index-embeddings-huggingface llama-index-llms-groq streamlit beam-client依赖清单与本仓库实际 import 一一对应pymilvusMilvus 客户端、llama-index核心及其 HuggingFace Embedding / Groq LLM 扩展、streamlitUI、beam-client云端部署。若按仓库目录直接运行还需python-dotenv以读取.env中的密钥。7.2 配置 Groq API Key在 Groq 控制台申请 API Key 后写入项目根目录的.envGROQ_API_KEYYOUR_GROQ_API_KEY应用启动时通过load_dotenv()app.py自动读取也可以在页面侧边栏手动填入二者等价。7.3 本地启动streamlit run app.py浏览器打开后上传一个 PDF仓库fastest-rag-milvus-groq/docs/raft.pdf可作为测试样例即可看到 “生成 Embeddings → 创建向量索引 → 存入向量库” 的进度条随后进入流式问答界面并实时显示每次提问的检索毫秒数。八、一键上云Beam Serverless 部署为了让“毫秒级”在真实服务中可被体验项目还提供了 Beam 部署脚本 start_server.py把 Streamlit 应用直接发布到 Beam 云from beam import Image, Pod streamlit_server Pod( imageImage().add_python_packages([ streamlit, pymilvus, llama-index, llama-index-embeddings-huggingface, llama-index-llms-groq ]), ports[8501], # Default port for streamlit gpuT4, memory2Gi, entrypoint[streamlit, run, app.py], ) res streamlit_server.create() print(✨ Streamlit server hosted at:, res.url)脚本的资源配置要点如下便于按需调整配置取值说明ports[8501]Streamlit 默认监听端口gpuT4指定 T4 GPU当前被注释的cpu4表明也可切换为纯 CPU 规格memory2Gi容器内存配额entrypoint[streamlit, run, app.py]容器启动命令image.add_python_packages5 个 Python 包云端运行环境依赖部署步骤README 原文依次是# 1. 注册 Beam进入控制台默认 token 会自动生成然后在终端绑定 beam configure default --token YOUR_BEAM_TOKEN # 2. 执行部署脚本等待 Pod 创建 python start_server.py脚本执行成功后会在终端打印 Streamlit 服务的公网地址把生成的链接粘贴到浏览器即可直接访问部署在 Beam 上的问答应用。九、代码路径速览与进一步探索围绕本主题可以在仓库中继续阅读以下文件从“会跑”进阶到“懂实现”fastest-rag-milvus-groq/README.md项目目标、完整安装与部署命令fastest-rag-milvus-groq/rag.py二值量化、Milvus 集合建库、检索器与 RAG 引擎的核心实现重点看EmbedData、MilvusVDB_BQ、Retriever、RAG四个类fastest-rag-milvus-groq/app.pyStreamlit 交互、会话级集合隔离、检索计时与流式输出fastest-rag-milvus-groq/start_server.pyBeam Pod 的云部署声明fastest-rag-milvus-groq/docs/raft.pdf仓库内置的可直接用于测试的示例 PDF。值得自行验证与思考的边界BIN_FLAT是精确扫描源码注释明确其为 “Exact search for binary vectors”检索耗时随库内向量数量线性增长。当语料达到较大规模时可在 Milvus 二进制索引体系内进一步调研近似检索ANN索引方案来换取舍与速度的平衡二值量化是有损压缩阈值 0 的符号化丢弃了向量幅值信息换取约 32 倍存储缩减与按位运算。对精度敏感的评测集建议先用本项目内置的Retrieval time计时 人工问答验证召回质量时延目标存在前提 15ms是设计目标实测值受硬件Beam T4 / 本地 CPU、语料量与文本长度影响务必以应用内展示的毫秒数为准。综上这个项目提供了一条“Embedding 二值化 → Milvus 二进制索引 → Groq 快速生成”的低时延 RAG 参考实现。从本地 Streamlit 验证到 Beam 云上发布链路完整、代码量小是理解二进制向量检索落地与 LLM 快速推理协同的极佳起点。【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 fastest-rag-milvus-groq 实战：Milvus 二值向量检索 + Groq 推理，构建毫秒级 RAG 应用
fastest-rag-milvus-groq 实战Milvus 二值向量检索 Groq 推理构建毫秒级 RAG 应用【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub导读本文基于 ai-engineering-hub 仓库中的 fastest-rag-milvus-groq 项目完整拆解一条以“低检索时延”为核心目标的 RAG 技术栈用 HuggingFace 模型产出稠密向量并做二值量化Binary Quantization存入 Milvus 向量库做二进制向量索引与 Hamming 距离检索再由 Groq 作为推理引擎驱动 Kimi K2 模型生成答案最后借助 Beam 完成 Serverless 部署。读完本文你将掌握二值向量检索的原理与落地代码、MilvusBINARY_VECTOR集合的建库与索引配置、Groq LLM 的接入方式以及一条从 PDF 上传到流式问答、并在本地/云端运行完整应用的实操路径。一、系统全景这条 RAG 技术栈为什么“快”该项目在 README.md 中给出了明确的设计目标构建检索时延 15ms的 RAG 应用。这一目标由四个核心组件协同实现组件在本项目中的角色代码位置LlamaIndex编排 RAG 应用提供文档加载、Embedding 与 LLM 封装rag.pyMilvus二进制向量BINARY_VECTOR的索引与存储rag.pyGroq推理引擎运行 MoonshotAI 的 Kimi K2moonshotai/kimi-k2-instructrag.pyBeam极速 Serverless 云端部署 Streamlit 应用start_server.py速度来自两条主线依据仓库源码 rag.py 实现推断检索侧稠密向量被符号化压缩为每个维度仅 1 bit 的二进制向量存储开销相对 float32 降至约 1/32Milvus 通过BIN_FLAT索引 HAMMING汉明距离直接按位计算相似度无需浮点乘法。生成侧Groq 作为高吞吐推理引擎驱动 LLM配合 LlamaIndex 的流式接口把首字延迟进一步摊薄。需要说明 15ms是项目 README 给出的目标指标实际数值与语料规模、索引类型本项目BIN_FLAT属于精确扫描、部署硬件本地 CPU 还是 Beam GPU强相关应以你所在环境的实测为准——本项目的 app.py 内置了毫秒级检索计时器可以直接量化验证。二、核心原理Embedding 的二值量化Binary Quantization二值量化是本项目检索侧提速的关键。它把 float32 的稠密向量变成仅含 0/1 的比特串。仓库在EmbedData中给出了完整实现模型默认为BAAI/bge-large-en-v1.5rag.pyclass EmbedData: def __init__(self, embed_model_nameBAAI/bge-large-en-v1.5, batch_size512): self.embed_model_name embed_model_name self.embed_model self._load_embed_model() # HuggingFaceEmbedding self.batch_size batch_size def _binary_quantize(self, embeddings): Convert float32 embeddings to binary vectors embeddings_array np.array(embeddings) binary_embeddings np.where(embeddings_array 0, 1, 0).astype(np.uint8) # Pack bits into bytes (8 dimensions per byte) packed_embeddings np.packbits(binary_embeddings, axis1) return [vec.tobytes() for vec in packed_embeddings] def embed(self, contexts): for batch_context in batch_iterate(contexts, self.batch_size): batch_embeddings self.generate_embedding(batch_context) # float32 self.embeddings.extend(batch_embeddings) binary_batch self._binary_quantize(batch_embeddings) # 二值化 self.binary_embeddings.extend(binary_batch)量化过程分两步符号化Sign 编码np.where(x 0, 1, 0)以 0 为阈值向量各维大于 0 记 1、否则记 0位打包np.packbits(binary_embeddings, axis1)沿向量维度方向把 8 个 bit 打包成 1 个 byte最终以bytes形式送入 Milvus这正是 MilvusBINARY_VECTOR字段所需的数据形态。代码中有两个值得注意的工程细节批处理batch_iterate按batch_size512分批生成 Embedding 与二进制向量避免一次处理整个语料造成内存压力模型缓存_load_embed_model()将模型缓存在./hf_cache目录rag.py首次运行需联网下载bge-large-en-v1.5之后可离线加载。值得一提的设计权衡符号化只用“正/负”信息而丢弃幅值因此对相似度精度是有损压缩。它换取的是 Milvus 端仅需按位比较的 Hamming 距离与约 32 倍的内存节省——这正是该项目换取检索速度的“代价”。对检索召回率要求极高的场景可在此基础上用重排序Re-ranking阶段补偿本项目未包含属于可自行扩展方向。三、向量库设计Milvus 二进制向量集合3.1 客户端与本地库文件项目使用MilvusClient(db_file)直接以本地文件的方式运行 Milvus Literag.py无需单独起 Milvus 服务非常适合单机/开发期验证def define_client(self): try: self.client MilvusClient(self.db_file) logger.info(fInitialized Milvus Lite client with database: {self.db_file}) except Exception as e: raise e在 app.py 中每个浏览器会话会生成独立的库文件milvus_{session_id}.db放在系统临时目录并在重新索引前先删除旧文件保证多会话数据隔离。3.2 集合 Schema 与索引参数MilvusVDB_BQ.create_collection()定义了二进制向量集合的完整结构rag.pyschema self.client.create_schema( auto_idTrue, enable_dynamic_fieldsTrue, ) schema.add_field(field_nameid, datatypeDataType.INT64, is_primaryTrue, auto_idTrue) schema.add_field(field_namecontext, datatypeDataType.VARCHAR, max_length65535) schema.add_field(field_namebinary_vector, datatypeDataType.BINARY_VECTOR, dimself.vector_dim) index_params self.client.prepare_index_params() index_params.add_index( field_namebinary_vector, index_namebinary_vector_index, index_typeBIN_FLAT, # Exact search for binary vectors metric_typeHAMMING # Hamming distance for binary vectors ) self.client.create_collection( collection_nameself.collection_name, schemaschema, index_paramsindex_params )关键参数逐项说明配置项取值含义与影响auto_idTrue主键id由 Milvus 自动生成插入时无需传 idenable_dynamic_fieldsTrue允许后续插入未预定义的字段contextVARCHAR(65535)存储原始文本片段检索时随结果一起取出binary_vectorBINARY_VECTOR二进制向量类型维度dim必须与量化前的 Embedding 维度一致8 的整数倍index_typeBIN_FLAT二进制向量的精确暴力扫描索引代码注释明确标注 “Exact search”metric_typeHAMMING相似度度量采用汉明距离数值越小表示两个比特串差异越少关于维度rag.py 默认vector_dim1024而 app.py 采用更稳妥的做法先对一条test文本取真实 Embedding 长度作为actual_dim再传入建库函数从根上避免维度不匹配的报错。3.3 批量写入ingest_data()将文本与二进制向量一一配对、逐批插入rag.pyfor batch_context, batch_binary_embeddings in zip( batch_iterate(embeddata.contexts, self.batch_size), batch_iterate(embeddata.binary_embeddings, self.batch_size) ): data_batch [ {context: context, binary_vector: binary_embedding} for context, binary_embedding in zip(batch_context, batch_binary_embeddings) ] self.client.insert(collection_nameself.collection_name, datadata_batch)注意binary_vector字段写入的是打包后的bytes对象_binary_quantize的返回值而非 0/1 矩阵或 float 数组这是 MilvusBINARY_VECTOR的输入约定。四、检索链路查询二值化 → Hamming 距离 → 相似度换算Retriever类负责把用户问题映射为同样的二进制空间再执行检索rag.pydef _binary_quantize_query(self, query_embedding): embedding_array np.array([query_embedding]) binary_embedding np.where(embedding_array 0, 1, 0).astype(np.uint8) packed_embedding np.packbits(binary_embedding, axis1) return packed_embedding[0].tobytes() def search(self, query, top_kNone): if top_k is None: top_k self.top_k # 默认 5 query_embedding self.embeddata.embed_model.get_query_embedding(query) binary_query self._binary_quantize_query(query_embedding) search_results self.vector_db.client.search( collection_nameself.vector_db.collection_name, data[binary_query], anns_fieldbinary_vector, search_params{metric_type: HAMMING, params: {}}, limittop_k, output_fields[context] ) formatted_results [] for result in search_results[0]: formatted_results.append({ id: result[id], score: 1.0 / (1.0 result[distance]), # 汉明距离 → 相似度 payload: {context: result[entity][context]}, }) return formatted_results四个要点值得展开查询也需同源量化get_query_embedding产生 float32 查询向量后必须用与建库阶段完全一致的 0 → 1阈值规则打包成 bytes否则检索结果无意义索引字段对齐anns_fieldbinary_vector指明在二进制向量字段上检索search_params中的度量必须是建索引时注册的HAMMING返回原文通过output_fields[context]让 Milvus 直接带回命中的文本片段省去二次查询相似度换算Milvus 返回的是汉明距离越小越相似项目用1.0 / (1.0 distance)将其转换成 “越大越相似” 的分数方便下游或 UI 直接展示。五、生成侧LlamaIndex Groq 驱动 Kimi K2检索到的片段由RAG类交给 Groq 上的 LLM 完成生成rag.py。初始化核心参数如下class RAG: def __init__(self, retriever, llm_modelmoonshotai/kimi-k2-instruct, groq_api_keyNone): self.llm_model llm_model self.groq_api_key groq_api_key or os.getenv(GROQ_API_KEY) self.llm self._setup_llm() self.retriever retriever self.prompt_template ( CONTEXT: {context}\n ---------------------\n Given the context information above I want you to think step by step to answer the users query in a crisp and concise manner. In case you dont know the answer simply say I dont know!. Dont try to make up an answer. Only answer based on facts and contextual information.\n QUERY: {query}\n ANSWER: ) def _setup_llm(self): if not self.groq_api_key: raise ValueError(Groq API key is required...) return Groq( modelself.llm_model, api_keyself.groq_api_key, temperature0.4, max_tokens1000 )需要特别说明的几处设计模型选择moonshotai/kimi-k2-instruct走 Groq 的推理端点API Key 通过GROQ_API_KEY环境变量或构造参数传入缺省时会抛出明确错误提示解码参数temperature0.4偏低、更保守稳定与max_tokens1000单次回复上限可结合自身场景调整防幻觉提示词Prompt 模板要求模型“基于事实与上下文回答、不知道就说 I dont know绝不编造”这是 RAG 落地中抑制幻觉的常见手段双入口 APIquery()走llm.stream_complete/llm.completechat_query()走llm.stream_chat/llm.chatUI 层选用stream_complete实现逐 token 流式输出。generate_context()会把top_k默认 5条检索结果按\n\n---\n\n拼接成单一上下文块rag.py再填入上述模板——检索与生成的衔接就在这里完成。六、前端交互Streamlit 应用与毫秒级计时app.py 用 Streamlit 把整套流程包装成可视化应用交互设计可总结为四步侧边栏输入 Groq API Key密码输入框支持从环境变量预填上传 PDF文件进入临时目录后由 LlamaIndex 的SimpleDirectoryReader(input_dir..., required_exts[.pdf], recursiveTrue)抽取文本app.py三步索引管线生成 Embedding进度 40%→ 创建二进制向量集合并写入 Milvus Lite60%→ 构建RAG查询引擎100%全程用st.progress反馈进度同一会话内已处理的文档存入st.session_state.file_cache避免重复索引流式对话 计时用户提问后走检索 → 组装提示词 →llm.stream_complete的流水线。其中计时逻辑是验证“快不快”的关键app.py# Measure retrieval time retrieval_start time.perf_counter() context_text query_engine.generate_context(queryprompt) retrieval_time time.perf_counter() - retrieval_start # ... stream LLM tokens and render ... retrieval_ms int(retrieval_time * 1000) st.caption(f⏱️ Retrieval time: {retrieval_ms} ms)这里测量的范围是generate_context()——即查询二值化 Milvus Hamming 检索 上下文拼接的纯检索阶段不含 LLM 生成因此该数字能直接刻画向量检索环节的时延用于验证 README 提出的毫秒级目标。其余 UI 细节还包括内置 PDF 的 base64 内联预览、每条会话独立的 Milvus 集合docs_{uuid8}与“Clear ↺”按钮重置对话并触发gc.collect()释放内存。七、环境准备与本地运行7.1 安装 uv 与依赖README 要求Python 3.11 及以上并推荐用 uv 管理项目与虚拟环境。先安装 uv# MacOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh # Windows powershell -ExecutionPolicy ByPass -c irm https://astral.sh/uv/install.ps1 | iex初始化项目并安装依赖# Create a new directory for our project uv init fastest-rag cd fastest-rag # Create virtual environment and activate it uv venv source .venv/bin/activate # MacOS/Linux .venv\Scripts\activate # Windows # Install dependencies uv add pymilvus llama-index llama-index-embeddings-huggingface llama-index-llms-groq streamlit beam-client依赖清单与本仓库实际 import 一一对应pymilvusMilvus 客户端、llama-index核心及其 HuggingFace Embedding / Groq LLM 扩展、streamlitUI、beam-client云端部署。若按仓库目录直接运行还需python-dotenv以读取.env中的密钥。7.2 配置 Groq API Key在 Groq 控制台申请 API Key 后写入项目根目录的.envGROQ_API_KEYYOUR_GROQ_API_KEY应用启动时通过load_dotenv()app.py自动读取也可以在页面侧边栏手动填入二者等价。7.3 本地启动streamlit run app.py浏览器打开后上传一个 PDF仓库fastest-rag-milvus-groq/docs/raft.pdf可作为测试样例即可看到 “生成 Embeddings → 创建向量索引 → 存入向量库” 的进度条随后进入流式问答界面并实时显示每次提问的检索毫秒数。八、一键上云Beam Serverless 部署为了让“毫秒级”在真实服务中可被体验项目还提供了 Beam 部署脚本 start_server.py把 Streamlit 应用直接发布到 Beam 云from beam import Image, Pod streamlit_server Pod( imageImage().add_python_packages([ streamlit, pymilvus, llama-index, llama-index-embeddings-huggingface, llama-index-llms-groq ]), ports[8501], # Default port for streamlit gpuT4, memory2Gi, entrypoint[streamlit, run, app.py], ) res streamlit_server.create() print(✨ Streamlit server hosted at:, res.url)脚本的资源配置要点如下便于按需调整配置取值说明ports[8501]Streamlit 默认监听端口gpuT4指定 T4 GPU当前被注释的cpu4表明也可切换为纯 CPU 规格memory2Gi容器内存配额entrypoint[streamlit, run, app.py]容器启动命令image.add_python_packages5 个 Python 包云端运行环境依赖部署步骤README 原文依次是# 1. 注册 Beam进入控制台默认 token 会自动生成然后在终端绑定 beam configure default --token YOUR_BEAM_TOKEN # 2. 执行部署脚本等待 Pod 创建 python start_server.py脚本执行成功后会在终端打印 Streamlit 服务的公网地址把生成的链接粘贴到浏览器即可直接访问部署在 Beam 上的问答应用。九、代码路径速览与进一步探索围绕本主题可以在仓库中继续阅读以下文件从“会跑”进阶到“懂实现”fastest-rag-milvus-groq/README.md项目目标、完整安装与部署命令fastest-rag-milvus-groq/rag.py二值量化、Milvus 集合建库、检索器与 RAG 引擎的核心实现重点看EmbedData、MilvusVDB_BQ、Retriever、RAG四个类fastest-rag-milvus-groq/app.pyStreamlit 交互、会话级集合隔离、检索计时与流式输出fastest-rag-milvus-groq/start_server.pyBeam Pod 的云部署声明fastest-rag-milvus-groq/docs/raft.pdf仓库内置的可直接用于测试的示例 PDF。值得自行验证与思考的边界BIN_FLAT是精确扫描源码注释明确其为 “Exact search for binary vectors”检索耗时随库内向量数量线性增长。当语料达到较大规模时可在 Milvus 二进制索引体系内进一步调研近似检索ANN索引方案来换取舍与速度的平衡二值量化是有损压缩阈值 0 的符号化丢弃了向量幅值信息换取约 32 倍存储缩减与按位运算。对精度敏感的评测集建议先用本项目内置的Retrieval time计时 人工问答验证召回质量时延目标存在前提 15ms是设计目标实测值受硬件Beam T4 / 本地 CPU、语料量与文本长度影响务必以应用内展示的毫秒数为准。综上这个项目提供了一条“Embedding 二值化 → Milvus 二进制索引 → Groq 快速生成”的低时延 RAG 参考实现。从本地 Streamlit 验证到 Beam 云上发布链路完整、代码量小是理解二进制向量检索落地与 LLM 快速推理协同的极佳起点。【免费下载链接】ai-engineering-hubIn-depth tutorials on LLMs, RAGs and real-world AI agent applications.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-engineering-hub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考