Усилить семантический AGENT gate для 1С

This commit is contained in:
2026-05-23 22:01:29 +03:00
parent 75e0e2c66e
commit a4dd9c7c66
9 changed files with 634 additions and 13 deletions
+3
View File
@@ -118,6 +118,9 @@ BUSINESS_TOP_LINE_SCAFFOLD_MARKERS = (
"\u043e\u0433\u0440\u0430\u043d\u0438\u0447\u0435\u043d\u043d\u044b\u0439 \u0431\u0438\u0437\u043d\u0435\u0441-\u043e\u0431\u0437\u043e\u0440",
"\u0447\u0442\u043e \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u043e",
"\u043f\u0440\u043e\u0432\u0435\u0440\u0435\u043d\u043d\u044b\u0435 \u043a\u043e\u043d\u0442\u0443\u0440\u044b",
"\u043f\u043e \u0434\u0430\u043d\u043d\u044b\u043c 1\u0441 \u043d\u0430\u0439\u0434\u0435\u043d\u044b",
"\u043d\u0430\u0439\u0434\u0435\u043d\u044b \u0441\u0442\u0440\u043e\u043a\u0438",
"\u0441\u0443\u043c\u043c\u0443 \u043c\u043e\u0436\u043d\u043e \u043d\u0430\u0437\u044b\u0432\u0430\u0442\u044c \u0442\u043e\u043b\u044c\u043a\u043e",
"\u0431\u043b\u043e\u043a 1",
"\u0441\u0442\u0430\u0442\u0443\u0441",
)
+23 -3
View File
@@ -1156,8 +1156,15 @@ def build_business_review_summary(spec: dict[str, Any], scenario_state: dict[str
step_outputs = scenario_state.get("step_outputs") if isinstance(scenario_state.get("step_outputs"), dict) else {}
steps: list[dict[str, Any]] = []
issue_counts: dict[str, int] = {}
review_failures = 0
review_warnings = 0
for index, step in enumerate(spec["steps"], start=1):
step_state = step_outputs.get(step["step_id"], {})
review_status = step_state.get("review_status") if isinstance(step_state, dict) else None
if review_status == "fail":
review_failures += 1
elif review_status == "warning":
review_warnings += 1
business_review = (
step_state.get("business_first_review")
if isinstance(step_state, dict) and isinstance(step_state.get("business_first_review"), dict)
@@ -1171,7 +1178,7 @@ def build_business_review_summary(spec: dict[str, Any], scenario_state: dict[str
"index": index,
"step_id": step["step_id"],
"question": step["question_template"],
"review_status": step_state.get("review_status") if isinstance(step_state, dict) else None,
"review_status": review_status,
"direct_answer": business_review.get("actual_direct_answer"),
"answer_length_chars": business_review.get("answer_length_chars"),
"direct_answer_required": business_review.get("direct_answer_required"),
@@ -1193,6 +1200,7 @@ def build_business_review_summary(spec: dict[str, Any], scenario_state: dict[str
)
)
warnings = sum(1 for step in steps if "business_answer_too_verbose" in step["issue_codes"])
semantic_status = "fail" if failed or review_failures else ("warning" if warnings or review_warnings else "pass")
return {
"schema_version": "business_first_run_review_v1",
"scenario_id": spec["scenario_id"],
@@ -1202,24 +1210,32 @@ def build_business_review_summary(spec: dict[str, Any], scenario_state: dict[str
"steps_total": len(steps),
"steps_with_business_failures": failed,
"steps_with_business_warnings": warnings,
"steps_with_review_failures": review_failures,
"steps_with_review_warnings": review_warnings,
"issue_counts": issue_counts,
"overall_business_status": "fail" if failed else ("warning" if warnings else "pass"),
"overall_semantic_status": semantic_status,
"semantic_gate_passed": semantic_status == "pass",
"steps": steps,
}
def build_business_review_markdown(business_review: dict[str, Any]) -> str:
def build_business_review_markdown(business_review: dict[str, Any], *, title: str = "Business-first review") -> str:
lines = [
"# Business-first review",
f"# {title}",
"",
f"- scenario_id: `{business_review.get('scenario_id') or 'n/a'}`",
f"- domain: `{business_review.get('domain') or 'n/a'}`",
f"- title: {business_review.get('title') or 'n/a'}",
f"- session_id: `{business_review.get('session_id') or 'n/a'}`",
f"- overall_business_status: `{business_review.get('overall_business_status') or 'n/a'}`",
f"- overall_semantic_status: `{business_review.get('overall_semantic_status') or 'n/a'}`",
f"- semantic_gate_passed: `{business_review.get('semantic_gate_passed') is True}`",
f"- steps_total: `{business_review.get('steps_total')}`",
f"- steps_with_business_failures: `{business_review.get('steps_with_business_failures')}`",
f"- steps_with_business_warnings: `{business_review.get('steps_with_business_warnings')}`",
f"- steps_with_review_failures: `{business_review.get('steps_with_review_failures')}`",
f"- steps_with_review_warnings: `{business_review.get('steps_with_review_warnings')}`",
f"- issue_counts: `{dump_json(business_review.get('issue_counts') or {})}`",
"",
"## Human Answer Surface",
@@ -1479,6 +1495,8 @@ def review_export(spec: dict[str, Any], export_path: Path, output_dir: Path) ->
write_text(output_dir / "truth_review.md", review_markdown)
write_json(output_dir / "business_review.json", business_review)
write_text(output_dir / "business_review.md", build_business_review_markdown(business_review))
write_json(output_dir / "semantic_audit.json", business_review)
write_text(output_dir / "semantic_audit.md", build_business_review_markdown(business_review, title="Semantic audit"))
acceptance_bundle = write_acceptance_artifacts(output_dir, spec, scenario_state, review_summary)
return {
"scenario_state": scenario_state,
@@ -1575,6 +1593,8 @@ def run_live(spec: dict[str, Any], output_dir: Path, args: argparse.Namespace) -
write_text(output_dir / "truth_review.md", review_markdown)
write_json(output_dir / "business_review.json", business_review)
write_text(output_dir / "business_review.md", build_business_review_markdown(business_review))
write_json(output_dir / "semantic_audit.json", business_review)
write_text(output_dir / "semantic_audit.md", build_business_review_markdown(business_review, title="Semantic audit"))
acceptance_bundle = write_acceptance_artifacts(output_dir, spec, scenario_state, review_summary)
print(f"[truth-harness] saved artifacts to {output_dir}")
print(f"[truth-harness] overall_status={review_summary['overall_status']}")
@@ -927,6 +927,91 @@ class DomainCaseLoopStepStateTests(unittest.TestCase):
self.assertIn("technical_garbage_in_answer", review["issue_codes"])
self.assertNotIn("business_direct_answer_missing", review["issue_codes"])
def test_business_first_review_rejects_found_rows_scaffold_as_direct_answer(self) -> None:
question = "\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u0432\u0445\u043e\u0434\u044f\u0449\u0438\u0445 \u0434\u0435\u043d\u0435\u0433 \u0437\u0430 2020"
step_state = dcl.build_scenario_step_state(
scenario_id="semantic_gate_demo",
domain="value_flow",
step={
"step_id": "step_01",
"title": "Direct money answer",
"depends_on": [],
"question_template": question,
},
step_index=1,
question_resolved=question,
analysis_context={},
turn_artifact={
"assistant_message": {
"reply_type": "factual",
"text": "\u041a\u043e\u0440\u043e\u0442\u043a\u043e: \u041f\u043e \u0434\u0430\u043d\u043d\u044b\u043c 1\u0421 \u043d\u0430\u0439\u0434\u0435\u043d\u044b \u0441\u0442\u0440\u043e\u043a\u0438 \u0432\u0445\u043e\u0434\u044f\u0449\u0438\u0445 \u0434\u0435\u043d\u0435\u0436\u043d\u044b\u0445 \u043f\u043e\u0441\u0442\u0443\u043f\u043b\u0435\u043d\u0438\u0439; \u0441\u0443\u043c\u043c\u0443 \u043c\u043e\u0436\u043d\u043e \u043d\u0430\u0437\u044b\u0432\u0430\u0442\u044c \u0442\u043e\u043b\u044c\u043a\u043e \u0432 \u0440\u0430\u043c\u043a\u0430\u0445 \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0438.\n\u0427\u0442\u043e \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u043e: 47 628 853,03 \u0440\u0443\u0431.",
"message_id": "msg-1",
"trace_id": "trace-1",
},
"technical_debug_payload": {},
"session_summary": {},
},
entries=[],
)
review = step_state["business_first_review"]
self.assertFalse(review["direct_answer_first_ok"])
self.assertFalse(review["answer_layering_ok"])
self.assertIn("business_direct_answer_missing", review["issue_codes"])
self.assertIn("answer_layering_noise", review["issue_codes"])
def test_business_first_review_allows_direct_answer_with_evidence_boundary(self) -> None:
question = "\u043a\u0430\u043a\u043e\u0435 \u043d\u0435\u0442\u0442\u043e \u043f\u043e \u0434\u0435\u043d\u044c\u0433\u0430\u043c \u0441 \u0413\u0440\u0443\u043f\u043f\u0430 \u0421\u0412\u041a \u0437\u0430 2020"
step_state = dcl.build_scenario_step_state(
scenario_id="semantic_gate_demo",
domain="value_flow",
step={
"step_id": "step_01",
"title": "Counterparty net flow",
"depends_on": [],
"question_template": question,
},
step_index=1,
question_resolved=question,
analysis_context={},
turn_artifact={
"assistant_message": {
"reply_type": "factual_with_explanation",
"text": "\u041a\u043e\u0440\u043e\u0442\u043a\u043e: \u043f\u043e \u043a\u043e\u043d\u0442\u0440\u0430\u0433\u0435\u043d\u0442\u0443 \u0413\u0440\u0443\u043f\u043f\u0430 \u0421\u0412\u041a \u0437\u0430 2020 \u043f\u043e \u043d\u0430\u0439\u0434\u0435\u043d\u043d\u044b\u043c \u0441\u0442\u0440\u043e\u043a\u0430\u043c 1\u0421 \u043f\u043e\u043b\u0443\u0447\u0438\u043b\u0438 12 093 465 \u0440\u0443\u0431., \u0437\u0430\u043f\u043b\u0430\u0442\u0438\u043b\u0438 0 \u0440\u0443\u0431.; \u043d\u0435\u0442\u0442\u043e \u0432 \u043d\u0430\u0448\u0443 \u0441\u0442\u043e\u0440\u043e\u043d\u0443 12 093 465 \u0440\u0443\u0431.",
"message_id": "msg-1",
"trace_id": "trace-1",
},
"technical_debug_payload": {},
"session_summary": {},
},
entries=[],
)
review = step_state["business_first_review"]
self.assertTrue(review["direct_answer_first_ok"])
self.assertTrue(review["answer_layering_ok"])
self.assertEqual(review["issue_codes"], [])
def test_semantic_audit_markdown_exposes_semantic_gate(self) -> None:
markdown = dth.build_business_review_markdown(
{
"scenario_id": "semantic_gate_demo",
"domain": "value_flow",
"title": "Semantic gate demo",
"session_id": "session-1",
"overall_business_status": "fail",
"steps_total": 1,
"steps_with_business_failures": 1,
"steps_with_business_warnings": 0,
"issue_counts": {"answer_layering_noise": 1},
"steps": [],
},
title="Semantic audit",
)
self.assertIn("# Semantic audit", markdown)
self.assertIn("semantic_gate_passed: `False`", markdown)
def test_truth_harness_promotes_business_review_issues_to_findings(self) -> None:
step_state = dcl.build_scenario_step_state(
scenario_id="business_surface_demo",