iomgaa
779a9b29a3
llm_calls gains a nullable stream_end TEXT column (auto backfill, manual-mode trimming as in 1.3.8). TelemetryRecorder.record_llm_call takes a new required keyword stream_end; the emitter reads it from the response on attempt rows and writes NULL for failed, cache-hit and terminal rows. Red: 18 tests in tests/unit/test_telemetry.py and test_ports.py failed against the pre-change source (missing column, 38-field signature, stream_end absent from emitted rows).
362 lines
13 KiB
Python
362 lines
13 KiB
Python
"""全部端口协议与治理快照类型(M1 设计 §4;最内层,只依赖 types/errors/标准库)。
|
||
|
||
决策逻辑与状态存储分离(D3): 中间件只面向这里的 Protocol,
|
||
后端(memory/redis)各自实现同一契约并共用一套契约测试。
|
||
时间量纲一律**秒**(CHS Redis 实现内部的毫秒换算是后端私事,不进契约)。
|
||
"""
|
||
|
||
import asyncio
|
||
from collections.abc import Awaitable, Callable
|
||
from dataclasses import dataclass
|
||
from enum import StrEnum
|
||
from typing import Any, Protocol, runtime_checkable
|
||
|
||
from .types import (
|
||
ChatRequest,
|
||
Effort,
|
||
EmbeddingTransportResult,
|
||
LLMResponse,
|
||
OcrLayoutResult,
|
||
OcrLayoutTransportResult,
|
||
OcrTextResult,
|
||
OcrTextTransportResult,
|
||
ResponseMode,
|
||
SourceConfig,
|
||
SourceStats,
|
||
TelemetryStatus,
|
||
TransportResult,
|
||
)
|
||
|
||
CallNext = Callable[[ChatRequest], Awaitable[LLMResponse]]
|
||
|
||
|
||
@runtime_checkable
|
||
class Middleware(Protocol):
|
||
"""洋葱层: 包裹 call_next,层与层正交(D1)。"""
|
||
|
||
async def __call__(self, request: ChatRequest, call_next: CallNext) -> LLMResponse: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class Transport(Protocol):
|
||
"""一次原始调用的协议细节(请求组装/流式解析/错误翻译);不含任何治理。
|
||
|
||
`response_mode` 为必填返回契约：openai 成功必须提供 completion，
|
||
不支持者应 RequestRejected，不能静默退回文本。工具的实际增量与首次身份
|
||
同正文/推理共用活性事件；同帧只置位一次，重复身份不算进展。
|
||
|
||
`reasoning_effort` 是本次调用要求的推理档位(`None` = 不表态,随源级配置)。
|
||
它必须走**协议参数**而不能让 transport 自己去读 `ChatRequest`: 端口只收拆开的
|
||
请求要素,是为了让 transport 不依赖洋葱内部的请求类型(P7 端口最内层)。
|
||
|
||
该参数**不设默认值**,与 `TelemetryRecorder.record_llm_call` 同一既有约定:
|
||
库外无第三方实现者,写全签名的成本为零,而默认值会把"某一层漏传"变成静默的
|
||
"调用方没表态"——一次本该报错的漏配就此变成一次悄悄涨价的调用。
|
||
|
||
`first_token_event` 同一约定(1.3.7 对冲 H2): `None` = 调用方不观测首 token
|
||
(未启用对冲);非流式实现**永不置位**(物理上无中途信号,事件自然退化为纯时间
|
||
阈值),流式实现在首个增量(内容或思考)到达时置位。
|
||
|
||
`content_event` 同一约定(1.3.8 中段沉默对冲,issue #27/#28): `None` = 调用方
|
||
不观测增量 chunk(未启用中段对冲);**流式**实现每到达一个增量 chunk(`content`
|
||
或 `reasoning_content`,与看门狗/首 token 活性口径一致)置位一次;空 chunk、
|
||
usage 帧、SSE keepalive 不算;非流式实现**永不置位**;对冲路恒收 `None`
|
||
(v1 不梯次,设计 §5.2 N3③)。
|
||
"""
|
||
|
||
async def complete(
|
||
self,
|
||
*,
|
||
messages: list[dict[str, Any]],
|
||
source: SourceConfig,
|
||
stream: bool,
|
||
overlay: dict[str, Any],
|
||
call_id: str,
|
||
reasoning_effort: Effort | None,
|
||
first_token_event: asyncio.Event | None,
|
||
content_event: asyncio.Event | None,
|
||
response_mode: ResponseMode,
|
||
) -> TransportResult: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class EmbeddingTransport(Protocol):
|
||
"""一次原始 embedding 调用的协议细节(M2 §7);不含任何治理。"""
|
||
|
||
async def embed(
|
||
self, *, texts: list[str], source: SourceConfig, call_id: str
|
||
) -> EmbeddingTransportResult: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class OcrTextPort(Protocol):
|
||
"""OCR 文本转录端口(D9;对应 MonkeyOCR POST /ocr/text)。"""
|
||
|
||
async def recognize_text(self, image: bytes) -> OcrTextResult: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class OcrLayoutPort(Protocol):
|
||
"""OCR 版面解析端口(D9;对应 MonkeyOCR POST /parse → GET ZIP)。"""
|
||
|
||
async def parse_layout(self, image: bytes) -> OcrLayoutResult: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class OcrTransport(Protocol):
|
||
"""一次原始 OCR 调用的协议细节(M3 设计 §3.2);不含任何治理。
|
||
|
||
check_health 是探测不是调用: 2xx → True,异常 → False,
|
||
CancelledError 穿透;由消费方(启动门)决定成败语义。
|
||
"""
|
||
|
||
async def recognize_text(
|
||
self, *, image: bytes, source: SourceConfig, call_id: str
|
||
) -> OcrTextTransportResult: ...
|
||
|
||
async def parse_layout(
|
||
self, *, image: bytes, source: SourceConfig, call_id: str
|
||
) -> OcrLayoutTransportResult: ...
|
||
|
||
async def check_health(self, *, source: SourceConfig) -> bool: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class Permit(Protocol):
|
||
"""限流入场许可;settle/release 均幂等,finally 中必然执行。"""
|
||
|
||
async def release(self) -> None: ...
|
||
|
||
async def settle(self, actual_tokens: int) -> None: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class RateLimiter(Protocol):
|
||
"""限流契约(CHS limiter.py 形态): 并发/RPM/TPM × 全局/单源 六道闸。"""
|
||
|
||
async def try_acquire(self, source_key: str, est_tokens: int) -> Permit | None: ...
|
||
|
||
async def acquire(self, source_key: str, est_tokens: int) -> Permit: ...
|
||
|
||
async def source_stats(self, source_key: str) -> SourceStats: ...
|
||
|
||
async def mark_progress(self) -> None: ...
|
||
|
||
async def progress_age_s(self) -> float: ...
|
||
|
||
|
||
class GateState(StrEnum):
|
||
"""熔断状态机三态。"""
|
||
|
||
CLOSED = "closed"
|
||
OPEN = "open"
|
||
HALF_OPEN = "half_open"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class GateDecision:
|
||
"""一次源准入决定,同时是结果写回所需的 fencing token(CHS ports.py:405)。"""
|
||
|
||
source_name: str
|
||
allowed: bool
|
||
state: GateState
|
||
epoch: int
|
||
is_probe: bool
|
||
probe_owner: str | None
|
||
retry_after_s: float
|
||
|
||
def __post_init__(self) -> None:
|
||
# 校验逐条移植 CHS ports.py:405-440,保证快照内部一致
|
||
self._check_bounds()
|
||
self._check_admission()
|
||
self._check_probe()
|
||
|
||
def _check_bounds(self) -> None:
|
||
if not self.source_name.strip():
|
||
raise ValueError("source_name 不能为空")
|
||
if self.epoch < 0:
|
||
raise ValueError("epoch 不能为负")
|
||
if self.retry_after_s < 0:
|
||
raise ValueError("retry_after_s 不能为负")
|
||
|
||
def _check_admission(self) -> None:
|
||
if self.allowed and self.state is GateState.OPEN:
|
||
raise ValueError("OPEN 状态不得准入")
|
||
if self.allowed and self.state is GateState.HALF_OPEN and not self.is_probe:
|
||
raise ValueError("HALF_OPEN 准入必须是探针")
|
||
|
||
def _check_probe(self) -> None:
|
||
if self.is_probe:
|
||
probe_valid = (
|
||
self.allowed
|
||
and self.state is GateState.HALF_OPEN
|
||
and self.probe_owner is not None
|
||
and bool(self.probe_owner.strip())
|
||
)
|
||
if not probe_valid:
|
||
raise ValueError("探针决定必须同时包含准入、HALF_OPEN 与 owner")
|
||
elif self.probe_owner is not None:
|
||
raise ValueError("非探针决定不得携带 probe_owner")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class GateUpdate:
|
||
"""门控写回快照;applied=False 表示 fencing 已拒绝迟到结果(CHS ports.py:442)。"""
|
||
|
||
applied: bool
|
||
state: GateState
|
||
epoch: int
|
||
failure_count: int
|
||
retry_after_s: float
|
||
|
||
def __post_init__(self) -> None:
|
||
if self.epoch < 0 or self.failure_count < 0 or self.retry_after_s < 0:
|
||
raise ValueError("门控更新快照不能包含负数")
|
||
|
||
|
||
@runtime_checkable
|
||
class ProviderGate(Protocol):
|
||
"""熔断契约(CHS provider_gate.py 形态): 半开单探针租约 + epoch fencing 在契约内。"""
|
||
|
||
async def try_enter(self, source_name: str, owner: str) -> GateDecision: ...
|
||
|
||
async def record_success(
|
||
self, entry: GateDecision, *, count_attempt: bool = True
|
||
) -> GateUpdate: ...
|
||
|
||
async def record_failure(
|
||
self, entry: GateDecision, reason: str, force_open: bool
|
||
) -> GateUpdate: ...
|
||
|
||
async def release_probe(self, entry: GateDecision) -> GateUpdate: ...
|
||
|
||
async def retry_after_s(self, sources: tuple[str, ...]) -> float: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class CacheBackend(Protocol):
|
||
"""响应缓存后端: 笨 KV;key 公式归 CacheMW(算法一份)。"""
|
||
|
||
async def get(self, key: str) -> str | None: ...
|
||
|
||
async def set(self, key: str, value: str, ttl_s: int) -> None: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class SourceSelector(Protocol):
|
||
"""选源策略(CHS selector.py 形态): 返回本次尝试的候选顺序。"""
|
||
|
||
def order(
|
||
self, sources: list[SourceConfig], stats: dict[str, SourceStats]
|
||
) -> list[SourceConfig]: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class OutcomeAwareSelector(Protocol):
|
||
"""可选选源器扩展(M2.5 设计 §3.2): 消费尝试结果以维护健康视图。
|
||
|
||
RetryMW 构造时 isinstance 判定一次;非本 Protocol 的选源器不受影响。
|
||
喂数口径: 真实成功 ok=True;Transient/SourceDead/429 ok=False;
|
||
ResultInvalid 与"网关健康拒坏请求"不喂(坏结果 ≠ 坏服务)。
|
||
"""
|
||
|
||
def record_outcome(self, source_name: str, ok: bool) -> None: ...
|
||
|
||
def health(self, source_name: str) -> float: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class StructuredOutputStrategy(Protocol):
|
||
"""结构化输出策略(D7/D14): 请求侧叠加 + 响应侧解析。"""
|
||
|
||
def request_overlay(self, schema: dict[str, Any] | None) -> dict[str, Any]: ...
|
||
|
||
def parse(self, text: str) -> Any: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class TelemetryStatusProvider(Protocol):
|
||
"""可自述可写状态的遥测后端;`TelemetryRecorder` 的**可选**伴生端口(issue #15)。
|
||
|
||
与 `TelemetryRecorder` 分开而不是给它加成员,是因为后者是 `@runtime_checkable`
|
||
而运行时检查按属性存在性做: 加一个属性会让所有只实现 `record_llm_call` 的
|
||
实现**当场不再是** `TelemetryRecorder`,下游若有同款 isinstance 断言,升级即断
|
||
(设计 §3.3)。消费方一律先 isinstance 再取值,取不到就当没有状态可报。
|
||
"""
|
||
|
||
@property
|
||
def telemetry_status(self) -> TelemetryStatus: ...
|
||
|
||
|
||
@runtime_checkable
|
||
class TelemetryRecorder(Protocol):
|
||
"""遥测后端;39 字段冻结(M1 设计 §4.4 + issue #3/#4/#11/#16/#20 + 1.3.5/1.3.8/1.3.10),唯一调用点是 TelemetryEmitter。
|
||
|
||
新增参数不设默认值: 库外无第三方实现者(三项目迁移时删除了各自的同名
|
||
Protocol),完整签名的成本为零,而少写一列会被 emitter 的降级吞成 warning。
|
||
|
||
`tenant_id` 与 `meta` 到达 recorder 时**已由 emitter 归一化**——`tenant_id`
|
||
的 `None` 已转空串,`meta` 已序列化为 JSON 字符串(空 dict 为 `'{}'`)。
|
||
`thinking_observation` 同理: emitter 已把 `ThinkingObservation` 取成 `.value`
|
||
的裸 `str`(`StrEnum` 是 `str` 子类,而 asyncpg 的参数编码对子类不保证接受,
|
||
遥测写失败又只降级成 warning——PG 那一路会静默少一列数据)。
|
||
`reasoning_effort` 同一先例(issue #20): emitter 已把 `Effort` 取成 `.value`
|
||
的裸 `str`,`None` 表示调用方没表态——它与 `'none'`(明确要求不推理)不可折叠。
|
||
|
||
recorder 只负责落库,不做任何语义判断,与 `sampling` 列由
|
||
`canonical_sampling_json()` 在 emitter 侧定型是同一先例。
|
||
|
||
1.3.5 新增十列同理已在 emitter 侧定型: `operation` 是**公开方法**四值之一
|
||
(与 `PolyGatewayError.operation` 这个 HTTP 子操作是两个语义);`event_kind` 区分
|
||
attempt / cache_hit / terminal_failure 三类行;`attempts` 与 `total_latency_ms`
|
||
只在终态行非空;`error_body` 沿用 transport 侧 `summarize_body` 的上限,
|
||
**不进 `PGW_TELEMETRY_TEXT_CAP` 的覆盖面**。
|
||
|
||
**本签名是装配闸的唯一事实源**: `TelemetryEmitter.__init__` 按它派生参数名做
|
||
一次 `signature.bind` 形状校验(设计 §7),改本签名即改闸的判据。
|
||
"""
|
||
|
||
async def record_llm_call(
|
||
self,
|
||
*,
|
||
call_id: str,
|
||
parent_call_id: str | None,
|
||
session_id: str | None,
|
||
model: str,
|
||
provider: str,
|
||
source_name: str,
|
||
messages: str,
|
||
response: str,
|
||
thinking: str,
|
||
prompt_tokens: int,
|
||
completion_tokens: int,
|
||
usage_source: str,
|
||
latency_ms: int,
|
||
ttft_ms: float | None,
|
||
max_inter_token_ms: float | None,
|
||
cache_hit: bool,
|
||
error: str | None,
|
||
cost: float | None,
|
||
cached_prompt_tokens: int | None,
|
||
model_reported: str | None,
|
||
sampling: str | None,
|
||
reasoning_tokens: int | None,
|
||
tenant_id: str,
|
||
meta: str,
|
||
thinking_observation: str,
|
||
reasoning_effort: str | None,
|
||
scope: str,
|
||
operation: str,
|
||
logical_call_id: str | None,
|
||
event_kind: str,
|
||
http_status_code: int | None,
|
||
error_type: str | None,
|
||
cause_type: str | None,
|
||
error_body: str | None,
|
||
attempts: int | None,
|
||
total_latency_ms: int | None,
|
||
stall_ms: int | None,
|
||
stall_count: int | None,
|
||
stream_end: str | None,
|
||
) -> None: ...
|