feat: fast route optimization and avatar-origin AI refusal rule

qingjian_cs/app/orchestrator.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from typing import Any
 
 from .agents import AfterSalesAgent, PreSalesAgent, QuoteAgent, RiskAgent, RouterAgent
+from .config import FAST_ROUTE_ENABLED
 from .image_quote_analyzer import analyze_image_for_quote
 from .models import Decision
 from .state_machine import evolve_after_sales_state, migrate_state_schema
@@ -18,6 +19,23 @@ class Orchestrator:
         self.risk = RiskAgent()
         self.store = ConversationStore()
 
+    @staticmethod
+    def _fast_route(context: dict[str, Any]) -> tuple[str, str] | None:
+        if not FAST_ROUTE_ENABLED:
+            return None
+        msg = str(context.get("msg", "") or "")
+        lower = msg.lower()
+        if bool(context.get("auto_quote_trigger")):
+            return "quote", "fast:auto_quote_trigger"
+        current_urls = context.get("current_image_urls") or []
+        if isinstance(current_urls, list) and len(current_urls) > 0:
+            return "quote", "fast:current_image"
+        if int(context.get("pending_images", 0) or 0) > 0 and any(k in lower for k in ("报价", "多少钱", "价格", "找得到", "找图", "能做", "有没有")):
+            return "quote", "fast:pending_image_and_pricing"
+        if any(k in lower for k in ("退款", "退钱", "不满意", "重做", "售后", "返工", "重发")):
+            return "after_sales", "fast:after_sales_keyword"
+        return None
+
     async def decide(self, context: dict[str, Any]) -> tuple[str, Decision, dict[str, Any]]:
         customer_key = context["customer_key"]
         session = self.store.get_session(customer_key)
@@ -36,7 +54,11 @@ class Orchestrator:
             "order_status": order_status,
         }
 
-        route, route_reason = await self.router.route(merged_ctx)
+        fast = self._fast_route(merged_ctx)
+        if fast is not None:
+            route, route_reason = fast
+        else:
+            route, route_reason = await self.router.route(merged_ctx)
 
         if route == "quote":
             latest_image_url = str(context.get("latest_image_url", "") or "").strip()
@@ -56,6 +78,10 @@ class Orchestrator:
             decision = await self.pre_sales.decide(merged_ctx)
 
         merged_state = {**prev_state, **(decision.state_patch or {})}
+        if route == "quote":
+            analysis_obj = merged_ctx.get("image_quote_analysis")
+            if isinstance(analysis_obj, dict) and analysis_obj:
+                merged_state["last_image_quote_analysis"] = analysis_obj
         new_state = evolve_after_sales_state(
             merged_state,
             route=route,

qingjian_cs/app/rules.py

@@ -103,6 +103,7 @@ def rules_prompt() -> str:
         "   - 外国平台相关内容由 AI 按上下文给出简短拒绝，不要展开解释。\n"
         "3) 命中硬规则后禁止改口：后续客户追问“能不能做”，仍保持同一结论。\n"
         "4) 多图场景若部分可做、部分不可做，必须明确“哪张可做、哪张不做”，禁止含糊表述。\n\n"
+        "5) 若客户诉求是“找头像原图/头像能不能找到/头像原图有没有”，固定 action=reply，reply='这个没有'。\n\n"
         "MASTER_RULES:\n"
         "A. 统一动作语义\n"
         "1) reply: 直接回复客户。\n"

qingjian_cs/services/service_gemini_stable.py

@@ -11,7 +11,7 @@ import aiohttp
 
 logger = logging.getLogger(__name__)
 
-GEMINI_API_KEY = "sk-YOUR_NEW_KEY"
+GEMINI_API_KEY = "sk-8i7uYE0RtnQwDImV8a5f7014DcAb46F6BcEb72Df92218aC8"
 GEMINI_API_URL = "https://api.laozhang.ai/v1beta/models"
 GEMINI_MODEL = "gemini-3.1-flash-image-preview"
 GEMINI_IMAGE_SIZE = "4K"