§ · Nested detail page

路由与运行时限流深度解析

拆解 select_model、fallback candidate、ProviderRuntime.admit 和 complete/stream 的并发边界。

§ 00 · IMPORT CHAIN

导入链与契约

▸入口契约

这条链路从 HTTP 入口开始：sources/my-free-code/my_free_code/api/routes.py:49 接收 Anthropic Messages，sources/my-free-code/my_free_code/api/routes.py:51 先做代理鉴权，sources/my-free-code/my_free_code/api/routes.py:52 调用模型选择，再把非流式和流式分开。入口函数本身不认识 provider 细节，这是后续可维护性的关键。

sources/my-free-code/my_free_code/api/routes.py:49 - sources/my-free-code/my_free_code/api/routes.py:67Python

49@router.post("/v1/messages")
50async def messages(body: dict, authorization: str | None = Header(None)):
51    require_auth(authorization)
52    selected = select_model(body.get("model"), settings)
53
54    if not body.get("stream"):
55        last = None
56        for model in candidates(selected, settings):
57            try:
58                return await runtime.complete(body, model)
59            except Exception as exc:
60                last = exc
61        raise HTTPException(status_code=502, detail=str(last))
62
63    return StreamingResponse(
64        stream_messages(body, selected),
65        media_type="text/event-stream",
66        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
67    )

§ 01 · CORE IMPL

核心实现，逐段拆

▸模型别名和 fallback 候选

路由器的设计重点是“显式优先，别名其次”。sources/my-free-code/my_free_code/core/router.py:5 到 sources/my-free-code/my_free_code/core/router.py:6 保护 provider/model 直连模型，sources/my-free-code/my_free_code/core/router.py:7 到 sources/my-free-code/my_free_code/core/router.py:15 才处理 Claude 档位别名，sources/my-free-code/my_free_code/core/router.py:17 到 sources/my-free-code/my_free_code/core/router.py:18 让 fallback 列表去重保序。

sources/my-free-code/my_free_code/core/router.py:3 - sources/my-free-code/my_free_code/core/router.py:24Python

3def select_model(requested: str | None, s: Settings) -> str:
4    value = (requested or "").lower()
5    if "/" in value:
6        return requested or s.model
7    if "opus" in value and s.opus:
8        return s.opus
9    if "sonnet" in value and s.sonnet:
10        return s.sonnet
11    if "haiku" in value and s.haiku:
12        return s.haiku
13    if "fable" in value and s.fable:
14        return s.fable
15    return s.model
16
17def candidates(primary: str, s: Settings) -> list[str]:
18    return list(dict.fromkeys([primary, *s.fallbacks]))
19
20def split_model(model: str) -> tuple[str, str]:
21    provider, sep, remote = model.partition("/")
22    if not sep or not provider or not remote:
23        raise ValueError("model must have the form provider/model")
24    return provider, remote

▸ProviderRuntime 的限流状态

ProviderRuntime 把每个 provider 当成独立资源池。sources/my-free-code/my_free_code/providers/runtime.py:13 建并发 semaphore，sources/my-free-code/my_free_code/providers/runtime.py:33 到 sources/my-free-code/my_free_code/providers/runtime.py:43 用 monotonic 时间实现窗口限流，sources/my-free-code/my_free_code/providers/runtime.py:35 到 sources/my-free-code/my_free_code/providers/runtime.py:36 阻断处于恢复 backoff 的 provider。

sources/my-free-code/my_free_code/providers/runtime.py:10 - sources/my-free-code/my_free_code/providers/runtime.py:43Python

10class ProviderRuntime:
11    def __init__(self, settings: Settings):
12        self.settings = settings
13        self.semaphores = {p.id: asyncio.Semaphore(settings.max_concurrency) for p in PROVIDER_MAP.values()}
14        self.calls = {p.id: deque() for p in PROVIDER_MAP.values()}
15        self.unhealthy_until = {}
16
17    def context(self, provider_id, remote_model):
18        p = PROVIDER_MAP[provider_id]
19        base = p.base_url
20        if provider_id == "ollama":
21            base = self.settings.key("OLLAMA_BASE_URL") or base
22        elif provider_id == "lmstudio":
23            base = self.settings.key("LM_STUDIO_BASE_URL") or base
24        elif provider_id == "llamacpp":
25            base = self.settings.key("LLAMACPP_BASE_URL") or base
26        if not base:
27            raise RuntimeError(f"{p.name} requires a specialized adapter")
28        return ProviderContext(provider_id, remote_model, self.settings.key(p.key), base, self.settings.http_timeout)
29
30    def adapter(self, provider_id):
31        return SPECIALIZED.get(provider_id, OpenAIChatAdapter)()
32
33    async def admit(self, provider_id):
34        now = time.monotonic()
35        if now < self.unhealthy_until.get(provider_id, 0):
36            raise RuntimeError(f"provider {provider_id} is in recovery backoff")
37        q = self.calls[provider_id]
38        while q and now - q[0] > self.settings.rate_window:
39            q.popleft()
40        if len(q) >= self.settings.rate_limit:
41            await asyncio.sleep(max(0.01, self.settings.rate_window - (now - q[0])))
42        q.append(time.monotonic())
43        return self.semaphores[provider_id]

§ 02 · KEY FLOW

关键流程

request.model -> select_model -> candidates -> ProviderRuntime.admit -> adapter.complete/stream -> unhealthy_until on failure -> next fallback

完整请求进入 runtime 后，sources/my-free-code/my_free_code/providers/runtime.py:45 到 sources/my-free-code/my_free_code/providers/runtime.py:56 负责非流式 complete，sources/my-free-code/my_free_code/providers/runtime.py:58 到 sources/my-free-code/my_free_code/providers/runtime.py:73 负责 stream。两个分支都先 admit，再 adapter，再 context，这让限流、鉴权和协议转换保持同一条顺序。

§ 03 · EDGE CASES

边界 / 失败模式

sources/my-free-code/my_free_code/providers/runtime.py:40 只在队列长度达到 limit 后 sleep；突发流量会在 coroutine 层等待，而不是直接返回 429。
sources/my-free-code/my_free_code/api/routes.py:103 到 sources/my-free-code/my_free_code/api/routes.py:106 说明流式输出一旦 started，就不会继续 fallback，避免混合两个模型的 token。
sources/my-free-code/my_free_code/core/failures.py:14 到 sources/my-free-code/my_free_code/core/failures.py:22 的 retryable 分类尚未被 adapter 使用，错误语义还有断点。

Sub-page · 2026-08-31 Source · hkqr/my-free-code ↑ my-free-code 主页面

路由与运行时限流 深度解析