Spaces:
Sleeping
Sleeping
File size: 6,687 Bytes
ce8f04a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 | """Expectation quality loop pure control plane (no LLM)."""
from __future__ import annotations
from agents.world_class_advisor import evaluate_application
from core.generation.expectation_loop import (
is_regulation_ready,
run_advisor_expectation_loop,
run_expectation_loop,
)
BRIEF = {
"key_rules": ["Wnioskodawca musi posiadać status MŚP", "Projekt musi spełniać zasadę DNSH"],
"required_sections": ["Opis projektu", "Budżet"],
"required_attachments": [],
"attention_points": ["Sprawdź DNSH / wpływ środowiskowy w opisie projektu."],
"usable": True,
}
def test_is_regulation_ready_false_for_structure_only():
assert (
is_regulation_ready(
{
"passed": True,
"score": 95,
"regulation_grounded_pass": False,
"blockers": [],
"grounding_mode": "structure_only",
}
)
is False
)
def test_loop_stops_when_ready():
sections = {
"Opis projektu": (
"Projekt MŚP z DNSH i pełnym opisem innowacji. " * 10
),
"Budżet": (
"Budżet z wkładem własnym i kosztami kwalifikowalnymi. " * 10
),
}
def evaluate(state):
return evaluate_application(
sections=state.get("generated_sections") or {},
brief=BRIEF,
grounding_mode="regulation",
)
result = run_expectation_loop(
evaluate=evaluate,
apply_fixes=None,
initial_state={"generated_sections": sections},
max_iterations=3,
min_score=70,
)
assert result.ready is True
assert result.stop_reason == "ready"
assert result.regulation_grounded_pass is True
assert result.iterations >= 1
assert not result.remaining_blockers
def test_loop_max_iter_with_blockers():
weak = {"Opis projektu": "za mało"}
def evaluate(state):
return evaluate_application(
sections=state.get("generated_sections") or {},
brief=BRIEF,
grounding_mode="regulation",
)
def fix(state, report):
# Improves slightly but not enough to cover Budżet
gen = dict(state.get("generated_sections") or {})
gen["Opis projektu"] = (gen.get("Opis projektu") or "") + " dodatek MŚP DNSH " * 5
state = dict(state)
state["generated_sections"] = gen
state["fixed_sections"] = ["Opis projektu"]
return state
result = run_expectation_loop(
evaluate=evaluate,
apply_fixes=fix,
initial_state={"generated_sections": weak},
max_iterations=2,
min_score=70,
)
assert result.ready is False
assert result.stop_reason == "max_iter"
assert result.iterations == 2
assert result.remaining_blockers
assert result.regulation_grounded_pass is False
def test_loop_structure_only_never_ready():
sections = {
"Opis projektu": "x" * 250,
"Budżet": "y" * 250,
}
def evaluate(state):
return evaluate_application(
sections=state.get("generated_sections") or {},
brief=BRIEF,
grounding_mode="structure_only",
)
result = run_expectation_loop(
evaluate=evaluate,
apply_fixes=lambda s, r: s,
initial_state={
"generated_sections": sections,
"external_context": {"grounding_mode": "structure_only"},
"grounding_mode": "structure_only",
},
max_iterations=2,
)
assert result.ready is False
assert result.regulation_grounded_pass is False
assert result.stop_reason in ("max_iter", "blocked_grounding")
def test_quality_expectation_step_wired_for_production():
"""Production helper run_quality_expectation_step drives real advisor + fix path."""
from core.generation.quality_loop import run_quality_expectation_step
state = {
"sections_plan": [
{"title": "Opis projektu", "type": "desc"},
{"title": "Budżet", "type": "budget"},
],
"generated_sections": {
"Opis projektu": "Pełny opis projektu MŚP z DNSH i celami. " * 8,
"Budżet": "Budżet z wkładem własnym 30% i kosztami kwalifikowalnymi. " * 8,
},
"external_context": {
"grounding_mode": "regulation",
"advisor_brief": BRIEF,
"required_sections": BRIEF["required_sections"],
"key_rules": BRIEF["key_rules"],
},
}
out = run_quality_expectation_step(state, source="holistic", min_score=70)
assert "advisor_before" in out
assert "regulation_ready" in out
assert out["stop_reason"] in ("ready", "needs_retry")
# Strong content with brief signals should be ready without needing rewrite
assert out["regulation_ready"] is True
assert out["stop_reason"] == "ready"
def test_soft_pass_gate_requires_regulation_grounded_pass_semantics():
"""Mirror generator gate: usable brief → only regulation_grounded_pass is ready."""
from core.generation.expectation_loop import is_regulation_ready
fluff_rep = evaluate_application(
sections={
"Opis projektu": "Projekt innowacyjny z bogatym doświadczeniem zespołu. " * 20,
"Budżet": "Budżet obejmuje koszty osobowe i sprzęt w pełnym zakresie. " * 20,
},
brief=BRIEF,
grounding_mode="regulation",
)
assert fluff_rep.regulation_grounded_pass is False
assert is_regulation_ready(fluff_rep) is False
def test_run_advisor_expectation_loop_improves_to_ready():
state = {
"generated_sections": {
"Opis projektu": "krótki start",
},
"external_context": {
"grounding_mode": "regulation",
"advisor_brief": BRIEF,
"required_sections": BRIEF["required_sections"],
},
}
def pure_fix(st, report):
# Simulate targeted rewrite filling required sections
st = dict(st)
st["generated_sections"] = {
"Opis projektu": (
"Pełny opis projektu MŚP z DNSH, innowacją i celami programu. " * 8
),
"Budżet": (
"Szczegółowy budżet, wkład własny 30%, koszty kwalifikowalne. " * 8
),
}
st["fixed_sections"] = ["Opis projektu", "Budżet"]
return st
result = run_advisor_expectation_loop(
state,
max_iterations=3,
min_score=70,
apply_fixes=pure_fix,
)
assert result.ready is True
assert result.stop_reason == "ready"
assert result.regulation_grounded_pass is True
|