1. 为什么 LLM 网关绕不开依赖注入如果你正在用 FastAPI 搭一个本地 Claude 代理服务大概率会遇到这样的场景路由函数里直接import配置模块用全局变量持有 Provider 实例HTTP 客户端在模块加载时就被初始化。写起来很快但一到单元测试就崩了——你没法在不发真实网络请求的前提下验证路由逻辑没法给不同测试用例塞不同的 API Key更没法在并发测试里隔离 Provider 状态。Free-Claude-Code 这个项目要处理的问题更棘手。它需要同时管理 NVIDIA NIM、OpenRouter、DeepSeek、Ollama 等多个上游 Provider 的生命周期处理请求级的模型路由、Token 计数、流式响应编排还要在测试环境里快速替换掉真实的外部依赖。这些需求堆在一起如果没有一套清晰的依赖注入Dependency InjectionDI体系代码会迅速退化成难以维护的意大利面条。FastAPI 的Depends()恰好提供了三个关键能力声明式依赖树让路由只声明我需要什么请求级缓存保证同一请求内同一依赖只执行一次运行时覆盖允许在测试中无缝替换任何依赖。Free-Claude-Code 把这三点用到了极致从最简单的get_settings开始一步步搭起了整个服务编排体系。这篇文章会从源码层面拆解它的 DI 设计给出可复制的依赖注入骨架、测试替身注入配置和验证动作。同时说明如何通过统一的 Key/API 通道接入 TaoToken让本地代理服务有一个稳定的上游入口。适合正在搭建本地代理服务、或者想理解 FastAPI 服务解耦与可测性设计的开发者。2. TaoToken 前置统一 Key 与 API 通道在动手写 DI 骨架之前先把上游通道准备好。Free-Claude-Code 这类代理服务的核心价值在于统一入口、多 Provider 路由而 TaoToken 提供的正是这样一个统一 Key/API 通道让本地代理不必为每个上游单独维护认证逻辑。你需要先拿到一个可用的 API Key。访问控制台创建密钥https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite创建完成后在 API Keys 页面可以查看和管理你的密钥https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite接入文档里有完整的端点说明和参数格式建议先过一遍https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewriteAPI 基础地址是https://taotoken.net/api这个地址在后面的配置里会用到。拿到 Key 之后把它写进项目的.env文件不要硬编码在代码里。Free-Claude-Code 的Settings类会从环境变量加载配置所以你的.env大概长这样# .env ANTHROPIC_AUTH_TOKENsk-your-taotoken-key PROVIDER_TYPEopenai_compat OPENAI_COMPAT_BASE_URLhttps://taotoken.net/api OPENAI_COMPAT_API_KEYsk-your-taotoken-key HTTP_READ_TIMEOUT600.0 HTTP_WRITE_TIMEOUT20.0 HTTP_CONNECT_TIMEOUT5.0这里有个细节值得注意ANTHROPIC_AUTH_TOKEN是服务端用来校验客户端请求的而OPENAI_COMPAT_API_KEY是服务端向上游发起请求时用的。两者可以相同也可以不同取决于你的部署场景。如果你只是本地自用用同一个 Key 最省事。提示如果你打算长期跑编码类任务或 Agent 工作流可以了解一下 Coding Plan它在长会话场景下的配额策略更友好https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite3. 可复制配置依赖注入骨架与 ClaudeProxyService 编排这一章是全文的核心我会给出可以直接抄进项目的 DI 骨架。整个体系分三层基础依赖配置、认证、资源解析Provider 缓存、服务编排ClaudeProxyService。3.1 基础依赖get_settings 与 require_api_key配置入口放在config/settings.py用 Pydantic Settings 加载环境变量并用lru_cache保证进程内只实例化一次# config/settings.py from functools import lru_cache from pydantic_settings import BaseSettings class Settings(BaseSettings): anthropic_auth_token: str provider_type: str openai_compat openai_compat_base_url: str https://taotoken.net/api openai_compat_api_key: str http_read_timeout: float 600.0 http_write_timeout: float 20.0 http_connect_timeout: float 5.0 log_api_error_tracebacks: bool False class Config: env_file .env lru_cache def get_settings() - Settings: return Settings()然后在api/dependencies.py里做一个 FastAPI 友好的桥接避免路由层直接引用配置模块# api/dependencies.py from config.settings import Settings from config.settings import get_settings as _get_settings def get_settings() - Settings: return _get_settings()认证依赖require_api_key是最典型的横切关注点。它同时依赖Request和Settings支持三种认证头并且在未配置 Token 时自动放行# api/dependencies.py import secrets from fastapi import Depends, HTTPException, Request def require_api_key( request: Request, settings: Settings Depends(get_settings), ) - None: anthropic_auth_token settings.anthropic_auth_token if not anthropic_auth_token: return header ( request.headers.get(x-api-key) or request.headers.get(authorization) or request.headers.get(anthropic-auth-token) ) if not header: raise HTTPException(status_code401, detailMissing API key) token header if header.lower().startswith(bearer ): token header.split( , 1)[1] if : in token: token token.split(:, 1)[0] if not secrets.compare_digest( token.encode(utf-8), anthropic_auth_token.encode(utf-8), ): raise HTTPException(status_code401, detailInvalid API key)这里用secrets.compare_digest而不是普通字符串比较是为了消除时序侧信道风险。这个细节在自建代理服务时经常被忽略但成本极低建议保留。3.2 双层 Provider 缓存进程级与应用级Free-Claude-Code 最核心的设计之一是 Provider 解析的双层架构。原因很简单单元测试里没有运行中的 FastAPI 应用实例也没有 HTTP 请求上下文此时需要一个进程全局的缓存而生产环境通过 Lifespan 管理生命周期应用状态保存在app.state中如果所有请求共享进程级缓存会导致不同应用实例之间状态污染。进程级缓存是一个模块级字典配合三个工具函数# api/dependencies.py from providers.base import BaseProvider from providers.registry import ProviderRegistry _providers: dict[str, BaseProvider] {} def get_provider_for_type(provider_type: str) - BaseProvider: return resolve_provider(provider_type, appNone, settingsget_settings()) def get_provider() - BaseProvider: return get_provider_for_type(get_settings().provider_type) async def cleanup_provider(): global _providers await ProviderRegistry(_providers).cleanup() _providers {}生产环境的核心是resolve_provider它显式接受app参数不通过隐式全局变量获取应用实例# api/dependencies.py from starlette.applications import Starlette def resolve_provider( provider_type: str, *, app: Starlette | None, settings: Settings, ) - BaseProvider: if app is not None: reg getattr(app.state, provider_registry, None) if reg is None: raise RuntimeError( Provider registry is not configured. Ensure AppRuntime startup ran or assign app.state.provider_registry for test apps. ) return _resolve_with_registry(reg, provider_type, settings) return _resolve_with_registry(ProviderRegistry(_providers), provider_type, settings)ProviderRegistry本身是一个轻量的懒加载缓存管理器# providers/registry.py from collections.abc import MutableMapping from providers.base import BaseProvider from providers.factory import create_provider class ProviderRegistry: def __init__(self, providers: MutableMapping[str, BaseProvider] | None None): self._providers providers if providers is not None else {} def is_cached(self, provider_id: str) - bool: return provider_id in self._providers def get(self, provider_id: str, settings) - BaseProvider: if provider_id not in self._providers: self._providers[provider_id] create_provider(provider_id, settings) return self._providers[provider_id] async def cleanup(self): for provider in self._providers.values(): await provider.aclose() self._providers.clear()懒加载策略确保 Provider 只在首次使用时才初始化对于拥有多个 Provider 的应用来说可以显著减少启动时间和不必要的资源占用。3.3 ClaudeProxyService构造函数注入为什么ClaudeProxyService不直接作为依赖函数返回而是让get_proxy_service来组装因为 Service 的构造需要请求级上下文——它需要一个ProviderGetter而这个 getter 必须绑定到当前请求的request.app才能使用应用级的ProviderRegistry。# api/services.py from collections.abc import Callable from typing import Any from config.settings import Settings from core.anthropic import get_token_count from providers.base import BaseProvider TokenCounter Callable[[list[Any], str | list[Any] | None, list[Any] | None], int] ProviderGetter Callable[[str], BaseProvider] class ClaudeProxyService: def __init__( self, settings: Settings, provider_getter: ProviderGetter, model_router: ModelRouter | None None, token_counter: TokenCounter get_token_count, ): self._settings settings self._provider_getter provider_getter self._model_router model_router or ModelRouter(settings) self._token_counter token_counter def create_message(self, request_data): try: routed self._model_router.resolve_messages_request(request_data) provider self._provider_getter(routed.resolved.provider_id) input_tokens self._token_counter( request_data.messages, None, None ) return provider.stream_response( routed.request, input_tokensinput_tokens ) except ProviderError: raise except Exception as e: raise HTTPException(status_code500, detailInternal error) from e这里体现了几个设计原则依赖倒置Service 不直接导入resolve_provider而是依赖抽象的ProviderGetter、默认参数即文档token_counter的默认值明确了接口契约、可选依赖有兜底model_router为 None 时自动创建默认实例。3.4 get_proxy_service依赖的依赖路由层的get_proxy_service负责把 FastAPI 的依赖系统和 Service 的构造函数对接起来# api/routes.py from fastapi import Depends, Request from api.dependencies import get_settings, resolve_provider from api.services import ClaudeProxyService from core.anthropic import get_token_count def get_proxy_service( request: Request, settings: Settings Depends(get_settings), ) - ClaudeProxyService: return ClaudeProxyService( settings, provider_getterlambda provider_type: resolve_provider( provider_type, apprequest.app, settingssettings, ), token_counterget_token_count, ) router.post(/v1/messages) async def create_message( request_data: MessagesRequest, service: ClaudeProxyService Depends(get_proxy_service), _authDepends(require_api_key), ): return service.create_message(request_data)这段代码是整个 DI 体系里最精妙的环节request: Request作为隐式依赖由 FastAPI 自动注入settings通过嵌套依赖解析同一请求内结果被缓存闭包捕获request.app把请求级上下文绑定到 Provider 解析逻辑中而ClaudeProxyService本身是纯业务类不知道 FastAPI 的存在可以单独单元测试。4. 验证请求从启动到流式响应配置写完了接下来验证整条链路是否跑通。启动服务uvicorn api.app:app --host 0.0.0.0 --port 8000 --reload启动日志里应该能看到 Provider 注册表初始化和模型校验的信息。如果app.state.provider_registry没有正确发布resolve_provider会抛出明确的错误信息提示你检查AppRuntime的启动流程。4.1 用 curl 验证认证与路由先测认证。不带任何认证头请求/v1/messages应该返回 401curl -s -o /dev/null -w %{http_code}\n \ -X POST http://localhost:8000/v1/messages \ -H Content-Type: application/json \ -d {model:claude-3-sonnet,messages:[{role:user,content:hi}],max_tokens:10}带上正确的x-api-key再试一次应该返回 200 并开始流式输出curl -N -X POST http://localhost:8000/v1/messages \ -H Content-Type: application/json \ -H x-api-key: sk-your-taotoken-key \ -d { model: claude-3-sonnet, messages: [{role: user, content: 用一句话解释依赖注入}], max_tokens: 100, stream: true }如果上游通道配置正确你会看到 SSE 格式的流式响应以event: message_start开头以[DONE]结尾。这说明从客户端认证、依赖解析、Provider 获取到上游请求的整条链路都通了。4.2 用模型对话页面快速验证如果你不想写 curl也可以直接在模型对话页面验证 Key 和通道是否可用https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite在页面里选一个模型发一条消息能正常返回就说明 Key 和 API 通道没问题。这一步可以帮你快速区分是上游通道的问题还是本地代理代码的问题。4.3 验证依赖覆盖是否生效在测试环境里app.dependency_overrides是验证 DI 设计是否正确的关键。写一个最小测试# tests/test_di_smoke.py from fastapi.testclient import TestClient from api.app import create_app from api.dependencies import get_settings from config.settings import Settings app create_app() def test_auth_override(): client TestClient(app) settings Settings() settings.anthropic_auth_token s3cr3t app.dependency_overrides[get_settings] lambda: settings payload {model: claude-3-sonnet, messages: [{role: user, content: hello}]} r client.post(/v1/messages/count_tokens, jsonpayload) assert r.status_code 401 r client.post( /v1/messages/count_tokens, jsonpayload, headers{X-API-Key: s3cr3t}, ) assert r.status_code 200 app.dependency_overrides.clear()运行pytest tests/test_di_smoke.py -v两个断言都通过说明依赖覆盖机制工作正常。这个测试不需要任何真实网络请求完全在内存中完成。5. 本篇常见错排查5.1 Provider registry is not configured这是最常见的报错通常出现在测试环境或手动构造TestClient时。原因是app.state.provider_registry没有被设置。生产环境由AppRuntime.startup()负责发布测试环境需要手动赋值from providers.registry import ProviderRegistry app.state.provider_registry ProviderRegistry()或者更简单的方式在测试 fixture 里直接 patch 掉resolve_providerfrom unittest.mock import patch, MagicMock with patch(api.dependencies.resolve_provider, return_valuemock_provider): with TestClient(app) as client: yield client5.2 依赖缓存导致的状态不更新FastAPI 在同一请求内会缓存依赖函数的返回值。如果你写了一个返回可变对象的依赖两个路由参数拿到的是同一个对象def get_counter(): return {count: 0} app.get(/demo) def demo(c1: dict Depends(get_counter), c2: dict Depends(get_counter)): c1[count] 1 return {c1: c1[count], c2: c2[count]} # 返回 {c1: 1, c2: 1}如果你需要每次调用都获得新实例应该使用类依赖或工厂模式而不是返回可变字典。5.3 进程级缓存污染测试_providers是模块级全局变量测试之间会互相污染。解决方案是用autouseTrue的 fixture 在每个测试前后保存和恢复状态import pytest import api.dependencies pytest.fixture(autouseTrue) def reset_provider(): saved api.dependencies._providers api.dependencies._providers {} yield api.dependencies._providers saved5.4 异步依赖的传染性Free-Claude-Code 的所有依赖函数都是同步的这是有意为之。get_settings只涉及内存访问Provider 创建也是同步的。保持依赖同步可以简化测试避免async/await的传染性。如果你的依赖需要执行 IO 操作比如查数据库可以定义为async defFastAPI 会自动在事件循环中运行它们但要注意这会改变依赖的调用语义。5.5 循环依赖当依赖函数之间形成循环引用时FastAPI 会在运行时抛出异常。避免的方式是分层导入api/dependencies.py只导入接口类型BaseProvider不在模块级实例化任何 Provider工厂函数使用局部导入避免模块加载时的循环引用ClaudeProxyService依赖ProviderGetter类型别名而非具体函数。6. 长期编码与 Agent 场景的接入建议如果你只是偶尔用本地代理跑几条测试请求上面的配置已经够用了。但如果你打算把它作为长期编码助手或 Agent 工作流的底座有几个点值得提前考虑。第一是超时配置。编码类任务经常涉及长上下文和流式输出HTTP_READ_TIMEOUT建议设到 600 秒以上否则容易在生成到一半时被截断。第二是 Provider 的懒加载策略在多 Provider 场景下首次请求会有额外的初始化开销如果你对首字延迟敏感可以在AppRuntime.startup()里预热常用的 Provider。第三是配额管理。长期跑 Agent 工作流会消耗大量 Token建议用 Coding Plan 这类针对长会话优化的方案https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite如果你在接入过程中遇到认证或路由问题优先检查 API Keys 配置和接入文档里的端点说明https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite最后回到 DI 本身。Free-Claude-Code 的依赖注入体系之所以值得反复研读不是因为它用了什么高深技巧而是因为它把Depends()这个看似简单的语法糖用成了整个架构解耦的基石。双层 Provider 缓存解决了脚本测试与生产环境的不同需求构造函数注入让 Service 保持纯业务类的可测试性app.dependency_overrides让测试替身注入变得毫无侵入性。如果你正在构建需要对接多个第三方服务、对可测试性有高要求的 FastAPI 项目这套设计模式可以直接复用。