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

资讯详情

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

大模型上下文与工具链搭建:RAG、记忆、API、MCP与鉴权审计实战

大模型上下文与工具链搭建:RAG、记忆、API、MCP与鉴权审计实战 1. 从标题拆解这套系统的真实骨架1.1 为什么是上下文工具链而不是单纯的RAG看到大模型上下文与工具链搭建这个说法很多人第一反应就是又一个RAG教程。但把标题拆开看RAG只是四个关键词里的一个后面还跟着记忆、API、MCP最后落到带鉴权审计应用实践。这说明它要解决的不是怎么让模型答得准这一个问题而是怎么让一个模型驱动的应用在真实环境里跑起来、记得住、调得动外部能力、还能被管住这一整套问题。我做过几个类似的项目最大的体会是单点技术从来不是瓶颈拼装才是。RAG检索增强能解决知识时效性和私域知识注入但它解决不了用户上句话说了什么记忆机制能解决多轮对话的连贯性但它解决不了这个用户有没有权限查这条数据API能打通外部系统但裸奔的API调用就是灾难MCP作为模型与工具之间的标准化协议能把工具接入这件事从每个模型写一套适配变成写一次到处能用。这四样东西叠在一起才构成一个能上生产的上下文系统。所以这篇内容适合谁看如果你已经跑通过一个最简RAG demo但一到多轮对话就露馅、一接外部工具就乱套、一上线就担心安全和审计那这套组合拳就是给你准备的。如果你是完全零基础也能看但建议先把RAG的基本检索流程跑一遍再回来不然会有点吃力。1.2 四个模块各自的职责边界先把职责划清楚后面搭起来才不会互相打架。我习惯用一张表把边界钉死模块核心职责不负责什么典型失败表现RAG注入外部知识提升事实准确性不负责对话连贯、不负责权限答非所问、检索到无权限文档记忆维持长短期上下文连贯不负责知识准确性忘记前文、重复提问API连接外部业务系统与数据源不负责编排、不负责鉴权策略401、超时、限流打爆MCP标准化工具接入与调用不负责具体业务逻辑工具描述歧义、参数错配这张表看着简单但实际项目里最常见的翻车就是职责越界。比如有人把权限判断塞进RAG的检索环节结果检索器要依赖用户身份缓存全废也有人把长期记忆直接当知识库用导致模型把三个月前的临时结论当成事实反复引用。边界清晰是这套系统能维护的前提。1.3 鉴权审计为什么必须从第一天就设计标题最后落在带鉴权审计这是最容易被忽略、也最不该忽略的部分。我见过太多项目前期为了跑通功能API key硬编码在代码里工具调用不记录等要上线了才发现谁在什么时候调了什么工具、拿到了什么数据一概查不到。这时候再补审计等于把整个调用链重写一遍。正确的做法是鉴权审计不是一层外挂而是贯穿在每一次模型调用、每一次检索、每一次工具执行里的横切关注点。用户身份从入口进来一路透传到RAG的过滤条件、记忆的命名空间、API的凭证选择、MCP工具的权限校验最后所有关键动作落审计日志。这条链路设计好了后面加功能就是往里插而不是推倒重来。2. 核心模块的选型逻辑与实现细节2.1 RAG检索增强从朴素向量到混合检索RAG这块新手最容易踩的坑就是一切皆向量。把所有文档切块、embedding、丢进向量库然后指望它精准召回。实测下来纯向量检索在专有名词、编号、代码片段这类场景上召回率很难看。我的做法是混合检索向量召回负责语义相似关键词召回BM25或倒排负责精确匹配两路结果用RRFReciprocal Rank Fusion融合。具体参数上切块大小我一般从512 token起步重叠128 token。为什么是这个数太小了语义不完整太大了检索精度下降且浪费上下文窗口。重叠是为了防止关键信息正好被切在边界上。这个值不是拍脑袋是要拿真实query去测的我通常会准备50到100条真实问题做召回率评估看hit rate能不能到85%以上达不到就调切块策略。关于rag知识库能存储图片嘛这个高频疑问答案是能但方式有讲究。图片本身不进向量库进库的是图片的文本描述或OCR结果。常见做法是用多模态模型给图片生成caption把caption和图片URL一起存检索命中后把URL返回给前端渲染。如果图片里有大量文字先走OCR再切块效果比直接caption好。还有一个绕不开的话题是GraphRAG和本体RAG。普通RAG回答某文档里说了什么很在行但回答这几个实体之间是什么关系就吃力。GraphRAG的思路是先抽实体和关系建图检索时走图遍历。代价是构建成本高、更新麻烦。我的建议是如果你的问题大量涉及多跳关系推理再上GraphRAG否则普通混合检索加个好点的rerank模型性价比高得多。2.2 记忆机制长短期分层与时间衰减记忆这块热词里长短期记忆网络双网络记忆模型记忆score时间半衰期其实指向同一个核心问题怎么让模型既记得住近期对话又不被陈旧信息干扰。我的分层方案是这样的短期记忆当前会话的最近N轮对话直接进上下文窗口。N取决于模型窗口大小和单轮长度一般保留最近10到20轮。工作记忆当前任务相关的中间结论、工具返回结果用结构化格式暂存任务结束可丢弃。长期记忆跨会话的用户偏好、历史结论、重要事实持久化存储检索时按相关性注入。长期记忆的检索不能只按语义相似度排必须加时间衰减。热词里那个score时间半衰期的公式很实用我一般写成final_score similarity * exp(-λ * Δt)其中Δt是记忆产生到现在的时间差λ是衰减系数。λ怎么定如果业务变化快比如股票、新闻λ取大一点半衰期设几天如果是用户偏好这种稳定信息λ取小半衰期可以到几个月。这个参数一定要可配置不同记忆类型用不同衰减曲线。一台电脑上的记忆配置怎么用到另一台这个问题本质是记忆的可移植性。我的做法是把记忆存储和配置分离记忆数据存外部数据库或文件配置衰减系数、命名空间、注入策略用版本化的配置文件管理。换机器时导出数据加配置文件导入即可。千万别把记忆写死在本地缓存里那等于给自己挖坑。2.3 API接入凭证管理与错误处理API这块热词里那一堆401、400错误太真实了。unexpected status 401 unauthorized: incorrect api key provided这个报错十有八九是key过期、环境变量没加载、或者复制时带了空格。我排查这类问题的顺序是先打印key的前后各4位确认没截断再确认环境变量在当前进程可见最后确认key对应的账号没被禁用。api error: 400 this models maximum context length is 1048576 tokens这种是上下文超限。1048576 token看着很大但如果你把整个知识库往prompt里塞照样爆。解决办法就是RAG——先检索再注入而不是全量塞。这也是为什么RAG和API接入必须一起设计。凭证管理我的原则是三条绝不硬编码、按环境隔离、最小权限。生产key和测试key分开每个key只开必要的权限范围。调用外部API时统一走一个封装层在这一层做重试、超时、限流、日志。重试策略用指数退避超时按接口特性设一般3到10秒。限流要防的是自己把对方打爆本地做个令牌桶。调用国产模型API比如DeepSeek、智谱、讯飞星火时注意它们的鉴权方式和OpenAI不完全一样有的用Bearer有的要签名。封装层要能适配不同provider别把某家的调用方式写死在业务代码里。2.4 MCP协议工具接入的标准化MCP是这四个模块里最新、也最容易被误解的。热词里有人问mcp是软件协议还是硬件协议明确说MCP是软件层的协议全称Model Context Protocol解决的是模型怎么标准化地发现和调用外部工具、资源、提示模板。它跟硬件没关系跟接口协议是同一类概念。为什么需要MCP在没有它之前每接一个工具你都要为每个模型写一套适配代码。MCP把这个适配抽象成标准工具提供方实现一个MCP server暴露工具列表和调用接口模型侧实现MCP client按标准协议发现和调用。写一次多个模型都能用。实际搭建时MCP server可以用stdio或SSE/WebSocket传输。本地工具用stdio最简单远程工具用SSE。工具描述一定要写清楚因为模型是靠描述来决定调不调、怎么调的。描述里要包含工具做什么、参数含义、什么场景用、什么场景别用。我见过太多工具调用失败根因就是描述太模糊模型猜错了参数。关于browser use mcp和playwright mcp有什么区别简单说playwright mcp是把浏览器自动化能力标准化暴露给模型模型能精确控制点击、输入、导航browser use类工具更偏向让模型自主决策操作序列。前者可控性强适合确定性任务后者灵活但容易跑偏。生产环境我倾向playwright mcp这种可控的。3. 完整搭建流程与关键环节实现3.1 环境准备与依赖清单先把地基打好。我用的技术栈是Python为主因为生态最全。核心依赖pip install fastapi uvicorn pip install langchain langchain-community pip install chromadb pip install openai pip install pydantic pip install tiktoken向量库我选Chroma本地开发够用要上生产可以换Milvus或Qdrant。embedding模型用本地部署的bge系列省钱且数据不出域。如果要用云端embedding注意把key放进环境变量。目录结构我习惯这样组织project/ config/ settings.yaml memory_policy.yaml rag/ ingest.py retriever.py memory/ short_term.py long_term.py tools/ mcp_client.py api_wrapper.py auth/ guard.py audit.py app/ main.py这个结构的好处是每个模块独立可测鉴权和审计单独成层方便横切。3.2 RAG索引构建与检索链路索引构建分三步加载、切块、入库。from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings splitter RecursiveCharacterTextSplitter( chunk_size512, chunk_overlap128, separators[\n\n, \n, 。, , , , ] ) embeddings HuggingFaceEmbeddings(model_nameBAAI/bge-large-zh-v1.5) vectorstore Chroma( collection_namekb_main, embedding_functionembeddings, persist_directory./chroma_db )注意separators里我把中文标点加进去了中文文档按句号切比按空格切合理得多。切完块每块要带上元数据来源文档、页码、权限标签。权限标签是后面鉴权过滤的关键别省。检索链路我做成两路融合def hybrid_retrieve(query, user_roles, top_k5): vec_results vectorstore.similarity_search(query, ktop_k*2) kw_results bm25_search(query, ktop_k*2) fused rrf_fuse(vec_results, kw_results) filtered [r for r in fused if has_permission(r, user_roles)] return rerank(query, filtered)[:top_k]这里有个关键点权限过滤放在融合之后、rerank之前。为什么如果先过滤再检索检索池被缩小召回率下降先检索再过滤虽然多算了一些但召回质量有保证。当然如果权限粒度很细导致过滤掉大量结果就得考虑在检索阶段就带过滤条件这是权衡。3.3 记忆系统的落地实现短期记忆直接用会话ID做key存最近N轮class ShortTermMemory: def __init__(self, max_turns15): self.store {} self.max_turns max_turns def add(self, session_id, role, content): self.store.setdefault(session_id, []).append( {role: role, content: content} ) if len(self.store[session_id]) self.max_turns * 2: self.store[session_id] self.store[session_id][-self.max_turns*2:] def get(self, session_id): return self.store.get(session_id, [])长期记忆带时间衰减存数据库import math, time def memory_score(similarity, created_at, half_life_days30): delta_days (time.time() - created_at) / 86400 decay math.exp(-math.log(2) * delta_days / half_life_days) return similarity * decay半衰期30天意味着30天后权重降到一半。这个值按业务调用户偏好可以设90天临时结论设7天。检索长期记忆时先按语义召回候选再用这个公式重排取top注入上下文。记忆的命名空间要按用户隔离namespace fuser_{user_id}防止A用户的记忆串到B用户。这是隐私底线不能马虎。3.4 MCP工具接入与API封装MCP client连接server发现工具class MCPClient: def __init__(self, server_config): self.config server_config self.tools {} async def discover(self): resp await self._request(tools/list) for tool in resp[tools]: self.tools[tool[name]] tool async def call(self, name, arguments, user_ctx): if not self._authorized(name, user_ctx): raise PermissionError(ftool {name} not allowed) audit_log(user_ctx, tool_call, name, arguments) return await self._request(tools/call, { name: name, arguments: arguments })注意call里先鉴权再审计再执行顺序不能乱。审计要在执行前记录意图执行后再记录结果这样即使执行崩溃也有迹可循。API封装层统一处理重试和错误import httpx from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) async def call_external_api(url, payload, headers): async with httpx.AsyncClient(timeout10.0) as client: resp await client.post(url, jsonpayload, headersheaders) if resp.status_code 401: raise AuthError(凭证失效检查key) resp.raise_for_status() return resp.json()401单独抛出来因为它是配置问题不是网络问题重试没意义要立刻告警。3.5 鉴权审计的横切实现鉴权用中间件统一拦截把用户身份解析出来放进请求上下文from fastapi import Request, HTTPException async def auth_middleware(request: Request, call_next): token request.headers.get(Authorization, ).replace(Bearer , ) user verify_token(token) if not user: raise HTTPException(status_code401, detailunauthorized) request.state.user user response await call_next(request) return response审计日志结构化落盘字段至少包含时间戳、用户ID、会话ID、动作类型、目标资源、参数摘要、结果状态。参数摘要要脱敏别把敏感数据原样记进去。def audit_log(user, action, target, params, resultpending): record { ts: time.time(), user_id: user.id, session_id: user.session_id, action: action, target: target, params_digest: hash_sensitive(params), result: result } audit_sink.write(record)审计日志要单独存储、只追加、定期归档。别和业务库混在一起否则查询慢还容易被误删。4. 常见问题排查与避坑实录4.1 鉴权与API错误速查报错根因解决401 incorrect api keykey错误/过期/带空格打印key前后4位核对检查环境变量400 maximum context length上下文超限走RAG检索注入别全量塞400 organization disabled账号被禁用联系账号管理员工具调用参数错配工具描述模糊补全描述加参数示例检索召回率低切块/embedding不合适调切块换embedding加关键词路401这类问题我踩过最坑的一次是本地测试好好的一上容器就401。查了半天发现是容器环境变量没注入代码读了个空字符串。所以现在我的启动脚本第一件事就是校验关键环境变量存在不存在直接启动失败别让它带着空key跑起来。4.2 RAG效果不达标的排查顺序召回率低按这个顺序查先看切块是不是把关键信息切碎了再看embedding模型是不是不适合中文再看是不是纯向量没加关键词路最后看rerank模型是不是拖后腿。我一般会写个小脚本拿20条真实query跑一遍打印每条的召回结果肉眼一看就知道问题在哪。rag瓶颈这个词很准RAG的瓶颈往往不在检索算法而在数据质量。文档本身格式乱、有大量表格图片、有重复内容检索再好也白搭。所以索引构建前一定要做数据清洗这一步花的时间往往比调模型多。4.3 记忆串味与上下文膨胀记忆最常见的两个问题串味和膨胀。串味是命名空间没隔离A用户的记忆被B用户检索到。解决办法是检索时强制带namespace过滤且这个过滤在向量库层面做别在应用层做。膨胀是长期记忆越积越多每次注入一大堆把上下文窗口占满。解决办法是设上限每次最多注入top 3到5条且加时间衰减让旧记忆自然沉底。定期还要做记忆压缩把多条相关记忆合并成一条摘要。4.4 MCP工具调用的稳定性MCP工具调用不稳定八成是工具描述的问题。模型是靠描述决定调不调的描述里没写清楚什么时候用模型就会乱调或漏调。我的经验是每个工具描述里必须包含功能一句话、参数逐个说明、一个正例、一个反例。反例尤其重要告诉模型什么情况别调这个工具。还有一点工具调用要设超时和熔断。外部工具挂了不能把整个对话卡死超时后返回降级结果让模型基于已有信息继续。5. 上线前的检查清单与个人经验5.1 上线前必须过的几道关第一道鉴权链路全通。从入口token到RAG过滤、记忆命名空间、工具权限每一环都要有测试用例覆盖。我一般会写一个越权测试用低权限用户去尝试访问高权限资源确认全部被拦。第二道审计日志完整。随便跑一轮对话检查审计日志里能不能还原出谁、什么时候、调了什么、拿到什么。还原不出来就是日志字段缺了。第三道错误处理兜底。把外部API的key故意改错看系统是不是优雅报错而不是崩溃把向量库停掉看RAG失败时是不是降级到纯对话而不是整个挂掉。第四道上下文预算。算一下最坏情况下单次请求的token消耗系统提示短期记忆长期记忆检索结果工具返回加起来别超过模型窗口的70%留30%给输出和波动。5.2 我踩过的几个真实坑第一个坑是embedding模型和检索query不同源。索引用的模型A查询用的模型B向量空间对不上召回全是噪声。这个错误很隐蔽因为不报错只是效果差。记住索引和查询必须用同一个embedding模型。第二个坑是MCP工具的参数类型。模型有时候会把数字传成字符串工具端没做类型转换就崩了。解决办法是在MCP client层做参数校验和类型强制转换别指望模型每次都传对。第三个坑是审计日志写成了同步阻塞。每次工具调用都同步写磁盘高并发下直接把响应时间拖垮。改成异步队列写入日志落盘和业务响应解耦。第四个坑是记忆的时间衰减参数全局统一。结果用户偏好这种长期信息被快速衰减掉了模型老是忘记用户说过喜欢什么。后来改成按记忆类型配置不同半衰期问题解决。5.3 后续可以怎么扩展这套骨架搭好之后扩展方向其实很多。比如把RAG的rerank换成更小的交叉编码器做本地推理省调用成本把记忆做成可解释的让用户能看到系统记住了什么、能手动删除把MCP工具做成插件市场团队里谁都能贡献工具把审计日志接上可视化面板实时看调用分布和异常。我个人最想加的是记忆的主动遗忘机制。现在是被动衰减未来可以让模型自己判断某条记忆是否还有价值主动清理。不过这涉及模型对自身记忆的元认知还在探索阶段。最后分享一个实操小技巧搭这套系统时先别追求四个模块全上。我的建议顺序是RAG打底跑通检索再加记忆解决多轮再加API打通外部最后上MCP和鉴权审计。每加一层都做回归测试确保前面的功能没被破坏。一次性全上出了问题你都不知道是哪层的锅。这个顺序是我踩了无数次坑之后总结出来的能帮你省下大量排查时间。
返回列表