qizhong/inquiry_robot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

空运「货交」只记提货地址，不再当成起运地。舱位旧说法「释放」先问航线是否取消，陆运通关口岸示例改为友谊关。

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jillion886
2026-10-11 20:45:53 +08:00
co-authored by Cursor
parent b070a9b7a7
commit 185abbf45e
16 changed files with 533 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
inquiry-agent/tests/test_air_group_collab.py
+85 -4
View File
@@ -20,6 +20,7 @@ from agent.policy import inquiry_copy as copy
from agent.routing.air_group_intent import (
INTENT_LOCK,
INTENT_RELEASE,
INTENT_RELEASE_ASK,
classify_air_group_text,
)
from agent.routing.dispatch import dispatch_inbound
@@ -201,7 +202,12 @@ class AirGroupCollabTests(unittest.TestCase):
def test_intent_lock_is_not_hardcoded(self) -> None:
self.assertEqual(classify_air_group_text("帮我锁舱 WO202609160001", allow_network=False), INTENT_LOCK)
self.assertEqual(classify_air_group_text("WO202609160001释放舱位", allow_network=False), INTENT_RELEASE)
self.assertEqual(classify_air_group_text("WO202609160001取消舱位", allow_network=False), INTENT_RELEASE)
self.assertEqual(classify_air_group_text("放舱", allow_network=False), INTENT_RELEASE)
self.assertEqual(classify_air_group_text("解锁舱", allow_network=False), INTENT_RELEASE)
self.assertEqual(classify_air_group_text("取消锁舱", allow_network=False), INTENT_RELEASE)
self.assertEqual(classify_air_group_text("WO202609160001释放舱位", allow_network=False), INTENT_RELEASE_ASK)
self.assertEqual(classify_air_group_text("把舱位释放了", allow_network=False), INTENT_RELEASE_ASK)
def test_airline_fee_wording_is_quote_not_adjust(self) -> None:
"""航线写「空运费改为900」是手工报价，不能先判成调价。比例和改总价仍是调价。"""
@@ -1278,9 +1284,84 @@ class AirGroupCollabTests(unittest.TestCase):
)
self.assertEqual(phase, "release_ok")
added = "\n".join(self._texts()[before:])
self.assertIn("释放", added)
self.assertIn("取消舱位", added)
self.assertNotIn("请确认成交状态", added)
def test_release_wording_asks_before_cancel(self) -> None:
"""说出释放舱位只提问。销售回「是」不算，航线确认后才取消。"""
wo = self._create_air()
handle_air_group(
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_ask", mid="ask0"),
flow=self.flow,
ledger=self.ledger,
injected_intent="air_lock",
)
self.groups.sent.clear()
phase = handle_air_group(
self._msg(sender="air1", text=f"{wo}释放舱位", chat="wr_air_ask", mid="ask1"),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "release_ask")
self.assertIn("是否需要取消舱位？", "\n".join(self._texts()))
self.assertNotIn("已通知TMS执行空运取消舱位成功", "\n".join(self._texts()))
phase = handle_air_group(
self._msg(sender="sales1", text="是", chat="wr_air_ask", mid="ask2"),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "not_airline")
self.assertNotIn("已通知TMS执行空运取消舱位成功", "\n".join(self._texts()))
phase = handle_air_group(
self._msg(sender="air1", text="是", chat="wr_air_ask", mid="ask3"),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "release_ok")
self.assertIn("已通知TMS执行空运取消舱位成功", "\n".join(self._texts()))
def test_release_wording_can_be_declined(self) -> None:
wo = self._create_air()
handle_air_group(
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_ask_no", mid="no0"),
flow=self.flow,
ledger=self.ledger,
injected_intent="air_lock",
)
self.groups.sent.clear()
handle_air_group(
self._msg(sender="air1", text=f"{wo}把舱位释放了", chat="wr_air_ask_no", mid="no1"),
flow=self.flow,
ledger=self.ledger,
)
phase = handle_air_group(
self._msg(sender="air1", text="不用", chat="wr_air_ask_no", mid="no2"),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "release_ask_dropped")
self.assertIn("已取消这次操作", "\n".join(self._texts()))
self.assertNotIn("已通知TMS执行空运取消舱位成功", "\n".join(self._texts()))
def test_cancel_cabin_words_release_immediately(self) -> None:
wo = self._create_air()
handle_air_group(
self._msg(sender="air1", text=f"{wo}锁舱", chat="wr_air_now", mid="now0"),
flow=self.flow,
ledger=self.ledger,
injected_intent="air_lock",
)
self.groups.sent.clear()
phase = handle_air_group(
self._msg(sender="air1", text=f"{wo}取消舱位", chat="wr_air_now", mid="now1"),
flow=self.flow,
ledger=self.ledger,
)
self.assertEqual(phase, "release_ok")
joined = "\n".join(self._texts())
self.assertIn("已通知TMS执行空运取消舱位成功", joined)
self.assertNotIn("是否需要取消舱位？", joined)
def test_sales_cannot_lock(self) -> None:
wo = self._create_air()
phase = handle_air_group(
@@ -1825,9 +1906,9 @@ class AirGroupCollabTests(unittest.TestCase):
)
self.assertEqual(phase, "system_exception")
joined = "\n".join(self._texts())
self.assertIn(f"工单{wo} 空运释放舱位失败。", joined)
self.assertIn(f"工单{wo} 空运取消舱位失败。", joined)
self.assertIn("操作人：HangXian", joined)
self.assertIn("释放舱位失败的原因：舱位不存在", joined)
self.assertIn("取消舱位失败的原因：舱位不存在", joined)
self.assertIn("已通知IT运维排查，请稍等。", joined)
self.assertNotIn("TMS接口异常，IT运维排查中", joined)
inquiry-agent/tests/test_extract_oral.py
+73
View File
@@ -624,6 +624,79 @@ AIR_QUOTE_SHEET = """
"""
class CargoHandoverSynonymTests(unittest.TestCase):
"""货交是提货地址，不能当成起运地。"""
def test_huojiao_shanghai_is_pickup_not_origin(self) -> None:
text = (
"KUL 空运 普货 散货\n"
"8686KG//9PLT//11.6CBM\n"
"110*110*118CM*7\n"
"110*110*80CM*1\n"
"110*110*53CM*1\n"
"货交上海\n"
"看看价格 几号"
)
guessed = parse_extract_payload(
{
"transport_mode": "AIR",
"fields": {
"起运港": "上海",
"目的港": "KUL",
"品名": "普货",
"件数": "9",
"毛重": "8686KG",
"体积": "11.6CBM",
"包装方式": "散货",
},
}
)
facts = extract_inquiry_snapshot(text, b_result=guessed)["facts"]
self.assertEqual(facts.get("提货地址"), "上海")
self.assertNotIn("起运港", facts)
self.assertEqual(facts.get("目的港"), "KUL")
def test_huojiao_colon_and_dao_same_as_pickup(self) -> None:
for spoken in ("货交：上海", "货交到上海", "货交地：上海"):
guessed = parse_extract_payload(
{
"transport_mode": "AIR",
"fields": {"起运港": "上海", "目的港": "KUL", "品名": "普货"},
}
)
facts = extract_inquiry_snapshot(
f"空运 {spoken} 目的地KUL", b_result=guessed
)["facts"]
self.assertEqual(facts.get("提货地址"), "上海", spoken)
self.assertNotIn("起运港", facts, spoken)
def test_explicit_route_keeps_origin(self) -> None:
text = "SZX-CGK 空运 货交深圳"
guessed = parse_extract_payload(
{
"transport_mode": "AIR",
"fields": {"起运港": "SZX", "目的港": "CGK", "品名": "普货"},
}
)
facts = extract_inquiry_snapshot(text, b_result=guessed)["facts"]
self.assertEqual(facts.get("提货地址"), "深圳")
self.assertEqual(facts.get("起运港"), "SZX")
self.assertEqual(facts.get("目的港"), "CGK")
def test_fly_from_city_keeps_origin(self) -> None:
text = "从南京飞吉隆坡 空运 货交上海"
guessed = parse_extract_payload(
{
"transport_mode": "AIR",
"fields": {"起运港": "南京", "目的港": "吉隆坡", "品名": "普货"},
}
)
facts = extract_inquiry_snapshot(text, b_result=guessed)["facts"]
self.assertEqual(facts.get("提货地址"), "上海")
self.assertEqual(facts.get("起运港"), "南京")
self.assertEqual(facts.get("目的港"), "吉隆坡")
class AirQuoteSheetInquiryOnlyTests(unittest.TestCase):
def test_quote_sheet_keeps_inquiry_fields_only(self) -> None:
"""费用明细、报价条件、落款不能进补问。起运地不能收成操作费。"""
inquiry-agent/tests/test_land_copy.py
+1 -1
View File
@@ -123,7 +123,7 @@ class LandCopyTest(unittest.TestCase):
self.assertIn("始发站（必填）：广州", text)
self.assertIn(copy.CLARIFY_PASTE_HINT, text)
self.assertIn("1.体积（CBM）（必填）：\n(如：100CBM）", text)
self.assertIn("2.通关口岸（非必填）：\n(如：皇岗）", text)
self.assertIn("2.通关口岸（非必填）：\n(如：友谊关）", text)
self.assertNotIn("体积（必填）：(如：", text)
self.assertNotIn(copy.LAND_CONFIRM_TAIL, text)
inquiry-agent/tests/test_system_exception.py
+4 -4
View File
@@ -65,7 +65,7 @@ class TestSystemExceptionCopy(unittest.TestCase):
)
self.assertEqual(
copy.system_exception_impact(copy.STEP_TMS_RELEASE),
"释放舱位失败，工单状态保持不变",
"取消舱位失败，工单状态保持不变",
)
def test_alert_title_never_has_test_prefix(self) -> None:
@@ -118,9 +118,9 @@ class TestSystemExceptionCopy(unittest.TestCase):
)
self.assertEqual(
text,
"工单WO202609210001 空运释放舱位失败。\n"
"工单WO202609210001 空运取消舱位失败。\n"
"操作人：吴季林\n"
"释放舱位失败的原因：舱位不存在\n"
"取消舱位失败的原因：舱位不存在\n"
"已通知IT运维排查，请稍等。",
)
@@ -409,7 +409,7 @@ class TestConfirmSystemException(unittest.TestCase):
)
card = queued[0]["payload"]["template_card"]
values = {row["keyname"]: row["value"] for row in card["horizontal_content_list"]}
self.assertEqual(values["影响范围"], "释放舱位失败，工单状态保持不变")
self.assertEqual(values["影响范围"], "取消舱位失败，工单状态保持不变")
def test_closed_ticket_not_marked(self) -> None:
self.ledger.transition(work_order_no=self.wo, to_status="已关闭")