实现第一个GPT聊天机器人

章节介绍

本章围绕基于OpenAI API实现GPT聊天机器人展开，从基础聊天接口开发入手，逐步完成接口调试、数据校验、统一响应格式、数据库集成、单元测试等企业级开发环节，最终实现可落地的GPT聊天机器人后端服务，同时约定项目开发与测试规范。

一、基础准备：配置分离与接口请求校验

1.1 .env环境变量配置分离

实现工具：使用python-dotenv库将.env文件中的配置加载为Python环境变量，实现代码与配置分离，便于后期维护。
配置获取：通过os.getenv()方法获取环境变量中的配置信息（如OpenAI API秘钥）。
安装：需提前安装依赖pip install python-dotenv。

1.2 接口请求校验的必要性

安全性：避免SQL注入等安全漏洞，确保输入数据符合预期；
数据完整性：校验数据格式、类型、长度等，防止不合法数据导致系统出错；
用户体验：对不合法请求给出友好提示，引导用户规范输入。

1.3 Flask-wtf实现请求数据校验

工具介绍：Flask-wtf是基于wtforms封装的Flask插件，支持CSRF保护、快速提取数据、自定义验证规则，是Flask项目主流的数据校验工具。
wtforms核心能力
- 支持提取的数据类型：布尔值、日期、文本、时间、浮点数、邮箱、文件、整数、自定义类型等；
- 支持的校验规则：必填、长度、邮箱、正则、范围、UUID、URL、数据可选等。
安装：pip install flask-wtf
课程小结：完成Flask项目配置加载，通过Flask-wtf+wtforms实现请求数据统一校验，从源头保证数据安全。

二、接口规范化：统一响应与异常错误设计

2.1 统一响应接口设计

2.1.1 设计原因

建立前后端统一的接口约定，避免响应格式混乱（文本、JSON、异常页面混用），减少前端数据处理的复杂度，提升前后端交互效率。

2.1.2 核心设计原则

所有接口HTTP状态码统一返回200，通过业务状态码判断操作是否成功；
错误信息通过message和data字段携带，避免直接抛出异常页面。

2.1.3 基础统一响应格式

json

{
  "code": "success", // 业务状态码
  "message": "获取AI应用数据成功", // 业务消息提示
  "data": {} // 业务数据体
}

约定业务状态码：success、fail、not_found、unauthorized、forbidden、validate_error。

2.1.4 分页数据响应格式

针对列表型分页数据，data字段包含paginator（分页信息）和list（数据列表）：

json

{
  "code": "success",
  "message": "",
  "data": {
    "paginator": {
      "page_size": 10, // 每页条数
      "current_page": 1, // 当前页数
      "total_page": 10, // 总页数
      "total_record": 100 // 总记录数
    },
    "list": [] // 分页数据列表
  }
}

2.2 异常错误状态统一设计

统一项目错误异常分类，为项目添加日志功能，将错误/日志信息统一存储到本地文件，便于问题定位和排查（文档未展开具体实现，为后续实操内容）。

三、单元测试：PyTest配置与API测试用例编写

3.1 Pytest简介与安装

工具定位：Python主流的测试框架，支持单元测试、集成测试、功能测试，相比Python内置的unittest框架更简洁、灵活。
核心优势
- 语法简单，学习曲线低，支持函数式/类式/模块级测试；
- 插件生态丰富，可扩展功能满足各类测试场景；
- 与Jenkins、Coverage.py等工具无缝集成；
- 测试输出信息丰富，便于定位问题。
官网：https://docs.pytest.org/en/8.0.x/
安装：pip install pytest

3.2 Pytest测试用例编写规则

测试文件以test_开头或_test结尾；
测试类以Test开头，且无__init__方法；
测试函数/方法以test_开头；
断言直接使用Python原生assert语句。

3.3 Pytest常用运行命令

bash

# 运行指定文件夹下所有符合规则的用例
pytest folder_name
# 运行指定测试文件
pytest test_file.py
# 运行指定文件中的指定函数
pytest test_file.py::test_func
# 运行指定文件中测试类的指定方法
pytest test_file.py::TestClass::test_method
# 显示测试中的print输出
pytest -s
# 显示详细测试信息（用例名称、结果）
pytest -v
# 组合使用
pytest -vs test_file.py

3.4 fixture：测试环境预设

功能：为测试函数提供预设数据/测试环境的机制，避免重复代码。
存放文件：通常写在conftest.py中，当前目录及子目录均可使用，嵌套目录取最接近的conftest.py。
使用方式：定义好的fixture可作为测试函数的参数，Pytest自动注入。
示例：配置Flask测试客户端

python

# test/conftest.py
import pytest
from ..app.server import app

@pytest.fixture
def client():
    app.config["TESTING"] = True
    with app.test_client() as client:
        yield client

# 测试用例中使用
def test_completion(client):
    r = client.post("/app/completion", json={"query": "你好！有什么可以帮助你的吗？"})
    assert r.status_code == 200
    assert r.json.get("code") == "success"

3.5 参数化测试

实现装饰器：@pytest.mark.parametrize，支持多参数值多次运行同一测试用例，覆盖多种测试场景。
参数格式
- 单个参数：@pytest.mark.parametrize("param", [值1, 值2])；
- 多个参数：@pytest.mark.parametrize("p1,p2", [(值1,值2), (值3,值4)])。
示例：测试聊天接口的参数校验

python

import pytest
from ..pkg.response import HttpCode

class TestAppHandler:
    @pytest.mark.parametrize("query", [None, "你好，你是？"])
    def test_completion(self, query, client):
        r = client.post("/app/completion", json={"query": query})
        assert r.status_code == 200
        if query is None:
            assert r.json.get("code") == HttpCode.VALIDATE_ERROR
        else:
            assert r.json.get("code") == HttpCode.SUCCESS

3.6 pytest.ini：运行时配置

功能：Pytest的基础配置文件，统一设置测试规则，避免重复输入命令行参数。
核心配置项示例

ini

[pytest]
# 测试缓存文件存储路径
cache_dir = tmp/.pytest_cache
# 默认命令行选项
addopts = -v -s
# 指定测试文件匹配规则
python_files = test_*.py *_test.py
# 指定测试类匹配规则
python_classes = Test*
# 指定测试函数匹配规则
python_functions = test_*

四、数据库集成：Postgres与Flask-SQLAlchemy

4.1 ORM概念介绍

ORM（Object-Relational Mapping，对象关系映射）：将数据库中的表映射为Python的类，表的行映射为类的实例，表的列映射为类的属性，通过操作Python类实现数据库增删改查，无需直接编写SQL。

ORM核心优缺点

优点	缺点
有语法提示，保证SQL语法正确性	语法转换带来性能损耗，比原生SQL慢
提供方言功能，支持多数据库迁移，降低学习成本	复杂查询的ORM语法繁琐，需结合原生SQL
面向对象，可读性强，开发效率高	-
防止SQL注入攻击，提升安全性	-
搭配数据库迁移工具，更新表结构方便	-

4.2 Flask-SQLAlchemy安装与配置

工具介绍：Flask的SQLAlchemy扩展，简化Flask项目中ORM的使用，SQLAlchemy是Python最强大的ORM框架。
官方文档
- Flask-SQLAlchemy：https://flask-sqlalchemy.palletsprojects.com/en/3.1.x/
- SQLAlchemy：https://www.sqlalchemy.org/

安装

bash

# 基础安装
pip install flask-sqlalchemy
# Postgres数据库驱动（必装）
pip install psycopg2
# 一键安装
pip install flask-sqlalchemy psycopg2

核心配置项

配置项	说明	示例
SQLALCHEMY_DATABASE_URI	数据库连接地址	postgresql://root:123456@127.0.0.1:5432/test31
SQLALCHEMY_BINDS	多数据库连接配置
SQLALCHEMY_ECHO	是否打印底层SQL语句	True/False
SQLALCHEMY_TRACK_MODIFICATIONS	是否追踪数据库变化（消耗内存）	False
SQLALCHEMY_ENGINE_OPTIONS	SQLAlchemy本体配置

初始化：db.init_app(flask_app)，将配置加载到Flask应用中。

4.3 ORM模型创建

4.3.1 核心规范

模型类继承db.Model；
通过__tablename__指定数据库表名（单数）；
通过__table_args__设置主键、索引、唯一约束等；
字段类型从sqlalchemy导入，Postgres特有类型（如JSONB）从sqlalchemy.dialects.postgresql导入；
所有字段默认nullable=False，String类型默认String(255)且默认值""，UUID类型默认uuid.uuid4()；
所有模型包含updated_at和created_at字段（DateTime类型），updated_at带onupdate=datetime.now。

4.3.2 常用字段类型

python

from sqlalchemy import (
    Column, UUID, String, Text, DateTime,
    Integer, Decimal, PrimaryKeyConstraint,
    Index, UniqueConstraint
)
# Postgres特有
from sqlalchemy.dialects.postgresql import JSONB

4.3.3 模型创建示例（App模型）

python

import uuid
from datetime import datetime
from sqlalchemy import (
    Column, UUID, String, DateTime, PrimaryKeyConstraint, Index
)
from sqlalchemy.dialects.postgresql import JSONB
from ..internal.extension.database_extension import db

class App(db.Model):
    """AI应用模型"""
    __tablename__ = "app"
    __table_args__ = (
        PrimaryKeyConstraint("id", name="pk_app_id"),
        Index("idx_app_account_id", "account_id"),
    )
    id = Column(UUID, default=uuid.uuid4, nullable=False)
    account_id = Column(UUID, nullable=False)
    name = Column(String(255), default="", nullable=False)
    icon = Column(String(255), default="", nullable=False)
    config = Column(JSONB, default={}, nullable=False)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False)
    created_at = Column(DateTime, default=datetime.now, nullable=False)

4.3.4 表创建

通过db.create_all()将模型映射为数据库表，需在Flask应用上下文中执行：

python

from flask import Flask
from ..internal.model import App
app = Flask(__name__)
with app.app_context():
    db.create_all()

4.3.5 ChatGPT辅助生成ORM模型

可使用预设Prompt让ChatGPT快速生成符合规范的ORM模型，Prompt包含角色定义、技能要求、格式规范、示例（详见文档，核心为约定上述模型创建规范）。

4.4 数据库增删改查（CRUD）

4.4.1 操作方式约定

单表操作：使用ORM模型自身的query属性（如App.query.xxx），简洁高效；
多表操作：使用db.session.query(Model).xxx，支持复杂联表查询。

4.4.2 新增数据

通过db.session.add(model)添加实例，db.session.commit()提交事务：

python

def add_app():
    app = App(account_id=uuid.uuid4(), name="GPT聊天机器人", icon="icon.png")
    db.session.add(app)
    db.session.commit()
    return {"code": "success", "message": "创建成功"}

4.4.3 查询数据

Flask-SQLAlchemy提供丰富的查询方法，核心示例：

python

# 获取所有数据
all_apps = App.query.all()
# 获取第一条数据
first_app = App.query.first()
# 通过ID查询
app = App.query.get(uuid.UUID("xxx-xxx-xxx"))
# 条件筛选（filter_by，简单条件）
filter_apps = App.query.filter_by(name="GPT聊天机器人").all()
# 条件筛选（filter，复杂条件）
from sqlalchemy import or_
filter_apps = App.query.filter(or_(App.name=="GPT", App.account_id==uuid.uuid4())).all()
# 唯一结果查询（无结果返回None）
app = App.query.filter_by(name="GPT聊天机器人").one_or_none()

4.4.4 修改数据

两种方式，均需commit()提交（批量更新可直接调用update）：

python

# 方式1：批量更新
App.query.filter_by(name="GPT聊天机器人").update({"name": "GPT聊天助手"})
db.session.commit()

# 方式2：修改模型实例属性
app = App.query.first()
app.name = "GPT聊天助手"
db.session.commit()

4.4.5 删除数据

两种方式，均需commit()提交：

python

# 方式1：批量删除
App.query.filter_by(name="GPT聊天助手").delete()
db.session.commit()

# 方式2：删除模型实例
app = App.query.first()
db.session.delete(app)
db.session.commit()

4.5 重写SQLAlchemy实现自动提交

4.5.1 问题背景

原生操作中，新增/修改/删除需手动调用db.session.commit()，异常时需手动rollback()，代码冗余。

4.5.2 实现方案

重写SQLAlchemy核心类，通过@contextlib.contextmanager装饰器实现自动提交/回滚：

python

from contextlib import contextmanager
from flask_sqlalchemy import SQLAlchemy as _SQLAlchemy

class SQLAlchemy(_SQLAlchemy):
    """重写SQLAlchemy，实现自动提交"""
    @contextmanager
    def auto_commit(self):
        try:
            yield
            self.session.commit()
        except Exception as e:
            self.session.rollback()
            raise e

# 初始化db
db = SQLAlchemy()

4.5.3 使用方式

通过with db.auto_commit()上下文管理器包裹数据库操作，无需手动commit/rollback：

python

with db.auto_commit():
    app = App.query.first()
    app.name = "新名称"
# 异常时自动回滚，无需额外代码

4.6 Flask-Migrate：数据库迁移

4.6.1 迁移必要性

db.create_all()仅能创建新表，无法更新现有表的结构（新增/修改字段），直接删除表重建会丢失数据，Flask-Migrate可在不破坏数据的前提下更新表结构。

4.6.2 安装与初始化

安装：pip install flask-migrate

初始化

python

from flask_migrate import Migrate
migrate = Migrate()
# 关联Flask应用和db，指定迁移文件目录
migrate.init_app(app, db, directory="internal/migration")

4.6.3 核心迁移命令

LLMOps项目专用命令格式：flask --app app.server.app db [命令]，普通Flask项目可简化为flask db [命令]。

初始化迁移环境（首次使用）
bash
```
flask --app app.server.app db init
```
会在指定目录生成迁移环境文件。
生成迁移脚本
bash
```
# -m 为迁移添加注释（可选）
flask --app app.server.app db migrate -m "create app table"
```
生成的脚本包含upgrade()（应用改动）和downgrade()（撤销改动）函数，生产环境需手动检查脚本正确性。
应用迁移（更新表结构）
bash
```
flask --app app.server.app db upgrade
```
每次修改ORM模型后，需先migrate生成脚本，再upgrade应用。

回滚迁移

bash

# 回滚上一个版本
flask --app app.server.app db downgrade
# 回滚到最原始版本（删除所有表）
flask --app app.server.app db downgrade base
# 回滚到指定版本
flask --app app.server.app db downgrade [版本号]

五、核心实现：30行代码实现GPT聊天机器人

结合上述所有知识点，通过OpenAI API+Flask+请求校验+统一响应实现基础GPT聊天机器人接口，核心步骤：

加载环境变量中的OpenAI API秘钥；
编写聊天接口路由，通过Flask-wtf校验入参（如query必填）；
调用OpenAI API获取聊天响应；
按统一格式返回数据；
编写PyTest测试用例对接口进行单元测试；
将聊天记录通过Flask-SQLAlchemy存储到Postgres数据库。

（文档未给出具体30行代码，为实操核心环节，需结合OpenAI Library使用技巧实现）

六、本章核心总结

开发规范：实现配置分离、请求数据校验、统一响应/异常格式，建立企业级Flask项目开发规范；
测试体系：基于PyTest完成接口单元测试，掌握fixture、参数化测试、配置文件等核心用法；
数据库操作：通过Flask-SQLAlchemy实现Postgres的ORM操作，掌握模型创建、增删改查、自动提交；
数据库迁移：使用Flask-Migrate实现表结构的无数据丢失更新，掌握迁移全流程命令；
业务实现：基于OpenAI API实现GPT聊天机器人基础接口，完成从接口开发到测试、数据库存储的全流程落地。

实现第一个GPT聊天机器人 ​

章节介绍 ​

一、基础准备：配置分离与接口请求校验 ​

1.1 .env环境变量配置分离 ​

1.2 接口请求校验的必要性 ​

1.3 Flask-wtf实现请求数据校验 ​

二、接口规范化：统一响应与异常错误设计 ​

2.1 统一响应接口设计 ​

2.1.1 设计原因 ​

2.1.2 核心设计原则 ​

2.1.3 基础统一响应格式 ​

2.1.4 分页数据响应格式 ​

2.2 异常错误状态统一设计 ​

三、单元测试：PyTest配置与API测试用例编写 ​

3.1 Pytest简介与安装 ​

3.2 Pytest测试用例编写规则 ​

3.3 Pytest常用运行命令 ​

3.4 fixture：测试环境预设 ​

3.5 参数化测试 ​

3.6 pytest.ini：运行时配置 ​

四、数据库集成：Postgres与Flask-SQLAlchemy ​

4.1 ORM概念介绍 ​

ORM核心优缺点 ​

4.2 Flask-SQLAlchemy安装与配置 ​

4.3 ORM模型创建 ​

4.3.1 核心规范 ​

4.3.2 常用字段类型 ​

4.3.3 模型创建示例（App模型） ​

4.3.4 表创建 ​

4.3.5 ChatGPT辅助生成ORM模型 ​

4.4 数据库增删改查（CRUD） ​

4.4.1 操作方式约定 ​

4.4.2 新增数据 ​

4.4.3 查询数据 ​

4.4.4 修改数据 ​

4.4.5 删除数据 ​

4.5 重写SQLAlchemy实现自动提交 ​

4.5.1 问题背景 ​

4.5.2 实现方案 ​

4.5.3 使用方式 ​

4.6 Flask-Migrate：数据库迁移 ​

4.6.1 迁移必要性 ​

4.6.2 安装与初始化 ​

4.6.3 核心迁移命令 ​

五、核心实现：30行代码实现GPT聊天机器人 ​

六、本章核心总结 ​

实现第一个GPT聊天机器人

章节介绍

一、基础准备：配置分离与接口请求校验

1.1 .env环境变量配置分离

1.2 接口请求校验的必要性

1.3 Flask-wtf实现请求数据校验

二、接口规范化：统一响应与异常错误设计

2.1 统一响应接口设计

2.1.1 设计原因

2.1.2 核心设计原则

2.1.3 基础统一响应格式

2.1.4 分页数据响应格式

2.2 异常错误状态统一设计

三、单元测试：PyTest配置与API测试用例编写

3.1 Pytest简介与安装

3.2 Pytest测试用例编写规则

3.3 Pytest常用运行命令

3.4 fixture：测试环境预设

3.5 参数化测试

3.6 pytest.ini：运行时配置

四、数据库集成：Postgres与Flask-SQLAlchemy

4.1 ORM概念介绍

ORM核心优缺点

4.2 Flask-SQLAlchemy安装与配置

4.3 ORM模型创建

4.3.1 核心规范

4.3.2 常用字段类型

4.3.3 模型创建示例（App模型）

4.3.4 表创建

4.3.5 ChatGPT辅助生成ORM模型

4.4 数据库增删改查（CRUD）

4.4.1 操作方式约定

4.4.2 新增数据

4.4.3 查询数据

4.4.4 修改数据

4.4.5 删除数据

4.5 重写SQLAlchemy实现自动提交

4.5.1 问题背景

4.5.2 实现方案

4.5.3 使用方式

4.6 Flask-Migrate：数据库迁移

4.6.1 迁移必要性

4.6.2 安装与初始化

4.6.3 核心迁移命令

五、核心实现：30行代码实现GPT聊天机器人

六、本章核心总结