diff --git a/qingjian_cs/app/client.py b/qingjian_cs/app/client.py index 4d6c0cd..eb989c7 100644 --- a/qingjian_cs/app/client.py +++ b/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 diff --git a/qingjian_cs/app/config.py b/qingjian_cs/app/config.py index fbfd4c3..ceb5eda 100644 --- a/qingjian_cs/app/config.py +++ b/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() diff --git a/qingjian_cs/app/rules.py b/qingjian_cs/app/rules.py index 06ce170..ab2cd71 100644 --- a/qingjian_cs/app/rules.py +++ b/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" diff --git a/qingjian_cs/services/service_gemini.py b/qingjian_cs/services/service_gemini.py index dee77ec..fba8808 100644 --- a/qingjian_cs/services/service_gemini.py +++ b/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(BaseService): +class GeminiExtractV2Service: """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, - "max_retries": 2, + "api_model": "gemini-2.5-flash-image", # 更稳定的模型 + "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, - input_path: str, - output_path: str, - custom_prompt: str = None, - aspect_ratio: str = "1:1", - image_size: str = "", - person_generation: str = "", - thinking_level: str = "", + self, + input_path: str, + output_path: str, + custom_prompt: str = None ) -> 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"], - **({"imageConfig": image_config} if image_config else {}), - **({"thinkingConfig": thinking_config} if thinking_config else {}), + "responseModalities": ["IMAGE"] # 只生成图片 + # 不传imageConfig，让输出图片比例与输入图片一致 } } - 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( - input_path: str, - output_path: str, - custom_prompt: str = None, - aspect_ratio: str = "1:1", -) -> tuple[bool, str, dict]: - """Gemini V2印花提取便捷函数""" +async def extract_pattern_v2(input_path: str, output_path: str, custom_prompt: str = None) -> 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" - output_path = "test_output_v2.png" + input_path = "image.png" + output_path = f"image_output_{int(time.time())}.png" success, message, data = await service.extract_pattern(input_path, output_path)