请求生命周期与服务器/进程骨架

30 秒导读: OGX 是一台把 OpenAI 风格 API(Responses、Chat Completions、Embeddings 等)对外暴露、后面接各种推理后端的服务器。本章只讲两件事:一个 HTTP 请求进来、在被分发到具体业务逻辑之前经历了什么,以及这台服务器进程是怎么被拼起来的。不深入 provider 解析算法(那是 02),也不讲具体 API 逻辑。

本章属于 OGX 讲解系列,建议先读 index 建立全景。相邻章节:

• 02 · Provider 架构:注册、解析与自动路由装配
• 03 · 路由层与 OpenAI 兼容适配
• 06 · 存储、分发注册表与多租户隔离

---

1. 这是什么(零基础也能懂)

一句话: 本章讲的是 OGX 的"前门"——从 uvicorn 收到一个 HTTP 请求,到这个请求即将进入某个 API 路由处理函数之间,发生的全部事情;外加"这台服务器开机时把自己组装好"的那段启动代码。

为什么值得单独讲一章? 因为 OGX 的很多关键能力(版本兼容、认证、多租户、限流指标、压缩、错误格式统一)都不在业务代码里,而在请求进门时的一层层中间件里。看懂这一层,你才知道一个请求"还没到业务逻辑就可能被挡在哪、被改写成什么样"。

一个直觉类比: 把服务器想成一栋写字楼。

• 进程装配(Stack.initialize)= 开业前的装修:接好水电(存储后端)、把各部门(provider 实现)招进来、门牌挂好(注册路由)。
• 中间件栈 = 大楼门口的一排安检闸机:先过压缩包检查、再刷工牌(认证)、再确认你属于哪家公司(租户)、再看你能不能进这层楼(路由授权)……一道一道过完,才放你上楼(路由分发)。

用起来什么样? 你几乎感觉不到它——这正是设计目标。你把现成的 OpenAI SDK 指向 OGX 的地址就行:

# 示意,非源码:客户端视角,请求打到 OGX
from openai import OpenAI

client = OpenAI(base_url="http://localhost:8321/v1", api_key="...")
resp = client.chat.completions.create(
    model="ollama/llama3.2:3b",       # provider/model 形式
    messages=[{"role": "user", "content": "hi"}],
)
# 这一个 POST /v1/chat/completions 请求,会先穿过本章讲的整条中间件链,
# 才到达 03 章讲的路由与适配层。

服务器这一侧是怎么被拉起来的?靠 CLI 里的一行 uvicorn 调用,用 factory 模式指向 create_app:

# 真实调用:src/ogx/cli/stack/run.py:193
uvicorn.run("ogx.core.server.server:create_app", factory=True, ...)

factory=True 表示 uvicorn 会调用 create_app() 拿到 app 实例——所以 create_app 就是整台服务器的组装入口。

---

2. 顶层全景(它大概怎么转)

分两条线看:装配线(开机时跑一次)和请求线(每个请求跑一次)。

2.1 两条线的关系

                        ┌──────────────────────── 装配线(开机一次)────────────────────────┐
  uvicorn 调用 factory  │  create_app()                                                      │
  ───────────────────▶  │    读 OGX_CONFIG → 解析 YAML → replace_env_vars → StackConfig       │
                        │    new StackApp(config)                                             │
                        │        └─ 临时线程里 asyncio.run(Stack.initialize())  ← 拼装 provider │
                        │    add_middleware(...) × N   ← 搭中间件栈                            │
                        │    include_router(...)       ← 挂载自动发现的路由                    │
                        │    注册 exception_handler                                            │
                        └────────────────────────────────┬───────────────────────────────────┘
                                                          │ 装配完成,进入服务循环
                                                          ▼
                        ┌──────────────────────── 请求线(每请求一次)───────────────────────┐
  HTTP 请求 ──────────▶ │  中间件栈(外→内,见 §3)→ Route Dispatch → 路由处理函数(见 03)   │
                        └───────────────────────────────────────────────────────────────────┘

2.2 本章涉及的部件一句话职责

部件	干什么	在哪
create_app	组装整台 FastAPI 服务器的工厂函数	core/server/server.py:358
StackApp	FastAPI 子类,持有 Stack 实例以便启停后台任务	core/server/server.py:126
中间件栈	版本校验 / 认证 / 租户 / 授权 / 指标 / 压缩 / provider 数据	core/server/server.py、core/server/auth.py、core/server/metrics.py
global_exception_handler	把各种异常翻译成统一(且多 SDK 兼容)的 JSON 错误	core/server/server.py:94
lifespan	FastAPI 生命周期:开机启后台任务,关机时优雅 shutdown	core/server/server.py:171
路由自动发现	从 ogx_api 各包里自动找出 create_router 工厂	core/server/fastapi_router_registry.py:37
Stack.initialize	进程级装配:存储、内部实现、provider 解析、资源注册	core/stack.py:749
replace_env_vars	配置里 ${env.X} 语法的环境变量替换	core/stack.py:485

---

3. 请求线:一个请求怎么穿过中间件栈

先讲最容易踩坑的一点:中间件的执行顺序,和代码里 add_middleware 的书写顺序是反的。

3.1 关键规则:后加的先跑(外层)

Starlette/FastAPI 里,每次 add_middleware 都把新中间件插到最外层。于是:

最后 add_middleware 的那个,是请求第一个碰到的(最外层);最先加的,反而最贴近路由(最内层)。

create_app 里对这条规则是有意识利用的。看认证三件套的注释就知道(server.py:417):它故意先加 RouteAuthorization、后加 Authentication,好让运行时顺序变成 Auth → Tenancy → RouteAuth。

3.2 实际的执行顺序(外 → 内)

把 create_app 里所有 add_middleware 按"后加先跑"翻过来,一个请求真正经历的顺序是:

HTTP 请求
   │
   ▼  ① ZstdDecompressionMiddleware      解压 zstd 请求体(server.py:264)
   ▼  ② RequestMetricsMiddleware         计时/计数/并发数(metrics.py:133)
   ▼  ③ AuthenticationMiddleware*        校验 Bearer token,写入 principal/tenant(auth.py:30)
   ▼  ④ TenancyMiddleware*               强制租户模式 disabled/single/multi(auth.py:419)
   ▼  ⑤ RouteAuthorizationMiddleware*    按 route_policy 判断这条路由能不能走(auth.py:209)
   ▼  ⑥ ProviderDataMiddleware           解析 X-OGX-Provider-Data + 测试上下文(server.py:228)
   ▼  ⑦ ClientVersionMiddleware**        客户端 major.minor 版本兼容校验(server.py:200)
   │
   ▼  Route Dispatch  → FastAPI 路由处理函数(见 03 章)

  *  仅在配置了 auth / tenancy / route_policy 时才加入
  ** 除非设了环境变量 OGX_DISABLE_VERSION_CHECK

怎么读这张图: 从上到下就是一个请求被"逐层剥开"的顺序。任何一层都可能提前短路返回错误(如 ③ 认证失败返回 401),后面的层就不会执行。

注意这和仓库自带 ARCHITECTURE.md 里画的简化流程不完全一样——那张图只强调了认证三件套。上面这张是把 create_app 里全部 add_middleware 调用还原后的真实全序(逐条核对过 server.py:405-475 的加入次序)。

3.3 每个中间件在解决什么问题

① Zstd 解压 —— 给带宽敏感的客户端省流量。 只有 content-encoding: zstd 的请求体才处理;解压后超过 100 MB 直接返回 413;解压失败则记一条 warning、退回用原始压缩体继续(容错而非直接失败)。核心在 server.py:264 的 ZstdDecompressionMiddleware.__call__,内部用 asyncio.to_thread 把解压放到线程池避免阻塞事件循环。它放在最外层是合理的:得先把包解开,后面的层才读得懂 body。

② 请求指标 —— 可观测性。 记三个 OpenTelemetry 指标:requests_total(按 api/method/status)、request_duration_seconds、concurrent_requests。它把 path 反查成 api + method 名字,靠的是首个请求时懒构建的路由映射表(metrics.py:168 的 if self._patterns is None),映射来自 build_route_to_api_map。放得比认证更外,是为了连"认证失败的请求"也能计入指标。

③ 认证 —— 谁在调。 从 Authorization: Bearer <token> 取 token,交给配置的 auth provider 校验,成功后把 principal、user_attributes、tenant_id 写进 ASGI scope 供下游用。两个细节值得记:

• 公开路由可豁免。 路由若在 webmethod 上标了 require_authentication=False(如 /health、/version),直接放行(auth.py:130)。
• WebSocket 也认证。 握手请求同样带 token,scope["type"] in ("http", "websocket") 两种都处理,拒绝时用 WebSocket 关闭码 4401(auth.py:196)。

④ 租户 —— 你属于哪家。 三种模式(auth.py:419 的 TenancyMiddleware):

模式	行为
disabled	直接放行,什么都不做
single	把 scope["tenant_id"] 强制设为默认租户;principal 缺失则填 "system"
multi	认证后仍无 tenant_id 就拒绝(租户上下文必须解析出来)

single/multi 模式下,公开路由(如 health)仍会被豁免(_is_public_route)。

⑤ 路由授权 —— 你能不能走这条路由。 按 route_policy 规则顺序匹配、首条命中即决定(auth.py:251 的 _is_route_allowed);没有任何规则命中则默认拒绝。路径匹配支持精确、前缀通配 *、全通配 *、以及 regex: 前缀正则(auth.py:317 的 _route_matches);还能配 when/unless 条件对用户属性做 ABAC 判断。

⑥ Provider 数据 —— 把每请求上下文塞进 contextvar。 解析 X-OGX-Provider-Data 头(客户端临时传给 provider 的数据,如某个 API key),连同用户信息装进一个 request_provider_data_context(server.py:244),让深处的 provider 代码不用层层传参也能取到。测试模式(OGX_TEST_INFERENCE_MODE)下还会同步一份确定性 ID 生成用的测试上下文。

⑦ 客户端版本校验 —— 挡住不兼容的旧客户端。 读 x-ogx-client-version 头,和服务器版本比 major.minor,不一致就返回 426 Upgrade Required(server.py:215);解析失败则放行(容错)。可用 OGX_DISABLE_VERSION_CHECK 关掉。

3.4 出错时:统一而"多 SDK 兼容"的错误格式

请求在任何一层抛出未捕获异常,最终都会落到 global_exception_handler(server.py:94)。它注册在一串具体异常 + 兜底 Exception 上(server.py:479-486)。它做的不只是"翻译成 JSON",还要按目标 API 的方言给不同的错误信封:

              异常 exc
                │
                ▼
        translate_exception(exc)        ← core/exceptions/translation.py:15
                │  (ValidationError→400 带 errors 列表;OGXError→自带 status;
                │   provider SDK 异常保留其 status_code;否则 500)
                ▼
     请求路径是 /v1alpha/interactions ?
        │是                                   │否
        ▼                                     ▼
  Google 错误信封                      ResourceNotFoundError 且路径以
  {"error":{"code",                    /v1/vector_stores 开头 ?
   "message"}}                            │是                    │否
  (_format_google_                       ▼                      ▼
   error_response:81)              404 改写成 400          OpenAIErrorResponse
                                    (OpenAI 客户端期望)     .from_message(...).to_dict()

三条要点:

• Google Interactions API(/v1alpha/interactions)走 Google 风格 {"error": {"code", "message"}},由 _is_interactions_path(server.py:89)判定。
• OpenAI 兼容的 Vector Stores 端点把很多"not found"当成 400 而非 404(server.py:118),因为该仓库里的集成测试和 OpenAI 客户端行为期望如此。
• 其余一律 OpenAI 风格错误体(OpenAIErrorResponse)。

translate_exception 里还有个实用细节:provider SDK 抛的异常(如 OpenAI 的 AuthenticationError 401、PermissionDeniedError 403)带 status_code 属性,会被原样保留,而不是一律压成 500(translation.py:43)。

---

4. 进程线:服务器怎么被拼起来

现在回到"开机装配"。入口是 create_app(server.py:358),它干六件事,顺序如下。

4.1 create_app 的六步

create_app()  (server.py:358)
  1. migrate_legacy_config_dir()            迁移旧配置目录
  2. 读 OGX_CONFIG → 打开 YAML                无此环境变量则直接报错
        └─ 配好日志、parse_and_maybe_upgrade_config → StackConfig
  3. app = StackApp(config)                  ← 关键:构造时就跑完 Stack.initialize(见 §4.2)
  4. add_middleware(...) × N                 按 §3 的规则搭中间件栈
  5. include_router(...)                     挂载自动发现的路由(见 §4.4)
  6. app.exception_handler(...)(...)          注册 §3.4 的错误处理器
  return app

第 4 步里决定"要服务哪些 API"的逻辑值得一提(server.py:443):若 config 显式列了 apis 就用它,否则用所有已解析出的实现;再对每个"自动路由"的 API 补上它对应的 routing-table API;最后无条件加上 admin、inspect、providers、prompts、conversations 这几个内置 API。

4.2 StackApp:在临时事件循环里把 provider 拼好

StackApp 是 FastAPI 的子类(server.py:126),它存在的理由写在类注释里:持有 Stack 实例,好让 lifespan 能启停后台任务。

它的构造函数里藏着一个非常关键、也很微妙的技巧:

# 真实源码:src/ogx/core/server/server.py:139
with concurrent.futures.ThreadPoolExecutor() as executor:
    future = executor.submit(asyncio.run, self.stack.initialize())
    future.result()

为什么要在一个临时线程里用临时事件循环跑 initialize? 因为路由注册需要先知道有哪些 provider/资源,而 Stack.initialize 是异步的;但此刻还没进入 uvicorn 的正式事件循环。于是先开一个一次性事件循环把 impls 拼出来。

代价与善后: 有些客户端(SQL 引擎、Google genai 的内部 httpx.AsyncClient)会急切地把自己绑到当时那个临时事件循环上;临时循环一结束它们就废了。所以构造函数随后要主动重置这些客户端(server.py:146 的 reset_sqlstore_engines,以及 server.py:159 起遍历 impls/impls_by_provider_id 调用各自的 _reset_client),让它们在 uvicorn 的正式循环里被懒重建。这段是"为什么这里要多写一堆 reset"的答案。

4.3 Stack.initialize:进程级装配的主线

Stack.initialize(stack.py:749)是把一个空 Stack 变成"能干活"的核心序列:

Stack.initialize()  (stack.py:749)
  1. 若 OGX_TEST_INFERENCE_MODE → 开启 API 录制上下文
  2. _initialize_storage(run_config)                        注册 kv_*/sql_* 存储后端(stack.py:718)
  3. create_dist_registry(stores.metadata, distro_name)     建分发注册表(见 06 章)
  4. add_internal_implementations(...)                      装 inspect/providers/admin/
                                                            prompts/conversations/connectors
  5. 先 initialize 内部 SQL 表(prompts/conversations/connectors)
        ← 必须在 resolve_impls 之前,否则共享引擎被别处先绑,后注册的表建不出来
  6. resolve_impls(...)                                     解析出全部 provider 实现(→ 02 章)
  7. register_resources(...)                                注册 models / vector_stores(stack.py:160)
  8. auto_register_tool_groups(...)                         按 tool_runtime provider 自动登记工具组
  9. register_connectors(...) + refresh_registry_once(...)
 10. validate_vector_stores_config(...)                     校验默认 embedding/reranker 模型存在
 11. self.impls = impls

其中第 6 步 resolve_impls(把 provider 规格变成活的实现、并装好自动路由)是本系列 02 章的主题,这里只当黑盒。

第 7 步 register_resources(stack.py:160)有个值得记的分支:配置里 provider_id: "all" 的 model 会用"第一个活跃 inference provider"注册一个无前缀别名(stack.py:180);__disabled__ 或空 provider_id 的资源直接跳过。

关于 _initialize_storage(stack.py:718): 它按后端类型名前缀把配置分成 kv_(键值)和 sql_ 两类,分别调用 register_kvstore_backends / register_sqlstore_backends,并把租户配置设为 SQL 层的默认——这是 06 章多租户隔离的地基。

4.4 环境变量替换:${env.X} 的三种语义

配置 YAML 在变成 StackConfig 之前,会被 replace_env_vars(stack.py:485)递归展开。它实现了一套类 bash 的语法(核心正则在 stack.py:585,替换逻辑在 get_env_var):

写法	名称	行为
${env.X}	必填	X 未设或为空 → 抛 EnvVarError(明确报错,不静默)
${env.X:=default}	默认值(操作符 =)	X 有值用 X,否则用 default;${env.X:=} 表示"没有就空串"
${env.X:+value}	条件值(操作符 +)	X 有值才代入 value;X 没值则得空串;${env.X:+} 恒为空串

替换完还会做类型归一(stack.py:643 的 _convert_string_to_proper_type):空串 → None,"true"/"false" → 布尔,能转就转 int/float,否则留字符串。值里的 ~ 会展开成家目录(os.path.expanduser)。

replace_env_vars 里还有两处"整块跳过"的特判,对读配置很重要:

• auth 关闭。 当 path == "server.auth" 且 provider_config.type 用条件语法解析成空时,把整个 provider_config 设为 None 从而关掉认证(stack.py:491)——这样就能用一个环境变量开关 auth,而不会因为其他 ${env.KEYCLOAK_URL} 这类裸变量没设而报错。
• 禁用的 provider / 空 ID 资源跳过。 列表里某项的 provider_id 解析成 "__disabled__",或 model_id/vector_store_id 这类 ID 字段(RESOURCE_ID_FIELDS)解析成空,则整条跳过、不再展开其内部配置,避免为被禁用的东西报缺变量的错(stack.py:534、stack.py:549)。

4.5 路由怎么被"自动发现"

create_app 不是手写一堆 include_router,而是靠 fastapi_router_registry.py 自动发现。

发现机制(fastapi_router_registry.py:37 的 _discover_router_factories):遍历 Api 枚举的每个值,尝试导入 ogx_api.<package>.fastapi_routes 模块并取出里面的 create_router 工厂函数。

两个细节:

• 枚举名 ≠ 包名的做特判:_API_TO_PACKAGE(fastapi_router_registry.py:31)把 inspect → inspect_api、tool_groups → tools。
• 缺模块是正常的(不是每个 API 都有路由,如 vector_stores、tool_runtime),用 importlib.util.find_spec 先探测、缺了就静默跳过;但导入报错会记 warning——把"没有路由"和"路由坏了"区分开。

发现结果缓存在模块级 _ROUTER_FACTORIES。create_app 随后对每个要服务的 API 调 build_fastapi_router(fastapi_router_registry.py:89)拿到 router 并 include_router。外部(第三方)API 还能通过 register_external_api_routers(fastapi_router_registry.py:71)注册自己的 create_router。

4.6 lifespan:开机与关机

lifespan(server.py:171)是 FastAPI 的生命周期钩子:

• 开机(yield 之前):app.stack.create_registry_refresh_task() 启动后台注册表刷新任务。
• 关机(yield 之后):await app.stack.shutdown() 优雅关闭。

后台刷新任务(stack.py:791 的 create_registry_refresh_task):用 asyncio.create_task 跑 refresh_registry_task(stack.py:859),它是个死循环——refresh_registry_once + asyncio.sleep(interval),间隔取自 config.server.registry_refresh_interval_seconds(默认 300 秒)。refresh_registry_once(stack.py:851)会挑出所有 CommonRoutingTableImpl 并调用它们的 refresh(),好让新上线/下线的模型被感知到。任务上还挂了 done 回调,把取消/异常/正常完成分别记日志。

shutdown(stack.py:811)则反向收尾:对每个 impl 调 shutdown()(带 5 秒超时保护)、退出录制上下文、取消刷新任务、关掉 kv/sql 存储后端。

---

5. 巧妙之处(可借鉴的技术)

• "后加先跑"被当成显式排序工具用。 认证三件套故意逆序添加以得到 Auth → Tenancy → RouteAuth 的运行顺序,并在注释里写清意图(server.py:417)。别人容易踩的坑,这里被当成特性。
• 临时事件循环装配 + 事后重置客户端。 用一次性事件循环跑异步 initialize 以完成同步的路由注册,再主动 reset 那些"急切绑定事件循环"的客户端让其懒重建(server.py:139 起)。这是"异步初始化撞上同步组装"这一经典难题的一种务实解法。
• 解压放线程池、并设上限 + 容错回退。 zstd 解压用 asyncio.to_thread 不阻塞事件循环、超过 100 MB 返回 413、失败则退回原始压缩体(server.py:298 起),兼顾性能、安全与健壮。
• 错误格式按调用方言分流。 同一个异常处理器,能给 Google Interactions、OpenAI、Vector Stores 三种不同的错误信封(server.py:110-123),让不同 SDK 的客户端都"感觉像原生服务"。
• env 替换里为"禁用项"整块短路。 被禁用的 provider / 空 ID 资源在展开阶段就跳过,避免为不需要的东西报缺变量错(stack.py:534、stack.py:549)——配置容错的好范式。

---

6. 边界与局限(本章范围内)

• provider 怎么被解析、自动路由怎么装配 —— 本章把 resolve_impls 当黑盒,细节见 02。
• 请求进入路由处理函数之后 的模型调用如何落到后端 —— 见 03。
• 多租户在存储层如何做到非旁路隔离 —— 本章只提到 set_default_tenancy_config 这一入口,机制见 06。
• 中间件是否加入取决于配置:无 auth 配置时认证/授权中间件根本不挂载(但 tenancy 可在无 auth 时单独启用,见 server.py:433);OGX_DISABLE_VERSION_CHECK 会摘掉版本校验。上面的全序图是"全部启用"下的形态。

---

7. 代码地图(导航索引)

主题	文件路径	符号名
服务器工厂/组装入口	src/ogx/core/server/server.py	create_app
FastAPI 子类,持有 Stack	src/ogx/core/server/server.py	StackApp
生命周期(启后台任务/优雅关机)	src/ogx/core/server/server.py	lifespan
统一 + 多 SDK 错误格式	src/ogx/core/server/server.py	global_exception_handler、_format_google_error_response、_is_interactions_path
客户端版本校验中间件	src/ogx/core/server/server.py	ClientVersionMiddleware
provider 数据/测试上下文中间件	src/ogx/core/server/server.py	ProviderDataMiddleware
zstd 请求体解压中间件	src/ogx/core/server/server.py	ZstdDecompressionMiddleware
异常翻译(状态码/校验/provider SDK)	src/ogx/core/exceptions/translation.py	translate_exception
认证中间件	src/ogx/core/server/auth.py	AuthenticationMiddleware
租户强制中间件	src/ogx/core/server/auth.py	TenancyMiddleware
路由级授权中间件	src/ogx/core/server/auth.py	RouteAuthorizationMiddleware、_is_route_allowed、_route_matches
请求指标中间件	src/ogx/core/server/metrics.py	RequestMetricsMiddleware、build_route_to_api_map
路由自动发现	src/ogx/core/server/fastapi_router_registry.py	_discover_router_factories、build_fastapi_router、register_external_api_routers
进程级装配主线	src/ogx/core/stack.py	Stack.initialize
存储后端注册	src/ogx/core/stack.py	_initialize_storage
内部实现装配	src/ogx/core/stack.py	add_internal_implementations
资源注册(models/vector_stores)	src/ogx/core/stack.py	register_resources
环境变量替换语义	src/ogx/core/stack.py	replace_env_vars、get_env_var、_convert_string_to_proper_type
后台注册表刷新	src/ogx/core/stack.py	create_registry_refresh_task、refresh_registry_task、refresh_registry_once
uvicorn factory 启动	src/ogx/cli/stack/run.py	_uvicorn_run