221 lines
12 KiB
Python
221 lines
12 KiB
Python
"""Execute whitelisted model requests under the current server identity."""
|
|
|
|
import json
|
|
import re
|
|
from copy import deepcopy
|
|
|
|
from .customer_service import order_login_response
|
|
from .visitor_service import customer_tenant_ids
|
|
|
|
|
|
def validate_value(value, schema, depth=0):
|
|
"""Validate the supported schema subset before any business tool executes."""
|
|
if depth > 6:
|
|
raise ValueError('Tool argument nesting is too deep')
|
|
if 'anyOf' in schema:
|
|
for candidate in schema['anyOf']:
|
|
try:
|
|
validate_value(value, candidate, depth + 1)
|
|
return
|
|
except ValueError:
|
|
continue
|
|
raise ValueError('Invalid union value')
|
|
if 'enum' in schema and value not in schema['enum']:
|
|
raise ValueError('Invalid tool choice')
|
|
kind = schema.get('type')
|
|
kinds = kind if isinstance(kind, list) else [kind]
|
|
matches = {'object': type(value) is dict, 'array': type(value) is list,
|
|
'string': isinstance(value, str), 'integer': type(value) is int,
|
|
'number': type(value) in (int, float), 'boolean': type(value) is bool,
|
|
'null': value is None}
|
|
if kind and not any(matches.get(item, False) for item in kinds):
|
|
raise ValueError('Invalid tool value type')
|
|
if isinstance(value, dict):
|
|
properties = schema.get('properties', {})
|
|
if not set(schema.get('required', ())) <= value.keys():
|
|
raise ValueError('Missing tool arguments')
|
|
if schema.get('additionalProperties') is False and value.keys() - properties.keys():
|
|
raise ValueError('Unknown tool arguments')
|
|
for key, item in value.items():
|
|
if key in properties:
|
|
validate_value(item, properties[key], depth + 1)
|
|
elif isinstance(value, list):
|
|
if not schema.get('minItems', 0) <= len(value) <= schema.get('maxItems', 100):
|
|
raise ValueError('Too many tool values')
|
|
for item in value:
|
|
validate_value(item, schema.get('items', {}), depth + 1)
|
|
elif isinstance(value, str):
|
|
if not schema.get('minLength', 0) <= len(value) <= schema.get('maxLength', 2000):
|
|
raise ValueError('Invalid tool string length')
|
|
if schema.get('pattern') and not re.fullmatch(schema['pattern'], value):
|
|
raise ValueError('Invalid tool string')
|
|
elif type(value) in (int, float):
|
|
import math
|
|
if (type(value) is float and not math.isfinite(value)
|
|
or value < schema.get('minimum', -1e12) or value > schema.get('maximum', 1e12)):
|
|
raise ValueError('Invalid tool number')
|
|
|
|
|
|
def validate_calls(message):
|
|
from .model_prompts import TOOLS
|
|
|
|
calls = message.get('tool_calls')
|
|
if not isinstance(calls, list) or not 1 <= len(calls) <= 4:
|
|
raise ValueError('Invalid tool count')
|
|
schemas = {tool['function']['name']: tool['function']['parameters'] 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 schemas:
|
|
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')
|
|
raw = function.get('arguments', '')
|
|
if not isinstance(raw, str) or len(raw) > 16000:
|
|
raise ValueError('Invalid tool arguments')
|
|
arguments = json.loads(raw)
|
|
name = function['name']
|
|
validate_value(arguments, schemas[name])
|
|
question = arguments['question']
|
|
if not isinstance(question, str) or not 0 < len(question.strip()) <= 2000:
|
|
raise ValueError('Invalid tool question')
|
|
for key, value in arguments.items():
|
|
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 name == 'query_orders' and 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, *, context=None):
|
|
question = arguments['question']
|
|
if name == 'query_orders':
|
|
if not customer_tenant_ids():
|
|
return order_login_response()
|
|
from .order_data import answer_query
|
|
state = context.order_list_state if context is not None else None
|
|
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}
|
|
answer = {**answer_summary(query, customer_tenant_ids(), conv_id, state=state), '_customer_data': True}
|
|
if context is not None:
|
|
context.remember_order(answer)
|
|
return answer
|
|
label = ('异常' if condition == 'exception' else '')
|
|
label += STAGE_LABELS.get(arguments['stage'], '') + STATUS_LABELS.get(arguments['status'], '')
|
|
question = '列出我的' + label + '订单'
|
|
answer = answer_query(question, conv_id, list_state=state)
|
|
if answer is None:
|
|
answer = {'t': '请提供入仓码，或说明要查询的订单统计条件。'}
|
|
answer = {**answer, '_customer_data': True}
|
|
if context is not None:
|
|
context.remember_order(answer)
|
|
return answer
|
|
if name == 'quote_shipping':
|
|
if context is not None:
|
|
if getattr(context, 'replay_quote', None):
|
|
arguments = {key: value for key, value in arguments.items()
|
|
if key not in ('shipment', 'changes', 'new_shipment', 'option_id')}
|
|
arguments['shipment'] = {}
|
|
answer = context.quote.run(arguments)
|
|
context.quote.replay_state = context.quote.candidate
|
|
context.quote.candidate = None
|
|
return answer
|
|
return context.quote.run(arguments)
|
|
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 in ('search_knowledge', 'read_knowledge'):
|
|
if context is None:
|
|
raise ValueError('Knowledge tools need a pinned turn context')
|
|
from .knowledge_service import search_business_knowledge, read_business_knowledge
|
|
if name == 'search_knowledge':
|
|
country = arguments['country']
|
|
services = arguments['services']
|
|
result = search_business_knowledge(question, country=country, services=services,
|
|
topic=arguments['topic'], book=context.book)
|
|
resolved_country = result.get('data', {}).get('resolved_country', country)
|
|
for match in result.get('data', {}).get('matches', []):
|
|
# An entry can cover several regions (EU,GB) or be global.
|
|
# Preserve the actual query scope, not the entry's applicability
|
|
# label, so read rechecks exactly the country search authorized.
|
|
context.references[match['reference']] = (resolved_country, services)
|
|
return result
|
|
scope = context.references.get(arguments['reference'])
|
|
if scope is None:
|
|
return {'t': '请先检索适用资料，再读取本轮返回的条款引用。',
|
|
'status': 'not_found', 'data': {}, 'sources': []}
|
|
return read_business_knowledge(arguments['reference'], country=scope[0], services=scope[1], book=context.book)
|
|
if name == 'read_service_info':
|
|
if arguments.get('clarification') == 'quote_option':
|
|
return {'t': '请明确要选择的报价方案编号或渠道，或说明优先时效还是运费，我再继续核对。',
|
|
'status': 'needs_clarification'}
|
|
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')
|
|
|
|
|
|
class ToolContext:
|
|
"""Keep a quote book, candidate shipment and allowed document references per turn."""
|
|
|
|
def __init__(self, conv_id, question):
|
|
from .quote_data import load_book
|
|
from .quote_tool import QuoteTurn
|
|
self.book = load_book()
|
|
self.quote = QuoteTurn(conv_id, question, self.book)
|
|
self.references = {}
|
|
from .order_listing import load_state
|
|
self.order_list_state = load_state(conv_id)
|
|
self.order_revision = 0
|
|
|
|
def remember_order(self, answer):
|
|
"""Keep the displayed list and selection local until the final atomic save."""
|
|
candidate = answer.get('_order_list_state')
|
|
if candidate is None and 'card' in answer and self.order_list_state is not None:
|
|
from .order_followups import selected_row
|
|
from .tracking import CARD_PREFIX
|
|
candidate = deepcopy(self.order_list_state)
|
|
shown = [(item['inbound_code'] or '未登记', item['order_no']) for item in candidate.get('shown', [])]
|
|
index = selected_row(CARD_PREFIX + json.dumps(answer['card'], ensure_ascii=False), shown)
|
|
candidate.update(selected=index is not None, selected_index=index)
|
|
answer['_order_list_state'] = candidate
|
|
if candidate is None:
|
|
return
|
|
candidate = deepcopy(candidate)
|
|
candidate.setdefault('selected', False)
|
|
candidate.setdefault('selected_index', None)
|
|
if candidate != self.order_list_state:
|
|
self.order_revision += 1
|
|
self.order_list_state = candidate
|