355 lines
14 KiB
Python
355 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
from .shared import *
|
|
|
|
class LLMChatTransportMixin:
|
|
async def _chat_completion_stream(
|
|
self,
|
|
*,
|
|
provider: str,
|
|
prompt: str,
|
|
system_prompt: str,
|
|
api_key: str,
|
|
base_url: str,
|
|
model: str,
|
|
temperature: float,
|
|
max_tokens: int,
|
|
function_name: str = "",
|
|
response_mode: str = "complete",
|
|
routing_meta: dict[str, Any] | None = None,
|
|
audit_context: dict[str, Any] | None = None,
|
|
) -> AsyncIterator[str]:
|
|
provider_name = _normalize_provider(provider)
|
|
normalized_routing_meta = _normalize_routing_meta(routing_meta)
|
|
structured_output = self._structured_output_required(function_name, routing_meta=normalized_routing_meta)
|
|
if structured_output:
|
|
raise ValueError("streaming_structured_output_not_supported")
|
|
request_timeout = self._request_timeout_for_function(
|
|
function_name,
|
|
response_mode=response_mode,
|
|
routing_meta=normalized_routing_meta,
|
|
)
|
|
is_control_plane_gateway_request = self._is_control_plane_gateway_request(
|
|
provider_name=provider_name,
|
|
base_url=base_url,
|
|
)
|
|
control_plane_agents_timeout_seconds = self._control_plane_agents_timeout_seconds()
|
|
if is_control_plane_gateway_request:
|
|
request_timeout = max(float(request_timeout), float(control_plane_agents_timeout_seconds) + 5.0)
|
|
key = api_key.strip()
|
|
messages: list[dict[str, str]] = []
|
|
cleaned_system_prompt = self._inject_function_guardrails(
|
|
system_prompt=system_prompt,
|
|
function_name=function_name,
|
|
)
|
|
if cleaned_system_prompt:
|
|
messages.append({"role": "system", "content": cleaned_system_prompt})
|
|
messages.append({"role": "user", "content": prompt})
|
|
request_trace_id = uuid4().hex[:12]
|
|
function_label = str(function_name or "llm_generate")
|
|
if provider_name == "ollama":
|
|
async for chunk in self._chat_completion_stream_via_ollama(
|
|
base_url=base_url,
|
|
model=model,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
messages=messages,
|
|
request_timeout=request_timeout,
|
|
audit_context=audit_context,
|
|
):
|
|
yield chunk
|
|
return
|
|
if not key:
|
|
raise ValueError("未配置 LLM API Key")
|
|
async for chunk in self._chat_completion_stream_via_openai_compatible(
|
|
key=key,
|
|
base_url=base_url,
|
|
model=model,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
messages=messages,
|
|
request_timeout=request_timeout,
|
|
function_label=function_label,
|
|
provider_name=provider_name,
|
|
is_control_plane_gateway_request=is_control_plane_gateway_request,
|
|
control_plane_agents_timeout_seconds=control_plane_agents_timeout_seconds,
|
|
request_trace_id=request_trace_id,
|
|
routing_meta=normalized_routing_meta,
|
|
audit_context=audit_context,
|
|
):
|
|
yield chunk
|
|
|
|
async def _chat_completion(
|
|
self,
|
|
*,
|
|
provider: str,
|
|
prompt: str,
|
|
system_prompt: str,
|
|
api_key: str,
|
|
base_url: str,
|
|
model: str,
|
|
temperature: float,
|
|
max_tokens: int,
|
|
function_name: str = "",
|
|
response_mode: str = "complete",
|
|
routing_meta: dict[str, Any] | None = None,
|
|
) -> str:
|
|
result = await self._chat_completion_raw(
|
|
provider=provider,
|
|
prompt=prompt,
|
|
system_prompt=system_prompt,
|
|
api_key=api_key,
|
|
base_url=base_url,
|
|
model=model,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
function_name=function_name,
|
|
response_mode=response_mode,
|
|
routing_meta=routing_meta,
|
|
)
|
|
return result.text
|
|
|
|
async def _chat_completion_raw(
|
|
self,
|
|
*,
|
|
provider: str,
|
|
prompt: str,
|
|
system_prompt: str,
|
|
api_key: str,
|
|
base_url: str,
|
|
model: str,
|
|
temperature: float,
|
|
max_tokens: int,
|
|
function_name: str = "",
|
|
response_mode: str = "complete",
|
|
routing_meta: dict[str, Any] | None = None,
|
|
) -> ChatCompletionResult:
|
|
provider_name = _normalize_provider(provider)
|
|
normalized_routing_meta = _normalize_routing_meta(routing_meta)
|
|
structured_output = self._structured_output_required(function_name, routing_meta=normalized_routing_meta)
|
|
request_timeout = self._request_timeout_for_function(
|
|
function_name,
|
|
response_mode=response_mode,
|
|
routing_meta=normalized_routing_meta,
|
|
)
|
|
is_control_plane_gateway_request = self._is_control_plane_gateway_request(
|
|
provider_name=provider_name,
|
|
base_url=base_url,
|
|
)
|
|
control_plane_agents_timeout_seconds = self._control_plane_agents_timeout_seconds()
|
|
if is_control_plane_gateway_request:
|
|
# 网关内部 agent 超时需小于客户端超时，避免本地请求先被中止。
|
|
request_timeout = max(float(request_timeout), float(control_plane_agents_timeout_seconds) + 5.0)
|
|
use_provider_circuit_breaker = self._use_provider_circuit_breaker(
|
|
function_name,
|
|
routing_meta=normalized_routing_meta,
|
|
)
|
|
circuit_key = self._provider_circuit_key(
|
|
provider=provider_name,
|
|
base_url=base_url,
|
|
model=model,
|
|
)
|
|
if use_provider_circuit_breaker and self._is_provider_circuit_open(circuit_key):
|
|
raise ValueError(f"provider_circuit_open:{provider_name}")
|
|
key = api_key.strip()
|
|
messages: list[dict[str, str]] = []
|
|
cleaned_system_prompt = self._inject_function_guardrails(
|
|
system_prompt=system_prompt,
|
|
function_name=function_name,
|
|
)
|
|
if cleaned_system_prompt:
|
|
messages.append({"role": "system", "content": cleaned_system_prompt})
|
|
messages.append({"role": "user", "content": prompt})
|
|
request_trace_id = uuid4().hex[:12]
|
|
request_started_at_perf = time.perf_counter()
|
|
function_label = str(function_name or "llm_generate")
|
|
control_plane_agent_timeout = int(control_plane_agents_timeout_seconds) if is_control_plane_gateway_request else None
|
|
|
|
if is_control_plane_gateway_request:
|
|
_runtime_log(
|
|
"PMFlow 本地路由层调用",
|
|
(
|
|
f"start ts={_cn_now_text_ms()} trace_id={request_trace_id} "
|
|
f"function={function_label} "
|
|
f"request_timeout={float(request_timeout):.2f}s "
|
|
f"agent_timeout={control_plane_agent_timeout if control_plane_agent_timeout is not None else 'N/A'}"
|
|
),
|
|
)
|
|
|
|
try:
|
|
if provider_name == "ollama":
|
|
result = await self._chat_completion_via_ollama(
|
|
base_url=base_url,
|
|
model=model,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
messages=messages,
|
|
request_timeout=request_timeout,
|
|
)
|
|
else:
|
|
if not key:
|
|
raise ValueError("未配置 LLM API Key")
|
|
result = await self._chat_completion_via_openai_compatible(
|
|
key=key,
|
|
base_url=base_url,
|
|
model=model,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
messages=messages,
|
|
request_timeout=request_timeout,
|
|
structured_output=structured_output,
|
|
function_label=function_label,
|
|
provider_name=provider_name,
|
|
is_control_plane_gateway_request=is_control_plane_gateway_request,
|
|
control_plane_agents_timeout_seconds=control_plane_agents_timeout_seconds,
|
|
request_started_at_perf=request_started_at_perf,
|
|
request_trace_id=request_trace_id,
|
|
control_plane_agent_timeout=control_plane_agent_timeout,
|
|
routing_meta=normalized_routing_meta,
|
|
)
|
|
if use_provider_circuit_breaker:
|
|
self._record_provider_success(circuit_key)
|
|
return result
|
|
except Exception as exc:
|
|
if _is_control_plane_timeout_message(str(exc)):
|
|
elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0
|
|
_runtime_log(
|
|
"LLM超时异常",
|
|
(
|
|
f"detected ts={_cn_now_text_ms()} trace_id={request_trace_id} "
|
|
f"function={function_label} elapsed_ms={elapsed_ms:.0f} "
|
|
f"error={type(exc).__name__}:{str(exc)[:260]}"
|
|
),
|
|
)
|
|
logger.warning(
|
|
"llm_timeout_exception_detected",
|
|
trace_id=request_trace_id,
|
|
function_name=function_label,
|
|
request_timeout_seconds=float(request_timeout),
|
|
agent_timeout_seconds=control_plane_agent_timeout,
|
|
error_type=type(exc).__name__,
|
|
error=str(exc)[:260],
|
|
)
|
|
if is_control_plane_gateway_request:
|
|
elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0
|
|
_runtime_log(
|
|
"PMFlow 本地路由层调用",
|
|
(
|
|
f"exception ts={_cn_now_text_ms()} trace_id={request_trace_id} "
|
|
f"function={function_label} elapsed_ms={elapsed_ms:.0f} "
|
|
f"error={type(exc).__name__}:{str(exc)[:260]}"
|
|
),
|
|
)
|
|
if use_provider_circuit_breaker:
|
|
self._record_provider_failure(circuit_key, exc)
|
|
raise
|
|
|
|
async def _chat_completion_via_ollama(
|
|
self,
|
|
*,
|
|
base_url: str,
|
|
model: str,
|
|
temperature: float,
|
|
max_tokens: int,
|
|
messages: list[dict[str, str]],
|
|
request_timeout: float,
|
|
) -> ChatCompletionResult:
|
|
endpoint = f"{base_url.rstrip('/')}/api/chat"
|
|
payload = {
|
|
"model": model,
|
|
"messages": messages,
|
|
"stream": False,
|
|
"options": {
|
|
"temperature": temperature,
|
|
"num_predict": max_tokens,
|
|
},
|
|
}
|
|
async with httpx.AsyncClient(timeout=request_timeout, trust_env=False) as client:
|
|
resp = await client.post(endpoint, json=payload)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
message = data.get("message") if isinstance(data, dict) else {}
|
|
content = message.get("content") if isinstance(message, dict) else None
|
|
text = self._sanitize_visible_answer(self._extract_text_from_value(content))
|
|
if not text:
|
|
text = self._sanitize_visible_answer(self._extract_text_from_value(data.get("response")))
|
|
if not text and not isinstance(data, dict):
|
|
raise ValueError("Ollama 返回空内容")
|
|
return ChatCompletionResult(
|
|
text=text,
|
|
raw_request=payload,
|
|
raw_response=data,
|
|
provider_attempts=[
|
|
{
|
|
"provider": "ollama",
|
|
"endpoint": endpoint,
|
|
"request_payload": payload,
|
|
"response_payload": data,
|
|
"status": "success",
|
|
}
|
|
],
|
|
)
|
|
|
|
async def _chat_completion_stream_via_ollama(
|
|
self,
|
|
*,
|
|
base_url: str,
|
|
model: str,
|
|
temperature: float,
|
|
max_tokens: int,
|
|
messages: list[dict[str, str]],
|
|
request_timeout: float,
|
|
audit_context: dict[str, Any] | None = None,
|
|
) -> AsyncIterator[str]:
|
|
endpoint = f"{base_url.rstrip('/')}/api/chat"
|
|
payload = {
|
|
"model": model,
|
|
"messages": messages,
|
|
"stream": True,
|
|
"options": {
|
|
"temperature": temperature,
|
|
"num_predict": max_tokens,
|
|
},
|
|
}
|
|
chunks_seen = 0
|
|
response_samples: list[Any] = []
|
|
if audit_context is not None:
|
|
audit_context["provider_request_payload"] = payload
|
|
async with httpx.AsyncClient(timeout=request_timeout, trust_env=False) as client:
|
|
async with client.stream("POST", endpoint, json=payload) as resp:
|
|
if resp.status_code >= 400:
|
|
body = (await resp.aread()).decode("utf-8", errors="replace")
|
|
raise httpx.HTTPStatusError(
|
|
f"Ollama stream HTTP {resp.status_code}: {body[:300]}",
|
|
request=resp.request,
|
|
response=resp,
|
|
)
|
|
async for line in resp.aiter_lines():
|
|
text_line = str(line or "").strip()
|
|
if not text_line:
|
|
continue
|
|
try:
|
|
data = json.loads(text_line)
|
|
except Exception:
|
|
continue
|
|
if len(response_samples) < 8:
|
|
response_samples.append(data)
|
|
message = data.get("message") if isinstance(data, dict) else {}
|
|
chunk = self._sanitize_visible_answer(self._extract_text_from_value(message.get("content") if isinstance(message, dict) else ""))
|
|
if chunk:
|
|
chunks_seen += 1
|
|
yield chunk
|
|
if isinstance(data, dict) and data.get("done") is True:
|
|
break
|
|
if audit_context is not None:
|
|
audit_context["provider_response_payload"] = {"chunks_seen": chunks_seen, "samples": response_samples}
|
|
audit_context["provider_attempts"] = [
|
|
{
|
|
"provider": "ollama",
|
|
"endpoint": endpoint,
|
|
"request_payload": payload,
|
|
"response_payload": audit_context["provider_response_payload"],
|
|
"status": "success" if chunks_seen else "missing_stream_chunks",
|
|
}
|
|
]
|