Coverage for src/qdrant_loader_mcp_server/config.py: 76%
135 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-08 06:06 +0000
1"""Configuration settings for the RAG MCP Server."""
3import json
4import logging
5import os
6from typing import Annotated
8from dotenv import load_dotenv
9from pydantic import BaseModel, Field
11# Load environment variables from .env file
12load_dotenv()
14# Module logger
15logger = logging.getLogger(__name__)
18# --- Helpers -----------------------------------------------------------------
20# Accepted boolean truthy/falsey strings (case-insensitive)
21TRUE_VALUES = {"1", "true", "t", "yes", "y", "on"}
22FALSE_VALUES = {"0", "false", "f", "no", "n", "off"}
25def parse_bool_env(var_name: str, default: bool) -> bool:
26 """Parse a boolean from an environment variable robustly.
28 Accepted true values: 1, true, t, yes, y, on
29 Accepted false values: 0, false, f, no, n, off
31 Raises:
32 ValueError: If the variable is set but not a valid boolean value.
33 """
34 raw_value = os.getenv(var_name)
35 if raw_value is None:
36 return default
37 normalized = raw_value.strip().lower()
38 if normalized in TRUE_VALUES:
39 return True
40 if normalized in FALSE_VALUES:
41 return False
42 raise ValueError(
43 f"Invalid boolean for {var_name}: {raw_value!r}. "
44 f"Expected one of {sorted(TRUE_VALUES | FALSE_VALUES)}"
45 )
48def parse_int_env(
49 var_name: str,
50 default: int,
51 *,
52 min_value: int | None = None,
53 max_value: int | None = None,
54) -> int:
55 """Parse an integer from an environment variable with bounds checking.
57 Args:
58 var_name: Environment variable name to read.
59 default: Value to use when the variable is not set.
60 min_value: Optional lower bound (inclusive).
61 max_value: Optional upper bound (inclusive).
63 Raises:
64 ValueError: If the variable is set but not an int, or out of bounds.
65 """
66 raw_value = os.getenv(var_name)
67 if raw_value is None or raw_value.strip() == "":
68 return default
69 try:
70 value = int(raw_value)
71 except (TypeError, ValueError) as exc:
72 raise ValueError(f"Invalid integer for {var_name}: {raw_value!r}") from exc
73 if min_value is not None and value < min_value:
74 raise ValueError(f"{var_name} must be >= {min_value}; got {value}")
75 if max_value is not None and value > max_value:
76 raise ValueError(f"{var_name} must be <= {max_value}; got {value}")
77 return value
80def parse_float_env(
81 var_name: str,
82 default: float,
83 *,
84 min_value: float | None = None,
85 max_value: float | None = None,
86) -> float:
87 """Parse a float from an environment variable with bounds checking.
89 Args:
90 var_name: Environment variable name to read.
91 default: Value to use when the variable is not set.
92 min_value: Optional lower bound (inclusive).
93 max_value: Optional upper bound (inclusive).
95 Raises:
96 ValueError: If the variable is set but not a float, or out of bounds.
97 """
98 raw_value = os.getenv(var_name)
99 if raw_value is None or raw_value.strip() == "":
100 return default
101 try:
102 value = float(raw_value)
103 except (TypeError, ValueError) as exc:
104 raise ValueError(f"Invalid float for {var_name}: {raw_value!r}") from exc
105 if min_value is not None and value < min_value:
106 raise ValueError(f"{var_name} must be >= {min_value}; got {value}")
107 if max_value is not None and value > max_value:
108 raise ValueError(f"{var_name} must be <= {max_value}; got {value}")
109 return value
112class ServerConfig(BaseModel):
113 """Server configuration settings."""
115 host: str = "0.0.0.0"
116 port: int = 8000
117 log_level: str = "INFO"
120class QdrantConfig(BaseModel):
121 """Qdrant configuration settings."""
123 url: str = "http://localhost:6333"
124 api_key: str | None = None
125 collection_name: str = "documents"
127 def __init__(self, **data):
128 """Initialize with environment variables if not provided."""
129 if "url" not in data:
130 data["url"] = os.getenv("QDRANT_URL", "http://localhost:6333")
131 if "api_key" not in data:
132 data["api_key"] = os.getenv("QDRANT_API_KEY")
133 if "collection_name" not in data:
134 data["collection_name"] = os.getenv("QDRANT_COLLECTION_NAME", "documents")
135 super().__init__(**data)
138class SearchConfig(BaseModel):
139 """Search optimization configuration settings."""
141 # Search result caching
142 cache_enabled: bool = True
143 cache_ttl: Annotated[int, Field(ge=0, le=86_400)] = 300 # 0s..24h
144 cache_max_size: Annotated[int, Field(ge=1, le=100_000)] = 500
146 # Search parameters optimization
147 hnsw_ef: Annotated[int, Field(ge=1, le=32_768)] = 128 # HNSW search parameter
148 use_exact_search: bool = False # Use exact search when needed
150 # Conflict detection performance controls (defaults calibrated for P95 ~8–10s)
151 conflict_limit_default: Annotated[int, Field(ge=2, le=50)] = 10
152 conflict_max_pairs_total: Annotated[int, Field(ge=1, le=200)] = 24
153 conflict_tier_caps: dict = {
154 "primary": 12,
155 "secondary": 8,
156 "tertiary": 4,
157 "fallback": 0,
158 }
159 conflict_use_llm: bool = True
160 conflict_max_llm_pairs: Annotated[int, Field(ge=0, le=10)] = 2
161 conflict_llm_model: str = "gpt-4o-mini"
162 conflict_llm_timeout_s: Annotated[float, Field(gt=0, le=60)] = 12.0
163 conflict_overall_timeout_s: Annotated[float, Field(gt=0, le=60)] = 9.0
164 conflict_text_window_chars: Annotated[int, Field(ge=200, le=8000)] = 2000
165 conflict_embeddings_timeout_s: Annotated[float, Field(gt=0, le=30)] = 2.0
166 conflict_embeddings_max_concurrency: Annotated[int, Field(ge=1, le=20)] = 5
168 def __init__(self, **data):
169 """Initialize with environment variables if not provided.
171 Performs robust boolean parsing and strict numeric validation to avoid
172 subtle runtime issues from malformed environment inputs.
173 """
174 if "cache_enabled" not in data:
175 data["cache_enabled"] = parse_bool_env("SEARCH_CACHE_ENABLED", True)
176 if "cache_ttl" not in data:
177 data["cache_ttl"] = parse_int_env(
178 "SEARCH_CACHE_TTL", 300, min_value=0, max_value=86_400
179 )
180 if "cache_max_size" not in data:
181 data["cache_max_size"] = parse_int_env(
182 "SEARCH_CACHE_MAX_SIZE", 500, min_value=1, max_value=100_000
183 )
184 if "hnsw_ef" not in data:
185 data["hnsw_ef"] = parse_int_env(
186 "SEARCH_HNSW_EF", 128, min_value=1, max_value=32_768
187 )
188 if "use_exact_search" not in data:
189 data["use_exact_search"] = parse_bool_env("SEARCH_USE_EXACT", False)
191 # Conflict detection env overrides (optional; safe defaults used if unset)
192 def _get_env_dict(name: str, default: dict) -> dict:
193 raw = os.getenv(name)
194 if not raw:
195 return default
196 try:
197 parsed = json.loads(raw)
198 if isinstance(parsed, dict):
199 return parsed
200 return default
201 except (json.JSONDecodeError, ValueError) as exc:
202 # Shorten raw value to avoid logging excessively large strings
203 raw_preview = raw if len(raw) <= 200 else f"{raw[:200]}..."
204 logger.warning(
205 "Failed to parse JSON for env var %s; raw=%r; falling back to default. Error: %s",
206 name,
207 raw_preview,
208 exc,
209 exc_info=True,
210 )
211 return default
213 if "conflict_limit_default" not in data:
214 data["conflict_limit_default"] = parse_int_env(
215 "SEARCH_CONFLICT_LIMIT_DEFAULT", 10, min_value=2, max_value=50
216 )
217 if "conflict_max_pairs_total" not in data:
218 data["conflict_max_pairs_total"] = parse_int_env(
219 "SEARCH_CONFLICT_MAX_PAIRS_TOTAL", 24, min_value=1, max_value=200
220 )
221 if "conflict_tier_caps" not in data:
222 data["conflict_tier_caps"] = _get_env_dict(
223 "SEARCH_CONFLICT_TIER_CAPS",
224 {"primary": 12, "secondary": 8, "tertiary": 4, "fallback": 0},
225 )
226 if "conflict_use_llm" not in data:
227 data["conflict_use_llm"] = parse_bool_env("SEARCH_CONFLICT_USE_LLM", True)
228 if "conflict_max_llm_pairs" not in data:
229 data["conflict_max_llm_pairs"] = parse_int_env(
230 "SEARCH_CONFLICT_MAX_LLM_PAIRS", 2, min_value=0, max_value=10
231 )
232 if "conflict_llm_model" not in data:
233 data["conflict_llm_model"] = os.getenv(
234 "SEARCH_CONFLICT_LLM_MODEL", "gpt-4o-mini"
235 )
236 if "conflict_llm_timeout_s" not in data:
237 data["conflict_llm_timeout_s"] = parse_float_env(
238 "SEARCH_CONFLICT_LLM_TIMEOUT_S", 12.0, min_value=1.0, max_value=60.0
239 )
240 if "conflict_overall_timeout_s" not in data:
241 data["conflict_overall_timeout_s"] = parse_float_env(
242 "SEARCH_CONFLICT_OVERALL_TIMEOUT_S", 9.0, min_value=1.0, max_value=60.0
243 )
244 if "conflict_text_window_chars" not in data:
245 data["conflict_text_window_chars"] = parse_int_env(
246 "SEARCH_CONFLICT_TEXT_WINDOW_CHARS", 2000, min_value=200, max_value=8000
247 )
248 if "conflict_embeddings_timeout_s" not in data:
249 data["conflict_embeddings_timeout_s"] = parse_float_env(
250 "SEARCH_CONFLICT_EMBEDDINGS_TIMEOUT_S",
251 2.0,
252 min_value=1.0,
253 max_value=30.0,
254 )
255 if "conflict_embeddings_max_concurrency" not in data:
256 data["conflict_embeddings_max_concurrency"] = parse_int_env(
257 "SEARCH_CONFLICT_EMBEDDINGS_MAX_CONCURRENCY",
258 5,
259 min_value=1,
260 max_value=20,
261 )
262 super().__init__(**data)
265class OpenAIConfig(BaseModel):
266 """OpenAI configuration settings."""
268 api_key: str
269 model: str = "text-embedding-3-small"
270 chat_model: str = "gpt-3.5-turbo"
273class Config(BaseModel):
274 """Main configuration class."""
276 server: ServerConfig = Field(default_factory=ServerConfig)
277 qdrant: QdrantConfig = Field(default_factory=QdrantConfig)
278 openai: OpenAIConfig = Field(
279 default_factory=lambda: OpenAIConfig(api_key=os.getenv("OPENAI_API_KEY"))
280 )
281 search: SearchConfig = Field(default_factory=SearchConfig)