AI项目博客
记录 AI 工程实践与大模型应用

从 0 搭一套 Python AI 知识库(一):先把服务跑起来

本系列共 9 篇,目标是从一个空目录开始,搭出一套能上传文档、能检索、能带出处回答的 Python AI 知识库。这是第 1 篇。

这一篇不碰业务,只做三件事:把项目骨架立起来、把配置和返回格式定好、让服务能启动并通过健康检查。听起来简单,但这三件事如果一开始没定好,后面每加一个接口都要返工。

AI 知识库系统知识库列表

图:服务跑起来后,知识库列表能看到两个库及切片与检索参数。

一、先看最终形态

本篇只创建最小的目录,后面每篇往里加东西:

kb-api/

├─ pyproject.toml

├─ .env

├─ dev.py

└─ app/

   ├─ main.py

   └─ core/

      ├─ config.py

      ├─ response.py

      ├─ errors.py

      ├─ db.py

      └─ redis.py

技术栈是 FastAPI + pydantic-settings + SQLAlchemy(async)+ PostgreSQL/pgvector + Redis + MinIO + LangChain/LangGraph。依赖一次写进 pyproject.toml,后面用到哪个讲哪个。

二、创建项目和依赖

需要 Python 3.13+ 和 uv。没有 uv 的话,先按官网装一个:

mkdir kb-api && cd kb-api

uv init --python 3.13

uv add fastapi "uvicorn[standard]" pydantic-settings

uv add "sqlalchemy[asyncio]" asyncpg psycopg[binary,pool] redis

uv add httpx minio pypdf python-docx openpyxl

uv add langchain langchain-openai langgraph langgraph-checkpoint-postgres

uv add apscheduler

不用一次记住每个包的作用,先知道分工:Web 用 FastAPI,配置用 pydantic-settings,数据库用 SQLAlchemy + asyncpg,文件进 MinIO,问答与 Agent 用 LangChain/LangGraph。

三、配置:连接信息只进 .env

所有配置统一加 KB_ 前缀,字段名自动映射:db_url 对应 KB_DB_URL,redis_url 对应 KB_REDIS_URL。

from functools import lru_cache

from pathlib import Path

from pydantic_settings import BaseSettings, SettingsConfigDict

BASE_DIR = Path(__file__).resolve().parents[2]

class Settings(BaseSettings):

    model_config = SettingsConfigDict(

        env_file=str(BASE_DIR / ".env"),

        env_prefix="KB_",

        env_file_encoding="utf-8",

        case_sensitive=False,

        extra="ignore",          # .env 里多余的键不报错

    )

    service_name: str = "kb-api"

    port: int = 18015

    log_level: str = "INFO"

    db_url: str = ""

    redis_url: str = "redis://127.0.0.1:6379/4"

@lru_cache(maxsize=1)

def get_settings() -> Settings:

    return Settings()

为什么用 lru_cache:配置是单例,没必要每个请求都读一遍文件。代价是改完 .env 要重启进程,不做热刷新。

四、返回信封:前端只认一种格式

所有接口统一返回 {"code":200,"message":"操作成功","data":...}。注意字段是 message,不是 msg。

def envelope(code: int = 200, message: str = "操作成功", data=None) -> dict:

    return {"code": code, "message": message, "data": data}

def success(data=None) -> dict:

    return envelope(200, "操作成功", data)

def fail(code: int, message: str, data=None) -> dict:

    return envelope(code, message, data)

业务错误也返回 HTTP 200,错误码放在 body.code。这不是偷懒:前端拦截器按 body.code 判断,HTTP 非 2xx 会被当成网络异常,用户反而看不到具体提示。

五、异常处理:别让 FastAPI 默认格式漏出去

业务异常抛 PlatformException,全局处理器把它转成统一信封;参数校验、404、未知异常也各有一个处理器。最关键的是最后那个 Exception 兜底,否则一旦漏掉异常,前端收到的就是 FastAPI 默认的 {"detail": ...},整个解析链路都断了。

class PlatformException(Exception):

    def __init__(self, message="操作失败", code=400):

        super().__init__(message)

        self.message = message

        self.code = code

@app.exception_handler(PlatformException)

async def _platform_exception(_, exc: PlatformException):

    return json_response(fail(exc.code, exc.message))

@app.exception_handler(Exception)

async def _unexpected_exception(_, exc: Exception):

    logger.exception("未预期异常", exc_info=exc)

    return json_response(fail(5000, "系统异常,请稍后重试"))

六、启动与健康检查

数据库和 Redis 都做懒加载:模块导入时不连,用到才连。这样即使库暂时不可用,服务也能起来,交给探针去报状态。

@asynccontextmanager

async def lifespan(_: FastAPI):

    logger.info("kb-api 启动:scheduler=%s", settings.scheduler_enabled)

    start_scheduler()

    try:

        yield

    finally:

        stop_scheduler()

        await close_redis()

        await dispose_db()

app = FastAPI(title="kb-api", lifespan=lifespan)

@app.get("/healthz")

async def healthz():

    return success({

        "service": settings.service_name,

        "db": await ping_db(),

        "redis": await ping_redis(),

    })

/healthz 永远返回 200,db 和 redis 分别给出真假。进程活着但依赖挂了,监控能一眼区分。

启动:

uv run python dev.py

curl -s localhost:18015/healthz

期望输出:

{"code":200,"message":"操作成功","data":{"service":"kb-api","db":true,"redis":true}}

七、本篇踩坑

  • 不要在模块导入时建连接。 一导入就连库,库挂一次整个服务起不来,本地开发会非常难受。
  • 配置读不到先查前缀。 环境变量优先级高于 .env;字段名大小写不敏感,但前缀 KB_ 不能少。
  • 端口以启动命令为准。 dev.py 会读 KB_PORT,而 fastapi dev 默认监听 8000,不读这个变量,别被“端口没生效”绕进去。

下一篇

骨架有了,下一步把数据层和登录链路接上:PostgreSQL 建表、SQLAlchemy 异步会话、SSO 换票、Redis 会话,以及贯穿所有查询的多租户隔离。

下一篇:数据库、登录与多租户

赞(0)
未经允许不得转载:AI项目博客 » 从 0 搭一套 Python AI 知识库(一):先把服务跑起来

评论 抢沙发

AI项目博客

专注 AI 工程化落地 —— 覆盖智能体、大模型、Java AI 与 Python 的学习笔记,以及 AI 平台的项目实战记录。

关于本站文章归档

登录

找回密码

注册