444 lines
20 KiB
Python
444 lines
20 KiB
Python
from __future__ import annotations
|
||
|
||
from .shared import *
|
||
|
||
|
||
class LLMEmptyContentError(ValueError):
|
||
"""Raised when the model still returns empty content after one forced retry."""
|
||
|
||
|
||
class LLMChatOpenAIMixin:
|
||
@staticmethod
|
||
def _attach_provider_audit_to_exception(
|
||
exc: Exception,
|
||
*,
|
||
request_payload: Any = None,
|
||
response_payload: Any = None,
|
||
attempts: list[dict[str, Any]] | None = None,
|
||
) -> Exception:
|
||
setattr(exc, "provider_request_payload", request_payload)
|
||
setattr(exc, "provider_response_payload", response_payload)
|
||
setattr(exc, "provider_attempts", attempts or [])
|
||
return exc
|
||
|
||
@staticmethod
|
||
def _provider_audit_copy(value: Any) -> Any:
|
||
try:
|
||
return json.loads(json.dumps(value, ensure_ascii=False, default=str))
|
||
except Exception:
|
||
return value
|
||
|
||
@staticmethod
|
||
def _is_provider_billing_blocked_http_error(*, status_code: int, body_preview: str) -> bool:
|
||
if int(status_code or 0) not in {400, 401, 402, 403}:
|
||
return False
|
||
lowered = str(body_preview or "").strip().lower()
|
||
if not lowered:
|
||
return False
|
||
return (
|
||
"arrearage" in lowered
|
||
or "overdue-payment" in lowered
|
||
or ("access denied" in lowered and "good standing" in lowered)
|
||
or "provider_billing_blocked" in lowered
|
||
)
|
||
|
||
@staticmethod
|
||
def _append_direct_answer_retry_instruction(
|
||
messages: list[dict[str, str]],
|
||
*,
|
||
structured_output: bool = False,
|
||
) -> None:
|
||
retry_instruction = (
|
||
"请直接输出最终 JSON 对象，不要解释，不要 Markdown，不要把正文放在 reasoning_content。"
|
||
if structured_output
|
||
else "请直接输出答案"
|
||
)
|
||
for idx in range(len(messages) - 1, -1, -1):
|
||
row = messages[idx]
|
||
if not isinstance(row, dict):
|
||
continue
|
||
role = str(row.get("role") or "").strip().lower()
|
||
if role != "user":
|
||
continue
|
||
original = str(row.get("content") or "").strip()
|
||
if retry_instruction in original:
|
||
return
|
||
patched = dict(row)
|
||
patched["content"] = f"{original}\n\n{retry_instruction}" if original else retry_instruction
|
||
messages[idx] = patched
|
||
return
|
||
messages.append({"role": "user", "content": retry_instruction})
|
||
|
||
async def _chat_completion_via_openai_compatible(
|
||
self,
|
||
*,
|
||
key: str,
|
||
base_url: str,
|
||
model: str,
|
||
temperature: float,
|
||
max_tokens: int,
|
||
messages: list[dict[str, str]],
|
||
request_timeout: float,
|
||
structured_output: bool,
|
||
function_label: str,
|
||
provider_name: str,
|
||
is_control_plane_gateway_request: bool,
|
||
control_plane_agents_timeout_seconds: int,
|
||
request_started_at_perf: float,
|
||
request_trace_id: str,
|
||
control_plane_agent_timeout: Optional[int],
|
||
routing_meta: dict[str, Any] | None = None,
|
||
) -> ChatCompletionResult:
|
||
endpoint = f"{base_url.rstrip('/')}/chat/completions"
|
||
headers = {
|
||
"Authorization": f"Bearer {key}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
current_max_tokens = max(int(max_tokens or 0), 1)
|
||
http_status_retry_budget = self._http_status_retry_budget_for_function(function_label)
|
||
http_status_retry_attempt = 0
|
||
network_retry_budget = 1
|
||
network_retry_attempt = 0
|
||
explicit_retry_budget = self._max_retries_from_routing_meta(routing_meta)
|
||
transient_retries_left = explicit_retry_budget
|
||
provider_attempts: list[dict[str, Any]] = []
|
||
empty_chat_retry_budget = self._empty_chat_retry_budget(
|
||
function_name=function_label,
|
||
current_max_tokens=current_max_tokens,
|
||
)
|
||
use_json_response_format = bool(structured_output)
|
||
direct_answer_retry_used = False
|
||
while True:
|
||
payload = {
|
||
"model": model,
|
||
"messages": messages,
|
||
"temperature": temperature,
|
||
"max_tokens": current_max_tokens,
|
||
}
|
||
if use_json_response_format:
|
||
payload["response_format"] = {"type": "json_object"}
|
||
if is_control_plane_gateway_request:
|
||
payload["agents"] = {
|
||
"defaults": {
|
||
"timeoutSeconds": int(control_plane_agents_timeout_seconds),
|
||
}
|
||
}
|
||
attempt_record: dict[str, Any] = {
|
||
"provider": provider_name,
|
||
"model": model,
|
||
"endpoint": endpoint,
|
||
"request_payload": self._provider_audit_copy(payload),
|
||
}
|
||
try:
|
||
async with httpx.AsyncClient(timeout=request_timeout, trust_env=False) as client:
|
||
resp = await client.post(endpoint, headers=headers, json=payload)
|
||
except httpx.RequestError as exc:
|
||
attempt_record.update(
|
||
{
|
||
"status": "request_error",
|
||
"error_type": type(exc).__name__,
|
||
"error": str(exc)[:2000],
|
||
}
|
||
)
|
||
provider_attempts.append(attempt_record)
|
||
can_retry_network_error = (
|
||
transient_retries_left > 0
|
||
if transient_retries_left is not None
|
||
else network_retry_budget > 0
|
||
)
|
||
if can_retry_network_error:
|
||
retry_delay = min(0.8 * (2 ** network_retry_attempt), 3.0)
|
||
retries_left_after = (
|
||
max(int(transient_retries_left or 0) - 1, 0)
|
||
if transient_retries_left is not None
|
||
else network_retry_budget - 1
|
||
)
|
||
_runtime_log(
|
||
"LLM",
|
||
(
|
||
"网络异常重试："
|
||
f"function={function_label}, "
|
||
f"retry_in={retry_delay:.2f}s, left={retries_left_after}, "
|
||
f"error={type(exc).__name__}"
|
||
),
|
||
)
|
||
logger.warning(
|
||
"llm_network_retry trace_id=%s function_name=%s retry_attempt=%s retry_delay_seconds=%.2f error_type=%s error=%s",
|
||
request_trace_id,
|
||
function_label,
|
||
network_retry_attempt,
|
||
retry_delay,
|
||
type(exc).__name__,
|
||
str(exc)[:260],
|
||
)
|
||
if transient_retries_left is not None:
|
||
transient_retries_left -= 1
|
||
else:
|
||
network_retry_budget -= 1
|
||
network_retry_attempt += 1
|
||
await asyncio.sleep(retry_delay)
|
||
continue
|
||
raise self._attach_provider_audit_to_exception(
|
||
exc,
|
||
request_payload=payload,
|
||
response_payload=None,
|
||
attempts=provider_attempts,
|
||
)
|
||
else:
|
||
if is_control_plane_gateway_request:
|
||
elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0
|
||
_runtime_log(
|
||
"PMFlow 本地路由层调用",
|
||
(
|
||
f"http ts={_cn_now_text_ms()} trace_id={request_trace_id} "
|
||
f"function={function_label} status={resp.status_code} elapsed_ms={elapsed_ms:.0f}"
|
||
),
|
||
)
|
||
if resp.status_code >= 400:
|
||
body_preview = str(resp.text or "").replace("\n", " ").strip()[:260]
|
||
response_payload: Any
|
||
try:
|
||
response_payload = resp.json()
|
||
except Exception:
|
||
response_payload = {"raw_text": str(resp.text or "")}
|
||
attempt_record.update(
|
||
{
|
||
"status": "http_error",
|
||
"status_code": int(resp.status_code),
|
||
"response_payload": response_payload,
|
||
}
|
||
)
|
||
provider_attempts.append(attempt_record)
|
||
if is_control_plane_gateway_request:
|
||
_runtime_log(
|
||
"PMFlow 本地路由层调用",
|
||
(
|
||
f"http_error ts={_cn_now_text_ms()} trace_id={request_trace_id} "
|
||
f"function={function_label} status={resp.status_code} body={body_preview}"
|
||
),
|
||
)
|
||
status_code = int(resp.status_code)
|
||
if self._is_provider_billing_blocked_http_error(
|
||
status_code=status_code,
|
||
body_preview=body_preview,
|
||
):
|
||
exc = ValueError(f"provider_billing_blocked:{provider_name}:{body_preview}")
|
||
raise self._attach_provider_audit_to_exception(
|
||
exc,
|
||
request_payload=payload,
|
||
response_payload=response_payload,
|
||
attempts=provider_attempts,
|
||
)
|
||
can_retry_http_status = self._is_retryable_http_status(status_code) and (
|
||
transient_retries_left > 0
|
||
if transient_retries_left is not None
|
||
else http_status_retry_budget > 0
|
||
)
|
||
if can_retry_http_status:
|
||
retry_delay = self._http_retry_delay_seconds(
|
||
status_code=status_code,
|
||
attempt=http_status_retry_attempt,
|
||
retry_after_header=resp.headers.get("Retry-After"),
|
||
)
|
||
retries_left_after = (
|
||
max(int(transient_retries_left or 0) - 1, 0)
|
||
if transient_retries_left is not None
|
||
else http_status_retry_budget - 1
|
||
)
|
||
_runtime_log(
|
||
"LLM",
|
||
(
|
||
"HTTP状态重试："
|
||
f"function={function_label}, "
|
||
f"status={status_code}, retry_in={retry_delay:.2f}s, "
|
||
f"left={retries_left_after}"
|
||
),
|
||
)
|
||
if transient_retries_left is not None:
|
||
transient_retries_left -= 1
|
||
else:
|
||
http_status_retry_budget -= 1
|
||
http_status_retry_attempt += 1
|
||
await asyncio.sleep(retry_delay)
|
||
continue
|
||
if use_json_response_format and status_code in {400, 422}:
|
||
lower_preview = body_preview.lower()
|
||
if "response_format" in lower_preview or "json_object" in lower_preview or "json schema" in lower_preview:
|
||
_runtime_log(
|
||
"LLM",
|
||
(
|
||
"结构化参数降级重试："
|
||
f"function={function_label}, status={status_code}"
|
||
),
|
||
)
|
||
use_json_response_format = False
|
||
continue
|
||
try:
|
||
resp.raise_for_status()
|
||
except Exception as exc:
|
||
raise self._attach_provider_audit_to_exception(
|
||
exc,
|
||
request_payload=payload,
|
||
response_payload=response_payload,
|
||
attempts=provider_attempts,
|
||
)
|
||
try:
|
||
data = resp.json()
|
||
except Exception as exc:
|
||
attempt_record.update(
|
||
{
|
||
"status": "invalid_json_response",
|
||
"status_code": int(resp.status_code),
|
||
"response_payload": {"raw_text": str(resp.text or "")},
|
||
}
|
||
)
|
||
provider_attempts.append(attempt_record)
|
||
raise self._attach_provider_audit_to_exception(
|
||
exc,
|
||
request_payload=payload,
|
||
response_payload={"raw_text": str(resp.text or "")},
|
||
attempts=provider_attempts,
|
||
)
|
||
attempt_record.update(
|
||
{
|
||
"status": "success",
|
||
"status_code": int(resp.status_code),
|
||
"response_payload": data,
|
||
}
|
||
)
|
||
provider_attempts.append(attempt_record)
|
||
choices = data.get("choices") or []
|
||
if not choices:
|
||
exc = ValueError("LLM 响应无 choices")
|
||
raise self._attach_provider_audit_to_exception(
|
||
exc,
|
||
request_payload=payload,
|
||
response_payload=data,
|
||
attempts=provider_attempts,
|
||
)
|
||
first_choice = choices[0] if isinstance(choices[0], dict) else {}
|
||
message = first_choice.get("message") if isinstance(first_choice, dict) else {}
|
||
text = self._extract_text_from_chat_response(
|
||
message=message,
|
||
choice=first_choice,
|
||
response=data,
|
||
structured_output=structured_output,
|
||
)
|
||
if not text and structured_output:
|
||
reasoning_text = self._extract_reasoning_fallback_text(
|
||
message=message,
|
||
choice=first_choice,
|
||
response=data,
|
||
)
|
||
normalized_reasoning = self._strip_code_fence(reasoning_text)
|
||
if self._contains_json_container(normalized_reasoning):
|
||
candidate = self._extract_json_candidate(normalized_reasoning)
|
||
if candidate:
|
||
if self._call_is_invalid_structured_output(
|
||
function_name=function_label,
|
||
response_text=candidate,
|
||
routing_meta=routing_meta,
|
||
):
|
||
_runtime_log(
|
||
"LLM",
|
||
(
|
||
"结构化reasoning_content候选无效，继续空content重试："
|
||
f"function={function_label}, provider={provider_name}"
|
||
),
|
||
)
|
||
else:
|
||
_runtime_log(
|
||
"LLM",
|
||
(
|
||
"结构化reasoning_content回退："
|
||
f"function={function_label}, provider={provider_name}"
|
||
),
|
||
)
|
||
text = candidate
|
||
if text:
|
||
break
|
||
|
||
msg_keys = list(message.keys()) if isinstance(message, dict) else []
|
||
choice_keys = list(first_choice.keys()) if isinstance(first_choice, dict) else []
|
||
choice_preview = ""
|
||
try:
|
||
choice_preview = json.dumps(first_choice, ensure_ascii=False)[:300]
|
||
except Exception:
|
||
choice_preview = str(first_choice)[:300]
|
||
empty_error = LLMEmptyContentError(
|
||
"LLM 返回空内容: "
|
||
f"message_keys={msg_keys}, choice_keys={choice_keys}, choice_preview={choice_preview}"
|
||
)
|
||
if not direct_answer_retry_used:
|
||
direct_answer_retry_used = True
|
||
self._append_direct_answer_retry_instruction(
|
||
messages,
|
||
structured_output=structured_output,
|
||
)
|
||
_runtime_log(
|
||
"LLM",
|
||
(
|
||
"空content定向重试："
|
||
f"function={function_label}, retry_with=请直接输出答案"
|
||
),
|
||
)
|
||
continue
|
||
if empty_chat_retry_budget > 0 and self._should_retry_empty_chat_response(
|
||
structured_output=structured_output,
|
||
message=message,
|
||
choice=first_choice,
|
||
current_max_tokens=current_max_tokens,
|
||
):
|
||
next_max_tokens = self._next_retry_max_tokens_for_empty_chat(current_max_tokens)
|
||
if next_max_tokens > current_max_tokens:
|
||
empty_chat_retry_budget -= 1
|
||
current_max_tokens = next_max_tokens
|
||
_runtime_log(
|
||
"LLM",
|
||
(
|
||
"空content扩容重试："
|
||
f"function={function_label}, next_max_tokens={current_max_tokens}, "
|
||
f"retries_left={empty_chat_retry_budget}"
|
||
),
|
||
)
|
||
continue
|
||
raise self._attach_provider_audit_to_exception(
|
||
empty_error,
|
||
request_payload=payload,
|
||
response_payload=data,
|
||
attempts=provider_attempts,
|
||
)
|
||
if _is_control_plane_timeout_message(text):
|
||
elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0
|
||
_runtime_log(
|
||
"PMFlow 本地路由层超时" if is_control_plane_gateway_request else "LLM超时文本",
|
||
(
|
||
f"detected ts={_cn_now_text_ms()} trace_id={request_trace_id} "
|
||
f"function={function_label} elapsed_ms={elapsed_ms:.0f} "
|
||
f"text={text.replace(chr(10), ' ')[:260]}"
|
||
),
|
||
)
|
||
logger.warning(
|
||
"llm_timeout_payload_detected",
|
||
trace_id=request_trace_id,
|
||
function_name=function_label,
|
||
request_timeout_seconds=float(request_timeout),
|
||
agent_timeout_seconds=control_plane_agent_timeout,
|
||
text_preview=text.replace("\n", " ")[:260],
|
||
)
|
||
if is_control_plane_gateway_request:
|
||
elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0
|
||
_runtime_log(
|
||
"PMFlow 本地路由层调用",
|
||
(
|
||
f"end ts={_cn_now_text_ms()} trace_id={request_trace_id} "
|
||
f"function={function_label} elapsed_ms={elapsed_ms:.0f} text_len={len(text)}"
|
||
),
|
||
)
|
||
return ChatCompletionResult(
|
||
text=text,
|
||
raw_request=payload,
|
||
raw_response=data,
|
||
provider_attempts=provider_attempts,
|
||
)
|