引言过去两年大语言模型LLM应用经历了从概念验证到生产部署的快速演进。然而构建一个稳定、高效、可维护的LLM应用远不止调用API那么简单。架构设计直接决定了系统的响应速度、成本和可靠性。本文将深入探讨LLM应用架构设计的核心最佳实践并带你从零构建一个带有语义缓存的RAG检索增强生成智能问答系统。一、核心概念LLM应用的分层架构一个成熟的LLM应用通常遵循四层架构┌─────────────────────────────────┐ │ 表示层 (API / UI) │ ← 用户交互入口 ├─────────────────────────────────┤ │ 业务编排层 (Orchestration) │ ← 流程控制、Prompt管理 ├─────────────────────────────────┤ │ 服务层 (Services) │ ← 检索、缓存、预处理 ├─────────────────────────────────┤ │ 基础设施层 (Infra) │ ← 向量库、LLM、嵌入模型 └─────────────────────────────────┘1.1 基础设施层负责与外部系统交互包括LLM调用、向量数据库操作、嵌入模型请求等。这一层应当完全与业务逻辑解耦方便随时替换底层模型或存储方案。1.2 服务层封装可复用的能力如语义缓存、文档检索、后处理等。服务层是系统性能和成本优化的关键所在。1.3 业务编排层负责将多个服务组合成完整的业务流程管理Prompt模板处理条件分支和异常流程。1.4 表示层提供对外的API接口或UI界面处理请求校验、流式响应等。二、实战构建带语义缓存的RAG问答系统下面我们实现一个完整的分层架构系统。核心特性- 模块化设计各层通过接口解耦- 基于FAISS的向量检索- 语义缓存层相似问题直接返回缓存结果- 流式输出支持2.1 项目依赖pip install langchain langchain-openai faiss-cpu openai numpy python-dotenv2.2 完整代码实现pythonLLM应用架构最佳实践 —— 分层RAG系统完整示例作者技术博客依赖langchain, langchain-openai, faiss-cpu, openai, numpyimport osimport timeimport hashlibimport loggingfrom typing import List, Optional, Dict, Anyfrom dataclasses import dataclass, fieldfrom abc import ABC, abstractmethodimport numpy as npfrom dotenv import load_dotenvfrom openai import OpenAIfrom langchain_openai import OpenAIEmbeddingsfrom langchain_community.vectorstores import FAISSfrom langchain_core.documents import Documentload_dotenv()logging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s)logger logging.getLogger(name)═══════════════════════════════════════════1. 配置层 — 集中管理所有可配置项═══════════════════════════════════════════dataclassclass AppConfig:应用配置支持环境变量覆盖openai_api_key: str field(default_factorylambda: os.getenv(OPENAI_API_KEY, ))embedding_model: str text-embedding-3-smallllm_model: str gpt-4o-minicache_similarity_threshold: float 0.92 # 语义缓存相似度阈值max_retries: int 3 # LLM调用最大重试次数retry_backoff: float 1.5 # 重试退避因子top_k_retrieval: int 4 # 检索返回文档数stream_output: bool True # 是否启用流式输出config AppConfig()═══════════════════════════════════════════2. 基础设施层 — LLM封装带重试和可观测性═══════════════════════════════════════════class LLMProvider(ABC):LLM抽象接口方便替换底层模型abstractmethoddef generate(self, prompt: str, stream: bool False) - str:...class OpenAILLMProvider(LLMProvider):OpenAI实现内置指数退避重试definit(self, cfg: AppConfig):self.client OpenAI(api_keycfg.openai_api_key)self.model cfg.llm_modelself.max_retries cfg.max_retriesself.backoff cfg.retry_backoffdef generate(self, prompt: str, stream: bool False) - str: for attempt in range(self.max_retries): try: start time.time() response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], streamstream, temperature0.3, ) if stream: collected [] for chunk in response: delta chunk.choices[0].delta.content or collected.append(delta) print(delta, end, flushTrue) # 流式打印 print() result .join(collected) else: result response.choices[0].message.content elapsed time.time() - start logger.info(fLLM调用完成耗时 {elapsed:.2f}s响应长度 {len(result)}) return result except Exception as e: wait self.backoff ** attempt logger.warning(fLLM调用失败(attempt {attempt1}){e}{wait:.1f}s后重试) time.sleep(wait) raise RuntimeError(LLM调用失败已达最大重试次数)═══════════════════════════════════════════3. 服务层 — 语义缓存核心优化═══════════════════════════════════════════class SemanticCache:语义缓存对用户查询做向量化若与历史缓存中的查询语义相似度超过阈值则直接返回缓存答案避免重复调用LLM。definit(self, embedder, threshold: float 0.92):self.embedder embedderself.threshold thresholdself._cache: List[Dict[str, Any]] [] # [{query, embedding, answer}]def lookup(self, query: str) - Optional[str]: 查找语义相似的缓存条目 if not self._cache: return None query_vec np.array(self.embedder.embed_query(query), dtypenp.float32) best_score -1.0 best_answer None for entry in self._cache: cached_vec np.array(entry[embedding], dtypenp.float32) # 余弦相似度 score np.dot(query_vec, cached_vec) / ( np.linalg.norm(query_vec) * np.linalg.norm(cached_vec) 1e-10 ) if score best_score: best_score score best_answer entry[answer] if best_score self.threshold: logger.info(f语义缓存命中相似度{best_score:.4f}) return best_answer logger.info(f缓存未命中最高相似度{best_score:.4f}阈值{self.threshold}) return None def store(self, query: str, answer: str): 将查询-答案对存入缓存 embedding self.embedder.embed_query(query) self._cache.append({query: query, embedding: embedding, answer: answer}) logger.info(f已缓存当前缓存条目数{len(self._cache)})═══════════════════════════════════════════4. 服务层 — 向量检索器═══════════════════════════════════════════class VectorRetriever:封装FAISS向量库的文档检索逻辑definit(self, embedder, documents: List[Document]):self.embedder embedderself.vectorstore FAISS.from_documents(documents, embedder)logger.info(f向量库初始化完成文档数{len(documents)})def retrieve(self, query: str, top_k: int 4) - List[Document]: docs self.vectorstore.similarity_search(query, ktop_k) logger.info(f检索到 {len(docs)} 篇相关文档) return docs═══════════════════════════════════════════5. 业务编排层 — RAG管道═══════════════════════════════════════════class RAGPipeline:RAG管道编排 缓存查询 → 文档检索 → LLM生成 的完整流程SYSTEM_PROMPT 你是一个专业的知识问答助手。请严格基于以下提供的文档内容回答问题。如果文档中没有相关信息请明确说明根据现有资料无法回答不要编造内容。参考文档{context}def __init__(self, llm: LLMProvider, retriever: VectorRetriever, cache: SemanticCache, cfg: AppConfig): self.llm llm self.retriever retriever self.cache cache self.cfg cfg def ask(self, question: str) - str: # Step 1: 查询语义缓存 cached self.cache.lookup(question) if cached: return cached # Step 2: 检索相关文档 docs self.retriever.retrieve(question, top_kself.cfg.top_k_retrieval) context \n\n.join([f【文档{i1}】{d.page_content} for i, d in enumerate(docs)]) # Step 3: 构建Prompt并调用LLM prompt self.SYSTEM_PROMPT.format(contextcontext) f\