回收站分类管理 RAG 知识库检索 AI 对话 Agent技术栈清单后端FastAPI、Tortoise-ORM数据库、ChromaDB 向量库、DashScope 通义千问 Embedding前端Vue3 Axios Vite 代理跨域业务功能回收站、分类、居民 CRUD 管理接口本地 MD 文档向量化入库RAG 知识库基于知识库的语义检索AI 智能问答对话接口前置准备Python 3.10 环境Node.js 16前端运行阿里云百炼 DashScope API Key免费开通后文替换密钥Windows/Mac 本地文件夹存放项目官方文档参考标注 FastAPI 官方文档https://fastapi.tiangolo.com/ ChromaDB 向量库官方文档https://docs.trychroma.com/ 阿里云 DashScope Embedding 接口文档https://help.aliyun.com/document_detail/241186.htm Vite 官方代理文档https://cn.vitejs.dev/config/server-options#server-proxy第一部分后端项目从零搭建FastAPIRAG 核心步骤 1创建项目文件夹与目录结构新建项目根目录recycle_b_project完整目录结构直接对照创建文件夹recycle_b_project/ ├─ app/ # 后端核心业务目录 │ ├─ config/ # 全局配置 │ │ └─ settings.py # 密钥、全局变量配置 │ ├─ models/ # ORM数据库模型 │ │ └─ recycle.py │ ├─ schemas/ # 数据校验Schema │ │ └─ recycle.py │ ├─ services/ # 业务服务层RAG/回收管理/Agent对话 │ │ ├─ recycle_service.py # 回收站业务逻辑 │ │ ├─ rag_service.py # RAG向量知识库你提供的代码完整版注释 │ │ └─ agent_service.py # AI对话Agent服务 │ └─ api/ # 路由管理 │ └─ recycle_router.py # 全部API路由 ├─ 知识库/ # 存放回收指南MD文档必须新建 │ └─ 社区旧物回收指南.md # RAG读取的知识库文件 ├─ vector_store/ # 自动生成存放向量数据库文件无需手动建 ├─ main.py # FastAPI程序入口 └─ requirements.txt # Python依赖清单步骤 2安装 Python 依赖在项目根目录打开终端新建requirements.txt复制以下内容# FastAPI web框架 fastapi0.104.1 uvicorn0.24.0 # ORM数据库 tortoise-orm0.19.3 # 向量数据库chromadb chromadb0.4.18 # 阿里云大模型兼容OpenAI SDK openai1.9.0 # 数据处理 python-multipart python-dotenv终端执行安装命令pip install -r requirements.txt步骤 3全局配置文件 app/config/settings.py替换密钥业务说明这里统一存放 DashScope 密钥、数据库配置新手只需要修改DASHSCOPE_API_KEY即可其余无需改动 全局配置文件存放密钥、数据库、向量库通用配置 参考文档FastAPI 全局配置最佳实践 https://fastapi.tiangolo.com/advanced/settings/ from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): # 【新手仅需修改此处】阿里云百炼API密钥 # 登录阿里云百炼平台获取自己的key替换下面字符串 DASHSCOPE_API_KEY: str sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx # # DashScope兼容OpenAI接口地址 DASHSCOPE_BASE_URL: str https://你的业务空间ID.cn-beijing.maas.aliyuncs.com/compatible-mode/v1 # 向量库集合名称全局统一 VECTOR_COLLECTION_NAME: str recycle_knowledge # Sqlite数据库文件路径轻量本地数据库无需装MySQL DB_URL: str sqlite://recycle_db.sqlite3 # 实例化全局配置所有业务文件直接导入使用 settings Settings()步骤 4数据库模型 app/models/recycle.py使用 Tortoise ORM自动创建数据表无需手动操作数据库 数据库ORM模型回收站、分类、居民数据表 参考文档Tortoise ORM https://tortoise-orm.readthedocs.io/ from tortoise import fields, Model class RecycleCategory(Model): 回收分类表塑料、金属、纸张等分类 id fields.IntField(pkTrue, description分类主键ID) name fields.CharField(max_length50, description分类名称) is_delete fields.IntField(default0, description逻辑删除0未删 1已删除) create_time fields.DatetimeField(auto_now_addTrue, description创建时间) class Meta: table recycle_category class RecycleStation(Model): 回收站站点表 id fields.IntField(pkTrue, description站点ID) name fields.CharField(max_length100, description站点名称) address fields.CharField(max_length200, description站点地址) category_id fields.IntField(description关联回收分类ID) status fields.IntField(default1, description站点状态1开放 0暂停) is_delete fields.IntField(default0, description逻辑删除) create_time fields.DatetimeField(auto_now_addTrue) class Meta: table recycle_station class Resident(Model): 居民用户表 id fields.IntField(pkTrue) name fields.CharField(max_length20, description居民姓名) phone fields.CharField(max_length11, description手机号) is_delete fields.IntField(default0) class Meta: table resident步骤 5数据校验 Schema app/schemas/recycle.py用于接口入参、出参格式校验FastAPI 自动返回友好报错 接口数据校验Schema规范前端传入参数、后端返回格式 FastAPI Schema文档https://fastapi.tiangolo.com/tutorial/schema-extra-example/ from pydantic import BaseModel, Field from typing import Optional, List # 分类返回格式 class CategoryOut(BaseModel): id: int name: str class Config: from_attributes True # 新增回收站入参 class StationCreate(BaseModel): name: str Field(description站点名称) address: str Field(description站点地址) category_id: int Field(description所属分类ID) status: Optional[int] Field(default1, description站点状态) # 单站点返回 class StationOut(BaseModel): id: int name: str address: str category_id: int status: int # 分页站点列表返回 class StationListOut(BaseModel): total: int page: int page_size: int data: List[StationOut] # 居民返回格式 class ResidentOut(BaseModel): id: int name: str phone: str class Config: from_attributes True # AI对话请求体 class ChatRequest(BaseModel): user_query: str Field(description用户提问内容) resident_id: int Field(description提问居民ID) # AI对话返回体 class ChatResponse(BaseModel): code: int message: str data: str sources: Optional[List[str]] None步骤 6核心 RAG 向量库服务 app/services/rag_service.py全注释版业务逻辑读取项目根目录知识库/社区旧物回收指南.md文档自动切片、调用阿里云 Embedding 生成向量存入 ChromaDB提供初始化入库、语义检索接口 B卷 RAG知识库向量服务 完整注释版 ChromaDB官方文档https://docs.trychroma.com/getting-started 阿里云Embedding接口文档https://help.aliyun.com/document_detail/241186.htm import os from pathlib import Path import chromadb from chromadb.config import Settings from openai import OpenAI # 导入全局配置仅需修改settings里的API Key from app.config.settings import settings class RAGService: # -------------------------- 全局路径常量配置 -------------------------- # 向量库持久化存储文件夹自动生成在项目根目录 VECTOR_DB_PATH Path(__file__).parent.parent.parent / vector_store # 向量库集合名称全局统一 COLLECTION_NAME settings.VECTOR_COLLECTION_NAME # 知识库MD文档路径项目根目录/知识库/社区旧物回收指南.md # 新手注意必须手动新建【知识库】文件夹放入对应md文件否则报文档不存在 DOC_PATH Path(__file__).parent.parent.parent / 知识库 / 社区旧物回收指南.md classmethod def get_embedding_client(cls): 获取阿里云DashScope Embedding客户端 兼容OpenAI SDK格式无需额外适配代码 client OpenAI( api_keysettings.DASHSCOPE_API_KEY, base_urlsettings.DASHSCOPE_BASE_URL ) return client classmethod async def init_vector_store(cls): 接口POST /rag/init 功能初始化向量库读取本地MD文档切片后存入向量数据库 返回(是否成功, 提示信息) try: # 第一步校验知识库文件是否存在新手最容易报错的地方 if not cls.DOC_PATH.exists(): return False, f知识库文档不存在: {cls.DOC_PATH} # 第二步读取本地md文档utf-8编码防止中文乱码 with open(cls.DOC_PATH, r, encodingutf-8) as f: content f.read() # 第三步调用文档切片方法把长文档切割成小片段 chunks cls._split_document(content) if not chunks: return False, 文档切分失败文档内容为空 # 第四步创建向量库文件夹不存在则自动创建 cls.VECTOR_DB_PATH.mkdir(parentsTrue, exist_okTrue) # 第五步初始化持久化ChromaDB客户端断电不丢失向量数据 chroma_client chromadb.PersistentClient(pathstr(cls.VECTOR_DB_PATH)) # 第六步如果集合已存在先删除避免重复入库重复数据 try: chroma_client.delete_collection(cls.COLLECTION_NAME) except Exception: pass # 第七步新建向量集合使用余弦相似度匹配文本 collection chroma_client.create_collection( namecls.COLLECTION_NAME, metadata{hnsw:space: cosine} ) # 第八步调用阿里云Embedding接口把文本转为向量数组 embedding_client cls.get_embedding_client() response embedding_client.embeddings.create( modeltext-embedding-v3, inputchunks ) embeddings [item.embedding for item in response.data] # 第九步批量存入向量数据库 ids [fchunk_{i} for i in range(len(chunks))] collection.add( idsids, embeddingsembeddings, documentschunks, metadatas[{source: 社区旧物回收指南.md} for _ in chunks] ) # 全部流程执行完成返回成功 return True, f向量库初始化成功共 {len(chunks)} 个文档片段 except Exception as e: # 捕获全部异常返回错误信息给前端 return False, f初始化失败: {str(e)} classmethod async def search(cls, query: str, k: int 3): 接口GET /rag/search 功能用户提问在向量库检索最相似的3段知识库文本 :param query: 用户输入的回收相关问题 :param k: 返回匹配片段数量默认3条 :return: 拼接后的知识库参考文本 try: # 连接本地向量库 chroma_client chromadb.PersistentClient(pathstr(cls.VECTOR_DB_PATH)) collection chroma_client.get_collection(namecls.COLLECTION_NAME) # 将用户提问转为向量 client cls.get_embedding_client() response client.embeddings.create( modeltext-embedding-v3, input[query] ) query_embedding response.data[0].embedding # 相似度检索 results collection.query( query_embeddings[query_embedding], n_resultsk ) # 格式化返回文本给AI对话使用 documents results[documents][0] formatted_results [] for idx, doc in enumerate(documents, 1): formatted_results.append(f[参考片段{idx}]: {doc}) return \n\n.join(formatted_results) except Exception as e: return f检索失败: {str(e)} staticmethod def _split_document(content: str, chunk_size: int 500, overlap: int 50): 私有工具方法文档智能切片 规则按markdown标题分割单段最大500字符相邻片段保留50字符重叠避免上下文丢失 sections [] current_section [] current_length 0 # 按换行分割全文 lines content.split(\n) for line in lines: line line.strip() if not line: continue # 遇到Markdown标题 / 文本长度超出限制新建片段 if line.startswith(#) or current_length len(line) chunk_size: if current_section: sections.append(\n.join(current_section)) current_section [line] current_length len(line) else: current_section.append(line) current_length len(line) 1 # 存入最后一段未完成文本 if current_section: sections.append(\n.join(current_section)) # 处理片段重叠提升检索连贯性 chunks [] for i in range(len(sections)): chunk sections[i] # 添加上一段尾部重叠文本 if i 0 and overlap 0: prev_text sections[i-1] prev_tail prev_text[-overlap:] if len(prev_text) overlap else prev_text chunk prev_tail \n chunk chunks.append(chunk) # 兜底无分段时返回全文 return chunks if chunks else [content]步骤 7回收站业务服务 app/services/recycle_service.py实现分类、站点、居民分页查询、新增逻辑 回收站业务服务层处理分类、站点、居民数据库CRUD逻辑 from typing import Optional from app.models.recycle import RecycleCategory, RecycleStation, Resident from app.schemas.recycle import StationCreate class RecycleService: staticmethod async def get_stations(name: Optional[str], status: Optional[int], page: int, page_size: int): 分页查询回收站站点支持名称、状态模糊筛选 # 初始化查询过滤条件 filter_dict {is_delete: 0} if name: filter_dict[name__contains] name if status is not None: filter_dict[status] status # 总条数 total await RecycleStation.filter(**filter_dict).count() # 分页数据 station_list await RecycleStation.filter(**filter_dict).offset((page-1)*page_size).limit(page_size).all() return { total: total, page: page, page_size: page_size, data: station_list } staticmethod async def create_station(station_data: StationCreate): 新增回收站站点 station await RecycleStation.create( namestation_data.name, addressstation_data.address, category_idstation_data.category_id, statusstation_data.status ) return station步骤 8AI 对话 Agent 服务 app/services/agent_service.py基于 RAG 检索结果结合大模型生成智能回收问答回复 AI对话Agent服务结合RAG知识库检索结果调用大模型生成回答 from openai import OpenAI from app.config.settings import settings from app.services.rag_service import RAGService class AgentService: classmethod async def chat(cls, user_query: str, resident_id: int): AI问答主逻辑 1. 调用RAG检索知识库参考文本 2. 拼接系统提示词交给大模型生成答案 # 第一步检索相关回收知识库内容 rag_context await RAGService.search(user_query) # 第二步初始化大模型客户端 client OpenAI( api_keysettings.DASHSCOPE_API_KEY, base_urlsettings.DASHSCOPE_BASE_URL ) # 系统提示词限定AI只回答回收相关问题 system_prompt f 你是社区旧物回收智能客服仅根据以下知识库内容回答用户问题 知识库参考内容 {rag_context} 如果知识库没有相关信息直接回复暂时没有找到该回收相关知识。不要编造内容。 # 调用通义千问大模型生成回复 response client.chat.completions.create( modelqwen-turbo, messages[ {role: system, content: system_prompt}, {role: user, content: user_query} ] ) reply response.choices[0].message.content return { reply: reply, sources: rag_context.split(\n\n) }步骤 9统一 API 路由 app/api/recycle_router.py整合全部业务接口分三大模块回收站管理、RAG 知识库、AI 对话 B卷全部API路由汇总 FastAPI路由文档https://fastapi.tiangolo.com/tutorial/bigger-applications/ from fastapi import APIRouter, Query, HTTPException from typing import Optional from app.models.recycle import RecycleCategory, RecycleStation, Resident from app.schemas.recycle import CategoryOut, StationCreate, StationListOut, ResidentOut, ChatRequest, ChatResponse from app.services.recycle_service import RecycleService from app.services.rag_service import RAGService from app.services.agent_service import AgentService # ---------------- 1. 回收站管理路由 前缀/recycle ---------------- recycle_router APIRouter(prefix/recycle, tags[回收站管理]) recycle_router.get(/categories, response_modellist[CategoryOut], summary获取所有回收分类) async def get_categories(): categories await RecycleCategory.filter(is_delete0).all() return categories recycle_router.get(/stations, response_modelStationListOut, summary回收站分页列表) async def get_stations( name: Optional[str] Query(None, description站点名称模糊搜索), status: Optional[int] Query(None, description1开放 0暂停), page: int Query(1, ge1), page_size: int Query(10, ge1, le100) ): return await RecycleService.get_stations(name, status, page, page_size) recycle_router.post(/stations, summary新增回收站) async def create_station(station: StationCreate): category await RecycleCategory.get_or_none(idstation.category_id, is_delete0) if not category: raise HTTPException(status_code400, detail所选分类不存在) res await RecycleService.create_station(station) return {code: 1, message: 新增成功, data: res} recycle_router.get(/residents, response_modellist[ResidentOut], summary获取全部居民) async def get_residents(): return await Resident.filter(is_delete0).all() # ---------------- 2. RAG知识库路由 前缀/rag ---------------- rag_router APIRouter(prefix/rag, tags[RAG向量知识库]) rag_router.post(/init, summary初始化向量库读取本地MD文档入库) async def init_vector_store(): success, msg await RAGService.init_vector_store() return {code: 1 if success else 0, message: msg} rag_router.get(/search, summary知识库语义检索) async def rag_search(query: str Query(..., description用户回收问题)): data await RAGService.search(query) return {code: 1, data: data} # ---------------- 3. AI对话路由 前缀/chat ---------------- chat_router APIRouter(prefix/chat, tags[AI智能对话]) chat_router.post(/message, response_modelChatResponse, summary智能问答对话接口) async def chat(request: ChatRequest): res await AgentService.chat(request.user_query, request.resident_id) return ChatResponse(code1, messageOK, datares[reply], sourcesres[sources])步骤 10程序入口 main.py启动 FastAPI、初始化数据库、挂载全部路由 FastAPI项目启动入口 运行命令uvicorn main:app --reload --port 8000 访问接口文档http://127.0.0.1:8000/docs from fastapi import FastAPI from tortoise import Tortoise from app.api.recycle_router import recycle_router, rag_router, chat_router from app.config.settings import settings # 创建FastAPI实例 app FastAPI(title社区旧物回收管理系统API, versionB卷1.0) # 挂载全部业务路由 app.include_router(recycle_router) app.include_router(rag_router) app.include_router(chat_router) # 项目启动事件自动创建数据库表 app.on_event(startup) async def startup(): await Tortoise.init( db_urlsettings.DB_URL, modules{models: [app.models.recycle]} ) await Tortoise.generate_schemas() # 自动生成数据表 # 项目关闭事件断开数据库连接 app.on_event(shutdown) async def shutdown(): await Tortoise.close_connections() if __name__ __main__: import uvicorn uvicorn.run(main:app, host127.0.0.1, port8000, reloadTrue)步骤 11知识库文档创建新手必做解决文档不存在报错在项目根目录新建文件夹知识库在文件夹内新建文件社区旧物回收指南.md写入测试内容示例# 社区旧物回收指南 ## 一、塑料回收规则 1. 矿泉水瓶、塑料外卖盒属于可回收物投放至塑料回收站点 2. 带油污塑料盒需要清洗干净后回收 ## 二、金属回收 易拉罐、铁锅、金属衣架可回收电池属于有害垃圾禁止投放普通回收站 ## 三、站点开放时间 所有社区回收站工作日8:00-18:00开放周末9:00-17:00后端启动运行教程终端进入项目根目录执行启动命令uvicorn main:app --reload --port 8000打开浏览器访问接口文档http://127.0.0.1:8000/docs先调用/rag/init接口初始化向量库解决前端知识库报错第二部分前端 Vue3 Vite 跨域配置新手配套步骤 1前端 vite.config.js 完整配置适配后端 8000 端口路由import { defineConfig } from vite import vue from vitejs/plugin-vue import { resolve } from path export default defineConfig({ plugins: [vue()], resolve: { alias: { : resolve(__dirname, src) } }, server: { port: 3000, proxy: { // 匹配前端/api/recycle 请求转发后端/recycle /api/recycle: { target: http://localhost:8000, changeOrigin: true, rewrite: (path) path.replace(/^\/api\/recycle/, /recycle) }, // /api/rag 直接转发后端/rag /api/rag: { target: http://localhost:8000, changeOrigin: true, rewrite: (path) path.replace(/^\/api/, ) }, // /api/chat 直接转发后端/chat /api/chat: { target: http://localhost:8000, changeOrigin: true, rewrite: (path) path.replace(/^\/api/, ) } } } })步骤 2前端 axios 请求封装 src/api/index.jsimport axios from axios // 统一基础路径 /api由vite代理转发 const api axios.create({ baseURL: /api/, timeout: 30000 }) // 回收站管理接口 // 获取全部回收分类 export const getCategories () api.get(/recycle/categories) // 分页查询回收站 export const getStations (params) api.get(/recycle/stations, { params }) // 新增回收站 export const addStation (data) api.post(/recycle/stations, data) // 获取全部居民 export const getResidents () api.get(/recycle/residents) // RAG知识库接口 // 初始化向量库解决文档不存在报错的核心接口 export const initVectorStore () api.post(/rag/init) // 知识库检索 export const ragSearch (query) api.get(/rag/search, { params: { query } }) // AI对话接口 export const chat (data) api.post(/chat/message, data) export default api第三部分完整业务流程演示修改app/config/settings.py替换自己的 DashScope 密钥创建知识库/社区旧物回收指南.md写入测试文档后端启动uvicorn main:app --reload --port 8000打开 http://127.0.0.1:8000/docs 调用/rag/init初始化向量库启动前端npm run dev页面调用接口先执行initVectorStore()初始化知识库再查询分类、回收站、调用 AI 问答全部业务链路跑通前端请求→Vite 代理→FastAPI 后端→RAG 向量检索→AI 大模型返回回答