wangwensheng/AM516
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: paginate candidate model recommendations

Browse Source
This commit is contained in:
ar51
2026-09-01 09:48:01 +08:00
parent d97b788e0e
commit 8a695900f3
9 changed files with 598 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

210
tests/test_feishu_bot_bridge.py
View File

@@ -28,7 +28,36 @@ if "lark_oapi" not in sys.modules:
from shared.runtime import feishu_bot_bridge as bridge
def successful_payload() -> dict[str, object]:
return {
"data": {
"时间戳": "2026-08-31 12:00:00",
"交期汇总": {
"AM516预测交期": "2026-09-01",
"AM516总交期": 1,
"AM516主导因素": "现货",
},
"库存信息": {"可用库存(扣除了待产数量)": 100},
"BOM信息": {"是否有BOM(False则需提醒无BOM)": True},
"采购信息": {"缺料明细": []},
}
}
def message_event(message_id: str, chat_id: str, open_id: str) -> SimpleNamespace:
return SimpleNamespace(
event=SimpleNamespace(
message=SimpleNamespace(message_id=message_id, chat_id=chat_id),
sender=SimpleNamespace(sender_id=SimpleNamespace(open_id=open_id)),
)
)
class TriggerGateTests(unittest.TestCase):
def setUp(self) -> None:
with bridge._candidate_continuation_lock:
bridge._candidate_continuations.clear()
def test_complete_model_and_quantity_trigger(self) -> None:
request = bridge.parse_trigger_request("eRob70H50I-BHM-18CTC[V5] 10台")
@@ -102,6 +131,187 @@ class TriggerGateTests(unittest.TestCase):
)
reply.assert_called_once_with("client", "om_valid", "report")
def test_candidate_continuation_phrase_is_recognized(self) -> None:
valid_phrases = (
"还有没有其他候选型号？",
"还有吗？",
"还有别的吗",
"其他呢",
"还有类似的吗",
"还有什么推荐",
"有没有其他选择",
"还有其他方案吗",
"再推荐几个",
"可以再给我几个吗",
"继续吧",
"接着来",
"下一批",
"换一批",
"多看几个候选",
"more candidates",
"any other options?",
)
for phrase in valid_phrases:
with self.subTest(phrase=phrase):
self.assertTrue(bridge.is_candidate_continuation_request(phrase))
invalid_phrases = (
"这个机器人头像有点丑",
"more interesting",
"继续查询英语例句",
"下一课",
"再来一首歌",
"有其他事情吗",
)
for phrase in invalid_phrases:
with self.subTest(phrase=phrase):
self.assertFalse(bridge.is_candidate_continuation_request(phrase))
def test_continuation_without_same_conversation_context_is_silent(self) -> None:
data = message_event("om_more", "oc_other", "ou_other")
with (
mock.patch.object(
bridge,
"message_text",
return_value="还有没有其他候选型号？",
),
mock.patch.object(bridge, "build_report") as build_report,
mock.patch.object(bridge, "reply") as reply,
):
bridge.handle_message("client", data)
build_report.assert_not_called()
reply.assert_not_called()
def test_follow_up_continues_from_fifth_candidate_in_same_conversation(self) -> None:
request_text = "eRob70H50I-FS-18CN[V5] 10台"
request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10)
initial_data = message_event("om_initial", "oc_same", "ou_same")
follow_up_data = message_event("om_follow_up", "oc_same", "ou_same")
with (
mock.patch.object(
bridge,
"message_text",
side_effect=[request_text, "还有吗？"],
),
mock.patch.object(
bridge,
"build_report",
side_effect=["initial report", "continued report"],
) as build_report,
mock.patch.object(bridge, "reply") as reply,
):
bridge.handle_message("client", initial_data)
bridge.handle_message("client", follow_up_data)
self.assertEqual(
build_report.call_args_list,
[mock.call(request), mock.call(request, candidate_offset=4)],
)
self.assertEqual(
reply.call_args_list,
[
mock.call("client", "om_initial", "initial report"),
mock.call("client", "om_follow_up", "continued report"),
],
)
def test_context_is_cleared_when_all_candidates_fit_in_first_batch(self) -> None:
request_text = "eRob70H50I-BHM-18CTC[V5] 10台"
initial_data = message_event("om_small", "oc_small", "ou_small")
follow_up_data = message_event("om_small_more", "oc_small", "ou_small")
with (
mock.patch.object(
bridge,
"message_text",
side_effect=[request_text, "还有没有其他候选型号？"],
),
mock.patch.object(bridge, "build_report", return_value="report") as build_report,
mock.patch.object(bridge, "reply") as reply,
):
bridge.handle_message("client", initial_data)
bridge.handle_message("client", follow_up_data)
build_report.assert_called_once_with(
bridge.RequestInput(model="eRob70H50I-BHM-18CTC[V5]", quantity=10)
)
reply.assert_called_once_with("client", "om_small", "report")
def test_expired_candidate_context_is_removed(self) -> None:
key = "oc_expired:ou_expired"
request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10)
bridge._candidate_continuations[key] = bridge.CandidateContinuation(
request=request,
next_offset=4,
total_candidates=10,
expires_at=0,
)
self.assertIsNone(bridge.get_candidate_continuation(key))
self.assertNotIn(key, bridge._candidate_continuations)
class CandidateBatchTests(unittest.TestCase):
def setUp(self) -> None:
self.request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10)
self.specs = bridge.generate_candidate_specs(self.request.model)
self.assertGreater(len(self.specs), bridge.CANDIDATE_BATCH_SIZE * 2)
def test_initial_report_queries_baseline_and_only_four_candidates(self) -> None:
with (
mock.patch.object(
bridge,
"run_controlled_query",
return_value=successful_payload(),
) as run_query,
mock.patch.object(bridge, "write_record"),
):
report = bridge.build_report(self.request)
queried_models = [call.args[0].model for call in run_query.call_args_list]
self.assertEqual(len(queried_models), 1 + bridge.CANDIDATE_BATCH_SIZE)
self.assertEqual(queried_models[0], self.request.model)
self.assertCountEqual(
queried_models[1:],
[spec["model"] for spec in self.specs[: bridge.CANDIDATE_BATCH_SIZE]],
)
self.assertIn("本次先推荐4个候选型号（原型号另作基准）", report)
self.assertIn("可说“还有吗”“再推荐几个”“继续”等类似表达", report)
self.assertNotIn(self.specs[bridge.CANDIDATE_BATCH_SIZE]["model"], report)
def test_follow_up_report_queries_next_four_without_requerying_baseline(self) -> None:
with (
mock.patch.object(
bridge,
"run_controlled_query",
return_value=successful_payload(),
) as run_query,
mock.patch.object(bridge, "write_record"),
):
report = bridge.build_report(
self.request,
candidate_offset=bridge.CANDIDATE_BATCH_SIZE,
)
queried_models = [call.args[0].model for call in run_query.call_args_list]
expected_models = [
spec["model"]
for spec in self.specs[
bridge.CANDIDATE_BATCH_SIZE : bridge.CANDIDATE_BATCH_SIZE * 2
]
]
self.assertEqual(len(queried_models), bridge.CANDIDATE_BATCH_SIZE)
self.assertNotIn(self.request.model, queried_models)
self.assertCountEqual(queried_models, expected_models)
self.assertIn("本次继续推荐第5-8个候选型号", report)
for model in expected_models:
self.assertIn(model, report)
for model in [spec["model"] for spec in self.specs[:4]]:
self.assertNotIn(model, report)
if __name__ == "__main__":
unittest.main()