423 lines
16 KiB
Python
423 lines
16 KiB
Python
import os
|
||
import unittest
|
||
from unittest.mock import AsyncMock
|
||
|
||
from core.pydantic_ai_agent import CustomerServiceAgent, CustomerMessage
|
||
from db.customer_db import db
|
||
|
||
|
||
class RegressionPipelineTest(unittest.IsolatedAsyncioTestCase):
|
||
def setUp(self):
|
||
self.customer_id = "__regression_test_customer__"
|
||
db.clear_pending_quote_state(self.customer_id)
|
||
os.environ["FEATURE_BATCH_QUOTE_ENABLED"] = "true"
|
||
os.environ["FEATURE_BATCH_QUOTE_PERCENT"] = "100"
|
||
os.environ["FEATURE_BATCH_QUOTE_SHOPS"] = ""
|
||
os.environ["AI_DYNAMIC_COLLECTION_REPLIES"] = "false"
|
||
os.environ["AI_REWRITE_ALL_REPLIES"] = "false"
|
||
os.environ["BATCH_QUOTE_DELAY_TURNS"] = "0"
|
||
|
||
async def test_collect_images_then_ack(self):
|
||
agent = CustomerServiceAgent()
|
||
msg = CustomerMessage(
|
||
msg_id="m1",
|
||
acc_id="test_shop",
|
||
msg="https://img.alicdn.com/a.jpg#*#https://img.alicdn.com/b.jpg",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
self.assertTrue(resp.reply.strip())
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
self.assertEqual(len(st.pending_image_urls), 2)
|
||
|
||
async def test_finish_signal_triggers_batch_quote(self):
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
|
||
st.pending_requirements = ["去背景"]
|
||
agent._sync_pending_quote_state(self.customer_id, st)
|
||
agent._quote_pending_images = AsyncMock(return_value={"reply": "打包15元，确认我就安排", "need_transfer": False})
|
||
|
||
msg = CustomerMessage(
|
||
msg_id="m2",
|
||
acc_id="test_shop",
|
||
msg="发完了，报价吧",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
self.assertIn("15", resp.reply)
|
||
agent._quote_pending_images.assert_awaited()
|
||
|
||
async def test_single_image_requirement_intent_triggers_quote(self):
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
|
||
st.pending_requirements = []
|
||
agent._sync_pending_quote_state(self.customer_id, st)
|
||
agent._quote_pending_images = AsyncMock(return_value={"reply": "这张20元，定了我马上做", "need_transfer": False})
|
||
|
||
msg = CustomerMessage(
|
||
msg_id="m3",
|
||
acc_id="test_shop",
|
||
msg="这个门头上面的字做一下",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
self.assertIn("20", resp.reply)
|
||
agent._quote_pending_images.assert_awaited()
|
||
|
||
async def test_multi_image_finish_intent_triggers_quote(self):
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg", "https://img.alicdn.com/b.jpg"]
|
||
st.pending_requirements = ["改字"]
|
||
agent._sync_pending_quote_state(self.customer_id, st)
|
||
agent._quote_pending_images = AsyncMock(return_value={"reply": "两张打包45，定了我就开做", "need_transfer": False})
|
||
|
||
msg = CustomerMessage(
|
||
msg_id="m4",
|
||
acc_id="test_shop",
|
||
msg="就这几张，先按这些报个价",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
self.assertIn("45", resp.reply)
|
||
agent._quote_pending_images.assert_awaited()
|
||
|
||
async def test_finish_signal_with_meile_triggers_quote(self):
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
|
||
st.pending_requirements = []
|
||
agent._sync_pending_quote_state(self.customer_id, st)
|
||
agent._quote_pending_images = AsyncMock(return_value={"reply": "这张15元，确认就开始", "need_transfer": False})
|
||
|
||
msg = CustomerMessage(
|
||
msg_id="m4b",
|
||
acc_id="test_shop",
|
||
msg="没有了，报下价",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
self.assertIn("15", resp.reply)
|
||
agent._quote_pending_images.assert_awaited()
|
||
|
||
async def test_finish_signal_defers_quote_when_delay_enabled(self):
|
||
os.environ["BATCH_QUOTE_DELAY_TURNS"] = "1"
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
|
||
st.pending_requirements = []
|
||
agent._sync_pending_quote_state(self.customer_id, st)
|
||
agent._quote_pending_images = AsyncMock(return_value={"reply": "这张20元，确认就开做", "need_transfer": False})
|
||
|
||
msg1 = CustomerMessage(
|
||
msg_id="m4c-1",
|
||
acc_id="test_shop",
|
||
msg="没有了，报价吧",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp1 = await agent.process_message(msg1)
|
||
self.assertTrue(resp1.should_reply)
|
||
self.assertNotIn("20", resp1.reply)
|
||
agent._quote_pending_images.assert_not_awaited()
|
||
|
||
msg2 = CustomerMessage(
|
||
msg_id="m4c-2",
|
||
acc_id="test_shop",
|
||
msg="有吗",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp2 = await agent.process_message(msg2)
|
||
self.assertTrue(resp2.should_reply)
|
||
self.assertIn("20", resp2.reply)
|
||
agent._quote_pending_images.assert_awaited()
|
||
os.environ["BATCH_QUOTE_DELAY_TURNS"] = "0"
|
||
|
||
async def test_cross_image_composite_intent_triggers_quote(self):
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg", "https://img.alicdn.com/b.jpg"]
|
||
st.pending_requirements = []
|
||
agent._sync_pending_quote_state(self.customer_id, st)
|
||
agent._quote_pending_images = AsyncMock(return_value={"reply": "这个跨图合成可以做，打包50元", "need_transfer": False})
|
||
|
||
msg = CustomerMessage(
|
||
msg_id="m5",
|
||
acc_id="test_shop",
|
||
msg="A图的图案转换到B图上去",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
self.assertIn("50", resp.reply)
|
||
agent._quote_pending_images.assert_awaited()
|
||
|
||
async def test_result_followup_query_uses_progress_reply(self):
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
|
||
st.pending_requirements = []
|
||
agent._sync_pending_quote_state(self.customer_id, st)
|
||
agent._quote_pending_images = AsyncMock(return_value={"reply": "不应触发报价", "need_transfer": False})
|
||
|
||
msg = CustomerMessage(
|
||
msg_id="m5b",
|
||
acc_id="test_shop",
|
||
msg="没找到吗",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
self.assertNotIn("不太懂你的意思", resp.reply)
|
||
self.assertNotIn("没完全理解", resp.reply)
|
||
self.assertNotIn("没听明白", resp.reply)
|
||
agent._quote_pending_images.assert_not_awaited()
|
||
|
||
async def test_short_youma_followup_uses_progress_reply(self):
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
|
||
st.pending_requirements = []
|
||
agent._sync_pending_quote_state(self.customer_id, st)
|
||
agent._quote_pending_images = AsyncMock(return_value={"reply": "不应触发报价", "need_transfer": False})
|
||
|
||
msg = CustomerMessage(
|
||
msg_id="m5bb",
|
||
acc_id="test_shop",
|
||
msg="有吗",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
self.assertNotIn("不太懂你的意思", resp.reply)
|
||
self.assertNotIn("没完全理解", resp.reply)
|
||
self.assertNotIn("没听明白", resp.reply)
|
||
st2 = agent._get_conversation_state(self.customer_id)
|
||
self.assertEqual(st2.quote_phase, "waiting_result")
|
||
agent._quote_pending_images.assert_not_awaited()
|
||
|
||
def test_short_text_classifier(self):
|
||
agent = CustomerServiceAgent()
|
||
self.assertEqual(agent._classify_short_customer_text("有吗"), "progress_query")
|
||
self.assertEqual(agent._classify_short_customer_text("没有了"), "finish_signal")
|
||
self.assertEqual(agent._classify_short_customer_text("好的"), "ack")
|
||
|
||
async def test_related_screenshot_followup_is_marked(self):
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
|
||
st.pending_requirements = []
|
||
agent._sync_pending_quote_state(self.customer_id, st)
|
||
|
||
msg = CustomerMessage(
|
||
msg_id="m5c",
|
||
acc_id="test_shop",
|
||
msg="这是上一张的截图#*#https://img.alicdn.com/b.jpg",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
st2 = agent._get_conversation_state(self.customer_id)
|
||
self.assertIn("与上一张相关（截图/局部细节）", st2.pending_requirements)
|
||
|
||
async def test_find_image_not_edit_conflict_triggers_clarification(self):
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
|
||
st.pending_requirements = []
|
||
agent._sync_pending_quote_state(self.customer_id, st)
|
||
agent._quote_pending_images = AsyncMock(return_value={"reply": "不该触发", "need_transfer": False})
|
||
|
||
msg = CustomerMessage(
|
||
msg_id="m6",
|
||
acc_id="test_shop",
|
||
msg="我要你帮我找图，不是让你做图",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
self.assertIn("找图", resp.reply)
|
||
self.assertIn("不是做图", resp.reply)
|
||
agent._quote_pending_images.assert_not_awaited()
|
||
|
||
async def test_pending_state_restore(self):
|
||
db.update_pending_quote_state(self.customer_id, ["u1", "u2"], ["r1"])
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
self.assertEqual(st.pending_image_urls, ["u1", "u2"])
|
||
self.assertEqual(st.pending_requirements, ["r1"])
|
||
|
||
async def test_collection_reply_never_returns_none(self):
|
||
os.environ["AI_DYNAMIC_COLLECTION_REPLIES"] = "true"
|
||
agent = CustomerServiceAgent()
|
||
agent.agent_natural_reply.run = AsyncMock(side_effect=RuntimeError("mock ai fail"))
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
msg = CustomerMessage(
|
||
msg_id="m-collection-fallback",
|
||
acc_id="test_shop",
|
||
msg="收到没",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
reply = await agent._render_collection_reply_with_ai(
|
||
message=msg,
|
||
state=st,
|
||
scene="collect_ack",
|
||
intent_hint="确认已收到图片",
|
||
fallback="图片收到了，你继续发就行。",
|
||
)
|
||
self.assertEqual(reply, "图片收到了，你继续发就行。")
|
||
|
||
async def test_first_image_ack_avoids_unified_quote_wording(self):
|
||
os.environ["AI_DYNAMIC_COLLECTION_REPLIES"] = "true"
|
||
agent = CustomerServiceAgent()
|
||
st = agent._get_conversation_state(self.customer_id)
|
||
st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
|
||
msg = CustomerMessage(
|
||
msg_id="m-first-ack",
|
||
acc_id="test_shop",
|
||
msg="https://img.alicdn.com/a.jpg",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
reply = await agent._render_collection_reply_with_ai(
|
||
message=msg,
|
||
state=st,
|
||
scene="collect_ack",
|
||
intent_hint="确认收到后先承接",
|
||
fallback="图片收到了，你继续发就行。",
|
||
)
|
||
self.assertIn("稍等", reply)
|
||
self.assertNotIn("统一报价", reply)
|
||
self.assertNotIn("发完", reply)
|
||
|
||
async def test_map_inquiry_is_rejected(self):
|
||
agent = CustomerServiceAgent()
|
||
msg = CustomerMessage(
|
||
msg_id="m-map-reject",
|
||
acc_id="test_shop",
|
||
msg="这个地图能做吗",
|
||
from_id=self.customer_id,
|
||
from_name="t",
|
||
cy_id=self.customer_id,
|
||
acc_type="AliWorkbench",
|
||
msg_type=0,
|
||
cy_name="t",
|
||
goods_name="专业找图",
|
||
goods_order="",
|
||
)
|
||
resp = await agent.process_message(msg)
|
||
self.assertTrue(resp.should_reply)
|
||
self.assertIn("地图", resp.reply)
|
||
self.assertIn("不做", resp.reply)
|
||
self.assertTrue(resp.reply.endswith(("嗯", "哦", "好的", "嗯咯", "嗯啦")))
|
||
|
||
def tearDown(self):
|
||
db.clear_pending_quote_state(self.customer_id)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|