feat: add backend application foundation

This commit is contained in:
gqt
2026-07-13 11:24:05 +08:00
parent 81b0faa67a
commit 44e01e2f93
10 changed files with 314 additions and 0 deletions
+1
View File
@@ -0,0 +1 @@
"""HTTP API infrastructure."""
+85
View File
@@ -0,0 +1,85 @@
import logging
from typing import Any
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
logger = logging.getLogger(__name__)
class AppError(Exception):
def __init__(
self,
code: str,
message: str,
status_code: int,
*,
retryable: bool = False,
) -> None:
super().__init__(message)
self.code = code
self.message = message
self.status_code = status_code
self.retryable = retryable
def _payload(
request: Request,
*,
code: str,
message: str,
retryable: bool = False,
details: Any | None = None,
) -> dict[str, Any]:
error: dict[str, Any] = {
"code": code,
"message": message,
"request_id": request.state.request_id,
}
if retryable:
error["retryable"] = True
if details is not None:
error["details"] = details
return {"error": error}
def register_exception_handlers(app: FastAPI) -> None:
@app.exception_handler(AppError)
async def handle_app_error(request: Request, exc: AppError) -> JSONResponse:
return JSONResponse(
status_code=exc.status_code,
content=_payload(
request,
code=exc.code,
message=exc.message,
retryable=exc.retryable,
),
)
@app.exception_handler(RequestValidationError)
async def handle_validation_error(
request: Request, exc: RequestValidationError
) -> JSONResponse:
return JSONResponse(
status_code=422,
content=_payload(
request,
code="VALIDATION_ERROR",
message="请求参数校验失败",
details=exc.errors(),
),
)
@app.exception_handler(Exception)
async def handle_unexpected_error(request: Request, exc: Exception) -> JSONResponse:
logger.exception("Unhandled request error request_id=%s", request.state.request_id)
return JSONResponse(
status_code=500,
content=_payload(
request,
code="INTERNAL_ERROR",
message="服务暂时不可用,请稍后重试",
retryable=True,
),
)
+14
View File
@@ -0,0 +1,14 @@
from uuid import uuid4
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import Response
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
request_id = request.headers.get("X-Request-ID") or str(uuid4())
request.state.request_id = request_id
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
+3
View File
@@ -0,0 +1,3 @@
from fastapi import APIRouter
api_router = APIRouter(prefix="/api/v1")
+66
View File
@@ -0,0 +1,66 @@
from functools import lru_cache
from pathlib import Path
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_prefix="KBQA_",
extra="ignore",
)
app_name: str = "KBQA"
debug: bool = False
dashscope_api_key: str = ""
dashscope_base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1"
chat_model: str = "qwen3.5-flash"
embedding_model: str = "text-embedding-v4"
embedding_dim: int = 1024
database_url: str = "sqlite+aiosqlite:///./data/kbqa.sqlite3"
milvus_uri: str = "./data/milvus/kbqa.db"
raw_data_dir: Path = Path("./data/raw")
chunk_size: int = Field(default=500, ge=1)
chunk_overlap: int = Field(default=50, ge=0)
retrieval_top_k: int = Field(default=5, ge=1, le=100)
max_research_searches: int = Field(default=3, ge=1, le=10)
max_sources: int = Field(default=8, ge=1, le=50)
history_message_limit: int = Field(default=20, ge=1, le=200)
max_upload_mib: int = Field(default=20, ge=1, le=1024)
max_query_length: int = Field(default=10_000, ge=1, le=100_000)
cors_origins: list[str] = [
"http://localhost:5173",
"http://127.0.0.1:5173",
]
run_live_tests: bool = False
live_test_file: Path | None = None
@field_validator("chunk_overlap")
@classmethod
def overlap_must_be_smaller_than_chunk(cls, value: int, info) -> int:
chunk_size = info.data.get("chunk_size", 500)
if value >= chunk_size:
raise ValueError("chunk_overlap must be smaller than chunk_size")
return value
@lru_cache
def get_settings() -> Settings:
return Settings()
def validate_live_settings(settings: Settings) -> None:
missing: list[str] = []
if not settings.dashscope_api_key:
missing.append("KBQA_DASHSCOPE_API_KEY")
if not settings.dashscope_base_url:
missing.append("KBQA_DASHSCOPE_BASE_URL")
if missing:
joined = ", ".join(missing)
raise RuntimeError(f"Missing live model configuration: {joined}")
+48
View File
@@ -0,0 +1,48 @@
from collections.abc import AsyncIterator
from fastapi import Request
from sqlalchemy import event
from sqlalchemy.engine import Engine
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
class Database:
def __init__(self, url: str) -> None:
self.engine: AsyncEngine = create_async_engine(url, pool_pre_ping=True)
self.session_factory = async_sessionmaker(
bind=self.engine,
expire_on_commit=False,
class_=AsyncSession,
)
self._install_sqlite_pragmas(self.engine.sync_engine)
@staticmethod
def _install_sqlite_pragmas(engine: Engine) -> None:
@event.listens_for(engine, "connect")
def configure_sqlite(dbapi_connection, _connection_record) -> None:
cursor = dbapi_connection.cursor()
try:
cursor.execute("PRAGMA foreign_keys=ON")
cursor.execute("PRAGMA journal_mode=WAL")
cursor.execute("PRAGMA busy_timeout=5000")
finally:
cursor.close()
async def dispose(self) -> None:
await self.engine.dispose()
async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
database: Database = request.app.state.database
async with database.session_factory() as session:
yield session
+1
View File
@@ -0,0 +1 @@
"""Health check feature."""
+29
View File
@@ -0,0 +1,29 @@
import asyncio
from fastapi import APIRouter, Request
from sqlalchemy import text
router = APIRouter(tags=["health"])
@router.get("/health")
async def health(request: Request) -> dict[str, object]:
components: dict[str, dict[str, str]] = {
"app": {"status": "ok"},
"sqlite": {"status": "ok"},
"milvus": {"status": "ok"},
}
try:
async with request.app.state.database.engine.connect() as connection:
await connection.execute(text("SELECT 1"))
except Exception:
components["sqlite"] = {"status": "error"}
try:
await asyncio.to_thread(request.app.state.milvus_client.list_collections)
except Exception:
components["milvus"] = {"status": "error"}
overall = "ok" if all(item["status"] == "ok" for item in components.values()) else "degraded"
return {"status": overall, "components": components}
+38
View File
@@ -0,0 +1,38 @@
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
from pymilvus import MilvusClient
from kbqa.config import get_settings
from kbqa.database import Database
def _ensure_runtime_directories(database_url: str, milvus_uri: str, raw_data_dir: Path) -> None:
raw_data_dir.mkdir(parents=True, exist_ok=True)
if database_url.startswith("sqlite+aiosqlite:///./"):
Path(database_url.removeprefix("sqlite+aiosqlite:///./")).parent.mkdir(
parents=True, exist_ok=True
)
Path(milvus_uri).parent.mkdir(parents=True, exist_ok=True)
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
settings = get_settings()
_ensure_runtime_directories(
settings.database_url,
settings.milvus_uri,
settings.raw_data_dir,
)
database = Database(settings.database_url)
milvus_client = MilvusClient(uri=settings.milvus_uri)
app.state.settings = settings
app.state.database = database
app.state.milvus_client = milvus_client
try:
yield
finally:
milvus_client.close()
await database.dispose()
+29
View File
@@ -0,0 +1,29 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from kbqa.api.errors import register_exception_handlers
from kbqa.api.middleware import RequestIDMiddleware
from kbqa.api.router import api_router
from kbqa.config import get_settings
from kbqa.health.routes import router as health_router
from kbqa.lifecycle import lifespan
def create_app() -> FastAPI:
settings = get_settings()
app = FastAPI(title=settings.app_name, version="0.1.0", lifespan=lifespan)
app.add_middleware(RequestIDMiddleware)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
)
register_exception_handlers(app)
app.include_router(health_router)
app.include_router(api_router)
return app
app = create_app()