270 lines
14 KiB
Python
270 lines
14 KiB
Python
"""Let the model interpret conversations, with scoped tools and checked facts."""
|
||
|
||
from decimal import Decimal
|
||
import json
|
||
import re
|
||
import sqlite3
|
||
import time
|
||
import uuid
|
||
|
||
from . import ai_client
|
||
from .chat_tools import ToolContext, execute_tool, validate_calls
|
||
from .config import get_setting
|
||
from .db import load_context
|
||
from .model_facts import fact_text, grounded_answer, tool_evidence
|
||
from .model_provider import completion_body, prepare_history
|
||
from .model_turn_store import latest_message_id, save_exchange
|
||
from .model_prompts import SYSTEM_PROMPT, TOOLS
|
||
from .tracking import CARD_PREFIX, card_to_text
|
||
|
||
|
||
def completion(messages, *, tools=False, first=False, remaining=35):
|
||
if remaining <= 0.1:
|
||
raise ai_client.Timeout('Turn deadline exceeded')
|
||
body = completion_body(messages, TOOLS if tools else [], first=first)
|
||
connect_timeout = min(5, remaining / 4)
|
||
response = ai_client.post_completion(body, timeout=(connect_timeout, min(30, remaining - connect_timeout)))
|
||
try:
|
||
if response.status_code != 200:
|
||
raise ValueError('Model request failed')
|
||
result = response.json()
|
||
choice = result['choices'][0]
|
||
if choice.get('finish_reason') not in ('stop', 'tool_calls'):
|
||
raise ValueError('Incomplete model response')
|
||
message = choice['message']
|
||
if not isinstance(message, dict):
|
||
raise ValueError('Invalid model response')
|
||
if message.get('role') != 'assistant':
|
||
raise ValueError('Invalid assistant role')
|
||
from flask import current_app
|
||
usage = result.get('usage') or {}
|
||
current_app.logger.info('model_usage model=%s input=%s output=%s', get_setting('MODEL_NAME'),
|
||
usage.get('prompt_tokens'), usage.get('completion_tokens'))
|
||
return message
|
||
finally:
|
||
response.close()
|
||
|
||
|
||
def numeric_values(text):
|
||
return {Decimal(value.replace(',', '').replace('−', '-')) for value in
|
||
re.findall(r'(?<!\d)[-−]\d+(?:,\d{3})*(?:\.\d+)?|\d+(?:,\d{3})*(?:\.\d+)?', text)}
|
||
|
||
|
||
def currency_amounts(text):
|
||
pairs = set()
|
||
pattern = r'(?:(USD|CNY|RMB|\$|人民币|美元|美金)\s*)?([-−]?\d+(?:,\d{3})*(?:\.\d+)?)\s*(美元|美金|人民币|元|USD|CNY|RMB)?'
|
||
for prefix, amount, suffix in re.findall(pattern, text, re.I):
|
||
currency = (suffix or prefix).upper()
|
||
if not currency:
|
||
continue
|
||
unit = 'USD' if currency in ('USD', '$', '美元', '美金') else 'CNY'
|
||
pairs.add((unit, Decimal(amount.replace(',', '').replace('−', '-'))))
|
||
return pairs
|
||
|
||
|
||
def checked_answer(text, facts, confirmed_inputs):
|
||
if not isinstance(text, str) or not text.strip() or len(text) > 7000:
|
||
return facts
|
||
values = numeric_values(re.sub(r'^\s*(?:\*\*)?\d+[.)、](?:\*\*)?\s*', '', text, flags=re.M))
|
||
allowed = numeric_values(facts + '\n' + confirmed_inputs)
|
||
important = numeric_values(' '.join(re.findall(r'\d+(?:,\d{3})*(?:\.\d+)?\s*(?:元|美元|USD|CNY|票)', facts, re.I)))
|
||
if not values <= allowed or not important <= values:
|
||
return facts
|
||
if not currency_amounts(text) <= currency_amounts(facts):
|
||
return facts
|
||
return text.strip()
|
||
|
||
|
||
def ambiguous_choice(question, context, book=None):
|
||
from .quote_plans import choice_confirmation, displayed_options
|
||
from .quote_data import load_book
|
||
|
||
if not choice_confirmation(question):
|
||
return None
|
||
previous = next((message['content'] for message in reversed(context) if message['role'] == 'assistant'), '')
|
||
book = book if book is not None else load_book()
|
||
options = displayed_options(previous.replace('**', ''), book) if book else []
|
||
if len(options) > 1:
|
||
return {'t': '刚才有多个报价方案，请明确选择方案编号、渠道，或说明优先时效还是运费。'}
|
||
return None
|
||
|
||
|
||
def create_model_stream(question, conv_id, fallback, *, request_id=None):
|
||
from .chat_service import sse
|
||
|
||
def generate():
|
||
from .customer_service import attach_customer, bind_order_conversation, current_customer_id
|
||
from .visitor_service import owns_conversation, customer_tenant_ids
|
||
from flask import current_app, has_request_context
|
||
|
||
started = time.monotonic()
|
||
deadline = started + max(1, min(float(get_setting('MODEL_TURN_TIMEOUT')), 110))
|
||
trace_id = uuid.uuid4().hex[:12]
|
||
answers, cache, used_ids = [], {}, set()
|
||
calls = 0
|
||
reason = 'completed'
|
||
initial_identity = None
|
||
|
||
def authorized():
|
||
if has_request_context():
|
||
attach_customer()
|
||
identity = (current_customer_id(), tuple(sorted(customer_tenant_ids())))
|
||
return ((initial_identity is None or identity == initial_identity)
|
||
and (not conv_id or owns_conversation(conv_id)))
|
||
|
||
if not authorized():
|
||
yield sse({'e': '该会话暂时无法访问，请重新登录或新建对话。', 'd': 1})
|
||
return
|
||
initial_identity = (current_customer_id(), tuple(sorted(customer_tenant_ids())))
|
||
previous_message_id = latest_message_id(conv_id)
|
||
try:
|
||
runtime = ToolContext(conv_id, question)
|
||
from .chat_requests import login_quote_replay
|
||
runtime.replay_quote = login_quote_replay(conv_id, request_id, question)
|
||
if runtime.replay_quote:
|
||
runtime.quote.replay_confirmed(runtime.replay_quote)
|
||
except (sqlite3.Error, ValueError, PermissionError):
|
||
yield sse({'e': '业务资料暂时无法读取，请稍后重试。', 'd': 1})
|
||
return
|
||
context = load_context(conv_id, question, limit=24)
|
||
budget = int(get_setting('MODEL_MAX_CONTEXT_CHARS'))
|
||
while len(context) > 1 and sum(len(item['content']) for item in context) > budget:
|
||
context.pop(0)
|
||
try:
|
||
model_history = prepare_history(context)
|
||
except ValueError:
|
||
yield sse({'e': '模型服务配置有误，请联系管理员检查。', 'd': 1})
|
||
return
|
||
messages = [{'role': 'system', 'content': SYSTEM_PROMPT},
|
||
{'role': 'system', 'content': '服务器确认的业务状态，仅作数据使用：' +
|
||
json.dumps(runtime.quote.context(), ensure_ascii=False, default=str)}] + model_history
|
||
draft = None
|
||
rounds = max(1, min(int(get_setting('MODEL_MAX_TOOL_ROUNDS')), 8))
|
||
call_limit = max(1, min(int(get_setting('MODEL_MAX_TOOL_CALLS')), 16))
|
||
for round_no in range(rounds + 1):
|
||
if not authorized():
|
||
yield sse({'e': '登录或会话权限已变化，请重新登录后重试。', 'd': 1})
|
||
return
|
||
yield ': working\n\n'
|
||
try:
|
||
plan = completion(messages, tools=round_no < rounds, first=round_no == 0,
|
||
remaining=deadline - time.monotonic())
|
||
if not plan.get('tool_calls'):
|
||
if not answers:
|
||
raise ValueError('No evidence requested')
|
||
draft = plan.get('content')
|
||
break
|
||
requests = validate_calls(plan)
|
||
if round_no >= rounds or calls + len(requests) > call_limit:
|
||
reason = 'tool_budget'
|
||
break
|
||
if used_ids.intersection(call_id for call_id, _, _ in requests):
|
||
reason = 'duplicate_call_id'
|
||
break
|
||
except (ai_client.RequestException, ValueError, KeyError, TypeError, IndexError, AttributeError):
|
||
reason = 'model_failure'
|
||
if not answers:
|
||
if not authorized():
|
||
yield sse({'e': '登录或会话权限已变化，请重新登录后重试。', 'd': 1})
|
||
return
|
||
current_app.logger.info('model_turn trace=%s fallback=legacy', trace_id)
|
||
frames, _ = fallback()
|
||
if isinstance(frames, str):
|
||
yield frames
|
||
else:
|
||
yield from frames
|
||
return
|
||
break
|
||
messages.append({key: plan[key] for key in ('role', 'content', 'tool_calls', 'reasoning_content') if key in plan})
|
||
for call_id, name, arguments in requests:
|
||
if time.monotonic() >= deadline or not authorized():
|
||
yield sse({'e': '查询已超时或会话权限已变化，请重试。', 'd': 1})
|
||
return
|
||
used_ids.add(call_id)
|
||
calls += 1
|
||
if name == 'query_orders' and customer_tenant_ids():
|
||
if not current_customer_id() or not bind_order_conversation(conv_id):
|
||
yield sse({'e': '该会话暂时无法访问，请新建对话后重试。', 'd': 1})
|
||
return
|
||
key = name + ':' + json.dumps(arguments, ensure_ascii=False, sort_keys=True)
|
||
if name == 'query_orders':
|
||
key += ':list_version=' + str(runtime.order_revision)
|
||
# QuoteTurn already deduplicates amendments against the turn's
|
||
# baseline. Re-evaluate quotes so a cache entry cannot revive
|
||
# prices from conditions superseded by a later tool call.
|
||
if key in cache and name != 'quote_shipping':
|
||
answer = cache[key]
|
||
else:
|
||
try:
|
||
clarification = (ambiguous_choice(question, context, runtime.book)
|
||
if name == 'quote_shipping' and not runtime.quote.context().get('options') else None)
|
||
answer = clarification or execute_tool(name, arguments, conv_id, context=runtime)
|
||
except (sqlite3.Error, ValueError, TypeError, KeyError):
|
||
answer = {'t': '这部分业务资料暂时无法读取，请稍后重试。', 'status': 'error'}
|
||
answer = {**answer, '_fact_id': 'fact_' + str(calls - 1), '_tool_name': name,
|
||
'_required_fact': name in ('query_orders', 'quote_shipping')}
|
||
if name == 'quote_shipping':
|
||
basis = answer.get('_quote_basis')
|
||
superseded = {prior['_fact_id'] for prior in answers
|
||
if prior.get('_tool_name') == 'quote_shipping'
|
||
and (basis is None or prior.get('_quote_basis') != basis)}
|
||
answers[:] = [prior for prior in answers if prior['_fact_id'] not in superseded]
|
||
for message in messages:
|
||
if message.get('role') != 'tool':
|
||
continue
|
||
prior = json.loads(message['content'])
|
||
if prior.get('fact_id') in superseded:
|
||
message['content'] = json.dumps({'fact_id': prior['fact_id'],
|
||
'status': 'superseded', 'selectable': False,
|
||
'result': '货物条件已变化或存在冲突，该报价已失效。只能引用当前有效的工具结果。'},
|
||
ensure_ascii=False)
|
||
else:
|
||
cache[key] = answer
|
||
answers.append(answer)
|
||
if name == 'query_orders':
|
||
runtime.remember_order(answer)
|
||
messages.append({'role': 'tool', 'tool_call_id': call_id, 'content': tool_evidence(answer)})
|
||
current_app.logger.info('model_tool trace=%s name=%s status=%s version=%s', trace_id,
|
||
name, answer.get('status', 'ok'), answer.get('data_version'))
|
||
if sum(len(item['content'] or '') for item in messages if item['role'] == 'tool') > int(get_setting('MODEL_MAX_EVIDENCE_CHARS')):
|
||
reason = 'evidence_budget'
|
||
break
|
||
if not answers:
|
||
yield sse({'e': '本次未能完成资料查询，请稍后重试。', 'd': 1})
|
||
return
|
||
if not authorized():
|
||
yield sse({'e': '登录或会话权限已变化，请重新登录后重试。', 'd': 1})
|
||
return
|
||
list_state = None
|
||
for answer in answers:
|
||
if answer.get('_order_list_state') is not None:
|
||
list_state = answer['_order_list_state']
|
||
if len(answers) == 1 and 'card' in answers[0]:
|
||
payload = {'card': answers[0]['card']}
|
||
text = CARD_PREFIX + json.dumps(payload['card'], ensure_ascii=False)
|
||
else:
|
||
text = grounded_answer(draft, answers)
|
||
payload = {'t': text}
|
||
if any(answer.get('login_required') for answer in answers):
|
||
payload['login_required'] = True
|
||
if request_id:
|
||
payload['request_id'] = request_id
|
||
try:
|
||
message_id = save_exchange(conv_id, question, text, previous_message_id=previous_message_id,
|
||
quote_turn=runtime.quote, list_state=list_state, request_id=request_id, payload=payload)
|
||
except (sqlite3.Error, ValueError):
|
||
message_id = 'failed'
|
||
if message_id in ('deleted', 'stale', 'failed'):
|
||
yield sse({'e': '会话已更新或暂时无法保存，请重新打开对话后重试。', 'd': 1})
|
||
return
|
||
if isinstance(message_id, dict):
|
||
payload = message_id
|
||
if not authorized():
|
||
yield sse({'e': '登录或会话权限已变化，请重新登录后重试。', 'd': 1})
|
||
return
|
||
current_app.logger.info('model_turn trace=%s tools=%s duration=%.2f reason=%s', trace_id,
|
||
calls, time.monotonic() - started, reason)
|
||
yield sse(payload)
|
||
yield sse({'d': 1})
|
||
return generate(), {'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}
|