lambdagent/agentexample/qaagent67lambda/scripts/config.py

"""
Unified config loader for LambdaRAG scripts.
All scripts import paths from here instead of hardcoding.

Usage:
    from config import cfg
    print(cfg.raw_dir)       # Path to raw files
    print(cfg.processed_dir) # Path to processed text
    print(cfg.wiki_dir)      # Path to wiki
    print(cfg.base_dir)      # Knowledge base directory
"""
import os
import sys
from pathlib import Path

import yaml


class Config:
    """Loads paths from agent-config.yml + environment variables."""

    def __init__(self):
        self._cfg = {}
        self._load()

    def _find_agent_config(self):
        """Search for agent-config.yml in standard locations"""
        candidates = [
            os.environ.get('AGENT_CONFIG', ''),
            # Relative to this script: scripts/ -> parent = agent dir
            str(Path(__file__).parent.parent / 'agent-config.yml'),
            # Docker paths
            '/app/agentexample/qaagent67lambda/agent-config.yml',
            '/app/agentexample/qaagent67/agent-config.yml',
        ]
        for p in candidates:
            if p and Path(p).is_file():
                return Path(p)
        return None

    def _find_instance_config(self):
        """Search for instance.yml.

        Resolution order:
        1. INSTANCE_CONFIG env var
        2. {baseDir}/instance.yml (after agent-config.yml is loaded)
        3. instance.yml in the directory containing this config.py file
           (lets build scripts auto-detect domain when run from a domain dir)
        """
        here = Path(__file__).parent
        base_dir = (self._cfg.get("knowledge", {}) or {}).get("baseDir", "")
        candidates = [
            os.environ.get("INSTANCE_CONFIG", ""),
            str(Path(base_dir) / "instance.yml") if base_dir else "",
            str(here / "instance.yml"),
        ]
        for p in candidates:
            if p and Path(p).is_file():
                return Path(p)
        return None

    def _load(self):
        config_path = self._find_agent_config()
        if config_path:
            with open(config_path) as f:
                self._cfg = yaml.safe_load(f) or {}

        # Load instance.yml (overrides agent-config.yml)
        instance_path = self._find_instance_config()
        if instance_path and Path(instance_path).is_file():
            with open(instance_path) as f:
                instance = yaml.safe_load(f) or {}
            # Deep merge: instance overrides agent config
            self._cfg = self._deep_merge(self._cfg, instance)

    def _deep_merge(self, base, override):
        import copy
        result = copy.deepcopy(base)
        for k, v in override.items():
            if k in result and isinstance(result[k], dict) and isinstance(v, dict):
                result[k] = self._deep_merge(result[k], v)
            else:
                result[k] = copy.deepcopy(v)
        return result

    # ── Knowledge paths ──
    @property
    def base_dir(self):
        return Path(
            self._cfg.get('knowledge', {}).get('baseDir', '')
            or os.environ.get('KNOWLEDGE_DIR', '')
            or str(Path(__file__).parent.parent / 'knowledge')
        )

    @property
    def raw_dir(self):
        return Path(
            self._cfg.get('knowledge', {}).get('rawDir', '')
            or str(self.base_dir / 'raw')
        )

    @property
    def processed_dir(self):
        return Path(
            self._cfg.get('knowledge', {}).get('processedDir', '')
            or str(self.base_dir / 'processed')
        )

    @property
    def index_file(self):
        return Path(
            self._cfg.get('knowledge', {}).get('indexFile', '')
            or str(self.base_dir / 'rag_index.json')
        )

    @property
    def keyword_file(self):
        return Path(
            self._cfg.get('knowledge', {}).get('keywordFile', '')
            or str(self.base_dir / 'rag_index.keywords.json')
        )

    @property
    def vector_index(self):
        return Path(
            self._cfg.get('knowledge', {}).get('vectorIndex', '')
            or str(self.base_dir / 'rag_vectors_v2.npy')
        )

    @property
    def vector_meta(self):
        return Path(
            self._cfg.get('knowledge', {}).get('vectorMeta', '')
            or str(self.base_dir / 'rag_vectors_meta_v2.pkl')
        )

    # ── Wiki paths ──
    @property
    def wiki_dir(self):
        return Path(
            self._cfg.get('wiki', {}).get('dir', '')
            or os.environ.get('WIKI_DIR', '')
            or str(Path(__file__).parent.parent / 'wiki')
        )

    @property
    def relations_file(self):
        return Path(
            self._cfg.get('wiki', {}).get('relationsFile', '')
            or str(self.wiki_dir / 'relations.json')
        )

    # ── Embedding config ──
    @property
    def embed_url(self):
        return (
            self._cfg.get('retrieval', {}).get('skills', {}).get('embedding', {}).get('baseUrl', '')
            or os.environ.get('OLLAMA_EMBED_URL', 'http://127.0.0.1:11435/api/embed')
        )

    @property
    def embed_model(self):
        return (
            self._cfg.get('retrieval', {}).get('skills', {}).get('embedding', {}).get('model', '')
            or os.environ.get('EMBED_MODEL', 'bge-m3')
        )

    # ── LLM config ──
    @property
    def vllm_url(self):
        return os.environ.get('VLLM_URL', 'http://127.0.0.1:8000/v1/chat/completions')

    @property
    def llm_model(self):
        return self._cfg.get('model', {}).get('name', 'qwen2.5-32b')

    # ── Retrieval weights ──
    @property
    def weight_profiles(self):
        return self._cfg.get('retrieval', {}).get('weights', {
            'fact':      {'bm25': 1.0, 'vector': 0.5, 'graph': 0.1, 'wiki': 0.1},
            'compare':   {'bm25': 0.5, 'vector': 0.8, 'graph': 0.4, 'wiki': 0.7},
            'relation':  {'bm25': 0.5, 'vector': 0.3, 'graph': 0.7, 'wiki': 0.8},
            'synthesis': {'bm25': 0.6, 'vector': 0.5, 'graph': 0.5, 'wiki': 0.6},
            'temporal':  {'bm25': 1.0, 'vector': 0.2, 'graph': 0.05, 'wiki': 0.2},
        })

    @property
    def default_weights(self):
        return self._cfg.get('retrieval', {}).get('weights', {}).get('default',
            {'bm25': 0.7, 'vector': 0.5, 'graph': 0.3, 'wiki': 0.3})

    @property
    def rrf_k(self):
        return self._cfg.get('retrieval', {}).get('orchestration', {}).get('rrfK', 60)

    # ── RAG params ──
    @property
    def chunk_size(self):
        return self._cfg.get('rag', {}).get('chunkSize', 512)

    @property
    def chunk_overlap(self):
        return self._cfg.get('rag', {}).get('chunkOverlap', 64)

    @property
    def top_k(self):
        return self._cfg.get('rag', {}).get('topK', 5)

    # ── Workspace ──
    @property
    def workspace_dir(self):
        return Path(__file__).parent.parent / 'workspace'

    # ── Full config dict ──
    @property
    def raw(self):
        return self._cfg

    def __repr__(self):
        return (f"Config(base_dir={self.base_dir}, raw_dir={self.raw_dir}, "
                f"wiki_dir={self.wiki_dir}, embed_url={self.embed_url})")


# Singleton
cfg = Config()

if __name__ == '__main__':
    print(cfg)
    print(f"\n  base_dir:      {cfg.base_dir}")
    print(f"  raw_dir:       {cfg.raw_dir}")
    print(f"  processed_dir: {cfg.processed_dir}")
    print(f"  index_file:    {cfg.index_file}")
    print(f"  wiki_dir:      {cfg.wiki_dir}")
    print(f"  relations:     {cfg.relations_file}")
    print(f"  embed_url:     {cfg.embed_url}")
    print(f"  vllm_url:      {cfg.vllm_url}")
    print(f"  weights:       {cfg.weight_profiles}")
    print(f"  chunk_size:    {cfg.chunk_size}")
    print(f"  workspace:     {cfg.workspace_dir}")