构建第一个AI聊天机器人：Flask+DeepSeek+Postgres实战

LLM到Al Agent的技术演进

想入门AI Agent开发，先要了解一下LLM/RAG/Agent的技术路线： LLM/RAG/Agent已经成为人工智能领域进步的关键技术理解这三者的概念与关系是做好面向AI编程开发的基础。

• Agent无论传递了多少的工具、记忆、规划、行动等，本质上都是统一编码到Prompt中并传递给大语言模型，然后程序根据输出的内容执行不同的操作，反复执行，直到完成。
• 任何一个Agent都可以在ChatGPT或者大模型的对话界面进行模拟，并且实现类似的效果。
• 实现一个集ip地址查询、数学表达式计算、网络搜索的简单Agent应用。

为什么需要LLMops

一个AIAgent应用涵盖了LLM、记忆、工具、Prompt、规划、知识库、执行者等模块，但每个应用的流程又比较接近，对开发者和非开发者都不友好。

有没有一个平台，能在可视化界面通过鼠标拖拽对应的模块+提示词等内容，就可以让无编程基础的人也可以快速创建一个AlAgent应用并调试部署?

LLMOps 是一个基于LLM的应用程序的生命周期管理平台或者工具，涵盖了LLM应用的开发、部署、配置、运维。LLMOps的旨在简化和优化LLM应用程序的各个环节以确保LLM应用高效、可靠和安全地运行。

LLMOps对使用者友好，极大降低了企业创建AI Agent应用的成本， 把复杂的部分留给了LLMOps开发者。

不使用LLMOps 整理需求、Prompt编写、部署LLM、对接LLM接口、处理应用数据、记录日志、应用工具开发、AI工作流开发、前后端交互开发、前后端部署。

使用LLMOps:需求整理、Prompt编写、上传数据、勾选关联插件、可视化编排工作流、选择LLM模型、发布。

LLMops7层架构

1.表示层：面向用户与生态

• 能力：PC 网站、移动端 APP、开放 API 网关
• 定位：统一入口、多端一致
• 价值：对外提供交互界面，对内屏蔽底层实现，便于第三方对接与生态扩展

2. 接入层：流量入口与安全

• 能力：Nginx、负载均衡、限流、鉴权
• 定位：统一网关、高可用入口
• 价值：抗流量冲击、统一日志、安全防护，是生产系统必备的 “大门”

3. 控制层：业务编排与权限

• 能力：AI 应用、用户、授权、插件模块
• 定位：业务逻辑中枢、权限管控点
• 价值：解耦 “业务规则” 与 “AI 能力”，支持多租户、多应用、多角色管理

4. 服务层：原子化能力输出

• 能力：LLM 应用服务、文件上传、授权、文本嵌入等
• 定位：可复用、可坚控、可编排
• 价值：将 AI 能力拆分为细粒度服务，支持接口复用、独立扩容、链路追踪

5. 核心层：大模型与 Agent 引擎

• 能力：多 LLM 集成、Agent 调度、LangChain 封装
• 定位：LLM 能力内核、智能化引擎
• 价值：屏蔽不同模型差异，统一调用、统一计费、统一异常，支持工具链与复杂推理

6. 存储层：多模数据统一管理

• 能力：PostgreSQL、Redis、向量数据库、对象存储
• 定位：数据底座、RAG 基础
• 价值：同时支撑业务数据、缓存、向量知识库、文件 / 图片 / 音频，满足 LLM 全场景数据需求

7. 资源层：基础设施与弹性算力

• 能力：Linux、Docker、K8s
• 定位：环境统一、弹性伸缩
• 价值：支持容器化部署、CI/CD、自动扩缩容，解决 GPU/CPU 资源调度问题

核心特点

1. 分层解耦，团队可分工

• 前后端分离、业务与 AI 分离、应用与基础设施分离
• 适合中大型团队并行开发：前端组、后端组、AI 组、运维 / 平台组

2. 生产级工程化能力

• 高可用：接入层负载均衡 + 服务层集群部署
• 可观测：每层可独立坚控（QPS、耗时、异常、Token 消耗）
• 可运维：Docker + K8s 实现一键发布、弹性扩缩容

3. 天生适配 LLM 应用痛点

• 支持多厂商模型统一管理
• 内置向量存储 + RAG 能力
• 支持Agent 工具链与复杂对话
• 可做计费、限流、权限、审计

4. 极强扩展性

• 横向：服务独立扩容
• 纵向：新增模型、存储、插件不影响上层业务
• 生态：开放 API 支持外部系统对接

5. 安全与合规可控

• 统一授权、统一鉴权、统一日志
• 支持敏感词、内容审核、数据隔离
• 满足企业级合规与审计要求

适用场景

• 企业级 AI 助手 / 知识库 / 客服
• 多模型统一管理平台（LLM Gateway）
• 支持 RAG + Agent 的私有化部署系统
• 可商用、可交付的标准化 LLMOps 平台

第一个GPT聊天机器人

使用 Flask+DeepSeek+Postgres封装一个简单对话的机器人，首先要接入那个大模型都要先注册一个用户，创建一个应用，会有唯一的一个API_KEY进行网络通信，一个粗糙的开始就是一个好的开始，万丈高楼平地起~

Python 开发环境配置

1.虚拟环境

• 创建虚拟环境命令: python-m venv env 命令在Python 2.7+或Python 3.3+以上可以使用。
• Python虚拟环境与系统环境之间是隔离的一个项目一个虚拟环境，不同的虚拟环境不会相互影响，最大程度避免包冲突。
• 占用更大的空间，进入虚拟环境都需要使用命令activate激活才能使用，使用完，要使用deactivate命令退出。

# 创建虚拟环境 (env)是当前的虚拟环境
PS E:pythonProjectllmopsllmops-api> python -m venv env
PS E:pythonProjectllmopsllmops-api> .envScriptsactivate
(env) PS E:pythonProjectllmopsllmops-api> deactivate
PS E:pythonProjectllmopsllmops-api>

2.配置国内 PyPI 加速I器

命令行配置（Windows/macOS/Linux 通用）, 激活虚拟环境后执行 pip config set global.index-url

(env) PS E:pythonProjectllmopsllmops-api> pip config set global.index-url 
Writing to C:Users15117AppDataRoamingpippip.ini
(env) PS E:pythonProjectllmopsllmops-api> pip config list
global.index-url='https://mirrors.cloud.tencent.com/pypi/simple/'

3.Postgres数据库

可以直接去官网下载安装客户端，支持Linux、Mac、Windows，安装的过程中有两个目录特别重要，一个是安装目录里面包含了很多应用程序，一个目录是数据储存的目录。

# 

D:Program Files (x86)PostgreSQL16bin
D:Program Files (x86)PostgreSQL16data

可以让AI帮我写一个Windows的启动/停止的.bat脚本，脚本如下(亲测可用)：启动脚本 start_postgres.bat

@echo off
chcp 65001 > nul
title PostgreSQL 16 启动脚本

:: 设置PostgreSQL的路径
set "PG_BIN=D:Program Files (x86)PostgreSQL16bin"
set "PG_DATA=D:Program Files (x86)PostgreSQL16data"

:: 切换到bin目录
cd /d "%PG_BIN%"

:: 检查是否以管理员身份运行
net session >nul 2>&1
if %errorLevel% neq 0 (
    echo 请以管理员身份运行此脚本！
    pause
    exit /b 1
)

echo 正在启动PostgreSQL 16...
pg_ctl.exe start -D "%PG_DATA%" -w

:: 检查启动结果
if %errorLevel% equ 0 (
    echo PostgreSQL 16 启动成功！
) else (
    echo PostgreSQL 16 启动失败，请检查日志或路径是否正确！
)

pause

单独的停止脚本（stop_postgres.bat）

@echo off
chcp 65001 > nul
title PostgreSQL 16 停止脚本

:: 设置PostgreSQL的路径
set "PG_BIN=D:Program Files (x86)PostgreSQL16bin"
set "PG_DATA=D:Program Files (x86)PostgreSQL16data"

:: 切换到bin目录
cd /d "%PG_BIN%"

:: 检查是否以管理员身份运行
net session >nul 2>&1
if %errorLevel% neq 0 (
    echo 请以管理员身份运行此脚本！
    pause
    exit /b 1
)

echo 正在停止PostgreSQL 16...
pg_ctl.exe stop -D "%PG_DATA%" -m fast -w

:: 检查停止结果
if %errorLevel% equ 0 (
    echo PostgreSQL 16 停止成功！
) else (
    echo PostgreSQL 16 停止失败，请检查服务是否正在运行！
)

pause

实践

我使用的是Flask框架进行实践，再实践之前我已经在Deepseek应用平台创建了应用生成了key值，敏感的配置信息，一般生成在根目录下的.env文件中，这已经是很多程序员的共识。

1.加载.env 文件，实现配置文件分离

在Python项目中，如果想将.env文件的信息加载成环境变量，可以使用python-dotenv库,通过os.getenv即可获取在环境变量中的配置信息，将代码与配置分离，方便后期维护。

# 安装
# pip install python-dotenv

# 加载
import dotenv
dotenv.load_dotenv()

2. Flask-wtf 参数安全校验

• 安全性:通过验证请求的数据，可以确保输入符合预期，避免存在的安全漏洞，如SQL注入等。
• 数据完整性:确保数据的完整，例如数据格式、类型、长度等，避免因为不完整或不正确的数据导致系统出错。
• 提高用户体验:对用户提交的不合法请求给出友好提示，引导用户按照规定格式或要求重新输入，从而提高用户体验。
• Flask-wtf是基于wtforms封装的Flask插件，支持CSRF保护快速提取数据、自定义验证规则及wtforms的所有规则。
• wtforms支持提取的数据类型:布尔值、日期、文本、时间、浮点数、邮箱、文件、音频、整数、自定义数据类型等。
• wtforms支持校验的规则:必填、长度、邮箱、正则、范围UUID、数据可选、URL等。

# pip install -U Flask-WTF 

from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField
from wtforms.validators import DataRequired, Length

class CompletionRequest(FlaskForm):
    """ 基础聊天参数校验 """
    query = StringField('query', validators=[
        DataRequired(message="用户的提问是必填"),
        Length(min=1, max=1000, message="用户的提问长度不能超过1000")
    ])

3. 响应接口设计与实现

先定义一个枚举字典HttpCode类，封装统一的输入输出，其实就是统一的结构体，封装完统一的输出如下：

{
    "code": "success",
    "data": {
        "description": "这是一个简单的Deepseek对话的测试机器人",
        "id": "cab840dc-29c1-4f8d-b2eb-b792a7f1235a",
        "name": "划水鸭AI聊天机器人"
    },
    "message": ""
}

python的代码简单如下：

from enum import Enum

class HttpCode(str, Enum):
    SUCCESS = 'success'  # 成功
    FAIL = 'fail'  # 失败

class Response:
    code: HttpCode = HttpCode.SUCCESS
    message: str = ""
    data: Any = field(default_factory=dict)

def json(data: Response = None):
    return jsonify(data), 200

def success_json(data: Any = None):
    return json(Response(code=HttpCode.SUCCESS, message="", data=data))

4. 异常处理

也是同样的道理，对于项目中的错误处理和异常也统一处理

class CustomException(Exception):
    code: HttpCode = HttpCode.FAIL
    message: str = ""
    data: Any = field(default_factory=dict)

    def __init__(self, message: str = None, data: Any = None):
        super().__init__()
        self.message = message
        self.data = data

class FailException(CustomException):
    pass

class NotFoundException(CustomException):
    code = HttpCode.NOT_FOUND

5. PyTest测试用例

Pytest是一个功能强大且易于使用的Python测试框架，用于编写和运行“单元测试“用例、集成测试和功能测试，对比 unitest框架，Pytest的优势非常明显:

• 简单易用: Pytest的语法简单明了，学习曲线低，它采用了自然的测试用例编写方式，使得编写测试代码变得更加简单和直观。
• 灵活性:Pytest支持多种测试样式，包括函数式、类式、模块级别的测试，甚至支持测试生成器，而且插件生态非常丰富，可以通过 插件来扩展Pytest的功能，满足各种测试场景。
• 集成性:Pytest 可以很好地与其他测试工具和框架集成，比如与Jenkins、Travis CI、Coverage.py等。它还支持与其他测试框架(如 unittest和doctest)协同工作。
• 丰富的输出信息:Pytest在测试运行过程中提供丰富的输出信息，包括测试结果、失败原因、详细的堆栈跟踪等，便于开发人员定为和修复问题。

pip install pytest

pytest会递归查找当前目录及子目录下的所有以test开始或者_test结尾的Python脚本，执行其中符合规则的函数和方法，不需要显 示调用，在Pytest中，基础的配置可以放在 pytest.ini配置文件中，格式如下:

[pytest]
# pytest cache
cache_dir = tmp/.pytest_cache

# pytest options
addopts = -v -s

2.fixture 设置测试环境

import pytest
from app.http.app import app

@pytest.fixture
def client():
    """获取Flask应用的测试应用，并返回"""
    app.config["TESTING"] = True
    with app.test_client() as client:
        yield client

3.参数化测试，调用方法：

import pytest

from pkg.response import HttpCode

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

pytest 执行结果：

(env) PS E:pythonProjectllmopsllmops-apitest> pytest -v -s
=========================== test session starts ===========================
platform win32 -- Python 3.10.11, pytest-9.0.2, pluggy-1.6.0 -- E:pythonProjectllmopsllmops-apienvScriptspython.exe
cachedir: tmp.pytest_cache
rootdir: E:pythonProjectllmopsllmops-api
configfile: pytest.ini
plugins: anyio-4.12.1
collected 2 items                                                                                
internalhandlertest_app_handler.py::TestAppHandler::test_completion[None] PASSED
internalhandlertest_app_handler.py::TestAppHandler::test_completion[u4f60u597duff0cu4f60u662fu8c01uff1f] PASSED
=========================== 2 passed in 6.08s ===========================

4.Flask_SQLAlchemy 数据操作

# pip install flask-sqlalchemy psycopg2

#SQLAlchemy数据库配置
SQLALCHEMY_DATABASE_URI=postgresql://user:password@127.0.0.1:5432/llmops?client_encoding=utf8
SQLALCHEMY_POOL_SIZE=30
SQLALCHEMY_POOL_RECYCLE=3600
SQLALCHEMY_ECHO=True

5.项目的实践步骤

项目的加载流程和步骤：

1、加载根目录下的.env文件，读取敏感配置

2、Injector是全局加载的依赖注入库

3、加载路由Router，Flask的路由蓝图完成，注册到蓝图中

4、SQLAlchemy 在启动Http服务的时候自动加载了

import dotenv
from pkg.sqlalchemy import SQLAlchemy
from injector import Injector

from app.http.module import ExtensionModule
from internal.router import Router
from internal.server import Http
from config import Config

dotenv.load_dotenv()

conf = Config()

injector = Injector([ExtensionModule])
app = Http(__name__, conf=conf, db=injector.get(SQLAlchemy), router=injector.get(Router))

if __name__ == '__main__':
    app.run(debug=True)

至此，一个接入Deepseek简单的聊天机器人就已经运行起来了，有需要源码的小伙伴可以私聊我，一起加油~