尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

从Pragmatik Labs看大模型工程化落地:RAG系统搭建与实战

从Pragmatik Labs看大模型工程化落地:RAG系统搭建与实战 林俊旸官宣创业公司 Pragmatik Labs 的消息在技术圈里引起的讨论其实比一般创业新闻更多一层深意。作为长期关注大模型工程化落地的人我更关心的是这支团队会用什么样的技术路线去兑现“务实主义”这个命名里隐含的承诺。本文不聊八卦也不做商业预测而是从 Pragmatik Labs 这个名字和创始团队的技术背景出发梳理当前大模型应用落地最值得关注的几个工程方向技术栈选型、RAG 系统搭建、数据闭环、模型评估、生产环境运维。文章会给出可复用的代码示例、配置思路和排错清单方便想在大模型方向深入的同学直接参考。1. 事件背景Pragmatik Labs 意味着什么1.1 先从创始人背景看起林俊旸在深度学习框架和 AI 基础设施领域有深厚积累尤其在分布式训练、模型底层优化方向上有过不少工程实践。这类背景的创始人出来做 AI 应用层或模型层创业通常会更重视“可落地性”而不是“纯论文创新”。因此Pragmatik Labs 这个名字非常值得玩味。Pragmatik 与英文 pragmatic 同源强调实用主义、务实主义。结合大模型行业的现状这个命名传递出的信号很一致不追求宏大叙事优先解决真实业务问题。1.2 务实主义与 LLM 工程化的关系当前大模型行业的一个典型矛盾是模型能力提升很快但真正能跑通业务闭环的团队并不多。很多项目停在 Demo 阶段原因不是模型不够聪明而是工程链路没有打通。Pragmatik Labs 强调的务实主义刚好对应了 LLM 工程化的核心痛点模型选型要克制不是越大的模型越好推理成本要可控不能只追求单次效果数据质量比模型微调更值得投入评估体系必须前置否则根本无法迭代。换句话说Pragmatik Labs 切入的可能是这样一个方向把大模型从“能聊天”变成“能干活”并且让业务方算得清投入产出比。1.3 对开发者的启发不管 Pragmatik Labs 最终的产品方向是什么对普通开发者的启发是明确的大模型时代工程能力重新变得重要。过去一年我见过太多团队把精力花在“换更强模型”上而忽略了检索链路、提示词管理、数据管线、评估回归这些基础工程。Pragmatik Labs 这种从基础设施背景出来的团队反而更可能把力气花在这些“不性感但决定成败”的地方。2. 大模型应用落地务实的技术栈选型思路2.1 选型的第一原则先定义业务场景很多团队做技术选型时第一个问号就是“该用哪个模型”。这个顺序其实反了。正确顺序应该是业务场景 - 效果指标 - 成本预算 - 延迟要求 - 模型选型 - 架构设计举一个典型例子业务场景是客服知识库问答效果指标是答案准确率和引用可追溯率成本预算要求单次问答成本控制在 0.01 元以内延迟要求是用户可感知 3 秒内返回。在这个约束下你可能不需要微调一个 700 亿参数的大模型而是用一个中等规模的模型配合 RAG 检索就能满足需求。2.2 模型服务化与推理引擎模型选型确定之后下一步是推理服务化。常见方案有 vLLM、TGIText Generation Inference、SGLang 等。以 vLLM 为例它的核心优势是 PagedAttention 显存管理机制可以在同样显存条件下服务更长的上下文、支持更高的并发。部署方式也比较标准可以通过 OpenAI 兼容接口对外提供服务。# 使用 vLLM 启动一个 OpenAI 兼容服务 python -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen2.5-7B-Instruct \ --served-model-name qwen2.5-7b \ --port 8000 \ --gpu-memory-utilization 0.9 \ --max-model-len 8192启动后客户端可以通过标准 OpenAI SDK 访问# 文件路径test_vllm.py from openai import OpenAI client OpenAI( base_urlhttp://localhost:8000/v1, api_keyEMPTY, ) resp client.chat.completions.create( modelqwen2.5-7b, messages[ {role: system, content: 你是一个运维知识库助手。}, {role: user, content: 请介绍一下容器健康检查的配置方式。}, ], temperature0.2, max_tokens512, ) print(resp.choices[0].message.content)注意具体模型名称、参数默认值会随着 vLLM 版本变化实际部署时以你本地的版本和模型为准。2.3 向量数据库选型RAG 架构中向量数据库负责存储文档向量并执行相似度检索。常见选项有面向生产、分布式的Milvus、Qdrant、Weaviate轻量级、适合原型验证的Chroma、FAISS复用现有关系库的pgvector。选型时不要只看性能测试报告要重点考察是否有成熟的监控体系、数据备份方案、权限管理机制、客户端 SDK 是否活跃维护。2.4 应用编排层应用编排层建议用轻量方案避免引入过重框架。目前 LangChain 和 LlamaIndex 都在快速迭代但它们的抽象层更新频繁长期维护成本其实不低。我的建议是业务简单时直接用原生 Python 写流程控制把检索、拼接提示词、调用模型、解析输出封装成几个函数。业务复杂时再引入框架并且一定要把框架版本锁死。3. 核心原理RAG 系统的四段式拆解3.1 为什么 RAG 是当前最务实的技术路线RAGRetrieval-Augmented Generation检索增强生成的思路是在模型生成答案之前先从外部知识库中检索相关内容作为上下文提供给模型。它解决的核心问题是幻觉问题。因为模型不再纯粹依赖训练时学到的参数化记忆而是依赖检索到的实时内容进行生成。对于企业知识库、产品文档、客服问答这类场景RAG 是当前性价比最高的方案。微调的优势在于改变模型的行为风格和领域能力但成本高、迭代慢。RAG 的优势在于知识更新快、可控性强、可追溯更适合业务知识频繁变化的场景。3.2 索引阶段文档加载与切分RAG 的第一步是把原始文档处理成可检索的向量索引。文档加载要考虑格式多样性包括 PDF、Word、Markdown、HTML 等。切分策略直接影响检索质量。# 文件路径rag_pipeline/indexing.py from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter loader PyPDFLoader(./docs/产品手册.pdf) documents loader.load() text_splitter RecursiveCharacterTextSplitter( chunk_size512, chunk_overlap64, separators[\n\n, \n, 。, , , ., !, ?, , ;, , ,], ) chunks text_splitter.split_documents(documents) print(f文档共切分为 {len(chunks)} 个切片)切分参数中chunk_size 代表每个文本块的最大字符数chunk_overlap 代表相邻块之间的重叠字符数。重叠的目的是避免一句话被拦腰切断保证语义完整性。3.3 检索阶段向量召回与重排序向量召回解决的是语义相似问题但向量检索的结果往往夹杂噪声。行业实践是在向量召回之后增加一个重排序环节。# 文件路径rag_pipeline/retriever.py from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-v2-m3, use_fp16True) def rerank(query: str, candidates: list[str], top_k: int 3) - list[str]: pairs [(query, doc) for doc in candidates] scores reranker.compute_score(pairs) doc_scores list(zip(candidates, scores)) doc_scores.sort(keylambda x: x[1], reverseTrue) return [doc for doc, _ in doc_scores[:top_k]]重排序模型的作用是让真正和问题相关的文本排到前面减少噪声对生成阶段的影响。建议在向量召回阶段多召回一些候选比如 top_k20再通过重排序压缩到 top_k3。3.4 生成阶段提示词拼接与输出约束生成阶段的关键是提示词质量。RAG 提示词需要明确约束模型的职责强调只依据检索内容作答。# 文件路径rag_pipeline/generator.py SYSTEM_PROMPT 你是一个严谨的文档助手。请严格基于以下检索片段回答用户问题。 要求 1. 如果检索片段中没有答案请明确回答“根据现有资料无法回答”。 2. 不要编造检索片段中不存在的信息。 3. 回答时用简洁的语言组织正文需要时给出条目化说明。 4. 在回答末尾列出参考片段序号。 检索片段 {context} 用户问题{question} 这里要特别注意即使提示词里写了“请严格基于检索片段回答”模型仍有可能自由发挥。因此有条件时建议在输出端增加规则校验或者使用结构化输出格式做二次解析。4. 完整实战从零搭建一个 RAG 问答服务4.1 项目结构设计rag-demo/ ├── app.py # FastAPI 应用入口 ├── requirements.txt # 依赖清单 ├── rag_pipeline/ │ ├── __init__.py │ ├── indexing.py # 文档加载与切分 │ ├── vector_store.py # 向量库封装 │ ├── retriever.py # 检索与重排序 │ └── generator.py # 提示词与生成 ├── docs/ # 原始文档目录 └── .env.example # 环境变量示例4.2 添加依赖# 文件路径requirements.txt fastapi0.115.6 uvicorn0.30.6 openai1.51.0 langchain0.3.7 langchain-community0.3.7 pypdf5.1.0 sentence-transformers3.2.1 FlagEmbedding1.2.10 qdrant-client1.12.1 python-dotenv1.0.1需要说明的是FlagEmbedding 的安装依赖 torch 和 CUDA 环境建议先在虚拟环境中单独验证安装避免污染全局环境。4.3 编写向量库封装# 文件路径rag_pipeline/vector_store.py import os from dotenv import load_dotenv from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams from sentence_transformers import SentenceTransformer load_dotenv() QDRANT_URL os.getenv(QDRANT_URL, http://localhost:6333) COLLECTION_NAME os.getenv(COLLECTION_NAME, documents) EMBED_MODEL_NAME os.getenv(EMBED_MODEL_NAME, BAAI/bge-m3) encoder SentenceTransformer(EMBED_MODEL_NAME) def get_client() - QdrantClient: return QdrantClient(urlQDRANT_URL) def create_collection() - None: client get_client() if not client.collection_exists(COLLECTION_NAME): client.create_collection( collection_nameCOLLECTION_NAME, vectors_configVectorParams( sizeencoder.get_sentence_embedding_dimension(), distanceDistance.COSINE, ), ) print(f集合 {COLLECTION_NAME} 创建成功) else: print(f集合 {COLLECTION_NAME} 已存在跳过创建) def upsert_documents(chunks: list) - None: client get_client() points [] for idx, chunk in enumerate(chunks): vector encoder.encode(chunk.page_content).tolist() points.append( { id: idx, vector: vector, payload: { text: chunk.page_content, source: chunk.metadata.get(source, ), }, } ) client.upsert(collection_nameCOLLECTION_NAME, pointspoints) print(f已写入 {len(points)} 条向量数据)4.4 编写 FastAPI 应用# 文件路径app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from rag_pipeline.generator import SYSTEM_PROMPT from rag_pipeline.retriever import rerank from rag_pipeline.vector_store import get_client, COLLECTION_NAME, encoder from openai import OpenAI import os from dotenv import load_dotenv load_dotenv() LLM_API_BASE os.getenv(LLM_API_BASE, http://localhost:8000/v1) LLM_API_KEY os.getenv(LLM_API_KEY, EMPTY) LLM_MODEL os.getenv(LLM_MODEL, qwen2.5-7b) app FastAPI(titleRAG 问答服务) class QueryRequest(BaseModel): question: str Field(..., min_length1, description用户问题) top_k: int Field(5, ge1, le20, description召回数量) class QueryResponse(BaseModel): answer: str sources: list[str] app.post(/query, response_modelQueryResponse) async def query(req: QueryRequest) - QueryResponse: try: client get_client() question_vector encoder.encode(req.question).tolist() search_result client.query_points( collection_nameCOLLECTION_NAME, queryquestion_vector, limitreq.top_k, ) candidates [hit.payload[text] for hit in search_result.points] if not candidates: raise HTTPException(status_code404, detail未检索到任何相关文档) best_docs rerank(req.question, candidates, top_k3) context \n\n.join( f[片段{i 1}] {doc} for i, doc in enumerate(best_docs) ) llm_client OpenAI(base_urlLLM_API_BASE, api_keyLLM_API_KEY) completion llm_client.chat.completions.create( modelLLM_MODEL, messages[ {role: system, content: SYSTEM_PROMPT.format( contextcontext, questionreq.question, )}, {role: user, content: req.question}, ], temperature0.2, max_tokens2048, ) return QueryResponse( answercompletion.choices[0].message.content, sources[doc[:200] for doc in best_docs], ) except HTTPException: raise except Exception as exc: raise HTTPException(status_code500, detailf服务内部错误: {exc}) from exc4.5 运行与验证先确保向量数据库已启动。如果是本地 Docker 方式docker run -d --name qdrant \ -p 6333:6333 \ -v $(pwd)/qdrant_storage:/qdrant/storage \ qdrant/qdrant然后执行索引脚本把文档写入向量库python -m rag_pipeline.indexing接着启动推理服务vLLM和问答服务python -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen2.5-7B-Instruct \ --served-model-name qwen2.5-7b \ --port 8000 uvicorn app:app --host 0.0.0.0 --port 8080最后用 curl 验证curl -X POST http://localhost:8080/query \ -H Content-Type: application/json \ -d {question: 产品的最大并发连接数是多少, top_k: 5}预期响应结构{ answer: 根据产品手册第 3 章最大并发连接数为 5000。, sources: [ 产品的最大并发连接数为 5000超过该值后系统将拒绝新的连接请求。 ] }5. 常见问题与排查清单问题现象常见原因解决思路启动 Qdrant 失败端口被占用或镜像版本问题检查端口监听情况指定不同端口查看容器日志定位异常向量维度不匹配更换了 embedding 模型但集合已存在删除集合重建或在创建集合时固定 embedding 模型检索结果质量差文档切分不合理或召回数量不足调整 chunk_size 和 chunk_overlap增加向量召回数量再通过重排序筛选模型回答与检索内容不一致提示词约束较弱或模型能力不足加强提示词约束检查检索内容是否真正有效覆盖问题API 请求超时模型服务并发压力大或网络连接异常检查推理服务监控指标适当调大超时时间考虑增加并发实例重新索引后内容未更新写入时未做去重或清理旧数据使用 upsert 前先删除 collection 或按 payload key 去重排查顺序建议先确认向量数据库状态再确认 embedding 服务正常接着检查召回内容是否相关最后才考虑模型生成问题。6. 生产环境最佳实践与工程建议6.1 数据闭环从评估到迭代务实主义的核心是数据闭环。不要等到上线后才发现问题而是从一开始就建设评估集。评估集至少包含以下几类问题有标准答案的简单事实问题需要跨多个片段归纳的复杂问题知识库中确实没有答案的问题边界模糊、容易产生歧义的问题。每次修改索引策略、提示词或模型版本后都要跑一遍回归测试。推荐用 Ragas 这类评估框架从忠实度Faithfulness、答案相关性Answer Relevance、上下文相关度Context Relevance等维度量化评估。6.2 提示词版本管理提示词是代码的一部分但项目里经常有人直接改线上 Prompt改完就忘了记录。建议把提示词作为独立文件管理纳入 Git 仓库。prompts/ ├── rag_system_v1.txt ├── rag_system_v2.txt └── changelog.md每次调整提示词时记录变更原因和评估结果方便追溯效果变化。6.3 缓存与性能优化对重复问题加缓存能显著降低成本。缓存 Key 建议用问题文本的哈希值同时设置合理的过期时间因为知识库内容会更新不能永久命中缓存。# 文件路径rag_pipeline/cache.py import hashlib import time from collections import OrderedDict class LRUCache: def __init__(self, capacity: int 1024, ttl: int 3600): self.capacity capacity self.ttl ttl self.cache OrderedDict() def get(self, key: str): if key not in self.cache: return None value, expire_at self.cache[key] if time.time() expire_at: del self.cache[key] return None self.cache.move_to_end(key) return value def set(self, key: str, value): self.cache[key] (value, time.time() self.ttl) self.cache.move_to_end(key) if len(self.cache) self.capacity: self.cache.popitem(lastFalse) cache LRUCache(capacity2048, ttl1800) def make_cache_key(question: str) - str: return hashlib.sha256(question.encode(utf-8)).hexdigest()6.4 安全边界与内容合规RAG 系统在生产环境面临的安全风险主要集中在提示词注入、权限绕过、敏感内容泄露三方面。建议至少做到对知识库文档做分类分级敏感内容单独隔离对用户输入做基本的异常检测识别可疑的注入模式系统提示词中不要拼接待执行的指令外部 API 调用必须走服务端不要把内网地址暴露给前端上线前准备内容安全过滤策略在模型输入和输出两侧都做检查。6.5 监控与可观测性RAG 服务不能只看接口延迟。建议至少监控以下指标召回相关度分布答案为空率来源引用比例LLM Token 消耗重排序后命中率。日志方面每一轮问答都应该记录问题原文、检索到的候选片段、重排序后的最终上下文、模型返回结果、耗时和成本估算。这些日志是后续优化的重要依据。7. 给技术人的学习路线与思考Pragmatik Labs 的成立某种程度上验证了一个趋势大模型行业正在从“模型竞赛”转向“工程竞赛”。真正稀缺的不是会调用 API 的人而是能搭建完整链路、控制成本、评估效果、定位问题的工程型人才。如果你也想沿着这个方向深入建议按以下路径学习掌握大模型基础原理重点理解 Transformer 结构、注意力机制、Token 化和生成策略熟悉推理服务部署掌握 vLLM 或类似引擎的部署、参数调优和并发评估方法深入 RAG 链路把文档加载、切分、向量化、检索、重排序、生成每一环都亲手实现一遍学习评估与数据闭环理解为什么评估集是决定系统迭代速度的关键了解生产运维包括容器化部署、监控告警、成本控制和安全防护。关于 Pragmatik Labs 后续会发布什么产品、采用什么模型架构目前公开信息还比较有限。但从团队背景和命名逻辑来看技术务实、工程落地、成本可控会是核心关键词。对于我们这些在一线写代码、调模型、修 Bug 的开发者来说与其关注一家创业公司的具体商业计划不如把注意力放在它验证的行业方向上大模型应用的胜负手最终还是要回到工程效率和数据质量上。这两件事做好了不管模型底座怎么换你的系统都能快速跟上。
返回列表