1. 这不是又一个“Hello World”教程：FastAPI + LangChain 的真实战场在哪里？

你点开这个标题，大概率不是想学怎么写个返回{"message": "Hello World"}的接口。你真正关心的是：当用户同时发起3000次请求去调用一个带RAG检索、需要调用3个外部API、还要做LLM流式响应的AI服务时，你的后端会不会在第2987次请求上直接跪倒？是不是每次改一行代码就得重启整个服务，等15秒才能看到效果？是不是调试Agent逻辑时，光是理清tool calling的嵌套回调就花了半天？这些不是理论问题，是我在给三家客户落地AI服务时，连续踩过的坑。

FastAPI + LangChain 组合之所以被反复提起，根本原因在于它直击当前AI工程化落地的两个核心矛盾：一边是LangChain这类框架为快速构建复杂AI工作流提供了抽象层，另一边是传统Web框架在高并发、低延迟、热重载、类型安全这些生产级要求上明显吃力。而FastAPI，恰恰是那个能把LangChain的“智能逻辑”稳稳托住的底座——它不是靠堆砌异步关键字来假装高性能，而是从设计哲学上就拒绝阻塞。比如它的依赖注入系统，天然适配LangChain里各种需要复用的组件（LLM、Embeddings、VectorStore），你不用自己手写单例管理器；它的Pydantic v2模型校验，能在请求刚进来的毫秒级就把非法的query参数拦在外面，避免无效请求一路穿透到昂贵的LLM调用环节。

我见过太多团队，把LangChain当成万能胶水，一股脑把所有逻辑塞进一个Chain里，结果服务一上线就内存暴涨、响应延迟抖动剧烈。真正的实战，从来不是“能不能跑起来”，而是“能不能扛住、能不能查、能不能扩”。所以这篇内容不会教你如何用LangChain写一个聊天机器人demo，而是带你从零开始，搭建一个具备生产可用性的AI后端骨架：它有清晰的项目分层、可独立部署的模块、带熔断和降级的外部API调用、支持流式SSE响应的前端对接能力、以及一套能快速定位性能瓶颈的日志和指标体系。如果你正打算用这套技术栈接一个真实的业务需求——比如给客服系统加知识库问答、给销售工具加智能话术生成、或者给内部运营平台加数据分析助手——那接下来的内容，就是你跳过试错周期、直接抄作业的路线图。

2. 架构设计：为什么必须放弃“一个main.py打天下”的思维？

2.1 三层解耦：让AI逻辑、业务逻辑、传输逻辑各司其职

很多初学者的FastAPI+LangChain项目，目录结构长得像这样：

project/ ├── main.py # 所有东西都在这里：路由、Chain定义、LLM初始化、数据库连接... ├── requirements.txt └── README.md

这种结构在本地跑通demo没问题，但一旦进入真实场景，就会立刻暴露出三个致命问题：第一，修改一个Prompt模板，得重启整个服务；第二，想给某个特定API加Rate Limit，发现所有路由共享同一个中间件配置；第三，测试Agent逻辑时，不得不启动整个FastAPI应用，连带启动Redis、PostgreSQL、向量数据库，测试一次要40秒。

我的解决方案是强制分层，每一层只解决一类问题：

• Transport Layer（传输层）：只负责HTTP协议处理。定义路由、接收请求、校验参数、返回响应。它不碰任何AI模型、不读数据库、不调用外部API。它的唯一职责，是把干净的、经过强类型校验的数据，交给下一层。

• Application Layer（应用层）：这是业务逻辑的核心。它持有对LLM、Retriever、Tool等LangChain组件的引用，但不直接创建它们。它定义了“用户问什么，系统该做什么”的规则，比如“当query包含‘价格’且上下文有产品ID时，触发price_lookup_tool”。这一层是单元测试的重点区域，可以完全脱离FastAPI运行。

• Infrastructure Layer（基础设施层）：负责所有外部依赖的初始化和生命周期管理。LLM客户端、向量数据库连接池、缓存客户端、监控上报器……全部在这里集中创建、配置、销毁。它向上只暴露干净的接口（比如get_llm()、get_vectorstore()），向下屏蔽所有底层细节（比如Ollama是用HTTP还是gRPC通信、Chroma是内存模式还是持久化模式）。

这种分层不是为了炫技，而是为了可维护性。举个实际例子：某次客户要求把LLM从Ollama切换到Azure OpenAI。如果按老方式，你得在main.py里找到所有ChatOllama(...)实例，逐个替换，再检查每个Chain的参数是否兼容。而按分层架构，你只需要修改Infrastructure层里的get_llm()函数，把Ollama实例换成AzureOpenAI实例，其他两层代码一行都不用动。因为Application层只认get_llm()这个契约，Transport层只认Application层提供的service接口。

2.2 LangChain组件的“懒加载”与连接池设计

LangChain里最常被忽视的性能陷阱，是LLM和Embeddings的初始化方式。很多人写：

# ❌ 危险！每次请求都新建一个LLM实例 @app.get("/chat") def chat(query: str): llm = ChatOllama(model="qwen2:7b", temperature=0.3) # 每次调用都new一次！ chain = LLMChain(llm=llm, prompt=CHAT_PROMPT) return chain.invoke({"query": query})

这会导致两个严重后果：一是LLM客户端的TCP连接无法复用，频繁建立/关闭连接；二是Ollama服务端会为每个请求创建新的推理线程，内存占用呈线性增长。实测下来，Qwen2-7B模型在并发100时，内存占用从1.2GB飙升到4.8GB。

正确做法是利用FastAPI的依赖注入，实现全局单例+连接池：

# ✅ infrastructure/llm.py from langchain_community.chat_models import ChatOllama from langchain_openai import ChatOpenAI from typing import Union class LLMFactory: _instance: Union[ChatOllama, ChatOpenAI, None] = None @classmethod def get_llm(cls, provider: str = "ollama") -> Union[ChatOllama, ChatOpenAI]: if cls._instance is not None: return cls._instance if provider == "ollama": # 关键：启用连接池和超时控制 cls._instance = ChatOllama( model="qwen2:7b", temperature=0.3, timeout=30, # 网络超时 keep_alive="24h", # Ollama的keep-alive num_ctx=4096, # 上下文长度 # 以下参数直接影响Ollama服务端资源占用 num_batch=512, # 批处理大小 num_gpu=1, # GPU数量 verbose=False # 关闭日志，避免干扰 ) elif provider == "azure": cls._instance = ChatOpenAI( azure_deployment="gpt-4o-mini", openai_api_version="2024-05-01-preview", azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"), api_key=os.getenv("AZURE_OPENAI_KEY"), # Azure特有的连接池配置 max_retries=3, timeout=30 ) return cls._instance

然后在Application层通过依赖注入获取：

# ✅ application/services/chat_service.py from infrastructure.llm import LLMFactory class ChatService: def __init__(self, llm_provider: str = "ollama"): self.llm = LLMFactory.get_llm(llm_provider) # 懒加载，首次调用才初始化 def invoke_chain(self, query: str) -> str: # 这里可以复用同一个llm实例，连接池自动管理 chain = LLMChain(llm=self.llm, prompt=CHAT_PROMPT) return chain.invoke({"query": query})

提示：Ollama的num_batch参数非常关键。默认值是512，但对于Qwen2-7B这种7B模型，在4x A10G卡上，设为1024反而更高效；而在单卡3090上，设为256能显著降低OOM风险。这不是玄学，是Ollama源码里根据GPU显存带宽和计算单元数做的硬编码优化，必须根据你的硬件实测调整。

2.3 高性能的关键：别让LangChain的“链式调用”拖垮你的QPS

LangChain的Chain设计初衷是提升开发效率，但它在高并发场景下，会成为性能瓶颈。原因在于：Chain的invoke()方法内部做了大量动态反射、参数序列化、中间状态保存。我做过压测，一个简单的LLMChain，在并发500时，CPU时间有37%花在_validate_inputs()和_prepare_args()这些非业务逻辑上。

实战中，我采用“链拆解+手动编排”策略：

# ❌ Chain方式（简洁但慢） chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=vectorstore.as_retriever(), return_source_documents=True ) # ✅ 手动编排（快3.2倍） def manual_rag_query(query: str) -> dict: # Step 1: 向量检索（并行化） docs = vectorstore.similarity_search(query, k=5) # Step 2: 构建Prompt（避免Chain的模板渲染开销） context = "\n\n".join([doc.page_content for doc in docs]) final_prompt = f"""你是一个专业的客服助手。请基于以下信息回答用户问题： 【参考信息】 {context} 【用户问题】 {query} 请用中文回答，简洁准确，不要编造信息。""" # Step 3: 直接调用LLM（绕过Chain封装） response = llm.invoke(final_prompt) return { "answer": response.content, "sources": [doc.metadata for doc in docs] }

这个改动带来的收益是实实在在的：在同等硬件下，QPS从128提升到412，P99延迟从842ms降到217ms。代价是代码行数多了20行，但换来的是可预测的性能和更低的运维成本。记住，AI后端的“高性能”，从来不是靠框架的魔法，而是靠对每一毫秒开销的精确控制。

3. 核心细节解析：从环境准备到生产部署的避坑指南

3.1 FastAPI项目目录结构：为什么src/和app/不能混用？

一个被严重低估的细节是项目根目录结构。很多教程直接把main.py放在项目根目录，导致pip install -e .时，Python会把整个项目根目录当作包来导入，引发循环引用。正确的结构必须明确区分“可安装包”和“可执行入口”：

my-ai-backend/ ├── src/ # ✅ 可安装的Python包，所有业务代码放这里 │ ├── my_ai_backend/ # 包名，对应import my_ai_backend │ │ ├── __init__.py │ │ ├── transport/ # Transport Layer │ │ │ ├── __init__.py │ │ │ └── api.py # 路由定义 │ │ ├── application/ # Application Layer │ │ │ ├── __init__.py │ │ │ └── services/ │ │ ├── infrastructure/ # Infrastructure Layer │ │ │ ├── __init__.py │ │ │ └── llm.py │ │ └── core/ # 配置、异常、常量 ├── app/ # ✅ 可执行入口，只放启动脚本 │ ├── __init__.py │ └── main.py # uvicorn.run(...)，只负责启动 ├── pyproject.toml # ✅ 使用PEP 621标准，替代setup.py └── Dockerfile

pyproject.toml的关键配置：

[build-system] requires = ["setuptools>=45", "wheel", "setuptools_scm[toml]>=6.2"] build-backend = "setuptools.build_meta" [project] name = "my-ai-backend" version = "0.1.0" description = "High-performance AI backend with FastAPI and LangChain" dependencies = [ "fastapi>=0.111.0", "langchain>=0.2.0", "langchain-community>=0.2.0", "langchain-openai>=0.1.0", "chromadb>=0.4.25", "uvicorn>=0.29.0", "prometheus-client>=0.19.0", ] requires-python = ">=3.10" [project.optional-dependencies] dev = ["pytest>=7.0", "black>=24.0", "mypy>=1.10"]

注意：src/目录的存在，是为了让pip install -e .时，Python能正确识别my_ai_backend为一个包，而不是把当前目录当作包。否则你在app/main.py里写from my_ai_backend.transport.api import router，会报ModuleNotFoundError。这是Python打包机制的硬性要求，不是风格偏好。

3.2 Windows下的打包陷阱：为什么pyinstaller会失败？

Windows开发者常遇到的问题是：本地开发一切正常，但用pyinstaller打包后，服务启动报错ModuleNotFoundError: No module named 'langchain_community'。根本原因在于LangChain 0.2.x版本大量使用了动态导入（importlib.import_module），而PyInstaller默认无法静态分析这些导入路径。

解决方案不是放弃打包，而是显式告诉PyInstaller哪些模块必须包含：

# ✅ 正确的打包命令（Windows） pyinstaller --onefile ^ --add-data "src;my_ai_backend" ^ --collect-all langchain ^ --collect-all langchain_community ^ --collect-all langchain_openai ^ --collect-all chromadb ^ --hidden-import langchain_community.llms.ollama ^ --hidden-import langchain_community.vectorstores.chroma ^ app/main.py

其中--collect-all参数会递归收集指定包的所有子模块，--hidden-import则强制包含那些PyInstaller无法自动发现的动态导入模块。实测下来，这样打包后的exe文件大小约280MB（含Ollama的Windows二进制），但能稳定运行。

实操心得：在Windows上打包前，务必先用--debug模式运行一次，观察日志里是否有WARNING: Hidden import提示，这些就是你需要手动添加--hidden-import的模块。我曾经漏掉了langchain_community.embeddings.fastembed，结果服务启动时报AttributeError: module 'langchain_community' has no attribute 'embeddings'，排查了3小时才发现是打包遗漏。

3.3 LangChain的Embeddings性能优化：别让向量检索成为瓶颈

RAG系统的性能，70%取决于Embeddings的生成速度和向量检索的效率。很多人直接用HuggingFaceEmbeddings，结果发现单次query的embedding耗时高达1.2秒（CPU）或380ms（GPU），彻底拖垮整体体验。

我的优化方案是三级加速：

1. 模型选择：放弃all-MiniLM-L6-v2这类通用小模型，改用专为中文优化的bge-m3。实测在Intel i9-13900K上，bge-m3的吞吐量是MiniLM的2.3倍，且召回率更高。关键是bge-m3支持normalize_embeddings=True，能显著提升Chroma的ANN检索精度。

2. 批处理：Embeddings API天生适合批处理。不要为每个文档单独调用embed_documents()，而是积攒一批（比如50个）再统一调用：

# ✅ 批处理Embeddings def batch_embed_documents(documents: List[str], batch_size: int = 50) -> List[List[float]]: embeddings = [] for i in range(0, len(documents), batch_size): batch = documents[i:i+batch_size] # LangChain的embed_documents()原生支持batch batch_embeddings = embedding_model.embed_documents(batch) embeddings.extend(batch_embeddings) return embeddings

3. 缓存层：对高频Query做Embedding缓存。用Redis存储md5(query)->embedding_vector映射，TTL设为1小时。实测在客服场景下，缓存命中率可达68%，直接抹平Embeddings的延迟毛刺。

# ✅ Redis缓存Embeddings import redis import hashlib import numpy as np redis_client = redis.Redis(host='localhost', port=6379, db=0) def cached_embed_query(query: str) -> List[float]: key = f"emb:{hashlib.md5(query.encode()).hexdigest()}" cached = redis_client.get(key) if cached: return np.frombuffer(cached, dtype=np.float32).tolist() # 缓存未命中，调用模型 embedding = embedding_model.embed_query(query) # 存入Redis，序列化为bytes redis_client.setex(key, 3600, np.array(embedding, dtype=np.float32).tobytes()) return embedding

注意：bge-m3模型输出是1024维向量，每个float32占4字节，一个embedding就是4KB。Redis的setex操作对4KB数据的延迟在0.2ms内，远低于模型推理的300ms，所以缓存收益巨大。

4. 实操过程：从零开始搭建一个可监控、可扩展的AI服务

4.1 初始化项目与依赖管理

第一步，创建干净的Python环境（强烈建议用conda，避免Windows下pip编译C扩展失败）：

# 创建独立环境 conda create -n ai-backend python=3.10 conda activate ai-backend # 安装核心依赖（注意版本锁定） pip install fastapi==0.111.0 \ langchain==0.2.10 \ langchain-community==0.2.10 \ langchain-openai==0.1.16 \ chromadb==0.4.25 \ uvicorn==0.29.0 \ prometheus-client==0.20.0 \ redis==5.0.5 \ python-dotenv==1.0.1

为什么锁死版本？因为LangChain的API在0.1.x到0.2.x之间有重大breaking change，比如LLMChain的predict()方法被移除，invoke()成为唯一入口。生产环境绝不允许pip install langchain这种不带版本号的操作，否则某天pip install -U就可能让你的服务全线崩溃。

4.2 构建Transport Layer：定义健壮的API契约

src/my_ai_backend/transport/api.py：

from fastapi import APIRouter, Depends, HTTPException, status, Request from fastapi.responses import StreamingResponse, JSONResponse from pydantic import BaseModel, Field from typing import List, Optional, AsyncGenerator import json import time router = APIRouter() # 请求模型：强类型约束，避免前端传错字段 class ChatRequest(BaseModel): query: str = Field(..., min_length=1, max_length=2000, description="用户提问") session_id: str = Field(..., min_length=1, max_length=64, description="会话ID，用于历史追溯") model: str = Field("qwen2:7b", description="指定LLM模型") # 响应模型：明确告知前端返回什么 class ChatResponse(BaseModel): answer: str = Field(..., description="AI生成的回答") sources: List[dict] = Field(default_factory=list, description="引用的文档来源") latency_ms: float = Field(..., description="端到端延迟（毫秒）") @router.post("/v1/chat", response_model=ChatResponse) async def chat_endpoint( request: ChatRequest, # 依赖注入：从Infrastructure层获取服务 chat_service: ChatService = Depends(get_chat_service), ): start_time = time.time() try: result = await chat_service.invoke_chain( query=request.query, session_id=request.session_id, model=request.model ) latency_ms = (time.time() - start_time) * 1000 return ChatResponse( answer=result["answer"], sources=result.get("sources", []), latency_ms=round(latency_ms, 2) ) except Exception as e: # 统一错误处理，避免暴露内部细节 raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"AI service error: {str(e)[:100]}" ) # 流式响应端点：前端用EventSource监听 @router.post("/v1/chat/stream") async def chat_stream_endpoint( request: ChatRequest, chat_service: ChatService = Depends(get_chat_service), ): async def event_generator(): try: # 调用流式方法 async for chunk in chat_service.stream_chain( query=request.query, session_id=request.session_id ): yield f"data: {json.dumps({'chunk': chunk})}\n\n" except Exception as e: yield f"data: {json.dumps({'error': str(e)})}\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache"} )

这个API设计体现了三个生产级原则：第一，ChatRequest用Field做了严格的长度和格式校验，前端传超过2000字符的query，FastAPI会在路由层就返回422错误，根本不会走到LLM调用；第二，latency_ms作为响应字段，让前端能自行监控API健康度；第三，/stream端点用SSE协议，比WebSocket更轻量，且浏览器原生支持，无需额外JS库。

4.3 Application Layer实战：构建可测试的ChatService

src/my_ai_backend/application/services/chat_service.py：

from typing import Dict, Any, AsyncGenerator, List, Optional from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.llms import Ollama from infrastructure.llm import LLMFactory from infrastructure.vectorstore import VectorStoreFactory import asyncio class ChatService: def __init__( self, llm_provider: str = "ollama", vectorstore_provider: str = "chroma" ): self.llm = LLMFactory.get_llm(llm_provider) self.vectorstore = VectorStoreFactory.get_vectorstore(vectorstore_provider) self.embedding = HuggingFaceEmbeddings( model_name="BAAI/bge-m3", model_kwargs={'device': 'cpu'}, # Windows上优先用CPU，避免CUDA初始化失败 encode_kwargs={'normalize_embeddings': True} ) async def invoke_chain(self, query: str, session_id: str, model: str) -> Dict[str, Any]: """同步调用，适用于简单问答""" # Step 1: 检索相关文档 docs = self.vectorstore.similarity_search(query, k=3) # Step 2: 构建上下文 context = "\n\n".join([doc.page_content for doc in docs]) # Step 3: 构建Prompt（避免Chain开销） system_prompt = f"""你是一个专业客服助手。请严格基于【参考信息】回答问题，不要编造。如果信息不足，请说'暂无相关信息'。 【参考信息】 {context}""" messages = [ SystemMessage(content=system_prompt), HumanMessage(content=query) ] # Step 4: 调用LLM response = self.llm.invoke(messages) return { "answer": response.content, "sources": [doc.metadata for doc in docs] } async def stream_chain(self, query: str, session_id: str) -> AsyncGenerator[str, None]: """流式响应，适用于长回答""" # 使用LangChain的stream方法，但手动控制 docs = self.vectorstore.similarity_search(query, k=3) context = "\n\n".join([doc.page_content for doc in docs]) system_prompt = f"""你是一个专业客服助手。请严格基于【参考信息】回答问题，不要编造。如果信息不足，请说'暂无相关信息'。 【参考信息】 {context}""" messages = [ SystemMessage(content=system_prompt), HumanMessage(content=query) ] # 关键：使用stream方法，获得token级流式输出 async for chunk in self.llm.astream(messages): if hasattr(chunk, 'content') and chunk.content: yield chunk.content

这个ChatService的设计亮点在于：它完全不依赖langchain.chains模块，而是直接调用llm.invoke()和llm.astream()。这意味着你可以随时替换底层LLM（比如从Ollama换成OpenAI），而无需修改任何Chain定义。所有业务逻辑都集中在invoke_chain和stream_chain这两个方法里，单元测试时，只需mockself.llm和self.vectorstore即可，无需启动FastAPI。

4.4 Infrastructure Layer：Ollama + Chroma的本地化部署

src/my_ai_backend/infrastructure/vectorstore.py：

from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_core.documents import Document import os class VectorStoreFactory: _instance: Chroma = None @classmethod def get_vectorstore(cls, provider: str = "chroma") -> Chroma: if cls._instance is not None: return cls._instance # Chroma配置：持久化到本地，避免每次重启丢失 persist_directory = os.path.join(os.getcwd(), "data", "chroma_db") os.makedirs(persist_directory, exist_ok=True) embedding = HuggingFaceEmbeddings( model_name="BAAI/bge-m3", model_kwargs={'device': 'cpu'}, encode_kwargs={'normalize_embeddings': True} ) # 关键：设置chroma_db的持久化路径和嵌入模型 cls._instance = Chroma( persist_directory=persist_directory, embedding_function=embedding, collection_name="knowledge_base" ) return cls._instance @classmethod def init_from_docs(cls, documents: List[Document]): """初始化向量库，仅在首次运行时调用""" vectorstore = cls.get_vectorstore() # 批量添加，避免单条插入的性能损耗 vectorstore.add_documents(documents, batch_size=50) print(f"✅ Initialized Chroma with {len(documents)} documents")

Ollama的安装和模型拉取（Windows）：

# 下载Ollama Windows版（https://github.com/ollama/ollama/releases） # 安装后，以管理员身份运行PowerShell，执行： ollama serve # 启动服务 # 拉取模型（国内用户推荐清华源） $env:OLLAMA_BASE_URL="https://mirrors.tuna.tsinghua.edu.cn/ollama/" ollama pull qwen2:7b ollama pull bge-m3 # 验证 ollama list # NAME ID SIZE LAST MODIFIED # qwen2:7b 8f1b7... 4.2 GB 2 hours ago # bge-m3 a1c2d... 2.1 GB 1 day ago

注意：Ollama在Windows上默认绑定127.0.0.1:11434，FastAPI代码里不需要改任何URL，LangChain的ChatOllama默认就连接这个地址。但如果在Docker里部署，需要把Ollama容器的11434端口映射出来，并在FastAPI里配置base_url="http://host.docker.internal:11434"。

4.5 生产监控：用Prometheus暴露关键指标

src/my_ai_backend/core/metrics.py：

from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry from prometheus_client.exposition import make_asgi_app import time # 创建自定义注册表，避免污染全局 registry = CollectorRegistry() # 请求计数器 REQUEST_COUNT = Counter( 'ai_backend_requests_total', 'Total number of requests', ['endpoint', 'status_code'], registry=registry ) # 延迟直方图 REQUEST_LATENCY = Histogram( 'ai_backend_request_latency_seconds', 'Latency of requests', ['endpoint'], registry=registry, buckets=[0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0] ) # 当前活跃连接数 ACTIVE_CONNECTIONS = Gauge( 'ai_backend_active_connections', 'Number of active connections', registry=registry ) # 在Transport Layer的中间件中记录指标 async def metrics_middleware(request: Request, call_next): start_time = time.time() ACTIVE_CONNECTIONS.inc() try: response = await call_next(request) REQUEST_COUNT.labels( endpoint=request.url.path, status_code=response.status_code ).inc() return response finally: latency = time.time() - start_time REQUEST_LATENCY.labels(endpoint=request.url.path).observe(latency) ACTIVE_CONNECTIONS.dec()

然后在app/main.py中挂载：

from src.my_ai_backend.core.metrics import registry, metrics_middleware from fastapi.middleware.base import BaseHTTPMiddleware app = FastAPI() app.add_middleware(BaseHTTPMiddleware, dispatch=metrics_middleware) # 挂载Prometheus指标端点 metrics_app = make_asgi_app(registry=registry) app.mount("/metrics", metrics_app)

启动服务后，访问http://localhost:8000/metrics，就能看到实时指标：

# HELP ai_backend_requests_total Total number of requests # TYPE ai_backend_requests_total counter ai_backend_requests_total{endpoint="/v1/chat",status_code="200"} 1245 ai_backend_requests_total{endpoint="/v1/chat/stream",status_code="200"} 892 # HELP ai_backend_request_latency_seconds Latency of requests # TYPE ai_backend_request_latency_seconds histogram ai_backend_request_latency_seconds_bucket{endpoint="/v1/chat",le="0.1"} 321 ai_backend_request_latency_seconds_bucket{endpoint="/v1/chat",le="0.25"} 892 ai_backend_request_latency_seconds_bucket{endpoint="/v1/chat",le="+Inf"} 1245

配合Grafana，你可以做出这样的看板：实时QPS、P99延迟趋势、错误率、LLM调用成功率。这才是真正的“可观测性”，而不是靠print()和日志grep。

5. 常见问题与排查技巧实录：那些文档里不会写的真相

5.1 典型问题速查表

5.2 “为什么我的RAG总是答非所问？”——召回率调试三步法

RAG效果差，90%的问题出在检索阶段，而不是LLM。我用这套方法快速定位：

Step 1：绕过LLM，直接看检索结果

写一个调试端点：

@router.post("/debug/retrieve") def debug_retrieve(query: str): docs = vectorstore.similarity_search(query, k=5) return { "query": query, "retrieved": [ { "content": doc.page_content[:100] + "...", "score": doc.metadata.get("score", 0), "source": doc.metadata.get("source", "unknown") } for doc in docs ] }

访问POST /debug/retrieve，输入你的典型query，观察返回的文档是否真的相关。如果返回的都是无关内容，说明Embeddings或向量库有问题。

Step 2：检查Embeddings质量

用bge-m3生成两个句子的向量，计算余弦相似度：

from sklearn.metrics.pairwise import cosine_similarity import numpy as np embedding = HuggingFaceEmbeddings(model_name="BAAI/bge-m3") vec1 = embedding.embed_query("苹果手机电池续航怎么样？") vec2 = embedding.embed_query("iPhone的battery life is how long?") vec3 = embedding.embed_query("华为手机屏幕分辨率多少？") sim12 = cosine_similarity([vec1], [vec2])[