Files
fastapi/backend/app/core/config.py
T
2026-06-01 09:25:26 +08:00

109 lines
5.8 KiB
Python

import os
from pathlib import Path
from typing import Any
from pydantic import BaseModel, Field
def _load_dotenv_file() -> None:
"""环境加载:轻量读取项目根目录 `.env`,避免强依赖 python-dotenv。"""
env_path = Path(__file__).resolve().parents[3] / ".env"
if not env_path.exists():
return
for line in env_path.read_text(encoding="utf-8").splitlines():
if not line or line.strip().startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
os.environ.setdefault(key.strip(), value.strip())
_load_dotenv_file()
def _env_first(*keys: str, default: str = "") -> str:
"""环境读取:按优先级读取多个环境变量。"""
for key in keys:
value = os.getenv(key)
if value:
return value
return default
def _normalize_sync_database_url(url: str) -> str:
"""数据库连接:将异步 MySQL URL 转换为当前同步 ORM 可用的 URL。"""
if url.startswith("mysql+aiomysql://"):
return url.replace("mysql+aiomysql://", "mysql+pymysql://", 1)
if url.startswith("mysql://"):
return url.replace("mysql://", "mysql+pymysql://", 1)
return url
class Settings(BaseModel):
"""系统配置:集中管理数据库、DeepSeek、报告和短期 memory 配置。"""
app_name: str = Field(default_factory=lambda: os.getenv("APP_NAME", "Medical Consultation Agent Demo"))
app_env: str = Field(default_factory=lambda: os.getenv("APP_ENV", "local"))
app_debug: bool = Field(default_factory=lambda: os.getenv("APP_DEBUG", "true").lower() == "true")
api_v1_prefix: str = Field(default_factory=lambda: os.getenv("API_V1_PREFIX", "/api/v1"))
mysql_url: str = Field(default_factory=lambda: os.getenv("MYSQL_URL", ""))
database_url: str = Field(
default_factory=lambda: _normalize_sync_database_url(
_env_first("DATABASE_URL", "MYSQL_URL", default="sqlite:///./storage/demo.db")
)
)
llm_api_key: str = Field(default_factory=lambda: _env_first("LLM_API_KEY", "DEEPSEEK_API_KEY", default=""))
llm_base_url: str = Field(
default_factory=lambda: _env_first("LLM_BASE_URL", "DEEPSEEK_BASE_URL", default="https://api.deepseek.com")
)
llm_model: str = Field(default_factory=lambda: _env_first("LLM_MODEL", "DEEPSEEK_FAST_MODEL", default="deepseek-chat"))
llm_fast_model: str = Field(default_factory=lambda: _env_first("LLM_FAST_MODEL", "LLM_MODEL", "DEEPSEEK_FAST_MODEL", default="deepseek-chat"))
llm_reason_model: str = Field(
default_factory=lambda: _env_first("LLM_REASON_MODEL", "LLM_MODEL", "DEEPSEEK_REASON_MODEL", default="deepseek-reasoner")
)
llm_timeout_seconds: int = Field(default_factory=lambda: int(os.getenv("LLM_TIMEOUT_SECONDS", "45")))
llm_chat_timeout_seconds: int = Field(default_factory=lambda: int(os.getenv("LLM_CHAT_TIMEOUT_SECONDS", "20")))
llm_stream_first_token_timeout_seconds: int = Field(
default_factory=lambda: int(os.getenv("LLM_STREAM_FIRST_TOKEN_TIMEOUT_SECONDS", "15"))
)
llm_stream_total_timeout_seconds: int = Field(default_factory=lambda: int(os.getenv("LLM_STREAM_TOTAL_TIMEOUT_SECONDS", "45")))
llm_stream_enabled: bool = Field(default_factory=lambda: os.getenv("LLM_STREAM_ENABLED", "true").lower() == "true")
llm_mock_enabled: bool = Field(default_factory=lambda: os.getenv("LLM_MOCK_ENABLED", "true").lower() == "true")
llm_fallback_to_mock: bool = Field(default_factory=lambda: os.getenv("LLM_FALLBACK_TO_MOCK", "true").lower() == "true")
llm_fast_thinking_enabled: bool = Field(default_factory=lambda: os.getenv("LLM_FAST_THINKING_ENABLED", "false").lower() == "true")
llm_reason_thinking_enabled: bool = Field(default_factory=lambda: os.getenv("LLM_REASON_THINKING_ENABLED", "false").lower() == "true")
llm_reasoning_effort: str = Field(default_factory=lambda: os.getenv("LLM_REASONING_EFFORT", "low"))
llm_fast_max_tokens: int = Field(default_factory=lambda: int(os.getenv("LLM_FAST_MAX_TOKENS", "512")))
llm_hint_max_tokens: int = Field(default_factory=lambda: int(os.getenv("LLM_HINT_MAX_TOKENS", "1200")))
llm_scoring_json_response: bool = Field(default_factory=lambda: os.getenv("LLM_SCORING_JSON_RESPONSE", "true").lower() == "true")
llm_scoring_max_tokens: int = Field(default_factory=lambda: int(os.getenv("LLM_SCORING_MAX_TOKENS", "4096")))
report_storage_dir: str = Field(default_factory=lambda: os.getenv("REPORT_STORAGE_DIR", "./storage/reports"))
runtime_memory_ttl_seconds: int = Field(default_factory=lambda: int(os.getenv("RUNTIME_MEMORY_TTL_SECONDS", "7200")))
runtime_memory_backend: str = Field(default_factory=lambda: os.getenv("RUNTIME_MEMORY_BACKEND", "memory"))
redis_url: str = Field(default_factory=lambda: os.getenv("REDIS_URL", "redis://127.0.0.1:6379/0"))
def as_public_dict(self) -> dict[str, Any]:
"""配置展示:返回允许暴露给 Demo 前端的功能开关。"""
mock_enabled = self.llm_mock_enabled or not self.llm_api_key
return {
"stream_chat": self.llm_stream_enabled,
"score_types": ["percentage", "five_point"],
"pdf_export": True,
"knowledge_search": True,
"llm_mock_enabled": mock_enabled,
"llm_mode": "mock" if mock_enabled else "real",
"llm_fallback_to_mock": self.llm_fallback_to_mock,
"llm_fast_model": self.llm_fast_model,
"llm_reason_model": self.llm_reason_model,
"llm_fast_thinking_enabled": self.llm_fast_thinking_enabled,
"llm_reason_thinking_enabled": self.llm_reason_thinking_enabled,
"llm_reasoning_effort": self.llm_reasoning_effort,
"llm_fast_max_tokens": self.llm_fast_max_tokens,
"runtime_memory_backend": self.runtime_memory_backend,
}
settings = Settings()