尧图网页设计 
如果你已经做过 RAG 相关的 LLM 应用开发一定很熟悉这种场景文档加载了、向量化跑通了、模型也能回答几个问题了可一旦放进真实业务环境效果就开始失控。文档频繁变更、权限不断缺失、召回结果不稳定、回答缺少可验证的依据。最后团队真正花掉的时间几乎都消耗在数据治理、检索调优和系统稳定性上而不是“调大模型”本身。业界有一个直击要害的说法Connecting an LLM to Your Data Is the 21% Solution。直译就是把大语言模型接到你的数据上只是完成了 21% 的解决方案。剩下 79% 的工作藏在最开始觉得很“不重要”的地方——数据准备、清洗、切分、索引、检索优化、权限隔离、评估监控、反馈闭环以及配套的运维治理。这篇文章不打算介绍某个具体的商业产品也不重复“RAG 是什么”的入门科普。我会从工程视角拆解为什么连接数据只是 21% 的完整方案剩下的 79% 具体落在哪里。同时给出一套可以照着做的最小实现数据接入、分块、向量检索、生成、评估、问题排查以及落地时容易被忽略的最佳实践。正在做 LLMData 项目的开发者和技术负责人适合静下心读一遍。已经做过 RAG 原型但没上过生产的人也会在里面看到很多熟悉又扎心的细节。1. 这篇文章真正要解决的问题我见过太多团队把“接入 LLM”当作项目的核心工作量。计划排两周第一周接 API第二周做前端演示原型看起来非常顺利。但进入验收阶段问题接踵而至同一个问题换个说法就答错新导入的文档没有覆盖到旧知识不同部门的人访问同样的知识库却看到了一样的数据权限形同虚设线上模型偶发返回乱码查了一圈发现是推理服务混用了不同精度。这些问题的共同点是什么它们几乎都不在“模型”层而在“数据和工程”层。所以这篇文章真正要解决的问题是为什么“连接 LLM 到数据”在完整系统中只占 21%剩下的 79% 是一个什么样的工程结构一个最小可用的 LLM 数据应用代码该怎么写出、怎么验证、怎么排查从原型到生产哪些工作必须提前做哪些问题可以后置这不是一篇劝退文。恰恰相反理解 21% 这个数字能让你把有限的资源投入到真正决定项目成败的地方。模型永远在快速迭代API 调用成本也在下降但数据治理和系统工程带来的壁垒才是长期价值的来源。2. 基础概念与核心原理2.1 从“模型推理”到“数据产品”传统认知里LLM 应用最关键的是模型GPT-4o、Claude、Llama选个能力强的就完事。但现实中模型只是一个推理内核它需要被“喂养”正确的数据才能输出有价值的结果。做一个知识库问答产品真正运行的是一条完整的数据链路数据源 - 采集 - 清洗 - 分块 - 嵌入 - 索引 - 检索 - 重排 - 生成 - 校验 - 返回任何一个环节松动最终回答质量都会下降。多数团队把精力集中在“嵌入 检索 生成”这三个环节也就是大约 21% 的范围。而数据采集、清洗、更新、权限、评估、监控这些 79% 的部分往往被严重低估。2.2 21% 是一个判断不是一个精确统计严格来说21% 不是某个研究报告里测出来的精确数字它是一种工程判断在可演示的原型里模型接入和数据连接确实是主干但在可交付的生产系统里它只是第一公里。用一个类比理解修一条高速公路“通车”这件事看起来最重要但真正决定公路能不能长期安全运营的是路基、排水、护栏、标识、监控和管理制度。模型连接是“铺沥青”数据工程和系统工程才是“整条路的基础设施”。2.3 LLM、RAG、Agent、MCP 的关系这四个概念经常出现在同一个项目里容易混在一起。这里用一句话区分LLM大语言模型负责理解和生成是系统的“大脑”。RAG检索增强生成从外部数据中检索相关内容再交给 LLM 生成回答。它解决的是“模型不知道私有知识”的问题。Agent智能体让模型能调用外部工具、执行动作、多步规划。它解决的是“模型只能动嘴不能动手”的问题。MCP模型上下文协议提供一套统一的工具/数据接入协议让 Agent 与外部系统连接时不用为每个工具单独写适配器。一句话总结LLM 是内核RAG 管读数据Agent 管做事情MCP 管连接协议。近期讨论度很高的“llm wiki”范式本质上也是在回答同一个问题与其堆一套复杂的 RAG 管线不如先把知识整理成结构化的、可验证的 Wiki 形式再让模型基于它产出。它强调的仍然是数据侧的系统工程。2.4 21% 与 79% 的职责分解层面具体工作占比参考技术重心LLM 接入API 调用、模型选择、提示词模板、Function Calling约 10%接口设计、Prompt 工程数据连接向量化、向量库、基础检索约 11%Embedding、相似度检索数据工程采集、清洗、去重、分块、版本管理、更新策略约 35%数据管线、质量校验系统与产品工程权限隔离、评估回归、监控告警、缓存、成本治理、反馈闭环约 30%工程架构、SRE 方法论模型与部署治理推理服务精度选择、GPU 资源、灰度发布、可观测性约 14%推理优化、MLOps你会发现真正让项目“能用”和“好用”的几乎都落在 79% 里。这也是为什么很多 RAG 项目原型很快、上线很难。3. 环境准备与前置条件为了让后面的示例跑起来你需要准备一个标准 Python 环境。版本请以实际项目为准本文重点演示通用思路不绑定某个具体版本。3.1 基础运行环境操作系统Windows / macOS / Linux 均可Python建议 3.10 或更高版本包管理工具pip 或 poetry网络需要能访问 LLM API 服务3.2 依赖安装本文示例代码尽量精简依赖只用了三个库pip install requests python-dotenv numpyrequests调用 LLM 的 HTTP API比直接封装 SDK 更容易看清请求结构。python-dotenv读取.env文件中的环境变量避免把密钥写死在代码里。numpy做向量点积和归一化计算简单实现一个向量检索。如果你要上生产可以把numpy部分替换成正式的向量数据库比如 FAISS、ChromaDB、Qdrant、Milvus 等。本文用最小实现演示原理。3.3 配置文件在项目根目录创建.env文件# .env OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.openai.com/v1 EMBEDDING_MODELtext-embedding-3-small CHAT_MODELgpt-4o-mini CHUNK_SIZE500 CHUNK_OVERLAP50 TOP_K3说明OPENAI_API_KEY你的 API 密钥。如果你用的是国内兼容服务或私有部署也可以指向自定义 endpoint。OPENAI_BASE_URL兼容接口的地址默认是 OpenAI 的公开接口。CHUNK_SIZE/CHUNK_OVERLAP文档分块参数越小越精确但成本越高越大越省 token 但容易丢失局部上下文。4. 核心流程拆解从零开始搭一个“LLM 你的数据”应用完整流程可以拆成六步。每一步都影响最终效果不要跳步。4.1 数据源对接常见数据源有三类本地文件TXT、Markdown、PDF、Word。业务数据库MySQL、PostgreSQL、SQLite 等。外部 API企业内部系统、网页、第三方平台。这一步的核心问题是数据长什么样更新频率如何谁能访问它。很多人上来直接做向量化忽略了数据源的权限边界导致后续权限隔离非常被动。4.2 清洗与分块原始文档往往包含大量噪音页眉页脚、导航栏、表格、重复段落。清洗的目标是让进入模型上下文的内容都是有效信息。分块是 RAG 工程里最容易踩坑的地方。块太大检索到的是大段无关内容浪费上下文窗口块太小语义被切断召回效果变差。最稳妥的经验是按段落优先切分再按字符长度和重叠窗口调整。中文场景里推荐块大小 300~800 字重叠 50~100 字。4.3 嵌入与索引把清洗后的文本块通过 Embedding 模型转成向量然后存入向量索引。常见的嵌入模型包括 OpenAI 的text-embedding-3-small、BGE、M3E 等。向量索引用 FAISS 或者商业向量数据库都可以。关键点是用于检索的嵌入模型必须和实际部署保持一致。如果开发时用模型 A线上用模型 B检索效果会明显抖动。4.4 检索策略拿到用户问题后先对问题做嵌入再在向量库里找最相似的 TopK 个文本块。这一步看似简单实际有大量策略细节是否需要混合检索向量 关键词 BM25是否需要重排模型Reranker把召回的候选重新排序是否需要按用户权限过滤数据范围是否需要按时间、部门、文档类型做元数据过滤如果只做向量相似度检索很容易出现“语义相关但答案错误”的情况。这也是“连接数据只解决 21%”的典型例证向量库本身不会告诉你哪些内容可答、哪些内容不可答。4.5 生成与引用把检索到的文本块组装进 Prompt让模型基于上下文生成回答。这一步有两个硬性要求模型必须在上下文中找不到答案时明确说“不知道”而不是编造。模型必须为关键结论标注引用来源方便用户验证。4.6 评估与反馈上线前需要准备一组评估用例集覆盖常见问题、边界问题、越权问题、幻觉问题。每次改动数据或模型都要在用例集上回归一遍。没有评估体系的 RAG 项目后期会陷入“改一个错误引入三个新错误”的泥潭。5. 完整示例与代码实现这一节给出一个最小可运行的项目。代码按文件拆开你可以原样拷贝到自己项目里跑通再按实际业务替换数据源。文件结构llm-data-demo/ ├── .env ├── config.py ├── simple_rag.py ├── db_connector.py ├── agent_example.py └── docs/ └── product_manual.txt5.1 配置文件读取# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY, ) OPENAI_BASE_URL os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, text-embedding-3-small) CHAT_MODEL os.getenv(CHAT_MODEL, gpt-4o-mini) CHUNK_SIZE int(os.getenv(CHUNK_SIZE, 500)) CHUNK_OVERLAP int(os.getenv(CHUNK_OVERLAP, 50)) TOP_K int(os.getenv(TOP_K, 3))这段代码把配置集中管理避免在业务代码里散落环境变量。load_dotenv()会自动读取项目根目录的.env文件。5.2 最小 RAG 实现# simple_rag.py import re import requests import numpy as np import config HEADERS { Authorization: fBearer {config.OPENAI_API_KEY}, Content-Type: application/json, } def get_embedding(text: str) - list: 调用 Embedding 接口将文本转为向量。 payload { model: config.EMBEDDING_MODEL, input: text, } resp requests.post( f{config.OPENAI_BASE_URL}/embeddings, jsonpayload, headersHEADERS, timeout60, ) resp.raise_for_status() return resp.json()[data][0][embedding] def chat(messages: list, temperature: float 0.2) - str: 调用 Chat Completion 接口。 payload { model: config.CHAT_MODEL, messages: messages, temperature: temperature, } resp requests.post( f{config.OPENAI_BASE_URL}/chat/completions, jsonpayload, headersHEADERS, timeout60, ) resp.raise_for_status() return resp.json()[choices][0][message][content] def load_document(path: str) - str: 读取本地纯文本文件。 with open(path, r, encodingutf-8) as f: return f.read() def chunk_text(text: str, chunk_size: int, overlap: int) - list: 简单按字符长度分块。 生产环境建议改成按段落/句子优先切分再合并到指定长度。 text re.sub(r\s, , text).strip() chunks [] start 0 while start len(text): end min(start chunk_size, len(text)) if end len(text): chunks.append(text[start:end]) break split_point text.rfind(。, start, end) if split_point ! -1 and split_point start chunk_size // 2: end split_point 1 chunks.append(text[start:end]) start max(end - overlap, start 1) return chunks class SimpleVectorStore: 基于 numpy 的最小向量索引仅用于演示原理。 def __init__(self): self.vectors [] self.metadatas [] def add(self, metadata: dict, embedding: list): self.vectors.append(np.array(embedding, dtypenp.float32)) self.metadatas.append(metadata) def search(self, query_embedding: list, top_k: int 3) - list: query_vec np.array(query_embedding, dtypenp.float32) scores [] for vec, meta in zip(self.vectors, self.metadatas): dot float(np.dot(vec, query_vec)) norm_a float(np.linalg.norm(vec)) norm_b float(np.linalg.norm(query_vec)) score dot / (norm_a * norm_b 1e-9) scores.append((score, meta)) scores.sort(reverseTrue, keylambda x: x[0]) return scores[:top_k] def build_index(doc_paths: list) - SimpleVectorStore: store SimpleVectorStore() for path in doc_paths: text load_document(path) for chunk_id, chunk in enumerate(chunk_text(text, config.CHUNK_SIZE, config.CHUNK_OVERLAP)): embedding get_embedding(chunk) store.add({source: path, chunk_id: chunk_id, text: chunk}, embedding) print(f已索引: {path} chunk {chunk_id}, 长度 {len(chunk)}) return store def ask(store: SimpleVectorStore, question: str) - dict: query_embedding get_embedding(question) results store.search(query_embedding, top_kconfig.TOP_K) if not results: return {answer: 未找到相关资料请补充文档后重试。, contexts: []} context_parts [ f[来源 {i 1}] {meta[source]}:\n{meta[text]} for i, (_, meta) in enumerate(results) ] context \n\n.join(context_parts) system_prompt ( 你是一个严谨的企业知识助手。请根据用户提供的资料回答问题。\n 规则\n 1. 只根据资料内容回答不要使用资料之外的知识。\n 2. 如果资料中没有答案请直接说“资料中未找到相关内容”。\n 3. 每个重要结论后面标注引用来源例如 [来源 1]。\n ) user_prompt f资料\n{context}\n\n问题{question} answer chat([ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ]) return { answer: answer, contexts: [{source: meta[source], score: round(score, 4)} for score, meta in results], } if __name__ __main__: store build_index([docs/product_manual.txt]) while True: q input(\n请输入问题输入 exit 退出).strip() if q.lower() exit: break result ask(store, q) print(\n 回答 ) print(result[answer]) print(\n 引用片段 ) for ctx in result[contexts]: print(ctx)这里包含了一个完整的 RAG 主链路。chunk_text做的并不是最精细的分块但演示了“先按长度切再回退到句号”的通用做法比纯字符硬切更合理。5.3 数据库数据接入实际业务中数据不只在文档里更多在数据库里。下面是一个从 SQLite 拉业务数据并格式化上下文的最小示例# db_connector.py import sqlite3 import config from simple_rag import get_embedding, chat def fetch_sales_summary() - str: 从业务库查询销售汇总结构化成文本。 conn sqlite3.connect(business.db) cursor conn.execute( SELECT region, SUM(amount) as total FROM orders GROUP BY region ORDER BY total DESC ) rows cursor.fetchall() conn.close() if not rows: return 数据库中没有销售数据。 lines [各区域销售额汇总] for region, total in rows: lines.append(f- {region}{total}) return \n.join(lines) def ask_sales(question: str) - str: context fetch_sales_summary() system_prompt 你是数据分析助手只能基于给定的数据回答不要编造数字。 user_prompt f数据\n{context}\n\n问题{question} return chat([ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ]) if __name__ __main__: q 哪个地区销售额最高 print(ask_sales(q))这段代码展示了“边查询边构造上下文”的模式。相比把所有数据灌进向量库结构化查询往往更精确也更省钱。它提醒我们不是所有数据都需要向量化先把能结构化的数据用结构化方式消费是 79% 工程里的重要策略。5.4 Function Calling 与 Agent 工具调用当模型需要调用外部工具时要用 Function Calling。这里给一个最小实现演示“模型决定调用工具 - 执行工具 - 再把结果交给模型生成最终回答”的闭环# agent_example.py import json import requests import config HEADERS { Authorization: fBearer {config.OPENAI_API_KEY}, Content-Type: application/json, } def get_sales_by_region(region: str) - str: 模拟一个业务查询函数。生产环境换成真实接口。 mock_data { 华东: 120000, 华南: 98000, 华北: 76000, } return json.dumps({region: region, sales: mock_data.get(region, 0)}, ensure_asciiFalse) TOOLS [ { type: function, function: { name: get_sales_by_region, description: 按地区查询销售额, parameters: { type: object, properties: { region: {type: string, description: 地区名称例如华东} }, required: [region], }, }, } ] def run_agent(question: str) - str: messages [{role: user, content: question}] payload { model: config.CHAT_MODEL, messages: messages, tools: TOOLS, tool_choice: auto, } resp requests.post( f{config.OPENAI_BASE_URL}/chat/completions, jsonpayload, headersHEADERS, timeout60, ) resp.raise_for_status() msg resp.json()[choices][0][message] if not msg.get(tool_calls): return msg[content] tool_call msg[tool_calls][0] function_name tool_call[function][name] arguments json.loads(tool_call[function][arguments]) print(f模型决定调用工具: {function_name}, 参数: {arguments}) if function_name get_sales_by_region: result get_sales_by_region(arguments[region]) else: result json.dumps({error: unknown tool}) messages.append(msg) messages.append({ role: tool, tool_call_id: tool_call[id], content: result, }) final_payload { model: config.CHAT_MODEL, messages: messages, } final_resp requests.post( f{config.OPENAI_BASE_URL}/chat/completions, jsonfinal_payload, headersHEADERS, timeout60, ) final_resp.raise_for_status() return final_resp.json()[choices][0][message][content] if __name__ __main__: print(run_agent(华东区销售额是多少))需要说明的是不同服务商的 Function Calling 协议字段可能有差异。在实际项目中请以你所用的 API 文档为准。这里的代码展示的是一种通用范式不是某个 SDK 的固定写法。如果你想用 MCP 统一管理工具思路是一样的把工具暴露成 MCP Server然后在客户端通过 MCP Client 调用。差别只在协议层帮你做了标准化省掉为每个业务系统写适配器的重复工作。6. 运行结果与效果验证6.1 运行方式在项目根目录执行pip install requests python-dotenv numpy python simple_rag.py前提是docs/product_manual.txt里已经放入了你的业务文档。启动后脚本会逐块索引文档然后进入交互问答模式。6.2 预期输出索引阶段输出类似已索引: docs/product_manual.txt chunk 0, 长度 485 已索引: docs/product_manual.txt chunk 1, 长度 512 已索引: docs/product_manual.txt chunk 2, 长度 498输入问题后输出包含模型回答和引用片段请输入问题输入 exit 退出产品支持哪些退货方式 回答 根据资料产品支持两种退货方式线上申请退货和联系客服人工退货。[来源 1] 引用片段 {source: docs/product_manual.txt, score: 0.8723}6.3 成功判定标准回答内容能在引用片段里找到对应依据。回答没有明显编造事实。引用片段来源与文档内容一致。连续问 5 个不同问题没有 API 报错。如果运行失败按下面顺序排查环境变量是否加载成功打印config.OPENAI_API_KEY是否非空。网络链路是否正常单独调用一次get_embedding(test)看是否报超时。API Key 是否有权限检查 HTTP 返回码401 是鉴权失败429 是限流。分块是否异常如果索引阶段没打印任何 chunk检查文档编码和读取路径。向量维度是否一致如果使用了本地模型确保查询和入库用的是同一个 Embedding 模型。7. 常见问题与排查思路问题现象可能原因排查方式解决方案检索不到相关内容分块太大、嵌入模型不匹配、文档内容被清洗过度打印召回结果检查相似度得分调小分块切换嵌入模型加入关键词检索辅助回答出现幻觉上下文包含无关片段Prompt 约束不足检查实际进入 Prompt 的上下文在系统提示中强制“无资料不得编造”限制 TopK 数量引用来源对不上引用编号和上下文顺序错位检查生成答案的 prompt 模板让模型严格沿用“来源 N”的编号体系并在解析时校验响应速度太慢向量库全量扫描、请求串行、模型推理慢分阶段压测向量检索耗时、API 耗时换用 FAISS/Qdrant增加缓存批量嵌入减少 TopK上下文长度超限检索到的文本块过多加上历史消息累积查看 token 使用量统计减小分块大小压缩历史消息限制单轮上下文长度权限越权向量库没有按用户过滤所有用户能搜到全部文档检查检索接口的过滤参数在元数据中加入文档权限字段检索时动态过滤API 返回 400 或参数错误API 版本不同字段名有差异查看错误返回的 message固定 SDK 版本或使用原生 HTTP 请求对照官方文档校验模型精度设置不一致推理服务混用 fp16/bf16/fp32导致输出异常查看推理日志和 GPU 显存状态统一推理精度固定混精度策略8. 最佳实践与工程建议8.1 数据治理先行数据接入前先回答三个问题数据的来源系统是什么更新频率如何数据的所有者和权限边界在哪里数据质量出现问题时谁来负责修正没有数据版本文档和更新策略的 RAG 项目一周后就会面临“文档乱了不知道哪一版是新的”的问题。建议给每个数据源建一个last_updated字段并在索引更新时记录版本号。8.2 建立评估回归集无论你用的是 RAG、Agent 还是 MCP都要准备一份至少 30~50 条问题的评估集包含正常的业务问题边界问题资料中没有答案越权问题不同角色询问不该看的信息数据更新问题旧文档已过期每次修改 Prompt、数据源或检索策略都跑一遍评估集。只有评估分数稳定才允许上线。8.3 权限和安全边界连接数据时最常见的安全漏洞是“整个知识库对所有人开放”。解决方案是给每个文档块打上元数据标签比如{ text: 某项目的合同金额说明..., source: contract_2024.pdf, department: finance, visible_to: [finance, admin] }检索时根据当前用户角色在向量索引中做过滤。这个逻辑必须在后端完成不能依赖前端隐藏。8.4 推理精度与部署稳定性模型精度的选择是容易忽略的运维问题。LLM 推理服务在部署时常见的精度有 fp16、bf16、fp32fp32数值精度高但显存占用大推理速度慢。fp16训练和推理常用速度快但对数值范围敏感。bf16动态范围大训练推理更稳定是当前大模型主流的推理精度。实践建议是把精度作为部署配置的一部分而不是每次临时设置。上线前固定模型版本和精度组合避免开发环境 fp32、生产环境 fp16 导致的输出漂移。8.5 日志、监控与反馈闭环每一个用户问题都要记录输入、检索 TopK 结果、响应、耗时、用户是否点赞/点踩。这些数据不仅支撑后续优化也能用来做模型漂移检测。当某个问题的回答质量开始下降往往是数据源过期或模型版本切换导致的日志会告诉你答案。8.6 用最小闭环先跑通再扩展不要一上来就引入一整套编排框架。先用本文的最小代码跑通业务确认数据链路是通畅的再逐步引入框架。框架解决的是规模化后的复杂性问题它不会自动解决数据质量问题。一个没清洗好的数据集放进任何框架都会产出错误答案。9. 总结与后续学习方向回到标题Connecting an LLM to Your Data Is the 21% Solution。这句话真正的含义是把模型连上数据只是完成了从“没有系统”到“有原型”的跨越。真正决定一个 LLM 数据项目能不能长期稳定运行的关键是数据质量、权限管控、评估回归和运维治理。这 79% 的工作没有捷径但也不是无章可循。读完这篇文章你可以先做三件事用simple_rag.py跑通一个最小 RAG把你手头最常见的业务文档放进去。为这个项目准备 20 条测试问题记录当前效果。分析你现在的数据从哪里来、更新频率、权限如何规划那 79% 的工程清单。后续值得深入的方向包括混合检索向量 关键词、Reranker 重排、Agent 多工具编排、MCP 协议接入、模型评测体系建设以及推理服务的精度与成本治理。每个方向都能单独写一篇长文。如果你正在规划新的 LLM 应用建议把这篇收藏备用。构建数据链路时遇到对应环节再回来对照比硬啃完整框架文档要高效得多。 21%法则：LLM数据连接之外的79%工程实践
如果你已经做过 RAG 相关的 LLM 应用开发一定很熟悉这种场景文档加载了、向量化跑通了、模型也能回答几个问题了可一旦放进真实业务环境效果就开始失控。文档频繁变更、权限不断缺失、召回结果不稳定、回答缺少可验证的依据。最后团队真正花掉的时间几乎都消耗在数据治理、检索调优和系统稳定性上而不是“调大模型”本身。业界有一个直击要害的说法Connecting an LLM to Your Data Is the 21% Solution。直译就是把大语言模型接到你的数据上只是完成了 21% 的解决方案。剩下 79% 的工作藏在最开始觉得很“不重要”的地方——数据准备、清洗、切分、索引、检索优化、权限隔离、评估监控、反馈闭环以及配套的运维治理。这篇文章不打算介绍某个具体的商业产品也不重复“RAG 是什么”的入门科普。我会从工程视角拆解为什么连接数据只是 21% 的完整方案剩下的 79% 具体落在哪里。同时给出一套可以照着做的最小实现数据接入、分块、向量检索、生成、评估、问题排查以及落地时容易被忽略的最佳实践。正在做 LLMData 项目的开发者和技术负责人适合静下心读一遍。已经做过 RAG 原型但没上过生产的人也会在里面看到很多熟悉又扎心的细节。1. 这篇文章真正要解决的问题我见过太多团队把“接入 LLM”当作项目的核心工作量。计划排两周第一周接 API第二周做前端演示原型看起来非常顺利。但进入验收阶段问题接踵而至同一个问题换个说法就答错新导入的文档没有覆盖到旧知识不同部门的人访问同样的知识库却看到了一样的数据权限形同虚设线上模型偶发返回乱码查了一圈发现是推理服务混用了不同精度。这些问题的共同点是什么它们几乎都不在“模型”层而在“数据和工程”层。所以这篇文章真正要解决的问题是为什么“连接 LLM 到数据”在完整系统中只占 21%剩下的 79% 是一个什么样的工程结构一个最小可用的 LLM 数据应用代码该怎么写出、怎么验证、怎么排查从原型到生产哪些工作必须提前做哪些问题可以后置这不是一篇劝退文。恰恰相反理解 21% 这个数字能让你把有限的资源投入到真正决定项目成败的地方。模型永远在快速迭代API 调用成本也在下降但数据治理和系统工程带来的壁垒才是长期价值的来源。2. 基础概念与核心原理2.1 从“模型推理”到“数据产品”传统认知里LLM 应用最关键的是模型GPT-4o、Claude、Llama选个能力强的就完事。但现实中模型只是一个推理内核它需要被“喂养”正确的数据才能输出有价值的结果。做一个知识库问答产品真正运行的是一条完整的数据链路数据源 - 采集 - 清洗 - 分块 - 嵌入 - 索引 - 检索 - 重排 - 生成 - 校验 - 返回任何一个环节松动最终回答质量都会下降。多数团队把精力集中在“嵌入 检索 生成”这三个环节也就是大约 21% 的范围。而数据采集、清洗、更新、权限、评估、监控这些 79% 的部分往往被严重低估。2.2 21% 是一个判断不是一个精确统计严格来说21% 不是某个研究报告里测出来的精确数字它是一种工程判断在可演示的原型里模型接入和数据连接确实是主干但在可交付的生产系统里它只是第一公里。用一个类比理解修一条高速公路“通车”这件事看起来最重要但真正决定公路能不能长期安全运营的是路基、排水、护栏、标识、监控和管理制度。模型连接是“铺沥青”数据工程和系统工程才是“整条路的基础设施”。2.3 LLM、RAG、Agent、MCP 的关系这四个概念经常出现在同一个项目里容易混在一起。这里用一句话区分LLM大语言模型负责理解和生成是系统的“大脑”。RAG检索增强生成从外部数据中检索相关内容再交给 LLM 生成回答。它解决的是“模型不知道私有知识”的问题。Agent智能体让模型能调用外部工具、执行动作、多步规划。它解决的是“模型只能动嘴不能动手”的问题。MCP模型上下文协议提供一套统一的工具/数据接入协议让 Agent 与外部系统连接时不用为每个工具单独写适配器。一句话总结LLM 是内核RAG 管读数据Agent 管做事情MCP 管连接协议。近期讨论度很高的“llm wiki”范式本质上也是在回答同一个问题与其堆一套复杂的 RAG 管线不如先把知识整理成结构化的、可验证的 Wiki 形式再让模型基于它产出。它强调的仍然是数据侧的系统工程。2.4 21% 与 79% 的职责分解层面具体工作占比参考技术重心LLM 接入API 调用、模型选择、提示词模板、Function Calling约 10%接口设计、Prompt 工程数据连接向量化、向量库、基础检索约 11%Embedding、相似度检索数据工程采集、清洗、去重、分块、版本管理、更新策略约 35%数据管线、质量校验系统与产品工程权限隔离、评估回归、监控告警、缓存、成本治理、反馈闭环约 30%工程架构、SRE 方法论模型与部署治理推理服务精度选择、GPU 资源、灰度发布、可观测性约 14%推理优化、MLOps你会发现真正让项目“能用”和“好用”的几乎都落在 79% 里。这也是为什么很多 RAG 项目原型很快、上线很难。3. 环境准备与前置条件为了让后面的示例跑起来你需要准备一个标准 Python 环境。版本请以实际项目为准本文重点演示通用思路不绑定某个具体版本。3.1 基础运行环境操作系统Windows / macOS / Linux 均可Python建议 3.10 或更高版本包管理工具pip 或 poetry网络需要能访问 LLM API 服务3.2 依赖安装本文示例代码尽量精简依赖只用了三个库pip install requests python-dotenv numpyrequests调用 LLM 的 HTTP API比直接封装 SDK 更容易看清请求结构。python-dotenv读取.env文件中的环境变量避免把密钥写死在代码里。numpy做向量点积和归一化计算简单实现一个向量检索。如果你要上生产可以把numpy部分替换成正式的向量数据库比如 FAISS、ChromaDB、Qdrant、Milvus 等。本文用最小实现演示原理。3.3 配置文件在项目根目录创建.env文件# .env OPENAI_API_KEYsk-xxxx OPENAI_BASE_URLhttps://api.openai.com/v1 EMBEDDING_MODELtext-embedding-3-small CHAT_MODELgpt-4o-mini CHUNK_SIZE500 CHUNK_OVERLAP50 TOP_K3说明OPENAI_API_KEY你的 API 密钥。如果你用的是国内兼容服务或私有部署也可以指向自定义 endpoint。OPENAI_BASE_URL兼容接口的地址默认是 OpenAI 的公开接口。CHUNK_SIZE/CHUNK_OVERLAP文档分块参数越小越精确但成本越高越大越省 token 但容易丢失局部上下文。4. 核心流程拆解从零开始搭一个“LLM 你的数据”应用完整流程可以拆成六步。每一步都影响最终效果不要跳步。4.1 数据源对接常见数据源有三类本地文件TXT、Markdown、PDF、Word。业务数据库MySQL、PostgreSQL、SQLite 等。外部 API企业内部系统、网页、第三方平台。这一步的核心问题是数据长什么样更新频率如何谁能访问它。很多人上来直接做向量化忽略了数据源的权限边界导致后续权限隔离非常被动。4.2 清洗与分块原始文档往往包含大量噪音页眉页脚、导航栏、表格、重复段落。清洗的目标是让进入模型上下文的内容都是有效信息。分块是 RAG 工程里最容易踩坑的地方。块太大检索到的是大段无关内容浪费上下文窗口块太小语义被切断召回效果变差。最稳妥的经验是按段落优先切分再按字符长度和重叠窗口调整。中文场景里推荐块大小 300~800 字重叠 50~100 字。4.3 嵌入与索引把清洗后的文本块通过 Embedding 模型转成向量然后存入向量索引。常见的嵌入模型包括 OpenAI 的text-embedding-3-small、BGE、M3E 等。向量索引用 FAISS 或者商业向量数据库都可以。关键点是用于检索的嵌入模型必须和实际部署保持一致。如果开发时用模型 A线上用模型 B检索效果会明显抖动。4.4 检索策略拿到用户问题后先对问题做嵌入再在向量库里找最相似的 TopK 个文本块。这一步看似简单实际有大量策略细节是否需要混合检索向量 关键词 BM25是否需要重排模型Reranker把召回的候选重新排序是否需要按用户权限过滤数据范围是否需要按时间、部门、文档类型做元数据过滤如果只做向量相似度检索很容易出现“语义相关但答案错误”的情况。这也是“连接数据只解决 21%”的典型例证向量库本身不会告诉你哪些内容可答、哪些内容不可答。4.5 生成与引用把检索到的文本块组装进 Prompt让模型基于上下文生成回答。这一步有两个硬性要求模型必须在上下文中找不到答案时明确说“不知道”而不是编造。模型必须为关键结论标注引用来源方便用户验证。4.6 评估与反馈上线前需要准备一组评估用例集覆盖常见问题、边界问题、越权问题、幻觉问题。每次改动数据或模型都要在用例集上回归一遍。没有评估体系的 RAG 项目后期会陷入“改一个错误引入三个新错误”的泥潭。5. 完整示例与代码实现这一节给出一个最小可运行的项目。代码按文件拆开你可以原样拷贝到自己项目里跑通再按实际业务替换数据源。文件结构llm-data-demo/ ├── .env ├── config.py ├── simple_rag.py ├── db_connector.py ├── agent_example.py └── docs/ └── product_manual.txt5.1 配置文件读取# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY, ) OPENAI_BASE_URL os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, text-embedding-3-small) CHAT_MODEL os.getenv(CHAT_MODEL, gpt-4o-mini) CHUNK_SIZE int(os.getenv(CHUNK_SIZE, 500)) CHUNK_OVERLAP int(os.getenv(CHUNK_OVERLAP, 50)) TOP_K int(os.getenv(TOP_K, 3))这段代码把配置集中管理避免在业务代码里散落环境变量。load_dotenv()会自动读取项目根目录的.env文件。5.2 最小 RAG 实现# simple_rag.py import re import requests import numpy as np import config HEADERS { Authorization: fBearer {config.OPENAI_API_KEY}, Content-Type: application/json, } def get_embedding(text: str) - list: 调用 Embedding 接口将文本转为向量。 payload { model: config.EMBEDDING_MODEL, input: text, } resp requests.post( f{config.OPENAI_BASE_URL}/embeddings, jsonpayload, headersHEADERS, timeout60, ) resp.raise_for_status() return resp.json()[data][0][embedding] def chat(messages: list, temperature: float 0.2) - str: 调用 Chat Completion 接口。 payload { model: config.CHAT_MODEL, messages: messages, temperature: temperature, } resp requests.post( f{config.OPENAI_BASE_URL}/chat/completions, jsonpayload, headersHEADERS, timeout60, ) resp.raise_for_status() return resp.json()[choices][0][message][content] def load_document(path: str) - str: 读取本地纯文本文件。 with open(path, r, encodingutf-8) as f: return f.read() def chunk_text(text: str, chunk_size: int, overlap: int) - list: 简单按字符长度分块。 生产环境建议改成按段落/句子优先切分再合并到指定长度。 text re.sub(r\s, , text).strip() chunks [] start 0 while start len(text): end min(start chunk_size, len(text)) if end len(text): chunks.append(text[start:end]) break split_point text.rfind(。, start, end) if split_point ! -1 and split_point start chunk_size // 2: end split_point 1 chunks.append(text[start:end]) start max(end - overlap, start 1) return chunks class SimpleVectorStore: 基于 numpy 的最小向量索引仅用于演示原理。 def __init__(self): self.vectors [] self.metadatas [] def add(self, metadata: dict, embedding: list): self.vectors.append(np.array(embedding, dtypenp.float32)) self.metadatas.append(metadata) def search(self, query_embedding: list, top_k: int 3) - list: query_vec np.array(query_embedding, dtypenp.float32) scores [] for vec, meta in zip(self.vectors, self.metadatas): dot float(np.dot(vec, query_vec)) norm_a float(np.linalg.norm(vec)) norm_b float(np.linalg.norm(query_vec)) score dot / (norm_a * norm_b 1e-9) scores.append((score, meta)) scores.sort(reverseTrue, keylambda x: x[0]) return scores[:top_k] def build_index(doc_paths: list) - SimpleVectorStore: store SimpleVectorStore() for path in doc_paths: text load_document(path) for chunk_id, chunk in enumerate(chunk_text(text, config.CHUNK_SIZE, config.CHUNK_OVERLAP)): embedding get_embedding(chunk) store.add({source: path, chunk_id: chunk_id, text: chunk}, embedding) print(f已索引: {path} chunk {chunk_id}, 长度 {len(chunk)}) return store def ask(store: SimpleVectorStore, question: str) - dict: query_embedding get_embedding(question) results store.search(query_embedding, top_kconfig.TOP_K) if not results: return {answer: 未找到相关资料请补充文档后重试。, contexts: []} context_parts [ f[来源 {i 1}] {meta[source]}:\n{meta[text]} for i, (_, meta) in enumerate(results) ] context \n\n.join(context_parts) system_prompt ( 你是一个严谨的企业知识助手。请根据用户提供的资料回答问题。\n 规则\n 1. 只根据资料内容回答不要使用资料之外的知识。\n 2. 如果资料中没有答案请直接说“资料中未找到相关内容”。\n 3. 每个重要结论后面标注引用来源例如 [来源 1]。\n ) user_prompt f资料\n{context}\n\n问题{question} answer chat([ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ]) return { answer: answer, contexts: [{source: meta[source], score: round(score, 4)} for score, meta in results], } if __name__ __main__: store build_index([docs/product_manual.txt]) while True: q input(\n请输入问题输入 exit 退出).strip() if q.lower() exit: break result ask(store, q) print(\n 回答 ) print(result[answer]) print(\n 引用片段 ) for ctx in result[contexts]: print(ctx)这里包含了一个完整的 RAG 主链路。chunk_text做的并不是最精细的分块但演示了“先按长度切再回退到句号”的通用做法比纯字符硬切更合理。5.3 数据库数据接入实际业务中数据不只在文档里更多在数据库里。下面是一个从 SQLite 拉业务数据并格式化上下文的最小示例# db_connector.py import sqlite3 import config from simple_rag import get_embedding, chat def fetch_sales_summary() - str: 从业务库查询销售汇总结构化成文本。 conn sqlite3.connect(business.db) cursor conn.execute( SELECT region, SUM(amount) as total FROM orders GROUP BY region ORDER BY total DESC ) rows cursor.fetchall() conn.close() if not rows: return 数据库中没有销售数据。 lines [各区域销售额汇总] for region, total in rows: lines.append(f- {region}{total}) return \n.join(lines) def ask_sales(question: str) - str: context fetch_sales_summary() system_prompt 你是数据分析助手只能基于给定的数据回答不要编造数字。 user_prompt f数据\n{context}\n\n问题{question} return chat([ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ]) if __name__ __main__: q 哪个地区销售额最高 print(ask_sales(q))这段代码展示了“边查询边构造上下文”的模式。相比把所有数据灌进向量库结构化查询往往更精确也更省钱。它提醒我们不是所有数据都需要向量化先把能结构化的数据用结构化方式消费是 79% 工程里的重要策略。5.4 Function Calling 与 Agent 工具调用当模型需要调用外部工具时要用 Function Calling。这里给一个最小实现演示“模型决定调用工具 - 执行工具 - 再把结果交给模型生成最终回答”的闭环# agent_example.py import json import requests import config HEADERS { Authorization: fBearer {config.OPENAI_API_KEY}, Content-Type: application/json, } def get_sales_by_region(region: str) - str: 模拟一个业务查询函数。生产环境换成真实接口。 mock_data { 华东: 120000, 华南: 98000, 华北: 76000, } return json.dumps({region: region, sales: mock_data.get(region, 0)}, ensure_asciiFalse) TOOLS [ { type: function, function: { name: get_sales_by_region, description: 按地区查询销售额, parameters: { type: object, properties: { region: {type: string, description: 地区名称例如华东} }, required: [region], }, }, } ] def run_agent(question: str) - str: messages [{role: user, content: question}] payload { model: config.CHAT_MODEL, messages: messages, tools: TOOLS, tool_choice: auto, } resp requests.post( f{config.OPENAI_BASE_URL}/chat/completions, jsonpayload, headersHEADERS, timeout60, ) resp.raise_for_status() msg resp.json()[choices][0][message] if not msg.get(tool_calls): return msg[content] tool_call msg[tool_calls][0] function_name tool_call[function][name] arguments json.loads(tool_call[function][arguments]) print(f模型决定调用工具: {function_name}, 参数: {arguments}) if function_name get_sales_by_region: result get_sales_by_region(arguments[region]) else: result json.dumps({error: unknown tool}) messages.append(msg) messages.append({ role: tool, tool_call_id: tool_call[id], content: result, }) final_payload { model: config.CHAT_MODEL, messages: messages, } final_resp requests.post( f{config.OPENAI_BASE_URL}/chat/completions, jsonfinal_payload, headersHEADERS, timeout60, ) final_resp.raise_for_status() return final_resp.json()[choices][0][message][content] if __name__ __main__: print(run_agent(华东区销售额是多少))需要说明的是不同服务商的 Function Calling 协议字段可能有差异。在实际项目中请以你所用的 API 文档为准。这里的代码展示的是一种通用范式不是某个 SDK 的固定写法。如果你想用 MCP 统一管理工具思路是一样的把工具暴露成 MCP Server然后在客户端通过 MCP Client 调用。差别只在协议层帮你做了标准化省掉为每个业务系统写适配器的重复工作。6. 运行结果与效果验证6.1 运行方式在项目根目录执行pip install requests python-dotenv numpy python simple_rag.py前提是docs/product_manual.txt里已经放入了你的业务文档。启动后脚本会逐块索引文档然后进入交互问答模式。6.2 预期输出索引阶段输出类似已索引: docs/product_manual.txt chunk 0, 长度 485 已索引: docs/product_manual.txt chunk 1, 长度 512 已索引: docs/product_manual.txt chunk 2, 长度 498输入问题后输出包含模型回答和引用片段请输入问题输入 exit 退出产品支持哪些退货方式 回答 根据资料产品支持两种退货方式线上申请退货和联系客服人工退货。[来源 1] 引用片段 {source: docs/product_manual.txt, score: 0.8723}6.3 成功判定标准回答内容能在引用片段里找到对应依据。回答没有明显编造事实。引用片段来源与文档内容一致。连续问 5 个不同问题没有 API 报错。如果运行失败按下面顺序排查环境变量是否加载成功打印config.OPENAI_API_KEY是否非空。网络链路是否正常单独调用一次get_embedding(test)看是否报超时。API Key 是否有权限检查 HTTP 返回码401 是鉴权失败429 是限流。分块是否异常如果索引阶段没打印任何 chunk检查文档编码和读取路径。向量维度是否一致如果使用了本地模型确保查询和入库用的是同一个 Embedding 模型。7. 常见问题与排查思路问题现象可能原因排查方式解决方案检索不到相关内容分块太大、嵌入模型不匹配、文档内容被清洗过度打印召回结果检查相似度得分调小分块切换嵌入模型加入关键词检索辅助回答出现幻觉上下文包含无关片段Prompt 约束不足检查实际进入 Prompt 的上下文在系统提示中强制“无资料不得编造”限制 TopK 数量引用来源对不上引用编号和上下文顺序错位检查生成答案的 prompt 模板让模型严格沿用“来源 N”的编号体系并在解析时校验响应速度太慢向量库全量扫描、请求串行、模型推理慢分阶段压测向量检索耗时、API 耗时换用 FAISS/Qdrant增加缓存批量嵌入减少 TopK上下文长度超限检索到的文本块过多加上历史消息累积查看 token 使用量统计减小分块大小压缩历史消息限制单轮上下文长度权限越权向量库没有按用户过滤所有用户能搜到全部文档检查检索接口的过滤参数在元数据中加入文档权限字段检索时动态过滤API 返回 400 或参数错误API 版本不同字段名有差异查看错误返回的 message固定 SDK 版本或使用原生 HTTP 请求对照官方文档校验模型精度设置不一致推理服务混用 fp16/bf16/fp32导致输出异常查看推理日志和 GPU 显存状态统一推理精度固定混精度策略8. 最佳实践与工程建议8.1 数据治理先行数据接入前先回答三个问题数据的来源系统是什么更新频率如何数据的所有者和权限边界在哪里数据质量出现问题时谁来负责修正没有数据版本文档和更新策略的 RAG 项目一周后就会面临“文档乱了不知道哪一版是新的”的问题。建议给每个数据源建一个last_updated字段并在索引更新时记录版本号。8.2 建立评估回归集无论你用的是 RAG、Agent 还是 MCP都要准备一份至少 30~50 条问题的评估集包含正常的业务问题边界问题资料中没有答案越权问题不同角色询问不该看的信息数据更新问题旧文档已过期每次修改 Prompt、数据源或检索策略都跑一遍评估集。只有评估分数稳定才允许上线。8.3 权限和安全边界连接数据时最常见的安全漏洞是“整个知识库对所有人开放”。解决方案是给每个文档块打上元数据标签比如{ text: 某项目的合同金额说明..., source: contract_2024.pdf, department: finance, visible_to: [finance, admin] }检索时根据当前用户角色在向量索引中做过滤。这个逻辑必须在后端完成不能依赖前端隐藏。8.4 推理精度与部署稳定性模型精度的选择是容易忽略的运维问题。LLM 推理服务在部署时常见的精度有 fp16、bf16、fp32fp32数值精度高但显存占用大推理速度慢。fp16训练和推理常用速度快但对数值范围敏感。bf16动态范围大训练推理更稳定是当前大模型主流的推理精度。实践建议是把精度作为部署配置的一部分而不是每次临时设置。上线前固定模型版本和精度组合避免开发环境 fp32、生产环境 fp16 导致的输出漂移。8.5 日志、监控与反馈闭环每一个用户问题都要记录输入、检索 TopK 结果、响应、耗时、用户是否点赞/点踩。这些数据不仅支撑后续优化也能用来做模型漂移检测。当某个问题的回答质量开始下降往往是数据源过期或模型版本切换导致的日志会告诉你答案。8.6 用最小闭环先跑通再扩展不要一上来就引入一整套编排框架。先用本文的最小代码跑通业务确认数据链路是通畅的再逐步引入框架。框架解决的是规模化后的复杂性问题它不会自动解决数据质量问题。一个没清洗好的数据集放进任何框架都会产出错误答案。9. 总结与后续学习方向回到标题Connecting an LLM to Your Data Is the 21% Solution。这句话真正的含义是把模型连上数据只是完成了从“没有系统”到“有原型”的跨越。真正决定一个 LLM 数据项目能不能长期稳定运行的关键是数据质量、权限管控、评估回归和运维治理。这 79% 的工作没有捷径但也不是无章可循。读完这篇文章你可以先做三件事用simple_rag.py跑通一个最小 RAG把你手头最常见的业务文档放进去。为这个项目准备 20 条测试问题记录当前效果。分析你现在的数据从哪里来、更新频率、权限如何规划那 79% 的工程清单。后续值得深入的方向包括混合检索向量 关键词、Reranker 重排、Agent 多工具编排、MCP 协议接入、模型评测体系建设以及推理服务的精度与成本治理。每个方向都能单独写一篇长文。如果你正在规划新的 LLM 应用建议把这篇收藏备用。构建数据链路时遇到对应环节再回来对照比硬啃完整框架文档要高效得多。