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