SILVER
/
NODEDC_1C
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Укрепить валидацию агентного лупа по сценарным якорям

This commit is contained in:
dctouch 2026-06-15 20:01:22 +03:00
parent bbb3620956
commit e2b928ac19
2 changed files with 188 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
scripts/domain_case_loop.py
View File

@ -1695,6 +1695,47 @@ def resolve_question_template(question_template: str, scenario_state: dict[str,
return pattern.sub(replace, question_template).strip()
def resolve_scenario_contract_placeholders(value: Any, scenario_state: dict[str, Any]) -> Any:
if isinstance(value, str):
runtime_resolved = str(resolve_runtime_placeholders(value))
if "{{" not in runtime_resolved:
return runtime_resolved
pattern = re.compile(r"{{\s*([^{}]+?)\s*}}")
def replace(match: re.Match[str]) -> str:
value = lookup_placeholder_value(match.group(1), scenario_state)
if isinstance(value, (dict, list)):
return dump_json(value)
return str(value)
return pattern.sub(replace, runtime_resolved)
if isinstance(value, list):
return [resolve_scenario_contract_placeholders(item, scenario_state) for item in value]
if isinstance(value, dict):
return {
key: resolve_scenario_contract_placeholders(item, scenario_state)
for key, item in value.items()
}
return value
def resolve_step_contract_placeholders(step: dict[str, Any], scenario_state: dict[str, Any]) -> dict[str, Any]:
resolved_step = dict(step)
resolved_step["required_filters"] = normalize_validation_filters(
resolve_scenario_contract_placeholders(step.get("required_filters") or {}, scenario_state)
)
for pattern_field in (
"forbidden_answer_patterns",
"required_answer_patterns_any",
"required_answer_patterns_all",
):
if pattern_field in step:
resolved_step[pattern_field] = normalize_pattern_list(
resolve_scenario_contract_placeholders(step.get(pattern_field) or [], scenario_state)
)
return resolved_step
def build_failed_step_state(
*,
scenario_id: str,
@ -2313,6 +2354,9 @@ def step_intent_matches_expected(state: dict[str, Any], expected_intents: list[s
observed_intents = [
str(state.get("detected_intent") or "").strip(),
*normalize_string_list(state.get("mcp_discovery_effective_intents")),
str(state.get("mcp_discovery_selected_chain_id") or "").strip(),
str(state.get("mcp_discovery_route_candidate_fact_family") or "").strip(),
str(state.get("mcp_discovery_route_candidate_action_family") or "").strip(),
]
return any(identifier_in_list(intent, expected_intents) for intent in observed_intents if intent)
@ -2551,6 +2595,8 @@ def is_validated_memory_checkpoint_answer(
) -> bool:
if is_validated_comparison_boundary_memory_checkpoint(state, business_review, violations):
return True
if is_validated_bare_organization_scope_checkpoint(state, business_review, violations):
return True
tags = set(normalize_string_list(state.get("semantic_tags")))
if "memory" not in tags:
return False
@ -2566,6 +2612,39 @@ def is_validated_memory_checkpoint_answer(
)
def is_validated_bare_organization_scope_checkpoint(
state: dict[str, Any],
business_review: dict[str, Any],
violations: list[str],
) -> bool:
if violations:
return False
tags = set(normalize_string_list(state.get("semantic_tags")))
if not tags.intersection({"bare_org_scope", "company_scope", "company_selected"}):
return False
if str(state.get("living_chat_response_source") or "").strip() != "deterministic_data_scope_selection_contract":
return False
if str(state.get("living_chat_data_scope_probe_status") or "").strip() != "resolved":
return False
selected_organization = first_non_empty_string(
state.get("living_chat_selected_organization"),
state.get("assistant_active_organization"),
)
if not selected_organization:
return False
assistant_text = str(state.get("assistant_text") or "")
normalized_answer = normalize_entity_filter_value_for_comparison(assistant_text)
normalized_organization = normalize_entity_filter_value_for_comparison(selected_organization)
if not normalized_organization or normalized_organization not in normalized_answer:
return False
return (
business_review.get("business_usefulness_ok") is True
and business_review.get("direct_answer_first_ok") is True
and business_review.get("answer_layering_ok") is True
and business_review.get("technical_garbage_present") is False
)
def is_validated_comparison_boundary_memory_checkpoint(
state: dict[str, Any],
business_review: dict[str, Any],
@ -3235,6 +3314,10 @@ def build_scenario_step_state(
"detected_intent": debug.get("detected_intent"),
"selected_recipe": debug.get("selected_recipe"),
"capability_id": debug.get("capability_id"),
"living_chat_response_source": debug.get("living_chat_response_source"),
"living_chat_data_scope_probe_status": debug.get("living_chat_data_scope_probe_status"),
"living_chat_selected_organization": debug.get("living_chat_selected_organization"),
"assistant_active_organization": debug.get("assistant_active_organization"),
"capability_contract_id": first_non_empty_string(
debug.get("capability_contract_id"),
runtime_contract.get("capability_contract_id"),
@ -3600,6 +3683,7 @@ def execute_scenario_manifest(
step_dir = steps_dir / step["step_id"]
try:
resolved_question = resolve_question_template(step["question_template"], scenario_state)
resolved_step = resolve_step_contract_placeholders(step, scenario_state)
step_analysis_context = merge_analysis_context(manifest.get("analysis_context"), step.get("analysis_context"))
step_analysis_context = carry_forward_analysis_context(
scenario_state,
@ -3611,7 +3695,7 @@ def execute_scenario_manifest(
args=args,
domain=manifest["domain"],
scenario_id=manifest["scenario_id"],
step=step,
step=resolved_step,
step_index=step_index,
session_id=scenario_state.get("session_id"),
question_resolved=resolved_question,

103
scripts/test_domain_case_loop_step_state.py
View File

@ -190,6 +190,109 @@ class DomainCaseLoopStepStateTests(unittest.TestCase):
self.assertIn("required_answer_patterns_any_missing", step_state["violated_invariants"])
self.assertEqual(step_state["acceptance_status"], "rejected")
def test_binding_placeholder_required_filter_is_resolved_before_step_validation(self) -> None:
step = dcl.normalize_step_definition(
1,
{
"step_id": "counterparty_binding_filter",
"title": "Counterparty binding filter",
"question": "Отдельно по {{bindings.counterparty}}: какой это поток?",
"required_filters": {"counterparty": "{{bindings.counterparty}}"},
"required_answer_patterns_all": ["(?i){{bindings.counterparty}}|банк"],
},
)
scenario_state = {
"bindings": {"counterparty": "СБЕРБАНК"},
"step_outputs": {},
"semantic_memory": {},
}
resolved_step = dcl.resolve_step_contract_placeholders(step, scenario_state)
self.assertEqual(resolved_step["required_filters"]["counterparty"], "СБЕРБАНК")
self.assertEqual(resolved_step["required_answer_patterns_all"], ["(?i)СБЕРБАНК|банк"])
step_state = dcl.build_scenario_step_state(
scenario_id="binding_filter_demo",
domain="agentic_loop",
step=resolved_step,
step_index=1,
question_resolved="Отдельно по СБЕРБАНК: какой это поток?",
analysis_context={},
turn_artifact={
"assistant_message": {
"reply_type": "factual",
"text": "СБЕРБАНК подтвержден как банковский/финансовый поток.",
"message_id": "msg-1",
"trace_id": "trace-1",
},
"technical_debug_payload": {
"detected_mode": "address_query",
"detected_intent": "bank_operations_by_counterparty",
"selected_recipe": "address_bank_operations_by_counterparty_v1",
"capability_id": "bank_operations_drilldown",
"truth_gate_contract_status": "full_confirmed",
"mcp_call_status": "matched_non_empty",
"extracted_filters": {"counterparty": "СБЕРБАНК"},
},
"session_summary": {},
},
entries=[],
)
self.assertNotIn("missing_required_filter", step_state["violated_invariants"])
def test_expected_intent_accepts_discovery_chain_id(self) -> None:
self.assertTrue(
dcl.step_intent_matches_expected(
{
"detected_intent": "supplier_payouts_profile",
"mcp_discovery_selected_chain_id": "value_flow",
"mcp_discovery_route_candidate_fact_family": "value_flow",
"mcp_discovery_route_candidate_action_family": "payout",
},
["value_flow"],
)
)
def test_bare_organization_scope_selection_is_validated_memory_checkpoint(self) -> None:
step_state = dcl.build_scenario_step_state(
scenario_id="org_scope_demo",
domain="agentic_loop",
step={
"step_id": "step_01",
"title": "Choose organization",
"depends_on": [],
"question_template": "ООО Альтернатива Плюс",
"semantic_tags": ["bare_org_scope", "company_scope"],
"required_answer_patterns_all": ["(?i)фиксир|рабоч.*организац|контур"],
},
step_index=1,
question_resolved="ООО Альтернатива Плюс",
analysis_context={},
turn_artifact={
"assistant_message": {
"reply_type": "factual_with_explanation",
"text": 'Отлично, фиксирую рабочую организацию: ООО "Альтернатива Плюс".',
"message_id": "msg-1",
"trace_id": "trace-1",
},
"technical_debug_payload": {
"detected_mode": "chat",
"fallback_type": "none",
"living_chat_response_source": "deterministic_data_scope_selection_contract",
"living_chat_data_scope_probe_status": "resolved",
"living_chat_selected_organization": 'ООО "Альтернатива Плюс"',
"assistant_active_organization": 'ООО "Альтернатива Плюс"',
},
"session_summary": {},
},
entries=[],
)
self.assertEqual(step_state["status"], "validated")
self.assertTrue(step_state["memory_checkpoint_validated"])
def test_repair_targets_promote_route_candidate_enablement_gaps(self) -> None:
repair_targets = dcl.build_deterministic_repair_targets(
{"pack_id": "route_candidate_pack", "domain": "open_world", "final_status": "accepted"},