zuowei/tw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: extract collection intent helpers from pydantic agent

Browse Source
This commit is contained in:
jimi
2026-03-01 15:38:20 +08:00
parent 54b1db17a7
commit fc05b60d1a
2 changed files with 463 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files

433
core/collection_intent_helpers.py Normal file
View File

@@ -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

326
core/pydantic_ai_agent.py
View File

@@ -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]:
"""