File size: 3,582 Bytes
d840c10
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
"""Configuration helpers for the GCMD classifier MVP."""

from __future__ import annotations

import os
from collections.abc import Mapping

from pydantic import BaseModel, ConfigDict, Field


class ModelSettings(BaseModel):
    """Model and prompt settings loaded from environment-compatible mappings."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    provider: str = "fake"
    model_name: str = "fake-model"
    temperature: float = Field(default=0.0, ge=0.0)
    timeout_seconds: float = Field(default=30.0, gt=0.0)
    max_retries: int = Field(default=2, ge=0)
    prompt_version_topic: str = "topic-v1"
    prompt_version_term: str = "term-v1"
    prompt_version_variable: str = "variable-v1"
    api_key_env_var: str = "OPENAI_API_KEY"
    include_cost_metadata: bool = True

    @classmethod
    def from_environment(cls, env: Mapping[str, str] | None = None) -> ModelSettings:
        """Load model settings from an environment mapping without reading secrets."""
        source = os.environ if env is None else env
        return cls(
            provider=source.get("MODEL_PROVIDER", cls.model_fields["provider"].default),
            model_name=source.get("MODEL_NAME", cls.model_fields["model_name"].default),
            temperature=_float_env(
                source,
                "MODEL_TEMPERATURE",
                cls.model_fields["temperature"].default,
            ),
            timeout_seconds=_float_env(
                source,
                "MODEL_TIMEOUT_SECONDS",
                cls.model_fields["timeout_seconds"].default,
            ),
            max_retries=_int_env(
                source,
                "MODEL_MAX_RETRIES",
                cls.model_fields["max_retries"].default,
            ),
            prompt_version_topic=source.get(
                "PROMPT_VERSION_TOPIC",
                cls.model_fields["prompt_version_topic"].default,
            ),
            prompt_version_term=source.get(
                "PROMPT_VERSION_TERM",
                cls.model_fields["prompt_version_term"].default,
            ),
            prompt_version_variable=source.get(
                "PROMPT_VERSION_VARIABLE",
                cls.model_fields["prompt_version_variable"].default,
            ),
            api_key_env_var=source.get(
                "MODEL_API_KEY_ENV_VAR",
                cls.model_fields["api_key_env_var"].default,
            ),
            include_cost_metadata=_bool_env(
                source,
                "MODEL_INCLUDE_COST_METADATA",
                cls.model_fields["include_cost_metadata"].default,
            ),
        )

    def prompt_version_for_stage(self, stage: str) -> str:
        """Return the configured prompt version for a model-call stage."""
        if stage == "topic":
            return self.prompt_version_topic
        if stage == "term":
            return self.prompt_version_term
        if stage == "variable":
            return self.prompt_version_variable
        raise ValueError(f"Unknown prompt stage: {stage}")


def _float_env(source: Mapping[str, str], name: str, default: float) -> float:
    value = source.get(name)
    return default if value is None else float(value)


def _int_env(source: Mapping[str, str], name: str, default: int) -> int:
    value = source.get(name)
    return default if value is None else int(value)


def _bool_env(source: Mapping[str, str], name: str, default: bool) -> bool:
    value = source.get(name)
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}