284 lines
16 KiB
Python
284 lines
16 KiB
Python
"""Compose knowledge, freight and model output into the existing SSE protocol."""
|
||
|
||
import json
|
||
import re
|
||
|
||
from . import ai_client
|
||
from .ai_client import compress_answer, gen_followups, has_chinese
|
||
from .config import get_setting
|
||
from .db import connect, last_user_question, load_context, save_message
|
||
from .knowledge import search_knowledge
|
||
from .pricing import (parse_shipping_intent, detect_dim_unit, freight_reply,
|
||
calc_warehouse_card, detect_surcharge, surcharge_reply, check_island,
|
||
is_shipping_destination_reply)
|
||
from .tracking import CARD_PREFIX
|
||
from .order_data import answer_query, extract_reference
|
||
|
||
|
||
|
||
def sse(payload):
|
||
"""统一的事件格式：data: {"t":正文} / {"e":错误} / {"card":卡片} / {"d":1结束}
|
||
JSON 编码可以避免正文里的换行把 SSE 帧撑断。"""
|
||
return 'data: ' + json.dumps(payload, ensure_ascii=False) + '\n\n'
|
||
|
||
|
||
QUOTE_FOLLOWUP_PREFIX = r'^(?:那(?:么)?|换成|改成|改为|如果|还是|再算|同样)'
|
||
|
||
|
||
def is_shipping_followup(message, intent=None):
|
||
"""Recognize bounded quote edits without carrying facts across new topics."""
|
||
if re.search(r'新(?:的)?一(?:票|批)|另(?:一|外一?)(?:票|批)|新(?:的)?(?:货|订单)|换个话题|理赔|赔偿|破损|丢失|签收|退件|退款|投诉', message):
|
||
return False
|
||
intent = intent if intent is not None else parse_shipping_intent(message)
|
||
text = message.strip().strip('，,。.!！?？、：: ')
|
||
prefix = re.match(QUOTE_FOLLOWUP_PREFIX, text)
|
||
if prefix and any(intent.get(key) is not None for key in ('weight', 'dims', 'channel', 'region')):
|
||
return True
|
||
if intent['region'] is not None and is_shipping_destination_reply(text):
|
||
return True
|
||
short = re.sub(QUOTE_FOLLOWUP_PREFIX, '', text, count=1).strip()
|
||
short = re.sub(r'^(?:请问|请|再问一下)', '', short).strip()
|
||
if re.fullmatch(r'(?:海运|海派|空运|空派|铁路|班列|卡航|卡车|sea(?:\s+freight)?|air(?:\s+freight)?|rail|truck)'
|
||
r'(?:报价|价格|费用|运费|时效)?(?:呢|怎么样|如何|多少钱|多久|几天)?', short, re.I):
|
||
return True
|
||
if re.fullmatch(r'(?:价格|费用|运费|时效)(?:呢|是多少|多少|怎么样|如何|多久|几天)?|多少钱', short):
|
||
return True
|
||
if intent['weight'] is not None and re.fullmatch(r'\d+(?:\.\d+)?\s*(?:kg|公斤|千克|吨|lbs?|tons?|kilograms?)(?:呢)?', short, re.I):
|
||
return True
|
||
return bool(intent['dims'] and re.fullmatch(r'\d+(?:\.\d+)?\s*[*xX×]\s*\d+(?:\.\d+)?\s*[*xX×]\s*\d+(?:\.\d+)?\s*(?:cm|mm|m|厘米|毫米|米)?(?:呢)?', short, re.I))
|
||
|
||
|
||
def shipping_intent_with_context(message, conv_id):
|
||
intent = parse_shipping_intent(message)
|
||
intent['dim_unit'] = detect_dim_unit(message)
|
||
if conv_id and is_shipping_followup(message, intent):
|
||
connection = connect()
|
||
try:
|
||
recent = connection.execute("SELECT content FROM messages WHERE conv_id=? AND role='user' ORDER BY id DESC LIMIT 6", (conv_id,)).fetchall()
|
||
finally:
|
||
connection.close()
|
||
for (previous,) in recent:
|
||
if extract_reference(previous):
|
||
break
|
||
earlier = parse_shipping_intent(previous)
|
||
if not any(earlier.get(key) is not None for key in ('weight', 'dims', 'channel', 'region')):
|
||
if is_shipping_followup(previous, earlier):
|
||
continue
|
||
break
|
||
for key in ('weight', 'channel', 'region'):
|
||
if intent.get(key) is None:
|
||
intent[key] = earlier.get(key)
|
||
if intent['dims'] is None and earlier['dims']:
|
||
intent['dims'] = earlier['dims']
|
||
intent['dim_unit'] = detect_dim_unit(previous)
|
||
if not is_shipping_followup(previous, earlier):
|
||
break
|
||
if intent['channel']:
|
||
base = intent['channel'].removesuffix('_uk')
|
||
intent['channel'] = base + '_uk' if intent['region'] == 'uk' else base
|
||
return intent
|
||
|
||
|
||
def create_chat_stream(msg, conv_id):
|
||
if not msg:
|
||
return (sse({'e': '问题不能为空'}), {})
|
||
if conv_id:
|
||
from .visitor_service import owns_conversation
|
||
if not owns_conversation(conv_id):
|
||
from .customer_service import needs_customer_login, order_login_response
|
||
if needs_customer_login(conv_id):
|
||
return (sse(order_login_response()) + sse({'d':1}), {})
|
||
return (sse({'e':'该会话已删除，请新建对话后重试。'}), {})
|
||
order_answer = answer_query(msg, conv_id)
|
||
if order_answer is None and re.fullmatch(r'\[图片\]/uploads/\S+', msg):
|
||
order_answer = {'t':'收到您的图片，请描述一下您的问题。'}
|
||
ship = shipping_intent_with_context(msg, conv_id) if order_answer is None else None
|
||
if ship and ship['region'] is None and (ship['weight'] is not None or ship['dims'] or
|
||
re.search(r'\d+(?:\.\d+)?\s*(?:cbm|方|立方)', msg, re.I) or
|
||
(ship['channel'] and (is_shipping_followup(msg, ship) or re.search(r'报价|价格|费用|运费|多少钱|\b(?:quote|cost|price)\b', msg, re.I)))):
|
||
order_answer = {'t':'请提供目的国家或地区、目的仓或邮编，以便匹配对应费率；当前无法确定报价，可联系人工客服确认。'}
|
||
if ship and ship['region'] == 'unsupported' and (ship['weight'] is not None or ship['dims'] or ship['channel'] or
|
||
is_shipping_destination_reply(msg) or re.search(r'报价|价格|运费|费用|多少钱', msg)):
|
||
order_answer = {'t':'当前报价表尚未包含该目的地的费率，请提供目的仓或邮编、货物品名、件数和尺寸，由人工客服确认报价。'}
|
||
if ship and ship['region'] == 'uk' and re.search(r'\d+(?:\.\d+)?\s*(?:cbm|方|立方)', msg, re.I) and not ship['weight'] and not ship['dims']:
|
||
order_answer = {'t':'当前报价表未提供该目的地的按方费率，请提供目的仓或邮编、实际重量和尺寸，由人工客服确认报价。'}
|
||
if order_answer is not None:
|
||
|
||
def track_stream():
|
||
if 'card' in order_answer:
|
||
from .customer_service import bind_order_conversation, current_customer_id
|
||
if current_customer_id() and not bind_order_conversation(conv_id):
|
||
yield sse({'e':'该会话暂时无法访问，请新建对话后重试。','d':1})
|
||
return
|
||
save_message(conv_id, 'user', msg)
|
||
content = (CARD_PREFIX + json.dumps(order_answer['card'], ensure_ascii=False)
|
||
if 'card' in order_answer else order_answer['t'])
|
||
save_message(conv_id, 'assistant', content)
|
||
yield sse(order_answer)
|
||
yield sse({'d': 1})
|
||
return (track_stream(), {'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'})
|
||
if not get_setting('DEEPSEEK_API_KEY'):
|
||
return (sse({'e': '服务未配置 API Key，请在 .env 或环境变量中设置 DEEPSEEK_API_KEY 后重启服务'}), {})
|
||
know_text = search_knowledge(msg)
|
||
freight_text = None
|
||
if ship['weight'] or ship['dims']:
|
||
freight_text = freight_reply(ship)
|
||
if not ship['channel']:
|
||
freight_text += '\n\n（客户没有指定渠道，上面已同时给出海运和空运，请在回答末尾询问更看重时效还是成本）'
|
||
extra_parts = []
|
||
cbm_m = re.search('(\\d+(?:\\.\\d+)?)\\s*(?:cbm|方|立方)', msg, re.I)
|
||
if cbm_m and ship['region'] != 'uk':
|
||
cbm = float(cbm_m.group(1))
|
||
ch = '海运'
|
||
if re.search('卡航|卡车', msg):
|
||
ch = '卡航'
|
||
elif re.search('铁路|班列', msg):
|
||
ch = '铁路'
|
||
elif re.search('普卡', msg):
|
||
ch = '普卡'
|
||
want = None
|
||
for kw, ctry in (('德国', 'DE'), ('法国', 'FR'), ('波兰', 'PL'), ('捷克', 'CZ'), ('西班牙', 'ES'), ('马德里', 'ES'), ('中国', '')):
|
||
if kw in msg:
|
||
want = ctry
|
||
break
|
||
card = calc_warehouse_card(cbm, ch, country=want)
|
||
if card:
|
||
store = card['warehouse'].splitlines()[0]
|
||
extra_parts.append(f"【海外仓卡派 · 按方计费】\n 仓点：{store}\n 计费：{card['note']}\n 渠道：{card['channel']}，适用档位 {card['tier']}\n 运费：{card['formula']}\n 说明：单箱实重超 35KG 加收 30 元/箱（封顶 3600 元）；不接受仓储，特殊情况 85 元/箱/周")
|
||
sur = detect_surcharge(msg)
|
||
if sur:
|
||
extra_parts.append(surcharge_reply(sur[0], sur[1], ship.get('weight')))
|
||
pc_m = re.search('(?<![0-9])(\\d{4,5})(?![0-9])', msg)
|
||
if pc_m and re.search('邮编|postcode|zip|偏远|岛屿', msg, re.I):
|
||
hit = check_island(pc_m.group(1))
|
||
if hit:
|
||
extra_parts.append(f'【偏远判断】邮编 {pc_m.group(1)} 属于{hit[0]}（{hit[1]}）的岛屿/偏远地区，需加收 DPD 岛屿附加费 35 欧元/件')
|
||
else:
|
||
extra_parts.append(f'【偏远判断】邮编 {pc_m.group(1)} 未命中 DPD 岛屿偏远清单；但仍需以 UPS/DPD 官网查询结果为准，偏远费以账单为准')
|
||
follow_src = None
|
||
if not know_text:
|
||
prev_q = last_user_question(conv_id)
|
||
if prev_q:
|
||
know_text = search_knowledge(prev_q)
|
||
if know_text:
|
||
follow_src = prev_q
|
||
_c = connect()
|
||
_msg_count = _c.execute("SELECT COUNT(*) FROM messages WHERE conv_id=? AND role='user'", (conv_id,)).fetchone()[0]
|
||
_c.close()
|
||
is_first_turn = _msg_count == 0
|
||
save_message(conv_id, 'user', msg)
|
||
LENGTH_RULE = '回答长度要求：全文不少于 120 字、不超过 350 字。内容要完整、信息充分，把关键点讲清楚，可分 2~4 段或用短列表呈现。可以补充相关的说明（例如适用范围、注意事项、下一步建议），让对方看完就能明白。不要寒暄客套，不要重复同一个数字，不要罗列与问题无关的信息。'
|
||
is_en = bool(re.search('[A-Za-z]{3,}', msg)) and (not re.search('[\\u4e00-\\u9fa5]', msg))
|
||
if is_en:
|
||
LANGUAGE_RULE = '【语言要求 · 最高优先级】用户用英文提问，你必须用英文（English）回答全文，包括所有小节标题（如 Conclusion / Calculation / Notes），不要出现任何中文。以下给出的中文结构与参考资料仅用于理解，输出必须翻译成英文。'
|
||
else:
|
||
LANGUAGE_RULE = '【语言要求】用户用中文提问，请用中文回答。'
|
||
if freight_text or extra_parts:
|
||
blocks = []
|
||
if freight_text:
|
||
blocks.append(freight_text)
|
||
blocks.extend(extra_parts)
|
||
body = '\n\n'.join(blocks)
|
||
system_prompt = f'{LANGUAGE_RULE}\n\n你是柏悦物流智能客服。\n下面是根据报价表计算好的结果，请严格按照这些数字向客户说明。\n不要重新计算、不要改动任何数字，但要讲清楚是怎么算出来的。\n\n{body}\n\n请按以下结构输出（用英文回答时请把标题也翻译成英文）：\n【结论】先给结论：哪种方式、多少钱、多久到。\n【计算过程】分点列出，最多 5 行，让客户看懂钱花在哪。\n【补充说明】2~3 行：适用范围（如单件重量、尺寸要求）、偏远地区或敏感货附加费、下一步建议（如提供仓点或运单号可进一步核算）。\n\n{LENGTH_RULE}\n客户没指定渠道时，把主要几种渠道（海运/铁路/卡航/空运）列出来对比，并问客户更看重时效还是成本。\n'
|
||
elif know_text:
|
||
extra = f'\n（以上资料对应上一句提问：{follow_src}）' if follow_src else ''
|
||
system_prompt = f'{LANGUAGE_RULE}\n\n你是柏悦物流智能客服，严格优先使用下面知识库内容回答，不要自己编造业务信息。\n回答语句通顺自然，分段清晰，让人一眼看清重点。\n要结合上文理解用户的追问（比如"那空运呢""这个多少钱"指的是什么）。\n\n{LENGTH_RULE}\n\n知识库内容（供你理解，输出需按语言要求翻译）：{know_text}{extra}\n'
|
||
else:
|
||
system_prompt = f'{LANGUAGE_RULE}\n\n你是柏悦物流智能客服。\n物流相关简单问题可以正常回答；不属于物流业务的问题礼貌说明无法解答，并建议联系人工客服。\n回答语句通顺，表述自然。\n\n{LENGTH_RULE}\n如果用户在追问，请结合上文的对话继续回答，不要重复问已经说过的信息。\n'
|
||
context = load_context(conv_id, msg)
|
||
|
||
def generate():
|
||
body = {'model': get_setting('MODEL_NAME'), 'messages': [{'role': 'system', 'content': system_prompt}] + context, 'stream': True}
|
||
try:
|
||
resp = ai_client.post_completion(body, stream=True, timeout=(5, 90))
|
||
except ai_client.Timeout:
|
||
yield sse({'e': '模型响应超时，请稍后重试'})
|
||
return
|
||
except ai_client.RequestException:
|
||
yield sse({'e': '无法连接模型服务，请检查网络后重试'})
|
||
return
|
||
except Exception:
|
||
yield sse({'e': '服务异常，请稍后重试'})
|
||
return
|
||
if resp.status_code != 200:
|
||
detail = ''
|
||
try:
|
||
detail = (resp.json().get('error') or {}).get('message', '')
|
||
except Exception:
|
||
detail = (resp.text or '')[:120]
|
||
tips = {401: '系统繁忙，请稍后再试', 402: '系统繁忙，请稍后再试', 429: '咨询量较大，请稍后再试'}
|
||
tip = tips.get(resp.status_code, '系统繁忙，请稍后再试')
|
||
resp.close()
|
||
yield sse({'e': tip})
|
||
return
|
||
parts = []
|
||
interrupted = False
|
||
answer_saved = False
|
||
try:
|
||
got_text = False
|
||
completed = False
|
||
finish_reason = None
|
||
for raw in resp.iter_lines():
|
||
if not raw:
|
||
continue
|
||
line = raw.decode('utf-8', errors='ignore')
|
||
if not line.startswith('data:'):
|
||
continue
|
||
data = line[5:].strip()
|
||
if data == '[DONE]':
|
||
completed = True
|
||
break
|
||
try:
|
||
j = json.loads(data)
|
||
choice = j['choices'][0]
|
||
chunk = choice.get('delta', {}).get('content') or ''
|
||
finish_reason = choice.get('finish_reason') or finish_reason
|
||
if finish_reason == 'stop':
|
||
completed = True
|
||
except Exception:
|
||
continue
|
||
if chunk:
|
||
got_text = True
|
||
parts.append(chunk)
|
||
yield sse({'t': chunk})
|
||
if not got_text:
|
||
yield sse({'e': '模型没有返回内容，请重试'})
|
||
return
|
||
if not completed or finish_reason not in (None, 'stop'):
|
||
interrupted = True
|
||
yield sse({'e': '回答未生成完整，请点击重试'})
|
||
return
|
||
full_text = ''.join(parts)
|
||
limit = get_setting('MAX_ANSWER_CHARS') if has_chinese(full_text) else get_setting('MAX_ANSWER_CHARS_EN')
|
||
if len(full_text) > limit:
|
||
short = compress_answer(full_text, limit)
|
||
if len(short) > limit:
|
||
short = compress_answer(short, limit - 60)
|
||
if short != full_text:
|
||
full_text = short
|
||
yield sse({'replace': short})
|
||
save_message(conv_id, 'assistant', full_text)
|
||
answer_saved = True
|
||
if is_first_turn:
|
||
try:
|
||
fu = gen_followups(msg, full_text)
|
||
if fu:
|
||
yield sse({'suggest': fu})
|
||
except Exception:
|
||
pass
|
||
yield sse({'d': 1})
|
||
except GeneratorExit:
|
||
interrupted = True
|
||
raise
|
||
except Exception:
|
||
interrupted = True
|
||
yield sse({'e': '传输中断，请点击重试'})
|
||
finally:
|
||
text = ''.join(parts)
|
||
try:
|
||
if interrupted and text and not answer_saved:
|
||
save_message(conv_id, 'assistant', text + '\n\n（回答中断，以上内容未生成完整）')
|
||
finally:
|
||
resp.close()
|
||
return (generate(), {'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'})
|