feat(phase-2.5): mutate_prompt_llm operator + weighted dispatcher + GA wiring
Implementazione completa Phase 2.5 (LLM prompt mutator) seguendo il piano in docs/superpowers/plans/2026-05-11-mutate-prompt-llm-phase-2-5.md. Nuovo modulo src/multi_swarm/genome/mutation_prompt_llm.py: - 6 mutation instructions (tighten_threshold, swap_comparator, add_condition, remove_condition, change_timeframe, add_temporal_gate) - mutate_prompt_llm(g, llm, rng, mutator_tier=ModelTier.B): clona genome con tier B per la call mutator, costruisce system+user prompt con istruzione scelta random, estrae prompt da tag <prompt>...</prompt>, valida - is_valid_prompt(): 3 check (lunghezza >= 50, keyword tecnica, diff > 5% Levenshtein-like via difflib.SequenceMatcher) - Fallback random_mutate su qualsiasi validation fail O LLM exception Esteso src/multi_swarm/genome/mutation.py con weighted_random_mutate dispatcher: con probabilità prompt_mutation_weight invoca mutate_prompt_llm, altrimenti random_mutate. Backward compat: llm=None oppure weight=0 → solo scalare. Integrazione GA loop + RunConfig: - GAConfig.prompt_mutation_weight: float = 0.0 (default off) - next_generation(llm=...) propagato all'invocazione mutator - RunConfig.prompt_mutation_weight con stesso default - run_phase1: passa llm a next_generation solo se weight > 0 - scripts/run_phase1.py: flag CLI --prompt-mutation-weight Tests (+18, 175 totale): - tests/unit/test_mutation_prompt_llm.py (12): extract_prompt, is_valid_prompt 3 check, operator success + fallback su 3 modi (invalid/identical/exception), tier B per LLM call, istruzione scelta dal pool - tests/unit/test_mutation_dispatcher.py (4): weight 0/1/None + distribuzione 30/70 su 1000 estrazioni con tolleranza ±5% - tests/integration/test_ga_loop_with_prompt_mutator.py (2): loop con weight=1.0 produce prompt evoluti; backward compat weight=0 invariato Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,104 @@
|
||||
"""Integration test Phase 2.5: GA loop con LLM mutator attivo.
|
||||
|
||||
Verifica che ``next_generation`` con ``prompt_mutation_weight > 0`` e ``llm``
|
||||
fornito produca figli con system_prompt mutato dall'LLM (e non solo scalari).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
from dataclasses import dataclass
|
||||
|
||||
from multi_swarm.ga.loop import GAConfig, next_generation
|
||||
from multi_swarm.genome.hypothesis import HypothesisAgentGenome, ModelTier
|
||||
|
||||
_PROMPT_TEMPLATES = (
|
||||
"Strategia mean-reversion 1h. Entry long RSI(14) < 30 e close > SMA(50). Stop 2%.",
|
||||
"Strategia momentum breakout. Entry long close > SMA(20) e ATR(14) crescente.",
|
||||
"Strategia trend-following 4h. Long SMA(20) > SMA(50). Short opposito.",
|
||||
)
|
||||
|
||||
|
||||
def _make_pop(n: int) -> list[HypothesisAgentGenome]:
|
||||
return [
|
||||
HypothesisAgentGenome(
|
||||
system_prompt=_PROMPT_TEMPLATES[i % len(_PROMPT_TEMPLATES)],
|
||||
feature_access=["close", "high"],
|
||||
temperature=0.9 + 0.01 * i,
|
||||
top_p=0.95,
|
||||
model_tier=ModelTier.C,
|
||||
lookback_window=200,
|
||||
cognitive_style="physicist",
|
||||
)
|
||||
for i in range(n)
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Result:
|
||||
text: str
|
||||
|
||||
|
||||
class _MutatorLLM:
|
||||
"""Mock che produce un prompt diverso (e valido) a ogni call."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.calls = 0
|
||||
|
||||
def complete(self, genome, system, user, max_tokens: int = 2000) -> _Result:
|
||||
self.calls += 1
|
||||
# Prompt sempre diverso per garantire validation pass.
|
||||
return _Result(
|
||||
text=(
|
||||
f"<prompt>Strategia evolved #{self.calls}. Entry long quando "
|
||||
f"RSI(14) < {25 + self.calls % 10} e close > SMA({40 + self.calls}). "
|
||||
f"Exit short quando momentum decade. Trade rule {self.calls}.</prompt>"
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def test_loop_with_prompt_mutator_produces_prompt_diversity() -> None:
|
||||
"""Con weight 1.0 + crossover 0 (solo mutation), tutti i child non-elite
|
||||
devono avere system_prompt diverso dai parent (LLM-mutated)."""
|
||||
rng = random.Random(0)
|
||||
pop = _make_pop(5)
|
||||
fitnesses = {g.id: 0.0 for g in pop}
|
||||
cfg = GAConfig(
|
||||
population_size=5,
|
||||
elite_k=1,
|
||||
tournament_k=2,
|
||||
p_crossover=0.0, # nessun crossover → tutta la non-elite è mutation
|
||||
prompt_mutation_weight=1.0,
|
||||
)
|
||||
llm = _MutatorLLM()
|
||||
|
||||
new_pop = next_generation(pop, fitnesses, cfg, rng, llm=llm)
|
||||
|
||||
assert len(new_pop) == 5
|
||||
# 4 non-elite figli, tutti con prompt evoluti.
|
||||
parent_prompts = {g.system_prompt for g in pop}
|
||||
evolved = [g for g in new_pop[1:] if g.system_prompt not in parent_prompts]
|
||||
assert len(evolved) >= 3, f"Solo {len(evolved)} figli con prompt mutato"
|
||||
assert llm.calls >= 4
|
||||
|
||||
|
||||
def test_loop_backward_compat_no_llm_no_prompt_mutation() -> None:
|
||||
"""Default weight=0.0 + llm=None → comportamento identico a Phase 2."""
|
||||
rng = random.Random(0)
|
||||
pop = _make_pop(5)
|
||||
fitnesses = {g.id: float(i) for i, g in enumerate(pop)}
|
||||
cfg = GAConfig(
|
||||
population_size=5,
|
||||
elite_k=1,
|
||||
tournament_k=2,
|
||||
p_crossover=0.0,
|
||||
prompt_mutation_weight=0.0,
|
||||
)
|
||||
|
||||
new_pop = next_generation(pop, fitnesses, cfg, rng, llm=None)
|
||||
|
||||
assert len(new_pop) == 5
|
||||
# Nessun child con prompt diverso dai parent: solo mutazioni scalari.
|
||||
parent_prompts = {g.system_prompt for g in pop}
|
||||
for child in new_pop:
|
||||
assert child.system_prompt in parent_prompts
|
||||
@@ -0,0 +1,102 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
from collections import Counter
|
||||
from dataclasses import dataclass
|
||||
|
||||
from multi_swarm.genome.hypothesis import HypothesisAgentGenome, ModelTier
|
||||
from multi_swarm.genome.mutation import weighted_random_mutate
|
||||
|
||||
_PROMPT = (
|
||||
"Strategia mean-reversion 1h BTC. Entry long quando RSI(14) < 30 e "
|
||||
"close > SMA(50). Exit short quando RSI(14) > 70."
|
||||
)
|
||||
|
||||
|
||||
def _make_genome() -> HypothesisAgentGenome:
|
||||
return HypothesisAgentGenome(
|
||||
system_prompt=_PROMPT,
|
||||
feature_access=["close"],
|
||||
temperature=0.9,
|
||||
top_p=0.95,
|
||||
model_tier=ModelTier.C,
|
||||
lookback_window=200,
|
||||
cognitive_style="physicist",
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _R:
|
||||
text: str
|
||||
|
||||
|
||||
class _AlwaysMutateLLM:
|
||||
"""Mock LLM che ritorna sempre un prompt mutato valido."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.calls = 0
|
||||
|
||||
def complete(self, genome, system, user, max_tokens: int = 2000) -> _R:
|
||||
self.calls += 1
|
||||
return _R(
|
||||
text=(
|
||||
"<prompt>Strategia momentum 1h BTC. Entry long quando close > "
|
||||
f"SMA(70) e ATR(14) crescente. Exit con stop loss 3% (call #{self.calls}).</prompt>"
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def test_weighted_dispatcher_zero_weight_never_calls_llm() -> None:
|
||||
llm = _AlwaysMutateLLM()
|
||||
rng = random.Random(0)
|
||||
parent = _make_genome()
|
||||
|
||||
for _ in range(50):
|
||||
weighted_random_mutate(parent, rng, llm=llm, prompt_mutation_weight=0.0)
|
||||
|
||||
assert llm.calls == 0
|
||||
|
||||
|
||||
def test_weighted_dispatcher_full_weight_always_calls_llm() -> None:
|
||||
llm = _AlwaysMutateLLM()
|
||||
rng = random.Random(0)
|
||||
parent = _make_genome()
|
||||
|
||||
for _ in range(20):
|
||||
child = weighted_random_mutate(
|
||||
parent, rng, llm=llm, prompt_mutation_weight=1.0
|
||||
)
|
||||
assert child.system_prompt != parent.system_prompt
|
||||
|
||||
assert llm.calls == 20
|
||||
|
||||
|
||||
def test_weighted_dispatcher_none_llm_falls_back_to_scalar() -> None:
|
||||
"""Senza llm passato (backward compat) → solo mutazione scalare."""
|
||||
rng = random.Random(0)
|
||||
parent = _make_genome()
|
||||
|
||||
for _ in range(50):
|
||||
child = weighted_random_mutate(parent, rng, llm=None, prompt_mutation_weight=0.5)
|
||||
assert child.system_prompt == parent.system_prompt
|
||||
|
||||
|
||||
def test_weighted_dispatcher_distribution_30_70() -> None:
|
||||
"""Su 1000 estrazioni con weight=0.3 il prompt mutator deve essere chiamato ~300 volte."""
|
||||
llm = _AlwaysMutateLLM()
|
||||
rng = random.Random(123)
|
||||
parent = _make_genome()
|
||||
|
||||
counter: Counter[str] = Counter()
|
||||
for _ in range(1000):
|
||||
child = weighted_random_mutate(
|
||||
parent, rng, llm=llm, prompt_mutation_weight=0.3
|
||||
)
|
||||
if child.system_prompt != parent.system_prompt:
|
||||
counter["prompt"] += 1
|
||||
else:
|
||||
counter["scalar"] += 1
|
||||
|
||||
# 30% ± 5% tolerance
|
||||
assert 250 <= counter["prompt"] <= 350, f"prompt mutations: {counter['prompt']}"
|
||||
assert 650 <= counter["scalar"] <= 750, f"scalar mutations: {counter['scalar']}"
|
||||
@@ -0,0 +1,178 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
from dataclasses import dataclass
|
||||
|
||||
from multi_swarm.genome.hypothesis import HypothesisAgentGenome, ModelTier
|
||||
from multi_swarm.genome.mutation_prompt_llm import (
|
||||
MUTATION_INSTRUCTIONS,
|
||||
_extract_prompt,
|
||||
is_valid_prompt,
|
||||
mutate_prompt_llm,
|
||||
)
|
||||
|
||||
_BASE_PROMPT = (
|
||||
"Strategia mean-reversion 1h su BTC. Entry long quando RSI(14) < 30 e "
|
||||
"close > SMA(50). Exit short quando RSI(14) > 70. Stop loss 2%."
|
||||
)
|
||||
|
||||
|
||||
def _make_genome(prompt: str = _BASE_PROMPT) -> HypothesisAgentGenome:
|
||||
return HypothesisAgentGenome(
|
||||
system_prompt=prompt,
|
||||
feature_access=["close", "high", "low"],
|
||||
temperature=0.9,
|
||||
top_p=0.95,
|
||||
model_tier=ModelTier.C,
|
||||
lookback_window=200,
|
||||
cognitive_style="physicist",
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeResult:
|
||||
text: str
|
||||
|
||||
|
||||
class _FakeLLM:
|
||||
"""Mock LLMClient: ritorna una risposta configurata in input."""
|
||||
|
||||
def __init__(self, response_text: str = "", raise_exc: bool = False) -> None:
|
||||
self.response_text = response_text
|
||||
self.raise_exc = raise_exc
|
||||
self.last_call: dict[str, object] | None = None
|
||||
|
||||
def complete(
|
||||
self,
|
||||
genome: HypothesisAgentGenome,
|
||||
system: str,
|
||||
user: str,
|
||||
max_tokens: int = 2000,
|
||||
) -> _FakeResult:
|
||||
self.last_call = {
|
||||
"genome_tier": genome.model_tier,
|
||||
"system": system,
|
||||
"user": user,
|
||||
"max_tokens": max_tokens,
|
||||
}
|
||||
if self.raise_exc:
|
||||
raise RuntimeError("simulated LLM failure")
|
||||
return _FakeResult(text=self.response_text)
|
||||
|
||||
|
||||
def test_extract_prompt_from_tag() -> None:
|
||||
raw = "preambolo blah\n<prompt>Strategia RSI > 75 short, SMA(60) trend.</prompt>\nblabla"
|
||||
assert _extract_prompt(raw) == "Strategia RSI > 75 short, SMA(60) trend."
|
||||
|
||||
|
||||
def test_extract_prompt_no_tag_returns_stripped_text() -> None:
|
||||
raw = " Strategia momentum breakout su 1h con ATR(14) "
|
||||
assert _extract_prompt(raw) == "Strategia momentum breakout su 1h con ATR(14)"
|
||||
|
||||
|
||||
def test_is_valid_prompt_accepts_proper_strategy() -> None:
|
||||
new = (
|
||||
"Strategia mean-reversion 1h su BTC. Entry long quando RSI(14) < 25 e "
|
||||
"close > SMA(50). Exit short quando RSI(14) > 75."
|
||||
)
|
||||
assert is_valid_prompt(new, _BASE_PROMPT) is True
|
||||
|
||||
|
||||
def test_is_valid_prompt_rejects_too_short() -> None:
|
||||
assert is_valid_prompt("short", _BASE_PROMPT) is False
|
||||
|
||||
|
||||
def test_is_valid_prompt_rejects_no_strategy_keywords() -> None:
|
||||
bad = "Questo è un testo a caso che parla del meteo di domani e della pioggia."
|
||||
assert is_valid_prompt(bad, _BASE_PROMPT) is False
|
||||
|
||||
|
||||
def test_is_valid_prompt_rejects_identical_prompt() -> None:
|
||||
assert is_valid_prompt(_BASE_PROMPT, _BASE_PROMPT) is False
|
||||
|
||||
|
||||
def test_mutate_prompt_llm_produces_mutated_child() -> None:
|
||||
mutated = (
|
||||
"Strategia mean-reversion 1h su BTC. Entry long quando RSI(14) < 25 e "
|
||||
"close > SMA(70). Exit short quando RSI(14) > 78."
|
||||
)
|
||||
llm = _FakeLLM(response_text=f"<prompt>{mutated}</prompt>")
|
||||
parent = _make_genome()
|
||||
rng = random.Random(0)
|
||||
|
||||
child = mutate_prompt_llm(parent, llm, rng)
|
||||
|
||||
assert child.system_prompt == mutated
|
||||
assert child.id != parent.id
|
||||
assert child.parent_ids == [*parent.parent_ids, parent.id]
|
||||
assert child.generation == parent.generation + 1
|
||||
assert child.model_tier == ModelTier.C # tier C preservato sul child
|
||||
|
||||
|
||||
def test_mutate_prompt_llm_uses_mutator_tier_b_for_llm_call() -> None:
|
||||
mutated = (
|
||||
"Strategia momentum breakout 1h. Entry long quando close > SMA(60) e "
|
||||
"ATR(14) crescente. Exit con stop loss 3%."
|
||||
)
|
||||
llm = _FakeLLM(response_text=f"<prompt>{mutated}</prompt>")
|
||||
parent = _make_genome()
|
||||
rng = random.Random(0)
|
||||
|
||||
mutate_prompt_llm(parent, llm, rng)
|
||||
|
||||
assert llm.last_call is not None
|
||||
assert llm.last_call["genome_tier"] == ModelTier.B
|
||||
|
||||
|
||||
def test_mutate_prompt_llm_falls_back_on_invalid_output() -> None:
|
||||
"""Output troppo corto -> fallback random_mutate (cambia uno scalare)."""
|
||||
llm = _FakeLLM(response_text="<prompt>nope</prompt>")
|
||||
parent = _make_genome()
|
||||
rng = random.Random(42)
|
||||
|
||||
child = mutate_prompt_llm(parent, llm, rng)
|
||||
|
||||
# random_mutate preserva system_prompt, cambia uno dei 4 scalari/style.
|
||||
assert child.system_prompt == parent.system_prompt
|
||||
assert child.parent_ids == [*parent.parent_ids, parent.id]
|
||||
assert child.generation == parent.generation + 1
|
||||
|
||||
|
||||
def test_mutate_prompt_llm_falls_back_on_identical_output() -> None:
|
||||
llm = _FakeLLM(response_text=f"<prompt>{_BASE_PROMPT}</prompt>")
|
||||
parent = _make_genome()
|
||||
rng = random.Random(42)
|
||||
|
||||
child = mutate_prompt_llm(parent, llm, rng)
|
||||
|
||||
assert child.system_prompt == parent.system_prompt
|
||||
assert child.generation == parent.generation + 1
|
||||
|
||||
|
||||
def test_mutate_prompt_llm_falls_back_on_llm_exception() -> None:
|
||||
llm = _FakeLLM(raise_exc=True)
|
||||
parent = _make_genome()
|
||||
rng = random.Random(7)
|
||||
|
||||
child = mutate_prompt_llm(parent, llm, rng)
|
||||
|
||||
# Fallback random_mutate sempre produce un child valido.
|
||||
assert child.system_prompt == parent.system_prompt
|
||||
assert child.generation == parent.generation + 1
|
||||
|
||||
|
||||
def test_mutate_prompt_llm_picks_one_of_six_instructions() -> None:
|
||||
"""Verifica che il system message dell'LLM includa una delle 6 istruzioni."""
|
||||
mutated = (
|
||||
"Strategia RSI 1h. Entry long quando RSI(14) < 28 e close > SMA(50). "
|
||||
"Exit short quando RSI(14) > 72."
|
||||
)
|
||||
llm = _FakeLLM(response_text=f"<prompt>{mutated}</prompt>")
|
||||
parent = _make_genome()
|
||||
|
||||
mutate_prompt_llm(parent, llm, random.Random(0))
|
||||
|
||||
assert llm.last_call is not None
|
||||
user_text = str(llm.last_call["user"])
|
||||
matched_keys = [k for k in MUTATION_INSTRUCTIONS if k in user_text]
|
||||
assert len(matched_keys) >= 1, f"User prompt non contiene istruzione: {user_text[:200]}"
|
||||
Reference in New Issue
Block a user