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

refactor: extract pre-rules and find-image quote flow from agent

Browse Source
This commit is contained in:
jimi
2026-03-01 14:54:11 +08:00
parent 3c825547cf
commit e62b39e0c3
3 changed files with 430 additions and 356 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
core/agent_pre_rules.py Normal file
View File

@@ -0,0 +1,200 @@
from __future__ import annotations
import random
from datetime import datetime
from typing import TYPE_CHECKING, Optional
from core.rules import Rule, RuleContext, RuleEngine, RuleResult
from services.risk_service import RiskService
if TYPE_CHECKING:
from core.pydantic_ai_agent import (
AgentResponse,
ConversationState,
CustomerMessage,
CustomerServiceAgent,
)
class AgentPreRuleService:
"""Pre-processing rule chain for short replies, cooldown, and text risk."""
def __init__(self, agent: "CustomerServiceAgent", risk_service: RiskService):
self.agent = agent
self.risk_service = risk_service
self.engine = self._build_engine()
async def run(
self,
*,
message: "CustomerMessage",
state: "ConversationState",
trace_id: str,
) -> Optional["AgentResponse"]:
ctx = RuleContext(data={"message": message, "state": state, "trace_id": trace_id})
result = await self.engine.run(ctx)
if not result.stop:
return None
response = result.payload.get("response")
return response
def _build_engine(self) -> RuleEngine:
return RuleEngine(
rules=[
Rule(
name="meaningless_short_text",
priority=10,
predicate=self._rule_pred_meaningless_short_text,
action=self._rule_act_meaningless_short_text,
),
Rule(
name="cooldown_silent",
priority=20,
predicate=self._rule_pred_cooldown_silent,
action=self._rule_act_cooldown_silent,
),
Rule(
name="manual_risk_block",
priority=30,
predicate=self._rule_pred_manual_risk_block,
action=self._rule_act_manual_risk_block,
),
Rule(
name="text_risk_block",
priority=40,
predicate=self._rule_pred_text_risk_block,
action=self._rule_act_text_risk_block,
),
]
)
async def _rule_pred_meaningless_short_text(self, ctx: RuleContext) -> bool:
from core.pydantic_ai_agent import _is_meaningless_short_text
message = ctx.get("message")
return _is_meaningless_short_text(message.msg)
async def _rule_act_meaningless_short_text(self, ctx: RuleContext) -> RuleResult:
from core.pydantic_ai_agent import AgentResponse
message = ctx.get("message")
state = ctx.get("state")
trace_id = ctx.get("trace_id", "")
ping = random.choice(("嗯咯", "嗯啦", "嗯", "哦"))
state.last_reply_at = datetime.now()
self.agent._activity_log(
"agent_ping_reply",
trace_id=trace_id,
customer_id=message.from_id,
msg=message.msg,
reply=ping,
)
return RuleResult(
matched=True,
stop=True,
action="agent_ping_reply",
payload={"response": AgentResponse(reply=ping, should_reply=True, need_transfer=False)},
)
async def _rule_pred_cooldown_silent(self, ctx: RuleContext) -> bool:
message = ctx.get("message")
state = ctx.get("state")
return self.agent._in_cooldown(state, message.msg)
async def _rule_act_cooldown_silent(self, ctx: RuleContext) -> RuleResult:
from core.pydantic_ai_agent import AgentResponse
message = ctx.get("message")
state = ctx.get("state")
trace_id = ctx.get("trace_id", "")
elapsed = int((datetime.now() - state.last_reply_at).total_seconds()) if state.last_reply_at else 0
print(f"[Agent] 冷却期静默（距上次回复 {elapsed}s）：{message.msg!r}")
self.agent._activity_log(
"agent_cooldown_silent",
trace_id=trace_id,
customer_id=message.from_id,
elapsed_s=elapsed,
)
return RuleResult(
matched=True,
stop=True,
action="agent_cooldown_silent",
payload={"response": AgentResponse(reply="", should_reply=False, need_transfer=False)},
)
async def _rule_pred_manual_risk_block(self, ctx: RuleContext) -> bool:
message = ctx.get("message")
decision = self.risk_service.check_manual_block(message.from_id)
ctx.set("manual_risk_decision", decision)
return decision.blocked
async def _rule_act_manual_risk_block(self, ctx: RuleContext) -> RuleResult:
from core.pydantic_ai_agent import AgentResponse, TRANSFER_MESSAGE
message = ctx.get("message")
trace_id = ctx.get("trace_id", "")
decision = ctx.get("manual_risk_decision")
self.agent._activity_log(
"agent_manual_risk_reject",
trace_id=trace_id,
customer_id=message.from_id,
risk=(decision.profile if decision else {}),
)
return RuleResult(
matched=True,
stop=True,
action="agent_manual_risk_reject",
payload={
"response": AgentResponse(
reply="这边无法继续为你处理该类需求，给你转人工专员对接。",
should_reply=True,
need_transfer=True,
transfer_msg=TRANSFER_MESSAGE,
)
},
)
async def _rule_pred_text_risk_block(self, ctx: RuleContext) -> bool:
message = ctx.get("message")
decision = await self.risk_service.check_text_block(
message.msg,
political_detector=self.agent._is_political_inquiry,
map_detector=self.agent._is_map_inquiry,
)
ctx.set("text_risk_decision", decision)
return decision.blocked
async def _rule_act_text_risk_block(self, ctx: RuleContext) -> RuleResult:
from core.pydantic_ai_agent import AgentResponse
message = ctx.get("message")
state = ctx.get("state")
trace_id = ctx.get("trace_id", "")
decision = ctx.get("text_risk_decision")
state.pending_image_urls.clear()
state.pending_requirements.clear()
self.agent._sync_pending_quote_state(message.from_id, state)
reject_text = self.risk_service.build_reject_text(decision.category if decision else "other")
reply = await self.agent._rewrite_reply_with_ai(
message=message,
state=state,
reply=reject_text,
scene="risk_reject",
)
state.last_reply_at = datetime.now()
print(f"{self.agent.C_REPLY}[REPLY->CUSTOMER]{self.agent.C_RESET} {reply}")
self.agent._activity_log(
"agent_risk_reject",
trace_id=trace_id,
customer_id=message.from_id,
risk_category=(decision.category if decision else "other"),
risk_source=(decision.source if decision else "unknown"),
reply=reply,
)
return RuleResult(
matched=True,
stop=True,
action="agent_risk_reject",
payload={"response": AgentResponse(reply=reply, should_reply=True, need_transfer=False)},
)

215
core/find_image_flow.py Normal file
View File

@@ -0,0 +1,215 @@
from __future__ import annotations
from datetime import datetime
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from core.pydantic_ai_agent import AgentResponse, ConversationState, CustomerMessage, CustomerServiceAgent
async def handle_find_image_batch_flow(
agent: "CustomerServiceAgent",
*,
message: "CustomerMessage",
state: "ConversationState",
customer_text: str,
shop_type: str,
) -> Optional["AgentResponse"]:
"""Handle find-image collecting/quote flow. Return response when handled."""
from core.pydantic_ai_agent import AgentResponse, TRANSFER_MESSAGE
if not (shop_type == "find_image" and agent._is_batch_quote_enabled(message.from_id, message.acc_id)):
return None
incoming_urls = agent._extract_image_urls(customer_text)
text_without_urls = agent._strip_urls_from_text(customer_text)
short_intent = agent._classify_short_customer_text(text_without_urls)
if incoming_urls:
is_related_followup = bool(text_without_urls and agent._is_related_image_followup_intent(text_without_urls))
for u in incoming_urls:
if u not in state.pending_image_urls:
state.pending_image_urls.append(u)
if text_without_urls:
agent._append_requirement(state, text_without_urls)
if is_related_followup:
agent._append_requirement(state, "与上一张相关（截图/局部细节）")
state.image_count = len(state.pending_image_urls)
agent._refresh_quote_phase(state, "collecting")
agent._sync_pending_quote_state(message.from_id, state)
if agent._is_batch_finish_intent(
text=customer_text,
state=state,
has_incoming_urls=bool(incoming_urls),
):
should_defer = agent._should_defer_batch_quote(state, mark_ready=True)
agent._sync_pending_quote_state(message.from_id, state)
if should_defer:
defer_fallback = "图片和需求我都收齐了，我先整理下，马上给你报总价。"
defer_reply = await agent._render_collection_reply_with_ai(
message=message,
state=state,
scene="quote_defer_notice",
intent_hint="确认已收齐图片与需求，先承接，告知稍后马上报价。",
fallback=defer_fallback,
)
state.last_reply_at = datetime.now()
print(f"{agent.C_REPLY}[REPLY->CUSTOMER]{agent.C_RESET} {defer_reply}")
return AgentResponse(reply=defer_reply, should_reply=True, need_transfer=False)
quote_res = await agent._quote_pending_images(state, message)
reply_text = agent._colloquialize_reply(quote_res.get("reply", ""))
reply_text = await agent._rewrite_reply_with_ai(
message=message,
state=state,
reply=reply_text,
scene="batch_quote_reply",
)
need_transfer = bool(quote_res.get("need_transfer"))
state.last_reply_at = datetime.now()
print(f"{agent.C_REPLY}[REPLY->CUSTOMER]{agent.C_RESET} {reply_text}")
return AgentResponse(
reply=reply_text,
should_reply=not need_transfer,
need_transfer=need_transfer,
transfer_msg=TRANSFER_MESSAGE if need_transfer else "",
)
ack_fallback = "图片收到了，你有补充就继续发，我这边一起看。"
ack_intent = (
"告知图片已收到；如果客户继续发图就继续收，发完可统一报价。"
if not is_related_followup
else "告知这是和上一张相关的截图/局部图，已按同一需求一起处理。"
)
ack = await agent._render_collection_reply_with_ai(
message=message,
state=state,
scene="collect_ack",
intent_hint=ack_intent,
fallback=ack_fallback,
)
state.last_reply_at = datetime.now()
print(f"{agent.C_REPLY}[REPLY->CUSTOMER]{agent.C_RESET} {ack}")
return AgentResponse(reply=ack, should_reply=True, need_transfer=False)
if not state.pending_image_urls:
return None
if text_without_urls:
if short_intent == "finish_signal":
agent._mark_quote_ready(state)
elif short_intent == "progress_query":
if state.quote_phase != "ready_to_quote":
agent._refresh_quote_phase(state, "waiting_result")
elif short_intent == "ack":
if state.quote_phase != "ready_to_quote":
agent._refresh_quote_phase(state, "collecting")
else:
agent._append_requirement(state, text_without_urls)
agent._refresh_quote_phase(state, "collecting")
agent._sync_pending_quote_state(message.from_id, state)
if agent._is_find_image_not_edit_conflict(text_without_urls):
clarify_fallback = "明白你是要找图，不是做图。你说下要找原图、同款还是高清版，我按这个给你找。"
clarify = await agent._render_collection_reply_with_ai(
message=message,
state=state,
scene="find_not_edit_clarify",
intent_hint="确认客户要找图不是做图，并追问是找原图/同款/高清版。",
fallback=clarify_fallback,
)
state.last_reply_at = datetime.now()
print(f"{agent.C_REPLY}[REPLY->CUSTOMER]{agent.C_RESET} {clarify}")
return AgentResponse(reply=clarify, should_reply=True, need_transfer=False)
if state.quote_phase == "ready_to_quote" and state.quote_ready_turns <= 0 and short_intent in {"progress_query", "ack", "finish_signal"}:
quote_res = await agent._quote_pending_images(state, message)
reply_text = agent._colloquialize_reply(quote_res.get("reply", ""))
reply_text = await agent._rewrite_reply_with_ai(
message=message,
state=state,
reply=reply_text,
scene="batch_quote_reply",
)
need_transfer = bool(quote_res.get("need_transfer"))
state.last_reply_at = datetime.now()
print(f"{agent.C_REPLY}[REPLY->CUSTOMER]{agent.C_RESET} {reply_text}")
return AgentResponse(
reply=reply_text,
should_reply=not need_transfer,
need_transfer=need_transfer,
transfer_msg=TRANSFER_MESSAGE if need_transfer else "",
)
if short_intent == "progress_query" or agent._is_result_followup_query(text_without_urls):
progress_fallback = "我这边在跟进了，一有结果马上发你。"
progress = await agent._render_collection_reply_with_ai(
message=message,
state=state,
scene="collect_progress",
intent_hint="承接客户的进度/结果追问，简短说明正在跟进，有结果会第一时间回复。",
fallback=progress_fallback,
)
state.last_reply_at = datetime.now()
print(f"{agent.C_REPLY}[REPLY->CUSTOMER]{agent.C_RESET} {progress}")
return AgentResponse(reply=progress, should_reply=True, need_transfer=False)
if agent._needs_clarification_in_collecting(text_without_urls):
ask_fallback = "你再补一句具体要什么效果，我马上按你的要求来。"
ask = await agent._render_collection_reply_with_ai(
message=message,
state=state,
scene="collect_clarify",
intent_hint="客户表达不清，礼貌请对方补充一句关键需求，不要机械，不要生硬。",
fallback=ask_fallback,
)
state.last_reply_at = datetime.now()
print(f"{agent.C_REPLY}[REPLY->CUSTOMER]{agent.C_RESET} {ask}")
return AgentResponse(reply=ask, should_reply=True, need_transfer=False)
if agent._is_batch_finish_intent(
text=customer_text,
state=state,
has_incoming_urls=False,
):
should_defer = agent._should_defer_batch_quote(state, mark_ready=True)
agent._sync_pending_quote_state(message.from_id, state)
if should_defer:
defer_fallback = "收到，我先把这批图过一遍，马上给你总价。"
defer_reply = await agent._render_collection_reply_with_ai(
message=message,
state=state,
scene="quote_defer_notice",
intent_hint="确认已收齐，先承接并告知稍后马上报价。",
fallback=defer_fallback,
)
state.last_reply_at = datetime.now()
print(f"{agent.C_REPLY}[REPLY->CUSTOMER]{agent.C_RESET} {defer_reply}")
return AgentResponse(reply=defer_reply, should_reply=True, need_transfer=False)
quote_res = await agent._quote_pending_images(state, message)
reply_text = agent._colloquialize_reply(quote_res.get("reply", ""))
reply_text = await agent._rewrite_reply_with_ai(
message=message,
state=state,
reply=reply_text,
scene="batch_quote_reply",
)
need_transfer = bool(quote_res.get("need_transfer"))
state.last_reply_at = datetime.now()
print(f"{agent.C_REPLY}[REPLY->CUSTOMER]{agent.C_RESET} {reply_text}")
return AgentResponse(
reply=reply_text,
should_reply=not need_transfer,
need_transfer=need_transfer,
transfer_msg=TRANSFER_MESSAGE if need_transfer else "",
)
remind_fallback = "需求我记上了，你继续发图，或者让我直接给你报价都行。"
remind = await agent._render_collection_reply_with_ai(
message=message,
state=state,
scene="collect_remind",
intent_hint="确认需求已记录，引导客户继续补图或直接让你报价。",
fallback=remind_fallback,
)
state.last_reply_at = datetime.now()
print(f"{agent.C_REPLY}[REPLY->CUSTOMER]{agent.C_RESET} {remind}")
return AgentResponse(reply=remind, should_reply=True, need_transfer=False)

365
core/pydantic_ai_agent.py
View File

@@ -24,8 +24,9 @@ from dotenv import load_dotenv
from utils.metrics_tracker import emit as metrics_emit
from utils.observability import emit_activity, build_trace_id
from core.quote_state_machine import QuoteStateMachine
from core.rules import Rule, RuleContext, RuleEngine, RuleResult
from services.risk_service import RiskService
from core.agent_pre_rules import AgentPreRuleService
from core.find_image_flow import handle_find_image_batch_flow
load_dotenv()
@@ -235,7 +236,7 @@ class CustomerServiceAgent:
self.batch_quote_delay_turns = 1
self.quote_state_machine = QuoteStateMachine(delay_turns=self.batch_quote_delay_turns)
self.risk_service = RiskService()
self._pre_rule_engine = self._build_pre_rule_engine()
self.pre_rule_service = AgentPreRuleService(self, self.risk_service)
if not self.api_key:
raise ValueError("请设置 OPENAI_API_KEY 环境变量")
@@ -1764,157 +1765,6 @@ class CustomerServiceAgent:
clean = msg.strip().rstrip("！!？?。.～~")
return clean in self._COOLDOWN_PATTERNS
def _build_pre_rule_engine(self) -> RuleEngine:
return RuleEngine(
rules=[
Rule(
name="meaningless_short_text",
priority=10,
predicate=self._rule_pred_meaningless_short_text,
action=self._rule_act_meaningless_short_text,
),
Rule(
name="cooldown_silent",
priority=20,
predicate=self._rule_pred_cooldown_silent,
action=self._rule_act_cooldown_silent,
),
Rule(
name="manual_risk_block",
priority=30,
predicate=self._rule_pred_manual_risk_block,
action=self._rule_act_manual_risk_block,
),
Rule(
name="text_risk_block",
priority=40,
predicate=self._rule_pred_text_risk_block,
action=self._rule_act_text_risk_block,
),
]
)
async def _rule_pred_meaningless_short_text(self, ctx: RuleContext) -> bool:
message: CustomerMessage = ctx.get("message")
return _is_meaningless_short_text(message.msg)
async def _rule_act_meaningless_short_text(self, ctx: RuleContext) -> RuleResult:
message: CustomerMessage = ctx.get("message")
state: ConversationState = ctx.get("state")
trace_id = ctx.get("trace_id", "")
ping = random.choice(("嗯咯", "嗯啦", "嗯", "哦"))
state.last_reply_at = datetime.now()
self._activity_log(
"agent_ping_reply",
trace_id=trace_id,
customer_id=message.from_id,
msg=message.msg,
reply=ping,
)
return RuleResult(
matched=True,
stop=True,
action="agent_ping_reply",
payload={"response": AgentResponse(reply=ping, should_reply=True, need_transfer=False)},
)
async def _rule_pred_cooldown_silent(self, ctx: RuleContext) -> bool:
message: CustomerMessage = ctx.get("message")
state: ConversationState = ctx.get("state")
return self._in_cooldown(state, message.msg)
async def _rule_act_cooldown_silent(self, ctx: RuleContext) -> RuleResult:
message: CustomerMessage = ctx.get("message")
state: ConversationState = ctx.get("state")
trace_id = ctx.get("trace_id", "")
elapsed = int((datetime.now() - state.last_reply_at).total_seconds()) if state.last_reply_at else 0
print(f"[Agent] 冷却期静默（距上次回复 {elapsed}s）：{message.msg!r}")
self._activity_log(
"agent_cooldown_silent",
trace_id=trace_id,
customer_id=message.from_id,
elapsed_s=elapsed,
)
return RuleResult(
matched=True,
stop=True,
action="agent_cooldown_silent",
payload={"response": AgentResponse(reply="", should_reply=False, need_transfer=False)},
)
async def _rule_pred_manual_risk_block(self, ctx: RuleContext) -> bool:
message: CustomerMessage = ctx.get("message")
decision = self.risk_service.check_manual_block(message.from_id)
ctx.set("manual_risk_decision", decision)
return decision.blocked
async def _rule_act_manual_risk_block(self, ctx: RuleContext) -> RuleResult:
message: CustomerMessage = ctx.get("message")
trace_id = ctx.get("trace_id", "")
decision = ctx.get("manual_risk_decision")
self._activity_log(
"agent_manual_risk_reject",
trace_id=trace_id,
customer_id=message.from_id,
risk=(decision.profile if decision else {}),
)
return RuleResult(
matched=True,
stop=True,
action="agent_manual_risk_reject",
payload={
"response": AgentResponse(
reply="这边无法继续为你处理该类需求，给你转人工专员对接。",
should_reply=True,
need_transfer=True,
transfer_msg=TRANSFER_MESSAGE,
)
},
)
async def _rule_pred_text_risk_block(self, ctx: RuleContext) -> bool:
message: CustomerMessage = ctx.get("message")
decision = await self.risk_service.check_text_block(
message.msg,
political_detector=self._is_political_inquiry,
map_detector=self._is_map_inquiry,
)
ctx.set("text_risk_decision", decision)
return decision.blocked
async def _rule_act_text_risk_block(self, ctx: RuleContext) -> RuleResult:
message: CustomerMessage = ctx.get("message")
state: ConversationState = ctx.get("state")
trace_id = ctx.get("trace_id", "")
decision = ctx.get("text_risk_decision")
state.pending_image_urls.clear()
state.pending_requirements.clear()
self._sync_pending_quote_state(message.from_id, state)
reject_text = self.risk_service.build_reject_text(decision.category if decision else "other")
reply = await self._rewrite_reply_with_ai(
message=message,
state=state,
reply=reject_text,
scene="risk_reject",
)
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {reply}")
self._activity_log(
"agent_risk_reject",
trace_id=trace_id,
customer_id=message.from_id,
risk_category=(decision.category if decision else "other"),
risk_source=(decision.source if decision else "unknown"),
reply=reply,
)
return RuleResult(
matched=True,
stop=True,
action="agent_risk_reject",
payload={"response": AgentResponse(reply=reply, should_reply=True, need_transfer=False)},
)
async def process_message(self, message: CustomerMessage) -> AgentResponse:
"""处理客户消息并生成回复"""
trace_id = build_trace_id(message.acc_id, message.from_id, message.msg_id, message.msg[:64])
@@ -1929,12 +1779,9 @@ class CustomerServiceAgent:
metrics_emit("inbound_msg", customer_id=message.from_id, acc_id=message.acc_id)
# 获取或创建对话状态
state = self._get_conversation_state(message.from_id)
pre_ctx = RuleContext(data={"message": message, "state": state, "trace_id": trace_id})
pre_result = await self._pre_rule_engine.run(pre_ctx)
if pre_result.stop:
response = pre_result.payload.get("response")
if isinstance(response, AgentResponse):
return response
pre_response = await self.pre_rule_service.run(message=message, state=state, trace_id=trace_id)
if isinstance(pre_response, AgentResponse):
return pre_response
# 检测售前/售后
new_stage = self._detect_stage(message.msg)
@@ -1979,205 +1826,17 @@ class CustomerServiceAgent:
print(f"[Agent] 订单通知静默（{pay_status or order_status}），跳过回复")
return AgentResponse(reply="", should_reply=False, need_transfer=False)
# 找图店：先收集图片和需求，等客户确认“发完”后统一报价
customer_text, _ = self._split_customer_text(message.msg)
shop_type = _get_shop_type(message.acc_id or "", message.goods_name or "")
if shop_type == "find_image" and self._is_batch_quote_enabled(message.from_id, message.acc_id):
incoming_urls = self._extract_image_urls(customer_text)
text_without_urls = self._strip_urls_from_text(customer_text)
short_intent = self._classify_short_customer_text(text_without_urls)
if incoming_urls:
is_related_followup = bool(text_without_urls and self._is_related_image_followup_intent(text_without_urls))
for u in incoming_urls:
if u not in state.pending_image_urls:
state.pending_image_urls.append(u)
if text_without_urls:
self._append_requirement(state, text_without_urls)
if is_related_followup:
self._append_requirement(state, "与上一张相关（截图/局部细节）")
state.image_count = len(state.pending_image_urls)
self._refresh_quote_phase(state, "collecting")
self._sync_pending_quote_state(message.from_id, state)
if self._is_batch_finish_intent(
text=customer_text,
state=state,
has_incoming_urls=bool(incoming_urls),
):
should_defer = self._should_defer_batch_quote(state, mark_ready=True)
self._sync_pending_quote_state(message.from_id, state)
if should_defer:
defer_fallback = "图片和需求我都收齐了，我先整理下，马上给你报总价。"
defer_reply = await self._render_collection_reply_with_ai(
flow_response = await handle_find_image_batch_flow(
self,
message=message,
state=state,
scene="quote_defer_notice",
intent_hint="确认已收齐图片与需求，先承接，告知稍后马上报价。",
fallback=defer_fallback,
customer_text=customer_text,
shop_type=shop_type,
)
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {defer_reply}")
return AgentResponse(reply=defer_reply, should_reply=True, need_transfer=False)
quote_res = await self._quote_pending_images(state, message)
reply_text = self._colloquialize_reply(quote_res.get("reply", ""))
reply_text = await self._rewrite_reply_with_ai(
message=message,
state=state,
reply=reply_text,
scene="batch_quote_reply",
)
need_transfer = bool(quote_res.get("need_transfer"))
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {reply_text}")
return AgentResponse(
reply=reply_text,
should_reply=not need_transfer,
need_transfer=need_transfer,
transfer_msg=TRANSFER_MESSAGE if need_transfer else "",
)
ack_fallback = "图片收到了，你有补充就继续发，我这边一起看。"
ack_intent = (
"告知图片已收到；如果客户继续发图就继续收，发完可统一报价。"
if not is_related_followup
else "告知这是和上一张相关的截图/局部图，已按同一需求一起处理。"
)
ack = await self._render_collection_reply_with_ai(
message=message,
state=state,
scene="collect_ack",
intent_hint=ack_intent,
fallback=ack_fallback,
)
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {ack}")
return AgentResponse(reply=ack, should_reply=True, need_transfer=False)
if state.pending_image_urls:
if text_without_urls:
# 短句先分类再路由，避免误追加为需求导致上下文漂移
if short_intent == "finish_signal":
self._mark_quote_ready(state)
elif short_intent == "progress_query":
if state.quote_phase != "ready_to_quote":
self._refresh_quote_phase(state, "waiting_result")
elif short_intent == "ack":
if state.quote_phase != "ready_to_quote":
self._refresh_quote_phase(state, "collecting")
else:
self._append_requirement(state, text_without_urls)
self._refresh_quote_phase(state, "collecting")
self._sync_pending_quote_state(message.from_id, state)
# 客户明确“找图，不是做图”时，先澄清意图，不继续报价链路
if self._is_find_image_not_edit_conflict(text_without_urls):
clarify_fallback = "明白你是要找图，不是做图。你说下要找原图、同款还是高清版，我按这个给你找。"
clarify = await self._render_collection_reply_with_ai(
message=message,
state=state,
scene="find_not_edit_clarify",
intent_hint="确认客户要找图不是做图，并追问是找原图/同款/高清版。",
fallback=clarify_fallback,
)
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {clarify}")
return AgentResponse(reply=clarify, should_reply=True, need_transfer=False)
# 已到报价就绪阶段且等待轮次结束：对“有吗/进度”等追问直接报价
if state.quote_phase == "ready_to_quote" and state.quote_ready_turns <= 0 and short_intent in {"progress_query", "ack", "finish_signal"}:
quote_res = await self._quote_pending_images(state, message)
reply_text = self._colloquialize_reply(quote_res.get("reply", ""))
reply_text = await self._rewrite_reply_with_ai(
message=message,
state=state,
reply=reply_text,
scene="batch_quote_reply",
)
need_transfer = bool(quote_res.get("need_transfer"))
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {reply_text}")
return AgentResponse(
reply=reply_text,
should_reply=not need_transfer,
need_transfer=need_transfer,
transfer_msg=TRANSFER_MESSAGE if need_transfer else "",
)
# 客户在追问“找到了吗/没找到吗/多久好”时，优先给进度承接，不走“没听懂”
if short_intent == "progress_query" or self._is_result_followup_query(text_without_urls):
progress_fallback = "我这边在跟进了，一有结果马上发你。"
progress = await self._render_collection_reply_with_ai(
message=message,
state=state,
scene="collect_progress",
intent_hint="承接客户的进度/结果追问，简短说明正在跟进，有结果会第一时间回复。",
fallback=progress_fallback,
)
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {progress}")
return AgentResponse(reply=progress, should_reply=True, need_transfer=False)
# 信息不足时先追问，避免误判为“直接报价”
if self._needs_clarification_in_collecting(text_without_urls):
ask_fallback = "你再补一句具体要什么效果，我马上按你的要求来。"
ask = await self._render_collection_reply_with_ai(
message=message,
state=state,
scene="collect_clarify",
intent_hint="客户表达不清，礼貌请对方补充一句关键需求，不要机械，不要生硬。",
fallback=ask_fallback,
)
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {ask}")
return AgentResponse(reply=ask, should_reply=True, need_transfer=False)
if self._is_batch_finish_intent(
text=customer_text,
state=state,
has_incoming_urls=False,
):
should_defer = self._should_defer_batch_quote(state, mark_ready=True)
self._sync_pending_quote_state(message.from_id, state)
if should_defer:
defer_fallback = "收到，我先把这批图过一遍，马上给你总价。"
defer_reply = await self._render_collection_reply_with_ai(
message=message,
state=state,
scene="quote_defer_notice",
intent_hint="确认已收齐，先承接并告知稍后马上报价。",
fallback=defer_fallback,
)
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {defer_reply}")
return AgentResponse(reply=defer_reply, should_reply=True, need_transfer=False)
quote_res = await self._quote_pending_images(state, message)
reply_text = self._colloquialize_reply(quote_res.get("reply", ""))
reply_text = await self._rewrite_reply_with_ai(
message=message,
state=state,
reply=reply_text,
scene="batch_quote_reply",
)
need_transfer = bool(quote_res.get("need_transfer"))
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {reply_text}")
return AgentResponse(
reply=reply_text,
should_reply=not need_transfer,
need_transfer=need_transfer,
transfer_msg=TRANSFER_MESSAGE if need_transfer else "",
)
remind_fallback = "需求我记上了，你继续发图，或者让我直接给你报价都行。"
remind = await self._render_collection_reply_with_ai(
message=message,
state=state,
scene="collect_remind",
intent_hint="确认需求已记录，引导客户继续补图或直接让你报价。",
fallback=remind_fallback,
)
state.last_reply_at = datetime.now()
print(f"{self.C_REPLY}[REPLY->CUSTOMER]{self.C_RESET} {remind}")
return AgentResponse(reply=remind, should_reply=True, need_transfer=False)
if isinstance(flow_response, AgentResponse):
return flow_response
# 构建提示词（包含对话状态 + 客户画像）
user_prompt = self._build_prompt(message, state)