企业官网建设流程全解析

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 rag""graphrag""ontology 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一套、国内厂商又一套。你写好的工具定义，换个模型就得改。MCP（Model Context Protocol）的价值在于标准化——工具提供方按MCP规范暴露能力，模型侧按MCP规范调用，两边解耦。

热词里"playwright mcp""chrome devtools mcp""browser 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_k=5): # 1. 查询改写（可选，提升召回） rewritten = rewrite_query(query) # 2. 向量检索召回 candidates = vector_store.search(rewritten, top_k=20) # 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_id=user_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(key=lambda x: x.score, reverse=True) 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 审计日志的字段设计与存储

审计日志的字段我建议至少包含这些：

存储上，审计日志写入独立的表或独立的库，只允许追加不允许修改。定期归档冷数据，热数据保留30-90天。查询接口要支持按trace_id、actor_id、时间范围检索。

注意：审计日志里的输入输出摘要要做脱敏，不能把用户的敏感信息原样记进去。我的做法是只记前100个字符加哈希值，既能追溯又不泄露。

4. 实操过程与关键环节实现

4.1 环境准备与依赖安装

先列一下这套系统需要的核心组件：

• Python 3.10+（3.11更稳，3.12部分库还没跟上）
• 向量数据库：Qdrant（Docker一键起）或Chroma（嵌入式，零配置）
• 关系数据库：PostgreSQL（存记忆和审计）或SQLite（原型阶段够用）
• 嵌入模型：本地用Ollama跑bge-m3，或用商用API
• 大模型API：任意兼容OpenAI格式的API
• MCP 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_PASSWORD=yourpass -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_KEY=sk-xxxx LLM_BASE_URL=https://api.example.com/v1 QDRANT_URL=http://localhost:6333 PG_DSN=postgresql://postgres:yourpass@localhost:5432/ai_app MCP_SERVER_URL=wss://mcp.example.com/mcp MCP_TOKEN=eyJhbGci...

4.2 知识库入库的完整流程

入库流程分五步：加载、切块、嵌入、存储、索引。

加载环节要处理多种格式：PDF用PyMuPDF或MinerU解析（热词里"mineru api"就是干这个的），Word用python-docx，Markdown直接读，网页用trafilatura提取正文。图片里的文字用OCR，图片本身如果要检索，需要多模态嵌入模型。

切块我用的是递归字符切分，优先按段落切，段落太长再按句子切，句子还长就按字符切。分隔符优先级：\n\n>\n>。>！>？>；>，。

from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=400, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", "；", "，", ""], length_function=len, ) chunks = splitter.split_text(document)

嵌入环节，批量嵌入比逐条嵌入快得多。Qdrant的批量upsert接口一次能传几百条。嵌入时要带上元数据：来源文档、页码、章节标题，检索时能一起返回，方便溯源。

from qdrant_client import QdrantClient from qdrant_client.models import PointStruct client = QdrantClient(url="http://localhost:6333") points = [] for i, chunk in enumerate(chunks): vector = embed(chunk) # 调用嵌入模型 points.append(PointStruct( id=hash(chunk) % (10**9), vector=vector, payload={ "text": chunk, "source": doc_name, "page": page_num, "chunk_index": i, } )) client.upsert(collection_name="knowledge", points=points)

注意：嵌入模型的维度要和向量库集合的维度一致，建集合时就要指定。换嵌入模型意味着要重建整个集合，所以选型时想清楚，别中途换。

4.3 对话主流程的编排

一轮完整对话的编排逻辑是这样的：

1. 接收用户请求，网关鉴权，生成trace_id。
2. 加载用户的事实记忆和情景记忆。
3. 判断是否需要RAG检索（可以用规则，也可以用模型判断）。
4. 如果需要，执行检索，得到上下文。
5. 组装Prompt：系统提示 + 记忆 + 检索上下文 + 对话历史 + 用户输入。
6. 调用大模型API，传入工具定义。
7. 如果模型返回工具调用请求，执行工具，把结果回填，再次调用模型。
8. 得到最终回答，返回给用户。
9. 异步写入审计日志和记忆。

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, tools=get_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, tools=get_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( name="query_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(type="text", text="权限不足")] # 执行查询（注意SQL注入防护） result = safe_query(arguments["sql"]) return [TextContent(type="text", text=str(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(), timeout=1.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 鉴权类问题速查

热词里"unexpected status 401 unauthorized: incorrect api key provided: sk-svcac****"这个错误，我遇到最多的情况是Key复制时带了空格，或者环境变量没加载成功。排查时先打印一下实际用的Key（脱敏后），确认和预期一致。

5.2 RAG效果差的排查思路

RAG效果差通常分三种情况：检索不到、检索到但不相关、相关但模型没用上。

检索不到，先看知识库里有没有相关内容。用原始问题直接搜向量库，看Top-20里有没有对的。如果没有，可能是切块切坏了，或者嵌入模型不适合这个领域。试试换嵌入模型，或者调整切块策略。

检索到但不相关，多半是召回太多噪声。加一个相似度阈值，低于阈值的直接丢弃。再加一个重排序模型，把真正相关的排上来。

相关但模型没用上，是Prompt的问题。检查检索结果有没有正确拼进Prompt，Prompt里有没有明确指示"基于以下资料回答"。有时候模型会忽略上下文，需要在系统提示里强调。

5.3 记忆系统的典型故障

记忆系统最常见的问题是记忆污染——错误的信息被写入长期记忆，之后一直影响对话。我的做法是：写入长期记忆前做一次校验，让模型判断这条信息是否确定、是否值得长期记住。不确定的信息只存短期。

第二个问题是记忆膨胀——记忆越存越多，检索越来越慢，而且噪声越来越大。解决方法是定期清理：超过一定时间且从未被召回的记忆，归档或删除；相似度极高的重复记忆，合并。

第三个问题是跨设备同步。热词里"一台电脑上的记忆配置如何用到另一台"就是这个场景。我的方案是记忆存服务端，客户端只做展示和触发，不存本地。这样换设备无缝衔接。

5.4 MCP工具调用的坑

MCP工具调用最容易出问题的地方是参数格式不匹配。模型生成的参数可能缺字段、类型不对、或者多传了字段。服务端要做参数校验，不合法就返回明确的错误信息，让模型能自我修正。

第二个坑是超时。工具执行可能很慢，模型等不了。要设超时，超时后返回"工具执行超时"，让模型决定是重试还是换方案。

第三个坑是工具结果太大。比如查询数据库返回了几万行，全塞进上下文会爆token。服务端要对结果做截断或摘要，只返回关键信息。

def truncate_result(result, max_chars=2000): text = str(result) if len(text) <= max_chars: return text return text[:max_chars] + f"\n...(结果被截断，共{len(text)}字符)"

5.5 上下文长度超限的处理

热词里"api error: 400 this model's 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 messages

6. 工具选型与性能优化

6.1 向量数据库选型对比

原型阶段我推荐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, threshold=0.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_keys=True) entry["hash"] = hashlib.sha256(entry_str.encode()).hexdigest() db.insert(entry) return entry["hash"]

8. 扩展方向与个人体会

这套系统跑通之后，扩展方向很多。RAG可以升级到Agentic RAG，让模型自主决定检索策略。记忆可以引入知识图谱，把事实记忆结构化。MCP工具可以接入更多业务系统，让模型真正能干活。审计可以加实时告警，发现异常调用立即通知。

我自己在实际操作中的体会是：这套系统最难的不是某个模块的技术实现，而是模块之间的协调。RAG检索的结果怎么和记忆融合，工具调用的结果怎么回填，审计日志怎么贯穿全链路，这些"胶水"工作才是工程量的主体。我的建议是先把主流程跑通，哪怕每个模块都用最简实现，然后再逐个优化。一上来就追求每个模块都完美，大概率会卡在半路。

最后分享一个小技巧：调试这套系统时，把每一轮的完整Prompt和模型原始输出都打到日志里（脱敏后），出问题时能快速定位是哪一环出的错。我踩过的坑是只记了最终回答，结果模型答错了根本不知道是检索错了还是模型理解错了，只能重跑，浪费大量时间。