From fc05b60d1a4369109205c6bf3080f30c0cb864b5 Mon Sep 17 00:00:00 2001 From: jimi <1847930177@qq.com> Date: Sun, 1 Mar 2026 15:38:20 +0800 Subject: [PATCH] refactor: extract collection intent helpers from pydantic agent --- core/collection_intent_helpers.py | 433 ++++++++++++++++++++++++++++++ core/pydantic_ai_agent.py | 326 +++------------------- 2 files changed, 463 insertions(+), 296 deletions(-) create mode 100644 core/collection_intent_helpers.py diff --git a/core/collection_intent_helpers.py b/core/collection_intent_helpers.py new file mode 100644 index 0000000..a450703 --- /dev/null +++ b/core/collection_intent_helpers.py @@ -0,0 +1,433 @@ +from __future__ import annotations + +import random +from typing import Any + + +def classify_short_customer_text(text: str) -> str: + """ + 短句分类器（状态机前置）： + - finish_signal: 发图完成，可报价 + - progress_query: 追问进度/结果 + - ack: 简短确认 + - unknown: 未识别 + """ + s = (text or "").strip() + if not s: + return "unknown" + if len(s) > 8: + return "unknown" + + finish_kw = ( + "没了", + "没有了", + "就这", + "就这张", + "就这一张", + "就这一个", + "就一个", + "先这些", + "就这些", + "发完了", + "都发完了", + ) + if any(k in s for k in finish_kw): + return "finish_signal" + + progress_kw = ( + "有吗", + "有没", + "有没有", + "找到了吗", + "找到了没", + "没找到吗", + "找到没", + "找到没有", + "进度", + "结果", + "多久好", + "什么时候好", + "好了没", + "弄好了吗", + "做了没", + "高清", + "发我", + "重新发", + "你重新发给我", + ) + if any(k in s for k in progress_kw) or s in {"?", "？", "在吗", "人呢"}: + return "progress_query" + + ack_kw = ("嗯", "嗯嗯", "好", "好的", "行", "可以", "ok", "OK", "收到", "明白") + if s in ack_kw: + return "ack" + return "unknown" + + +def is_batch_finish_signal(text: str) -> bool: + """客户是否表达“图发完了，可以统一报价”""" + if not text: + return False + if classify_short_customer_text(text) == "finish_signal": + return True + finish_keywords = [ + "发完了", + "都发完了", + "发齐了", + "齐了", + "先这些", + "就这些", + "全部", + "一起报", + "统一报价", + "总共多少钱", + "一共多少钱", + "打包价", + "总价", + "报价吧", + "报个总价", + "给个总价", + "没了", + "没有了", + "没图了", + "就这", + "就这张", + "就这一张", + "就这一个", + "就一个", + "先报吧", + "报下价", + "报个价", + "可以报价了", + "能报吗", + ] + return any(k in text for k in finish_keywords) + + +def is_cross_image_composite_intent(text: str) -> bool: + """ + 识别多图跨图修改意图（A图元素放到B图）。 + 例：A图的图案转到B图、这个图案放到另一张上。 + """ + s = (text or "").strip() + if not s: + return False + pair_marks = ("a图", "b图", "第一张", "第二张", "这张", "那张", "上一张", "另一张") + op_kw = ( + "转到", + "换到", + "放到", + "贴到", + "移到", + "套到", + "合成", + "融合", + "替换到", + "图案上去", + "字放到", + "元素放到", + "logo放到", + ) + return any(k in s.lower() for k in pair_marks) and any(k in s for k in op_kw) + + +def is_batch_finish_intent(text: str, state: Any, has_incoming_urls: bool) -> bool: + """ + 语义结束识别： + - 显式口令：发完了/统一报价 + - 隐式意图：询价/砍价 + - 单图需求明确：如“这个门头上面的字做一下”可直接进入报价 + """ + if not text: + return False + if is_batch_finish_signal(text): + return True + if has_incoming_urls: + return False + if not (getattr(state, "pending_image_urls", None) or []): + return False + + try: + from utils.intent_analyzer import detect_intent_embedding, detect_intent_keywords + + intent = detect_intent_embedding(text) or detect_intent_keywords(text) + except Exception: + intent = "" + if intent in ("询价", "砍价"): + return True + + msg = (text or "").strip() + if not msg: + return False + single_image_action_kw = ( + "做一下", + "改一下", + "处理一下", + "就这张", + "按这个做", + "照这个做", + "这个门头", + "上面的字", + "这个字", + "这个图做", + "能做吗", + ) + multi_image_finish_kw = ( + "就这些", + "就这几张", + "按这几张", + "这几张一起做", + "一起做一下", + "先按这些", + "先按这几张", + "直接报价", + "现在报价", + "看下报价", + "先报个总价", + "总价多少", + "一起多少钱", + "先做这几张", + ) + hold_kw = ("还有", "再发", "先等", "稍后", "等会", "回头") + image_count = len(getattr(state, "pending_image_urls", []) or []) + if image_count == 1: + if any(k in msg for k in single_image_action_kw) and not any(k in msg for k in hold_kw): + return True + elif image_count >= 2: + if any(k in msg for k in multi_image_finish_kw) and not any(k in msg for k in hold_kw): + return True + if is_cross_image_composite_intent(msg) and not any(k in msg for k in hold_kw): + return True + return False + + +def is_related_image_followup_intent(text: str) -> bool: + """识别“新发的是上一张的截图/局部细节”的关联意图。""" + s = (text or "").strip().lower() + if not s: + return False + relation_kw = ( + "截图", + "截屏", + "局部", + "细节", + "放大", + "裁剪", + "同一张", + "同一幅", + "上一张", + "上张", + "前一张", + "前面那张", + "刚才那张", + "这个是上面", + "这个是那张", + "补一张细节", + "补个截图", + ) + return any(k in s for k in relation_kw) + + +def is_result_followup_query(text: str) -> bool: + """识别客户在找图流程中的结果/进度追问。""" + if classify_short_customer_text(text) == "progress_query": + return True + s = (text or "").strip() + if not s: + return False + followup_kw = ( + "找到了吗", + "没找到吗", + "找到没", + "找到没有", + "找到了没", + "有吗", + "有没", + "有没有", + "有结果吗", + "结果呢", + "进度", + "多久好", + "什么时候好", + "好了没", + "弄好了吗", + "做了没", + "你重新发", + "重新发给我", + "高清", + "发我", + ) + if any(k in s for k in followup_kw): + return True + return s in {"?", "？", "在吗", "人呢"} + + +def build_collect_ack(count: int, related_followup: bool = False) -> str: + if related_followup and count >= 2: + related_templates = [ + "这张我收到了，看起来是上一张的截图/细节图，我按同一单一起处理。还有补充就继续发。", + "收到，这张是关联补图我记上了（按同一需求处理）。你还有图就继续发。", + "明白，这张是前图的局部截图，我会和前面那张一起算，不会分开漏掉。", + ] + return random.choice(related_templates) + if count <= 1: + one_templates = [ + "这张收到啦，还有图就继续发，我一起给你看。", + "图我看到了，后面还有就接着发，最后我一口价给你。", + "收到这张了，你有其他图也发来，我统一帮你算。", + "这张我先记上了，你那边还有的话接着发，我一起给你报。", + "第1张收到，你继续发就行，发完我这边一次给你算清楚。", + "这张没问题，我先收着。要是还有图，你直接连着发我就行。", + "我先看到了这张，你后面还有就一起发来，我统一给你报价。", + "这张图我已经记下了，后面有补充就继续甩过来哈。", + ] + return random.choice(one_templates) + templates = [ + "这几张我都收到了（现在{n}张）。还有的话继续发，我一起给你报。", + "好嘞，先看到{n}张了。你可以继续发，或者直接说“就这些”我现在就报价。", + "收到哈（共{n}张）。你还要补图就继续发，不补的话我现在也可以直接给价。", + "我这边先收到了{n}张。你继续补图，或者直接说“按这些算”我就开始报。", + "这波我已经记了{n}张，你要是还有就接着发，不补的话我立刻给总价。", + "先看到{n}张图了，后面你看是继续发，还是直接让我现在报价都可以。", + "好的，目前{n}张到位。你一句“就这些”，我马上给你打包价。", + "图我都看到了（{n}张）。你还发我就继续收，不发我现在就给你报。", + ] + return random.choice(templates).format(n=count) + + +def build_collect_progress_reply(count: int) -> str: + if count <= 1: + templates = [ + "我这边在处理了，这张有结果我第一时间回你。", + "在跟进中，这张一有进展我马上发你。", + "这张我正在看，稍等我一会儿，结果出来就回你。", + ] + return random.choice(templates) + templates = [ + "我这边在按你这{n}张一起处理，有结果我立刻同步你。", + "正在跟进这{n}张，出结果我第一时间发你，不会漏。", + "进度在跑了（共{n}张），你稍等一下，我这边有结果马上回。", + ] + return random.choice(templates).format(n=count) + + +def build_collect_remind(count: int) -> str: + if count <= 1: + one_templates = [ + "这个要求我记住了。你还有图就继续发，不补图我就按这张给你报价。", + "明白，这个需求我加上了。你继续发图也行，想直接报价也可以。", + "我先记下这张。你如果是要我找图，不是做图，直接说一声，我按找图思路给你走。", + "收到，这张我先按你的要求记好了。就做这一张的话，我现在直接给你报实价。", + "你这要求我记下了，后面还有图就发，没有的话我现在直接算价。", + "行，我按你这个要求来。继续补图也行，不补我就先报这张。", + "这个点我懂了，你还要补图就接着发，不补我立刻给你报价。", + "要求我已经加上了。你看是继续发，还是我现在直接报这张。", + ] + return random.choice(one_templates) + templates = [ + "需求我记下了（当前{n}张）。你继续补图，或者直接说“就这些”我现在报价。", + "好，这个要求也加上了（现在{n}张）。不再补图的话我立刻给你打包价。", + "收到（共{n}张）。你还发就继续，不发的话我现在就给总价。", + "这个需求我加进去了（现在{n}张）。你继续发也行，直接报价也行。", + "我这边都记好了（{n}张+需求）。你一句“先按这些算”，我马上报价。", + "要求同步好了，目前{n}张。要补图继续发，不补图我现在就给你打包价。", + "行，需求和图片我都收着了（{n}张）。你直接让我报价也可以。", + "好的，这条需求也算进去了（共{n}张）。你看要不要我现在直接报。", + ] + return random.choice(templates).format(n=count) + + +def is_find_image_not_edit_conflict(text: str) -> bool: + """识别客户明确声明“要找图，不是做图”的冲突语义。""" + s = (text or "").strip() + if not s: + return False + find_kw = ("找图", "找原图", "找素材", "找同款") + deny_edit_kw = ("不是让你做图", "不是做图", "不用做图", "不需要做图", "不是修图", "不用修图") + return any(k in s for k in find_kw) and any(k in s for k in deny_edit_kw) + + +def needs_clarification_in_collecting(text: str) -> bool: + """信息不足时先追问，不急着报价。""" + s = (text or "").strip() + if not s: + return False + short_non_vague_kw = ( + "？", + "?", + "没了", + "没有了", + "就这", + "行", + "好的", + "ok", + "报价", + "找到了吗", + "没找到吗", + "找到没", + "找到了没", + "有吗", + "有没", + "有没有", + "多久好", + "什么时候好", + "高清", + ) + if len(s) <= 4: + if any(k in s for k in short_non_vague_kw): + return False + return True + vague_kw = ( + "这个也是", + "一共几个图", + "几个图", + "啥意思", + "没明白", + "什么意思", + "这个呢", + "这个可以吗", + "然后呢", + "咋办", + "怎么搞", + ) + return any(k in s for k in vague_kw) + + +def build_find_image_clarify_reply(state: Any) -> str: + count = len(getattr(state, "pending_image_urls", []) or []) + return ( + f"明白，你是要我帮你找图，不是做图。现在我这边先记了{count}张，" + "你告诉我具体要找哪种：原图/同款/高清版，我按这个方向给你找。" + ) + + +def build_not_understood_reply() -> str: + """信息不足时的澄清话术（随机）。""" + templates = [ + "不好意思，不太懂你的意思，你再具体说下哈。", + "抱歉我这边没完全理解，你可以换个说法再说一次吗？", + "我有点没听明白，你是要找图还是要做图呀？", + "不好意思我没抓到重点，你再补一句我就能接着处理。", + "这句我理解得不太准，你再说具体一点我马上给你办。", + "抱歉，这里我没太看懂。你是想让我找原图，还是按图处理？", + "我这边还没完全明白你的意思，麻烦你再具体描述一下。", + "不好意思，这条我没读懂，你再详细说一点我马上跟上。", + ] + return random.choice(templates) + + +def append_requirement(state: Any, text: str) -> None: + """追加需求并做去重/截断，减少上下文噪音。""" + t = (text or "").strip() + if not t: + return + t = t[:120] + existing = list(getattr(state, "pending_requirements", []) or []) + if existing and existing[-1] == t: + return + if t in existing[-5:]: + return + existing.append(t) + if len(existing) > 20: + existing = existing[-20:] + state.pending_requirements = existing diff --git a/core/pydantic_ai_agent.py b/core/pydantic_ai_agent.py index 10232cd..6ba9edb 100755 --- a/core/pydantic_ai_agent.py +++ b/core/pydantic_ai_agent.py @@ -32,6 +32,22 @@ from core.ai_reply_flow import execute_ai_turn from core.reply_finalize_flow import finalize_ai_reply from core.prompt_flow import build_prompt_bundle from core.order_helpers import parse_order_info, order_instruction as build_order_instruction +from core.collection_intent_helpers import ( + append_requirement, + build_collect_ack, + build_collect_progress_reply, + build_collect_remind, + build_find_image_clarify_reply, + build_not_understood_reply, + classify_short_customer_text, + is_batch_finish_intent, + is_batch_finish_signal, + is_cross_image_composite_intent, + is_find_image_not_edit_conflict, + is_related_image_followup_intent, + is_result_followup_query, + needs_clarification_in_collecting, +) load_dotenv() @@ -1932,302 +1948,20 @@ class CustomerServiceAgent: text = re.sub(r'\s+', ' ', text) return text.strip("，,。.!！？?；;：: ") - def _is_batch_finish_signal(self, text: str) -> bool: - """客户是否表达“图发完了，可以统一报价”。""" - if not text: - return False - if self._classify_short_customer_text(text) == "finish_signal": - return True - finish_keywords = [ - "发完了", "都发完了", "发齐了", "齐了", "先这些", "就这些", "全部", "一起报", "统一报价", - "总共多少钱", "一共多少钱", "打包价", "总价", "报价吧", "报个总价", "给个总价", - "没了", "没有了", "没图了", "就这", "就这张", "就这一张", "就这一个", "就一个", - "先报吧", "报下价", "报个价", "可以报价了", "能报吗", - ] - return any(k in text for k in finish_keywords) - - def _is_batch_finish_intent( - self, - text: str, - state: ConversationState, - has_incoming_urls: bool, - ) -> bool: - """ - 语义结束识别： - - 显式口令：发完了/统一报价 - - 隐式意图：询价/砍价 - - 单图需求明确：如“这个门头上面的字做一下”可直接进入报价 - """ - if not text: - return False - if self._is_batch_finish_signal(text): - return True - if has_incoming_urls: - return False - if not state.pending_image_urls: - return False - - # 意图识别：询价/砍价通常意味着“可以报价了” - try: - from utils.intent_analyzer import detect_intent_embedding, detect_intent_keywords - intent = detect_intent_embedding(text) or detect_intent_keywords(text) - except Exception: - intent = "" - if intent in ("询价", "砍价"): - return True - - msg = (text or "").strip() - if not msg: - return False - # 单图场景：客户给出明确加工指令，可直接报价 - single_image_action_kw = ( - "做一下", "改一下", "处理一下", "就这张", "按这个做", "照这个做", - "这个门头", "上面的字", "这个字", "这个图做", "能做吗", - ) - multi_image_finish_kw = ( - "就这些", "就这几张", "按这几张", "这几张一起做", "一起做一下", - "先按这些", "先按这几张", "直接报价", "现在报价", "看下报价", - "先报个总价", "总价多少", "一起多少钱", "先做这几张", - ) - hold_kw = ("还有", "再发", "先等", "稍后", "等会", "回头") - if len(state.pending_image_urls) == 1: - if any(k in msg for k in single_image_action_kw) and not any(k in msg for k in hold_kw): - return True - elif len(state.pending_image_urls) >= 2: - if any(k in msg for k in multi_image_finish_kw) and not any(k in msg for k in hold_kw): - return True - if self._is_cross_image_composite_intent(msg) and not any(k in msg for k in hold_kw): - return True - - return False - - @staticmethod - def _is_cross_image_composite_intent(text: str) -> bool: - """ - 识别多图跨图修改意图（A图元素放到B图）。 - 例：A图的图案转到B图、这个图案放到另一张上。 - """ - s = (text or "").strip() - if not s: - return False - pair_marks = ("a图", "b图", "第一张", "第二张", "这张", "那张", "上一张", "另一张") - op_kw = ( - "转到", "换到", "放到", "贴到", "移到", "套到", "合成", "融合", "替换到", - "图案上去", "字放到", "元素放到", "logo放到", - ) - return any(k in s.lower() for k in pair_marks) and any(k in s for k in op_kw) - - @staticmethod - def _is_related_image_followup_intent(text: str) -> bool: - """ - 识别“新发的是上一张的截图/局部细节”的关联意图。 - 这类输入应与前图关联处理，避免当成完全独立需求。 - """ - s = (text or "").strip().lower() - if not s: - return False - relation_kw = ( - "截图", "截屏", "局部", "细节", "放大", "裁剪", "同一张", "同一幅", - "上一张", "上张", "前一张", "前面那张", "刚才那张", "这个是上面", - "这个是那张", "补一张细节", "补个截图", - ) - return any(k in s for k in relation_kw) - - @staticmethod - def _is_result_followup_query(text: str) -> bool: - """识别客户在找图流程中的结果/进度追问。""" - short_type = CustomerServiceAgent._classify_short_customer_text(text) - if short_type == "progress_query": - return True - s = (text or "").strip() - if not s: - return False - followup_kw = ( - "找到了吗", "没找到吗", "找到没", "找到没有", "找到了没", "有吗", "有没", "有没有", - "有结果吗", "结果呢", - "进度", "多久好", "什么时候好", "好了没", "弄好了吗", "做了没", - "你重新发", "重新发给我", "高清", "发我", - ) - if any(k in s for k in followup_kw): - return True - return s in {"?", "？", "在吗", "人呢"} - - @staticmethod - def _classify_short_customer_text(text: str) -> str: - """ - 短句分类器（状态机前置）： - - finish_signal: 发图完成，可报价 - - progress_query: 追问进度/结果 - - ack: 简短确认 - - unknown: 未识别 - """ - s = (text or "").strip() - if not s: - return "unknown" - if len(s) > 8: - return "unknown" - - finish_kw = ( - "没了", "没有了", "就这", "就这张", "就这一张", "就这一个", "就一个", - "先这些", "就这些", "发完了", "都发完了", - ) - if any(k in s for k in finish_kw): - return "finish_signal" - - progress_kw = ( - "有吗", "有没", "有没有", "找到了吗", "找到了没", "没找到吗", "找到没", "找到没有", - "进度", "结果", "多久好", "什么时候好", "好了没", "弄好了吗", "做了没", - "高清", "发我", "重新发", "你重新发给我", - ) - if any(k in s for k in progress_kw) or s in {"?", "？", "在吗", "人呢"}: - return "progress_query" - - ack_kw = ("嗯", "嗯嗯", "好", "好的", "行", "可以", "ok", "OK", "收到", "明白") - if s in ack_kw: - return "ack" - return "unknown" - - def _build_collect_ack(self, count: int, related_followup: bool = False) -> str: - if related_followup and count >= 2: - related_templates = [ - "这张我收到了，看起来是上一张的截图/细节图，我按同一单一起处理。还有补充就继续发。", - "收到，这张是关联补图我记上了（按同一需求处理）。你还有图就继续发。", - "明白，这张是前图的局部截图，我会和前面那张一起算，不会分开漏掉。", - ] - return random.choice(related_templates) - if count <= 1: - one_templates = [ - "这张收到啦，还有图就继续发，我一起给你看。", - "图我看到了，后面还有就接着发，最后我一口价给你。", - "收到这张了，你有其他图也发来，我统一帮你算。", - "这张我先记上了，你那边还有的话接着发，我一起给你报。", - "第1张收到，你继续发就行，发完我这边一次给你算清楚。", - "这张没问题，我先收着。要是还有图，你直接连着发我就行。", - "我先看到了这张，你后面还有就一起发来，我统一给你报价。", - "这张图我已经记下了，后面有补充就继续甩过来哈。", - ] - return random.choice(one_templates) - templates = [ - "这几张我都收到了（现在{n}张）。还有的话继续发，我一起给你报。", - "好嘞，先看到{n}张了。你可以继续发，或者直接说“就这些”我现在就报价。", - "收到哈（共{n}张）。你还要补图就继续发，不补的话我现在也可以直接给价。", - "我这边先收到了{n}张。你继续补图，或者直接说“按这些算”我就开始报。", - "这波我已经记了{n}张，你要是还有就接着发，不补的话我立刻给总价。", - "先看到{n}张图了，后面你看是继续发，还是直接让我现在报价都可以。", - "好的，目前{n}张到位。你一句“就这些”，我马上给你打包价。", - "图我都看到了（{n}张）。你还发我就继续收，不发我现在就给你报。", - ] - return random.choice(templates).format(n=count) - - def _build_collect_progress_reply(self, count: int) -> str: - if count <= 1: - templates = [ - "我这边在处理了，这张有结果我第一时间回你。", - "在跟进中，这张一有进展我马上发你。", - "这张我正在看，稍等我一会儿，结果出来就回你。", - ] - return random.choice(templates) - templates = [ - "我这边在按你这{n}张一起处理，有结果我立刻同步你。", - "正在跟进这{n}张，出结果我第一时间发你，不会漏。", - "进度在跑了（共{n}张），你稍等一下，我这边有结果马上回。", - ] - return random.choice(templates).format(n=count) - - def _build_collect_remind(self, count: int) -> str: - if count <= 1: - one_templates = [ - "这个要求我记住了。你还有图就继续发，不补图我就按这张给你报价。", - "明白，这个需求我加上了。你继续发图也行，想直接报价也可以。", - "我先记下这张。你如果是要我找图，不是做图，直接说一声，我按找图思路给你走。", - "收到，这张我先按你的要求记好了。就做这一张的话，我现在直接给你报实价。", - "你这要求我记下了，后面还有图就发，没有的话我现在直接算价。", - "行，我按你这个要求来。继续补图也行，不补我就先报这张。", - "这个点我懂了，你还要补图就接着发，不补我立刻给你报价。", - "要求我已经加上了。你看是继续发，还是我现在直接报这张。", - ] - return random.choice(one_templates) - templates = [ - "需求我记下了（当前{n}张）。你继续补图，或者直接说“就这些”我现在报价。", - "好，这个要求也加上了（现在{n}张）。不再补图的话我立刻给你打包价。", - "收到（共{n}张）。你还发就继续，不发的话我现在就给总价。", - "这个需求我加进去了（现在{n}张）。你继续发也行，直接报价也行。", - "我这边都记好了（{n}张+需求）。你一句“先按这些算”，我马上报价。", - "要求同步好了，目前{n}张。要补图继续发，不补图我现在就给你打包价。", - "行，需求和图片我都收着了（{n}张）。你直接让我报价也可以。", - "好的，这条需求也算进去了（共{n}张）。你看要不要我现在直接报。", - ] - return random.choice(templates).format(n=count) - - @staticmethod - def _is_find_image_not_edit_conflict(text: str) -> bool: - """识别客户明确声明“要找图，不是做图”的冲突语义。""" - s = (text or "").strip() - if not s: - return False - find_kw = ("找图", "找原图", "找素材", "找同款") - deny_edit_kw = ("不是让你做图", "不是做图", "不用做图", "不需要做图", "不是修图", "不用修图") - return any(k in s for k in find_kw) and any(k in s for k in deny_edit_kw) - - @staticmethod - def _needs_clarification_in_collecting(text: str) -> bool: - """ - 信息不足时先追问，不急着报价。 - 例：这个也是大图 / 一共几个图 / 啥意思 / 没明白 - """ - s = (text or "").strip() - if not s: - return False - short_non_vague_kw = ( - "？", "?", "没了", "没有了", "就这", "行", "好的", "ok", "报价", - "找到了吗", "没找到吗", "找到没", "找到了没", "有吗", "有没", "有没有", - "多久好", "什么时候好", "高清", - ) - if len(s) <= 4: - if any(k in s for k in short_non_vague_kw): - return False - return True - vague_kw = ( - "这个也是", "一共几个图", "几个图", "啥意思", "没明白", "什么意思", - "这个呢", "这个可以吗", "然后呢", "咋办", "怎么搞", - ) - return any(k in s for k in vague_kw) - - def _build_find_image_clarify_reply(self, state: ConversationState) -> str: - count = len(state.pending_image_urls or []) - return ( - f"明白，你是要我帮你找图，不是做图。现在我这边先记了{count}张，" - "你告诉我具体要找哪种：原图/同款/高清版，我按这个方向给你找。" - ) - - @staticmethod - def _build_not_understood_reply() -> str: - """信息不足时的澄清话术（随机）。""" - templates = [ - "不好意思，不太懂你的意思，你再具体说下哈。", - "抱歉我这边没完全理解，你可以换个说法再说一次吗？", - "我有点没听明白，你是要找图还是要做图呀？", - "不好意思我没抓到重点，你再补一句我就能接着处理。", - "这句我理解得不太准，你再说具体一点我马上给你办。", - "抱歉，这里我没太看懂。你是想让我找原图，还是按图处理？", - "我这边还没完全明白你的意思，麻烦你再具体描述一下。", - "不好意思，这条我没读懂，你再详细说一点我马上跟上。", - ] - return random.choice(templates) - - def _append_requirement(self, state: ConversationState, text: str): - """追加需求并做去重/截断，减少上下文噪音。""" - t = (text or "").strip() - if not t: - return - t = t[:120] - if state.pending_requirements and state.pending_requirements[-1] == t: - return - if t in state.pending_requirements[-5:]: - return - state.pending_requirements.append(t) - if len(state.pending_requirements) > 20: - state.pending_requirements = state.pending_requirements[-20:] + _is_batch_finish_signal = staticmethod(is_batch_finish_signal) + _is_batch_finish_intent = staticmethod(is_batch_finish_intent) + _is_cross_image_composite_intent = staticmethod(is_cross_image_composite_intent) + _is_related_image_followup_intent = staticmethod(is_related_image_followup_intent) + _is_result_followup_query = staticmethod(is_result_followup_query) + _classify_short_customer_text = staticmethod(classify_short_customer_text) + _build_collect_ack = staticmethod(build_collect_ack) + _build_collect_progress_reply = staticmethod(build_collect_progress_reply) + _build_collect_remind = staticmethod(build_collect_remind) + _is_find_image_not_edit_conflict = staticmethod(is_find_image_not_edit_conflict) + _needs_clarification_in_collecting = staticmethod(needs_clarification_in_collecting) + _build_find_image_clarify_reply = staticmethod(build_find_image_clarify_reply) + _build_not_understood_reply = staticmethod(build_not_understood_reply) + _append_requirement = staticmethod(append_requirement) def _calc_requirement_surcharge(self, requirements: List[str]) -> Dict[str, Any]: """