File size: 4,947 Bytes
8073bab
 
43199e3
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8073bab
0c44617
 
43199e3
 
 
 
8073bab
 
43199e3
 
 
 
 
8073bab
43199e3
 
 
2e8bb22
43199e3
 
 
 
 
8073bab
43199e3
 
 
 
 
8073bab
43199e3
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8073bab
 
43199e3
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8073bab
43199e3
 
 
 
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
import os
from pathlib import Path
from typing import Dict, List, Optional

from dotenv import load_dotenv, find_dotenv
from pydantic_settings import BaseSettings


class AgentConfig(BaseSettings):
    """
    Configuration class that works with environment variable manager.
    """

    # Core settings
    environment: str = "development"
    agent_name: str = "gaia_agent"
    debug: bool = False

    # Model configuration
    model_name: str = "gpt-4.1"
    response_processing_model_name: str = "gpt-4.1-mini"

    max_tokens: int = 20000

    project_root: Path = Path(__file__).parent.parent
    prompts_location: Path = project_root / "config" / "prompts.yaml"

    submission_mode_on: bool = False

    class Config:
        env_file = ".env"
        case_sensitive = False
        extra = "allow"


class EnvironmentVariableManager:
    """
    Manages loading .env files and setting environment variables that
    third-party libraries expect to find.
    """

    def __init__(self, env_file: Optional[str] = None):
        """
        Initialize the environment variable manager.

        Args:
            env_file: Path to .env file (if None, will search for it)
        """
        self.env_file = env_file or find_dotenv()
        self.loaded_vars = {}

    def load_env_file(self) -> Dict[str, str]:
        """Load the .env file and return all variables."""
        if not self.env_file or not Path(self.env_file).exists():
            print(f"Warning: .env file not found at {self.env_file}")
            return {}

        # Load .env file
        load_dotenv(self.env_file, override=True)

        # Read the file manually to get all variables
        env_vars = {}
        with open(self.env_file, 'r') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#') and '=' in line:
                    key, value = line.split('=', 1)
                    key = key.strip()
                    value = value.strip().strip('"\'')  # Remove quotes
                    env_vars[key] = value

        self.loaded_vars = env_vars
        return env_vars

    def get_required_env_vars(self, services: List[str]) -> List[str]:
        """
        Get list of required environment variables for specific services.

        Args:
            services: List of service names (e.g., ['openai', 'anthropic'])

        Returns:
            List of required environment variable names
        """
        service_requirements = {
            'openai': ['OPENAI_API_KEY'],
            'google': ['GOOGLE_API_KEY'],
            'tavily': ['TAVILY_API_KEY'],
        }

        required = []
        for service in services:
            if service.lower() in service_requirements:
                required.extend(service_requirements[service.lower()])

        return required

    def validate_required_env_vars(self, services: List[str]) -> List[str]:
        """
        Validate that required environment variables are set.

        Returns:
            List of missing environment variables
        """
        required = self.get_required_env_vars(services)
        missing = []

        for var in required:
            if not os.environ.get(var):
                missing.append(var)

        return missing


class ConfigLoader:
    """
    Main configuration loader that handles both Pydantic config and environment variables.
    """

    def __init__(self, env_file: Optional[str] = None):
        self.env_manager = EnvironmentVariableManager(env_file)
        self.config = None

    def load_config(self,
                    required_services: Optional[List[str]] = None,
                    validate: bool = True) -> AgentConfig:
        """
        Load configuration and set up environment variables.

        Args:
            required_services: List of services that must have API keys
            validate: Whether to validate required environment variables

        Returns:
            Configured LangGraphConfigWithEnvVars instance
        """

        # Step 1: Load .env file
        print("Loading .env file...")
        loaded_vars = self.env_manager.load_env_file()
        print(f"Loaded {len(loaded_vars)} variables from .env file")

        # Step 2: Load Pydantic configuration
        print("Loading Pydantic configuration...")
        self.config = AgentConfig()

        # Step 3: Validate required services
        if validate and required_services:
            print(f"Validating required services: {required_services}")
            missing = self.env_manager.validate_required_env_vars(required_services)
            if missing:
                raise ValueError(f"Missing required environment variables: {missing}")
            print("✓ All required environment variables are set")

        return self.config


loader = ConfigLoader()

config = loader.load_config(
    required_services=['openai', 'google', 'tavily'],
    validate=True
)