fix: prevent None reply in collection flow and harden response fallback

core/pydantic_ai_agent.py

@@ -384,6 +384,35 @@ class CustomerServiceAgent:
         """
         if not self.dynamic_collection_replies:
             return fallback
+        try:
+            deps = AgentDeps(
+                msg_id=message.msg_id,
+                acc_id=message.acc_id,
+                from_id=message.from_id,
+                platform=message.acc_type,
+            )
+            history = self.message_histories.get(message.from_id, [])
+            pending_req = "；".join((state.pending_requirements or [])[-4:]) or "无"
+            user_prompt = (
+                "请按下面意图生成给客户的自然回复。\n"
+                f"场景: {scene}\n"
+                f"回复意图: {intent_hint}\n"
+                f"客户原话: {message.msg}\n"
+                f"当前已收图片数: {len(state.pending_image_urls)}\n"
+                f"当前需求摘要: {pending_req}\n"
+                "输出要求: 不超过2句话，像真人店主聊天。"
+            )
+            result = await self.agent_natural_reply.run(user_prompt, deps=deps, message_history=history)
+            self.message_histories[message.from_id] = result.all_messages()[-30:]
+            text = self._colloquialize_reply(self._normalize_reply_text(result.output))
+            if not text:
+                return fallback
+            transfer_keywords = ("TRANSFER_REQUESTED", "[转移会话]", "转移会话")
+            if any(k in text for k in transfer_keywords):
+                return fallback
+            return text
+        except Exception:
+            return fallback
 
     async def _rewrite_reply_with_ai(
         self,
@@ -430,35 +459,6 @@ class CustomerServiceAgent:
             return polished
         except Exception:
             return text
-        try:
-            deps = AgentDeps(
-                msg_id=message.msg_id,
-                acc_id=message.acc_id,
-                from_id=message.from_id,
-                platform=message.acc_type,
-            )
-            history = self.message_histories.get(message.from_id, [])
-            pending_req = "；".join((state.pending_requirements or [])[-4:]) or "无"
-            user_prompt = (
-                "请按下面意图生成给客户的自然回复。\n"
-                f"场景: {scene}\n"
-                f"回复意图: {intent_hint}\n"
-                f"客户原话: {message.msg}\n"
-                f"当前已收图片数: {len(state.pending_image_urls)}\n"
-                f"当前需求摘要: {pending_req}\n"
-                "输出要求: 不超过2句话，像真人店主聊天。"
-            )
-            result = await self.agent_natural_reply.run(user_prompt, deps=deps, message_history=history)
-            self.message_histories[message.from_id] = result.all_messages()[-30:]
-            text = self._colloquialize_reply(self._normalize_reply_text(result.output))
-            if not text:
-                return fallback
-            transfer_keywords = ("TRANSFER_REQUESTED", "[转移会话]", "转移会话")
-            if any(k in text for k in transfer_keywords):
-                return fallback
-            return text
-        except Exception:
-            return fallback
 
     def _register_tools(self):
         """注册所有 Tool，让 Agent 可以主动调用"""
@@ -2116,7 +2116,7 @@ class CustomerServiceAgent:
         else:
             print(f"{self.C_MUTED}[REPLY->CUSTOMER]{self.C_RESET} <静默/不发送>")
 
-        return AgentResponse(reply=reply_text, should_reply=should_reply, need_transfer=need_transfer, transfer_msg=transfer_msg)
+        return AgentResponse(reply=reply_text or "", should_reply=should_reply, need_transfer=need_transfer, transfer_msg=transfer_msg)
 
     def _detect_price(self, reply: str, state: ConversationState):
         """从回复中提取价格，同步写入客户数据库（价格必须为5的整数倍）"""

tests/test_regression_pipeline.py

@@ -337,6 +337,33 @@ class RegressionPipelineTest(unittest.IsolatedAsyncioTestCase):
         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, "图片收到了，你继续发就行。")
+
     def tearDown(self):
         db.clear_pending_quote_state(self.customer_id)