本系列共 9 篇,目标是从一个空目录开始,搭出一套能上传文档、能检索、能带出处回答的 Python AI 知识库。这是第 1 篇。
这一篇不碰业务,只做三件事:把项目骨架立起来、把配置和返回格式定好、让服务能启动并通过健康检查。听起来简单,但这三件事如果一开始没定好,后面每加一个接口都要返工。

图:服务跑起来后,知识库列表能看到两个库及切片与检索参数。
一、先看最终形态
本篇只创建最小的目录,后面每篇往里加东西:
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 会话,以及贯穿所有查询的多租户隔离。
下一篇:数据库、登录与多租户

AI项目博客






