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"]) def tearDown(self): db.clear_pending_quote_state(self.customer_id) if __name__ == "__main__": unittest.main(verbosity=2)