Python 实现 AI 大模型智能对话系统

3.2 创建 FastAPI 应用

创建 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)

3.3 数据库模型设计

创建 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)

3.4 数据库配置

创建 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()

3.5 OpenAI 服务封装

创建 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 调用失败：{str(e)}")

    async def stream_chat_completion(
        self,
        messages: List[Dict[str, str]],
        model: str = "gpt-3.5-turbo",
        temperature: float = 0.7,
        max_tokens: int = 1000
    ):
        """ 流式聊天补全（用于实时打字效果） """
        try:
            response = self.client.chat.completions.create(
                model=model,
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens,
                stream=True
            )
            for chunk in response:
                if chunk.choices[0].delta.content is not None:
                    yield chunk.choices[0].delta.content
        except Exception as e:
            yield f"错误：{str(e)}"

# 创建全局实例
openai_service = OpenAIService()

3.6 聊天 API 实现

创建 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 = await get_conversation_history(session_id, db)

    # 添加用户新消息
    messages = conversation_history + [{"role": "user", "content": message}]

    try:
        # 调用 OpenAI 服务
        response = await openai_service.chat_completion(
            messages=messages,
            model=user_settings.model_preference,
            temperature=user_settings.temperature / 10.0, # 转换为 0-1 范围
            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()

        return {
            "session_id": session_id,
            "user_message": message,
            "ai_response": response,
            "timestamp": datetime.utcnow().isoformat()
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/history/{session_id}")
async def get_chat_history(
    session_id: str,
    limit: int = 50,
    db: Session = Depends(get_db)
):
    """ 获取指定会话的聊天历史 """
    conversations = db.query(Conversation).filter(
        Conversation.session_id == session_id
    ).order_by(Conversation.created_at.desc()).limit(limit).all()

    return [
        {
            "id": conv.id,
            "user_message": conv.user_message,
            "ai_response": conv.ai_response,
            "created_at": conv.created_at.isoformat(),
            "model_used": conv.model_used
        }
        for conv in reversed(conversations) # 按时间正序返回
    ]

async def get_conversation_history(session_id: str, db: Session) -> List[Dict[str, str]]:
    """ 获取格式化后的对话历史 """
    conversations = db.query(Conversation).filter(
        Conversation.session_id == session_id
    ).order_by(Conversation.created_at.asc()).limit(10).all()

    # 最近 10 条对话
    messages = []
    for conv in conversations:
        messages.append({"role": "user", "content": conv.user_message})
        messages.append({"role": "assistant", "content": conv.ai_response})
    return messages

3.7 前端界面实现

创建 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: 100vh;
        }
        .chat-container {
            max-width: 800px;
            margin: 0 auto;
            background: white;
            border-radius: 10px;
            box-shadow: 0 10px 30px rgba(0,0,0,0.1);
            overflow: hidden;
            height: 90vh;
            display: flex;
            flex-direction: column;
        }
        .chat-header {
            background: #2c3e50;
            color: white;
            padding: 20px;
            text-align: center;
        }
        .chat-messages {
            flex: 1;
            padding: 20px;
            overflow-y: auto;
            background: #f8f9fa;
        }
        .message {
            margin-bottom: 15px;
            display: flex;
        }
        .user-message {
            justify-content: flex-end;
        }
        .ai-message {
            justify-content: flex-start;
        }
        .message-bubble {
            max-width: 70%;
            padding: 12px 16px;
            border-radius: 18px;
            word-wrap: break-word;
        }
        .user-bubble {
            background: #007bff;
            color: white;
        }
        .ai-bubble {
            background: white;
            border: 1px solid #e9ecef;
        }
        .chat-input {
            padding: 20px;
            border-top: 1px solid #e9ecef;
            background: white;
        }
        .typing-indicator {
            display: flex;
            align-items: center;
            color: #6c757d;
            font-style: italic;
        }
        .dot {
            width: 8px;
            height: 8px;
            border-radius: 50%;
            background: #6c757d;
            margin: 0 2px;
            animation: bounce 1.4s infinite ease-in-out;
        }
        .dot:nth-child(1) { animation-delay: -0.32s; }
        .dot:nth-child(2) { animation-delay: -0.16s; }
        @keyframes bounce {
            0%, 80%, 100% { transform: scale(0); }
            40% { transform: scale(1); }
        }
    </style>
</head>
<body>
    <div id="app">
        <div class="chat-container">
            <div class="chat-header">
                <h2>🤖 AI 智能对话系统</h2>
                <p>基于 Python 和 AI 大模型的智能对话应用</p>
            </div>
            <div class="chat-messages" ref="messagesContainer">
                <div v-for="message in messages" :key="message.id" 
                     :class="['message', message.role === 'user' ? 'user-message' : 'ai-message']">
                    <div :class="['message-bubble', message.role === 'user' ? 'user-bubble' : 'ai-bubble']">
                        <div v-if="message.role === 'user'">
                            <strong>你:</strong> {{ message.content }}
                        </div>
                        <div v-else>
                            <strong>AI 助手:</strong>
                            <span v-if="message.isTyping">
                                <span></span><span></span><span></span>
                            </span>
                            <span v-else>{{ message.content }}</span>
                        </div>
                    </div>
                </div>
            </div>
            <div class="chat-input">
                <el-input v-model="inputMessage" 
                          placeholder="输入你的问题..." 
                          @keyup.enter="sendMessage" 
                          :disabled="isLoading">
                    <template #append>
                        <el-button @click="sendMessage" :loading="isLoading" type="primary">
                            发送
                        </el-button>
                    </template>
                </el-input>
            </div>
        </div>
    </div>
    <script>
        const { createApp, ref, onMounted, nextTick } = Vue;
        const { ElMessage } = ElementPlus;

        createApp({
            setup() {
                const messages = ref([]);
                const inputMessage = ref('');
                const isLoading = ref(false);
                const sessionId = ref('');
                const messagesContainer = ref(null);

                // 初始化会话 ID
                if (!sessionId.value) {
                    sessionId.value = generateSessionId();
                    loadChatHistory();
                }

                function generateSessionId() {
                    return 'session_' + Date.now() + '_' + Math.random().toString(36).substr(2, 9);
                }

                async function loadChatHistory() {
                    try {
                        const response = await fetch();
                         (response.) {
                             history =  response.();
                            messages. = history.( ({
                                : item.,
                                : ,
                                : item.,
                                : item.
                            })).(history.( ({
                                : item. + ,
                                : ,
                                : item.,
                                : item.
                            })).(  (a.) -  (b.));
                            ();
                        }
                    }  (error) {
                        .(, error);
                    }
                }

                  () {
                     (!inputMessage..() || isLoading.) ;
                     userMessage = inputMessage..();
                    inputMessage. = ;

                    
                     userMsg = {
                        : .(),
                        : ,
                        : userMessage,
                        :  ().()
                    };
                    messages..(userMsg);

                    
                     aiMsg = {
                        : .() + ,
                        : ,
                        : ,
                        : ,
                        :  ().()
                    };
                    messages..(aiMsg);
                    ();
                    isLoading. = ;

                     {
                         response =  (, {
                            : ,
                            : { : , },
                            : .({ : userMessage, : sessionId. })
                        });
                         (response.) {
                             data =  response.();
                            aiMsg. = data.;
                            aiMsg. = ;
                        }  {
                              ();
                        }
                    }  (error) {
                        aiMsg. =  + error.;
                        aiMsg. = ;
                        .();
                    }  {
                        isLoading. = ;
                        ();
                    }
                }

                 () {
                    ( {
                         (messagesContainer.) {
                            messagesContainer.. = messagesContainer..;
                        }
                    });
                }

                ( {
                    ();
                });

                 { messages, inputMessage, isLoading, messagesContainer, sendMessage };
            }
        }).().();
    </script>
</body>
</html>

3.8 更新主应用文件

更新 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"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

第四部分：高级功能扩展

4.1 环境变量配置

创建 .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))

4.2 流式响应实现

更新 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': str(e)})}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"}
    )

第五部分：部署与优化

5.1 Docker 容器化部署

创建 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

5.2 生产环境配置

创建 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

5.3 启动脚本

创建 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

完整项目运行指南

1. 环境准备

# 克隆项目（如果有 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

2. 配置环境变量

# 复制环境变量模板
cp .env.example .env

# 编辑.env 文件，设置你的 OpenAI API 密钥
OPENAI_API_KEY=sk-your-actual-api-key-here

3. 启动应用

# 开发模式
python -m uvicorn app.main:app --reload

# 或使用启动脚本
chmod +x start.sh
./start.sh

4. 访问应用

打开浏览器访问：http://localhost:8000/

总结

通过本教程，我们完整地实现了一个基于 Python 和 AI 大模型的智能对话系统。这个项目展示了：

1. 现代化 Python 开发：使用 FastAPI、SQLAlchemy 等现代框架
2. AI 集成能力：与 OpenAI API 的无缝集成
3. 完整功能实现：对话管理、历史记录、用户设置等
4. 生产就绪：Docker 部署、性能优化、安全配置

这个项目不仅是一个可用的智能对话应用，更是一个优秀的学习范例，可以帮助开发者掌握 Python AI 应用开发的核心技能。

引用文献

1. FastAPI 官方文档：https://fastapi.tiangolo.com/
2. OpenAI API 文档：https://platform.openai.com/docs
3. SQLAlchemy 文档：https://docs.sqlalchemy.org/
4. Vue.js 官方文档：https://vuejs.org/
5. Docker 官方文档：https://docs.docker.com/