拓冰网站定制 
1. 从标题拆解这套系统的真实骨架1.1 标题里藏着的五个独立模块看到大模型上下文与工具链搭建基于RAG、记忆、API和MCP构建带鉴权审计应用实践这个标题我第一反应是这不是一个单点技术而是一套完整的工程体系。拆开来看它至少包含五个可以独立成篇的模块——RAG检索增强、记忆系统、API网关、MCP工具协议、鉴权审计。这五个东西单独拎出来都不难难的是把它们串成一条能跑通的链路而且还要保证安全可控。我自己在搭建类似系统的时候最大的体会是很多人一上来就冲着我要做个知识库问答去结果RAG还没调明白就急着接工具调用最后整个系统变成一团乱麻。正确的做法应该是先把每个模块的边界划清楚再考虑它们之间怎么通信。这套系统的核心价值在于它让大模型不再是一个只会聊天的黑盒而是一个能查资料、能记事、能调工具、而且每一步操作都有据可查的数字员工。1.2 为什么是这五个模块的组合先说RAG。大模型的训练数据有截止日期而且不可能包含你公司内部的文档。RAG的作用就是在模型生成回答之前先从你的知识库里检索相关内容把检索结果作为上下文喂给模型。这样模型回答时就有据可依而不是凭空编造。热词里提到的rag知识库能存储图片嘛其实是个很实际的问题——标准RAG主要处理文本图片需要先做多模态嵌入或者OCR转文本这是后话。再说记忆。RAG解决的是外部知识的问题记忆解决的是对话连续性的问题。没有记忆的对话就像金鱼每轮都从零开始。记忆系统通常分短期记忆当前会话的上下文窗口和长期记忆跨会话持久化的用户偏好、历史事实。热词里记忆score时间半衰期这个说法很精准长期记忆不能无限堆积需要按重要性和时效性做衰减。API是模型能力的出口。无论是调用大模型本身还是调用外部服务搜索、数据库、业务系统都需要统一的API层来管理。热词里大量出现的unexpected status 401 unauthorized: incorrect api key provided说明鉴权是API层最容易踩的坑。MCP是工具调用的标准化协议。它让模型能以统一的方式发现和调用外部工具不用为每个工具写一套适配代码。热词里mcp协议和mcp是软件协议还是硬件协议的疑问说明这个概念还在普及阶段——MCP是软件层的通信协议跟硬件无关。鉴权审计是贯穿始终的安全底座。每一次模型调用、每一次工具执行、每一次知识库检索都要记录谁在什么时候做了什么并且验证调用方有没有权限。1.3 这套系统适合谁如果你是一个后端工程师想给自己的产品加上AI能力这套架构可以直接参考。如果你是一个AI应用开发者正在被模型胡说八道对话没有连续性工具调用混乱这些问题困扰这篇文章里的方案能帮你理清思路。如果你是一个技术负责人需要评估AI应用的落地成本和安全风险鉴权审计那部分值得重点看。哪怕你只是对RAG和MCP好奇想动手搭一个能跑的原型下面的步骤也可以直接抄。2. 整体架构设计与选型逻辑2.1 分层架构为什么不能把所有东西塞进一个服务我见过太多项目把RAG检索、模型调用、工具执行全写在一个Python文件里初期跑得挺欢一旦要加鉴权、要换模型、要接新工具就改不动了。这套系统的设计原则是分层解耦从上到下大致分四层接入层负责接收用户请求做身份认证和权限校验记录审计日志。编排层负责决定这一轮对话要不要检索、要不要调工具、要不要读写记忆相当于大脑的调度中心。能力层包含RAG检索服务、记忆服务、MCP工具网关、大模型API客户端每个能力独立部署、独立扩缩容。存储层向量数据库存知识库嵌入关系数据库存记忆和审计日志对象存储存原始文档。这样分层的好处是任何一层出问题都不会拖垮整个系统。比如向量数据库挂了RAG检索降级为不检索直接回答对话还能继续模型API超时了可以切换到备用模型。如果全塞在一起一个环节卡住整个请求就挂了。2.2 RAG方案选型朴素RAG还是Agentic RAG热词里同时出现了rag教程rag实战agentic raggraphragontology rag说明RAG本身也在进化。我的建议是从朴素RAG起步按需升级。朴素RAG的流程是文档切块→嵌入→存向量库→查询时检索Top-K→拼进Prompt。这套流程简单可靠80%的场景够用。它的瓶颈在于切块会丢失上下文检索靠语义相似度可能召回不相关内容热词里rag瓶颈说的就是这些。Agentic RAG是在朴素RAG基础上加了一层检索决策——模型先判断这个问题需不需要检索、需要检索什么甚至可以多轮检索、自我反思检索结果。GraphRAG则是把知识组织成图结构适合处理实体关系复杂的场景。Ontology RAG更进一步用本体论来约束知识表示。但我要泼一盆冷水不要一上来就上GraphRAG。图构建和维护的成本很高如果你的文档量不大、关系不复杂朴素RAG加一个好的重排序模型效果可能更好。选型的判断标准很简单如果你的问题经常涉及A和B是什么关系通过C能找到哪些D考虑图如果只是某文档里说了什么朴素RAG足够。2.3 记忆系统设计双网络记忆模型的落地热词里双网络记忆模型和agent记忆指向同一个问题Agent怎么记住东西。我的设计是把记忆分成两个网络事实记忆网络存储结构化的、相对稳定的事实。比如用户偏好中文回答用户所在团队是数据组上次讨论的项目代号是X。这些事实以键值对或三元组形式存在关系数据库里查询时精确匹配。情景记忆网络存储对话历史中的关键片段以向量形式存在向量库里。查询时用语义相似度召回。热词里记忆score时间半衰期说的就是情景记忆的排序公式最终得分 语义相似度得分 × 时间衰减因子。时间衰减因子通常用指数衰减半衰期设7天到30天不等看业务对时效性的要求。短期记忆就是当前会话的上下文窗口直接拼在Prompt里。这里有个工程细节上下文窗口有限热词里maximum context length is 1048576 tokens说明现在窗口很大了但依然有限所以短期记忆要做滑动窗口或摘要压缩。我的做法是保留最近N轮完整对话更早的对话做摘要后保留。2.4 MCP工具协议为什么不用传统的Function Calling传统Function Calling是每个模型厂商自己定义的格式OpenAI一套、Anthropic一套、国内厂商又一套。你写好的工具定义换个模型就得改。MCPModel Context Protocol的价值在于标准化——工具提供方按MCP规范暴露能力模型侧按MCP规范调用两边解耦。热词里playwright mcpchrome devtools mcpbrowser use mcp跟playwright mcp有什么区别说明MCP生态已经在工具层面铺开了。Playwright MCP让模型能操控浏览器Chrome DevTools MCP让模型能调试网页这些工具通过MCP协议接入后模型就能像调用本地函数一样调用它们。MCP的通信方式支持stdio和SSE/WebSocket。本地工具用stdio远程工具用SSE。热词里那个wss://api.xiaozhi.me/mcp/?token...就是远程MCP服务的WebSocket地址token用于鉴权。这里要特别注意MCP的token要当作敏感凭证管理不能硬编码在代码里要走密钥管理服务。2.5 鉴权审计安全底座怎么搭鉴权审计不是事后补的是设计之初就要考虑的。我的方案是三层第一层接入鉴权。每个请求带API Key或JWT网关验证签名和有效期。热词里incorrect api key provided: sk-svcac****这种错误就是这一层拦下来的。API Key要支持轮换和吊销不能一个Key用到底。第二层操作鉴权。验证通过身份后还要看这个身份有没有权限执行这个操作。比如普通用户只能检索公开知识库管理员才能写入知识库某些敏感工具只有特定角色能调用。这层用RBAC基于角色的访问控制实现。第三层审计日志。每一次模型调用、工具执行、知识库读写都记录时间戳、调用方身份、操作类型、输入摘要、输出摘要、耗时、是否成功。审计日志要写入独立的存储不能和业务数据混在一起而且要防篡改。3. 核心模块的实操细节3.1 RAG检索服务的搭建要点文档切块是RAG的第一个坑。切太大检索精度低切太小上下文丢失。我的经验值是中文文档按300-500字切英文按200-400词切块之间保留10%-20%的重叠。重叠的作用是防止关键信息正好被切在边界上。嵌入模型的选择上如果追求效果且预算充足用商用嵌入API如果追求成本和数据隐私用本地开源嵌入模型。热词里ollama 简易本地rag知识库就是本地方案的典型。Ollama跑嵌入模型配合本地向量库如Chroma、Qdrant整套不出内网。向量库的选型看规模十万级向量用Chroma或FAISS足够百万级考虑Qdrant或Milvus千万级以上要上分布式方案。索引类型上HNSW查询快但内存占用高IVF节省内存但需要训练小规模直接用扁平索引也行。检索环节我强烈建议加重排序。先用向量检索召回Top-20再用重排序模型如BGE-Reranker精排取Top-5。这一步能把rag hit rate提升一大截。重排序模型比嵌入模型小推理成本可接受。# RAG检索核心流程示意 def retrieve(query, top_k5): # 1. 查询改写可选提升召回 rewritten rewrite_query(query) # 2. 向量检索召回 candidates vector_store.search(rewritten, top_k20) # 3. 重排序精排 reranked reranker.rank(query, candidates) # 4. 取Top-K并组装上下文 contexts [c.text for c in reranked[:top_k]] return contexts注意查询改写这一步很多人忽略但对于口语化提问效果提升明显。比如用户问那个报销的事咋弄改写成报销流程是什么再检索命中率会高很多。3.2 记忆系统的读写时机记忆系统最难的不是存储是什么时候写、什么时候读。写入时机对话结束后异步写入不要阻塞主流程。写入前做一次记忆提取让模型判断这轮对话里有没有值得长期记住的信息。不是每轮对话都值得记闲聊就不用记。提取出的记忆要分类事实类进事实网络情景类进情景网络。读取时机每轮对话开始时先查事实记忆精确匹配用户ID再查情景记忆用当前问题做语义检索。两路结果合并后按得分排序取Top-N拼进系统提示。记忆的更新和删除同样重要。事实记忆如果变了比如用户换了团队要覆盖旧值。情景记忆要支持按时间衰减太老的记忆自动降权。热词里管理workbuddy的记忆一台电脑上的记忆配置如何用到另一台说的就是记忆的可移植性——记忆要能导出导入格式要标准化。# 记忆读取示意 def load_memory(user_id, query): # 事实记忆精确匹配 facts fact_store.get_by_user(user_id) # 情景记忆语义检索 时间衰减 episodes episode_store.search(query, user_iduser_id) now time.time() for ep in episodes: age_days (now - ep.timestamp) / 86400 decay 0.5 ** (age_days / HALF_LIFE_DAYS) ep.score ep.similarity * decay episodes.sort(keylambda x: x.score, reverseTrue) return facts, episodes[:5]3.3 API网关的鉴权实现API网关是所有请求的入口鉴权在这里做最合适。我的实现是每个调用方分配一对access_key和secret_key。请求时带access_key和签名签名算法用HMAC-SHA256签名内容包含请求方法、路径、时间戳、请求体哈希。网关验证签名和时间戳时间戳偏差超过5分钟拒绝防重放。import hmac, hashlib, time def sign_request(secret_key, method, path, body): timestamp str(int(time.time())) body_hash hashlib.sha256(body.encode()).hexdigest() message f{method}\n{path}\n{timestamp}\n{body_hash} signature hmac.new( secret_key.encode(), message.encode(), hashlib.sha256 ).hexdigest() return timestamp, signature注意热词里反复出现的401错误九成是签名或Key的问题。排查顺序是先确认Key有没有过期再确认签名算法对不对最后确认时间戳有没有偏差。我踩过的坑是服务器时间没同步导致时间戳校验一直失败查了半天才发现是NTP没配。权限校验用RBAC。角色定义在数据库里每个角色绑定一组权限每个权限对应一个资源操作。网关验证完签名后查这个access_key对应的角色再查角色有没有目标操作的权限。3.4 MCP工具接入的完整流程MCP工具的接入分三步发现、注册、调用。发现MCP服务端会暴露一个工具列表接口客户端启动时拉取得到每个工具的名称、描述、参数schema。这个schema是JSON Schema格式模型能直接理解。注册把工具列表转换成模型能识别的格式注入到系统提示或工具定义里。如果用支持Function Calling的模型就转成对应的格式如果用MCP原生协议就直接传。调用模型决定调用某个工具后客户端按MCP协议发送调用请求服务端执行后返回结果。结果要回填到对话上下文里让模型基于结果继续生成。{ name: search_knowledge_base, description: 在知识库中搜索相关内容, inputSchema: { type: object, properties: { query: {type: string, description: 搜索关键词}, top_k: {type: integer, default: 5} }, required: [query] } }MCP工具调用的鉴权token放在连接URL或请求头里。远程MCP服务建议用短时效token定期刷新。本地stdio方式没有网络传输但也要验证调用方身份防止本地其他进程冒用。3.5 审计日志的字段设计与存储审计日志的字段我建议至少包含这些字段类型说明trace_idstring全链路追踪ID贯穿一次请求的所有操作timestampdatetime操作发生时间精确到毫秒actor_idstring调用方身份标识actor_rolestring调用方角色actionstring操作类型llm_call/rag_search/tool_call/memory_writeresourcestring操作对象模型名/知识库名/工具名input_digeststring输入摘要脱敏后output_digeststring输出摘要脱敏后duration_msinteger耗时毫秒statusstringsuccess/failure/timeouterror_codestring失败时的错误码存储上审计日志写入独立的表或独立的库只允许追加不允许修改。定期归档冷数据热数据保留30-90天。查询接口要支持按trace_id、actor_id、时间范围检索。注意审计日志里的输入输出摘要要做脱敏不能把用户的敏感信息原样记进去。我的做法是只记前100个字符加哈希值既能追溯又不泄露。4. 实操过程与关键环节实现4.1 环境准备与依赖安装先列一下这套系统需要的核心组件Python 3.103.11更稳3.12部分库还没跟上向量数据库QdrantDocker一键起或Chroma嵌入式零配置关系数据库PostgreSQL存记忆和审计或SQLite原型阶段够用嵌入模型本地用Ollama跑bge-m3或用商用API大模型API任意兼容OpenAI格式的APIMCP SDK官方Python SDK# 起Qdrant docker run -d -p 6333:6333 -v $(pwd)/qdrant_data:/qdrant/storage qdrant/qdrant # 起PostgreSQL docker run -d -p 5432:5432 -e POSTGRES_PASSWORDyourpass -v $(pwd)/pg_data:/var/lib/postgresql/data postgres:16 # 安装Python依赖 pip install qdrant-client psycopg2-binary openai mcp ollama环境变量管理用.env文件配合python-dotenv加载。所有密钥、连接串都走环境变量代码里不出现明文。# .env 示例 LLM_API_KEYsk-xxxx LLM_BASE_URLhttps://api.example.com/v1 QDRANT_URLhttp://localhost:6333 PG_DSNpostgresql://postgres:yourpasslocalhost:5432/ai_app MCP_SERVER_URLwss://mcp.example.com/mcp MCP_TOKENeyJhbGci...4.2 知识库入库的完整流程入库流程分五步加载、切块、嵌入、存储、索引。加载环节要处理多种格式PDF用PyMuPDF或MinerU解析热词里mineru api就是干这个的Word用python-docxMarkdown直接读网页用trafilatura提取正文。图片里的文字用OCR图片本身如果要检索需要多模态嵌入模型。切块我用的是递归字符切分优先按段落切段落太长再按句子切句子还长就按字符切。分隔符优先级\n\n\n。。from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap80, separators[\n\n, \n, 。, , , , , ], length_functionlen, ) chunks splitter.split_text(document)嵌入环节批量嵌入比逐条嵌入快得多。Qdrant的批量upsert接口一次能传几百条。嵌入时要带上元数据来源文档、页码、章节标题检索时能一起返回方便溯源。from qdrant_client import QdrantClient from qdrant_client.models import PointStruct client QdrantClient(urlhttp://localhost:6333) points [] for i, chunk in enumerate(chunks): vector embed(chunk) # 调用嵌入模型 points.append(PointStruct( idhash(chunk) % (10**9), vectorvector, payload{ text: chunk, source: doc_name, page: page_num, chunk_index: i, } )) client.upsert(collection_nameknowledge, pointspoints)注意嵌入模型的维度要和向量库集合的维度一致建集合时就要指定。换嵌入模型意味着要重建整个集合所以选型时想清楚别中途换。4.3 对话主流程的编排一轮完整对话的编排逻辑是这样的接收用户请求网关鉴权生成trace_id。加载用户的事实记忆和情景记忆。判断是否需要RAG检索可以用规则也可以用模型判断。如果需要执行检索得到上下文。组装Prompt系统提示 记忆 检索上下文 对话历史 用户输入。调用大模型API传入工具定义。如果模型返回工具调用请求执行工具把结果回填再次调用模型。得到最终回答返回给用户。异步写入审计日志和记忆。async def chat(user_id, message, trace_id): # 1. 加载记忆 facts, episodes load_memory(user_id, message) # 2. 判断是否检索 need_rag should_retrieve(message) contexts retrieve(message) if need_rag else [] # 3. 组装Prompt messages build_prompt(facts, episodes, contexts, message) # 4. 调用模型带工具 response await llm_call(messages, toolsget_mcp_tools()) # 5. 处理工具调用循环 while response.has_tool_call(): tool_result await execute_mcp_tool(response.tool_call) messages.append(response.tool_call) messages.append(tool_result) response await llm_call(messages, toolsget_mcp_tools()) # 6. 异步写审计和记忆 asyncio.create_task(write_audit(trace_id, user_id, message, response)) asyncio.create_task(extract_and_save_memory(user_id, message, response)) return response.content工具调用循环要设最大轮次限制防止模型陷入无限调用。我设的是5轮超过就强制返回当前结果并记录异常。4.4 MCP工具服务端的实现MCP服务端可以用官方SDK快速搭。核心是定义工具函数注册到服务端然后启动监听。from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent server Server(my-tools) server.list_tools() async def list_tools(): return [ Tool( namequery_database, description查询业务数据库, inputSchema{ type: object, properties: { sql: {type: string, description: SQL查询语句} }, required: [sql] } ) ] server.call_tool() async def call_tool(name, arguments): if name query_database: # 鉴权检查 if not check_permission(arguments.get(_actor), db_query): return [TextContent(typetext, text权限不足)] # 执行查询注意SQL注入防护 result safe_query(arguments[sql]) return [TextContent(typetext, textstr(result))] async def main(): async with stdio_server() as (read, write): await server.run(read, write, server.create_initialization_options())注意工具服务端是安全重灾区。SQL查询工具必须做白名单或参数化不能让模型直接拼SQL。文件操作工具要限制目录范围。任何工具都要在服务端再做一次权限校验不能只信客户端传来的身份。4.5 鉴权审计的落地细节审计日志的写入用异步队列避免阻塞主流程。我用的是asyncio.Queue加后台消费者批量写入数据库。audit_queue asyncio.Queue() async def audit_consumer(): batch [] while True: try: item await asyncio.wait_for(audit_queue.get(), timeout1.0) batch.append(item) if len(batch) 100: await flush_audit(batch) batch [] except asyncio.TimeoutError: if batch: await flush_audit(batch) batch []审计日志的查询接口要单独做走独立的鉴权只有审计角色能查。查询要支持组合条件并且对结果做分页。鉴权方面API Key的存储要用哈希不能存明文。验证时把传来的Key哈希后比对。Key要支持设置有效期和权限范围过期自动失效。5. 常见问题与排查技巧实录5.1 鉴权类问题速查错误信息可能原因排查方法401 incorrect api keyKey错误/过期/被吊销检查Key是否复制完整是否在有效期内401 unauthorized签名验证失败检查签名算法、时间戳、请求体是否一致403 forbidden权限不足检查角色绑定的权限是否包含目标操作400 organization disabled账号状态异常联系服务方确认账号状态热词里unexpected status 401 unauthorized: incorrect api key provided: sk-svcac****这个错误我遇到最多的情况是Key复制时带了空格或者环境变量没加载成功。排查时先打印一下实际用的Key脱敏后确认和预期一致。5.2 RAG效果差的排查思路RAG效果差通常分三种情况检索不到、检索到但不相关、相关但模型没用上。检索不到先看知识库里有没有相关内容。用原始问题直接搜向量库看Top-20里有没有对的。如果没有可能是切块切坏了或者嵌入模型不适合这个领域。试试换嵌入模型或者调整切块策略。检索到但不相关多半是召回太多噪声。加一个相似度阈值低于阈值的直接丢弃。再加一个重排序模型把真正相关的排上来。相关但模型没用上是Prompt的问题。检查检索结果有没有正确拼进PromptPrompt里有没有明确指示基于以下资料回答。有时候模型会忽略上下文需要在系统提示里强调。5.3 记忆系统的典型故障记忆系统最常见的问题是记忆污染——错误的信息被写入长期记忆之后一直影响对话。我的做法是写入长期记忆前做一次校验让模型判断这条信息是否确定、是否值得长期记住。不确定的信息只存短期。第二个问题是记忆膨胀——记忆越存越多检索越来越慢而且噪声越来越大。解决方法是定期清理超过一定时间且从未被召回的记忆归档或删除相似度极高的重复记忆合并。第三个问题是跨设备同步。热词里一台电脑上的记忆配置如何用到另一台就是这个场景。我的方案是记忆存服务端客户端只做展示和触发不存本地。这样换设备无缝衔接。5.4 MCP工具调用的坑MCP工具调用最容易出问题的地方是参数格式不匹配。模型生成的参数可能缺字段、类型不对、或者多传了字段。服务端要做参数校验不合法就返回明确的错误信息让模型能自我修正。第二个坑是超时。工具执行可能很慢模型等不了。要设超时超时后返回工具执行超时让模型决定是重试还是换方案。第三个坑是工具结果太大。比如查询数据库返回了几万行全塞进上下文会爆token。服务端要对结果做截断或摘要只返回关键信息。def truncate_result(result, max_chars2000): text str(result) if len(text) max_chars: return text return text[:max_chars] f\n...(结果被截断共{len(text)}字符)5.5 上下文长度超限的处理热词里api error: 400 this models maximum context length is 1048576 tokens说明即使窗口很大也有超限的时候。处理策略分三层第一层预防。组装Prompt前估算token数超了就裁剪。裁剪优先级先裁情景记忆再裁检索上下文最后裁对话历史。系统提示和事实记忆不裁。第二层压缩。对话历史太长时把早期对话做摘要用摘要替代原文。摘要用便宜的小模型做成本低。第三层降级。如果实在超限减少检索Top-K或者切换到窗口更大的模型。def estimate_tokens(text): # 中文约1.5字符/token英文约4字符/token取保守估计 return len(text) // 2 def fit_context(messages, max_tokens): while estimate_tokens(str(messages)) max_tokens: # 从最早的对话开始删 if len(messages) 3: messages.pop(1) # 保留system和最新user else: break return messages6. 工具选型与性能优化6.1 向量数据库选型对比数据库部署方式适用规模优点缺点Chroma嵌入式十万级零配置Python原生不适合分布式FAISS库百万级快Facebook出品无持久化要自己管Qdrant服务百万到千万功能全Rust写的快要单独部署Milvus分布式千万级以上扩展性强运维复杂原型阶段我推荐Chroma一行代码起。生产环境推荐Qdrant性能和功能平衡得好。超大规模才考虑Milvus。6.2 嵌入模型的选择嵌入模型直接决定RAG的上限。中文场景我推荐bge-m3多语言支持好维度1024本地跑得动。如果追求极致效果可以用商用嵌入API但成本和数据隐私要权衡。嵌入模型的维度不是越高越好。1024维在大多数场景够用1536维提升有限但存储和计算成本增加。选型时看MTEB榜单但也要在自己的数据上实测。6.3 缓存策略大模型调用是成本和延迟的大头能缓存就缓存。缓存分两级精确缓存相同的Prompt直接返回缓存结果。用Prompt的哈希做Key存Redis。适合FAQ类高频问题。语义缓存语义相似的问题返回缓存结果。用嵌入向量做相似度匹配超过阈值就命中。适合问法多样但意图相同的场景。def get_cached_response(query, threshold0.95): query_vec embed(query) # 精确缓存 exact_key hashlib.md5(query.encode()).hexdigest() if cached : redis.get(exact_key): return cached # 语义缓存 similar semantic_cache.search(query_vec, threshold) if similar: return similar.response return None注意缓存要设过期时间知识库更新后旧缓存要失效。我的做法是知识库更新时清空相关缓存或者给缓存加版本号。6.4 并发与限流大模型API通常有QPS限制超了会报429。要做客户端限流用令牌桶或漏桶算法。同时要控制并发数避免打爆下游。import asyncio from asyncio import Semaphore llm_semaphore Semaphore(10) # 最多10个并发 async def llm_call_with_limit(messages): async with llm_semaphore: return await llm_call(messages)限流之外还要做重试。网络抖动、临时限流都可能失败重试要带指数退避。但注意非幂等操作不能盲目重试比如写操作重试可能重复写入。7. 安全加固与合规要点7.1 输入输出的安全过滤用户输入要过滤注入攻击。Prompt注入是AI应用特有的风险——用户在输入里写忽略之前的指令试图让模型执行非预期操作。防御方法是在系统提示里明确边界同时对输入做检测发现可疑模式就拒绝或转义。模型输出也要过滤。模型可能生成不当内容或者泄露系统提示。输出前过一遍敏感词库和格式校验。7.2 密钥管理所有密钥走密钥管理服务不写在代码和配置文件里。开发环境用.env生产环境用KMS或Vault。密钥要定期轮换轮换时新旧密钥并存一段时间避免服务中断。MCP的token、API Key、数据库密码都属于密钥。审计日志里不能出现密钥明文要脱敏。7.3 数据隔离多租户场景下不同租户的数据要隔离。向量库用payload过滤关系库用行级权限对象存储用前缀隔离。审计日志要记录租户ID方便追溯。知识库的检索要带租户过滤条件防止A租户检索到B租户的文档。这个过滤要在向量检索时就加上不能检索完再过滤否则有信息泄露风险。7.4 审计日志的防篡改审计日志写入后不可修改。技术上可以用只追加的表、WORM存储、或者区块链式的哈希链。哈希链的做法是每条日志包含前一条的哈希任何修改都会破坏链。def write_audit_log(entry, prev_hash): entry[prev_hash] prev_hash entry_str json.dumps(entry, sort_keysTrue) entry[hash] hashlib.sha256(entry_str.encode()).hexdigest() db.insert(entry) return entry[hash]8. 扩展方向与个人体会这套系统跑通之后扩展方向很多。RAG可以升级到Agentic RAG让模型自主决定检索策略。记忆可以引入知识图谱把事实记忆结构化。MCP工具可以接入更多业务系统让模型真正能干活。审计可以加实时告警发现异常调用立即通知。我自己在实际操作中的体会是这套系统最难的不是某个模块的技术实现而是模块之间的协调。RAG检索的结果怎么和记忆融合工具调用的结果怎么回填审计日志怎么贯穿全链路这些胶水工作才是工程量的主体。我的建议是先把主流程跑通哪怕每个模块都用最简实现然后再逐个优化。一上来就追求每个模块都完美大概率会卡在半路。最后分享一个小技巧调试这套系统时把每一轮的完整Prompt和模型原始输出都打到日志里脱敏后出问题时能快速定位是哪一环出的错。我踩过的坑是只记了最终回答结果模型答错了根本不知道是检索错了还是模型理解错了只能重跑浪费大量时间。 大模型上下文与工具链搭建：基于RAG、记忆、API和MCP的鉴权审计应用实践
1. 从标题拆解这套系统的真实骨架1.1 标题里藏着的五个独立模块看到大模型上下文与工具链搭建基于RAG、记忆、API和MCP构建带鉴权审计应用实践这个标题我第一反应是这不是一个单点技术而是一套完整的工程体系。拆开来看它至少包含五个可以独立成篇的模块——RAG检索增强、记忆系统、API网关、MCP工具协议、鉴权审计。这五个东西单独拎出来都不难难的是把它们串成一条能跑通的链路而且还要保证安全可控。我自己在搭建类似系统的时候最大的体会是很多人一上来就冲着我要做个知识库问答去结果RAG还没调明白就急着接工具调用最后整个系统变成一团乱麻。正确的做法应该是先把每个模块的边界划清楚再考虑它们之间怎么通信。这套系统的核心价值在于它让大模型不再是一个只会聊天的黑盒而是一个能查资料、能记事、能调工具、而且每一步操作都有据可查的数字员工。1.2 为什么是这五个模块的组合先说RAG。大模型的训练数据有截止日期而且不可能包含你公司内部的文档。RAG的作用就是在模型生成回答之前先从你的知识库里检索相关内容把检索结果作为上下文喂给模型。这样模型回答时就有据可依而不是凭空编造。热词里提到的rag知识库能存储图片嘛其实是个很实际的问题——标准RAG主要处理文本图片需要先做多模态嵌入或者OCR转文本这是后话。再说记忆。RAG解决的是外部知识的问题记忆解决的是对话连续性的问题。没有记忆的对话就像金鱼每轮都从零开始。记忆系统通常分短期记忆当前会话的上下文窗口和长期记忆跨会话持久化的用户偏好、历史事实。热词里记忆score时间半衰期这个说法很精准长期记忆不能无限堆积需要按重要性和时效性做衰减。API是模型能力的出口。无论是调用大模型本身还是调用外部服务搜索、数据库、业务系统都需要统一的API层来管理。热词里大量出现的unexpected status 401 unauthorized: incorrect api key provided说明鉴权是API层最容易踩的坑。MCP是工具调用的标准化协议。它让模型能以统一的方式发现和调用外部工具不用为每个工具写一套适配代码。热词里mcp协议和mcp是软件协议还是硬件协议的疑问说明这个概念还在普及阶段——MCP是软件层的通信协议跟硬件无关。鉴权审计是贯穿始终的安全底座。每一次模型调用、每一次工具执行、每一次知识库检索都要记录谁在什么时候做了什么并且验证调用方有没有权限。1.3 这套系统适合谁如果你是一个后端工程师想给自己的产品加上AI能力这套架构可以直接参考。如果你是一个AI应用开发者正在被模型胡说八道对话没有连续性工具调用混乱这些问题困扰这篇文章里的方案能帮你理清思路。如果你是一个技术负责人需要评估AI应用的落地成本和安全风险鉴权审计那部分值得重点看。哪怕你只是对RAG和MCP好奇想动手搭一个能跑的原型下面的步骤也可以直接抄。2. 整体架构设计与选型逻辑2.1 分层架构为什么不能把所有东西塞进一个服务我见过太多项目把RAG检索、模型调用、工具执行全写在一个Python文件里初期跑得挺欢一旦要加鉴权、要换模型、要接新工具就改不动了。这套系统的设计原则是分层解耦从上到下大致分四层接入层负责接收用户请求做身份认证和权限校验记录审计日志。编排层负责决定这一轮对话要不要检索、要不要调工具、要不要读写记忆相当于大脑的调度中心。能力层包含RAG检索服务、记忆服务、MCP工具网关、大模型API客户端每个能力独立部署、独立扩缩容。存储层向量数据库存知识库嵌入关系数据库存记忆和审计日志对象存储存原始文档。这样分层的好处是任何一层出问题都不会拖垮整个系统。比如向量数据库挂了RAG检索降级为不检索直接回答对话还能继续模型API超时了可以切换到备用模型。如果全塞在一起一个环节卡住整个请求就挂了。2.2 RAG方案选型朴素RAG还是Agentic RAG热词里同时出现了rag教程rag实战agentic raggraphragontology rag说明RAG本身也在进化。我的建议是从朴素RAG起步按需升级。朴素RAG的流程是文档切块→嵌入→存向量库→查询时检索Top-K→拼进Prompt。这套流程简单可靠80%的场景够用。它的瓶颈在于切块会丢失上下文检索靠语义相似度可能召回不相关内容热词里rag瓶颈说的就是这些。Agentic RAG是在朴素RAG基础上加了一层检索决策——模型先判断这个问题需不需要检索、需要检索什么甚至可以多轮检索、自我反思检索结果。GraphRAG则是把知识组织成图结构适合处理实体关系复杂的场景。Ontology RAG更进一步用本体论来约束知识表示。但我要泼一盆冷水不要一上来就上GraphRAG。图构建和维护的成本很高如果你的文档量不大、关系不复杂朴素RAG加一个好的重排序模型效果可能更好。选型的判断标准很简单如果你的问题经常涉及A和B是什么关系通过C能找到哪些D考虑图如果只是某文档里说了什么朴素RAG足够。2.3 记忆系统设计双网络记忆模型的落地热词里双网络记忆模型和agent记忆指向同一个问题Agent怎么记住东西。我的设计是把记忆分成两个网络事实记忆网络存储结构化的、相对稳定的事实。比如用户偏好中文回答用户所在团队是数据组上次讨论的项目代号是X。这些事实以键值对或三元组形式存在关系数据库里查询时精确匹配。情景记忆网络存储对话历史中的关键片段以向量形式存在向量库里。查询时用语义相似度召回。热词里记忆score时间半衰期说的就是情景记忆的排序公式最终得分 语义相似度得分 × 时间衰减因子。时间衰减因子通常用指数衰减半衰期设7天到30天不等看业务对时效性的要求。短期记忆就是当前会话的上下文窗口直接拼在Prompt里。这里有个工程细节上下文窗口有限热词里maximum context length is 1048576 tokens说明现在窗口很大了但依然有限所以短期记忆要做滑动窗口或摘要压缩。我的做法是保留最近N轮完整对话更早的对话做摘要后保留。2.4 MCP工具协议为什么不用传统的Function Calling传统Function Calling是每个模型厂商自己定义的格式OpenAI一套、Anthropic一套、国内厂商又一套。你写好的工具定义换个模型就得改。MCPModel Context Protocol的价值在于标准化——工具提供方按MCP规范暴露能力模型侧按MCP规范调用两边解耦。热词里playwright mcpchrome devtools mcpbrowser use mcp跟playwright mcp有什么区别说明MCP生态已经在工具层面铺开了。Playwright MCP让模型能操控浏览器Chrome DevTools MCP让模型能调试网页这些工具通过MCP协议接入后模型就能像调用本地函数一样调用它们。MCP的通信方式支持stdio和SSE/WebSocket。本地工具用stdio远程工具用SSE。热词里那个wss://api.xiaozhi.me/mcp/?token...就是远程MCP服务的WebSocket地址token用于鉴权。这里要特别注意MCP的token要当作敏感凭证管理不能硬编码在代码里要走密钥管理服务。2.5 鉴权审计安全底座怎么搭鉴权审计不是事后补的是设计之初就要考虑的。我的方案是三层第一层接入鉴权。每个请求带API Key或JWT网关验证签名和有效期。热词里incorrect api key provided: sk-svcac****这种错误就是这一层拦下来的。API Key要支持轮换和吊销不能一个Key用到底。第二层操作鉴权。验证通过身份后还要看这个身份有没有权限执行这个操作。比如普通用户只能检索公开知识库管理员才能写入知识库某些敏感工具只有特定角色能调用。这层用RBAC基于角色的访问控制实现。第三层审计日志。每一次模型调用、工具执行、知识库读写都记录时间戳、调用方身份、操作类型、输入摘要、输出摘要、耗时、是否成功。审计日志要写入独立的存储不能和业务数据混在一起而且要防篡改。3. 核心模块的实操细节3.1 RAG检索服务的搭建要点文档切块是RAG的第一个坑。切太大检索精度低切太小上下文丢失。我的经验值是中文文档按300-500字切英文按200-400词切块之间保留10%-20%的重叠。重叠的作用是防止关键信息正好被切在边界上。嵌入模型的选择上如果追求效果且预算充足用商用嵌入API如果追求成本和数据隐私用本地开源嵌入模型。热词里ollama 简易本地rag知识库就是本地方案的典型。Ollama跑嵌入模型配合本地向量库如Chroma、Qdrant整套不出内网。向量库的选型看规模十万级向量用Chroma或FAISS足够百万级考虑Qdrant或Milvus千万级以上要上分布式方案。索引类型上HNSW查询快但内存占用高IVF节省内存但需要训练小规模直接用扁平索引也行。检索环节我强烈建议加重排序。先用向量检索召回Top-20再用重排序模型如BGE-Reranker精排取Top-5。这一步能把rag hit rate提升一大截。重排序模型比嵌入模型小推理成本可接受。# RAG检索核心流程示意 def retrieve(query, top_k5): # 1. 查询改写可选提升召回 rewritten rewrite_query(query) # 2. 向量检索召回 candidates vector_store.search(rewritten, top_k20) # 3. 重排序精排 reranked reranker.rank(query, candidates) # 4. 取Top-K并组装上下文 contexts [c.text for c in reranked[:top_k]] return contexts注意查询改写这一步很多人忽略但对于口语化提问效果提升明显。比如用户问那个报销的事咋弄改写成报销流程是什么再检索命中率会高很多。3.2 记忆系统的读写时机记忆系统最难的不是存储是什么时候写、什么时候读。写入时机对话结束后异步写入不要阻塞主流程。写入前做一次记忆提取让模型判断这轮对话里有没有值得长期记住的信息。不是每轮对话都值得记闲聊就不用记。提取出的记忆要分类事实类进事实网络情景类进情景网络。读取时机每轮对话开始时先查事实记忆精确匹配用户ID再查情景记忆用当前问题做语义检索。两路结果合并后按得分排序取Top-N拼进系统提示。记忆的更新和删除同样重要。事实记忆如果变了比如用户换了团队要覆盖旧值。情景记忆要支持按时间衰减太老的记忆自动降权。热词里管理workbuddy的记忆一台电脑上的记忆配置如何用到另一台说的就是记忆的可移植性——记忆要能导出导入格式要标准化。# 记忆读取示意 def load_memory(user_id, query): # 事实记忆精确匹配 facts fact_store.get_by_user(user_id) # 情景记忆语义检索 时间衰减 episodes episode_store.search(query, user_iduser_id) now time.time() for ep in episodes: age_days (now - ep.timestamp) / 86400 decay 0.5 ** (age_days / HALF_LIFE_DAYS) ep.score ep.similarity * decay episodes.sort(keylambda x: x.score, reverseTrue) return facts, episodes[:5]3.3 API网关的鉴权实现API网关是所有请求的入口鉴权在这里做最合适。我的实现是每个调用方分配一对access_key和secret_key。请求时带access_key和签名签名算法用HMAC-SHA256签名内容包含请求方法、路径、时间戳、请求体哈希。网关验证签名和时间戳时间戳偏差超过5分钟拒绝防重放。import hmac, hashlib, time def sign_request(secret_key, method, path, body): timestamp str(int(time.time())) body_hash hashlib.sha256(body.encode()).hexdigest() message f{method}\n{path}\n{timestamp}\n{body_hash} signature hmac.new( secret_key.encode(), message.encode(), hashlib.sha256 ).hexdigest() return timestamp, signature注意热词里反复出现的401错误九成是签名或Key的问题。排查顺序是先确认Key有没有过期再确认签名算法对不对最后确认时间戳有没有偏差。我踩过的坑是服务器时间没同步导致时间戳校验一直失败查了半天才发现是NTP没配。权限校验用RBAC。角色定义在数据库里每个角色绑定一组权限每个权限对应一个资源操作。网关验证完签名后查这个access_key对应的角色再查角色有没有目标操作的权限。3.4 MCP工具接入的完整流程MCP工具的接入分三步发现、注册、调用。发现MCP服务端会暴露一个工具列表接口客户端启动时拉取得到每个工具的名称、描述、参数schema。这个schema是JSON Schema格式模型能直接理解。注册把工具列表转换成模型能识别的格式注入到系统提示或工具定义里。如果用支持Function Calling的模型就转成对应的格式如果用MCP原生协议就直接传。调用模型决定调用某个工具后客户端按MCP协议发送调用请求服务端执行后返回结果。结果要回填到对话上下文里让模型基于结果继续生成。{ name: search_knowledge_base, description: 在知识库中搜索相关内容, inputSchema: { type: object, properties: { query: {type: string, description: 搜索关键词}, top_k: {type: integer, default: 5} }, required: [query] } }MCP工具调用的鉴权token放在连接URL或请求头里。远程MCP服务建议用短时效token定期刷新。本地stdio方式没有网络传输但也要验证调用方身份防止本地其他进程冒用。3.5 审计日志的字段设计与存储审计日志的字段我建议至少包含这些字段类型说明trace_idstring全链路追踪ID贯穿一次请求的所有操作timestampdatetime操作发生时间精确到毫秒actor_idstring调用方身份标识actor_rolestring调用方角色actionstring操作类型llm_call/rag_search/tool_call/memory_writeresourcestring操作对象模型名/知识库名/工具名input_digeststring输入摘要脱敏后output_digeststring输出摘要脱敏后duration_msinteger耗时毫秒statusstringsuccess/failure/timeouterror_codestring失败时的错误码存储上审计日志写入独立的表或独立的库只允许追加不允许修改。定期归档冷数据热数据保留30-90天。查询接口要支持按trace_id、actor_id、时间范围检索。注意审计日志里的输入输出摘要要做脱敏不能把用户的敏感信息原样记进去。我的做法是只记前100个字符加哈希值既能追溯又不泄露。4. 实操过程与关键环节实现4.1 环境准备与依赖安装先列一下这套系统需要的核心组件Python 3.103.11更稳3.12部分库还没跟上向量数据库QdrantDocker一键起或Chroma嵌入式零配置关系数据库PostgreSQL存记忆和审计或SQLite原型阶段够用嵌入模型本地用Ollama跑bge-m3或用商用API大模型API任意兼容OpenAI格式的APIMCP SDK官方Python SDK# 起Qdrant docker run -d -p 6333:6333 -v $(pwd)/qdrant_data:/qdrant/storage qdrant/qdrant # 起PostgreSQL docker run -d -p 5432:5432 -e POSTGRES_PASSWORDyourpass -v $(pwd)/pg_data:/var/lib/postgresql/data postgres:16 # 安装Python依赖 pip install qdrant-client psycopg2-binary openai mcp ollama环境变量管理用.env文件配合python-dotenv加载。所有密钥、连接串都走环境变量代码里不出现明文。# .env 示例 LLM_API_KEYsk-xxxx LLM_BASE_URLhttps://api.example.com/v1 QDRANT_URLhttp://localhost:6333 PG_DSNpostgresql://postgres:yourpasslocalhost:5432/ai_app MCP_SERVER_URLwss://mcp.example.com/mcp MCP_TOKENeyJhbGci...4.2 知识库入库的完整流程入库流程分五步加载、切块、嵌入、存储、索引。加载环节要处理多种格式PDF用PyMuPDF或MinerU解析热词里mineru api就是干这个的Word用python-docxMarkdown直接读网页用trafilatura提取正文。图片里的文字用OCR图片本身如果要检索需要多模态嵌入模型。切块我用的是递归字符切分优先按段落切段落太长再按句子切句子还长就按字符切。分隔符优先级\n\n\n。。from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size400, chunk_overlap80, separators[\n\n, \n, 。, , , , , ], length_functionlen, ) chunks splitter.split_text(document)嵌入环节批量嵌入比逐条嵌入快得多。Qdrant的批量upsert接口一次能传几百条。嵌入时要带上元数据来源文档、页码、章节标题检索时能一起返回方便溯源。from qdrant_client import QdrantClient from qdrant_client.models import PointStruct client QdrantClient(urlhttp://localhost:6333) points [] for i, chunk in enumerate(chunks): vector embed(chunk) # 调用嵌入模型 points.append(PointStruct( idhash(chunk) % (10**9), vectorvector, payload{ text: chunk, source: doc_name, page: page_num, chunk_index: i, } )) client.upsert(collection_nameknowledge, pointspoints)注意嵌入模型的维度要和向量库集合的维度一致建集合时就要指定。换嵌入模型意味着要重建整个集合所以选型时想清楚别中途换。4.3 对话主流程的编排一轮完整对话的编排逻辑是这样的接收用户请求网关鉴权生成trace_id。加载用户的事实记忆和情景记忆。判断是否需要RAG检索可以用规则也可以用模型判断。如果需要执行检索得到上下文。组装Prompt系统提示 记忆 检索上下文 对话历史 用户输入。调用大模型API传入工具定义。如果模型返回工具调用请求执行工具把结果回填再次调用模型。得到最终回答返回给用户。异步写入审计日志和记忆。async def chat(user_id, message, trace_id): # 1. 加载记忆 facts, episodes load_memory(user_id, message) # 2. 判断是否检索 need_rag should_retrieve(message) contexts retrieve(message) if need_rag else [] # 3. 组装Prompt messages build_prompt(facts, episodes, contexts, message) # 4. 调用模型带工具 response await llm_call(messages, toolsget_mcp_tools()) # 5. 处理工具调用循环 while response.has_tool_call(): tool_result await execute_mcp_tool(response.tool_call) messages.append(response.tool_call) messages.append(tool_result) response await llm_call(messages, toolsget_mcp_tools()) # 6. 异步写审计和记忆 asyncio.create_task(write_audit(trace_id, user_id, message, response)) asyncio.create_task(extract_and_save_memory(user_id, message, response)) return response.content工具调用循环要设最大轮次限制防止模型陷入无限调用。我设的是5轮超过就强制返回当前结果并记录异常。4.4 MCP工具服务端的实现MCP服务端可以用官方SDK快速搭。核心是定义工具函数注册到服务端然后启动监听。from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent server Server(my-tools) server.list_tools() async def list_tools(): return [ Tool( namequery_database, description查询业务数据库, inputSchema{ type: object, properties: { sql: {type: string, description: SQL查询语句} }, required: [sql] } ) ] server.call_tool() async def call_tool(name, arguments): if name query_database: # 鉴权检查 if not check_permission(arguments.get(_actor), db_query): return [TextContent(typetext, text权限不足)] # 执行查询注意SQL注入防护 result safe_query(arguments[sql]) return [TextContent(typetext, textstr(result))] async def main(): async with stdio_server() as (read, write): await server.run(read, write, server.create_initialization_options())注意工具服务端是安全重灾区。SQL查询工具必须做白名单或参数化不能让模型直接拼SQL。文件操作工具要限制目录范围。任何工具都要在服务端再做一次权限校验不能只信客户端传来的身份。4.5 鉴权审计的落地细节审计日志的写入用异步队列避免阻塞主流程。我用的是asyncio.Queue加后台消费者批量写入数据库。audit_queue asyncio.Queue() async def audit_consumer(): batch [] while True: try: item await asyncio.wait_for(audit_queue.get(), timeout1.0) batch.append(item) if len(batch) 100: await flush_audit(batch) batch [] except asyncio.TimeoutError: if batch: await flush_audit(batch) batch []审计日志的查询接口要单独做走独立的鉴权只有审计角色能查。查询要支持组合条件并且对结果做分页。鉴权方面API Key的存储要用哈希不能存明文。验证时把传来的Key哈希后比对。Key要支持设置有效期和权限范围过期自动失效。5. 常见问题与排查技巧实录5.1 鉴权类问题速查错误信息可能原因排查方法401 incorrect api keyKey错误/过期/被吊销检查Key是否复制完整是否在有效期内401 unauthorized签名验证失败检查签名算法、时间戳、请求体是否一致403 forbidden权限不足检查角色绑定的权限是否包含目标操作400 organization disabled账号状态异常联系服务方确认账号状态热词里unexpected status 401 unauthorized: incorrect api key provided: sk-svcac****这个错误我遇到最多的情况是Key复制时带了空格或者环境变量没加载成功。排查时先打印一下实际用的Key脱敏后确认和预期一致。5.2 RAG效果差的排查思路RAG效果差通常分三种情况检索不到、检索到但不相关、相关但模型没用上。检索不到先看知识库里有没有相关内容。用原始问题直接搜向量库看Top-20里有没有对的。如果没有可能是切块切坏了或者嵌入模型不适合这个领域。试试换嵌入模型或者调整切块策略。检索到但不相关多半是召回太多噪声。加一个相似度阈值低于阈值的直接丢弃。再加一个重排序模型把真正相关的排上来。相关但模型没用上是Prompt的问题。检查检索结果有没有正确拼进PromptPrompt里有没有明确指示基于以下资料回答。有时候模型会忽略上下文需要在系统提示里强调。5.3 记忆系统的典型故障记忆系统最常见的问题是记忆污染——错误的信息被写入长期记忆之后一直影响对话。我的做法是写入长期记忆前做一次校验让模型判断这条信息是否确定、是否值得长期记住。不确定的信息只存短期。第二个问题是记忆膨胀——记忆越存越多检索越来越慢而且噪声越来越大。解决方法是定期清理超过一定时间且从未被召回的记忆归档或删除相似度极高的重复记忆合并。第三个问题是跨设备同步。热词里一台电脑上的记忆配置如何用到另一台就是这个场景。我的方案是记忆存服务端客户端只做展示和触发不存本地。这样换设备无缝衔接。5.4 MCP工具调用的坑MCP工具调用最容易出问题的地方是参数格式不匹配。模型生成的参数可能缺字段、类型不对、或者多传了字段。服务端要做参数校验不合法就返回明确的错误信息让模型能自我修正。第二个坑是超时。工具执行可能很慢模型等不了。要设超时超时后返回工具执行超时让模型决定是重试还是换方案。第三个坑是工具结果太大。比如查询数据库返回了几万行全塞进上下文会爆token。服务端要对结果做截断或摘要只返回关键信息。def truncate_result(result, max_chars2000): text str(result) if len(text) max_chars: return text return text[:max_chars] f\n...(结果被截断共{len(text)}字符)5.5 上下文长度超限的处理热词里api error: 400 this models maximum context length is 1048576 tokens说明即使窗口很大也有超限的时候。处理策略分三层第一层预防。组装Prompt前估算token数超了就裁剪。裁剪优先级先裁情景记忆再裁检索上下文最后裁对话历史。系统提示和事实记忆不裁。第二层压缩。对话历史太长时把早期对话做摘要用摘要替代原文。摘要用便宜的小模型做成本低。第三层降级。如果实在超限减少检索Top-K或者切换到窗口更大的模型。def estimate_tokens(text): # 中文约1.5字符/token英文约4字符/token取保守估计 return len(text) // 2 def fit_context(messages, max_tokens): while estimate_tokens(str(messages)) max_tokens: # 从最早的对话开始删 if len(messages) 3: messages.pop(1) # 保留system和最新user else: break return messages6. 工具选型与性能优化6.1 向量数据库选型对比数据库部署方式适用规模优点缺点Chroma嵌入式十万级零配置Python原生不适合分布式FAISS库百万级快Facebook出品无持久化要自己管Qdrant服务百万到千万功能全Rust写的快要单独部署Milvus分布式千万级以上扩展性强运维复杂原型阶段我推荐Chroma一行代码起。生产环境推荐Qdrant性能和功能平衡得好。超大规模才考虑Milvus。6.2 嵌入模型的选择嵌入模型直接决定RAG的上限。中文场景我推荐bge-m3多语言支持好维度1024本地跑得动。如果追求极致效果可以用商用嵌入API但成本和数据隐私要权衡。嵌入模型的维度不是越高越好。1024维在大多数场景够用1536维提升有限但存储和计算成本增加。选型时看MTEB榜单但也要在自己的数据上实测。6.3 缓存策略大模型调用是成本和延迟的大头能缓存就缓存。缓存分两级精确缓存相同的Prompt直接返回缓存结果。用Prompt的哈希做Key存Redis。适合FAQ类高频问题。语义缓存语义相似的问题返回缓存结果。用嵌入向量做相似度匹配超过阈值就命中。适合问法多样但意图相同的场景。def get_cached_response(query, threshold0.95): query_vec embed(query) # 精确缓存 exact_key hashlib.md5(query.encode()).hexdigest() if cached : redis.get(exact_key): return cached # 语义缓存 similar semantic_cache.search(query_vec, threshold) if similar: return similar.response return None注意缓存要设过期时间知识库更新后旧缓存要失效。我的做法是知识库更新时清空相关缓存或者给缓存加版本号。6.4 并发与限流大模型API通常有QPS限制超了会报429。要做客户端限流用令牌桶或漏桶算法。同时要控制并发数避免打爆下游。import asyncio from asyncio import Semaphore llm_semaphore Semaphore(10) # 最多10个并发 async def llm_call_with_limit(messages): async with llm_semaphore: return await llm_call(messages)限流之外还要做重试。网络抖动、临时限流都可能失败重试要带指数退避。但注意非幂等操作不能盲目重试比如写操作重试可能重复写入。7. 安全加固与合规要点7.1 输入输出的安全过滤用户输入要过滤注入攻击。Prompt注入是AI应用特有的风险——用户在输入里写忽略之前的指令试图让模型执行非预期操作。防御方法是在系统提示里明确边界同时对输入做检测发现可疑模式就拒绝或转义。模型输出也要过滤。模型可能生成不当内容或者泄露系统提示。输出前过一遍敏感词库和格式校验。7.2 密钥管理所有密钥走密钥管理服务不写在代码和配置文件里。开发环境用.env生产环境用KMS或Vault。密钥要定期轮换轮换时新旧密钥并存一段时间避免服务中断。MCP的token、API Key、数据库密码都属于密钥。审计日志里不能出现密钥明文要脱敏。7.3 数据隔离多租户场景下不同租户的数据要隔离。向量库用payload过滤关系库用行级权限对象存储用前缀隔离。审计日志要记录租户ID方便追溯。知识库的检索要带租户过滤条件防止A租户检索到B租户的文档。这个过滤要在向量检索时就加上不能检索完再过滤否则有信息泄露风险。7.4 审计日志的防篡改审计日志写入后不可修改。技术上可以用只追加的表、WORM存储、或者区块链式的哈希链。哈希链的做法是每条日志包含前一条的哈希任何修改都会破坏链。def write_audit_log(entry, prev_hash): entry[prev_hash] prev_hash entry_str json.dumps(entry, sort_keysTrue) entry[hash] hashlib.sha256(entry_str.encode()).hexdigest() db.insert(entry) return entry[hash]8. 扩展方向与个人体会这套系统跑通之后扩展方向很多。RAG可以升级到Agentic RAG让模型自主决定检索策略。记忆可以引入知识图谱把事实记忆结构化。MCP工具可以接入更多业务系统让模型真正能干活。审计可以加实时告警发现异常调用立即通知。我自己在实际操作中的体会是这套系统最难的不是某个模块的技术实现而是模块之间的协调。RAG检索的结果怎么和记忆融合工具调用的结果怎么回填审计日志怎么贯穿全链路这些胶水工作才是工程量的主体。我的建议是先把主流程跑通哪怕每个模块都用最简实现然后再逐个优化。一上来就追求每个模块都完美大概率会卡在半路。最后分享一个小技巧调试这套系统时把每一轮的完整Prompt和模型原始输出都打到日志里脱敏后出问题时能快速定位是哪一环出的错。我踩过的坑是只记了最终回答结果模型答错了根本不知道是检索错了还是模型理解错了只能重跑浪费大量时间。