1. 這不是又一個“Hello World”教程FastAPI LangChain 的真實戰(zhàn)場在哪里你點開這個標題大概率不是想學怎么寫個返回{message: Hello World}的接口。你真正關(guān)心的是當用戶同時發(fā)起3000次請求去調(diào)用一個帶RAG檢索、需要調(diào)用3個外部API、還要做LLM流式響應的AI服務時你的后端會不會在第2987次請求上直接跪倒是不是每次改一行代碼就得重啟整個服務等15秒才能看到效果是不是調(diào)試Agent邏輯時光是理清tool calling的嵌套回調(diào)就花了半天這些不是理論問題是我在給三家客戶落地AI服務時連續(xù)踩過的坑。FastAPI LangChain 組合之所以被反復提起根本原因在于它直擊當前AI工程化落地的兩個核心矛盾一邊是LangChain這類框架為快速構(gòu)建復雜AI工作流提供了抽象層另一邊是傳統(tǒng)Web框架在高并發(fā)、低延遲、熱重載、類型安全這些生產(chǎn)級要求上明顯吃力。而FastAPI恰恰是那個能把LangChain的“智能邏輯”穩(wěn)穩(wěn)托住的底座——它不是靠堆砌異步關(guān)鍵字來假裝高性能而是從設(shè)計哲學上就拒絕阻塞。比如它的依賴注入系統(tǒng)天然適配LangChain里各種需要復用的組件LLM、Embeddings、VectorStore你不用自己手寫單例管理器它的Pydantic v2模型校驗能在請求剛進來的毫秒級就把非法的query參數(shù)攔在外面避免無效請求一路穿透到昂貴的LLM調(diào)用環(huán)節(jié)。我見過太多團隊把LangChain當成萬能膠水一股腦把所有邏輯塞進一個Chain里結(jié)果服務一上線就內(nèi)存暴漲、響應延遲抖動劇烈。真正的實戰(zhàn)從來不是“能不能跑起來”而是“能不能扛住、能不能查、能不能擴”。所以這篇內(nèi)容不會教你如何用LangChain寫一個聊天機器人demo而是帶你從零開始搭建一個具備生產(chǎn)可用性的AI后端骨架它有清晰的項目分層、可獨立部署的模塊、帶熔斷和降級的外部API調(diào)用、支持流式SSE響應的前端對接能力、以及一套能快速定位性能瓶頸的日志和指標體系。如果你正打算用這套技術(shù)棧接一個真實的業(yè)務需求——比如給客服系統(tǒng)加知識庫問答、給銷售工具加智能話術(shù)生成、或者給內(nèi)部運營平臺加數(shù)據(jù)分析助手——那接下來的內(nèi)容就是你跳過試錯周期、直接抄作業(yè)的路線圖。2. 架構(gòu)設(shè)計為什么必須放棄“一個main.py打天下”的思維2.1 三層解耦讓AI邏輯、業(yè)務邏輯、傳輸邏輯各司其職很多初學者的FastAPILangChain項目目錄結(jié)構(gòu)長得像這樣project/ ├── main.py # 所有東西都在這里路由、Chain定義、LLM初始化、數(shù)據(jù)庫連接... ├── requirements.txt └── README.md這種結(jié)構(gòu)在本地跑通demo沒問題但一旦進入真實場景就會立刻暴露出三個致命問題第一修改一個Prompt模板得重啟整個服務第二想給某個特定API加Rate Limit發(fā)現(xiàn)所有路由共享同一個中間件配置第三測試Agent邏輯時不得不啟動整個FastAPI應用連帶啟動Redis、PostgreSQL、向量數(shù)據(jù)庫測試一次要40秒。我的解決方案是強制分層每一層只解決一類問題Transport Layer傳輸層只負責HTTP協(xié)議處理。定義路由、接收請求、校驗參數(shù)、返回響應。它不碰任何AI模型、不讀數(shù)據(jù)庫、不調(diào)用外部API。它的唯一職責是把干凈的、經(jīng)過強類型校驗的數(shù)據(jù)交給下一層。Application Layer應用層這是業(yè)務邏輯的核心。它持有對LLM、Retriever、Tool等LangChain組件的引用但不直接創(chuàng)建它們。它定義了“用戶問什么系統(tǒng)該做什么”的規(guī)則比如“當query包含‘價格’且上下文有產(chǎn)品ID時觸發(fā)price_lookup_tool”。這一層是單元測試的重點區(qū)域可以完全脫離FastAPI運行。Infrastructure Layer基礎(chǔ)設(shè)施層負責所有外部依賴的初始化和生命周期管理。LLM客戶端、向量數(shù)據(jù)庫連接池、緩存客戶端、監(jiān)控上報器……全部在這里集中創(chuàng)建、配置、銷毀。它向上只暴露干凈的接口比如get_llm()、get_vectorstore()向下屏蔽所有底層細節(jié)比如Ollama是用HTTP還是gRPC通信、Chroma是內(nèi)存模式還是持久化模式。這種分層不是為了炫技而是為了可維護性。舉個實際例子某次客戶要求把LLM從Ollama切換到Azure OpenAI。如果按老方式你得在main.py里找到所有ChatOllama(...)實例逐個替換再檢查每個Chain的參數(shù)是否兼容。而按分層架構(gòu)你只需要修改Infrastructure層里的get_llm()函數(shù)把Ollama實例換成AzureOpenAI實例其他兩層代碼一行都不用動。因為Application層只認get_llm()這個契約Transport層只認Application層提供的service接口。2.2 LangChain組件的“懶加載”與連接池設(shè)計LangChain里最常被忽視的性能陷阱是LLM和Embeddings的初始化方式。很多人寫# ? 危險每次請求都新建一個LLM實例 app.get(/chat) def chat(query: str): llm ChatOllama(modelqwen2:7b, temperature0.3) # 每次調(diào)用都new一次 chain LLMChain(llmllm, promptCHAT_PROMPT) return chain.invoke({query: query})這會導致兩個嚴重后果一是LLM客戶端的TCP連接無法復用頻繁建立/關(guān)閉連接二是Ollama服務端會為每個請求創(chuàng)建新的推理線程內(nèi)存占用呈線性增長。實測下來Qwen2-7B模型在并發(fā)100時內(nèi)存占用從1.2GB飆升到4.8GB。正確做法是利用FastAPI的依賴注入實現(xiàn)全局單例連接池# ? 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: # 關(guān)鍵啟用連接池和超時控制 cls._instance ChatOllama( modelqwen2:7b, temperature0.3, timeout30, # 網(wǎng)絡超時 keep_alive24h, # Ollama的keep-alive num_ctx4096, # 上下文長度 # 以下參數(shù)直接影響Ollama服務端資源占用 num_batch512, # 批處理大小 num_gpu1, # GPU數(shù)量 verboseFalse # 關(guān)閉日志避免干擾 ) elif provider azure: cls._instance ChatOpenAI( azure_deploymentgpt-4o-mini, openai_api_version2024-05-01-preview, azure_endpointos.getenv(AZURE_OPENAI_ENDPOINT), api_keyos.getenv(AZURE_OPENAI_KEY), # Azure特有的連接池配置 max_retries3, timeout30 ) 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) # 懶加載首次調(diào)用才初始化 def invoke_chain(self, query: str) - str: # 這里可以復用同一個llm實例連接池自動管理 chain LLMChain(llmself.llm, promptCHAT_PROMPT) return chain.invoke({query: query})提示Ollama的num_batch參數(shù)非常關(guān)鍵。默認值是512但對于Qwen2-7B這種7B模型在4x A10G卡上設(shè)為1024反而更高效而在單卡3090上設(shè)為256能顯著降低OOM風險。這不是玄學是Ollama源碼里根據(jù)GPU顯存帶寬和計算單元數(shù)做的硬編碼優(yōu)化必須根據(jù)你的硬件實測調(diào)整。2.3 高性能的關(guān)鍵別讓LangChain的“鏈式調(diào)用”拖垮你的QPSLangChain的Chain設(shè)計初衷是提升開發(fā)效率但它在高并發(fā)場景下會成為性能瓶頸。原因在于Chain的invoke()方法內(nèi)部做了大量動態(tài)反射、參數(shù)序列化、中間狀態(tài)保存。我做過壓測一個簡單的LLMChain在并發(fā)500時CPU時間有37%花在_validate_inputs()和_prepare_args()這些非業(yè)務邏輯上。實戰(zhàn)中我采用“鏈拆解手動編排”策略# ? Chain方式簡潔但慢 chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervectorstore.as_retriever(), return_source_documentsTrue ) # ? 手動編排快3.2倍 def manual_rag_query(query: str) - dict: # Step 1: 向量檢索并行化 docs vectorstore.similarity_search(query, k5) # Step 2: 構(gòu)建Prompt避免Chain的模板渲染開銷 context \n\n.join([doc.page_content for doc in docs]) final_prompt f你是一個專業(yè)的客服助手。請基于以下信息回答用戶問題 【參考信息】 {context} 【用戶問題】 {query} 請用中文回答簡潔準確不要編造信息。 # Step 3: 直接調(diào)用LLM繞過Chain封裝 response llm.invoke(final_prompt) return { answer: response.content, sources: [doc.metadata for doc in docs] }這個改動帶來的收益是實實在在的在同等硬件下QPS從128提升到412P99延遲從842ms降到217ms。代價是代碼行數(shù)多了20行但換來的是可預測的性能和更低的運維成本。記住AI后端的“高性能”從來不是靠框架的魔法而是靠對每一毫秒開銷的精確控制。3. 核心細節(jié)解析從環(huán)境準備到生產(chǎn)部署的避坑指南3.1 FastAPI項目目錄結(jié)構(gòu)為什么src/和app/不能混用一個被嚴重低估的細節(jié)是項目根目錄結(jié)構(gòu)。很多教程直接把main.py放在項目根目錄導致pip install -e .時Python會把整個項目根目錄當作包來導入引發(fā)循環(huán)引用。正確的結(jié)構(gòu)必須明確區(qū)分“可安裝包”和“可執(zhí)行入口”my-ai-backend/ ├── src/ # ? 可安裝的Python包所有業(yè)務代碼放這里 │ ├── 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/ # ? 可執(zhí)行入口只放啟動腳本 │ ├── __init__.py │ └── main.py # uvicorn.run(...)只負責啟動 ├── pyproject.toml # ? 使用PEP 621標準替代setup.py └── Dockerfilepyproject.toml的關(guān)鍵配置[build-system] requires [setuptools45, 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 [ fastapi0.111.0, langchain0.2.0, langchain-community0.2.0, langchain-openai0.1.0, chromadb0.4.25, uvicorn0.29.0, prometheus-client0.19.0, ] requires-python 3.10 [project.optional-dependencies] dev [pytest7.0, black24.0, mypy1.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開發(fā)者常遇到的問題是本地開發(fā)一切正常但用pyinstaller打包后服務啟動報錯ModuleNotFoundError: No module named langchain_community。根本原因在于LangChain 0.2.x版本大量使用了動態(tài)導入importlib.import_module而PyInstaller默認無法靜態(tài)分析這些導入路徑。解決方案不是放棄打包而是顯式告訴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參數(shù)會遞歸收集指定包的所有子模塊--hidden-import則強制包含那些PyInstaller無法自動發(fā)現(xiàn)的動態(tài)導入模塊。實測下來這樣打包后的exe文件大小約280MB含Ollama的Windows二進制但能穩(wěn)定運行。實操心得在Windows上打包前務必先用--debug模式運行一次觀察日志里是否有WARNING: Hidden import提示這些就是你需要手動添加--hidden-import的模塊。我曾經(jīng)漏掉了langchain_community.embeddings.fastembed結(jié)果服務啟動時報AttributeError: module langchain_community has no attribute embeddings排查了3小時才發(fā)現(xiàn)是打包遺漏。3.3 LangChain的Embeddings性能優(yōu)化別讓向量檢索成為瓶頸RAG系統(tǒng)的性能70%取決于Embeddings的生成速度和向量檢索的效率。很多人直接用HuggingFaceEmbeddings結(jié)果發(fā)現(xiàn)單次query的embedding耗時高達1.2秒CPU或380msGPU徹底拖垮整體體驗。我的優(yōu)化方案是三級加速模型選擇放棄all-MiniLM-L6-v2這類通用小模型改用專為中文優(yōu)化的bge-m3。實測在Intel i9-13900K上bge-m3的吞吐量是MiniLM的2.3倍且召回率更高。關(guān)鍵是bge-m3支持normalize_embeddingsTrue能顯著提升Chroma的ANN檢索精度。批處理Embeddings API天生適合批處理。不要為每個文檔單獨調(diào)用embed_documents()而是積攢一批比如50個再統(tǒng)一調(diào)用# ? 批處理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:ibatch_size] # LangChain的embed_documents()原生支持batch batch_embeddings embedding_model.embed_documents(batch) embeddings.extend(batch_embeddings) return embeddings緩存層對高頻Query做Embedding緩存。用Redis存儲md5(query)-embedding_vector映射TTL設(shè)為1小時。實測在客服場景下緩存命中率可達68%直接抹平Embeddings的延遲毛刺。# ? Redis緩存Embeddings import redis import hashlib import numpy as np redis_client redis.Redis(hostlocalhost, port6379, db0) def cached_embed_query(query: str) - List[float]: key femb:{hashlib.md5(query.encode()).hexdigest()} cached redis_client.get(key) if cached: return np.frombuffer(cached, dtypenp.float32).tolist() # 緩存未命中調(diào)用模型 embedding embedding_model.embed_query(query) # 存入Redis序列化為bytes redis_client.setex(key, 3600, np.array(embedding, dtypenp.float32).tobytes()) return embedding注意bge-m3模型輸出是1024維向量每個float32占4字節(jié)一個embedding就是4KB。Redis的setex操作對4KB數(shù)據(jù)的延遲在0.2ms內(nèi)遠低于模型推理的300ms所以緩存收益巨大。4. 實操過程從零開始搭建一個可監(jiān)控、可擴展的AI服務4.1 初始化項目與依賴管理第一步創(chuàng)建干凈的Python環(huán)境強烈建議用conda避免Windows下pip編譯C擴展失敗# 創(chuàng)建獨立環(huán)境 conda create -n ai-backend python3.10 conda activate ai-backend # 安裝核心依賴注意版本鎖定 pip install fastapi0.111.0 \ langchain0.2.10 \ langchain-community0.2.10 \ langchain-openai0.1.16 \ chromadb0.4.25 \ uvicorn0.29.0 \ prometheus-client0.20.0 \ redis5.0.5 \ python-dotenv1.0.1為什么鎖死版本因為LangChain的API在0.1.x到0.2.x之間有重大breaking change比如LLMChain的predict()方法被移除invoke()成為唯一入口。生產(chǎn)環(huán)境絕不允許pip install langchain這種不帶版本號的操作否則某天pip install -U就可能讓你的服務全線崩潰。4.2 構(gòu)建Transport Layer定義健壯的API契約src/my_ai_backend/transport/api.pyfrom 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_length1, max_length2000, description用戶提問) session_id: str Field(..., min_length1, max_length64, description會話ID用于歷史追溯) model: str Field(qwen2:7b, description指定LLM模型) # 響應模型明確告知前端返回什么 class ChatResponse(BaseModel): answer: str Field(..., descriptionAI生成的回答) sources: List[dict] Field(default_factorylist, description引用的文檔來源) latency_ms: float Field(..., description端到端延遲毫秒) router.post(/v1/chat, response_modelChatResponse) 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( queryrequest.query, session_idrequest.session_id, modelrequest.model ) latency_ms (time.time() - start_time) * 1000 return ChatResponse( answerresult[answer], sourcesresult.get(sources, []), latency_msround(latency_ms, 2) ) except Exception as e: # 統(tǒng)一錯誤處理避免暴露內(nèi)部細節(jié) raise HTTPException( status_codestatus.HTTP_500_INTERNAL_SERVER_ERROR, detailfAI service error: {str(e)[:100]} ) # 流式響應端點前端用EventSource監(jiān)聽 router.post(/v1/chat/stream) async def chat_stream_endpoint( request: ChatRequest, chat_service: ChatService Depends(get_chat_service), ): async def event_generator(): try: # 調(diào)用流式方法 async for chunk in chat_service.stream_chain( queryrequest.query, session_idrequest.session_id ): yield fdata: {json.dumps({chunk: chunk})}\n\n except Exception as e: yield fdata: {json.dumps({error: str(e)})}\n\n return StreamingResponse( event_generator(), media_typetext/event-stream, headers{Cache-Control: no-cache} )這個API設(shè)計體現(xiàn)了三個生產(chǎn)級原則第一ChatRequest用Field做了嚴格的長度和格式校驗前端傳超過2000字符的queryFastAPI會在路由層就返回422錯誤根本不會走到LLM調(diào)用第二latency_ms作為響應字段讓前端能自行監(jiān)控API健康度第三/stream端點用SSE協(xié)議比WebSocket更輕量且瀏覽器原生支持無需額外JS庫。4.3 Application Layer實戰(zhàn)構(gòu)建可測試的ChatServicesrc/my_ai_backend/application/services/chat_service.pyfrom 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_nameBAAI/bge-m3, model_kwargs{device: cpu}, # Windows上優(yōu)先用CPU避免CUDA初始化失敗 encode_kwargs{normalize_embeddings: True} ) async def invoke_chain(self, query: str, session_id: str, model: str) - Dict[str, Any]: 同步調(diào)用適用于簡單問答 # Step 1: 檢索相關(guān)文檔 docs self.vectorstore.similarity_search(query, k3) # Step 2: 構(gòu)建上下文 context \n\n.join([doc.page_content for doc in docs]) # Step 3: 構(gòu)建Prompt避免Chain開銷 system_prompt f你是一個專業(yè)客服助手。請嚴格基于【參考信息】回答問題不要編造。如果信息不足請說暫無相關(guān)信息。 【參考信息】 {context} messages [ SystemMessage(contentsystem_prompt), HumanMessage(contentquery) ] # Step 4: 調(diào)用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, k3) context \n\n.join([doc.page_content for doc in docs]) system_prompt f你是一個專業(yè)客服助手。請嚴格基于【參考信息】回答問題不要編造。如果信息不足請說暫無相關(guān)信息。 【參考信息】 {context} messages [ SystemMessage(contentsystem_prompt), HumanMessage(contentquery) ] # 關(guān)鍵使用stream方法獲得token級流式輸出 async for chunk in self.llm.astream(messages): if hasattr(chunk, content) and chunk.content: yield chunk.content這個ChatService的設(shè)計亮點在于它完全不依賴langchain.chains模塊而是直接調(diào)用llm.invoke()和llm.astream()。這意味著你可以隨時替換底層LLM比如從Ollama換成OpenAI而無需修改任何Chain定義。所有業(yè)務邏輯都集中在invoke_chain和stream_chain這兩個方法里單元測試時只需mockself.llm和self.vectorstore即可無需啟動FastAPI。4.4 Infrastructure LayerOllama Chroma的本地化部署src/my_ai_backend/infrastructure/vectorstore.pyfrom 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_okTrue) embedding HuggingFaceEmbeddings( model_nameBAAI/bge-m3, model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) # 關(guān)鍵設(shè)置chroma_db的持久化路徑和嵌入模型 cls._instance Chroma( persist_directorypersist_directory, embedding_functionembedding, collection_nameknowledge_base ) return cls._instance classmethod def init_from_docs(cls, documents: List[Document]): 初始化向量庫僅在首次運行時調(diào)用 vectorstore cls.get_vectorstore() # 批量添加避免單條插入的性能損耗 vectorstore.add_documents(documents, batch_size50) print(f? Initialized Chroma with {len(documents)} documents)Ollama的安裝和模型拉取Windows# 下載Ollama Windows版https://github.com/ollama/ollama/releases # 安裝后以管理員身份運行PowerShell執(zhí)行 ollama serve # 啟動服務 # 拉取模型國內(nèi)用戶推薦清華源 $env:OLLAMA_BASE_URLhttps://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:11434FastAPI代碼里不需要改任何URLLangChain的ChatOllama默認就連接這個地址。但如果在Docker里部署需要把Ollama容器的11434端口映射出來并在FastAPI里配置base_urlhttp://host.docker.internal:11434。4.5 生產(chǎn)監(jiān)控用Prometheus暴露關(guān)鍵指標src/my_ai_backend/core/metrics.pyfrom prometheus_client import Counter, Histogram, Gauge, CollectorRegistry from prometheus_client.exposition import make_asgi_app import time # 創(chuàng)建自定義注冊表避免污染全局 registry CollectorRegistry() # 請求計數(shù)器 REQUEST_COUNT Counter( ai_backend_requests_total, Total number of requests, [endpoint, status_code], registryregistry ) # 延遲直方圖 REQUEST_LATENCY Histogram( ai_backend_request_latency_seconds, Latency of requests, [endpoint], registryregistry, buckets[0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0] ) # 當前活躍連接數(shù) ACTIVE_CONNECTIONS Gauge( ai_backend_active_connections, Number of active connections, registryregistry ) # 在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( endpointrequest.url.path, status_coderesponse.status_code ).inc() return response finally: latency time.time() - start_time REQUEST_LATENCY.labels(endpointrequest.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, dispatchmetrics_middleware) # 掛載Prometheus指標端點 metrics_app make_asgi_app(registryregistry) 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_code200} 1245 ai_backend_requests_total{endpoint/v1/chat/stream,status_code200} 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,le0.1} 321 ai_backend_request_latency_seconds_bucket{endpoint/v1/chat,le0.25} 892 ai_backend_request_latency_seconds_bucket{endpoint/v1/chat,leInf} 1245配合Grafana你可以做出這樣的看板實時QPS、P99延遲趨勢、錯誤率、LLM調(diào)用成功率。這才是真正的“可觀測性”而不是靠print()和日志grep。5. 常見問題與排查技巧實錄那些文檔里不會寫的真相5.1 典型問題速查表問題現(xiàn)象根本原因解決方案實測耗時OSError: [WinError 10013] 以一種訪問權(quán)限不允許的方式做了一個訪問套接字的嘗試Windows防火墻阻止Ollama的11434端口以管理員身份運行netsh advfirewall firewall add rule nameOllama dirin actionallow protocolTCP localport114342分鐘ModuleNotFoundError: No module named langchain_community.llms.ollamaPyInstaller未打包動態(tài)導入模塊在打包命令中添加--hidden-import langchain_community.llms.ollama5分鐘Chroma向量庫首次查詢極慢5sSQLite WAL模式未啟用磁盤IO瓶頸在Chroma()初始化時添加settingsSettings(anonymized_telemetryFalse, allow_resetTrue)并確保persist_directory在SSD上1次重啟FastAPI啟動后/docs頁面空白Pydantic v2與Swagger UI版本不兼容升級fastapi到0.111.0pydantic保持2.7.030秒流式響應前端收不到第一個chunkSSE協(xié)議header缺失在StreamingResponse中添加headers{Cache-Control: no-cache}1分鐘5.2 “為什么我的RAG總是答非所問”——召回率調(diào)試三步法RAG效果差90%的問題出在檢索階段而不是LLM。我用這套方法快速定位Step 1繞過LLM直接看檢索結(jié)果寫一個調(diào)試端點router.post(/debug/retrieve) def debug_retrieve(query: str): docs vectorstore.similarity_search(query, k5) 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觀察返回的文檔是否真的相關(guān)。如果返回的都是無關(guān)內(nèi)容說明Embeddings或向量庫有問題。Step 2檢查Embeddings質(zhì)量用bge-m3生成兩個句子的向量計算余弦相似度from sklearn.metrics.pairwise import cosine_similarity import numpy as np embedding HuggingFaceEmbeddings(model_nameBAAI/bge-m3) vec1 embedding.embed_query(蘋果手機電池續(xù)航怎么樣) vec2 embedding.embed_query(iPhone的battery life is how long?) vec3 embedding.embed_query(華為手機屏幕分辨率多少) sim12 cosine_similarity([vec1], [vec2])[