rebuild(stage-10): deliver mentor and unified llm streaming
This commit is contained in:
leefer
@@ -0,0 +1,11 @@
|
||||
from backend.llm.gateway import LLMCall, LLMGateway, LLMGatewayError, LLMStreamEvent
|
||||
from backend.llm.provider import OpenAICompatibleClient, ProviderFailure
|
||||
|
||||
__all__ = (
|
||||
"LLMCall",
|
||||
"LLMGateway",
|
||||
"LLMGatewayError",
|
||||
"LLMStreamEvent",
|
||||
"OpenAICompatibleClient",
|
||||
"ProviderFailure",
|
||||
)
|
||||
@@ -0,0 +1,262 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
import uuid
|
||||
from collections.abc import Iterator
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from backend.database.connection import Database
|
||||
from backend.errors import BusinessError
|
||||
from backend.features.accounts.model_pool import ModelPoolService
|
||||
from backend.features.accounts.models import Principal
|
||||
from backend.features.accounts.service import MembershipService
|
||||
from backend.llm.provider import OpenAICompatibleClient, ProviderFailure
|
||||
from backend.llm.repository import LLMRepository
|
||||
|
||||
SHANGHAI = ZoneInfo("Asia/Shanghai")
|
||||
|
||||
|
||||
class LLMGatewayError(RuntimeError):
|
||||
def __init__(self, code: str, message: str, *, partial: bool = False) -> None:
|
||||
super().__init__(message)
|
||||
self.code = code
|
||||
self.partial = partial
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LLMProfile:
|
||||
model_id: int
|
||||
role: str
|
||||
base_url: str
|
||||
model_identifier: str
|
||||
api_key: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LLMCall:
|
||||
request_id: str
|
||||
user_id: int
|
||||
feature: str
|
||||
prompt_version: str
|
||||
business_id: str
|
||||
started_at: datetime
|
||||
usage_date: str
|
||||
input_chars: int
|
||||
quota_exempt: bool
|
||||
profiles: tuple[LLMProfile, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LLMStreamEvent:
|
||||
type: str
|
||||
content: str = ""
|
||||
request_id: str = ""
|
||||
|
||||
|
||||
class LLMGateway:
|
||||
def __init__(
|
||||
self,
|
||||
database: Database,
|
||||
repository: LLMRepository,
|
||||
memberships: MembershipService,
|
||||
model_pool: ModelPoolService,
|
||||
provider: OpenAICompatibleClient | None = None,
|
||||
) -> None:
|
||||
self._database = database
|
||||
self._repository = repository
|
||||
self._memberships = memberships
|
||||
self._model_pool = model_pool
|
||||
self._provider = provider or OpenAICompatibleClient()
|
||||
|
||||
def prepare(
|
||||
self,
|
||||
principal: Principal,
|
||||
*,
|
||||
feature: str,
|
||||
prompt_version: str,
|
||||
business_id: str,
|
||||
input_chars: int,
|
||||
) -> LLMCall:
|
||||
membership = self._memberships.view_for(principal)
|
||||
if not self._memberships.can_use_smart_features(principal):
|
||||
raise LLMGatewayError("membership_required", "该功能仅对会员开放。")
|
||||
try:
|
||||
runtime = self._model_pool.runtime_config()
|
||||
except BusinessError as exc:
|
||||
raise LLMGatewayError("not_configured", "智能解读服务尚未配置。") from exc
|
||||
records = (("primary", runtime.primary), ("fallback", runtime.fallback))
|
||||
profiles = tuple(
|
||||
LLMProfile(
|
||||
model_id=record.id,
|
||||
role=role,
|
||||
base_url=record.base_url,
|
||||
model_identifier=record.model_identifier,
|
||||
api_key=self._model_pool.decrypt_api_key(record),
|
||||
)
|
||||
for role, record in records
|
||||
if record is not None
|
||||
)
|
||||
now = datetime.now(SHANGHAI)
|
||||
request_id = uuid.uuid4().hex
|
||||
with self._database.transaction() as connection:
|
||||
used = self._repository.usage_today(
|
||||
connection, principal.user.id, now.date().isoformat()
|
||||
)
|
||||
active = self._repository.active_today(
|
||||
connection, principal.user.id, now.date().isoformat()
|
||||
)
|
||||
if not membership.quota_exempt and used + active >= membership.daily_limit:
|
||||
raise LLMGatewayError(
|
||||
"quota_exhausted",
|
||||
f"今日智能分析额度已用完（{membership.daily_limit} 次）。",
|
||||
)
|
||||
self._repository.reserve(
|
||||
connection,
|
||||
request_id=request_id,
|
||||
user_id=principal.user.id,
|
||||
feature=feature,
|
||||
business_id=business_id,
|
||||
prompt_version=prompt_version,
|
||||
started_at=now.isoformat(timespec="seconds"),
|
||||
input_chars=max(0, input_chars),
|
||||
)
|
||||
return LLMCall(
|
||||
request_id=request_id,
|
||||
user_id=principal.user.id,
|
||||
feature=feature,
|
||||
prompt_version=prompt_version,
|
||||
business_id=business_id,
|
||||
started_at=now,
|
||||
usage_date=now.date().isoformat(),
|
||||
input_chars=max(0, input_chars),
|
||||
quota_exempt=membership.quota_exempt,
|
||||
profiles=profiles,
|
||||
)
|
||||
|
||||
def stream(
|
||||
self, call: LLMCall, messages: list[dict[str, str]]
|
||||
) -> Iterator[LLMStreamEvent]:
|
||||
last_error = "unavailable"
|
||||
for profile in call.profiles:
|
||||
attempt_started = time.perf_counter()
|
||||
with self._database.transaction() as connection:
|
||||
attempt_id = self._repository.start_attempt(
|
||||
connection,
|
||||
call.request_id,
|
||||
profile.model_id,
|
||||
profile.role,
|
||||
_now(),
|
||||
call.input_chars,
|
||||
)
|
||||
output = ""
|
||||
try:
|
||||
upstream = iter(self._provider.stream(profile, messages))
|
||||
first = next(upstream)
|
||||
except (ProviderFailure, StopIteration) as exc:
|
||||
last_error = exc.code if isinstance(exc, ProviderFailure) else "empty_response"
|
||||
self._finish_attempt(attempt_id, "failed", attempt_started, last_error, 0)
|
||||
continue
|
||||
except Exception as exc:
|
||||
last_error = "unavailable"
|
||||
self._finish_attempt(attempt_id, "failed", attempt_started, last_error, 0)
|
||||
self._finish_request(call, "failed", last_error, 0)
|
||||
raise LLMGatewayError(
|
||||
"model_unavailable", "智能解读服务暂不可用，请稍后重试。"
|
||||
) from exc
|
||||
with self._database.transaction() as connection:
|
||||
self._repository.set_request_status(connection, call.request_id, "streaming")
|
||||
output = first
|
||||
try:
|
||||
yield LLMStreamEvent("delta", first, call.request_id)
|
||||
for chunk in upstream:
|
||||
output += chunk
|
||||
yield LLMStreamEvent("delta", chunk, call.request_id)
|
||||
except GeneratorExit:
|
||||
self._finish_attempt(
|
||||
attempt_id, "stopped", attempt_started, "cancelled", len(output)
|
||||
)
|
||||
self._finish_request(call, "stopped", "cancelled", len(output))
|
||||
raise
|
||||
except ProviderFailure as exc:
|
||||
self._finish_attempt(
|
||||
attempt_id, "failed", attempt_started, exc.code, len(output)
|
||||
)
|
||||
self._finish_request(call, "failed", exc.code, len(output))
|
||||
raise LLMGatewayError(
|
||||
"stream_interrupted", "智能解读连接中断，请稍后重试。", partial=True
|
||||
) from exc
|
||||
except Exception as exc:
|
||||
self._finish_attempt(
|
||||
attempt_id, "failed", attempt_started, "unavailable", len(output)
|
||||
)
|
||||
self._finish_request(call, "failed", "unavailable", len(output))
|
||||
raise LLMGatewayError(
|
||||
"stream_interrupted", "智能解读连接中断，请稍后重试。", partial=True
|
||||
) from exc
|
||||
self._finish_attempt(attempt_id, "success", attempt_started, "", len(output))
|
||||
self._finish_request(call, "success", "", len(output))
|
||||
yield LLMStreamEvent("done", request_id=call.request_id)
|
||||
return
|
||||
self._finish_request(call, "failed", last_error, 0)
|
||||
code, message = _safe_error(last_error)
|
||||
raise LLMGatewayError(code, message)
|
||||
|
||||
def _finish_attempt(
|
||||
self,
|
||||
attempt_id: int,
|
||||
status: str,
|
||||
started: float,
|
||||
error_type: str,
|
||||
output_chars: int,
|
||||
) -> None:
|
||||
with self._database.transaction() as connection:
|
||||
self._repository.finish_attempt(
|
||||
connection,
|
||||
attempt_id,
|
||||
status,
|
||||
_now(),
|
||||
round((time.perf_counter() - started) * 1000),
|
||||
error_type,
|
||||
output_chars,
|
||||
)
|
||||
|
||||
def _finish_request(
|
||||
self, call: LLMCall, status: str, error_type: str, output_chars: int
|
||||
) -> None:
|
||||
now = datetime.now(SHANGHAI)
|
||||
with self._database.transaction() as connection:
|
||||
current = connection.execute(
|
||||
"SELECT status FROM llm_requests WHERE id = ?", (call.request_id,)
|
||||
).fetchone()
|
||||
if current is None or str(current["status"]) in {"success", "failed", "stopped"}:
|
||||
return
|
||||
self._repository.set_request_status(
|
||||
connection,
|
||||
call.request_id,
|
||||
status,
|
||||
completed_at=now.isoformat(timespec="seconds"),
|
||||
duration_ms=round((now - call.started_at).total_seconds() * 1000),
|
||||
error_type=error_type,
|
||||
output_chars=output_chars,
|
||||
)
|
||||
if status == "success" and not call.quota_exempt:
|
||||
self._repository.increment_usage(
|
||||
connection, call.user_id, call.usage_date, now.isoformat(timespec="seconds")
|
||||
)
|
||||
|
||||
|
||||
def _now() -> str:
|
||||
return datetime.now(SHANGHAI).isoformat(timespec="seconds")
|
||||
|
||||
|
||||
def _safe_error(error_type: str) -> tuple[str, str]:
|
||||
messages = {
|
||||
"capacity": ("model_capacity", "智能解读服务当前繁忙，请稍后重试。"),
|
||||
"rate_limited": ("model_rate_limited", "智能解读请求过于频繁，请稍后重试。"),
|
||||
"authentication": ("model_authentication", "智能解读服务配置失效，请联系管理员。"),
|
||||
"timeout": ("model_timeout", "智能解读等待超时，请稍后重试。"),
|
||||
"network": ("model_network", "智能解读网络暂不可用，请稍后重试。"),
|
||||
}
|
||||
return messages.get(error_type, ("model_unavailable", "智能解读服务暂不可用，请稍后重试。"))
|
||||
@@ -0,0 +1,104 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from collections.abc import Iterator
|
||||
from typing import Any, Protocol
|
||||
|
||||
from backend.llm.streaming import TextAccumulator
|
||||
|
||||
|
||||
class RuntimeProfile(Protocol):
|
||||
api_key: str
|
||||
base_url: str
|
||||
model_identifier: str
|
||||
|
||||
|
||||
class ProviderFailure(RuntimeError):
|
||||
def __init__(self, code: str) -> None:
|
||||
super().__init__(code)
|
||||
self.code = code
|
||||
|
||||
|
||||
class OpenAICompatibleClient:
|
||||
def __init__(self, timeout_seconds: int = 90) -> None:
|
||||
self._timeout = timeout_seconds
|
||||
|
||||
def stream(
|
||||
self, profile: RuntimeProfile, messages: list[dict[str, str]]
|
||||
) -> Iterator[str]:
|
||||
payload = json.dumps(
|
||||
{"model": profile.model_identifier, "messages": messages, "stream": True},
|
||||
ensure_ascii=False,
|
||||
).encode("utf-8")
|
||||
request = urllib.request.Request(
|
||||
f"{profile.base_url.rstrip('/')}/chat/completions",
|
||||
data=payload,
|
||||
headers={
|
||||
"Accept": "text/event-stream",
|
||||
"Authorization": f"Bearer {profile.api_key}",
|
||||
"Content-Type": "application/json",
|
||||
"User-Agent": "XiaobaiReview/2",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=self._timeout) as response:
|
||||
accumulator = TextAccumulator()
|
||||
for raw_line in response:
|
||||
line = raw_line.decode("utf-8", errors="replace").strip()
|
||||
if not line or line.startswith(":"):
|
||||
continue
|
||||
if line.startswith("data:"):
|
||||
line = line[5:].strip()
|
||||
if line == "[DONE]":
|
||||
break
|
||||
try:
|
||||
payload = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
choices = payload.get("choices") if isinstance(payload, dict) else None
|
||||
if not isinstance(choices, list) or not choices:
|
||||
continue
|
||||
choice = choices[0]
|
||||
chunk = accumulator.feed(choice if isinstance(choice, dict) else {})
|
||||
if chunk:
|
||||
yield chunk
|
||||
if not accumulator.text:
|
||||
raise ProviderFailure("empty_response")
|
||||
except ProviderFailure:
|
||||
raise
|
||||
except urllib.error.HTTPError as exc:
|
||||
raise ProviderFailure(_http_error(exc)) from exc
|
||||
except TimeoutError as exc:
|
||||
raise ProviderFailure("timeout") from exc
|
||||
except (urllib.error.URLError, ConnectionError, OSError) as exc:
|
||||
raise ProviderFailure("network") from exc
|
||||
|
||||
|
||||
def _http_error(error: urllib.error.HTTPError) -> str:
|
||||
if error.code in {401, 403}:
|
||||
return "authentication"
|
||||
if error.code == 429:
|
||||
detail = _error_detail(error)
|
||||
return "capacity" if "capacity" in detail.casefold() else "rate_limited"
|
||||
if error.code in {408, 504}:
|
||||
return "timeout"
|
||||
if error.code >= 500:
|
||||
detail = _error_detail(error)
|
||||
return "capacity" if "capacity" in detail.casefold() else "upstream"
|
||||
return "request_rejected"
|
||||
|
||||
|
||||
def _error_detail(error: urllib.error.HTTPError) -> str:
|
||||
try:
|
||||
payload: Any = json.loads(error.read().decode("utf-8", errors="replace"))
|
||||
except (json.JSONDecodeError, OSError):
|
||||
return ""
|
||||
if not isinstance(payload, dict):
|
||||
return ""
|
||||
detail = payload.get("error") or payload.get("message") or ""
|
||||
if isinstance(detail, dict):
|
||||
detail = detail.get("message") or detail.get("code") or ""
|
||||
return str(detail)[:500]
|
||||
@@ -0,0 +1,125 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
|
||||
|
||||
class LLMRepository:
|
||||
@staticmethod
|
||||
def usage_today(connection: sqlite3.Connection, user_id: int, usage_date: str) -> int:
|
||||
row = connection.execute(
|
||||
"SELECT successful_calls FROM llm_usage_daily WHERE user_id = ? AND usage_date = ?",
|
||||
(user_id, usage_date),
|
||||
).fetchone()
|
||||
return int(row["successful_calls"]) if row else 0
|
||||
|
||||
@staticmethod
|
||||
def active_today(connection: sqlite3.Connection, user_id: int, day_prefix: str) -> int:
|
||||
row = connection.execute(
|
||||
"""
|
||||
SELECT COUNT(*) AS total FROM llm_requests
|
||||
WHERE user_id = ? AND started_at LIKE ? AND status IN ('reserved', 'streaming')
|
||||
""",
|
||||
(user_id, f"{day_prefix}%"),
|
||||
).fetchone()
|
||||
return int(row["total"])
|
||||
|
||||
@staticmethod
|
||||
def reserve(
|
||||
connection: sqlite3.Connection,
|
||||
*,
|
||||
request_id: str,
|
||||
user_id: int,
|
||||
feature: str,
|
||||
business_id: str,
|
||||
prompt_version: str,
|
||||
started_at: str,
|
||||
input_chars: int,
|
||||
) -> None:
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO llm_requests (
|
||||
id, user_id, feature, business_id, prompt_version,
|
||||
status, started_at, input_chars
|
||||
) VALUES (?, ?, ?, ?, ?, 'reserved', ?, ?)
|
||||
""",
|
||||
(
|
||||
request_id,
|
||||
user_id,
|
||||
feature,
|
||||
business_id,
|
||||
prompt_version,
|
||||
started_at,
|
||||
input_chars,
|
||||
),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def set_request_status(
|
||||
connection: sqlite3.Connection,
|
||||
request_id: str,
|
||||
status: str,
|
||||
*,
|
||||
completed_at: str | None = None,
|
||||
duration_ms: int = 0,
|
||||
error_type: str = "",
|
||||
output_chars: int = 0,
|
||||
) -> None:
|
||||
connection.execute(
|
||||
"""
|
||||
UPDATE llm_requests SET status = ?, completed_at = ?, duration_ms = ?,
|
||||
error_type = ?, output_chars = ? WHERE id = ?
|
||||
""",
|
||||
(status, completed_at, duration_ms, error_type, output_chars, request_id),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def increment_usage(
|
||||
connection: sqlite3.Connection, user_id: int, usage_date: str, updated_at: str
|
||||
) -> None:
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO llm_usage_daily (user_id, usage_date, successful_calls, updated_at)
|
||||
VALUES (?, ?, 1, ?)
|
||||
ON CONFLICT(user_id, usage_date) DO UPDATE SET
|
||||
successful_calls = successful_calls + 1,
|
||||
updated_at = excluded.updated_at
|
||||
""",
|
||||
(user_id, usage_date, updated_at),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def start_attempt(
|
||||
connection: sqlite3.Connection,
|
||||
request_id: str,
|
||||
model_id: int,
|
||||
role: str,
|
||||
started_at: str,
|
||||
input_chars: int,
|
||||
) -> int:
|
||||
cursor = connection.execute(
|
||||
"""
|
||||
INSERT INTO llm_attempts (
|
||||
request_id, model_id, role, status, started_at, input_chars
|
||||
) VALUES (?, ?, ?, 'streaming', ?, ?)
|
||||
""",
|
||||
(request_id, model_id, role, started_at, input_chars),
|
||||
)
|
||||
return int(cursor.lastrowid)
|
||||
|
||||
@staticmethod
|
||||
def finish_attempt(
|
||||
connection: sqlite3.Connection,
|
||||
attempt_id: int,
|
||||
status: str,
|
||||
completed_at: str,
|
||||
duration_ms: int,
|
||||
error_type: str,
|
||||
output_chars: int,
|
||||
) -> None:
|
||||
connection.execute(
|
||||
"""
|
||||
UPDATE llm_attempts SET status = ?, completed_at = ?, duration_ms = ?,
|
||||
error_type = ?, output_chars = ? WHERE id = ?
|
||||
""",
|
||||
(status, completed_at, duration_ms, error_type, output_chars, attempt_id),
|
||||
)
|
||||
@@ -0,0 +1,38 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
|
||||
class TextAccumulator:
|
||||
"""Normalize delta streams and providers that repeat full snapshots."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.text = ""
|
||||
self._saw_delta = False
|
||||
|
||||
def feed(self, choice: dict[str, Any]) -> str:
|
||||
delta = choice.get("delta")
|
||||
if isinstance(delta, dict) and delta.get("content") is not None:
|
||||
chunk = str(delta.get("content") or "")
|
||||
if chunk:
|
||||
self._saw_delta = True
|
||||
self.text += chunk
|
||||
return chunk
|
||||
message = choice.get("message")
|
||||
if not isinstance(message, dict) or message.get("content") is None:
|
||||
return ""
|
||||
snapshot = str(message.get("content") or "")
|
||||
if not snapshot:
|
||||
return ""
|
||||
if not self.text:
|
||||
self.text = snapshot
|
||||
return snapshot
|
||||
if snapshot == self.text or self.text.startswith(snapshot):
|
||||
return ""
|
||||
if snapshot.startswith(self.text):
|
||||
suffix = snapshot[len(self.text) :]
|
||||
self.text = snapshot
|
||||
return suffix
|
||||
if self._saw_delta:
|
||||
return ""
|
||||
return ""
|
||||
Reference in New Issue
Block a user