大模型上下文工程实战:RAG、记忆、API与MCP构建带鉴权审计的生产级应用
发布时间:2026/10/2 10:31:31来源:尧图网络
1. 从标题拆解这套系统的真实骨架1.1 标题里藏着的四个独立模块“大模型上下文与工具链搭建基于RAG、记忆、API和MCP构建带鉴权审计应用实践23.4”——这个标题信息密度很高我第一眼看到的时候就觉得它不是那种“跑个demo就完事”的玩具项目。拆开来看它至少包含四个可以独立成篇的子系统RAG检索增强负责让模型“知道得更多”记忆模块负责让模型“记得住上下文”API层负责对外暴露能力MCP负责把外部工具接进来。而“鉴权审计”这四个字才是真正让这套东西从“能跑”变成“能上线”的分水岭。很多人做RAG做到最后发现检索准确率上不去、上下文窗口被塞爆、多轮对话里模型把前面说过的信息忘得一干二净。这些问题单独看是检索问题、是记忆问题但放到一起看其实是上下文管理策略的问题。你不可能把所有东西都塞进prompt里1048576 tokens的窗口听起来很大但真跑起来光是一份产品手册加几轮对话历史就能吃掉大半。所以这套系统的核心矛盾就一句话在有限的上下文预算里让模型拿到最该拿到的信息同时保证每一次调用都可追溯、可审计、可控制。适合谁来参考如果你已经跑通过最简单的RAG demo但一上生产就发现检索命中率忽高忽低、多轮对话记忆混乱、API裸奔没有权限控制那这篇就是写给你的。如果你还在纠结“RAG知识库能不能存图片”这种问题建议先把基础链路跑通再回来看。1.2 为什么是RAG记忆APIMCP这个组合单独用RAG模型只能被动地“查资料”查完就忘单独用记忆模型记住的都是对话历史没有外部知识注入单独用API你只是把模型包了一层HTTP接口没有任何增强单独用MCP工具调用能力有了但工具返回的结果怎么和知识库结合、怎么和对话历史结合还是没解决。这四个东西的关系我习惯用一家餐厅来类比RAG是后厨的食材库需要什么菜现去取记忆是服务员的点单记录知道这桌客人之前点了什么、忌口什么API是餐厅的大门和菜单对外决定谁能进来、能点什么MCP是厨房里的各种设备接口烤箱、微波炉、搅拌机都通过统一标准接进来厨师不用管每个设备的具体操作方式。而鉴权审计就是门口的保安加监控谁进来了、点了什么、后厨做了什么全都有记录。这个组合之所以成立是因为它们解决的是同一个问题的不同侧面如何让大模型在真实业务场景里稳定、安全、可解释地工作。缺了RAG模型没有领域知识缺了记忆多轮对话就断片缺了API和鉴权系统没法对外服务缺了MCP工具调用就是一堆硬编码的if-else。四个模块拼在一起才是一个能交付给业务方用的东西。1.3 版本号23.4透露的信息标题末尾的“23.4”大概率是内部版本号说明这套方案已经迭代过很多轮了。我自己的经验是RAG记忆工具链这套东西第一版能跑通就不错了第二版开始处理检索精度第三版才会认真做鉴权和审计。能标到23.4说明作者已经在生产环境里踩过足够多的坑知道哪些地方容易出问题、哪些参数需要反复调。这也是为什么我在这篇里会重点讲“为什么这么选”而不是“怎么装”。2. RAG检索增强的工程化落地细节2.1 文档切分策略决定检索上限RAG的第一个瓶颈永远在文档切分。我见过太多人直接把整篇PDF丢进去切按固定字数一刀切结果检索出来的片段要么缺头少尾要么把两个不相关的段落拼在一起。正确的做法是按语义边界切分同时保留一定的重叠窗口。具体操作上我一般用递归字符切分加语义分割的组合策略。先按标题层级切大块再在大块内部按段落切最后对超长段落按句子边界切。每个chunk控制在300到500个token之间重叠50到80个token。为什么要重叠因为用户的问题可能刚好落在两个chunk的边界上没有重叠的话检索出来的片段可能只包含答案的一半。from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap60, separators[\n## , \n### , \n\n, \n, 。, , , , , , ], length_functionlen, )注意separators的顺序很关键中文场景下要把中文标点放在英文标点前面否则切出来的句子会很碎。另外chunk_size不要设太大400左右是个比较稳的值再大检索精度会下降再小语义完整性不够。提示如果你的知识库里包含表格和图片说明建议单独走一条处理链路。表格转成Markdown格式保留结构图片用多模态模型生成文字描述后再入库。RAG知识库本身不直接存图片但可以把图片的文本描述和图片路径一起存进去检索到之后返回路径给前端展示。2.2 向量化模型选型与混合检索向量化模型的选择直接决定检索质量。我实测下来中文场景下BGE系列和M3E系列表现比较稳英文场景OpenAI的text-embedding-3-small性价比很高。但纯向量检索有个致命问题对精确匹配不敏感。用户搜一个产品型号“XK-2024A”向量检索可能返回一堆语义相似但型号不对的文档。所以生产环境我强烈建议上混合检索向量检索加BM25关键词检索两路结果用RRF融合。这样既能抓住语义相似又能保证关键词精确命中。from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain_community.vectorstores import Chroma vector_retriever Chroma(...).as_retriever(search_kwargs{k: 10}) bm25_retriever BM25Retriever.from_documents(docs, k10) ensemble EnsembleRetriever( retrievers[vector_retriever, bm25_retriever], weights[0.6, 0.4], )权重怎么定我的经验是向量检索占0.6、BM25占0.4这个比例在大多数场景下比较平衡。如果你的知识库里有大量专有名词和编号可以把BM25权重提到0.5。2.3 重排序是提升命中率的关键一步混合检索拿到top-20之后不要直接塞给模型中间必须加一层重排序。重排序模型比如BGE-Reranker会对每个候选片段和query的相关性做精细打分把真正相关的排到前面。这一步能把检索命中率从60%左右拉到85%以上代价只是增加几十毫秒的延迟。from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-v2-m3, use_fp16True) pairs [[query, doc.page_content] for doc in candidates] scores reranker.compute_score(pairs) ranked sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue)[:5]注意重排序之后只取top-3到top-5就够了取太多反而会引入噪声。我一般取top-4然后根据token预算动态调整。2.4 上下文预算分配与截断策略这是很多人忽略的一步。假设模型窗口是128K你不能把检索结果全塞进去因为还要留空间给系统提示词、对话历史、工具返回结果。我的做法是给每个部分分配固定预算内容类型预算占比说明系统提示词5%固定不变优先保留对话历史20%按轮次从新到旧保留RAG检索结果40%按重排序分数截断工具返回结果20%按需动态分配输出预留15%留给模型生成当总预算超了优先砍对话历史里最旧的轮次再砍RAG里分数最低的片段。工具返回结果一般不动因为那是模型主动要的。3. 记忆模块的设计与上下文管理3.1 短期记忆与长期记忆的分层记忆模块最容易踩的坑是把所有对话历史都当短期记忆塞进prompt。正确的做法是分层短期记忆只保留最近N轮对话的原始文本长期记忆把更早的对话做摘要后存储需要时再检索出来。短期记忆我一般保留最近5到8轮具体看每轮的平均长度。如果一轮对话平均200个token8轮就是1600个token在预算范围内。超过8轮的对话用一个小模型做摘要摘要结果存到向量库里下次遇到相关话题时检索出来注入。def manage_memory(history, max_recent8): if len(history) max_recent: return history recent history[-max_recent:] old history[:-max_recent] summary summarize(old) # 调用小模型做摘要 return [{role: system, content: f历史对话摘要{summary}}] recent摘要的prompt要明确要求保留关键实体、决策和未完成事项不要泛泛地概括。我试过让模型自由发挥结果摘要里全是“用户询问了相关问题”这种废话一点用都没有。3.2 记忆检索的触发时机长期记忆不是每轮都检索那样太浪费。我的触发策略是当用户的问题里出现指代词“那个”“之前说的”“上次提到的”或者和当前对话主题的向量相似度低于阈值时才去检索长期记忆。这样既保证了连贯性又不会每轮都增加延迟。具体实现上我用一个轻量级的意图分类器判断是否需要检索记忆。分类器可以是一个小的BERT模型也可以直接用规则加关键词匹配。规则匹配虽然土但在大多数场景下够用而且零延迟。3.3 记忆冲突的处理多轮对话里经常出现用户改口的情况比如先说“我要订周五的票”后来说“算了改成周六”。如果记忆模块把两条都保留模型可能会困惑。我的处理方式是后发覆盖新信息覆盖旧信息同时在记忆里标记“已更新”。具体做法是在存储记忆时带上时间戳和状态标记检索时只取最新状态。注意覆盖策略要区分事实性信息和偏好性信息。事实性信息时间、地点、数量直接覆盖偏好性信息喜欢什么、讨厌什么要合并而不是覆盖因为偏好可能同时存在多个。4. API层设计与鉴权审计实现4.1 API网关的核心职责API层不只是把模型包一层HTTP接口那么简单。它要承担鉴权、限流、审计、路由四个职责。鉴权决定谁能调用限流防止滥用审计记录每一次调用路由决定请求发给哪个模型或哪个RAG链路。我一般用FastAPI做API层因为它的异步支持和依赖注入机制很适合这种场景。鉴权用JWT加API Key双机制内部服务调用走API Key外部用户走JWT。API Key存在数据库里带过期时间和权限范围JWT里带用户ID和角色每次请求验证签名和过期时间。from fastapi import FastAPI, Depends, HTTPException from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials security HTTPBearer() async def verify_token(credentials: HTTPAuthorizationCredentials Depends(security)): token credentials.credentials try: payload jwt.decode(token, SECRET_KEY, algorithms[HS256]) return payload except jwt.ExpiredSignatureError: raise HTTPException(status_code401, detailToken expired) except jwt.InvalidTokenError: raise HTTPException(status_code401, detailInvalid token)这里有个细节401错误返回的信息不要暴露太多比如“incorrect api key provided: sk-svcac****”这种信息在生产环境里是安全隐患。统一返回“Unauthorized”就够了具体原因记到审计日志里。4.2 审计日志的设计要点审计日志要记录什么我的清单是请求ID、用户ID、时间戳、请求路径、请求参数摘要、模型名称、token消耗、响应状态、响应耗时、RAG检索到的文档ID列表、工具调用记录。这些信息足够还原每一次调用的完整链路。存储上审计日志不要和业务数据放同一个库单独用一个时序数据库或者日志服务。写入用异步方式不要阻塞主请求链路。我一般用消息队列做缓冲请求结束后发一条消息到队列后台消费者批量写入。async def audit_log(request_id, user_id, path, params, model, tokens, status, duration, doc_ids, tool_calls): log_entry { request_id: request_id, user_id: user_id, timestamp: datetime.utcnow().isoformat(), path: path, params_summary: truncate(params, 200), model: model, tokens: tokens, status: status, duration_ms: duration, retrieved_docs: doc_ids, tool_calls: tool_calls, } await audit_queue.put(log_entry)提示审计日志里的请求参数要做脱敏处理用户输入的敏感信息手机号、身份证号要掩码后再记录。响应内容一般不记录全文只记录长度和状态。4.3 限流与配额管理限流我一般用令牌桶算法按用户维度限流。每个用户有一个令牌桶桶容量和补充速率根据用户等级配置。免费用户每分钟10次付费用户每分钟60次内部服务每分钟600次。超过限流返回429同时在响应头里带上重试等待时间。配额管理是限流的补充按天或按月统计token消耗。每个用户有一个配额上限消耗到80%时发预警到100%时拒绝新请求。配额数据存在Redis里用原子操作保证并发安全。5. MCP工具链接入与工具调用编排5.1 MCP协议的核心价值MCPModel Context Protocol解决的是工具调用的标准化问题。没有MCP之前每接一个工具就要写一套适配代码工具多了之后维护成本爆炸。MCP定义了一套统一的协议工具提供方按协议暴露能力模型侧按协议调用中间的适配层只需要实现一次。MCP的核心概念是Server和Client。Server暴露工具列表和调用接口Client负责发现工具、调用工具、处理返回结果。一个MCP Server可以暴露多个工具一个Client可以连接多个Server。这种设计让工具生态可以独立演进模型侧不用关心工具的具体实现。5.2 工具注册与发现机制在实际项目里我一般把工具分成三类内置工具计算器、时间查询、业务工具查订单、查库存、外部工具搜索引擎、第三方API。内置工具直接注册到Client里业务工具通过内部MCP Server暴露外部工具通过标准MCP协议接入。工具注册时要提供名称、描述、参数schema、返回值schema。描述要写得让模型能理解什么时候该用这个工具参数schema用JSON Schema格式模型会根据schema生成调用参数。tools [ { name: query_order, description: 根据订单号查询订单状态和详情, parameters: { type: object, properties: { order_id: {type: string, description: 订单号格式为ORD开头加12位数字} }, required: [order_id] } } ]描述里一定要写清楚参数的格式要求否则模型生成的参数可能不符合预期。我踩过的坑是模型把订单号里的字母小写了导致查询失败。后来在描述里明确写了“格式为ORD开头加12位数字区分大小写”问题就解决了。5.3 工具调用编排与错误处理模型一次可能调用多个工具工具之间可能有依赖关系。比如先查订单拿到订单里的商品ID再查商品详情。这种编排逻辑我一般不让模型自己决定而是在系统提示词里给出明确的编排规则或者用工作流引擎预先定义好。错误处理是工具调用里最容易被忽略的部分。工具调用失败时不要把原始错误信息直接返回给模型那样模型可能会胡编乱造。正确的做法是返回结构化的错误信息让模型知道是参数错了、超时了、还是权限不够。def handle_tool_error(error): if isinstance(error, TimeoutError): return {status: error, code: TIMEOUT, message: 工具调用超时请稍后重试} elif isinstance(error, PermissionError): return {status: error, code: FORBIDDEN, message: 当前用户无权调用此工具} else: return {status: error, code: UNKNOWN, message: 工具调用失败}模型拿到结构化错误后可以选择重试、换工具、或者告诉用户失败原因。这比让模型面对一堆堆栈信息要好得多。5.4 工具调用与RAG的协同工具调用和RAG不是互斥的很多时候需要协同。比如用户问“我的订单到哪了”模型先调用订单查询工具拿到订单状态发现状态是“已发货”然后需要查物流信息。物流信息可能在RAG知识库里比如物流公司的配送范围说明也可能需要调用物流查询工具。我的做法是在系统提示词里明确告诉模型先查工具拿实时数据再用RAG补充背景知识。工具返回的结果作为上下文的一部分注入RAG检索时把工具返回的关键实体作为query的一部分提高检索精度。6. 鉴权审计与全链路可观测性6.1 鉴权的三个层次鉴权不是简单的“有没有token”要分三个层次身份认证你是谁、权限校验你能做什么、配额检查你还能做多少。身份认证用JWT或API Key权限校验用RBAC模型配额检查用Redis计数器。RBAC模型里角色和权限的映射关系要可配置。我一般用数据库存角色、权限、角色权限关联三张表启动时加载到内存缓存变更时刷新缓存。权限粒度控制到API路径级别比如/api/chat需要chat:invoke权限/api/admin需要admin:access权限。6.2 审计日志的查询与分析审计日志写进去容易查出来难。我一般按三个维度建索引用户维度查某个用户的所有调用、时间维度查某个时间段的调用、请求ID维度查某次调用的完整链路。查询接口要支持组合条件同时做好分页和超时控制。分析方面我关注几个核心指标调用量趋势、平均响应时间、错误率、token消耗分布、工具调用成功率。这些指标用Grafana做可视化异常时触发告警。比如错误率超过5%持续5分钟就发告警到值班群。6.3 全链路追踪的实现一次请求可能经过API层、RAG检索、记忆检索、模型调用、工具调用多个环节每个环节的耗时和状态都要能追踪。我用OpenTelemetry做链路追踪每个环节生成一个Span最后聚合成一个Trace。from opentelemetry import trace tracer trace.get_tracer(__name__) async def handle_request(request): with tracer.start_as_current_span(handle_request) as span: span.set_attribute(user_id, request.user_id) with tracer.start_as_current_span(rag_retrieval): docs await retrieve(request.query) with tracer.start_as_current_span(llm_call): response await call_llm(request.query, docs) return responseTrace数据存到Jaeger或Tempo里出问题时可以快速定位是哪个环节慢了、哪个环节错了。我踩过的坑是RAG检索偶尔超时但日志里只看到总耗时不知道是向量检索慢还是重排序慢。加了Span之后一目了然发现是重排序模型首次加载慢后来做了预热就好了。7. 常见问题与排查技巧实录7.1 检索命中率低的排查思路检索命中率低是最常见的问题。排查顺序是先看切分再看向量化再看检索策略最后看重排序。切分问题占一半以上很多人的chunk切得太大或太小或者没有按语义边界切。向量化问题占三成主要是模型选型不对或者没有做归一化。检索策略问题占一成纯向量检索对关键词不敏感。重排序问题占一成没用重排序或者重排序模型选得不好。我一般用一组标准问题做回归测试每次调整参数后跑一遍看命中率变化。标准问题要覆盖事实查询、语义查询、多跳查询三种类型。7.2 上下文超长的处理上下文超长报错“maximum context length is 1048576 tokens”说明预算没管好。排查步骤先算系统提示词和对话历史的token数再算RAG检索结果的token数再算工具返回的token数看哪部分超了。我一般写一个token预算检查函数每次组装prompt前先跑一遍超了就按优先级砍。def check_budget(system_prompt, history, rag_docs, tool_results, max_tokens128000): total count_tokens(system_prompt) count_tokens(history) count_tokens(rag_docs) count_tokens(tool_results) if total max_tokens * 0.85: # 按优先级砍先砍历史再砍RAG history truncate_history(history, max_tokens * 0.15) rag_docs truncate_docs(rag_docs, max_tokens * 0.35) return system_prompt, history, rag_docs, tool_results7.3 工具调用失败的常见原因工具调用失败的原因我整理了一个速查表错误现象可能原因排查方法401 UnauthorizedAPI Key过期或错误检查Key配置和过期时间400 Bad Request参数格式不对检查JSON Schema和模型生成的参数超时工具服务响应慢检查工具服务负载和网络延迟返回空结果查询条件太严放宽查询条件或检查数据源模型不调用工具工具描述不清晰优化工具描述和系统提示词提示模型不调用工具是最隐蔽的问题。我遇到过模型明明应该调用工具却直接编答案的情况后来发现是工具描述里没写清楚“什么时候必须调用”。在描述里加上“当用户询问实时数据时必须调用此工具”之后问题就解决了。7.4 记忆混乱的排查记忆混乱表现为模型把不同轮次的信息搞混或者把摘要里的信息和当前对话搞混。排查方法是把每次注入的完整prompt打印出来人工检查记忆部分是否正确。常见原因有三个摘要质量差、记忆检索触发了不该触发的、覆盖策略不对。摘要质量差是最常见的小模型做摘要时容易丢失关键信息。我的对策是用大一点的模型做摘要或者在摘要prompt里明确列出必须保留的信息类型。记忆检索误触发一般是阈值设得太低调高阈值或者加规则过滤。覆盖策略问题一般是事实和偏好没区分按前面说的分类处理。8. 部署与性能优化的实操经验8.1 模型服务的部署方式模型服务我一般用vLLM或TGI部署支持连续批处理和PagedAttention吞吐量比裸跑高好几倍。如果用的是外部API比如DeepSeek、智谱要做好超时和重试配置。超时设30秒重试最多2次重试间隔用指数退避。import httpx from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(2), waitwait_exponential(multiplier1, min2, max10)) async def call_llm_api(prompt): async with httpx.AsyncClient(timeout30.0) as client: response await client.post(LLM_API_URL, json{prompt: prompt}) response.raise_for_status() return response.json()外部API的401错误一般是Key配置问题检查环境变量有没有正确加载。400错误里如果提到“organization has been disabled”说明账号状态有问题需要联系服务方。8.2 向量库的选型与调优向量库我主要用Chroma和Milvus。Chroma适合中小规模百万级以下部署简单和LangChain集成好。Milvus适合大规模千万级以上支持分布式和多种索引类型。选型时主要看数据量和查询并发。索引类型上HNSW适合高召回场景IVF适合高吞吐场景。我一般先用HNSW召回率不够再调参。HNSW的关键参数是M和efConstructionM控制图的连接度efConstruction控制建索引时的搜索深度。M设16到32efConstruction设100到200大多数场景够用。8.3 缓存策略缓存能大幅降低延迟和成本。我一般做三层缓存embedding缓存相同文本不重复向量化、检索结果缓存相同query不重复检索、模型响应缓存相同prompt不重复调用。缓存用Redis设置合理的过期时间。embedding缓存可以设长一点24小时检索结果缓存设短一点1小时模型响应缓存看场景5分钟到1小时。注意模型响应缓存要慎用因为同样的prompt在不同上下文下可能需要不同的响应。我一般只对确定性高的场景开响应缓存比如FAQ问答。8.4 监控与告警配置监控指标我分四类业务指标调用量、用户数、token消耗、性能指标延迟P50/P95/P99、吞吐量、质量指标检索命中率、工具调用成功率、用户满意度、资源指标CPU、内存、GPU利用率。每类指标设不同的告警阈值业务指标看趋势性能指标看突增质量指标看下降资源指标看瓶颈。告警渠道我一般用邮件加即时通讯工具严重告警加电话。告警信息要包含指标名称、当前值、阈值、时间范围、可能原因和建议操作。我踩过的坑是告警太多导致麻木后来做了告警聚合和降噪同类告警5分钟内只发一次。这套系统搭下来我最深的体会是RAG、记忆、API、MCP每一个单独做都不难难的是让它们协同工作并且在鉴权审计的框架下稳定运行。我见过太多项目在demo阶段跑得很好一上生产就各种问题根本原因就是没有把上下文管理、权限控制、可观测性这些“非功能需求”当回事。如果你正在搭类似的系统建议先把鉴权和审计的骨架搭好再往里填RAG和记忆的逻辑这样后期扩展会轻松很多。
网站建设高端定制企业官网