wangwei
f5c2dce64a
- rag_eval/advisor/: new package with rules engine, LLM analyzer, writer - rules.py: 7-metric diagnostic rules (warning/critical thresholds, top-3 low samples) - llm_analyzer.py: Chinese optimization report via judge_model, graceful fallback - writer.py: writes optimization_advice.md + log summary - __init__.py: run_advisor() entry point (no-op when optimization_advisor=False) - Scenario.optimization_advisor: new bool field (default False) - ScenarioModel: same field added, loader.py透传 - RunArtifactPaths.advice_md: new path field - factory.py: build_models() now public; build_metric_pipeline() accepts pre-built llm/embeddings - runner.py: lifts llm, passes to pipeline and advisor; calls run_advisor() at end - siemens online YAML: optimization_advisor: true enabled - tests: 9 rules tests + 6 writer tests, all pass - docs: advisor section added to engine-flow.md and architecture.md Co-Authored-By: Claude <noreply@anthropic.com>
114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
import shutil
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from rag_eval.advisor.rules import Diagnosis
|
|
from rag_eval.advisor.writer import write_advice, _format_log_summary
|
|
|
|
|
|
class TestWriteAdvice(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = Path("tests/.tmp/test_advisor_writer")
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
self.tmp.mkdir(parents=True, exist_ok=True)
|
|
self.advice_path = self.tmp / "optimization_advice.md"
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
|
|
def _make_diagnosis(self, metric="faithfulness", severity="warning"):
|
|
return Diagnosis(
|
|
metric=metric,
|
|
mean_score=0.55,
|
|
threshold=0.7,
|
|
severity=severity,
|
|
root_causes=["原因1", "原因2"],
|
|
suggested_actions=["建议1", "建议2"],
|
|
low_samples=[
|
|
{"sample_id": "s1", "question": "问题1", "answer": "答案1",
|
|
"ground_truth": "标准1", metric: 0.4},
|
|
],
|
|
)
|
|
|
|
def test_write_creates_file(self):
|
|
diag = self._make_diagnosis()
|
|
write_advice(
|
|
diagnoses=[diag],
|
|
llm_markdown="## faithfulness\n\nLLM 建议内容",
|
|
advice_path=self.advice_path,
|
|
scenario_name="test-scenario",
|
|
run_id="2026-01-01T00-00-00",
|
|
judge_model="deepseek-v4-flash",
|
|
)
|
|
self.assertTrue(self.advice_path.exists())
|
|
|
|
def test_write_contains_scenario_name_and_run_id(self):
|
|
diag = self._make_diagnosis()
|
|
write_advice(
|
|
diagnoses=[diag],
|
|
llm_markdown="## faithfulness\n\nLLM 建议",
|
|
advice_path=self.advice_path,
|
|
scenario_name="siemens-test",
|
|
run_id="2026-01-01T00-00-00",
|
|
judge_model="deepseek-v4-flash",
|
|
)
|
|
content = self.advice_path.read_text(encoding="utf-8")
|
|
self.assertIn("siemens-test", content)
|
|
self.assertIn("2026-01-01T00-00-00", content)
|
|
|
|
def test_write_contains_llm_markdown(self):
|
|
diag = self._make_diagnosis()
|
|
write_advice(
|
|
diagnoses=[diag],
|
|
llm_markdown="## faithfulness\n\n具体建议文本",
|
|
advice_path=self.advice_path,
|
|
scenario_name="test",
|
|
run_id="rid",
|
|
judge_model="model",
|
|
)
|
|
content = self.advice_path.read_text(encoding="utf-8")
|
|
self.assertIn("具体建议文本", content)
|
|
|
|
def test_write_fallback_when_no_llm_markdown(self):
|
|
"""When llm_markdown is empty, writer emits rule-only report."""
|
|
diag = self._make_diagnosis()
|
|
write_advice(
|
|
diagnoses=[diag],
|
|
llm_markdown="",
|
|
advice_path=self.advice_path,
|
|
scenario_name="test",
|
|
run_id="rid",
|
|
judge_model="model",
|
|
)
|
|
content = self.advice_path.read_text(encoding="utf-8")
|
|
self.assertIn("faithfulness", content)
|
|
self.assertIn("原因1", content)
|
|
|
|
def test_log_summary_format(self):
|
|
diags = [
|
|
self._make_diagnosis("faithfulness", "critical"),
|
|
self._make_diagnosis("context_recall", "warning"),
|
|
]
|
|
summary = _format_log_summary(diags, self.advice_path)
|
|
self.assertIn("faithfulness", summary)
|
|
self.assertIn("critical", summary)
|
|
self.assertIn("context_recall", summary)
|
|
self.assertIn("warning", summary)
|
|
|
|
def test_write_empty_diagnoses_still_creates_file(self):
|
|
write_advice(
|
|
diagnoses=[],
|
|
llm_markdown="",
|
|
advice_path=self.advice_path,
|
|
scenario_name="test",
|
|
run_id="rid",
|
|
judge_model="model",
|
|
)
|
|
self.assertTrue(self.advice_path.exists())
|
|
content = self.advice_path.read_text(encoding="utf-8")
|
|
self.assertIn("未发现明显指标异常", content)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|