ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

首页/资讯中心/资讯详情

从零搭建 FastAPI+RAG 垃圾分类回收智能问答系统

发布时间：2026/8/13 9:30:05 拓冰企业网站定制
从零搭建 FastAPI+RAG 垃圾分类回收智能问答系统

回收站分类管理 + RAG 知识库检索 + AI 对话 Agent

技术栈清单

  1. 后端：FastAPI、Tortoise-ORM（数据库）、ChromaDB 向量库、DashScope 通义千问 Embedding
  2. 前端：Vue3 + Axios + Vite 代理跨域
  3. 业务功能：
    • 回收站、分类、居民 CRUD 管理接口
    • 本地 MD 文档向量化入库（RAG 知识库）
    • 基于知识库的语义检索
    • AI 智能问答对话接口

前置准备

  1. Python 3.10+ 环境
  2. Node.js 16+（前端运行）
  3. 阿里云百炼 DashScope API Key（免费开通，后文替换密钥）
  4. 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

第一部分：后端项目从零搭建（FastAPI+RAG 核心）

步骤 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框架 fastapi==0.104.1 uvicorn==0.24.0 # ORM数据库 tortoise-orm==0.19.3 # 向量数据库chromadb chromadb==0.4.18 # 阿里云大模型兼容OpenAI SDK openai==1.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(pk=True, description="分类主键ID") name = fields.CharField(max_length=50, description="分类名称") is_delete = fields.IntField(default=0, description="逻辑删除：0未删 1已删除") create_time = fields.DatetimeField(auto_now_add=True, description="创建时间") class Meta: table = "recycle_category" class RecycleStation(Model): """回收站站点表""" id = fields.IntField(pk=True, description="站点ID") name = fields.CharField(max_length=100, description="站点名称") address = fields.CharField(max_length=200, description="站点地址") category_id = fields.IntField(description="关联回收分类ID") status = fields.IntField(default=1, description="站点状态：1开放 0暂停") is_delete = fields.IntField(default=0, description="逻辑删除") create_time = fields.DatetimeField(auto_now_add=True) class Meta: table = "recycle_station" class Resident(Model): """居民用户表""" id = fields.IntField(pk=True) name = fields.CharField(max_length=20, description="居民姓名") phone = fields.CharField(max_length=11, description="手机号") is_delete = fields.IntField(default=0) 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(default=1, 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_key=settings.DASHSCOPE_API_KEY, base_url=settings.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', encoding='utf-8') as f: content = f.read() # 第三步：调用文档切片方法，把长文档切割成小片段 chunks = cls._split_document(content) if not chunks: return False, "文档切分失败，文档内容为空" # 第四步：创建向量库文件夹，不存在则自动创建 cls.VECTOR_DB_PATH.mkdir(parents=True, exist_ok=True) # 第五步：初始化持久化ChromaDB客户端（断电不丢失向量数据） chroma_client = chromadb.PersistentClient(path=str(cls.VECTOR_DB_PATH)) # 第六步：如果集合已存在先删除，避免重复入库重复数据 try: chroma_client.delete_collection(cls.COLLECTION_NAME) except Exception: pass # 第七步：新建向量集合，使用余弦相似度匹配文本 collection = chroma_client.create_collection( name=cls.COLLECTION_NAME, metadata={"hnsw:space": "cosine"} ) # 第八步：调用阿里云Embedding接口，把文本转为向量数组 embedding_client = cls.get_embedding_client() response = embedding_client.embeddings.create( model="text-embedding-v3", input=chunks ) embeddings = [item.embedding for item in response.data] # 第九步：批量存入向量数据库 ids = [f"chunk_{i}" for i in range(len(chunks))] collection.add( ids=ids, embeddings=embeddings, documents=chunks, 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(path=str(cls.VECTOR_DB_PATH)) collection = chroma_client.get_collection(name=cls.COLLECTION_NAME) # 将用户提问转为向量 client = cls.get_embedding_client() response = client.embeddings.create( model="text-embedding-v3", input=[query] ) query_embedding = response.data[0].embedding # 相似度检索 results = collection.query( query_embeddings=[query_embedding], n_results=k ) # 格式化返回文本给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( name=station_data.name, address=station_data.address, category_id=station_data.category_id, status=station_data.status ) return station

步骤 8：AI 对话 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_key=settings.DASHSCOPE_API_KEY, base_url=settings.DASHSCOPE_BASE_URL ) # 系统提示词：限定AI只回答回收相关问题 system_prompt = f""" 你是社区旧物回收智能客服，仅根据以下知识库内容回答用户问题： 知识库参考内容： {rag_context} 如果知识库没有相关信息，直接回复：暂时没有找到该回收相关知识。不要编造内容。 """ # 调用通义千问大模型生成回复 response = client.chat.completions.create( model="qwen-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_model=list[CategoryOut], summary="获取所有回收分类") async def get_categories(): categories = await RecycleCategory.filter(is_delete=0).all() return categories @recycle_router.get("/stations", response_model=StationListOut, summary="回收站分页列表") async def get_stations( name: Optional[str] = Query(None, description="站点名称模糊搜索"), status: Optional[int] = Query(None, description="1开放 0暂停"), page: int = Query(1, ge=1), page_size: int = Query(10, ge=1, le=100) ): 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(id=station.category_id, is_delete=0) if not category: raise HTTPException(status_code=400, detail="所选分类不存在") res = await RecycleService.create_station(station) return {"code": 1, "message": "新增成功", "data": res} @recycle_router.get("/residents", response_model=list[ResidentOut], summary="获取全部居民") async def get_residents(): return await Resident.filter(is_delete=0).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_model=ChatResponse, summary="智能问答对话接口") async def chat(request: ChatRequest): res = await AgentService.chat(request.user_query, request.resident_id) return ChatResponse(code=1, message="OK", data=res["reply"], sources=res["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", version="B卷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_url=settings.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", host="127.0.0.1", port=8000, reload=True)

步骤 11：知识库文档创建（新手必做，解决文档不存在报错）

  1. 在项目根目录新建文件夹知识库
  2. 在文件夹内新建文件社区旧物回收指南.md，写入测试内容示例：
# 社区旧物回收指南 ## 一、塑料回收规则 1. 矿泉水瓶、塑料外卖盒属于可回收物，投放至塑料回收站点 2. 带油污塑料盒需要清洗干净后回收 ## 二、金属回收 易拉罐、铁锅、金属衣架可回收，电池属于有害垃圾，禁止投放普通回收站 ## 三、站点开放时间 所有社区回收站工作日8:00-18:00开放，周末9:00-17:00

后端启动运行教程

  1. 终端进入项目根目录，执行启动命令
uvicorn main:app --reload --port 8000
  1. 打开浏览器访问接口文档：http://127.0.0.1:8000/docs
  2. 先调用/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.js

import 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

第三部分：完整业务流程演示

  1. 修改app/config/settings.py替换自己的 DashScope 密钥
  2. 创建知识库/社区旧物回收指南.md写入测试文档
  3. 后端启动uvicorn main:app --reload --port 8000
  4. 打开 http://127.0.0.1:8000/docs 调用/rag/init初始化向量库
  5. 启动前端npm run dev
  6. 页面调用接口：
    • 先执行initVectorStore()初始化知识库
    • 再查询分类、回收站、调用 AI 问答
  7. 全部业务链路跑通：前端请求→Vite 代理→FastAPI 后端→RAG 向量检索→AI 大模型返回回答
网站建设 企业官网 运营推广
返回列表