【基于FastAPI+RAG开发本地智能知识助手 】ChainMind RAG智能知识库系统实战学习文档
·
嘿，朋友！欢迎来到 ChainMind 的学习之旅。这是一个非常有意思的项目，它能让你理解如何构建一个智能知识库问答系统——就像给 AI 装上了一个会思考的图书馆！通过这个项目，你会学到 RAG 技术、FastAPI 后端开发、前后端交互等实用技能。准备好了吗？让我们开始吧！
📋 学习前准备
在开始学习之前，你需要具备以下基础知识：
必学基础
- Python 基础：熟悉函数、类、异步编程（async/await）的基本概念
- HTTP 基础：理解 GET/POST 请求、JSON 数据格式
- 命令行操作：能够在终端执行命令
推荐了解
- JavaScript 基础：能读懂 React 组件（前端部分）
- 数据库概念：知道什么是存储和查询
环境准备
| 工具 | 版本要求 | 安装说明 |
|---|---|---|
| Python | 3.8+ | 官网下载或使用 conda |
| Node.js | 16+ | 官网下载 |
| Ollama | 最新版 | 用于运行本地 LLM 模型 |
环境检查清单
- Python 版本：
python --version显示 3.8+ - Node.js 版本：
node --version显示 16+ - 包管理器：pip 或 npm 可用
🗺️ 学习路线图
整个学习过程可以分为以下几个阶段，建议按顺序学习：
plaintext
第一阶段：理解概念（1-2小时）
│
├── 什么是 RAG？为什么需要它？
├── 向量数据库是什么？
└── FastAPI 为什么这么火？
│
▼
第二阶段：后端开发（3-4小时）
│
├── FastAPI 核心语法
├── 路由和中间件
├── 服务层架构
└── 单例模式的应用
│
▼
第三阶段：核心功能（4-5小时）
│
├── 文档上传和处理
├── 文本切分策略
├── 向量存储和检索
└── RAG 检索增强
│
▼
第四阶段：高级特性（2-3小时）
│
├── 记忆系统的实现
├── 智能代理模式
└── 工具注册表设计
│
▼
第五阶段：前端交互（2-3小时）
│
├── React 组件开发
├── 状态管理
├── API 调用封装
└── CORS 跨域处理
🎯 学完后你能做什么
学完这个项目后，你将具备以下能力：
1. 构建自己的 RAG 系统
- 能够上传文档（PDF、TXT、MD）
- 实现基于文档内容的智能问答
- 优化检索效果和回答质量
2. 掌握 FastAPI 全栈开发
- 快速构建高性能的 Python API
- 设计清晰的服务层架构
- 处理文件上传和复杂业务逻辑
3. 理解 AI 应用架构
- RAG 的原理和实现方式
- Agent 智能代理的设计模式
- 记忆系统和上下文管理
4. 面试加分项 💰
- 能讲清楚 RAG 的工作原理
- 能解释向量检索的思路
- 能说明前后端分离架构的设计
安装步骤：
```bash
# 1. 进入后端目录
cd backend
# 2. 安装依赖
pip install -r requirements.txt
# 3. 启动后端服务
cd app
uvicorn main:app --host 0.0.0.0 --port 8000 --reload# 1. 进入前端目录
cd frontend
# 2. 安装依赖
npm install
# 3. 启动前端服务
npm run dev第一部分：核心技术概念
1.1 RAG 技术详解
什么是 RAG？
RAG = Retrieval（检索）+ Augmented（增强）+ Generation（生成）
想象一下，如果你要回答"《红楼梦》里林黛玉葬花时说了什么"这个问题，你会怎么做？
没有 RAG 的做法（就像一个死记硬背的学生）：
- 直接凭记忆回答："花谢花飞花满天，红消香断有谁怜..."
- 问题：如果记错了呢？如果问题超出记忆范围呢？
有 RAG 的做法（就像一个会查资料的学生）
- 先去图书馆找到《红楼梦》的相关章节 🔍
- 仔细阅读找到具体内容 📖
- 根据找到的内容组织答案 ✍️
RAG 就是给 AI 装了一个"随时可以查阅的图书馆"！
RAG 的工作流程
plaintext
用户提问："什么是机器学习？"
│
▼
┌─────────────────────────────────────┐
│ 1️⃣ 检索阶段 (Retrieval) │
│ │
│ "把问题变成坐标，在图书馆（本地知识库）里找书" │
│ │
│ 问题 → 向量化 → 在向量库中搜索 │
│ ↓ │
│ 找到最相关的 3 个文档片段 │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 2️⃣ 增强阶段 (Augmented) │
│ │
│ "把找到的资料整理好" │
│ │
│ 将检索结果组合成上下文 │
│ 加上原始问题，组成完整提示 │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 3️⃣ 生成阶段 (Generation) │
│ │
│ "根据资料写出答案" │
│ │
│ LLM 阅读上下文 + 问题 → 生成答案 │
└─────────────────────────────────────┘
│
▼
最终答案输出
为什么需要 RAG？
| 方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 纯 LLM | 回答流畅、上下文理解好 | 可能胡编乱造、知识过时 | 创意写作、通用对话 |
| RAG | 答案有据可查、可更新知识 | 需要额外存储和检索 | 知识库问答、文档分析 |
加分点：RAG 解决了 LLM 的两个核心问题——幻觉问题（胡编乱造）和知识过时问题（无法获取最新信息）。
1.2 向量数据库详解
什么是向量？
向量 = 一串数字 = 空间中的一个点
想象一下：每个文档都可以被"翻译"成一串数字，这串数字就是文档的向量表示。
plaintext
"机器学习是人工智能的一个分支"
↓ 向量化 ↓
[0.23, -0.45, 0.78, 0.12, -0.33, ...] ← 这就是向量！
"深度学习使用神经网络"
↓ 向量化 ↓
[0.25, -0.42, 0.75, 0.15, -0.30, ...] ← 相似的句子，向量也相似！
什么是向量检索？
向量检索 = 在空间中找"最近的邻居
类比理解：
- 把每本书想象成图书馆里的一个书架位置 📚
- 相似的内容，书架位置就靠近
- 当你问问题时，先找到最近的几个书架，然后从这些书架上找答案
简单检索 vs 向量检索
plaintext
代码中的向量存储（简化版）
python
# 这是一个简化版的向量存储
# 实际生产环境会使用 Chroma、Pinecone、Milvus 等专业向量数据库
class VectorStore:
def __init__(self):
self.documents = [] # 存储文档内容
def add_documents(self, chunks):
"""添加文档"""
for chunk in chunks:
self.documents.append({
"content": chunk,
# 实际会存储：embedding（向量）、metadata（元数据）
})
def search(self, query, top_k=3):
"""检索相关文档"""
# 1. 将查询转为向量
# 2. 计算与所有文档的相似度
# 3. 返回 top_k 个最相似的
1.3 FastAPI 快速入门
为什么选择 FastAPI？
| 框架 | 特点 | 学习曲线 | 性能 |
|---|---|---|---|
| Flask | 轻量、灵活 | 简单 | 中等 |
| Django | 全功能、ORM | 陡峭 | 良好 |
| FastAPI | 现代、自动化、快 | 中等 | 极佳 |
FastAPI 的三大优势：
-
自动生成 API 文档 📚
- 写完代码就有 Swagger UI 可以测试
- 再也不用手动写 API 文档了！
-
类型安全 🛡️
- 用 Pydantic 自动验证请求数据
- 写错类型直接报错，不用等到运行时才发现
-
异步支持 ⚡
- 原生支持 async/await
- 高并发处理能力
FastAPI 核心概念
python
from fastapi import FastAPI
app = FastAPI()
# 装饰器定义路由
@app.get("/hello")
async def say_hello():
return {"message": "Hello World!"}
# 启动方式
# uvicorn main:app --host 0.0.0.0 --port 8000
请求流程图：
plaintext
用户请求 GET /hello
│
▼
┌───────────────────────┐
│ FastAPI 应用 │
│ │
│ 路由匹配 → /hello │
│ ↓ │
│ 执行 say_hello() │
│ ↓ │
│ Pydantic 验证响应 │
└───────────────────────┘
│
▼
返回 JSON 响应
1.4 CORS 跨域原理
什么是跨域？
跨域 = 浏览器的安全机制
浏览器规定：如果一个网页想从 A 网站获取数据，必须得到 A 网站的"许可"。
plaintext
浏览器地址：http://localhost:3000 (前端)
│
│ 发送请求
▼
http://localhost:8000/api/chat (后端)
❌ 浏览器拦截！因为端口不同（3000 vs 8000）
怎么解决？
python
# 后端 FastAPI 配置 CORS
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 允许所有来源（开发环境）
# 生产环境应该设置为具体域名：["http://localhost:3000"]
allow_credentials=True,
allow_methods=["*"], # 允许所有请求方法
allow_headers=["*"], # 允许所有请求头
)
类比理解：CORS 就像是在后端门口挂一个牌子："欢迎来自 3000 端口的访客！"
⚠️ 安全注意事项
python
# ❌ 生产环境不要这样写
allow_origins=["*"] # 允许所有来源
# ✅ 生产环境应该指定具体域名
allow_origins=["http://localhost:3000", "https://yourdomain.com"]
1.5 单例模式详解
什么是单例模式？
单例 = 只创建一个实例
类比理解：
- 餐厅只有一个厨房（实例）
- 所有服务员（调用方）都共用这个厨房
- 不会每来一个服务员就新建一个厨房
为什么需要单例？
python
# ❌ 每次调用都创建新实例（浪费资源）
def bad_example():
store = VectorStore() # 每次都 new 一个新的
return store
# ✅ 全局只创建一次（单例模式）
_vector_store = None # 全局变量
def get_vector_store():
global _vector_store
if _vector_store is None: # 懒加载：用到时才创建
_vector_store = VectorStore()
return _vector_store
单例的好处
- 节省资源：不需要重复创建和销毁对象
- 保持一致：所有地方用的是同一个实例，数据同步
- 延迟加载：用的时候才创建，提高启动速度
第二部分：后端核心模块详解
2.1 主入口文件 (main.py)
学习目标
- 理解 FastAPI 应用的创建和配置
- 掌握路由注册的方法
- 了解 CORS 中间件的配置
技术原理
FastAPI 应用本质上是一个 ASGI 应用，它接收请求、处理业务逻辑、返回响应。
python
"""
FastAPI 应用主入口
"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
# 以相对路径导入，就是从当前目录下的 routers/ 包中导入 upload, chat, rag, memory, agent
from .routers import upload, chat, rag, memory, agent
# 1️⃣ 创建FastAPI应用实例
app = FastAPI(
title="ChainMind API", # API 标题（文档中显示）
description="基于 FastAPI + Ollama + LangChain 的本地知识库问答系统API",
version="1.0.0" # 版本号
)
# 2️⃣ 配置 CORS 中间件（即是否允许前端跨域访问，例子：前端从3000端访问后端8000）
# 中间件：在请求到达路由之前/响应返回之前做一些处理
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 生产环境要改成具体域名这个项目是http://localhost:3000！
allow_credentials=True, # 是否允许携带 Cookie
allow_methods=["*"], # 允许的方法：GET, POST, PUT, DELETE...
allow_headers=["*"], # 允许的请求头
)
# 3️⃣ 注册路由
# 路由：将 URL 路径映射到处理函数
app.include_router(upload.router) # /upload/*
app.include_router(chat.router) # /chat/*
app.include_router(rag.router) # /rag/*
app.include_router(memory.router) # /memory/*
app.include_router(agent.router) # /agent/*
# 根路径的健康检查
@app.get("/")
async def root():
"""根路径 - 健康检查"""
return {
"name": "ChainMind API",
"status": "running",
"version": "1.0.0",
"endpoints": {
"docs": "/docs",
"upload": "/upload",
"chat": "/chat",
"rag": "/rag",
"memory": "/memory",
"agent": "/agent"
}
}
# 4️⃣ 健康检查端点
@app.get("/health")
async def health_check():
"""
健康检查 = 探测服务是否存活
类比：去医院体检，检查身体各项指标是否正常
用途：k8s/运维监控、负载均衡探测
"""
return {"status": "healthy", "message": "ChainMind API is running"}
# 本地开发服务器启动
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
代码执行流程
plaintext
启动服务: uvicorn main:app --port 8000
│
▼
┌─────────────────────────┐
│ 创建 FastAPI 应用 │
│ 加载配置文件 │
└─────────────────────────┘
│
▼
┌─────────────────────────┐
│ 注册 CORS 中间件 │
│ （跨域资源共享） │
└─────────────────────────┘
│
▼
┌─────────────────────────┐
│ 注册所有路由模块 │
│ - upload.router │
│ - chat.router │
│ - rag.router │
│ - memory.router │
│ - agent.router │
└─────────────────────────┘
│
▼
┌─────────────────────────┐
│ 服务启动完成 │
│ 监听 http://localhost │
│ :8000 │
└─────────────────────────┘
│
▼
接收请求 → 匹配路由 → 执行处理函数 → 返回响应
面试考点
Q1: FastAPI 和 Flask 有什么区别？
回答要点：
- FastAPI 自动生成 API 文档（Swagger UI）
- FastAPI 基于 Pydantic 提供自动类型验证
- FastAPI 原生支持异步（async/await）
- Flask 更轻量灵活，FastAPI 更适合构建大型 API 服务
Q2: 什么是 ASGI？
回答要点：
- ASGI = Asynchronous Server Gateway Interface（异步服务器网关接口）
- 是 WSGI 的异步升级版
- 支持长连接、WebSocket 等特性
- Uvicorn 是 ASGI 服务器的实现
Q3: 中间件是什么？执行顺序是怎样的？
回答要点：
- 中间件是拦截请求/响应的"拦截器"
- 请求顺序：外部中间件 → 内部中间件 → 路由处理
- 响应顺序：路由处理 → 内部中间件 → 外部中间
2.2 配置文件 (config.py)
学习目标
- 理解配置文件的最佳实践
- 掌握关键配置参数的含义
- 学会根据环境切换配置
技术原理
配置分离的好处：
- 开发/生产环境使用不同配置
- 敏感信息不硬编码在代码中
- 方便统一管理和修改
python
"""
应用配置文件
"""
import os
from pathlib import Path
# 项目根目录
BASE_DIR = Path(__file__).parent.parent.parent
# 数据目录
DATA_DIR = BASE_DIR / "data"
UPLOAD_DIR = DATA_DIR / "uploads"
CHROMA_DIR = DATA_DIR / "chroma_db"
# 确保目录存在
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
CHROMA_DIR.mkdir(parents=True, exist_ok=True)
# ========== Ollama 配置 ==========
# Ollama = 本地运行的 LLM 模型服务
# 类似 OpenAI API，但数据不会离开你的电脑 💻
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434") # Ollama 默认端口
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "llama3:8b") # 使用的模型名称
# ========== 文档处理配置RAG ==========
DEFAULT_TOP_K = 3
CHUNK_SIZE = 500 # 每个文本块的字符数
CHUNK_OVERLAP = 50 # 相邻块之间的重叠字符数
# ========== 向量存储配置 ==========
VECTOR_STORE_PERSIST_DIR = "./chroma_db" # 向量数据库的存储路径
# Embedding配置
EMBEDDING_MODEL = "all-MiniLM-L6-v2"
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
文档切分策略详解
为什么需要切分？
plaintext
假设有一本 10 万字的书 📚
❌ 直接处理整本书：
- LLM 的上下文窗口有限，装不下
- 检索粒度太粗，找到的内容太多
✅ 切成小块处理：
- 每个小块 500 字符，刚好合适
- 可以精确定位到相关内容
chunk_size 和 chunk_overlap 的关系：
plaintext
文本: "ABCDEFGHIJKLMNOPQRSTUVWXYZ" (26个字符)
chunk_size=10, chunk_overlap=5
块1: [0-10) → "ABCDEFGHIJ"
块2: [5-15) → "FGHIJLMNOP" ← 与块1重叠5个字符
块3: [10-20) → "LMNOPQRSTU"
块4: [15-25) → "PQRSTUVWXYZ"
好处: "LMNO" 不会因为切分而丢失上下文！
延伸思考：如何选择合适的 chunk_size？
| 场景 | 推荐 chunk_size | 原因 |
|---|---|---|
| 短问答 | 200-300 字符 | 问题简单，不需要太多上下文 |
| 文档分析 | 500-800 字符 | 保留更多语义完整性 |
| 代码检索 | 100-200 字符 | 代码块通常较小 |
| 长文章总结 | 1000+ 字符 | 需要足够上下文来总结 |
2.3 上传路由 (upload.py)
学习目标
- 掌握 FastAPI 文件上传的处理方式
- 理解文件类型验证和大小限制
- 学会处理批量上传
技术原理
文件上传的流程：
plaintext
用户选择文件 → 浏览器将文件封装为 FormData → POST 请求发送到服务器
│
▼
服务器接收 → 验证文件类型/大小 → 保存到磁盘
│
▼
调用文档处理服务 → 返回处理结果
核心代码解析
python
"""
文档上传路由
功能：接收用户上传的文档，保存到服务器，并调用文档处理服务
"""
import os
import shutil
from pathlib import Path
from typing import List
from fastapi import APIRouter, UploadFile, File, HTTPException
from fastapi.responses import JSONResponse
from ..services.document_processor import process_document
from ..services.vector_store import get_vector_store
# FastAPI 路由配置
router = APIRouter(prefix="/upload", tags=["文档上传"])
# 支持的文件类型（白名单机制）
SUPPORTED_EXTENSIONS = {".txt", ".md", ".pdf"}
def get_upload_dir():
"""获取上传目录的绝对路径"""
base_dir = Path(__file__).parent.parent
data_dir = base_dir / "data"
upload_dir = data_dir / "uploads"
os.makedirs(upload_dir, exist_ok=True)
return upload_dir
@router.post("/file")
async def upload_file(file: UploadFile = File(...)):
"""
上传单个文档
核心步骤：
1. 验证文件类型
2. 检查文件大小
3. 保存文件到磁盘
4. 处理文档内容
5. 添加到向量库
"""
# 📌 1. 验证文件类型
file_extension = Path(file.filename).suffix.lower()
if file_extension not in SUPPORTED_EXTENSIONS:
raise HTTPException(
status_code=400,
detail=f"不支持的文件类型: {file_extension}，仅支持: {', '.join(SUPPORTED_EXTENSIONS)}"
)
# 📌 2. 检查文件大小（限制 10MB）
contents = await file.read()
if len(contents) > 10 * 1024 * 1024: # 10MB = 10 * 1024 * 1024
# raise 语句用于主动触发异常
raise HTTPException(status_code=400, detail="文件大小超过 10MB 限制")
# 📌 3. 保存文件
upload_dir = get_upload_dir()
file_path = upload_dir / file.filename
with open(file_path, "wb") as buffer:
buffer.write(contents)
# 📌 4. 处理文档（切分成小块）
# 处理文档
try:
result = process_document(str(file_path))
result["status"] = "success"
# ✅ 自动将 chunks 添加到向量库
if result.get("chunks"):
vector_store = get_vector_store()
vector_store.add_documents(result["chunks"], file.filename)
result["vector_stored"] = True
return result
except Exception as e:
# 处理失败，删除已上传的文件
if file_path.exists():
file_path.unlink()
raise HTTPException(status_code=500, detail=f"文档处理失败: {str(e)}")
@router.post("/files")
async def upload_multiple_files(files: List[UploadFile] = File(...)):
"""
批量上传文档
- files: 要上传的文件列表
Returns:
所有文档的处理结果
"""
results = []
uploaded_files = []
for file in files:
# 检查文件类型
file_extension = Path(file.filename).suffix.lower()
if file_extension not in SUPPORTED_EXTENSIONS:
results.append({
"file_name": file.filename,
"status": "error",
"error": f"不支持的文件类型: {file_extension}"
})
continue
# 保存文件
contents = await file.read()
upload_dir = get_upload_dir()
file_path = upload_dir / file.filename
try:
with open(file_path, "wb") as buffer:
buffer.write(contents)
uploaded_files.append(file_path)
# 处理文档
result = process_document(str(file_path))
result["status"] = "success"
# ✅ 自动将 chunks 添加到向量库
if result.get("chunks"):
vector_store = get_vector_store()
vector_store.add_documents(result["chunks"], file.filename)
result["vector_stored"] = True
results.append(result)
except Exception as e:
results.append({
"file_name": file.filename,
"status": "error",
"error": str(e)
})
# 清理已上传的文件
if file_path.exists():
file_path.unlink()
return {
"total": len(files),
"success_count": sum(1 for r in results if r["status"] == "success"),
"results": results
}
@router.get("/list")
async def list_uploaded_files():
"""
列出已上传的所有文档
"""
files = []
upload_dir = get_upload_dir()
for file_path in upload_dir.iterdir():
if file_path.is_file():
files.append({
"name": file_path.name,
"size": file_path.stat().st_size,
"extension": file_path.suffix
})
return {"files": files}
@router.delete("/{filename}")
async def delete_file(filename: str):
"""
删除指定的已上传文件
"""
upload_dir = get_upload_dir()
file_path = upload_dir / filename
if not file_path.exists():
raise HTTPException(status_code=404, detail="文件不存在")
file_path.unlink()
return {"status": "deleted", "filename": filename}
代码执行流程图
plaintext
POST /upload/file
│
▼
┌───────────────────────────────┐
│ 接收 UploadFile 对象 │
└───────────────────────────────┘
│
▼
┌───────────────────────────────┐
│ 检查文件扩展名 │
│ (.txt/.md/.pdf?) │
│ ❌ 不符合 → 返回 400 错误 │
│ ✅ 符合 → 继续 │
└───────────────────────────────┘
│
▼
┌───────────────────────────────┐
│ 读取文件内容 │
│ 检查大小 > 10MB? │
│ ❌ 超限 → 返回 400 错误 │
│ ✅ 正常 → 继续 │
└───────────────────────────────┘
│
▼
┌───────────────────────────────┐
│ 保存文件到 ./data/uploads/ │
└───────────────────────────────┘
│
▼
┌───────────────────────────────┐
│ 调用 process_document() │
│ - 读取文件内容 │
│ - 切分成小块 (chunks) │
└───────────────────────────────┘
│
▼
┌───────────────────────────────┐
│ 调用 get_vector_store() │
│ - 获取单例向量库 │
│ - 添加 chunks 到向量库 │
└───────────────────────────────┘
│
▼
┌───────────────────────────────┐
│ 返回处理结果 │
│ { │
│ "file_name": "xxx.pdf", │
│ "total_chunks": 42, │
│ "vector_stored": true │
│ } │
└───────────────────────────────┘
实战技巧
1. 防止文件名冲突
python
# ❌ 直接使用原始文件名（可能重复）
file_path = upload_dir / file.filename
# ✅ 使用时间戳或 UUID 生成唯一文件名
import uuid
unique_name = f"{uuid.uuid4().hex}_{file.filename}"
file_path = upload_dir / unique_name
2. 异步文件读取
python
# ✅ 推荐：使用 await 异步读取
contents = await file.read()
# ❌ 不推荐：同步读取会阻塞事件循环
contents = file.file.read()
面试考点
Q1: FastAPI 中 File 和 UploadFile 的区别？
回答要点：
UploadFile提供异步文件操作、更方便的文件元数据访问File直接用于formdata["file"]的场景- 实际上传大文件时，
UploadFile配合async def使用异步读写效率更高
Q2: 为什么要在处理失败时删除已上传的文件？
回答要点：
- 避免磁盘空间浪费
- 保持文件系统整洁
- 防止因部分失败导致的数据不一致
2.4 聊天路由 (chat.py)
学习目标
- 理解 RAG 在聊天场景中的应用
- 掌握聊天历史的管理方式
- 学会构建 RAG 提示词
技术原理
聊天 + RAG 的工作流程：
plaintext
用户: "什么是机器学习？"
│
▼
┌─────────────────────────────────────┐
│ 1️⃣ 检索相关文档 │
│ vector_store.search("什么是机器学习") │
│ ↓ │
│ 找到最相关的 3 个文档片段 │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 2️⃣ 构建上下文 │
│ "来源: 机器学习.pdf" │
│ "内容: 机器学习是..." │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 3️⃣ 构建提示词 │
│ 基于以下上下文回答: │
│ {检索到的内容} │
│ 用户问题: 什么是机器学习？ │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 4️⃣ 调用 LLM 生成答案 │
│ (这里目前是模拟返回) │
└─────────────────────────────────────┘
│
▼
返回答案 + 来源列表
代码解析
python
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import List, Optional
import httpx
import json
import asyncio
from ..services.vector_store import get_vector_store
router = APIRouter(prefix="/chat", tags=["聊天"])
# 模拟的对话历史
chat_history = {}
# AI 对话 API 数据模型设计
class ChatRequest(BaseModel):
question: str
session_id: str = "default"
use_rag: bool = True # 是否启用 RAG 增强
top_k: int = 3 # RAG 检索数量
class SourceItem(BaseModel):
content: str
filename: str
distance: Optional[float] = None
class ChatResponse(BaseModel):
answer: str
sources: List[SourceItem]
used_rag: bool
# Ollama 本地模型 API 交互的流式响应生成器
async def generate_streaming_response(prompt: str, model: str = "llama3:8b"):
"""生成流式响应"""
try:
async with httpx.AsyncClient(timeout=120.0) as client:
async with client.stream(
"POST",
"http://localhost:11434/api/generate",
json={
"model": model,
"prompt": prompt,
"stream": True
}
) as response:
async for line in response.aiter_lines():
if line:
try:
data = json.loads(line)
if "response" in data:
yield data["response"]
except json.JSONDecodeError:
continue
except Exception as e:
yield f"错误: {str(e)}"
async def chat_with_rag(request: ChatRequest):
"""带RAG增强的聊天"""
# 获取向量存储实例
vector_store = get_vector_store()
retrieved_docs = vector_store.search(request.question, top_k=request.top_k)
sources = []
if retrieved_docs:
context_parts = []
for i, doc in enumerate(retrieved_docs, 1):
context_parts.append(f"[文档{i}] ({doc['filename']}):\n{doc['content']}")
context = "\n\n".join(context_parts)
prompt = f"""你是一个知识库问答助手。请根据以下参考资料回答用户的问题。
如果问题与参考资料相关，请基于参考资料回答；如果无关，你可以根据自己的知识回答。（但是务必用中文回答）
参考资料：
{context}
用户问题：{request.question}
回答："""
sources = [
SourceItem(
content=doc["content"],
filename=doc["filename"],
distance=doc.get("distance")
)
for doc in retrieved_docs
]
else:
prompt = f"你是一个知识库问答助手，请回答用户的问题。（用中文回答）\n\n问题：{request.question}"
return prompt, sources
@router.post("/", response_model=ChatResponse)
async def chat(request: ChatRequest):
"""
聊天接口：基于 RAG 的智能问答
📌 Pydantic 模型自动验证请求数据
"""
# 1. 如果启用 RAG，先检索相关文档
prompt = None
sources = []
used_rag = False
# 检索相关文档
# top_k=3 表示返回最相关的 3 个文档
sources = vector_store.search(request.message, top_k=3)
# 构建 RAG 提示词
if request.use_rag:
try:
prompt, sources = await chat_with_rag(request)
used_rag = True
except Exception as e:
prompt = f"你是一个知识库问答助手，请回答用户的问题。（用中文回答）\n\n问题：{request.question}"
else:
prompt = f"你是一个知识库问答助手，请回答用户的问题。（用中文回答）\n\n问题：{request.question}"
# 调用 LLM
answer = ""
try:
async with httpx.AsyncClient(timeout=120.0) as client:
response = await client.post(
"http://localhost:11434/api/generate",
json={
"model": "llama3:8b",
"prompt": prompt,
"stream": False
}
)
result = response.json()
answer = result.get("response", "抱歉，暂时无法回答")
except Exception as e:
answer = f"抱歉，AI 服务暂时不可用：{str(e)}"
# 保存到聊天历史
if request.session_id not in chat_history:
chat_history[request.session_id] = []
chat_history[request.session_id].append({
"question": request.question,
"answer": answer
})
return ChatResponse(
answer=answer,
sources=sources,
used_rag=used_rag
)
@router.post("/stream")
async def chat_stream(request: ChatRequest):
"""
流式聊天问答接口
支持 RAG 增强：从向量库检索相关文档，作为上下文回答
"""
# 1. 如果启用 RAG，先检索相关文档
prompt = None
sources = []
used_rag = False
if request.use_rag:
try:
prompt, sources = await chat_with_rag(request)
used_rag = True
except Exception as e:
prompt = f"你是一个知识库问答助手，请回答用户的问题。（用中文回答）\n\n问题：{request.question}"
else:
prompt = f"你是一个知识库问答助手，请回答用户的问题。（用中文回答）\n\n问题：{request.question}"
# 2. 保存历史
if request.session_id not in chat_history:
chat_history[request.session_id] = []
chat_history[request.session_id].append({
"question": request.question,
"answer": "" # 先占位，流式完成后再更新
})
# 3. 返回流式响应
async def stream_generator():
full_response = ""
try:
async for chunk in generate_streaming_response(prompt, "llama3:8b"):
full_response += chunk
# 发送 SSE 格式的数据
yield f"data: {json.dumps({'token': chunk, 'done': False})}\n\n"
# 流式结束后，发送完成信号
yield f"data: {json.dumps({'token': '', 'done': True, 'sources': [s.dict() for s in sources]})}\n\n"
# 更新历史记录
if chat_history[request.session_id]:
chat_history[request.session_id][-1]["answer"] = full_response
except Exception as e:
error_msg = f"错误: {str(e)}"
yield f"data: {json.dumps({'token': error_msg, 'done': True, 'error': True})}\n\n"
return StreamingResponse(
stream_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no"
}
)
@router.get("/history/{session_id}")
async def get_history(session_id: str):
"""获取聊天历史"""
return {"session_id": session_id, "messages": chat_history.get(session_id, [])}
@router.delete("/history/{session_id}")
async def clear_history(session_id: str):
"""清除聊天历史"""
if session_id in chat_history:
chat_history[session_id] = []
return {"status": "success", "message": f"会话 {session_id} 的历史已清除"}
RAG 提示词设计
好的 RAG 提示词应该包含：
plaintext
📌 角色设定（可选）
你是一个专业的知识库助手
📌 上下文信息
基于以下参考资料回答问题：
---
[文档1的内容]
[文档2的内容]
[文档3的内容]
---
📌 明确要求
1. 只基于提供的上下文回答
2. 如果上下文中没有相关信息，请说明"我没有找到相关信息"
3. 引用时说明来源
📌 用户问题
{用户的实际问题}
面试考点
Q1: RAG 相比纯 LLM 回答有什么优势？
回答要点：
- 可溯源性：答案来自哪个文档，清清楚楚
- 减少幻觉：不是凭空编造，有据可查
- 知识更新：更新文档即可更新知识，无需重新训练
- 成本更低：比 fine-tuning 便宜很多
Q2: 为什么需要保存聊天历史？
回答要点：
- 支持多轮对话，后面的问题可以引用前面的上下文
- 实现"记忆"功能，AI 能记住之前聊过的内容
- 方便用户回顾之前的对话
Q3: 为什么使用 Pydantic BaseModel？
回答要点：
- 自动数据验证（类型、必填、默认值）
- 自动生成文档
2.5 文档处理服务 (document_processor.py)
学习目标
- 掌握不同文档格式的读取方法
- 理解文本切分的策略和实现
- 学会处理边界情况
技术原理
文档处理的三步骤：
plaintext
原始文档 → 文本提取 → 文本切分 → 文本块列表
PDF ↓
TXT ↓
MD ↓
三种格式统一输出为纯文本
核心代码解析
python
"""
文档处理服务
功能：从各类文档（PDF、TXT、MD）中提取文本，并切分成小块
"""
import os
from pathlib import Path
from typing import List
def load_document(file_path: str, file_extension: str) -> str:
"""
根据文件类型加载文档
Args:
file_path: 文件路径
file_extension: 文件扩展名（.txt, .md, .pdf）
Returns:
文档文本内容
设计思路：策略模式
- 不同文件类型用不同的读取方法
- 统一输出为纯文本（str）
"""
if file_extension == ".txt" or file_extension == ".md":
# 文本文件直接读取
with open(file_path, 'r', encoding='utf-8') as f:
return f.read()
elif file_extension == ".pdf":
# 使用 pypdf 读取 PDF
from pypdf import PdfReader
reader = PdfReader(file_path)
text = ""
for page in reader.pages:
text += page.extract_text() + "\n"
return text
else:
raise ValueError(f"不支持的文件类型: {file_extension}")
def split_text(text: str, chunk_size: int = 500, chunk_overlap: int = 50) -> List[str]:
"""
将长文本切分成小块
Args:
text: 要切分的文本
chunk_size: 每个块的最大字符数
chunk_overlap: 相邻块之间的重叠字符数
Returns:
切分后的文本块列表
"""
chunks = []
start = 0
while start < len(text):
# 计算当前块的结束位置
end = start + chunk_size
# 如果不是最后一块，尝试在句子边界处切割
if end < len(text):
# 往后查找最后一个换行符或句子结束符
for sep in ['\n\n', '\n', '。', '！', '？', '. ', '! ', '? ']:
last_sep = text.rfind(sep, start, end)
if last_sep > start:
end = last_sep + len(sep)
break
# 提取当前块
chunk = text[start:end].strip()
if chunk:
chunks.append(chunk)
# 移动起始位置（考虑重叠）
start = end - chunk_overlap
if start <= 0:
start = end
return chunks
def process_document(file_path: str) -> dict:
"""
处理单个文档：加载 + 切分
Args:
file_path: 文件路径
Returns:
包含文档信息的字典
"""
file_path = Path(file_path)
file_extension = file_path.suffix.lower()
# 加载文档
text = load_document(str(file_path), file_extension)
# 切分文档
chunks = split_text(text)
return {
"file_name": file_path.name,
"file_size": os.path.getsize(file_path),
"total_chars": len(text),
"total_chunks": len(chunks),
"chunks": chunks
}
文本切分详解
python
def split_text(text: str, chunk_size: int = 500, chunk_overlap: int = 50) -> List[str]:
"""
智能文本切分
核心思路：
1. 按固定大小切分
2. 在句子边界处微调（避免把一句话切成两半）
3. 保持相邻块之间的重叠
"""
chunks = []
start = 0
while start < len(text):
# 1️⃣ 计算当前块的结束位置
end = start + chunk_size
# 2️⃣ 如果不是最后一块，在句子边界处切割
if end < len(text):
# 按优先级查找句子结束符
# 优先在段落边界切（\n\n）
# 其次是句号（。）
# 最后是换行（\n）
for sep in ['\n\n', '\n', '。', '！', '？', '. ', '! ', '? ']:
last_sep = text.rfind(sep, start, end) # 从后往前找
if last_sep > start: # 找到了
end = last_sep + len(sep) # 在此处切割
break
# 3️⃣ 提取当前块
chunk = text[start:end].strip()
if chunk:
chunks.append(chunk)
# 4️⃣ 移动起始位置（考虑重叠）
# overlap 确保相邻块有共同的上下文
start = end - chunk_overlap
if start <= 0:
start = end
return chunks
切分流程图解
plaintext
原文: "段落一的内容。\n\n段落二的内容。\n\n段落三的内容很长..."
chunk_size=20, chunk_overlap=5
迭代1:
start=0, end=20
在 "。" 处切割
块1: "段落一的内容。"
迭代2:
start=15 (20-5), end=35
在 "\n\n" 处切割
块2: "段落二的内容。"
迭代3:
start=30 (35-5), end=50
块3: "段落三的内容很长..."
实战技巧：PDF 解析的坑
python
# ⚠️ PDF 解析的常见问题：
# 1. 扫描版 PDF（图片）无法直接提取文字
# 需要用 OCR（如 pytesseract）处理
# 2. 复杂排版的 PDF 可能提取不完整
# 可以考虑：
# - 减小 chunk_size
# - 预处理清理格式
# 3. 中文 PDF 的编码问题
# 确保使用 utf-8 编码读取
面试考点
Q1: 如何选择 chunk_size？
回答要点：
- 太小：丢失上下文，每个块语义不完整
- 太大：超出 LLM 上下文窗口，检索粒度粗
- 经验值：500-1000 字符是常见选择
- 可以根据文档类型和业务场景调整
Q2: chunk_overlap 的作用是什么？
回答要点：
- 保证语义完整性：重要信息不会被"腰斩"
- 增加检索命中率：相关内容更容易被匹配到
- 代价是存储和计算量增加约 10%
2.6 向量存储服务 (vector_store.py)
学习目标
- 理解向量存储的设计思路
- 掌握单例模式的实现
- 理解向量检索的基本原理
技术原理
向量存储 = 文档的"数字坐标系统"
类比理解：
- 想象一座城市，每个建筑物都有 GPS 坐标
- 当你问"附近有什么餐厅"时，系统会：
- 把你当前位置转为坐标
- 找最近的几个坐标点
- 返回这些点对应的餐厅信息
核心代码解析：
python
class VectorStore:
"""向量存储类，提供文档存储和检索功能"""
def __init__(self, persist_directory: str = "./chroma_db"):
"""
初始化向量存储
Args:
persist_directory: 持久化目录
"""
self.persist_directory = persist_directory
os.makedirs(persist_directory, exist_ok=True)
# 使用内存存储（生产环境会用 Chroma/FAISS 等）
self.documents = []
def add_documents(self, chunks: List[str], filename: str) -> Dict:
"""
将文本块添加到存储
Args:
chunks: 文本块列表
filename: 来源文件名
Returns:
添加结果
实际生产中会做：
1. 调用 Embedding 模型将文本转为向量
2. 存储向量 + 元数据
"""
if not chunks:
return {"status": "success", "count": 0}
# 添加到内存
for chunk in chunks:
self.documents.append({
"id": f"{filename}_{uuid.uuid4().hex[:8]}",
"content": chunk,
"filename": filename
})
return {
"status": "success",
"count": len(chunks),
"filename": filename
}
def search(self, query: str, top_k: int = 5) -> List[Dict]:
"""
检索相关文档
Args:
query: 查询文本
top_k: 返回结果数量
Returns:
相关文档列表，每项包含 content, filename, distance
简化版：关键词匹配
生产版：向量相似度计算
"""
# 1️⃣ 计算每个文档与查询的匹配分数
results = []
for doc in self.documents:
# 计算简单的匹配分数
score = sum(1 for word in query.lower().split()
if word in doc['content'].lower())
if score > 0:
results.append({
"content": doc['content'],
"filename": doc['filename'],
"distance": 1.0 / (score + 1) # 分数越高，距离越小
})
# 2️⃣ 按相似度排序，返回 top_k 个
results.sort(key=lambda x: x['distance'])
return results[:top_k]
单例模式实现
python
# 全局变量存储单例实例
_vector_store: Optional[VectorStore] = None
def get_vector_store() -> VectorStore:
"""
获取全局向量存储实例（单例模式）
懒加载：用的时候才创建，创建后一直复用
"""
global _vector_store
if _vector_store is None:
_vector_store = VectorStore()
return _vector_store
# 使用示例
store1 = get_vector_store()
store2 = get_vector_store()
print(store1 is store2) # True，同一个实例！
向量检索进阶（了解即可）
生产环境的向量检索：
python
# 实际生产中会使用专业的向量数据库
# 这里以 Chroma 为例：
import chromadb
from chromadb.config import Settings
# 创建客户端
client = chromadb.Client(Settings(
persist_directory="./chroma_db"
))
# 创建集合（类似表）
collection = client.create_collection("documents")
# 添加文档时自动向量化
collection.add(
documents=["文本内容1", "文本内容2"],
ids=["id1", "id2"],
metadatas=[{"source": "file1.pdf"}, {"source": "file2.txt"}]
)
# 检索
results = collection.query(
query_texts=["查询内容"],
n_results=3
)
面试考点
Q1: 什么是向量嵌入（Embedding）？
回答要点：
- 将文本/图像等转换为固定长度的向量
- 相似的文本在向量空间中距离近
- 常用模型：OpenAI Embedding、BGE、M3E
Q2: 向量检索和关键词检索的区别？
回答要点：
维度 关键词检索 向量检索 原理 匹配关键词 计算向量距离 语义 ❌ 不支持 ✅ 支持 同义词 ❌ 不支持 ✅ 支持 速度 快 较慢（需要索引优化） 典型工具 Elasticsearch Chroma, FAISS
Q3: 为什么使用单例模式？
回答要点：
- 节省内存：避免创建多个相同实例
- 保持状态：向量库中的数据全局共享
- 线程安全：单例访问，避免竞态条件
2.7 记忆服务 (memory.py)
学习目标
- 理解对话记忆的设计思路
- 掌握记忆的增删改查操作
- 理解记忆召回的原理
技术原理
为什么需要记忆系统？
plaintext
对话1:
你：我是小明 🎓
AI：你好小明！
对话2:
你：今天吃了啥？
AI：❓ 不知道你是谁，也没记住你说过的话
有记忆系统后:
对话2:
AI：查一下记忆...小明之前说今天要去吃火锅？🍲
核心代码解析(不完整的代码，只做解析)
python
class ConversationBufferMemory:
"""
对话缓冲记忆
管理会话历史，支持 session_id 隔离
"""
def __init__(self):
"""初始化记忆管理器"""
self.sessions: Dict[str, List[dict]] = {}
def add_message(self, session_id: str, role: str, content: str) -> None:
"""
添加消息到会话历史
Args:
session_id: 会话ID
role: 角色 (user/assistant)
content: 消息内容
"""
if session_id not in self.sessions:
self.sessions[session_id] = []
self.sessions[session_id].append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
def get_history(self, session_id: str, limit: Optional[int] = None) -> List[dict]:
"""
获取会话历史
Args:
session_id: 会话ID
limit: 限制返回的消息数量（最近N条）
Returns:
消息历史列表
"""
history = self.sessions.get(session_id, [])
if limit is not None:
return history[-limit:]
return history
# 全局记忆实例
memory = ConversationBufferMemory()
def get_memory() -> ConversationBufferMemory:
"""获取记忆实例"""
return memory
def sync_chat_history_to_memory():
"""
将 chat.py 中的 chat_history 同步到 memory
保持与现有聊天功能的兼容性
"""
# 延迟导入以避免循环导入问题
try:
from ..routers.chat import chat_history
except Exception:
return
for session_id, messages in chat_history.items():
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
if role == "user":
memory.add_message(session_id, "user", content)
else:
memory.add_message(session_id, "assistant", content)记忆系统的设计模式
plaintext
┌─────────────────────────────────────────────┐
│ 对话流程中的记忆系统 │
│ │
│ 用户消息 → 存储到记忆 → 召回相关记忆 → 加入上下文 │
│ ↓ │
│ AI 回复 → 存储到记忆（可选） │
└─────────────────────────────────────────────┘
记忆类型：
1. 短期记忆：当前对话中的内容
2. 长期记忆：用户偏好、历史信息
3. 工作记忆：正在处理任务的相关信息
2.8 智能代理服务 (agent.py)
学习目标
- 理解 Agent（智能代理）的概念
- 掌握工具注册表的设计
- 理解 Tool Calling 的原理
技术原理
Agent = 会使用工具的 AI
类比理解：
- 普通 AI：只能聊天，像一个只会说话的人 🗣️
- Agent：可以调用工具，像一个能做事的人 🧑🔧
- 可以搜索资料
- 可以执行计算
- 可以访问网页
- 可以操作文件
核心代码解析：
python
class Agent:
"""智能代理类"""
def __init__(self, config: Optional[AgentConfig] = None):
"""
初始化Agent
Args:
config: Agent配置
"""
self.config = config or AgentConfig()
self.tool_registry = get_tool_registry()
self.memory = get_memory() # 记忆服务
self._init_tools()
async def run(self, query: str, session_id: str = "default") -> Dict[str, Any]:
"""
运行Agent处理查询
Args:
query: 用户查询
session_id: 会话ID
Returns:
处理结果
处理用户消息
Agent 的思考流程：
1. 理解用户意图
2. 决定是否需要调用工具
3. 执行工具获取结果
4. 根据结果生成回答
"""
# 1️⃣ 保存用户消息到记忆
self.memory.add_message(session_id, "user", query)
# 构建初始提示词
system_prompt = self._build_system_prompt()
# 获取对话历史
history = self.memory.get_history(session_id, limit=10)
history_context = self._format_history(history)
full_prompt = f"{system_prompt}\n\n{history_context}\n\n用户问题: {query}"
# ReAct循环
thoughts = []
iteration = 0
final_answer = None
while iteration < self.config.max_iterations:
iteration += 1
# 调用LLM
llm_output = await self._call_llm(full_prompt)
if self.config.verbose:
print(f"[Iteration {iteration}] LLM Output:\n{llm_output}\n")
# 解析输出
parsed = self._parse_llm_output(llm_output)
if parsed.get("is_final"):
final_answer = parsed.get("final_answer", llm_output)
break
# 执行工具调用
if parsed.get("action"):
tool_name = parsed["action"]
tool_input = parsed.get("action_input", {})
# 记录思考过程
thought = AgentThought(
thought=parsed.get("thought", ""),
action=tool_name,
action_input=tool_input
)
# 调用工具
tool_result = self.tool_registry.call_tool(tool_name, **tool_input)
thought.observation = str(tool_result)
thoughts.append(thought)
# 将工具结果添加到上下文
full_prompt += f"\n\n{llm_output}\n观察结果: {tool_result}"
else:
# 没有解析到action，直接作为最终回答
final_answer = llm_output
break
if final_answer is None:
final_answer = "抱歉，我无法完成这个任务。"
# 添加助手回复到记忆
self.memory.add_message(session_id, "assistant", final_answer)
return {
"answer": final_answer,
"thoughts": [t.dict() for t in thoughts],
"iterations": iteration,
"used_tools": [t.action for t in thoughts if t.action]
}
工具注册表设计（tool_registry.py）
核心代码解析：
python
class ToolRegistry:
"""工具注册表：管理所有可用的工具"""
def __init__(self):
"""初始化工具注册表"""
self._tools: Dict[str, Callable] = {}
self._tool_schemas: Dict[str, dict] = {}
self._register_builtin_tools()
def _register_builtin_tools(self):
"""注册内置工具"""
# 时间工具
self.register(
name="get_current_time",
func=self._get_current_time,
description="获取当前时间",
parameters={
"type": "object",
"properties": {},
"required": []
}
)
# 计算器工具
self.register(
name="calculator",
func=self._calculator,
description="执行数学计算，支持加减乘除和常见数学函数",
parameters={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如 '2+3*4' 或 'sqrt(16)'"
}
},
"required": ["expression"]
}
)
# 搜索工具（模拟）
self.register(
name="search",
func=self._search,
description="搜索互联网信息",
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
)Agent 工作流程图
plaintext
用户: "北京今天天气怎么样？"
│
▼
┌─────────────────────────────────────┐
│ Agent 思考： │
│ "这个问题需要查询天气工具" │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 调用工具: get_weather │
│ 参数: {"location": "北京"} │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 工具执行，返回结果： │
│ "北京今天晴，25度" │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ Agent 将工具结果整合到回答中： │
│ "北京今天天气晴朗，气温25度..." │
└─────────────────────────────────────┘
面试考点
Q1: 什么是 Tool Calling？Agent 如何决定调用哪个工具？
回答要点：
- Tool Calling = 让 LLM 能够调用外部工具
- 实现方式：给 LLM 提供工具定义（名称、描述、参数）
- LLM 根据用户问题判断是否需要调用工具、调用哪个
- 常见应用：联网搜索、API 调用、代码执行
Q2: Agent 和 普通 Chatbot 的区别？
回答要点：
- Chatbot：纯生成，输入→输出
- Agent：可以执行动作，输入→思考→工具调用→整合→输出
- Agent 有更强的"执行力"，能真正完成任务
第三部分：前端核心模块
3.1 主应用结构 (App.jsx)
学习目标
- 理解 React 组件的基本结构
- 掌握状态管理（useState）
- 理解条件渲染和组件组合
技术原理
React = 构建用户界面的库
类比理解：
- HTML = 静态页面（像一张打印好的纸）
- React = 动态页面（像一个活的、会变化的页面）
jsx
import { useState } from 'react'
import ChatWindow from './components/ChatWindow'
import RAGPanel from './components/RAGPanel'
export default function App() {
// 状态管理：存储当前活动的标签页
const [activeTab, setActiveTab] = useState('chat')
return (
<div className="min-h-screen bg-gradient-to-br from-blue-50 to-indigo-50 p-4 md:p-8">
<div className="max-w-6xl mx-auto">
<div className="text-center mb-8">
<h1 className="text-3xl md:text-4xl font-bold text-gray-800 mb-2">
ChainMind
</h1>
<p className="text-gray-600">本地AI知识库问答系统</p>
</div>
<div className="bg-white rounded-xl shadow-lg overflow-hidden">
<div className="tabs">
<button
onClick={() => setActiveTab('chat')}
className={`tab ${activeTab === 'chat' ? 'active' : 'inactive'}`}
>
💬 聊天
</button>
<button
onClick={() => setActiveTab('rag')}
className={`tab ${activeTab === 'rag' ? 'active' : 'inactive'}`}
>
📚 RAG
</button>
</div>
<div className="p-4 md:p-6">
{activeTab === 'chat' && <ChatWindow />}
{activeTab === 'rag' && <RAGPanel />}
</div>
</div>
<div className="mt-6 text-center text-sm text-gray-500">
<p>Powered by FastAPI + Ollama + LangChain</p>
</div>
</div>
</div>
)
}状态管理详解
jsx
// useState：React 的状态钩子
const [activeTab, setActiveTab] = useState('chat')
// ↑ ↑ ↑
// 当前值 更新函数 初始值
// 点击按钮时更新状态
<button onClick={() => setActiveTab('rag')}>
切换到 RAG
</button>
// 条件渲染
{activeTab === 'chat' && <ChatWindow />}
// 等价于：
if (activeTab === 'chat') {
return <ChatWindow />
}
3.2 文件上传组件 (UploadPanel.jsx)
学习目标
- 掌握文件上传的实现方式
- 理解拖拽上传的原理
- 学会处理上传状态和错误
技术原理
文件上传的三种方式：
plaintext
1️⃣ 点击选择文件
<input type="file" onChange={handleFileChange} />
2️⃣ 拖拽上传
onDragOver → onDrop → 处理文件
3️⃣ 批量上传
<input type="file" multiple />
核心代码解析
jsx
export default function UploadPanel() {
const [file, setFile] = useState(null) // 当前选中的文件
const [uploading, setUploading] = useState(false) // 上传状态
const [dragOver, setDragOver] = useState(false) // 拖拽状态
// 📌 处理文件选择
const handleFileChange = (e) => {
const selectedFile = e.target.files[0]
if (selectedFile) {
setFile(selectedFile)
}
}
// 📌 处理拖拽悬停
const handleDragOver = (e) => {
e.preventDefault() // 阻止默认行为（打开文件）
setDragOver(true)
}
// 📌 处理文件放下
const handleDrop = (e) => {
e.preventDefault()
setDragOver(false)
const droppedFile = e.dataTransfer.files[0]
if (droppedFile) {
setFile(droppedFile)
}
}
// 📌 上传文件
const handleUpload = async () => {
if (!file) return
setUploading(true)
// 1️⃣ 创建 FormData
const formData = new FormData()
formData.append('file', file)
try {
// 2️⃣ 发送请求
const response = await fetch('/api/upload/file', {
method: 'POST',
body: formData // FormData 自动设置 Content-Type
})
const data = await response.json()
setResult({ success: true, data })
} catch (error) {
setResult({ success: false, error: error.message })
} finally {
setUploading(false)
}
}
return (
<div onDragOver={handleDragOver} onDrop={handleDrop}>
{/* 拖拽区域 */}
<div className="border-2 border-dashed">
拖拽文件到此处
</div>
{/* 上传按钮 */}
<button onClick={handleUpload} disabled={uploading}>
{uploading ? '上传中...' : '上传'}
</button>
</div>
)
}
拖拽上传原理解析
plaintext
浏览器默认行为：
拖拽文件到页面 → 打开文件（不想要这个效果）
我们需要：
拖拽文件到上传区 → 显示高亮 → 放开 → 上传文件
实现步骤：
1. onDragOver: e.preventDefault() → 阻止默认打开
2. onDragLeave: 移除高亮样式
3. onDrop: e.preventDefault() → 阻止默认打开，获取文件
第四部分：面试实战总结
核心知识点速记卡
RAG 相关
| 问题 | 核心答案 |
|---|---|
| RAG 是什么？ | Retrieval-Augmented Generation，检索增强生成 |
| RAG 解决什么问题？ | LLM 幻觉、知识过时、可溯源性 |
| RAG 的三个步骤？ | 检索 → 增强 → 生成 |
| 为什么要切分文档？ | LLM 上下文有限、检索粒度需要适中 |
FastAPI 相关
| 问题 | 核心答案 |
|---|---|
| FastAPI 优势？ | 自动文档、类型验证、异步支持 |
| 如何配置 CORS？ | add_middleware(CORSMiddleware, ...) |
| 什么是中间件？ | 请求/响应的"拦截器" |
| Pydantic 的作用？ | 数据验证、类型转换、自动文档 |
系统设计相关
| 问题 | 核心答案 |
|---|---|
| 为什么要用单例模式？ | 节省资源、保持状态一致 |
| 前端如何调用后端 API？ | fetch/axios 发送 HTTP 请求 |
| 为什么需要记忆系统？ | 支持多轮对话、保持上下文 |
高频面试题详解
题目 1：画出 RAG 的工作流程
参考答案：
plaintext
用户问题
│
▼
┌─────────────────────────────────────┐
│ 1. 向量化问题 │
│ text → [0.23, -0.45, ...] │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 2. 向量检索 │
│ 在向量库中找到 top_k 个相似文档 │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 3. 构建 Prompt │
│ 系统提示 + 检索结果 + 用户问题 │
└─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 4. LLM 生成答案 │
│ 基于上下文生成准确答案 │
└─────────────────────────────────────┘
│
▼
返回答案 + 引用来源
题目 2：FastAPI 如何实现文件上传？
参考答案：
python
from fastapi import FastAPI, UploadFile, File
app = FastAPI()
@router.post("/upload")
async def upload_file(file: UploadFile = File(...)):
# 1. 验证文件类型和大小
# 2. 读取文件内容
contents = await file.read()
# 3. 保存到磁盘
with open(f"./uploads/{file.filename}", "wb") as f:
f.write(contents)
# 4. 返回结果
return {"filename": file.filename, "size": len(contents)}
关键点：
UploadFilevsFile：UploadFile 提供更好的异步支持await file.read()：异步读取，避免阻塞- 一定要验证文件类型，防止恶意文件
题目 3：什么是向量检索？
参考答案：
向量检索是一种基于语义相似度的搜索技术。
核心思想：将文本转换为向量（数字数组），相似的文本在向量空间中距离更近。
工作原理：
- 使用 Embedding 模型将文档转为向量
- 用户查询也转为向量
- 计算查询向量与所有文档向量的"距离"
- 返回距离最近（最相似）的文档
优势：能理解语义，找到"意思相近"的内容，不依赖关键词匹配。
题目 4：如何优化 RAG 的效果？
参考答案：
| 优化方向 | 具体方法 |
|---|---|
| 检索质量 | 更好的 Embedding 模型、混合检索（向量+关键词） |
| 文档切分 | 调整 chunk_size、使用语义切分而非固定长度 |
| 上下文 | 增加元数据过滤、重排序（Rerank） |
| 提示词 | 优化 Prompt 设计、few-shot 示例 |
| 答案质量 | 选择更强的 LLM、调整 temperature |
展示效果：
延伸学习资源
推荐阅读
- 《Building RAG Applications》 - RAG 系统构建指南
- FastAPI 官方文档 - fastapi.tiangolo.com
- LangChain 官方文档 - python.langchain.com
总结
- ✅ RAG 系统的核心原理和实现
- ✅ FastAPI 后端开发的最佳实践
- ✅ 前后端分离架构的设计思路
- ✅ 单例模式、工具注册表等设计模式
- ✅ 文件上传、异步处理等实战技能
这些知识不仅能帮助你理解 AI 应用开发的基础，也为你后续学习更高级的 Agent 系统打下了坚实的基础。
项目地址（需要把里面的本地模型换为你自己的模型API）https://github.com/AiW520/ChainMind-
https://github.com/AiW520/ChainMind-.git
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
11
0- 0
已为社区贡献4条内容
所有评论(0)