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 meta = _normalize_routing_meta(routing_meta) while True: payload = { "model": model, "messages": messages, "temperature": temperature, "max_tokens": current_max_tokens, } if meta.get("disable_thinking") is True and str(provider_name or "").strip().lower() == "dashscope": payload["enable_thinking"] = False if use_json_response_format: payload["response_format"] = {"type": "json_object"} if _dashscope_disable_thinking_for_strict_mode(provider=provider_name, model=model): payload["enable_thinking"] = False zhipu_disable_by_meta = ( meta.get("disable_thinking") is True and str(provider_name or "").strip().lower() == "zhipu" ) zhipu_disable_by_strict_mode = _zhipu_disable_thinking_for_strict_mode(provider=provider_name, model=model) and ( use_json_response_format or function_label.startswith("probe_capability_") or function_label == "test_engine_connection" ) if zhipu_disable_by_meta or zhipu_disable_by_strict_mode: payload["thinking"] = {"type": "disabled"} 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: if structured_output and self._should_retry_truncated_structured_response( function_name=function_label, response_text=text, routing_meta=routing_meta, choice=first_choice, current_max_tokens=current_max_tokens, ): next_max_tokens = self._next_retry_max_tokens_for_truncated_structured_response( current_max_tokens=current_max_tokens, function_name=function_label, routing_meta=routing_meta, ) if next_max_tokens > current_max_tokens: current_max_tokens = next_max_tokens _runtime_log( "LLM", ( "结构化输出截断扩容重试：" f"function={function_label}, next_max_tokens={current_max_tokens}" ), ) continue 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, ) async def _chat_completion_stream_via_openai_compatible( self, *, key: str, base_url: str, model: str, temperature: float, max_tokens: int, messages: list[dict[str, str]], request_timeout: float, function_label: str, provider_name: str, is_control_plane_gateway_request: bool, control_plane_agents_timeout_seconds: int, request_trace_id: str, routing_meta: dict[str, Any] | None = None, audit_context: dict[str, Any] | None = None, ) -> AsyncIterator[str]: endpoint = f"{base_url.rstrip('/')}/chat/completions" headers = { "Authorization": f"Bearer {key}", "Content-Type": "application/json", } payload: dict[str, Any] = { "model": model, "messages": messages, "temperature": temperature, "max_tokens": max(int(max_tokens or 0), 1), "stream": True, } meta = _normalize_routing_meta(routing_meta) if meta.get("disable_thinking") is True and str(provider_name or "").strip().lower() == "dashscope": payload["enable_thinking"] = False if ( meta.get("disable_thinking") is True and str(provider_name or "").strip().lower() == "zhipu" ): payload["thinking"] = {"type": "disabled"} if is_control_plane_gateway_request: payload["agents"] = { "defaults": { "timeoutSeconds": int(control_plane_agents_timeout_seconds), } } chunks_seen = 0 response_samples: list[Any] = [] provider_attempts: list[dict[str, Any]] = [] if audit_context is not None: audit_context["provider_request_payload"] = self._provider_audit_copy(payload) try: async with httpx.AsyncClient(timeout=request_timeout, trust_env=False) as client: async with client.stream("POST", endpoint, headers=headers, json=payload) as resp: if resp.status_code >= 400: body = (await resp.aread()).decode("utf-8", errors="replace") response_payload = {"raw_text": body, "status_code": int(resp.status_code)} provider_attempts.append( { "provider": provider_name, "model": model, "endpoint": endpoint, "request_payload": self._provider_audit_copy(payload), "response_payload": response_payload, "status": "http_error", "status_code": int(resp.status_code), } ) exc = httpx.HTTPStatusError( f"LLM stream HTTP {resp.status_code}: {body[:300]}", request=resp.request, response=resp, ) raise self._attach_provider_audit_to_exception( exc, request_payload=payload, response_payload=response_payload, attempts=provider_attempts, ) async for line in resp.aiter_lines(): text_line = str(line or "").strip() if not text_line: continue if text_line.startswith("data:"): text_line = text_line[5:].strip() if not text_line or text_line == "[DONE]": if text_line == "[DONE]": break continue try: data = json.loads(text_line) except Exception: continue if len(response_samples) < 12: response_samples.append(data) chunk = self._extract_stream_delta_text(data) if chunk: chunks_seen += 1 yield chunk except Exception as exc: if not getattr(exc, "provider_attempts", None): provider_attempts.append( { "provider": provider_name, "model": model, "endpoint": endpoint, "request_payload": self._provider_audit_copy(payload), "response_payload": {"chunks_seen": chunks_seen, "samples": response_samples}, "status": "exception", "error_type": type(exc).__name__, "error": str(exc)[:500], } ) raise self._attach_provider_audit_to_exception( exc, request_payload=payload, response_payload={"chunks_seen": chunks_seen, "samples": response_samples}, attempts=provider_attempts, ) raise finally: response_payload = {"chunks_seen": chunks_seen, "samples": response_samples} if audit_context is not None: audit_context["provider_response_payload"] = response_payload audit_context["provider_attempts"] = provider_attempts or [ { "provider": provider_name, "model": model, "endpoint": endpoint, "request_payload": self._provider_audit_copy(payload), "response_payload": response_payload, "status": "success" if chunks_seen else "missing_stream_chunks", "trace_id": request_trace_id, } ] @staticmethod def _extract_stream_delta_text(data: Any) -> str: if not isinstance(data, dict): return "" choices = data.get("choices") if isinstance(data.get("choices"), list) else [] if not choices: return "" first_choice = choices[0] if isinstance(choices[0], dict) else {} delta = first_choice.get("delta") if isinstance(first_choice.get("delta"), dict) else {} message = first_choice.get("message") if isinstance(first_choice.get("message"), dict) else {} for container in (delta, message, first_choice): value = container.get("content") if isinstance(container, dict) else None if isinstance(value, str) and value: return value return ""