vent_agent_2.0/config.py

"""全局配置：.env 读取与类型化访问的唯一入口。

设计取舍：不用 pydantic-settings，只用 ``python-dotenv`` + 三个 ``get_*``
帮助函数。理由是骨架阶段配置项还在快速增删，引入 settings 模型会让每次
加一个开关都要改三处；这里改一处即可，且每个配置项都带默认值——默认值
必须等于「什么都不配也能跑起来」的那套（沿用 1.5 的工程纪律：默认 = 可用的
最小配置，任何新开关的默认值都不应改变既有行为）。

## 关于环境变量污染（踩过的坑，别改回去）

机器上若存在全局的 ``OPENAI_API_KEY``，``load_dotenv(override=False)`` 会让
**进程环境胜出、项目 .env 被静默忽略**。表现是：配好的密钥明明是对的，服务
却报 401，且报错里显示的密钥后几位和 .env 里的对不上——足以浪费半天。

所以这里的策略是：
1. **项目 .env 优先**（``override=True``）。.env 是本项目对外声明的配置面
   （有 .env.example 为证），比机器上某个无关的全局变量更具体。
2. 若没有 .env 文件，行为与常规一致：纯用进程环境变量（容器/CI 场景不受影响）。
3. 两者**都存在且值不同**时，记一条 WARNING 并把冲突键名暴露给 /health，
   让「我改了配置为什么不生效」有迹可循。
"""

from __future__ import annotations

import os
from pathlib import Path

from dotenv import dotenv_values, load_dotenv

# 项目根目录（本文件所在目录）。所有相对路径以此为基准解析，
# 避免「从哪个目录启动服务」影响资源定位。
ROOT = Path(__file__).resolve().parent

_ENV_PATH = ROOT / ".env"
# 测试/CI 用：设 DEEP_AGENT_SKIP_DOTENV=1 可完全跳过 .env，只认进程环境变量
# 与代码默认值。没有这个开关，测试就会受开发者本机 .env 的内容影响——
# 同一套测试在不同机器上跑出不同结果，是最难查的一类失败。
_SKIP_DOTENV = os.getenv("DEEP_AGENT_SKIP_DOTENV", "").strip().lower() in {"1", "true", "yes", "on"}

# 直接读文件（不经 os.environ），用于检测冲突
_FILE_VALUES: dict[str, str] = (
    {} if _SKIP_DOTENV else {k: v for k, v in dotenv_values(_ENV_PATH).items() if v}
)
# 冲突记录：键名 → 说明。供 log_config_sources() 与 /health 使用
CONFLICTS: dict[str, str] = {}

for _key, _file_value in _FILE_VALUES.items():
    _process_value = os.environ.get(_key)
    if _process_value and _process_value != _file_value:
        CONFLICTS[_key] = "项目 .env 覆盖了同名进程环境变量"

# .env 优先：见模块 docstring 的「环境变量污染」一节
if not _SKIP_DOTENV:
    load_dotenv(_ENV_PATH, override=True)


def _mask(value: str) -> str:
    """给日志用的脱敏显示：只留前 6 后 4 位。"""
    if not value:
        return "(空)"
    if len(value) <= 12:
        return "*" * len(value)
    return f"{value[:6]}...{value[-4:]}"


def log_config_sources() -> None:
    """把配置来源与冲突写进日志。须在 setup_logging() 之后调用。"""
    from obs import get_logger

    logger = get_logger("config")
    if _ENV_PATH.exists():
        logger.info("配置来源：%s（%d 项，其中 %d 项覆盖了同名进程环境变量）",
                    _ENV_PATH, len(_FILE_VALUES), len(CONFLICTS))
    else:
        logger.warning("未找到 %s，全部配置取自进程环境变量与代码默认值", _ENV_PATH)

    for key, reason in CONFLICTS.items():
        # 只暴露键名与脱敏后的值，不打印完整密钥
        logger.warning(
            "配置冲突 %s：%s（生效值 %s）",
            key, reason, _mask(_FILE_VALUES.get(key, "")),
        )


def get_str(key: str, default: str = "") -> str:
    value = os.getenv(key)
    return default if value is None or value == "" else value


def get_int(key: str, default: int = 0) -> int:
    raw = os.getenv(key)
    if raw is None or raw.strip() == "":
        return default
    try:
        return int(raw.strip())
    except ValueError:
        return default


def get_float(key: str, default: float = 0.0) -> float:
    raw = os.getenv(key)
    if raw is None or raw.strip() == "":
        return default
    try:
        return float(raw.strip())
    except ValueError:
        return default


def get_bool(key: str, default: bool = False) -> bool:
    """把 ``1/true/yes/on`` 视为真，``0/false/no/off`` 视为假，其余回落默认值。"""
    raw = os.getenv(key)
    if raw is None or raw.strip() == "":
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


# ============================================================
# 服务
# ============================================================
HOST: str = get_str("HOST", "0.0.0.0")
PORT: int = get_int("PORT", 8080)
LOG_LEVEL: str = get_str("LOG_LEVEL", "INFO")
LOG_FORMAT: str = get_str("LOG_FORMAT", "text")

# ============================================================
# 模型
# ============================================================
MODEL: str = get_str("DEEPAGENT_MODEL", "deepseek-flash")
SUMMARY_MODEL: str = get_str("SUMMARY_MODEL", "") or MODEL
OPENAI_BASE_URL: str = get_str("OPENAI_BASE_URL", "")
OPENAI_API_KEY: str = get_str("OPENAI_API_KEY", "")
OPENAI_TIMEOUT: int = get_int("OPENAI_TIMEOUT", 180)

# 思考级别：off | high | highest，烧进模型实例（运行时切换需重建 agent）
THINKING_LEVEL: str = get_str("DEEPAGENT_THINKING", "off")

# ============================================================
# 存储
# ============================================================
# mysql（默认，生产/联调）| sqlite（本地开发与自动化测试）
DB_DIALECT: str = get_str("DB_DIALECT", "mysql").lower()
DB_HOST: str = get_str("DB_HOST", "127.0.0.1")
DB_PORT: int = get_int("DB_PORT", 3306)
DB_USER: str = get_str("DB_USER", "root")
DB_PASSWORD: str = get_str("DB_PASSWORD", "")
DB_NAME: str = get_str("DB_NAME", "vent_agent_v2")
DB_PING_IDLE_SECONDS: int = get_int("DB_PING_IDLE_SECONDS", 60)
# sqlite 档位的库文件位置
DB_SQLITE_PATH: str = get_str("DB_SQLITE_PATH", "data/dev.sqlite")

# ============================================================
# 图状态（checkpoint）
# ============================================================
CHECKPOINTER: str = get_str("CHECKPOINTER", "sqlite").lower()  # sqlite | memory
CHECKPOINT_DB_PATH: str = get_str("CHECKPOINT_DB_PATH", "data/checkpoints.sqlite")

# ============================================================
# 上下文与压缩
# ============================================================
# 声明的上下文窗口（0 = 用模型登记值）。
# 这是压缩的**唯一调节旋钮**：框架的摘要中间件在「声明窗口的 85%」处触发，
# 所以想让它更早压缩，就把 CONTEXT_LIMIT 调到 (真实窗口 × 期望比例 / 0.85)。
CONTEXT_LIMIT: int = get_int("CONTEXT_LIMIT", 0)
# 自动压缩开关。关闭时通过 HarnessProfile.excluded_middleware 摘掉框架的
# 摘要中间件——注意这会同时失去「淘汰历史自动落盘到 /conversation_history/」
# 这个能力，长会话会硬撑到超出模型窗口后报错。
CTX_COMPACT_ENABLED: bool = get_bool("CTX_COMPACT_ENABLED", True)
# 冷启动时从会话历史补种的最近消息条数（热线程只喂当轮新消息）
CTX_RECENT_LIMIT: int = get_int("CTX_RECENT_LIMIT", 30)

# 框架固定的压缩触发比例。仅用于展示与断言，改它无效——真正的旋钮是
# CONTEXT_LIMIT。放在这里是为了让 /health 能如实报出「几成触发」。
FRAMEWORK_COMPACT_TRIGGER: float = 0.85

# ============================================================
# 并发闸
# ============================================================
MAX_CONCURRENT_TASKS: int = get_int("TFAGENTS_MAX_CONCURRENT_TASKS", 10)
MAX_USER_TASKS: int = get_int("TFAGENTS_MAX_USER_TASKS", 2)
TASK_WALL_TIMEOUT: int = get_int("TFAGENTS_TASK_WALL_TIMEOUT", 1800)

# ============================================================
# 计费
# ============================================================
DAILY_TOKEN_LIMIT: int = get_int("TFAGENTS_DAILY_TOKEN_LIMIT", 0)  # 0 = 不限

# ============================================================
# 观测
# ============================================================
TRACE_MIDDLEWARE: bool = get_bool("TFA_TRACE_MIDDLEWARE", True)


def resolve_path(relative: str) -> Path:
    """把配置里的相对路径解析为项目根下的绝对路径，并确保父目录存在。"""
    path = Path(relative)
    if not path.is_absolute():
        path = ROOT / path
    path.parent.mkdir(parents=True, exist_ok=True)
    return path