refactor: extract prompt building and image workflow routing from agent

core/image_workflow_router.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+async def handle_image_workflow(*, workflow_router: Any, message: str, data: dict, image_urls: list) -> bool:
+    """处理图片工作流（根据客户说的话判断执行哪种工作流）。"""
+    if not image_urls:
+        return False
+
+    workflow_type, confidence = workflow_router.detect_workflow(message)
+
+    customer_id = data.get("from_id")
+    acc_id = data.get("acc_id", "")
+    acc_type = data.get("acc_type", "AliWorkbench")
+    image_url = image_urls[0]
+
+    print(f"[Agent] 检测到工作流类型：{workflow_type} (置信度：{confidence})")
+
+    if workflow_type == "find_image":
+        print(f"[Agent] 执行查找图片工作流 | 客户：{customer_id}")
+        from core.workflow import workflow
+
+        return await workflow.find_image_workflow(
+            customer_id=customer_id,
+            image_url=image_url,
+            acc_id=acc_id,
+            acc_type=acc_type,
+        )
+    if workflow_type == "process_image":
+        print(f"[Agent] 执行处理图片工作流 | 客户：{customer_id}")
+        from core.workflow import workflow
+
+        return await workflow.process_image_workflow(
+            customer_id=customer_id,
+            image_url=image_url,
+            acc_id=acc_id,
+            acc_type=acc_type,
+        )
+    if workflow_type == "transfer_human":
+        print(f"[Agent] 执行转人工派单工作流 | 客户：{customer_id}")
+        from core.workflow import workflow
+
+        return await workflow.transfer_to_designer_workflow(
+            customer_id=customer_id,
+            image_url=image_url,
+            acc_id=acc_id,
+            acc_type=acc_type,
+            reason="客户主动要求转人工",
+        )
+
+    return False

core/prompt_builder.py

@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+import re
+from typing import Any, Callable
+
+
+def split_customer_text(msg: str) -> tuple[str, str]:
+    """
+    把混合消息拆分为（客户真实文字, 系统订单块）。
+    平台有时把客户文字和系统订单通知拼在同一条消息里。
+    """
+    order_marker = re.search(r"\[系统订单信息\]|\[系统通知\]", msg or "")
+    if order_marker:
+        customer_text = (msg or "")[: order_marker.start()].strip()
+        order_block = (msg or "")[order_marker.start() :].strip()
+    else:
+        customer_text = (msg or "").strip()
+        order_block = ""
+    return customer_text, order_block
+
+
+def build_prompt(
+    *,
+    message: Any,
+    state: Any,
+    extract_image_url: Callable[[str], str],
+    shop_type_resolver: Callable[[str, str], str],
+    parse_order_info: Callable[[str], dict[str, str]],
+    build_order_instruction: Callable[[str, str], str],
+) -> str:
+    """构建提示词。"""
+    msg_content = message.msg
+    stage_info = f"【当前阶段】{state.stage}"
+
+    customer_text, order_block = split_customer_text(msg_content)
+    has_order = bool(order_block)
+
+    if has_order:
+        order = parse_order_info(order_block)
+        if order.get("order_id"):
+            state.last_order_id = order["order_id"]
+            stage_info += f"\n【订单号】{order['order_id']}"
+        if order.get("order_status"):
+            state.order_status = order["order_status"]
+            stage_info += f"\n【订单状态】{order['order_status']}"
+        if order.get("pay_status"):
+            stage_info += f"\n【支付状态】{order['pay_status']}"
+        if order.get("amount"):
+            stage_info += f"\n【订单金额】{order['amount']}元"
+        if order.get("quantity"):
+            stage_info += f"\n【数量】{order['quantity']}件"
+        if order.get("order_time"):
+            stage_info += f"\n【下单时间】{order['order_time']}"
+        if order.get("buyer_note"):
+            stage_info += f"\n【买家备注】{order['buyer_note']}"
+
+    if state.discount_count > 0:
+        stage_info += f"\n【客户压价次数】{state.discount_count}"
+
+    shop_type = shop_type_resolver(message.acc_id or "", message.goods_name or "")
+    shop_hint = ""
+    try:
+        from config.config import CONFIG_DIR
+        import json
+
+        cfg_path = CONFIG_DIR / "shop_prompts.json"
+        if cfg_path.exists():
+            with open(cfg_path, "r", encoding="utf-8") as f:
+                cfg = json.load(f)
+            hints = cfg.get("type_hints", {})
+            shop_hint = hints.get(shop_type, "")
+            if not shop_hint and message.acc_id:
+                sh = cfg.get("shops", {}).get(message.acc_id, {})
+                shop_hint = sh.get("hint", "")
+    except Exception:
+        pass
+
+    prompt = f"""收到新消息：
+{stage_info}
+
+发送者: {message.from_name} ({message.from_id})
+"""
+    if message.goods_name:
+        prompt += f"商品名称: {message.goods_name}\n"
+    if shop_hint:
+        prompt += f"\n{shop_hint}\n"
+
+    order_paid = False
+    order_unpaid = False
+    if has_order:
+        order = parse_order_info(order_block)
+        paid_kws = ["等待发货", "已付款", "付款成功", "买家已付款"]
+        unpaid_kws = ["等待买家付款", "待付款", "未付款"]
+        ps = order.get("pay_status", "")
+        os_ = order.get("order_status", "")
+        if any(kw in ps or kw in os_ for kw in paid_kws):
+            order_paid = True
+        elif any(kw in ps or kw in os_ for kw in unpaid_kws):
+            order_unpaid = True
+
+    progress_keywords = [
+        "安排了吗",
+        "安排好了吗",
+        "好了吗",
+        "做了吗",
+        "做好了吗",
+        "弄好了吗",
+        "好了没",
+        "做了没",
+        "什么时候好",
+        "多久好",
+        "进度",
+        "催一下",
+        "快点",
+        "什么时候能好",
+        "做完了吗",
+    ]
+
+    if customer_text:
+        prompt += f"\n客户说：{customer_text}\n"
+        image_url = extract_image_url(customer_text)
+        price_keywords = ["多少钱", "多少", "价格", "几块", "怎么收费", "报个价"]
+        size_keywords = [
+            "尺寸",
+            "比例",
+            "宽",
+            "高",
+            "米",
+            "厘米",
+            "mm",
+            "cm",
+            "横版",
+            "竖版",
+            "2米",
+            "3米",
+            "改成",
+            "做成",
+        ]
+        has_size_change = any(kw in customer_text.lower() for kw in [k.lower() for k in size_keywords])
+
+        if shop_type == "gemini_api":
+            prompt += "\n【Gemini API 店铺】客户问账号/pro/续费/套餐等，按 API 客服自然回复，不要求发图。"
+        elif image_url:
+            prompt += "\n客户在继续发图阶段：先确认“已收图”，并引导客户把图和要求一次发完；等客户明确“发完了/统一报价”后再统一报价。"
+        elif any(kw in customer_text for kw in price_keywords):
+            last_url = extract_image_url(msg_content)
+            if last_url:
+                prompt += "\n客户在询问价格：若客户已确认发完，则给总报价；若还在发图，先引导发完后统一报价。"
+            else:
+                prompt += "\n客户在询问价格但未发图：先简短承接（如“在看呢/收到”），不要机械连发；再自然引导对方发图。"
+            if has_size_change:
+                prompt += (
+                    "\n⚠️ 尺寸改动场景：优先判断图片主体是否会被拉伸变形，"
+                    "不是只看整张图宽高比。若会变形，要先提示“需要补图/扩边”，再给报价。"
+                )
+        elif has_size_change:
+            prompt += (
+                "\n客户在改尺寸/改比例：先按主体比例判断是否会变形，"
+                "不是只看整图比例。若目标尺寸会拉伸主体，先明确说明要补图（如上下补图/扩边）再报价。"
+            )
+        elif any(kw in customer_text for kw in progress_keywords):
+            if order_unpaid:
+                prompt += "\n⚠️【订单未付款】客户问安排进度，但订单还未付款。自然告知拍下付款后马上安排即可。"
+            elif order_paid:
+                prompt += "\n客户催单，订单已付款，自然回复在做了/快了之类。"
+            else:
+                prompt += "\n客户催单，查询当前处理状态后自然回复。"
+        elif any(kw in customer_text for kw in ["贵", "有点贵", "太贵", "算了", "便宜点", "少点", "打折", "贵哦"]):
+            prompt += "\n⚠️【客户嫌贵】客户已表达价格顾虑或要放弃。禁止追问「什么问题」「说清楚点」。\n→ 第一步：直接让价一次（如原价 20→15），话术自然。\n→ 第二步：如果让价后客户还是嫌贵，引导客户自己去注册星绘平台：\n   - 每个话术都必须带网址 xinhui.cloud\n   - 强调「自己去弄」「自己操作」「自己下单」\n   - 示例：「嫌贵就去 xinhui.cloud 注册个号，自己有余额可以弄」「xinhui.cloud 去注册下，新用户送体验金，自己弄便宜」\n   - 语气直接，让客户自己去操作」"
+        elif any(kw in customer_text for kw in ["擦边", "黄色", "色情", "大尺度", "性感图", "露点", "半裸"]):
+            prompt += "\n⚠️【拒绝】客户询问擦边/黄色/敏感内容。直接拒绝，不接单，不说「发图来看看」。自然回复如：这类不做/不接/做不了。"
+        else:
+            prompt += "\n根据客户说的内容自然回应，像真人聊天，不要套模板。"
+
+    if has_order:
+        order = parse_order_info(order_block)
+        order_instruction = build_order_instruction(order.get("pay_status", ""), order.get("order_status", ""))
+        if customer_text:
+            if not order_unpaid:
+                prompt += f"\n\n【背景参考-订单通知】{order_instruction}"
+        else:
+            prompt += f"\n\n{order_instruction}"
+
+    if not customer_text and not has_order:
+        prompt += f"\n消息内容: {msg_content}\n请按工作流规则回复。"
+
+    return prompt

core/pydantic_ai_agent.py

@@ -73,6 +73,8 @@ from core.batch_quote_helpers import (
     calc_requirement_surcharge,
     prepare_batch_intake,
 )
+from core.prompt_builder import build_prompt as build_agent_prompt, split_customer_text
+from core.image_workflow_router import handle_image_workflow as route_image_workflow
 
 load_dotenv()
 
@@ -1777,215 +1779,25 @@ class CustomerServiceAgent:
         self._finalize_batch_state(state, message.from_id, final_price=bundle_price)
         return {"reply": reply_text, "need_transfer": False}
 
-    def _split_customer_text(self, msg: str) -> tuple:
+    _split_customer_text = staticmethod(split_customer_text)
-        """
-        把混合消息拆分为（客户真实文字, 系统订单块）。
-        平台有时把客户文字和系统订单通知拼在同一条消息里。
-        """
-        import re
-        # 找到系统订单块的起始位置
-        order_marker = re.search(r'\[系统订单信息\]|\[系统通知\]', msg)
-        if order_marker:
-            customer_text = msg[:order_marker.start()].strip()
-            order_block   = msg[order_marker.start():].strip()
-        else:
-            customer_text = msg.strip()
-            order_block   = ""
-        return customer_text, order_block
 
     def _build_prompt(self, message: CustomerMessage, state: ConversationState) -> str:
-        """构建提示词"""
+        return build_agent_prompt(
-        msg_content   = message.msg
+            message=message,
-        stage_info    = f"【当前阶段】{state.stage}"
+            state=state,
-
+            extract_image_url=self._extract_image_url,
-        # 拆分：客户文字 vs 系统订单块
+            shop_type_resolver=_get_shop_type,
-        customer_text, order_block = self._split_customer_text(msg_content)
+            parse_order_info=parse_order_info,
-        has_order = bool(order_block)
+            build_order_instruction=build_order_instruction,
-
-        if has_order:
-            order = parse_order_info(order_block)
-            if order.get('order_id'):
-                state.last_order_id = order['order_id']
-                stage_info += f"\n【订单号】{order['order_id']}"
-            if order.get('order_status'):
-                state.order_status = order['order_status']
-                stage_info += f"\n【订单状态】{order['order_status']}"
-            if order.get('pay_status'):
-                stage_info += f"\n【支付状态】{order['pay_status']}"
-            if order.get('amount'):
-                stage_info += f"\n【订单金额】{order['amount']}元"
-            if order.get('quantity'):
-                stage_info += f"\n【数量】{order['quantity']}件"
-            if order.get('order_time'):
-                stage_info += f"\n【下单时间】{order['order_time']}"
-            if order.get('buyer_note'):
-                stage_info += f"\n【买家备注】{order['buyer_note']}"
-
-        if state.discount_count > 0:
-            stage_info += f"\n【客户压价次数】{state.discount_count}"
-
-        # 店铺类型：不同店铺不同回复策略
-        shop_type = _get_shop_type(message.acc_id or "", message.goods_name or "")
-        shop_hint = ""
-        try:
-            from config.config import CONFIG_DIR
-            import json
-            cfg_path = CONFIG_DIR / "shop_prompts.json"
-            if cfg_path.exists():
-                with open(cfg_path, "r", encoding="utf-8") as f:
-                    cfg = json.load(f)
-                hints = cfg.get("type_hints", {})
-                shop_hint = hints.get(shop_type, "")
-                if not shop_hint and message.acc_id:
-                    sh = cfg.get("shops", {}).get(message.acc_id, {})
-                    shop_hint = sh.get("hint", "")
-        except Exception:
-            pass
-
-        prompt = f"""收到新消息：
-{stage_info}
-
-发送者: {message.from_name} ({message.from_id})
-"""
-        if message.goods_name:
-            prompt += f"商品名称: {message.goods_name}\n"
-        if shop_hint:
-            prompt += f"\n{shop_hint}\n"
-
-        # ── 优先处理客户真实问题 ──
-        # ── 判断订单付款状态（供后续逻辑使用）──
-        order_paid = False
-        order_unpaid = False
-        if has_order:
-            order = parse_order_info(order_block)
-            paid_kws   = ["等待发货", "已付款", "付款成功", "买家已付款"]
-            unpaid_kws = ["等待买家付款", "待付款", "未付款"]
-            ps = order.get('pay_status', '')
-            os_ = order.get('order_status', '')
-            if any(kw in ps or kw in os_ for kw in paid_kws):
-                order_paid = True
-            elif any(kw in ps or kw in os_ for kw in unpaid_kws):
-                order_unpaid = True
-
-        # ── 催单/进度询问关键词 ──
-        progress_keywords = [
-            "安排了吗", "安排好了吗", "好了吗", "做了吗", "做好了吗",
-            "弄好了吗", "好了没", "做了没", "什么时候好", "多久好",
-            "进度", "催一下", "快点", "什么时候能好", "做完了吗",
-        ]
-
-        if customer_text:
-            prompt += f"\n客户说：{customer_text}\n"
-            image_url = self._extract_image_url(customer_text)
-            price_keywords = ["多少钱", "多少", "价格", "几块", "怎么收费", "报个价"]
-            size_keywords = [
-                "尺寸", "比例", "宽", "高", "米", "厘米", "mm", "cm",
-                "横版", "竖版", "2米", "3米", "改成", "做成",
-            ]
-            has_size_change = any(kw in customer_text.lower() for kw in [k.lower() for k in size_keywords])
-
-            # gemini_api 店铺：不触发找图流程，按 API 客服回复
-            if shop_type == "gemini_api":
-                prompt += "\n【Gemini API 店铺】客户问账号/pro/续费/套餐等，按 API 客服自然回复，不要求发图。"
-            elif image_url:
-                prompt += "\n客户在继续发图阶段：先确认“已收图”，并引导客户把图和要求一次发完；等客户明确“发完了/统一报价”后再统一报价。"
-            elif any(kw in customer_text for kw in price_keywords):
-                last_url = self._extract_image_url(msg_content)
-                if last_url:
-                    prompt += "\n客户在询问价格：若客户已确认发完，则给总报价；若还在发图，先引导发完后统一报价。"
-                else:
-                    prompt += "\n客户在询问价格但未发图：先简短承接（如“在看呢/收到”），不要机械连发；再自然引导对方发图。"
-                if has_size_change:
-                    prompt += (
-                        "\n⚠️ 尺寸改动场景：优先判断图片主体是否会被拉伸变形，"
-                        "不是只看整张图宽高比。若会变形，要先提示“需要补图/扩边”，再给报价。"
         )
-            elif has_size_change:
-                prompt += (
-                    "\n客户在改尺寸/改比例：先按主体比例判断是否会变形，"
-                    "不是只看整图比例。若目标尺寸会拉伸主体，先明确说明要补图（如上下补图/扩边）再报价。"
-                )
-            elif any(kw in customer_text for kw in progress_keywords):
-                # 客户问进度/催单，必须先核查付款状态
-                if order_unpaid:
-                    prompt += "\n⚠️【订单未付款】客户问安排进度，但订单还未付款。自然告知拍下付款后马上安排即可。"
-                elif order_paid:
-                    prompt += "\n客户催单，订单已付款，自然回复在做了/快了之类。"
-                else:
-                    prompt += "\n客户催单，查询当前处理状态后自然回复。"
-            elif any(kw in customer_text for kw in ["贵", "有点贵", "太贵", "算了", "便宜点", "少点", "打折", "贵哦"]):
-                # 客户嫌贵/要放弃 → 直接让价一次，不问「什么问题」
-                prompt += "\n⚠️【客户嫌贵】客户已表达价格顾虑或要放弃。禁止追问「什么问题」「说清楚点」。\n→ 第一步：直接让价一次（如原价 20→15），话术自然。\n→ 第二步：如果让价后客户还是嫌贵，引导客户自己去注册星绘平台：\n   - 每个话术都必须带网址 xinhui.cloud\n   - 强调「自己去弄」「自己操作」「自己下单」\n   - 示例：「嫌贵就去 xinhui.cloud 注册个号，自己有余额可以弄」「xinhui.cloud 去注册下，新用户送体验金，自己弄便宜」\n   - 语气直接，让客户自己去操作」"
-            elif any(kw in customer_text for kw in ["擦边", "黄色", "色情", "大尺度", "性感图", "露点", "半裸"]):
-                # 客户问擦边/黄色内容 → 直接拒绝，不说「发图来看看」
-                prompt += "\n⚠️【拒绝】客户询问擦边/黄色/敏感内容。直接拒绝，不接单，不说「发图来看看」。自然回复如：这类不做/不接/做不了。"
-            else:
-                prompt += "\n根据客户说的内容自然回应，像真人聊天，不要套模板。"
-
-        # ── 附加订单信息（不覆盖客户问题的优先级）──
-        if has_order:
-            order = parse_order_info(order_block)
-            order_instruction = build_order_instruction(
-                order.get('pay_status', ''),
-                order.get('order_status', '')
-            )
-            if customer_text:
-                if not order_unpaid:
-                    # 未付款情况已在上面明确处理，不重复添加背景
-                    prompt += f"\n\n【背景参考-订单通知】{order_instruction}"
-            else:
-                # 纯系统通知，没有客户文字
-                prompt += f"\n\n{order_instruction}"
-
-        if not customer_text and not has_order:
-            prompt += f"\n消息内容: {msg_content}\n请按工作流规则回复。"
-
-        return prompt
 
     async def _handle_image_workflow(self, message: str, data: dict, image_urls: list) -> bool:
-        """处理图片工作流（根据客户说的话判断执行哪种工作流）"""
+        return await route_image_workflow(
-        if not image_urls:
+            workflow_router=self.workflow_router,
-            return False
+            message=message,
-
+            data=data,
-        workflow_type, confidence = self.workflow_router.detect_workflow(message)
+            image_urls=image_urls,
-
-        customer_id = data.get('from_id')
-        acc_id = data.get('acc_id', '')
-        acc_type = data.get('acc_type', 'AliWorkbench')
-        image_url = image_urls[0]
-
-        print(f"[Agent] 检测到工作流类型：{workflow_type} (置信度：{confidence})")
-
-        if workflow_type == "find_image":
-            print(f"[Agent] 执行查找图片工作流 | 客户：{customer_id}")
-            from core.workflow import workflow
-            return await workflow.find_image_workflow(
-                customer_id=customer_id,
-                image_url=image_url,
-                acc_id=acc_id,
-                acc_type=acc_type
         )
-        elif workflow_type == "process_image":
-            print(f"[Agent] 执行处理图片工作流 | 客户：{customer_id}")
-            from core.workflow import workflow
-            return await workflow.process_image_workflow(
-                customer_id=customer_id,
-                image_url=image_url,
-                acc_id=acc_id,
-                acc_type=acc_type
-            )
-        elif workflow_type == "transfer_human":
-            print(f"[Agent] 执行转人工派单工作流 | 客户：{customer_id}")
-            from core.workflow import workflow
-            return await workflow.transfer_to_designer_workflow(
-                customer_id=customer_id,
-                image_url=image_url,
-                acc_id=acc_id,
-                acc_type=acc_type,
-                reason="客户主动要求转人工"
-            )
-
-        return False
 
 
 async def test_agent():