mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-27 23:47:46 +00:00
feat: Centralize all prompts in prompts/ directory
New structure:
- prompts/standard_prompts.yaml: Default prompts (committed to Git)
- prompts/local/: Your improved prompts (NOT in Git!)
- prompts/README.md: Documentation
- rdagent/components/loader.py: Prompt loader with priority
Features:
- Loader checks prompts/local/ first (your better prompts)
- Falls back to standard_prompts.yaml if no local version
- Supports sections (system/user)
- Lists available prompts
- Test function included
.gitignore updated:
- prompts/local/ excluded (your proprietary prompts)
- *.local.yaml excluded
- *_private.yaml excluded
Usage:
from rdagent.components.loader import load_prompt
prompt = load_prompt('factor_discovery') # Auto-loads your better version!
This commit is contained in:
@@ -0,0 +1,193 @@
|
||||
"""
|
||||
Predix Prompt Loader
|
||||
|
||||
Loads prompts from:
|
||||
1. prompts/local/*.yaml (your improved prompts - not in Git)
|
||||
2. prompts/standard_prompts.yaml (default prompts - in Git)
|
||||
|
||||
Usage:
|
||||
from rdagent.components.loader import load_prompt
|
||||
|
||||
# Load factor discovery prompt
|
||||
prompt = load_prompt("factor_discovery")
|
||||
|
||||
# Load with custom local prompt
|
||||
prompt = load_prompt("factor_discovery", local_only=True)
|
||||
"""
|
||||
|
||||
import os
|
||||
import yaml
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
|
||||
# Base paths
|
||||
BASE_DIR = Path(__file__).parent.parent.parent # Predix/
|
||||
PROMPTS_DIR = BASE_DIR / "prompts"
|
||||
LOCAL_PROMPTS_DIR = PROMPTS_DIR / "local"
|
||||
STANDARD_PROMPTS_FILE = PROMPTS_DIR / "standard_prompts.yaml"
|
||||
|
||||
|
||||
def get_local_prompt_path(name: str) -> Optional[Path]:
|
||||
"""Find local prompt file by name."""
|
||||
if not LOCAL_PROMPTS_DIR.exists():
|
||||
return None
|
||||
|
||||
# Try different file extensions
|
||||
for ext in ["yaml", "yml"]:
|
||||
path = LOCAL_PROMPTS_DIR / f"{name}.{ext}"
|
||||
if path.exists():
|
||||
return path
|
||||
|
||||
# Try subdirectories
|
||||
for subdir in LOCAL_PROMPTS_DIR.iterdir():
|
||||
if subdir.is_dir():
|
||||
for ext in ["yaml", "yml"]:
|
||||
path = subdir / f"{name}.{ext}"
|
||||
if path.exists():
|
||||
return path
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def load_yaml_file(path: Path) -> Dict[str, Any]:
|
||||
"""Load YAML file."""
|
||||
with open(path, 'r', encoding='utf-8') as f:
|
||||
return yaml.safe_load(f)
|
||||
|
||||
|
||||
def load_prompt(
|
||||
name: str,
|
||||
section: Optional[str] = None,
|
||||
local_only: bool = False,
|
||||
fallback_to_standard: bool = True
|
||||
) -> str:
|
||||
"""
|
||||
Load a prompt by name.
|
||||
|
||||
Priority:
|
||||
1. prompts/local/{name}.yaml (if exists)
|
||||
2. prompts/standard_prompts.yaml (if fallback_to_standard=True)
|
||||
|
||||
Args:
|
||||
name: Prompt name (e.g., "factor_discovery")
|
||||
section: Specific section in YAML (e.g., "system" or "user")
|
||||
local_only: Only load from local/, raise error if not found
|
||||
fallback_to_standard: If True, fall back to standard prompts
|
||||
|
||||
Returns:
|
||||
Prompt text
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If prompt not found
|
||||
"""
|
||||
# Try local prompts first
|
||||
local_path = get_local_prompt_path(name)
|
||||
|
||||
if local_path:
|
||||
print(f"✓ Loading prompt '{name}' from local: {local_path}")
|
||||
data = load_yaml_file(local_path)
|
||||
|
||||
if section:
|
||||
return data.get(section, "")
|
||||
|
||||
# If data is dict with 'system' and 'user', return full dict
|
||||
if isinstance(data, dict):
|
||||
return data
|
||||
return str(data)
|
||||
|
||||
# Local not found
|
||||
if local_only:
|
||||
raise FileNotFoundError(f"Local prompt '{name}' not found in {LOCAL_PROMPTS_DIR}")
|
||||
|
||||
# Try standard prompts
|
||||
if not fallback_to_standard:
|
||||
raise FileNotFoundError(f"Prompt '{name}' not found")
|
||||
|
||||
if not STANDARD_PROMPTS_FILE.exists():
|
||||
raise FileNotFoundError(f"Standard prompts file not found: {STANDARD_PROMPTS_FILE}")
|
||||
|
||||
print(f"✓ Loading prompt '{name}' from standard prompts")
|
||||
data = load_yaml_file(STANDARD_PROMPTS_FILE)
|
||||
|
||||
# Get section from standard prompts
|
||||
if name in data:
|
||||
prompt_data = data[name]
|
||||
|
||||
if section and isinstance(prompt_data, dict):
|
||||
return prompt_data.get(section, "")
|
||||
|
||||
return prompt_data
|
||||
|
||||
raise FileNotFoundError(f"Prompt '{name}' not found in standard prompts")
|
||||
|
||||
|
||||
def load_all_prompts() -> Dict[str, Any]:
|
||||
"""Load all available prompts."""
|
||||
result = {}
|
||||
|
||||
# Load standard prompts
|
||||
if STANDARD_PROMPTS_FILE.exists():
|
||||
result["standard"] = load_yaml_file(STANDARD_PROMPTS_FILE)
|
||||
|
||||
# Load local prompts
|
||||
if LOCAL_PROMPTS_DIR.exists():
|
||||
result["local"] = {}
|
||||
for path in LOCAL_PROMPTS_DIR.glob("*.yaml"):
|
||||
result["local"][path.stem] = load_yaml_file(path)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def list_available_prompts() -> Dict[str, list]:
|
||||
"""List all available prompts."""
|
||||
result = {"standard": [], "local": []}
|
||||
|
||||
# Standard prompts
|
||||
if STANDARD_PROMPTS_FILE.exists():
|
||||
data = load_yaml_file(STANDARD_PROMPTS_FILE)
|
||||
result["standard"] = list(data.keys())
|
||||
|
||||
# Local prompts
|
||||
if LOCAL_PROMPTS_DIR.exists():
|
||||
result["local"] = [p.stem for p in LOCAL_PROMPTS_DIR.glob("*.yaml")]
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# Convenience functions for specific prompts
|
||||
def get_factor_discovery_prompt() -> Dict[str, str]:
|
||||
"""Get factor discovery prompt (system + user)."""
|
||||
return load_prompt("factor_discovery")
|
||||
|
||||
|
||||
def get_factor_evolution_prompt() -> Dict[str, str]:
|
||||
"""Get factor evolution prompt."""
|
||||
return load_prompt("factor_evolution")
|
||||
|
||||
|
||||
def get_model_coder_prompt() -> Dict[str, str]:
|
||||
"""Get model coder prompt."""
|
||||
return load_prompt("model_coder")
|
||||
|
||||
|
||||
def get_trading_strategy_prompt() -> Dict[str, str]:
|
||||
"""Get trading strategy prompt."""
|
||||
return load_prompt("trading_strategy")
|
||||
|
||||
|
||||
# Test function
|
||||
if __name__ == "__main__":
|
||||
print("=== Available Prompts ===")
|
||||
available = list_available_prompts()
|
||||
print(f"Standard: {available['standard']}")
|
||||
print(f"Local: {available['local']}")
|
||||
|
||||
print("\n=== Testing Prompt Load ===")
|
||||
try:
|
||||
prompt = load_prompt("factor_discovery")
|
||||
print(f"✓ Loaded factor_discovery prompt")
|
||||
print(f" System: {len(prompt.get('system', ''))} chars")
|
||||
print(f" User: {len(prompt.get('user', ''))} chars")
|
||||
except FileNotFoundError as e:
|
||||
print(f"✗ Error: {e}")
|
||||
Reference in New Issue
Block a user