81 lines
3.5 KiB
Python
81 lines
3.5 KiB
Python
"""ORM model for model-stream truncation events."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import Boolean, Index, Integer, String, Text
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.types import BeijingDateTime
|
|
|
|
|
|
class StreamTruncationEventRow(Base):
|
|
"""One row per detected model-stream truncation.
|
|
|
|
Written from the patched OpenAI-compatible streaming layer when a provider
|
|
stream ends abnormally: without a terminal frame (silently committed half
|
|
answer), via ``IncompleteChatStreamError`` (strict integrity mode, retried),
|
|
on an output-length cap, or interrupted mid-stream by a transport error.
|
|
Rows are retained for at most ``RETENTION_DAYS`` by the SQL repository.
|
|
"""
|
|
|
|
__tablename__ = "stream_truncation_events"
|
|
__table_args__ = (
|
|
Index("ix_stream_truncation_created_model", "created_at", "model_name"),
|
|
Index("ix_stream_truncation_created_event", "created_at", "event_type"),
|
|
)
|
|
|
|
id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
BeijingDateTime(), default=lambda: datetime.now(UTC), index=True
|
|
)
|
|
|
|
# Configured model name (the config.yaml entry name), sanitized by the model layer.
|
|
model_name: Mapped[str | None] = mapped_column(String(128), index=True)
|
|
# Provider-side completion id when the gateway supplied one; joins a row to
|
|
# the upstream gateway's own request log.
|
|
provider_completion_id: Mapped[str | None] = mapped_column(String(128))
|
|
# LangGraph run id when the model layer could see one ("none" is dropped).
|
|
run_id: Mapped[str | None] = mapped_column(String(64), index=True)
|
|
|
|
# silent_no_terminal / incomplete_stream_error / length_truncated /
|
|
# content_filter / midstream_error
|
|
event_type: Mapped[str] = mapped_column(String(32), index=True)
|
|
# Whether the model ran with strict_stream_integrity enabled.
|
|
strict_mode: Mapped[bool] = mapped_column(Boolean, default=False)
|
|
|
|
# Raw provider finish reason bucket: none / stop / length / tool_calls /
|
|
# function_call / content_filter / end_turn / other.
|
|
finish_reason: Mapped[str] = mapped_column(String(16), default="none")
|
|
# Stream outcome from the model layer: success / error / cancelled.
|
|
status: Mapped[str] = mapped_column(String(16), default="success")
|
|
error_type: Mapped[str | None] = mapped_column(String(64))
|
|
|
|
chunk_count: Mapped[int] = mapped_column(Integer, default=0)
|
|
duration_ms: Mapped[int] = mapped_column(Integer, default=0)
|
|
first_chunk_ms: Mapped[int] = mapped_column(Integer, default=-1)
|
|
|
|
detail: Mapped[str | None] = mapped_column(Text)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"id": self.id,
|
|
"created_at": self.created_at.isoformat()
|
|
if isinstance(self.created_at, datetime)
|
|
else self.created_at,
|
|
"model_name": self.model_name,
|
|
"provider_completion_id": self.provider_completion_id,
|
|
"run_id": self.run_id,
|
|
"event_type": self.event_type,
|
|
"strict_mode": bool(self.strict_mode),
|
|
"finish_reason": self.finish_reason,
|
|
"status": self.status,
|
|
"error_type": self.error_type,
|
|
"chunk_count": self.chunk_count,
|
|
"duration_ms": self.duration_ms,
|
|
"first_chunk_ms": self.first_chunk_ms,
|
|
"detail": self.detail,
|
|
}
|