ai

1.本章目标 #

本章目标是实现一个支持 HTTP 协议的后端 API 服务。具体功能如下：

支持通过 .env 配置应用基础参数（如监听主机、端口、调试模式、文件上传大小限制、日志参数等），实现灵活的环境适配。
封装统一的应用配置模块 app/config.py，集中管理所有环境变量及应用常量，降低硬编码风险，便于后续扩展和维护。
提供统一的日志系统（app/utils/logger.py），支持多级别日志输出、日志轮转、文件和终端的可选输出方式，便于开发调试与生产问题追踪。
基于 Flask 框架进行服务搭建，封装应用工厂函数，统一初始化 CORS 支持等常见中间件。
提供简单的 HTTP 服务启动主程序入口 main.py，支持通过命令行一键运行服务，自动加载配置信息并输出规范的日志。
实现项目的基本目录结构划分，便于按层次维护代码、静态资源及模板。

通过本章实现，你将能够一键启动 HTTP 服务，为后续 API 及数据管理等高级功能模块奠定坚实基础。

2.目录结构 #

# 项目根目录
rag-lite/
    # 应用包目录
    ├── app/
        # 静态文件目录（如CSS、JS、图片等）
        │   ├── static/
        # 模板文件目录（如HTML模板文件）
        │   ├── templates/
        # 工具模块目录
        │   ├── utils/
                # 日志工具模块
        │   │   └── logger.py
        # 应用初始化模块
        │   ├── __init__.py
        # 应用配置模块
        │   └── config.py
    # 日志文件目录
    ├── logs/
            # 应用的日志文件
        │   └── rag_lite.log
    # 应用启动入口（主程序文件）
    ├── main.py
    # Python 项目依赖管理文件
    ├── pyproject.toml

3.安装依赖 #

uv add flask flask_cors  sqlalchemy python-dotenv langchain langchain-community langchain-openai langchain-chroma langchain-huggingface  langchain-deepseek pymysql minio langchain_milvus docx2txt pypdf sentence-transformers langchain_milvus

包名	作用简述
flask	轻量级Web应用框架，用于快速搭建Web服务器和API。
flask_cors	Flask 的跨域资源共享扩展，允许前端跨域访问后端API。
sqlalchemy	Python中强大的ORM工具，实现数据库操作对象化。
python-dotenv	用于加载和管理环境变量，支持从`.env`文件读取配置信息。
langchain	大模型相关应用的开发框架，便于构建与AI/LLM相关的工作流。
langchain-community	Langchain的社区贡献组件包，包含各类社区支持的工具与集成。
langchain-openai	Langchain针对OpenAI平台（如ChatGPT等）的集成和封装。
langchain-chroma	Langchain与Chroma向量数据库的集成，用于存储和检索向量数据。
langchain-huggingface	Langchain与Huggingface平台的集成，支持各类Huggingface模型调用。
langchain-deepseek	Langchain与DeepSeek平台的集成，用于接入DeepSeek相关的AI模型。
pymysql	纯 Python 实现的 MySQL 数据库驱动，SQLAlchemy 访问 MySQL 时的必要依赖。
minio	支持 MinIO 分布式对象存储服务，可用于大文件和分布式存储需求。
docx2txt	纯Python的docx文档（Word）解析库，用于提取docx文件中的文本内容。
pypdf	强大的Python PDF解析库，用于从PDF文件中读取、分割页面和提取文本数据。
sentence-transformers	强大的文本向量化库，支持多种预训练模型进行文本嵌入计算。
langchain_milvus	Langchain与Milvus向量数据库的集成，实现高效的文本向量存储与相似度检索。

4.启动HTTP #

4.1. .env #

.env

# 应用配置
APP_HOST=0.0.0.0
APP_PORT=5000
APP_DEBUG=True
MAX_FILE_SIZE=104857600
SECRET_KEY=dev-secret-key-change-in-production


# 日志配置
LOG_DIR=./logs
LOG_FILE=rag_lite.log
LOG_LEVEL=INFO
LOG_ENABLE_FILE=True
LOG_ENABLE_CONSOLE=True

4.2. init.py #

app/init.py

# RAG Lite 应用模块说明
"""
RAG Lite Application
"""

# 导入操作系统相关模块
import os
# 从 Flask 包导入 Flask 应用对象
from flask import Flask
# 导入 Flask 跨域资源共享支持
from flask_cors import CORS
# 导入应用配置类
from app.config import Config
# 导入日志工具，用于获取日志记录器
from app.utils.logger import get_logger

# 定义创建 Flask 应用的工厂函数
def create_app(config_class=Config):
    # 获取名为当前模块的日志记录器（在函数内部获取，避免模块导入时过早初始化）
    logger = get_logger(__name__)

    # 创建 Flask 应用对象，并指定模板和静态文件目录
    base_dir = os.path.abspath(os.path.dirname(__file__))
    # 创建 Flask 应用对象，并指定模板和静态文件目录
    app = Flask(
        __name__,
        # 指定模板文件目录
        template_folder=os.path.join(base_dir, 'templates'),
        # 指定静态文件目录
        static_folder=os.path.join(base_dir, 'static')
    )
    # 从给定配置类加载配置信息到应用
    app.config.from_object(config_class)

    # 启用跨域请求支持
    CORS(app)

    # 记录应用创建日志信息
    logger.info("Flask 应用已创建")

    # 定义首页路由
    @app.route('/')
    def index():
        return "Hello, World!"

    # 返回已配置的 Flask 应用对象
    return app

4.3. config.py #

app/config.py

"""
配置管理模块
"""

# 导入操作系统相关模块
import os
# 导入 Path，处理路径
from pathlib import Path
# 导入 dotenv，用于加载 .env 文件中的环境变量
from dotenv import load_dotenv

# 加载 .env 文件中的环境变量到系统环境变量
load_dotenv()

# 定义应用配置类
class Config:
    """应用配置类"""

    # 基础配置
    # 项目根目录路径（取上级目录）
    BASE_DIR = Path(__file__).parent.parent
    # 加载环境变量 SECRET_KEY，若未设置则使用默认开发密钥
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key-change-in-production'

    # 应用配置
    # 读取应用监听的主机地址，默认为本地所有地址
    APP_HOST = os.environ.get('APP_HOST', '0.0.0.0')
    # 读取应用监听的端口，默认为 5000，类型为 int
    APP_PORT = int(os.environ.get('APP_PORT', 5000))
    # 读取 debug 模式配置，字符串转小写等于 'true' 则为 True（开启调试）
    APP_DEBUG = os.environ.get('APP_DEBUG', 'false').lower() == 'true'
    # 读取允许上传的最大文件大小，默认为 100MB，类型为 int
    MAX_FILE_SIZE = int(os.environ.get('MAX_FILE_SIZE', 104857600))  # 100MB
    # 允许上传的文件扩展名集合
    ALLOWED_EXTENSIONS = {'pdf', 'docx', 'txt', 'md'}

    # 日志配置
    # 日志目录，默认 './logs'
    LOG_DIR = os.environ.get('LOG_DIR', './logs')
    # 日志文件名，默认 'rag_lite.log'
    LOG_FILE = os.environ.get('LOG_FILE', 'rag_lite.log')
    # 日志等级，默认 'INFO'
    LOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO')
    # 是否启用控制台日志，默认 True
    LOG_ENABLE_CONSOLE = os.environ.get('LOG_ENABLE_CONSOLE', 'true').lower() == 'true'
    # 是否启用文件日志，默认 True
    LOG_ENABLE_FILE = os.environ.get('LOG_ENABLE_FILE', 'true').lower() == 'true'

4.4. logger.py #

app/utils/logger.py

# 日志工具模块，提供统一的日志配置和管理功能
"""
日志工具模块
提供统一的日志配置和管理功能
"""
# 导入标准库 logging，用于日志管理
import logging
# 导入 sys，用于标准输出流
import sys
# 导入 RotatingFileHandler，用于日志文件轮转
from logging.handlers import RotatingFileHandler
# 导入 Path，用于文件/目录路径处理
from pathlib import Path
# 导入类型提示工具
from typing import Optional
# 导入应用配置类
from app.config import Config
# 日志管理器类
class LoggerManager:
    """日志管理器"""

    # 日志格式字符串
    FORMAT_STRING = '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
    # 最大日志文件大小（10MB）
    MAX_BYTES = 10 * 1024 * 1024
    # 日志文件保留份数
    BACKUP_COUNT = 5

    def __init__(self):
        """初始化日志管理器"""
        self.log_dir = Path(Config.LOG_DIR)
        self.log_file = Config.LOG_FILE
        self.level = getattr(logging, Config.LOG_LEVEL.upper(), logging.INFO)
        self.enable_file = Config.LOG_ENABLE_FILE
        self.enable_console = Config.LOG_ENABLE_CONSOLE
        # 初始化日志系统
        self._initialize()

    def _initialize(self):
        """初始化日志系统"""
        # 如果启用文件日志则创建日志目录
        if Config.LOG_ENABLE_FILE:
            Path(Config.LOG_DIR).mkdir(parents=True, exist_ok=True)
        # 获取根日志记录器
        root_logger = logging.getLogger()
        # 移除原有所有处理器，防止重复添加
        root_logger.handlers.clear()
        # 设置日志级别
        root_logger.setLevel(self.level)
        # 创建日志格式器
        formatter = logging.Formatter(self.FORMAT_STRING)
        # 如果启用控制台日志，则创建并添加控制台日志处理器
        if self.enable_console:
            # 创建控制台日志处理器
            console_handler = logging.StreamHandler(sys.stdout)
            # 设置日志级别
            console_handler.setLevel(self.level)
            # 设置日志格式
            console_handler.setFormatter(formatter)
            # 添加控制台日志处理器到根日志记录器
            root_logger.addHandler(console_handler)

        # 如果启用文件日志，则创建文件日志处理器，支持轮转
        if self.enable_file:
            # 创建日志文件路径
            log_path = self.log_dir / self.log_file
            # 创建文件日志处理器，支持轮转
            file_handler = RotatingFileHandler(
                # 日志文件路径
                str(log_path),
                # 最大日志文件大小
                maxBytes=self.MAX_BYTES,
                # 日志文件保留份数
                backupCount=self.BACKUP_COUNT,
                # 编码
                encoding='utf-8',
            )
            # 设置日志级别
            file_handler.setLevel(self.level)
            # 设置日志格式
            file_handler.setFormatter(formatter)
            # 添加文件日志处理器到根日志记录器
            root_logger.addHandler(file_handler)

        # 捕获 warnings 模块的警告作为日志
        logging.captureWarnings(True)

    # 获取日志记录器
    def get_logger(self, name: Optional[str] = None) -> logging.Logger:
        """
        获取日志记录器

        Args:
            name: 日志记录器名称，通常使用 __name__。如果为 None，返回根日志记录器

        Returns:
            logging.Logger 实例
        """
        # 如果日志记录器名称为 None，返回根日志记录器
        if name is None:
            # 返回根日志记录器
            return logging.getLogger()
        # 返回指定名称的日志记录器
        return logging.getLogger(name)

# 在模块级别创建 LoggerManager 实例并初始化
logger_manager = LoggerManager()
# 获取日志记录器
def get_logger(name: Optional[str] = None) -> logging.Logger:
    """
    获取日志记录器

    Args:
        name: 日志记录器名称，通常使用 __name__。如果为 None，返回根日志记录器
    """
    return logger_manager.get_logger(name)

4.5. main.py #

main.py

# 应用启动入口说明
+"""
+应用启动入口
+"""

# 导入操作系统相关模块
+import os

# 从 app 包导入创建 Flask 应用的工厂函数
+from app import create_app

# 导入应用配置类
+from app.config import Config

# 导入日志获取方法（日志系统会在首次使用时自动从 Config 获取配置并初始化）
+from app.utils.logger import get_logger

# 获取当前模块日志记录器（会自动初始化日志系统）
+logger = get_logger(__name__)


# 仅当直接运行该文件时才执行以下代码
+if __name__ == '__main__':
    # 创建 Flask 应用对象
+   app = create_app()

    # 记录应用启动的信息到日志
+   logger.info(f"Starting  RAG Lite server on {Config.APP_HOST}:{Config.APP_PORT}")

    # 启动 Flask 应用，监听指定主机和端口，是否开启调试模式由配置决定
+   app.run(
+       host=Config.APP_HOST,
+       port=Config.APP_PORT,
+       debug=Config.APP_DEBUG
+   )

导航菜单

1.本章目标 #

2.目录结构 #

3.安装依赖 #

4.启动HTTP #

4.1. .env #

4.2. init.py #

4.3. config.py #

4.4. logger.py #

4.5. main.py #

访问验证