08 Chroma 持久化 + FastAPI 封装

Chroma 持久化 FastAPI 封装安装依赖cdD:\yunyanshijie\skill_test\python_study .venv\Scripts\activate pipinstallchromadb langchain-chromaChroma 持久化为什么要换InMemoryVectorStoreChroma存储位置进程内存磁盘 chroma_db/重启后索引丢失需重新 Embedding直接加载秒级启动适用场景学习、调试接近真实部署核心 APIfromlangchain_chromaimportChromadefbuild_vectorstore(chunks,embeddings,chroma_dir:Path):chroma_dir.mkdir(parentsTrue,exist_okTrue)ifany(chroma_dir.iterdir()):# 目录非空 → 已有索引,库已存在直接加载print(从磁盘加载向量库...)returnChroma(persist_directorystr(chroma_dir),embedding_functionembeddings,collection_nameknowledge,)else:# 首次 → 建库文档切分后print(首次运行正在建索引...)returnChroma.from_documents(documentschunks,embeddingembeddings,persist_directorystr(chroma_dir),collection_nameknowledge,)FastAPI 封装架构POST /chatstartup 一次ClientFastAPIagent_corechroma_dbLangGraphDeepSeek重点向量库和 Graph 在 服务启动时初始化一次不是每个请求重建。生命周期lifespanFastAPI 推荐用 lifespan在 startup 里完成初始化fromcontextlibimportasynccontextmanager graphNoneasynccontextmanagerasyncdeflifespan(app:FastAPI):globalgraph# 在函数里要赋值模块级变量时必须写否则 Python 会当成函数局部变量graphinit_graph()# 内部调 build_vectorstore build_graphyield# 关服务 → yield 之后的代码再跑清理# shutdown 清理appFastAPI(lifespanlifespan)接口设计classChatRequest(BaseModel):message:strthread_id:strdefaultclassChatResponse(BaseModel):reply:strsources:list[str][]# 这是 RAG 的 溯源citation不只给答案还告诉用户「依据哪些文档」。路由方法作用/healthGET健康检查返回 {“status”: “ok”, “chunks”: N}/chatPOST发送消息返回回答 引用来源/chat 核心逻辑app.post(/chat,response_modelChatResponse)defchat(req:ChatRequest):config{configurable:{thread_id:req.thread_id}}resultgraph.invoke({messages:[HumanMessage(contentreq.message)],context:},configconfig,)replyresult[messages][-1].content sourcesresult.get(sources,[])# 见下方「溯源」returnChatResponse(replyreply,sourcessources)检索溯源在 AgentState 里加一个字段classAgentState(TypedDict):messages:Annotated[list,add_messages]context:strsources:list[str]# 新增retrieve 节点里sources[str(doc.metadata.get(source,unknown))fordocinresults]return{context:context,sources:sources}