"""Composition root + FastAPI dependencies. This is the only place that wires concrete adapters into the domain packages. Everything below it takes its collaborators as constructor arguments, which is what keeps ``packages/domain`` and ``packages/llm`` unit-testable without IO. """ from __future__ import annotations from collections.abc import AsyncIterator from decimal import Decimal from typing import Annotated from fastapi import Depends, Request from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy.ext.asyncio import AsyncSession from packages.agent.context import PROFILE_SUMMARY_MAX_CHARS from packages.agent.service import AgentService from packages.analytics.service import AnalyticsService from packages.billing.service import BillingService from packages.identity.service import IdentityService from packages.llm.budget import DailyBudgetGuard, RedisCostCounter from packages.llm.circuit import CircuitBreakerRegistry from packages.llm.cost import ( CompositeAICallRecorder, ModelRate, PeakAwareRateTable, RateTable, ) from packages.llm.gateway import LLMGateway from packages.llm.prompts import PromptRegistry from packages.llm.providers.base import LLMProvider from packages.llm.providers.deepseek import DeepSeekProvider from packages.llm.providers.zhipu import ZhipuGLMProvider from packages.llm.recording import PrometheusAICallRecorder, SqlAlchemyAICallRecorder from packages.llm.routing import SceneRouter, build_scene_configs from packages.report.profile import latest_profile_summary from packages.shared.config import Settings, get_settings from packages.shared.context import bind_context from packages.shared.db import get_session_factory from packages.shared.errors import AppError, ErrorCode from packages.shared.redis_client import get_redis from packages.shared.security import decode_access_token from packages.storage.service import StorageService from packages.task.service import TaskService bearer_scheme = HTTPBearer(auto_error=False) def build_llm_gateway(settings: Settings) -> LLMGateway: """Assemble the gateway from configuration (架构文档 §8.1).""" providers: dict[str, LLMProvider] = { "deepseek": DeepSeekProvider( api_key=settings.deepseek_api_key, base_url=settings.deepseek_base_url, default_model=settings.deepseek_model, ), "zhipu": ZhipuGLMProvider( api_key=settings.zhipu_api_key, base_url=settings.zhipu_base_url, default_model=settings.zhipu_model, ), } router = SceneRouter( build_scene_configs( primary_provider=settings.llm_primary_provider, backup_provider=settings.llm_backup_provider, structured_max_retries=settings.llm_structured_max_retries, overrides=settings.llm_scene_overrides, ) ) default_rate = ModelRate.from_mapping(settings.llm_default_cost_rate) rates = PeakAwareRateTable( RateTable( { model: ModelRate.from_mapping(rate) for model, rate in settings.llm_cost_rates.items() }, default=default_rate, ), RateTable( { model: ModelRate.from_mapping(rate) for model, rate in settings.llm_cost_rates_peak.items() }, default=default_rate, ) if settings.llm_cost_rates_peak else None, enabled=settings.llm_peak_pricing_enabled, ) recorder = CompositeAICallRecorder( SqlAlchemyAICallRecorder(get_session_factory()), PrometheusAICallRecorder(), ) budget = DailyBudgetGuard( RedisCostCounter(get_redis()), daily_budget_cents=Decimal(settings.llm_daily_budget_cents), enabled=settings.llm_daily_budget_enabled, tz_offset_hours=settings.llm_budget_timezone_offset_hours, tiers=list(settings.llm_budget_tiers), ) return LLMGateway( providers=providers, router=router, prompts=PromptRegistry( settings.prompts_dir, cache_prefix_mode=settings.llm_cache_prefix_mode ), rates=rates, recorder=recorder, budget=budget, breakers=CircuitBreakerRegistry( failure_threshold=settings.llm_circuit_failure_threshold, cooldown_seconds=settings.llm_circuit_cooldown_seconds, ), ) # --- request-scoped dependencies ------------------------------------------- def settings_dep() -> Settings: return get_settings() async def session_dep() -> AsyncIterator[AsyncSession]: factory = get_session_factory() async with factory() as session: try: yield session await session.commit() except Exception: await session.rollback() raise def gateway_dep(request: Request) -> LLMGateway: # app.state 是无类型的属性袋,取出来必须显式收窄 gateway: LLMGateway | None = getattr(request.app.state, "llm_gateway", None) if gateway is None: # pragma: no cover - startup guarantees this raise AppError(ErrorCode.CONFIG_ERROR, message="LLM 网关未初始化") return gateway SettingsDep = Annotated[Settings, Depends(settings_dep)] SessionDep = Annotated[AsyncSession, Depends(session_dep)] GatewayDep = Annotated[LLMGateway, Depends(gateway_dep)] async def current_user_id( credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)], settings: SettingsDep, ) -> int: """架构文档 §9.1: ``Authorization: Bearer ``.""" if credentials is None or not credentials.credentials: raise AppError(ErrorCode.UNAUTHENTICATED) claims = decode_access_token( credentials.credentials, secret=settings.jwt_secret, algorithm=settings.jwt_algorithm, issuer=settings.jwt_issuer, ) bind_context(user_id=claims.user_id) return claims.user_id CurrentUserId = Annotated[int, Depends(current_user_id)] def identity_service_dep(session: SessionDep) -> IdentityService: return IdentityService(session) def billing_service_dep(session: SessionDep, settings: SettingsDep) -> BillingService: return BillingService(session, tz_offset_hours=settings.llm_budget_timezone_offset_hours) def task_service_dep(session: SessionDep) -> TaskService: return TaskService(session) def analytics_service_dep(session: SessionDep) -> AnalyticsService: return AnalyticsService(session) def storage_service_dep(settings: SettingsDep) -> StorageService: return StorageService(settings) def budget_guard_dep(settings: SettingsDep) -> DailyBudgetGuard: """全局日预算的场景闸(架构 §13.3):hd_export 等可降级场景在请求入口先过它。 每次请求新建实例是刻意的:Guard 本身无状态(水位读 Redis), 与 build_llm_gateway 里给网关用的那只互不影响。 """ return DailyBudgetGuard( RedisCostCounter(get_redis()), daily_budget_cents=Decimal(settings.llm_daily_budget_cents), enabled=settings.llm_daily_budget_enabled, tz_offset_hours=settings.llm_budget_timezone_offset_hours, tiers=list(settings.llm_budget_tiers), ) def agent_service_dep( session: SessionDep, gateway: GatewayDep, billing: Annotated[BillingService, Depends(billing_service_dep)], settings: SettingsDep, ) -> AgentService: # agent 与 report 在 import-linter 分层里同层、不允许直接互调,画像摘要 # 以闭包在 apps 层接线;report 域内部仍只有 report.repo 触 rp_report。 async def profile_provider(user_id: int) -> str | None: return await latest_profile_summary( session, user_id=user_id, max_chars=PROFILE_SUMMARY_MAX_CHARS ) return AgentService( session, gateway=gateway, billing=billing, profile_provider=profile_provider, history_turns=settings.agent_history_turns, ) IdentityServiceDep = Annotated[IdentityService, Depends(identity_service_dep)] BillingServiceDep = Annotated[BillingService, Depends(billing_service_dep)] TaskServiceDep = Annotated[TaskService, Depends(task_service_dep)] AnalyticsServiceDep = Annotated[AnalyticsService, Depends(analytics_service_dep)] StorageServiceDep = Annotated[StorageService, Depends(storage_service_dep)] BudgetGuardDep = Annotated[DailyBudgetGuard, Depends(budget_guard_dep)] AgentServiceDep = Annotated[AgentService, Depends(agent_service_dep)]