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

Усилить answer contract и агентный аудит для phase105

Browse Source
This commit is contained in:
dctouch
2026-05-21 09:00:08 +03:00
parent 9c86407937
commit bbc257fd6c
34 changed files with 1533 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/domain_case_loop.py
+1
View File
@@ -184,6 +184,7 @@ DEFAULT_INVARIANT_SEVERITY: dict[str, str] = {
"top_level_noise_present": "P0",
"business_direct_answer_missing": "P0",
"technical_garbage_in_answer": "P0",
"counterparty_value_flow_misrouted_to_company_profit": "P0",
"answer_layering_noise": "P1",
"business_answer_too_verbose": "P1",
}
scripts/domain_truth_harness.py
+2
View File
@@ -322,6 +322,7 @@ def append_finding(
BUSINESS_REVIEW_FINDING_MESSAGES = {
"technical_garbage_in_answer": "User-facing answer leaked internal runtime or MCP identifiers.",
"business_direct_answer_missing": "The answer did not put the direct business answer first.",
"counterparty_value_flow_misrouted_to_company_profit": "Counterparty received/paid/net flow question was answered with company profit instead of counterparty cashflow.",
"answer_layering_noise": "The answer opened with scaffolding or report framing instead of a clean business result.",
"business_answer_too_verbose": "The answer is too verbose for a direct business question.",
}
@@ -329,6 +330,7 @@ BUSINESS_REVIEW_FINDING_MESSAGES = {
BUSINESS_REVIEW_FINDING_SEVERITY = {
"technical_garbage_in_answer": "critical",
"business_direct_answer_missing": "critical",
"counterparty_value_flow_misrouted_to_company_profit": "critical",
"answer_layering_noise": "critical",
"business_answer_too_verbose": "warning",
}
scripts/review_assistant_stage1_run.py
+65
View File
@@ -111,6 +111,30 @@ SAFE_FINANCIAL_BOUNDARY_MARKERS = (
"без назначения платеж",
"без договора",
)
COUNTERPARTY_VALUE_FLOW_QUESTION_RE = re.compile(
"(?:\u0441\u043a\u043e\u043b\u044c\u043a\u043e\\s+.*\u043f\u043e\u043b\u0443\u0447|\u0441\u043a\u043e\u043b\u044c\u043a\u043e\\s+.*\u0437\u0430\u043f\u043b\u0430\u0442|\u043a\u0430\u043a\u043e\u0435\\s+\u043d\u0435\u0442\u0442\u043e|\u043a\u0430\u043a\u043e\u0435\\s+\u0441\u0430\u043b\u044c\u0434\u043e)",
re.IGNORECASE,
)
COUNTERPARTY_SCOPE_QUESTION_RE = re.compile(
"(?:\u043a\u043e\u043d\u0442\u0440\u0430\u0433\u0435\u043d\u0442|\u043f\u043e\u0441\u0442\u0430\u0432\u0449\u0438\u043a|\u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b|\u043a\u043b\u0438\u0435\u043d\u0442|\u0441\u0432\u043a|\u0447\u0435\u043f\u0443\u0440\u043d\u043e\u0432)",
re.IGNORECASE,
)
COUNTERPARTY_VALUE_FLOW_ANSWER_RE = re.compile(
"(?:\u043f\u043e\u043b\u0443\u0447\u0438\u043b\u0438|\u043f\u043e\u043b\u0443\u0447\u0435\u043d\u043e|\u0437\u0430\u043f\u043b\u0430\u0442\u0438\u043b\u0438|\u043e\u043f\u043b\u0430\u0442\u0438\u043b\u0438|\u043d\u0435\u0442\u0442\u043e|\u0441\u0430\u043b\u044c\u0434\u043e|\u0432\\s+\u043d\u0430\u0448\u0443\\s+\u0441\u0442\u043e\u0440\u043e\u043d\u0443|\u0438\u0441\u0445\u043e\u0434\u044f\u0449\u0438\u0445\\s+\u043f\u043b\u0430\u0442\u0435\u0436\u0435\u0439|\u0432\u0445\u043e\u0434\u044f\u0449\u0438\u0445\\s+\u043f\u043e\u0441\u0442\u0443\u043f\u043b\u0435\u043d\u0438\u0439)",
re.IGNORECASE,
)
COUNTERPARTY_VALUE_FLOW_REQUIRED_ANSWER_RE = re.compile(
"(?:\u043f\u043e\u043b\u0443\u0447\u0438\u043b\u0438|\u043f\u043e\u043b\u0443\u0447\u0435\u043d\u043e|\u0437\u0430\u043f\u043b\u0430\u0442\u0438\u043b\u0438|\u043e\u043f\u043b\u0430\u0442\u0438\u043b\u0438|\u0438\u0441\u0445\u043e\u0434\u044f\u0449\u0438\u0445\\s+\u043f\u043b\u0430\u0442\u0435\u0436\u0435\u0439|\u0432\u0445\u043e\u0434\u044f\u0449\u0438\u0445\\s+\u043f\u043e\u0441\u0442\u0443\u043f\u043b\u0435\u043d\u0438\u0439)",
re.IGNORECASE,
)
COUNTERPARTY_ANSWER_SCOPE_RE = re.compile(
"(?:\u043a\u043e\u043d\u0442\u0440\u0430\u0433\u0435\u043d\u0442|\u043f\u043e\u0441\u0442\u0430\u0432\u0449\u0438\u043a|\u043f\u043e\u043a\u0443\u043f\u0430\u0442\u0435\u043b|\u043a\u043b\u0438\u0435\u043d\u0442|\u0441\u0432\u043a|\u0447\u0435\u043f\u0443\u0440\u043d\u043e\u0432)",
re.IGNORECASE,
)
COMPANY_PROFIT_ANSWER_RE = re.compile(
"(?:\u0447\u0438\u0441\u0442\u0430\u044f\\s+\u043f\u0440\u0438\u0431\u044b\u043b\u044c|\u043f\u0440\u0438\u0431\u044b\u043b\u044c\u044e|90/91/99|\u0444\u0438\u043d\u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442|\u0443\u0431\u044b\u0442\u043e\u043a)",
re.IGNORECASE,
)
def now_iso() -> str:
@@ -335,6 +359,7 @@ def build_step_for_pair(pair: dict[str, Any]) -> dict[str, Any]:
"answer_layering_noise": "P1",
"business_answer_too_verbose": "P1",
"bank_counterparty_misclassified_as_business_partner": "P1",
"counterparty_value_flow_misrouted_to_company_profit": "P0",
},
}
@@ -349,12 +374,41 @@ def marker_hits(text: str, markers: tuple[str, ...]) -> list[str]:
return [marker for marker in markers if marker and marker.casefold() in lowered]
def detect_counterparty_value_flow_profit_mismatch(question: str, assistant_text: str) -> dict[str, Any] | None:
question_text = str(question or "")
answer_text = str(assistant_text or "")
question_flow_match = COUNTERPARTY_VALUE_FLOW_QUESTION_RE.search(question_text)
question_scope_match = COUNTERPARTY_SCOPE_QUESTION_RE.search(question_text)
if not question_flow_match or not question_scope_match:
return None
profit_match = COMPANY_PROFIT_ANSWER_RE.search(answer_text)
if not profit_match:
return None
value_flow_match = COUNTERPARTY_VALUE_FLOW_ANSWER_RE.search(answer_text)
required_flow_match = COUNTERPARTY_VALUE_FLOW_REQUIRED_ANSWER_RE.search(answer_text)
answer_scope_match = COUNTERPARTY_ANSWER_SCOPE_RE.search(answer_text)
if required_flow_match and answer_scope_match:
return None
return {
"question_flow_hit": question_flow_match.group(0),
"question_scope_hit": question_scope_match.group(0),
"profit_hit": profit_match.group(0),
"value_flow_hit": value_flow_match.group(0) if value_flow_match else None,
"required_flow_hit": required_flow_match.group(0) if required_flow_match else None,
"answer_scope_hit": answer_scope_match.group(0) if answer_scope_match else None,
}
def augment_gui_business_review(step_state: dict[str, Any]) -> dict[str, Any]:
review = (
dict(step_state.get("business_first_review"))
if isinstance(step_state.get("business_first_review"), dict)
else {}
)
question = str(step_state.get("question_resolved") or step_state.get("question_template") or "")
assistant_text = str(step_state.get("assistant_text") or "")
issue_codes = [str(item) for item in review.get("issue_codes", []) if str(item).strip()]
root_layers = [str(item) for item in review.get("suggested_root_cause_layers", []) if str(item).strip()]
@@ -378,6 +432,17 @@ def augment_gui_business_review(step_state: dict[str, Any]) -> dict[str, Any]:
if "business_semantic_role_gap" not in root_layers:
root_layers.append("business_semantic_role_gap")
mismatch_details = detect_counterparty_value_flow_profit_mismatch(question, assistant_text)
if mismatch_details:
issue_code = "counterparty_value_flow_misrouted_to_company_profit"
if issue_code not in issue_codes:
issue_codes.append(issue_code)
if "followup_action_resolution_gap" not in root_layers:
root_layers.append("followup_action_resolution_gap")
if "answer_shape_mismatch" not in root_layers:
root_layers.append("answer_shape_mismatch")
review["semantic_mismatch_details"] = mismatch_details
review["technical_garbage_present"] = bool(technical_hits)
review["technical_garbage_hits"] = technical_hits
review["issue_codes"] = issue_codes
scripts/test_review_assistant_stage1_run.py
+80
View File
@@ -170,6 +170,86 @@ class AssistantStage1RunReviewTests(unittest.TestCase):
self.assertIn("technical_garbage_in_answer", review["summary"]["issue_counts"])
self.assertIn("bank_counterparty_misclassified_as_business_partner", review["summary"]["issue_counts"])
def test_review_flags_counterparty_net_flow_answer_that_slips_into_company_profit(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
sessions_dir = root / "sessions"
reports_dir = root / "reports"
run_id = "assistant-stage1-counterparty-profit-slip"
session_file = sessions_dir / f"{run_id}-SAVED-001.json"
report_file = reports_dir / f"{run_id}.md"
write_json(
session_file,
session_payload(
[
{
"role": "user",
"text": "А теперь по Группа СВК за 2020: сколько денег получили, сколько заплатили и какое нетто?",
},
{
"role": "assistant",
"text": "Нет, денежное операционное нетто не стоит считать чистой прибылью. "
"По закрытию 90/91/99 подтвержден учетный убыток −7 136 815,85 ₽.",
"reply_type": "factual_with_explanation",
"message_id": "a-counterparty-profit-slip",
"trace_id": "trace-counterparty-profit-slip",
"debug": {},
},
]
),
)
report_file.parent.mkdir(parents=True, exist_ok=True)
report_file.write_text(f"# Assistant Stage 1 Eval Run\n\n- run_id: {run_id}\n", encoding="utf-8")
review = reviewer.build_run_review(
run_id=run_id,
session_files=[session_file],
report_path=report_file,
)
self.assertEqual(review["summary"]["overall_business_status"], "fail")
self.assertIn("counterparty_value_flow_misrouted_to_company_profit", review["summary"]["issue_counts"])
target_by_issue = {item["issue_code"]: item for item in review["repair_targets"]}
self.assertEqual(target_by_issue["counterparty_value_flow_misrouted_to_company_profit"]["severity"], "P0")
def test_review_does_not_flag_counterparty_net_flow_when_received_paid_answer_is_present(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
sessions_dir = root / "sessions"
reports_dir = root / "reports"
run_id = "assistant-stage1-counterparty-net-clean"
session_file = sessions_dir / f"{run_id}-SAVED-001.json"
report_file = reports_dir / f"{run_id}.md"
write_json(
session_file,
session_payload(
[
{
"role": "user",
"text": "А теперь по Группа СВК за 2020: сколько денег получили, сколько заплатили и какое нетто?",
},
{
"role": "assistant",
"text": "По Группа СВК за 2020 получили 12 093 465 ₽, заплатили 0 ₽, денежное нетто +12 093 465 ₽ в нашу сторону.",
"reply_type": "factual",
"message_id": "a-counterparty-net-clean",
"trace_id": "trace-counterparty-net-clean",
"debug": {},
},
]
),
)
report_file.parent.mkdir(parents=True, exist_ok=True)
report_file.write_text(f"# Assistant Stage 1 Eval Run\n\n- run_id: {run_id}\n", encoding="utf-8")
review = reviewer.build_run_review(
run_id=run_id,
session_files=[session_file],
report_path=report_file,
)
self.assertNotIn("counterparty_value_flow_misrouted_to_company_profit", review["summary"]["issue_counts"])
def test_question_quality_treats_short_natural_followups_as_contextual(self) -> None:
pairs = [
{"pair_index": 1, "user": {"text": "приветик - че как там дела"}},