Принять agent replay маржинальности номенклатуры

This commit is contained in:
2026-05-22 16:21:59 +03:00
parent 09c6d1aa0e
commit 0bd631c160
4 changed files with 298 additions and 1 deletions
+97 -1
View File
@@ -386,6 +386,85 @@ def normalize_optional_bool(value: Any) -> bool | None:
return None
LEGACY_DEBUG_RESULT_MODES = {"confirmed_balance", "heuristic_candidates"}
BUSINESS_EXPECTED_RESULT_MODES = {
"clarification_required",
"limited_accounting_answer",
"evidence_or_honest_boundary",
"ranking_or_limited_accounting_answer",
"same_inventory_margin_context_or_clarification",
}
def _business_review_is_clean(step_state: dict[str, Any]) -> bool:
business_review = step_state.get("business_first_review")
if not isinstance(business_review, dict):
return True
return len(dcl.normalize_string_list(business_review.get("issue_codes"))) == 0
def business_expected_result_mode_matches(expected_result_mode: str, step_state: dict[str, Any]) -> bool:
reply_type = str(step_state.get("reply_type") or "").strip()
response_type = str(step_state.get("response_type") or "").strip()
truth_mode = str(step_state.get("truth_mode") or "").strip()
answer_shape = str(step_state.get("answer_shape") or "").strip()
detected_intent = str(step_state.get("detected_intent") or "").strip()
capability_id = str(step_state.get("capability_id") or "").strip()
assistant_text = str(step_state.get("assistant_text") or "").strip()
clean_business_review = _business_review_is_clean(step_state)
in_margin_context = (
detected_intent == "inventory_margin_ranking_for_nomenclature"
or capability_id == "inventory_inventory_margin_ranking_for_nomenclature"
)
if expected_result_mode == "clarification_required":
return (
clean_business_review
and (
truth_mode == "clarification_required"
or answer_shape == "clarification_required"
or (reply_type == "partial_coverage" and response_type == "LIMITED_WITH_REASON")
)
)
if expected_result_mode == "limited_accounting_answer":
return (
clean_business_review
and in_margin_context
and bool(assistant_text)
and reply_type in {"partial_coverage", "factual", "factual_with_explanation"}
)
if expected_result_mode == "evidence_or_honest_boundary":
return (
clean_business_review
and bool(assistant_text)
and reply_type in {"partial_coverage", "factual", "factual_with_explanation"}
)
if expected_result_mode == "ranking_or_limited_accounting_answer":
return (
clean_business_review
and in_margin_context
and bool(assistant_text)
and reply_type in {"partial_coverage", "factual", "factual_with_explanation"}
)
if expected_result_mode == "same_inventory_margin_context_or_clarification":
return (
clean_business_review
and bool(assistant_text)
and (
in_margin_context
or truth_mode == "clarification_required"
or answer_shape == "clarification_required"
)
and reply_type in {"partial_coverage", "factual", "factual_with_explanation"}
)
return False
def evaluate_truth_step(
*,
step: dict[str, Any],
@@ -627,7 +706,24 @@ def evaluate_truth_step(
resolve_nested_placeholders(step.get("expected_result_mode"), step_results, bindings, runtime_bindings) or ""
).strip()
actual_result_mode = str(step_state.get("result_mode") or "").strip()
if expected_result_mode and actual_result_mode and not dcl.identifiers_match(actual_result_mode, expected_result_mode):
if expected_result_mode in BUSINESS_EXPECTED_RESULT_MODES:
if not business_expected_result_mode_matches(expected_result_mode, step_state):
append_finding(
findings,
step,
"wrong_result_mode",
"Business answer mode does not match the expected semantic answer contract.",
actual={
"result_mode": actual_result_mode or None,
"reply_type": reply_type or None,
"truth_mode": step_state.get("truth_mode"),
"answer_shape": step_state.get("answer_shape"),
"intent": detected_intent or None,
"capability": capability_id or None,
},
expected=expected_result_mode,
)
elif expected_result_mode and actual_result_mode and not dcl.identifiers_match(actual_result_mode, expected_result_mode):
append_finding(
findings,
step,