evotraders/backend/agents/prompt_loader.py

# -*- coding: utf-8 -*-
"""
Prompt Loader - Unified management and loading of Agent Prompts
Supports Markdown and YAML formats
Uses simple string replacement, does not depend on Jinja2
"""
import re
from pathlib import Path
from typing import Any, Dict, Optional

import yaml

# Singleton instance
_prompt_loader_instance: Optional["PromptLoader"] = None


def get_prompt_loader() -> "PromptLoader":
    """Get the singleton PromptLoader instance."""
    global _prompt_loader_instance
    if _prompt_loader_instance is None:
        _prompt_loader_instance = PromptLoader()
    return _prompt_loader_instance


class PromptLoader:
    """Unified Prompt loader"""

    def __init__(self, prompts_dir: Optional[Path] = None):
        """
        Initialize Prompt loader

        Args:
            prompts_dir: Prompts directory path,
                         defaults to prompts/ directory of current file
        """
        if prompts_dir is None:
            self.prompts_dir = Path(__file__).parent / "prompts"
        else:
            self.prompts_dir = Path(prompts_dir)

    def load_prompt(
        self,
        agent_type: str,
        prompt_name: str,
        variables: Optional[Dict[str, Any]] = None,
    ) -> str:
        """
        Load and render Prompt.

        No caching — always reads fresh from disk (CoPaw-style).
        """
        prompt_path = self.prompts_dir / agent_type / f"{prompt_name}.md"

        if not prompt_path.exists():
            raise FileNotFoundError(
                f"Prompt file not found: {prompt_path}\n"
                f"Please create the prompt file or check the path.",
            )

        with open(prompt_path, "r", encoding="utf-8") as f:
            prompt_template = f.read()

        # If variables provided, use simple string replacement
        if variables:
            rendered = self._render_template(prompt_template, variables)
        else:
            rendered = prompt_template

        return rendered

    def _render_template(
        self,
        template: str,
        variables: Dict[str, Any],
    ) -> str:
        """
        Render template using simple string replacement
        Supports {{ variable }} syntax (compatible with previous Jinja2 format)

        Args:
            template: Template string
            variables: Variable dictionary

        Returns:
            Rendered string
        """
        rendered = template

        # Replace {{ variable }} format
        for key, value in variables.items():
            # Support both {{ key }} and {{key}} formats
            pattern1 = f"{{{{ {key} }}}}"
            pattern2 = f"{{{{{key}}}}}"
            rendered = rendered.replace(pattern1, str(value))
            rendered = rendered.replace(pattern2, str(value))

        return rendered

    def _escape_json_braces(self, text: str) -> str:
        """
        Escape braces in JSON code blocks, treating them as literals

        Args:
            text: Text to process

        Returns:
            Processed text
        """

        def replace_code_block(match):
            code_content = match.group(1)
            # Escape all braces within code block
            escaped = code_content.replace("{", "{{").replace("}", "}}")
            return f"```json\n{escaped}\n```"

        # Replace all braces in JSON code blocks
        text = re.sub(
            r"```json\n(.*?)\n```",
            replace_code_block,
            text,
            flags=re.DOTALL,
        )
        return text

    def load_yaml_config(
        self,
        agent_type: str,
        config_name: str,
    ) -> Dict[str, Any]:
        """
        Load YAML configuration file.

        No caching — always reads fresh from disk (CoPaw-style).
        """
        yaml_path = self.prompts_dir / agent_type / f"{config_name}.yaml"

        if not yaml_path.exists():
            raise FileNotFoundError(f"YAML config not found: {yaml_path}")

        with open(yaml_path, "r", encoding="utf-8") as f:
            return yaml.safe_load(f) or {}

    def clear_cache(self):
        """No-op — caching removed (CoPaw-style, always fresh reads)."""
        pass

    def reload_prompt(self, agent_type: str, prompt_name: str):
        """No-op — caching removed."""
        pass

    def reload_config(self, agent_type: str, config_name: str):
        """No-op — caching removed."""
        pass