114 lines
3.1 KiB
Python
114 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from app.enums.credit_record import CreditRecordChargeKind
|
|
from app.enums.llm_billing import LlmBillingLedgerState
|
|
from app.services.generation.billing_service import build_credit_biz_key
|
|
|
|
|
|
class LlmBillingConfigurationError(RuntimeError):
|
|
"""LLM 统一账务配置无效，必须在调用模型前终止。"""
|
|
|
|
|
|
class LlmBillingStateError(RuntimeError):
|
|
"""当前 attempt 的账务流水状态不允许继续执行。"""
|
|
|
|
|
|
@dataclass(slots=True, frozen=True)
|
|
class LlmBillingPolicy:
|
|
enabled: bool
|
|
hold_credits: float
|
|
config_key: str | None = None
|
|
source_key: str | None = None
|
|
valid: bool = True
|
|
error: str | None = None
|
|
|
|
@property
|
|
def bypassed(self) -> bool:
|
|
return not self.enabled
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LlmBillingContext:
|
|
user_id: str
|
|
owner_type: str
|
|
owner_id: str
|
|
attempt_no: int
|
|
charge_kind: str = CreditRecordChargeKind.TEXT_PROMPT.value
|
|
billing_scene: str | None = None
|
|
source_module: str | None = None
|
|
source_project_id: str | None = None
|
|
source_step_id: str | None = None
|
|
source_step_code: str | None = None
|
|
related_id: str | None = None
|
|
hold_credits: float | None = None
|
|
hold_config_key: str | None = None
|
|
description_prefix: str = "LLM"
|
|
trace_id: str | None = None
|
|
request_id: str | None = None
|
|
celery_task_id: str | None = None
|
|
provider: str | None = None
|
|
model_name: str | None = None
|
|
token_usage_id: str | None = None
|
|
|
|
@property
|
|
def hold_biz_key(self) -> str:
|
|
return build_credit_biz_key(
|
|
owner_type=self.owner_type,
|
|
owner_id=self.owner_id,
|
|
attempt_no=self.attempt_no,
|
|
charge_kind=self.charge_kind,
|
|
action="hold",
|
|
)
|
|
|
|
@property
|
|
def hold_release_biz_key(self) -> str:
|
|
return build_credit_biz_key(
|
|
owner_type=self.owner_type,
|
|
owner_id=self.owner_id,
|
|
attempt_no=self.attempt_no,
|
|
charge_kind=self.charge_kind,
|
|
action="hold_release",
|
|
)
|
|
|
|
@property
|
|
def charge_biz_key(self) -> str:
|
|
return build_credit_biz_key(
|
|
owner_type=self.owner_type,
|
|
owner_id=self.owner_id,
|
|
attempt_no=self.attempt_no,
|
|
charge_kind=self.charge_kind,
|
|
action="charge",
|
|
)
|
|
|
|
@property
|
|
def ledger_biz_keys(self) -> tuple[str, str, str]:
|
|
return self.hold_biz_key, self.hold_release_biz_key, self.charge_biz_key
|
|
|
|
|
|
@dataclass(slots=True, frozen=True)
|
|
class LlmHoldResult:
|
|
amount: float
|
|
state: LlmBillingLedgerState
|
|
created: bool = False
|
|
record_id: str | None = None
|
|
reason: str | None = None
|
|
|
|
@property
|
|
def bypassed(self) -> bool:
|
|
return self.state == LlmBillingLedgerState.BILLING_BYPASSED
|
|
|
|
|
|
@dataclass(slots=True, frozen=True)
|
|
class LlmHoldValidation:
|
|
can_execute: bool
|
|
amount: float
|
|
state: LlmBillingLedgerState
|
|
reason: str | None = None
|
|
hold_record_id: str | None = None
|
|
|
|
@property
|
|
def bypassed(self) -> bool:
|
|
return self.state == LlmBillingLedgerState.BILLING_BYPASSED
|