refactor: simplify gemini flow and tighten human-like short replies

qingjian_cs/app/client.py

@@ -113,8 +113,13 @@ class QingjianClient:
         t = self._humanize_reply(t)
         if len(t) <= max_len:
             return t
+        # 优先按句号切，避免把一句话硬腰斩成“AI半句”
         parts = re.split(r"[。！？!?]", t)
-        head = next((p.strip() for p in parts if p and p.strip()), t)
+        head = next((p.strip() for p in parts if p and p.strip()), "")
+        if not head:
+            # 无句号时按逗号切第一短分句
+            sub_parts = re.split(r"[，,；;：:]", t)
+            head = next((p.strip() for p in sub_parts if p and p.strip()), t)
         if len(head) > max_len:
             head = head[:max_len].rstrip("，,;；:： ")
         return head or t[:max_len]
@@ -122,6 +127,15 @@ class QingjianClient:
     @staticmethod
     def _humanize_reply(text: str) -> str:
         t = str(text or "").strip()
+        # 去AI腔常见口癖
+        t = re.sub(r"^(亲亲|宝子|宝贝|您好呀|您好哦)[，,]?\s*", "", t)
+        t = t.replace("我这边", "我")
+        t = t.replace("请问", "")
+        t = t.replace("可以先帮您评估看看哦", "我先看下")
+        t = t.replace("服务质量有保障", "质量没问题")
+        t = t.replace("这个价格已经是很优惠的啦", "这价已经很低了")
+        t = re.sub(r"(哈~|哦~|呀~|啦~)$", "", t)
+        t = re.sub(r"\s+", "", t)
         return t
 
     @staticmethod

qingjian_cs/app/config.py

@@ -16,7 +16,7 @@ MESSAGE_DEBOUNCE_SECONDS = int(os.getenv("MESSAGE_DEBOUNCE_SECONDS", "6"))
 AUTO_QUOTE_WAIT_SECONDS = int(os.getenv("AUTO_QUOTE_WAIT_SECONDS", "18"))
 AGENT_MAX_ITERS = int(os.getenv("AGENT_MAX_ITERS", "3"))
 FAST_ROUTE_ENABLED = os.getenv("FAST_ROUTE_ENABLED", "1").strip() in {"1", "true", "True", "yes", "on"}
-SHORT_REPLY_MAX_CHARS = int(os.getenv("SHORT_REPLY_MAX_CHARS", "20"))
+SHORT_REPLY_MAX_CHARS = int(os.getenv("SHORT_REPLY_MAX_CHARS", "18"))
 
 STORE_BACKEND = os.getenv("STORE_BACKEND", "sqlite").strip().lower()
 STORE_SQLITE_PATH = os.getenv("STORE_SQLITE_PATH", "").strip()

qingjian_cs/app/rules.py

@@ -128,7 +128,9 @@ def rules_prompt() -> str:
         "E. 对话质量\n"
         "1) 单次只做一个动作，不混合。\n"
         "2) 避免重复同一句话；若语义相同，换表达。\n"
-        "3) reply 必须短: 优先 1 句，口语化，避免AI腔。\n"
+        "3) reply 必须短: 只回 1 句，优先 12-18 字，口语化，避免AI腔。\n"
+        "   - 禁用词风: '亲亲'、'我这边'、'请问还有其他需求吗'、'服务质量有保障'、'作为AI'。\n"
+        "   - 优先自然短句: 例如'收到，我先看图。'、'发图我看下。'、'行，我现在算价。'\n"
         "4) 不要输出思考过程，不要输出 tool_use 文本给客户。\n"
         "5) 若上下文不足，先澄清 1 个关键问题，不要连续追问。\n\n"
         "F. 店铺人格\n"

qingjian_cs/services/service_gemini.py

@@ -14,22 +14,11 @@ import time
 from datetime import datetime
 from pathlib import Path
 import logging
-from dotenv import load_dotenv
-from utils.metrics_tracker import emit as metrics_emit
 
 
-from utils.service_base import BaseService
-
 logger = logging.getLogger(__name__)
-load_dotenv()
-GEMINI_IMAGE_MODEL = os.getenv("GEMINI_IMAGE_MODEL", "gemini-3.1-flash-image-preview")
-GEMINI_IMAGE_FALLBACK_MODEL = os.getenv("GEMINI_IMAGE_FALLBACK_MODEL", "gemini-2.5-flash-image")
-GEMINI_IMAGE_SIZE = os.getenv("GEMINI_IMAGE_SIZE", "1K")
-GEMINI_THINKING_LEVEL = os.getenv("GEMINI_THINKING_LEVEL", "MINIMAL")
-GEMINI_PERSON_GENERATION = os.getenv("GEMINI_PERSON_GENERATION", "")
 
-
+class GeminiExtractV2Service:
-class GeminiExtractV2Service(BaseService):
     """Gemini印花提取V2服务类 - 使用服务，更经济"""
     
     SERVICE_NAME = "gemini_extract_v2"
@@ -54,36 +43,29 @@ class GeminiExtractV2Service(BaseService):
             "name": "西风接口$0.014",
             "api_key": "sk-uRuvzLfIHsc3BiHZ2cyebk0cYsZ8NR9rLL326QqXCKIy9EpK",
             "api_url": "https://api.apiqik.online/v1beta/models",
-            "api_model": GEMINI_IMAGE_MODEL,
+            "api_model": "gemini-2.5-flash-image",  # 更稳定的模型
-            "max_retries": 2,
+            "max_retries": 2,  # 贵接口少重试
             "cost": "中",
             "use_gemini_format": True  # 使用Gemini原生API格式
         },
-        {
-            "name": "西风接口Fallback",
-            "api_key": "sk-uRuvzLfIHsc3BiHZ2cyebk0cYsZ8NR9rLL326QqXCKIy9EpK",
-            "api_url": "https://api.apiqik.online/v1beta/models",
-            "api_model": GEMINI_IMAGE_FALLBACK_MODEL,
-            "max_retries": 1,
-            "cost": "中",
-            "use_gemini_format": True
-        },
 
         {
             "name": "最贵的",
             "api_key": "sk-8i7uYE0RtnQwDImV8a5f7014DcAb46F6BcEb72Df92218aC8",
             "api_url": "https://api.laozhang.ai/v1/chat/completions", 
-            "api_model": GEMINI_IMAGE_MODEL,
+            "api_model": "gemini-2.5-flash-image-preview",
             "max_retries": 1,
             "cost": "高"
         }
     ]
 
     # 默认提示词
-    DEFAULT_PROMPT = "提取印花图案，把褶皱移除。补齐缺失的部分，要生成完整，细节丰富，严格按照原图的元素位置生成平面的印花图，不要相似的，相似度要100%，生成高质量的印刷图"
+    DEFAULT_PROMPT = (
+        "提取印花图案，去褶皱并补齐缺失区域，生成完整清晰的平面图。"
+        "严格保持原图元素位置、颜色和细节，不要改风格。"
+    )
   #  DEFAULT_PROMPT = "生成图片，把衣服的图案展开起来做成数码印花印刷平面图。去掉皱褶，生成图案增强细节。排除衣服图案以外内容"
     def __init__(self):
-        super().__init__(name="gemini_extract_v2")
         self.session = None
     
     def image_to_base64(self, image_path: str) -> str:
@@ -102,14 +84,10 @@ class GeminiExtractV2Service(BaseService):
             return None
     
     async def extract_pattern(
-        self,
+        self, 
-        input_path: str,
+        input_path: str, 
-        output_path: str,
+        output_path: str, 
-        custom_prompt: str = None,
+        custom_prompt: str = None
-        aspect_ratio: str = "1:1",
-        image_size: str = "",
-        person_generation: str = "",
-        thinking_level: str = "",
     ) -> tuple[bool, str, dict]:
         """
         使用多API配置进行印花图案提取
@@ -133,7 +111,6 @@ class GeminiExtractV2Service(BaseService):
         # 按优先级逐个尝试API配置
         for config_index, config in enumerate(self.API_CONFIGS):
             logger.info(f"尝试使用API: {config['name']} (成本: {config['cost']})")
-            metrics_emit("gemini_request", model=config.get("api_model", ""), provider=config.get("name", ""))
             
             # 对每个API配置进行重试
             for attempt in range(config['max_retries']):
@@ -147,26 +124,7 @@ class GeminiExtractV2Service(BaseService):
                         headers = {
                             "Content-Type": "application/json"
                         }
-
+                        
-                        # 有效比例列表（Auto 不传 aspectRatio）
-                        valid_ratios = {"1:1", "9:16", "16:9", "3:4", "4:3", "3:2", "2:3", "5:4", "4:5"}
-                        valid_sizes = {"1K", "2K", "4K"}
-                        valid_thinking = {"MINIMAL", "LOW", "MEDIUM", "HIGH"}
-                        image_config = {}
-                        if aspect_ratio in valid_ratios:
-                            image_config["aspectRatio"] = aspect_ratio
-                        size_val = (image_size or GEMINI_IMAGE_SIZE or "").upper().strip()
-                        if size_val in valid_sizes:
-                            image_config["imageSize"] = size_val
-                        person_val = (person_generation or GEMINI_PERSON_GENERATION or "").strip()
-                        if person_val:
-                            # 中转接口若支持该字段会生效；不设置时不发送，保证兼容
-                            image_config["personGeneration"] = person_val
-                        thinking_val = (thinking_level or GEMINI_THINKING_LEVEL or "").upper().strip()
-                        thinking_config = {}
-                        if thinking_val in valid_thinking:
-                            thinking_config["thinkingLevel"] = thinking_val
-
                         data = {
                             "contents": [
                                 {
@@ -174,7 +132,7 @@ class GeminiExtractV2Service(BaseService):
                                     "parts": [
                                         {
                                             "inlineData": {
-                                                "mimeType": "image/jpeg",
+                                                "mimeType": "image/png",
                                                 "data": img64
                                             }
                                         },
@@ -185,15 +143,10 @@ class GeminiExtractV2Service(BaseService):
                                 }
                             ],
                             "generationConfig": {
-                                "responseModalities": ["IMAGE"],
+                                "responseModalities": ["IMAGE"]  # 只生成图片
-                                **({"imageConfig": image_config} if image_config else {}),
+                                # 不传imageConfig，让输出图片比例与输入图片一致
-                                **({"thinkingConfig": thinking_config} if thinking_config else {}),
                             }
                         }
-                        logger.info(
-                            f"Gemini 生成配置: 比例={aspect_ratio} 尺寸={image_config.get('imageSize', '默认')} "
-                            f"person={image_config.get('personGeneration', '默认')} thinking={thinking_config.get('thinkingLevel', '默认')}"
-                        )
                     else:
                         # OpenAI兼容格式
                         api_url = config['api_url']
@@ -250,25 +203,6 @@ class GeminiExtractV2Service(BaseService):
                                     continue
                                 
                                 result = await response.json()
-                                # Gemini 偶发只返回文本不返回图片：NO_IMAGE 时快速重试/降级
-                                if config.get('use_gemini_format', False):
-                                    finish_reason = ""
-                                    try:
-                                        finish_reason = (
-                                            (result.get("candidates") or [{}])[0].get("finishReason", "")
-                                        )
-                                    except Exception:
-                                        finish_reason = ""
-                                    if finish_reason == "NO_IMAGE":
-                                        logger.warning(
-                                            f"{config['name']} 返回 NO_IMAGE (模型={config.get('api_model')})，第{attempt + 1}次"
-                                        )
-                                        metrics_emit("gemini_no_image", model=config.get("api_model", ""), provider=config.get("name", ""))
-                                        if attempt == config['max_retries'] - 1:
-                                            logger.warning(f"{config['name']} NO_IMAGE 重试已用完，切换下一个配置")
-                                            break
-                                        await asyncio.sleep(1 + attempt)
-                                        continue
                     
                     except (aiohttp.ClientError, asyncio.TimeoutError, AssertionError) as e:
                         logger.error(f"{config['name']} 网络连接错误 (第{attempt + 1}次): {str(e)}")
@@ -292,12 +226,6 @@ class GeminiExtractV2Service(BaseService):
                     
                     if success:
                         logger.info(f"使用 {config['name']} 成功完成印花提取")
-                        metrics_emit("gemini_success", model=config.get("api_model", ""), provider=config.get("name", ""))
-                        try:
-                            from utils.api_cost_tracker import record
-                            record("gemini_extract", count=1)
-                        except Exception:
-                            pass
                         return True, f"Gemini V2印花提取完成 - 使用{config['name']}", data
                     else:
                         logger.warning(f"{config['name']} 响应处理失败: {message}")
@@ -381,7 +309,9 @@ class GeminiExtractV2Service(BaseService):
     async def _save_image(self, image_data: bytes, output_path: str, api_name: str) -> tuple[bool, str, dict]:
         """保存图片文件"""
         try:
-            os.makedirs(os.path.dirname(output_path), exist_ok=True)
+            output_dir = os.path.dirname(output_path)
+            if output_dir:
+                os.makedirs(output_dir, exist_ok=True)
             
             with open(output_path, 'wb') as f:
                 f.write(image_data)
@@ -491,70 +421,36 @@ class GeminiExtractV2Service(BaseService):
         # 保存图片
         return await self._save_image(image_data, output_path, api_name)
     
-    async def correct_perspective(
-        self,
-        input_path: str,
-        output_path: str,
-        level: str = "mild",
-    ) -> tuple[bool, str, dict]:
-        """
-        透视矫正：先把有透视畸变的图还原为正面平铺视图，再做后续处理。
-
-        Args:
-            input_path:  本地图片路径
-            output_path: 矫正后输出路径
-            level:       "mild" 或 "strong"
-        """
-        if level == "strong":
-            prompt = (
-                "这张图存在明显透视畸变（俯拍/斜拍/贴墙）。"
-                "请对图片进行透视矫正：将主体变换为正面平铺视图，"
-                "使所有边缘变成水平或垂直，去除梯形形变，"
-                "保持图案颜色和细节完全不变，只矫正几何形状，输出矫正后的完整图片。"
-            )
-        else:
-            prompt = (
-                "这张图存在轻微透视畸变（衣物悬挂/桌面斜拍）。"
-                "请做轻度透视矫正：将主体调整为尽量正视角，"
-                "消除轻微的梯形拉伸感，保持图案颜色和细节不变，输出矫正后的图片。"
-            )
-
-        # 透视矫正使用 1:1 比例避免比例失真
-        return await self.extract_pattern(
-            input_path=input_path,
-            output_path=output_path,
-            custom_prompt=prompt,
-            aspect_ratio="1:1",
-        )
-
     async def cleanup(self):
         """清理资源"""
         if self.session and not self.session.closed:
             await self.session.close()
 
 # 便捷函数
-async def extract_pattern_v2(
+async def extract_pattern_v2(input_path: str, output_path: str, custom_prompt: str = None) -> tuple[bool, str, dict]:
-    input_path: str,
+    """
-    output_path: str,
+    Gemini V2印花提取便捷函数
-    custom_prompt: str = None,
+    """
-    aspect_ratio: str = "1:1",
-) -> tuple[bool, str, dict]:
-    """Gemini V2印花提取便捷函数"""
     service = GeminiExtractV2Service()
     try:
-        return await service.extract_pattern(input_path, output_path, custom_prompt, aspect_ratio)
+        return await service.extract_pattern(input_path, output_path, custom_prompt)
     finally:
         await service.cleanup()
 
 if __name__ == "__main__":
-    # 测试代码
+    logging.basicConfig(
+        level=logging.INFO,
+        format="[%(asctime)s] %(levelname)s %(name)s: %(message)s",
+        datefmt="%H:%M:%S",
+    )
+
     import asyncio
     
     async def test():
         service = GeminiExtractV2Service()
         
-        input_path = "F:/api/134.png"
+        input_path = "image.png"
-        output_path = "test_output_v2.png"
+        output_path = f"image_output_{int(time.time())}.png"
         
         success, message, data = await service.extract_pattern(input_path, output_path)