knowledge-base/backend/config.py
Pratik Narola 0f0addb36b chore: migrate to mem0ai v2.0.2 (V3 memory pipeline)
Pin mem0ai[nlp]==2.0.2 and fastembed for the new hybrid-search pipeline.
Drop OSS graph memory (removed upstream in 2.0.0, PR #4805): remove Neo4j
service, env vars, volumes, and driver deps; mark /graph/relationships
deprecated. Rewrite Memory.search/get_all/chat/health call sites to use
the v2 filters={} + top_k API (entity IDs at top level now raise
ValueError). Tighten MCP remove_memory ownership check to O(1)
verify_memory_ownership so it doesn't silently truncate at the new
top_k=20 default. Downgrade base image to python:3.12-slim for spaCy.

Adds scripts/migrate_qdrant_to_v3.py (scroll+upsert with per-user count
parity check) and docs/MIGRATION_RUNBOOK.md covering snapshot, dump,
collection rebuild, cutover, and rollback procedures.
2026-05-23 14:49:45 +05:30

100 lines
3.3 KiB
Python

"""Configuration management for Mem0 Interface POC."""
import os
import json
from typing import List, Optional, Dict
from pydantic import Field, AliasChoices
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""Application settings loaded from environment variables."""
model_config = SettingsConfigDict(
env_file=".env", case_sensitive=False, extra="ignore"
)
# API Configuration
# Accept both OPENAI_API_KEY (from docker-compose) and OPENAI_COMPAT_API_KEY (from direct .env)
openai_api_key: str = Field(
validation_alias=AliasChoices(
"OPENAI_API_KEY", "OPENAI_COMPAT_API_KEY", "openai_api_key"
)
)
openai_base_url: str = Field(
validation_alias=AliasChoices(
"OPENAI_BASE_URL", "OPENAI_COMPAT_BASE_URL", "openai_base_url"
)
)
cohere_api_key: str = Field(
validation_alias=AliasChoices("COHERE_API_KEY", "cohere_api_key")
)
# Database Configuration
qdrant_host: str = Field(
default="localhost", validation_alias=AliasChoices("QDRANT_HOST", "qdrant_host")
)
qdrant_port: int = Field(
default=6333, validation_alias=AliasChoices("QDRANT_PORT", "qdrant_port")
)
qdrant_collection_name: str = Field(
default="mem0",
validation_alias=AliasChoices(
"QDRANT_COLLECTION_NAME", "qdrant_collection_name"
),
)
# Application Configuration
log_level: str = Field(
default="INFO", validation_alias=AliasChoices("LOG_LEVEL", "log_level")
)
cors_origins: str = Field(
default="http://localhost:3000",
validation_alias=AliasChoices("CORS_ORIGINS", "cors_origins"),
)
# Model Configuration - Ultra-minimal (single model)
default_model: str = Field(
default="claude-sonnet-4",
validation_alias=AliasChoices("DEFAULT_MODEL", "default_model"),
)
# Embedder Configuration
ollama_base_url: str = Field(
default="http://host.docker.internal:11434",
validation_alias=AliasChoices("OLLAMA_BASE_URL", "ollama_base_url"),
)
embedding_model: str = Field(
default="qwen3-embedding:4b-q8_0",
validation_alias=AliasChoices("EMBEDDING_MODEL", "embedding_model"),
)
embedding_dims: int = Field(
default=2560, validation_alias=AliasChoices("EMBEDDING_DIMS", "embedding_dims")
)
# Authentication Configuration
# Format: JSON string mapping API keys to user IDs
# Example: {"api_key_123": "alice", "api_key_456": "bob"}
api_keys: str = Field(
default="{}", validation_alias=AliasChoices("API_KEYS", "api_keys")
)
@property
def cors_origins_list(self) -> List[str]:
"""Convert CORS origins string to list."""
return [origin.strip() for origin in self.cors_origins.split(",")]
@property
def api_key_mapping(self) -> Dict[str, str]:
"""Parse and return API key to user_id mapping."""
try:
mapping = json.loads(self.api_keys)
if not isinstance(mapping, dict):
raise ValueError("API_KEYS must be a JSON object/dict")
return mapping
except json.JSONDecodeError as e:
raise ValueError(f"Invalid API_KEYS JSON format: {e}")
# Global settings instance
settings = Settings()