98 lines
5.3 KiB
Python
98 lines
5.3 KiB
Python
"""Execute whitelisted model requests under the current server identity."""
|
|
|
|
import json
|
|
import re
|
|
|
|
from .customer_service import order_login_response
|
|
from .visitor_service import customer_tenant_ids
|
|
|
|
|
|
def validate_calls(message):
|
|
from .model_prompts import TOOLS, PARAMETERS
|
|
|
|
calls = message.get('tool_calls')
|
|
if not isinstance(calls, list) or not 1 <= len(calls) <= 4:
|
|
raise ValueError('Invalid tool count')
|
|
names = {tool['function']['name'] for tool in TOOLS}
|
|
requests = []
|
|
ids = set()
|
|
for call in calls:
|
|
function = call.get('function', {})
|
|
call_id = call.get('id')
|
|
if call.get('type') != 'function' or function.get('name') not in names:
|
|
raise ValueError('Unknown tool')
|
|
if not isinstance(call_id, str) or not 1 <= len(call_id) <= 200 or call_id in ids:
|
|
raise ValueError('Invalid call ID')
|
|
arguments = json.loads(function.get('arguments', ''))
|
|
name = function['name']
|
|
if not isinstance(arguments, dict) or set(arguments) != {'question', *PARAMETERS[name]}:
|
|
raise ValueError('Invalid tool arguments')
|
|
question = arguments['question']
|
|
if not isinstance(question, str) or not 0 < len(question.strip()) <= 2000:
|
|
raise ValueError('Invalid tool question')
|
|
for key, schema in PARAMETERS[name].items():
|
|
value = arguments[key]
|
|
if 'enum' in schema and value not in schema['enum']:
|
|
raise ValueError('Invalid tool choice')
|
|
if key == 'services' and (not isinstance(value, list) or len(value) > 12 or
|
|
any(not isinstance(service, str) or not 0 < len(service) <= 100 for service in value)):
|
|
raise ValueError('Invalid channels')
|
|
if key == 'reference' and (not isinstance(value, str) or len(value) > 64 or
|
|
value and not re.fullmatch(r'[A-Za-z0-9_-]+', value)):
|
|
raise ValueError('Invalid order reference')
|
|
if name == 'query_orders' and ((arguments['condition'] == 'stage' and not arguments['stage']) or
|
|
(arguments['condition'] == 'status' and not arguments['status'])):
|
|
raise ValueError('Missing order condition')
|
|
ids.add(call_id)
|
|
requests.append((call_id, name, {**arguments, 'question': question.strip()}))
|
|
return requests
|
|
|
|
|
|
def execute_tool(name, arguments, conv_id):
|
|
question = arguments['question']
|
|
if name == 'query_orders':
|
|
if not customer_tenant_ids():
|
|
return order_login_response()
|
|
from .order_data import answer_query
|
|
if arguments['action'] == 'lookup':
|
|
from .order_data import IDENTIFIER
|
|
if not arguments['reference']:
|
|
return {'t': '请提供单个入仓码查询实际物流轨迹。'}
|
|
question = arguments['reference'] + ' ' + IDENTIFIER.sub('', question)
|
|
if arguments['action'] in ('count', 'list'):
|
|
from .order_data import STATUS_LABELS
|
|
from .order_query import STAGE_LABELS
|
|
condition = arguments['condition']
|
|
if condition == 'unsupported':
|
|
return {'t': '当前尚不支持该筛选条件，请提供入仓码逐票查看实际轨迹。'}
|
|
if arguments['action'] == 'count':
|
|
from .order_listing import answer_summary
|
|
query = {'status': arguments['status'] or None, 'stage': arguments['stage'] or None,
|
|
'exception': condition == 'exception', 'grouped': condition == 'grouped', 'unsupported': False}
|
|
return {**answer_summary(query, customer_tenant_ids(), conv_id), '_customer_data': True}
|
|
label = ('异常' if condition == 'exception' else '')
|
|
label += STAGE_LABELS.get(arguments['stage'], '') + STATUS_LABELS.get(arguments['status'], '')
|
|
question = '列出我的' + label + '订单'
|
|
answer = answer_query(question, conv_id)
|
|
if answer is None:
|
|
answer = {'t': '请提供入仓码，或说明要查询的订单统计条件。'}
|
|
return {**answer, '_customer_data': True}
|
|
if name == 'quote_shipping':
|
|
from .quote_service import answer_quote
|
|
from .quote_scope import answer_us_fallback
|
|
request = {key: arguments[key] for key in ('request', 'selection', 'services', 'policy_topic')}
|
|
return (answer_quote(question, conv_id, request=request) or answer_us_fallback(question, conv_id)
|
|
or {'t': '请提供目的国家、货物重量及目的仓码或邮编，继续核对当前资料。'})
|
|
if name == 'read_service_info':
|
|
from .customer_faq import answer_capability
|
|
from .knowledge import search_knowledge
|
|
answer = answer_capability(question)
|
|
if answer is not None:
|
|
return answer
|
|
# Only service/contact information is universal across the old regional knowledge base.
|
|
if any(word in question for word in ('人工', '客服', '工作时间', '营业时间', '联系方式')):
|
|
return {'t': search_knowledge('人工客服工作时间')}
|
|
return {'t': '可以咨询物流发货、报价、订单和轨迹。具体订单需登录查询，具体运费需按货物条件查询报价工具。'
|
|
'本次没有查得其他业务事实。可以回应问候、询问客户想办什么，或针对原问题澄清缺失条件。'}
|
|
raise ValueError('Unknown tool')
|