Python 实现 AI 大模型智能对话系统
使用 Python 和 FastAPI 构建智能对话系统的完整流程。涵盖环境配置、项目结构、数据库设计、OpenAI API 集成、Vue 前端界面及 Docker 部署。教程包含多轮对话管理、历史记录存储及流式响应实现,适合希望掌握 AI 应用全栈开发的开发者参考。

使用 Python 和 FastAPI 构建智能对话系统的完整流程。涵盖环境配置、项目结构、数据库设计、OpenAI API 集成、Vue 前端界面及 Docker 部署。教程包含多轮对话管理、历史记录存储及流式响应实现,适合希望掌握 AI 应用全栈开发的开发者参考。

第一部分:技术背景与项目概述
近年来,人工智能大模型技术取得了突破性进展。从 GPT 系列到国内的文心一言、通义千问等,大模型正在改变我们与计算机交互的方式。Python 作为 AI 开发的首选语言,拥有丰富的生态系统和完善的工具链,是开发 AI 应用的最佳选择。
在本教程中,我们将使用 Python 开发一个完整的智能对话 Web 应用,具备以下功能:
第二部分:环境准备与基础配置
首先确保你的系统已安装 Python 3.8 或更高版本:
# 检查 Python 版本
python --version
# 或 python3 --version
使用虚拟环境隔离项目依赖:
# 创建项目目录
mkdir ai-chat-app
cd ai-chat-app
# 创建虚拟环境
python -m venv venv
# 激活虚拟环境
# Windows: venv\Scripts\activate
# Linux/Mac: source venv/bin/activate
创建 requirements.txt 文件:
fastapi==0.104.1
uvicorn==0.24.0
openai==1.3.0
sqlalchemy==2.0.23
alembic==1.12.1
python-multipart==0.0.6
python-dotenv==1.0.0
安装依赖:
pip install -r requirements.txt
第三部分:核心功能开发实战
让我们先规划项目结构:
ai-chat-app/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 应用入口
│ ├── models.py # 数据模型
│ ├── database.py # 数据库配置
│ ├── api/
│ │ ├── __init__.py
│ │ └── chat.py # 聊天 API
│ └── services/
│ ├── __init__.py
│ └── openai_service.py # OpenAI 服务
├── static/ # 静态文件
├── templates/ # 模板文件
├── requirements.txt
├── .env # 环境变量
└── README.md
创建 app/main.py 文件:
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.middleware.cors import CORSMiddleware
import os
# 创建 FastAPI 应用实例
app = FastAPI(
title="AI 智能对话系统",
description="基于 Python 和 AI 大模型的智能对话应用",
version="1.0.0"
)
# 配置 CORS 中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 配置静态文件和模板
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
@app.get("/")
async def root():
return {"message": "AI 智能对话系统 API 服务已启动"}
@app.get("/health")
async def health_check():
return {"status": "healthy"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
创建 app/models.py 文件:
from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class Conversation(Base):
__tablename__ = "conversations"
id = Column(Integer, primary_key=True, index=True)
session_id = Column(String(100), index=True, nullable=False)
user_message = Column(Text, nullable=False)
ai_response = Column(Text, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
model_used = Column(String(50), default="gpt-3.5-turbo")
class UserSettings(Base):
__tablename__ = "user_settings"
id = Column(Integer, primary_key=True, index=True)
session_id = Column(String(100), unique=True, index=True)
temperature = Column(Integer, default=7) # 0-10, 默认 7
max_tokens = Column(Integer, default=1000)
model_preference = Column(String(50), default="gpt-3.5-turbo")
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
创建 app/database.py 文件:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.models import Base
import os
# 数据库配置
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./chat_app.db")
# 创建数据库引擎
engine = create_engine(
DATABASE_URL,
connect_args={"check_same_thread": False} # SQLite 专用参数
)
# 创建 SessionLocal 类
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 创建数据库表
def create_tables():
Base.metadata.create_all(bind=engine)
# 数据库依赖
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
创建 app/services/openai_service.py 文件:
import openai
import os
from typing import List, Dict, Optional
from openai import OpenAI
class OpenAIService:
def __init__(self):
# 从环境变量获取 API 密钥
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise ValueError("OPENAI_API_KEY 环境变量未设置")
self.client = OpenAI(api_key=api_key)
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "gpt-3.5-turbo",
temperature: float = 0.7,
max_tokens: int = 1000
) -> str:
""" 调用 OpenAI 聊天补全 API """
try:
response = self.client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
stream=False
)
return response.choices[0].message.content
except Exception as e:
raise Exception(f"OpenAI API 调用失败:{(e)}")
():
:
response = .client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
stream=
)
chunk response:
chunk.choices[].delta.content :
chunk.choices[].delta.content
Exception e:
openai_service = OpenAIService()
创建 app/api/chat.py 文件:
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List, Dict
import uuid
from datetime import datetime
from app.database import get_db
from app.models import Conversation, UserSettings
from app.services.openai_service import openai_service
router = APIRouter(prefix="/api/chat", tags=["chat"])
@app.post("/message")
async def send_message(
message: str,
session_id: str = None,
db: Session = Depends(get_db)
):
""" 发送消息并获取 AI 回复 """
# 生成或使用现有的 session_id
if not session_id:
session_id = str(uuid.uuid4())
# 获取用户设置
user_settings = db.query(UserSettings).filter(
UserSettings.session_id == session_id
).first()
if not user_settings:
# 创建默认用户设置
user_settings = UserSettings(
session_id=session_id,
temperature=7,
max_tokens=1000,
model_preference="gpt-3.5-turbo"
)
db.add(user_settings)
db.commit()
db.refresh(user_settings)
# 构建对话历史
conversation_history = get_conversation_history(session_id, db)
messages = conversation_history + [{: , : message}]
:
response = openai_service.chat_completion(
messages=messages,
model=user_settings.model_preference,
temperature=user_settings.temperature / ,
max_tokens=user_settings.max_tokens
)
conversation = Conversation(
session_id=session_id,
user_message=message,
ai_response=response,
model_used=user_settings.model_preference
)
db.add(conversation)
db.commit()
{
: session_id,
: message,
: response,
: datetime.utcnow().isoformat()
}
Exception e:
HTTPException(status_code=, detail=(e))
():
conversations = db.query(Conversation).(
Conversation.session_id == session_id
).order_by(Conversation.created_at.desc()).limit(limit).()
[
{
: conv.,
: conv.user_message,
: conv.ai_response,
: conv.created_at.isoformat(),
: conv.model_used
}
conv (conversations)
]
() -> [[, ]]:
conversations = db.query(Conversation).(
Conversation.session_id == session_id
).order_by(Conversation.created_at.asc()).limit().()
messages = []
conv conversations:
messages.append({: , : conv.user_message})
messages.append({: , : conv.ai_response})
messages
创建 templates/index.html 文件:
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>AI 智能对话系统</title>
<script src="https://unpkg.com/vue@3/dist/vue.global.js"></script>
<link rel="stylesheet" href="https://unpkg.com/element-plus/dist/index.css">
<script src="https://unpkg.com/element-plus"></script>
<style>
body {
margin: 0;
padding: 0;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
min-height: ;
}
{
: ;
: auto;
: white;
: ;
: (,,,);
: hidden;
: ;
: flex;
: column;
}
{
: ;
: white;
: ;
: center;
}
{
: ;
: ;
: auto;
: ;
}
{
: ;
: flex;
}
{
: flex-end;
}
{
: flex-start;
}
{
: ;
: ;
: ;
: break-word;
}
{
: ;
: white;
}
{
: white;
: solid ;
}
{
: ;
: solid ;
: white;
}
{
: flex;
: center;
: ;
: italic;
}
{
: ;
: ;
: ;
: ;
: ;
: bounce infinite ease-in-out;
}
() { : -; }
() { : -; }
bounce {
, , { : (); }
{ : (); }
}
🤖 AI 智能对话系统
基于 Python 和 AI 大模型的智能对话应用
你: {{ message.content }}
AI 助手:
{{ message.content }}
发送
更新 app/main.py 文件,添加路由和模板支持:
from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.middleware.cors import CORSMiddleware
from app.database import create_tables
from app.api import chat
import os
# 创建数据库表
create_tables()
# 创建 FastAPI 应用实例
app = FastAPI(
title="AI 智能对话系统",
description="基于 Python 和 AI 大模型的智能对话应用",
version="1.0.0"
)
# 配置 CORS 中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 配置静态文件和模板
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
# 注册路由
app.include_router(chat.router)
@app.get("/")
async def root(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
@app.get("/health")
async def health_check():
return {"status": "healthy"}
__name__ == :
uvicorn
uvicorn.run(app, host=, port=)
第四部分:高级功能扩展
创建 .env 文件:
# OpenAI API 配置
OPENAI_API_KEY=your_openai_api_key_here
# 数据库配置
DATABASE_URL=sqlite:///./chat_app.db
# 应用配置
DEBUG=True
HOST=0.0.0.0
PORT=8000
创建 config.py 文件:
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./chat_app.db")
DEBUG = os.getenv("DEBUG", "False").lower() == "true"
HOST = os.getenv("HOST", "0.0.0.0")
PORT = int(os.getenv("PORT", 8000))
更新 app/api/chat.py,添加流式响应支持:
from fastapi import Response
from fastapi.responses import StreamingResponse
import json
@router.post("/message/stream")
async def stream_message(
message: str,
session_id: str = None,
db: Session = Depends(get_db)
):
""" 流式消息响应(实时打字效果) """
if not session_id:
session_id = str(uuid.uuid4())
# 获取用户设置(代码同上)
# ...
async def generate():
try:
full_response = ""
async for chunk in openai_service.stream_chat_completion(
messages=messages,
model=user_settings.model_preference,
temperature=user_settings.temperature / 10.0,
max_tokens=user_settings.max_tokens
):
full_response += chunk
yield f"data: {json.dumps({'content': chunk})}\n\n"
# 保存完整对话记录
conversation = Conversation(
session_id=session_id,
user_message=message,
ai_response=full_response,
model_used=user_settings.model_preference
)
db.add(conversation)
db.commit()
except Exception as e:
yield f"data: {json.dumps({'error': (e)})}\n\n"
StreamingResponse(
generate(),
media_type=,
headers={: }
)
第五部分:部署与优化
创建 Dockerfile:
FROM python:3.9-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
gcc \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件
COPY requirements.txt .
# 安装 Python 依赖
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 创建非 root 用户
RUN useradd -m -u 1000 user
USER user
# 暴露端口
EXPOSE 8000
# 启动命令
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
创建 docker-compose.yml:
version: '3.8'
services:
ai-chat-app:
build: .
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- DATABASE_URL=sqlite:///./chat_app.db
volumes:
- ./data:/app/data
restart: unless-stopped
创建 gunicorn_conf.py:
import multiprocessing
# 服务器绑定
bind = "0.0.0.0:8000"
# 工作进程数
workers = multiprocessing.cpu_count() * 2 + 1
# 工作模式
worker_class = "uvicorn.workers.UvicornWorker"
# 日志配置
accesslog = "-"
errorlog = "-"
loglevel = "info"
# 超时设置
timeout = 120
keepalive = 5
创建 start.sh:
#!/bin/bash
# 激活虚拟环境
source venv/bin/activate
# 启动应用
if [ "$ENVIRONMENT" = "production" ]; then
gunicorn -c gunicorn_conf.py app.main:app
else
uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
fi
完整项目运行指南
# 克隆项目(如果有 Git 仓库)
git clone <repository-url>
cd ai-chat-app
# 创建虚拟环境
python -m venv venv
source venv/bin/activate
# Windows: venv\Scripts\activate
# 安装依赖
pip install -r requirements.txt
# 复制环境变量模板
cp .env.example .env
# 编辑.env 文件,设置你的 OpenAI API 密钥
OPENAI_API_KEY=sk-your-actual-api-key-here
# 开发模式
python -m uvicorn app.main:app --reload
# 或使用启动脚本
chmod +x start.sh
./start.sh
打开浏览器访问:http://localhost:8000/
总结
通过本教程,我们完整地实现了一个基于 Python 和 AI 大模型的智能对话系统。这个项目展示了:
这个项目不仅是一个可用的智能对话应用,更是一个优秀的学习范例,可以帮助开发者掌握 Python AI 应用开发的核心技能。
引用文献

微信公众号「极客日志」,在微信中扫描左侧二维码关注。展示文案:极客日志 zeeklog
使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online