mirror of
https://github.com/langgenius/dify.git
synced 2026-07-26 06:08:38 +08:00
1157 lines
49 KiB
Python
1157 lines
49 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
import sqlalchemy as sa
|
|
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.orm import aliased
|
|
|
|
from configs import dify_config
|
|
from core.app.entities.app_invoke_entities import InvokeFrom
|
|
from graphon.enums import WorkflowNodeExecutionStatus
|
|
from libs.helper import convert_datetime_to_date, escape_like_pattern, to_timestamp
|
|
from models.agent import WorkflowAgentNodeBinding
|
|
from models.enums import CreatorUserRole, MessageStatus
|
|
from models.model import App, Conversation, Message
|
|
from models.workflow import WorkflowNodeExecutionModel, WorkflowRun, WorkflowType
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AgentLogQueryParams:
|
|
page: int = 1
|
|
limit: int = 20
|
|
keyword: str | None = None
|
|
statuses: tuple[str, ...] = ()
|
|
sources: tuple[str, ...] = ()
|
|
sort_by: str = "updated_at"
|
|
sort_order: str = "desc"
|
|
start: datetime | None = None
|
|
end: datetime | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AgentStatisticsQueryParams:
|
|
source: str | None = None
|
|
start: datetime | None = None
|
|
end: datetime | None = None
|
|
timezone: str = "UTC"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AgentSourceFilter:
|
|
kind: str
|
|
app_id: str | None = None
|
|
workflow_id: str | None = None
|
|
workflow_version: str | None = None
|
|
node_id: str | None = None
|
|
invoke_from: InvokeFrom | None = None
|
|
|
|
|
|
class AgentObservabilityService:
|
|
_SOURCE_ALIASES: dict[str, InvokeFrom] = {
|
|
"api": InvokeFrom.SERVICE_API,
|
|
"service-api": InvokeFrom.SERVICE_API,
|
|
"service_api": InvokeFrom.SERVICE_API,
|
|
"console": InvokeFrom.EXPLORE,
|
|
"explore": InvokeFrom.EXPLORE,
|
|
"explore-app": InvokeFrom.EXPLORE,
|
|
"explore_app": InvokeFrom.EXPLORE,
|
|
"web": InvokeFrom.WEB_APP,
|
|
"web-app": InvokeFrom.WEB_APP,
|
|
"web_app": InvokeFrom.WEB_APP,
|
|
"debugger": InvokeFrom.DEBUGGER,
|
|
"dev": InvokeFrom.DEBUGGER,
|
|
"openapi": InvokeFrom.OPENAPI,
|
|
"trigger": InvokeFrom.TRIGGER,
|
|
}
|
|
|
|
def __init__(self, session: Any):
|
|
self._session = session
|
|
|
|
@classmethod
|
|
def resolve_source(cls, source: str | None) -> InvokeFrom | None:
|
|
if not source or source == "all":
|
|
return None
|
|
normalized = source.strip().lower()
|
|
if not normalized or normalized == "all":
|
|
return None
|
|
try:
|
|
return cls._SOURCE_ALIASES[normalized]
|
|
except KeyError as exc:
|
|
raise ValueError(f"Unsupported source: {source}") from exc
|
|
|
|
@classmethod
|
|
def resolve_source_filter(cls, source: str | None) -> AgentSourceFilter:
|
|
if not source or source.strip().lower() == "all":
|
|
return AgentSourceFilter(kind="all")
|
|
normalized = source.strip()
|
|
lowered = normalized.lower()
|
|
if lowered == "webapp":
|
|
return AgentSourceFilter(kind="webapp")
|
|
if lowered.startswith("webapp:"):
|
|
return AgentSourceFilter(kind="webapp", app_id=normalized.split(":", 1)[1] or None)
|
|
if lowered == "workflow":
|
|
return AgentSourceFilter(kind="workflow")
|
|
if lowered.startswith("workflow:"):
|
|
parts = normalized.split(":")
|
|
if len(parts) == 2 and parts[1]:
|
|
return AgentSourceFilter(kind="workflow", app_id=parts[1])
|
|
if len(parts) < 5 or not all(parts[1:]):
|
|
raise ValueError(f"Unsupported source: {source}")
|
|
return AgentSourceFilter(
|
|
kind="workflow",
|
|
app_id=parts[1],
|
|
workflow_id=parts[2],
|
|
workflow_version=":".join(parts[3:-1]),
|
|
node_id=parts[-1],
|
|
)
|
|
return AgentSourceFilter(kind="webapp", invoke_from=cls.resolve_source(source))
|
|
|
|
@classmethod
|
|
def resolve_source_filters(cls, sources: tuple[str, ...]) -> list[AgentSourceFilter]:
|
|
if not sources:
|
|
return [AgentSourceFilter(kind="all")]
|
|
filters: list[AgentSourceFilter] = []
|
|
for source in sources:
|
|
source_filter = cls.resolve_source_filter(source)
|
|
if source_filter.kind == "all":
|
|
return [source_filter]
|
|
filters.append(source_filter)
|
|
return filters
|
|
|
|
@staticmethod
|
|
def _message_status(message: Message) -> str:
|
|
if message.error or message.status == MessageStatus.ERROR:
|
|
return "failed"
|
|
if message.status == MessageStatus.PAUSED:
|
|
return "paused"
|
|
return "success"
|
|
|
|
@staticmethod
|
|
def _total_tokens(message: Message) -> int:
|
|
return int(message.message_tokens or 0) + int(message.answer_tokens or 0)
|
|
|
|
@classmethod
|
|
def serialize_log_message(cls, message: Message, conversation: Conversation | None = None) -> dict[str, Any]:
|
|
invoke_from = message.invoke_from.value if message.invoke_from else None
|
|
return {
|
|
"id": message.id,
|
|
"message_id": message.id,
|
|
"conversation_id": message.conversation_id,
|
|
"conversation_name": conversation.name if conversation else None,
|
|
"query": message.query,
|
|
"answer": message.answer,
|
|
"status": cls._message_status(message),
|
|
"error": message.error,
|
|
"source": invoke_from,
|
|
"from_source": message.from_source.value if message.from_source else None,
|
|
"from_end_user_id": message.from_end_user_id,
|
|
"from_account_id": message.from_account_id,
|
|
"message_tokens": int(message.message_tokens or 0),
|
|
"answer_tokens": int(message.answer_tokens or 0),
|
|
"total_tokens": cls._total_tokens(message),
|
|
"total_price": str(message.total_price or Decimal(0)),
|
|
"currency": message.currency,
|
|
"latency": float(message.provider_response_latency or 0),
|
|
"created_at": to_timestamp(message.created_at),
|
|
"updated_at": to_timestamp(message.updated_at),
|
|
}
|
|
|
|
def list_logs(self, *, app: App, agent_id: str, params: AgentLogQueryParams) -> dict[str, Any]:
|
|
source_filters = self.resolve_source_filters(params.sources)
|
|
rows: list[dict[str, Any]] = []
|
|
for source_filter in source_filters:
|
|
if source_filter.kind in {"all", "webapp"}:
|
|
rows.extend(self._list_webapp_conversation_logs(app=app, params=params, source_filter=source_filter))
|
|
if source_filter.kind in {"all", "workflow"}:
|
|
rows.extend(
|
|
self._list_workflow_conversation_logs(
|
|
app=app,
|
|
agent_id=agent_id,
|
|
params=params,
|
|
source_filter=source_filter,
|
|
)
|
|
)
|
|
rows_by_scope = {(row["id"], row["source"]["id"] if row.get("source") else ""): row for row in rows}
|
|
rows = list(rows_by_scope.values())
|
|
sort_by = "created_at" if params.sort_by == "created_at" else "updated_at"
|
|
rows.sort(key=lambda row: (row[sort_by] or 0, row["id"]), reverse=params.sort_order != "asc")
|
|
|
|
total = len(rows)
|
|
start = (params.page - 1) * params.limit
|
|
end = start + params.limit
|
|
return {
|
|
"data": rows[start:end],
|
|
"page": params.page,
|
|
"limit": params.limit,
|
|
"total": total,
|
|
"has_more": end < total,
|
|
}
|
|
|
|
def list_log_messages(
|
|
self, *, app: App, agent_id: str, conversation_id: str, params: AgentLogQueryParams
|
|
) -> dict[str, Any]:
|
|
source_filters = self.resolve_source_filters(params.sources)
|
|
rows: list[dict[str, Any]] = []
|
|
for source_filter in source_filters:
|
|
if source_filter.kind in {"all", "webapp"}:
|
|
rows.extend(
|
|
self.serialize_log_message(message)
|
|
for message in self._list_webapp_messages(
|
|
app=app,
|
|
conversation_id=conversation_id,
|
|
params=params,
|
|
source_filter=source_filter,
|
|
)
|
|
)
|
|
if source_filter.kind in {"all", "workflow"}:
|
|
rows.extend(
|
|
self._list_workflow_messages(
|
|
app=app,
|
|
agent_id=agent_id,
|
|
conversation_id=conversation_id,
|
|
params=params,
|
|
source_filter=source_filter,
|
|
)
|
|
)
|
|
|
|
deduped = {row["id"]: row for row in rows}
|
|
sort_key = "created_at" if params.sort_by == "created_at" else "updated_at"
|
|
sorted_rows = sorted(
|
|
deduped.values(),
|
|
key=lambda row: (row[sort_key] or 0, row["id"]),
|
|
reverse=params.sort_order != "asc",
|
|
)
|
|
total = len(sorted_rows)
|
|
start = (params.page - 1) * params.limit
|
|
end = start + params.limit
|
|
return {
|
|
"data": sorted_rows[start:end],
|
|
"page": params.page,
|
|
"limit": params.limit,
|
|
"total": total,
|
|
"has_more": end < total,
|
|
}
|
|
|
|
def list_log_sources(self, *, app: App, agent_id: str) -> dict[str, Any]:
|
|
webapp_source = self._serialize_webapp_source(app)
|
|
workflow_sources = self._list_workflow_sources(app=app, agent_id=agent_id)
|
|
return {
|
|
"data": [webapp_source, *workflow_sources],
|
|
"groups": [
|
|
{"type": "webapp", "label": "WEBAPP", "sources": [webapp_source]},
|
|
{"type": "workflow", "label": "WORKFLOW", "sources": workflow_sources},
|
|
],
|
|
}
|
|
|
|
def _list_webapp_conversation_logs(
|
|
self, *, app: App, params: AgentLogQueryParams, source_filter: AgentSourceFilter
|
|
) -> list[dict[str, Any]]:
|
|
stmt = (
|
|
select(
|
|
Conversation,
|
|
func.count(Message.id).label("message_count"),
|
|
func.max(Message.created_at).label("created_at"),
|
|
func.max(Message.updated_at).label("updated_at"),
|
|
func.sum(sa.case((Message.status == MessageStatus.PAUSED, 1), else_=0)).label("paused_count"),
|
|
func.sum(
|
|
sa.case((or_(Message.error.is_not(None), Message.status == MessageStatus.ERROR), 1), else_=0)
|
|
).label("failed_count"),
|
|
)
|
|
.join(Message, Message.conversation_id == Conversation.id)
|
|
.where(Message.app_id == app.id, Conversation.app_id == app.id)
|
|
.group_by(Conversation.id)
|
|
)
|
|
stmt = self._apply_observability_filters(stmt, params=params, source_filter=source_filter)
|
|
rows = list(self._session.execute(stmt).all())
|
|
return [
|
|
self._serialize_conversation_log(
|
|
conversation=row[0],
|
|
message_count=row.message_count,
|
|
paused_count=row.paused_count,
|
|
failed_count=row.failed_count,
|
|
source=self._serialize_webapp_source(app),
|
|
created_at=row.created_at,
|
|
updated_at=row.updated_at,
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
def _list_workflow_conversation_logs(
|
|
self, *, app: App, agent_id: str, params: AgentLogQueryParams, source_filter: AgentSourceFilter
|
|
) -> list[dict[str, Any]]:
|
|
workflow_app = aliased(App)
|
|
stmt = (
|
|
select(
|
|
WorkflowNodeExecutionModel.id.label("node_execution_id"),
|
|
WorkflowNodeExecutionModel.title.label("node_title"),
|
|
WorkflowNodeExecutionModel.status.label("node_status"),
|
|
WorkflowNodeExecutionModel.created_by_role.label("node_created_by_role"),
|
|
WorkflowNodeExecutionModel.created_by.label("node_created_by"),
|
|
WorkflowNodeExecutionModel.created_at.label("node_created_at"),
|
|
WorkflowNodeExecutionModel.finished_at.label("node_finished_at"),
|
|
workflow_app,
|
|
WorkflowAgentNodeBinding.workflow_id,
|
|
WorkflowAgentNodeBinding.workflow_version,
|
|
WorkflowAgentNodeBinding.node_id,
|
|
)
|
|
.select_from(WorkflowNodeExecutionModel)
|
|
.join(WorkflowRun, WorkflowRun.id == WorkflowNodeExecutionModel.workflow_run_id)
|
|
.join(
|
|
WorkflowAgentNodeBinding,
|
|
and_(
|
|
WorkflowAgentNodeBinding.tenant_id == app.tenant_id,
|
|
WorkflowAgentNodeBinding.agent_id == agent_id,
|
|
WorkflowAgentNodeBinding.app_id == WorkflowRun.app_id,
|
|
WorkflowAgentNodeBinding.workflow_id == WorkflowRun.workflow_id,
|
|
WorkflowAgentNodeBinding.workflow_version == WorkflowRun.version,
|
|
),
|
|
)
|
|
.join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id)
|
|
.where(
|
|
WorkflowNodeExecutionModel.tenant_id == app.tenant_id,
|
|
WorkflowNodeExecutionModel.app_id == WorkflowAgentNodeBinding.app_id,
|
|
WorkflowNodeExecutionModel.workflow_id == WorkflowAgentNodeBinding.workflow_id,
|
|
WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id,
|
|
)
|
|
)
|
|
stmt = self._apply_workflow_node_filters(stmt, params=params, workflow_app=workflow_app)
|
|
stmt = self._apply_workflow_source_filter(stmt, source_filter)
|
|
rows = list(self._session.execute(stmt).all())
|
|
return [
|
|
self._serialize_workflow_execution_log(
|
|
node_execution_id=row.node_execution_id,
|
|
title=row.node_title,
|
|
status=row.node_status,
|
|
created_by_role=row.node_created_by_role,
|
|
created_by=row.node_created_by,
|
|
created_at=row.node_created_at,
|
|
finished_at=row.node_finished_at,
|
|
source=self._serialize_workflow_source(
|
|
app=row[7],
|
|
workflow_id=row.workflow_id,
|
|
workflow_version=row.workflow_version,
|
|
node_id=row.node_id,
|
|
),
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
def _list_webapp_messages(
|
|
self, *, app: App, conversation_id: str, params: AgentLogQueryParams, source_filter: AgentSourceFilter
|
|
) -> list[Message]:
|
|
stmt = select(Message).where(Message.app_id == app.id, Message.conversation_id == conversation_id)
|
|
stmt = self._apply_message_filters(stmt, params=params, source_filter=source_filter)
|
|
return list(self._session.scalars(stmt.order_by(Message.created_at.desc(), Message.id.desc())).all())
|
|
|
|
def _list_workflow_messages(
|
|
self,
|
|
*,
|
|
app: App,
|
|
agent_id: str,
|
|
conversation_id: str,
|
|
params: AgentLogQueryParams,
|
|
source_filter: AgentSourceFilter,
|
|
) -> list[dict[str, Any]]:
|
|
workflow_app = aliased(App)
|
|
stmt = (
|
|
select(WorkflowNodeExecutionModel)
|
|
.select_from(WorkflowNodeExecutionModel)
|
|
.join(WorkflowRun, WorkflowRun.id == WorkflowNodeExecutionModel.workflow_run_id)
|
|
.join(
|
|
WorkflowAgentNodeBinding,
|
|
and_(
|
|
WorkflowAgentNodeBinding.tenant_id == app.tenant_id,
|
|
WorkflowAgentNodeBinding.agent_id == agent_id,
|
|
WorkflowAgentNodeBinding.app_id == WorkflowRun.app_id,
|
|
WorkflowAgentNodeBinding.workflow_id == WorkflowRun.workflow_id,
|
|
WorkflowAgentNodeBinding.workflow_version == WorkflowRun.version,
|
|
),
|
|
)
|
|
.join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id)
|
|
.where(
|
|
WorkflowNodeExecutionModel.id == conversation_id,
|
|
WorkflowNodeExecutionModel.tenant_id == app.tenant_id,
|
|
WorkflowNodeExecutionModel.app_id == WorkflowAgentNodeBinding.app_id,
|
|
WorkflowNodeExecutionModel.workflow_id == WorkflowAgentNodeBinding.workflow_id,
|
|
WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id,
|
|
)
|
|
)
|
|
stmt = self._apply_workflow_node_filters(stmt, params=params, workflow_app=workflow_app)
|
|
stmt = self._apply_workflow_source_filter(stmt, source_filter)
|
|
executions = list(
|
|
self._session.scalars(
|
|
stmt.order_by(WorkflowNodeExecutionModel.created_at.desc(), WorkflowNodeExecutionModel.id.desc())
|
|
).all()
|
|
)
|
|
return [self.serialize_workflow_node_message(execution) for execution in executions]
|
|
|
|
def _list_workflow_sources(self, *, app: App, agent_id: str) -> list[dict[str, Any]]:
|
|
workflow_app = aliased(App)
|
|
stmt = (
|
|
select(workflow_app)
|
|
.select_from(WorkflowAgentNodeBinding)
|
|
.join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id)
|
|
.where(WorkflowAgentNodeBinding.tenant_id == app.tenant_id, WorkflowAgentNodeBinding.agent_id == agent_id)
|
|
.order_by(workflow_app.name.asc(), workflow_app.id.asc())
|
|
)
|
|
rows = self._session.execute(stmt).all()
|
|
deduped: dict[str, dict[str, Any]] = {}
|
|
for row in rows:
|
|
source_app = row[0]
|
|
deduped[source_app.id] = self._serialize_workflow_app_source(app=source_app)
|
|
return list(deduped.values())
|
|
|
|
@classmethod
|
|
def _apply_observability_filters(cls, stmt, *, params: AgentLogQueryParams, source_filter: AgentSourceFilter):
|
|
stmt = cls._apply_message_filters(stmt, params=params, source_filter=source_filter, include_keyword=False)
|
|
if params.keyword:
|
|
escaped_keyword = escape_like_pattern(params.keyword)
|
|
pattern = f"%{escaped_keyword}%"
|
|
stmt = stmt.where(
|
|
or_(
|
|
Message.query.ilike(pattern, escape="\\"),
|
|
Message.answer.ilike(pattern, escape="\\"),
|
|
Conversation.name.ilike(pattern, escape="\\"),
|
|
)
|
|
)
|
|
return stmt
|
|
|
|
@classmethod
|
|
def _apply_message_filters(
|
|
cls, stmt, *, params: AgentLogQueryParams, source_filter: AgentSourceFilter, include_keyword: bool = True
|
|
):
|
|
stmt = cls._apply_source_filter(stmt, source_filter.invoke_from)
|
|
if params.start:
|
|
stmt = stmt.where(Message.created_at >= params.start)
|
|
if params.end:
|
|
stmt = stmt.where(Message.created_at < params.end)
|
|
if include_keyword and params.keyword:
|
|
escaped_keyword = escape_like_pattern(params.keyword)
|
|
pattern = f"%{escaped_keyword}%"
|
|
stmt = stmt.where(
|
|
or_(
|
|
Message.query.ilike(pattern, escape="\\"),
|
|
Message.answer.ilike(pattern, escape="\\"),
|
|
)
|
|
)
|
|
if params.statuses:
|
|
stmt = cls._apply_status_filter(stmt, params.statuses)
|
|
return stmt
|
|
|
|
@classmethod
|
|
def _apply_workflow_node_filters(cls, stmt, *, params: AgentLogQueryParams, workflow_app):
|
|
if params.start:
|
|
stmt = stmt.where(WorkflowNodeExecutionModel.created_at >= params.start)
|
|
if params.end:
|
|
stmt = stmt.where(WorkflowNodeExecutionModel.created_at < params.end)
|
|
if params.keyword:
|
|
escaped_keyword = escape_like_pattern(params.keyword)
|
|
pattern = f"%{escaped_keyword}%"
|
|
stmt = stmt.where(
|
|
or_(
|
|
WorkflowNodeExecutionModel.inputs.ilike(pattern, escape="\\"),
|
|
WorkflowNodeExecutionModel.outputs.ilike(pattern, escape="\\"),
|
|
WorkflowNodeExecutionModel.error.ilike(pattern, escape="\\"),
|
|
WorkflowNodeExecutionModel.title.ilike(pattern, escape="\\"),
|
|
workflow_app.name.ilike(pattern, escape="\\"),
|
|
)
|
|
)
|
|
if params.statuses:
|
|
stmt = cls._apply_workflow_node_status_filter(stmt, params.statuses)
|
|
return stmt
|
|
|
|
@staticmethod
|
|
def _apply_workflow_node_status_filter(stmt, statuses: tuple[str, ...]):
|
|
conditions = []
|
|
for status in statuses:
|
|
normalized = status.strip().lower()
|
|
if normalized in {"success", "normal"}:
|
|
conditions.append(WorkflowNodeExecutionModel.status == WorkflowNodeExecutionStatus.SUCCEEDED)
|
|
elif normalized in {"failed", "error"}:
|
|
conditions.append(
|
|
WorkflowNodeExecutionModel.status.in_(
|
|
(
|
|
WorkflowNodeExecutionStatus.FAILED,
|
|
WorkflowNodeExecutionStatus.EXCEPTION,
|
|
WorkflowNodeExecutionStatus.STOPPED,
|
|
)
|
|
)
|
|
)
|
|
elif normalized == "paused":
|
|
conditions.append(
|
|
WorkflowNodeExecutionModel.status.in_(
|
|
(
|
|
WorkflowNodeExecutionStatus.PAUSED,
|
|
WorkflowNodeExecutionStatus.PENDING,
|
|
WorkflowNodeExecutionStatus.RUNNING,
|
|
WorkflowNodeExecutionStatus.RETRY,
|
|
)
|
|
)
|
|
)
|
|
else:
|
|
raise ValueError(f"Unsupported status: {status}")
|
|
if not conditions:
|
|
return stmt
|
|
return stmt.where(or_(*conditions))
|
|
|
|
@staticmethod
|
|
def _apply_workflow_source_filter(stmt, source_filter: AgentSourceFilter):
|
|
if source_filter.app_id:
|
|
stmt = stmt.where(WorkflowAgentNodeBinding.app_id == source_filter.app_id)
|
|
if source_filter.workflow_id:
|
|
stmt = stmt.where(WorkflowAgentNodeBinding.workflow_id == source_filter.workflow_id)
|
|
if source_filter.workflow_version:
|
|
stmt = stmt.where(WorkflowAgentNodeBinding.workflow_version == source_filter.workflow_version)
|
|
if source_filter.node_id:
|
|
stmt = stmt.where(WorkflowAgentNodeBinding.node_id == source_filter.node_id)
|
|
return stmt
|
|
|
|
@classmethod
|
|
def _apply_source_filter(cls, stmt, source: InvokeFrom | None):
|
|
if source is None:
|
|
return stmt.where(Message.invoke_from != InvokeFrom.DEBUGGER)
|
|
return stmt.where(Message.invoke_from == source)
|
|
|
|
@staticmethod
|
|
def _apply_status_filter(stmt, statuses: tuple[str, ...]):
|
|
conditions = []
|
|
for status in statuses:
|
|
normalized = status.strip().lower()
|
|
if normalized in {"success", "normal"}:
|
|
conditions.append(and_(Message.error.is_(None), Message.status == MessageStatus.NORMAL))
|
|
elif normalized in {"failed", "error"}:
|
|
conditions.append(or_(Message.error.is_not(None), Message.status == MessageStatus.ERROR))
|
|
elif normalized == "paused":
|
|
conditions.append(Message.status == MessageStatus.PAUSED)
|
|
else:
|
|
raise ValueError(f"Unsupported status: {status}")
|
|
if not conditions:
|
|
return stmt
|
|
return stmt.where(or_(*conditions))
|
|
|
|
@classmethod
|
|
def _serialize_conversation_log(
|
|
cls,
|
|
*,
|
|
conversation: Conversation,
|
|
message_count: int,
|
|
paused_count: int,
|
|
failed_count: int,
|
|
source: dict[str, Any],
|
|
created_at: datetime | None,
|
|
updated_at: datetime | None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"id": conversation.id,
|
|
"conversation_id": conversation.id,
|
|
"title": conversation.name,
|
|
"end_user_id": conversation.from_end_user_id,
|
|
"message_count": int(message_count or 0),
|
|
"user_rate": None,
|
|
"operation_rate": None,
|
|
"unread": conversation.read_at is None,
|
|
"source": source,
|
|
"status": cls._conversation_status(paused_count=paused_count, failed_count=failed_count),
|
|
"created_at": to_timestamp(created_at or conversation.created_at),
|
|
"updated_at": to_timestamp(updated_at or conversation.updated_at),
|
|
}
|
|
|
|
@classmethod
|
|
def _serialize_workflow_execution_log(
|
|
cls,
|
|
*,
|
|
node_execution_id: str,
|
|
title: str,
|
|
status: object,
|
|
created_by_role: object,
|
|
created_by: str,
|
|
created_at: datetime,
|
|
finished_at: datetime | None,
|
|
source: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
created_by_role_value = cls._enum_value(created_by_role)
|
|
return {
|
|
"id": node_execution_id,
|
|
"conversation_id": node_execution_id,
|
|
"title": title,
|
|
"end_user_id": created_by if created_by_role_value == CreatorUserRole.END_USER.value else None,
|
|
"message_count": 1,
|
|
"user_rate": None,
|
|
"operation_rate": None,
|
|
"unread": False,
|
|
"source": source,
|
|
"status": cls._workflow_node_status(status),
|
|
"created_at": to_timestamp(created_at),
|
|
"updated_at": to_timestamp(finished_at or created_at),
|
|
}
|
|
|
|
@classmethod
|
|
def serialize_workflow_node_message(cls, node_execution: WorkflowNodeExecutionModel) -> dict[str, Any]:
|
|
inputs = cls._json_mapping(node_execution.inputs)
|
|
outputs = cls._json_mapping(node_execution.outputs)
|
|
metadata = cls._json_mapping(node_execution.execution_metadata)
|
|
agent_log = cls._mapping_value(metadata, "agent_log")
|
|
agent_backend = cls._mapping_value(agent_log, "agent_backend")
|
|
usage = cls._mapping_value(agent_backend, "usage")
|
|
|
|
prompt_tokens = cls._int_value(usage.get("prompt_tokens"))
|
|
completion_tokens = cls._int_value(usage.get("completion_tokens"))
|
|
total_tokens = cls._int_value(usage.get("total_tokens") or metadata.get("total_tokens"))
|
|
if not total_tokens:
|
|
total_tokens = prompt_tokens + completion_tokens
|
|
created_by_role = cls._enum_value(node_execution.created_by_role)
|
|
|
|
return {
|
|
"id": node_execution.id,
|
|
"message_id": node_execution.id,
|
|
"conversation_id": node_execution.id,
|
|
"query": cls._workflow_node_query(inputs, fallback=node_execution.title),
|
|
"answer": cls._workflow_node_answer(outputs),
|
|
"status": cls._workflow_node_status(node_execution.status),
|
|
"error": node_execution.error,
|
|
"from_end_user_id": (
|
|
node_execution.created_by if created_by_role == CreatorUserRole.END_USER.value else None
|
|
),
|
|
"from_account_id": (
|
|
node_execution.created_by if created_by_role == CreatorUserRole.ACCOUNT.value else None
|
|
),
|
|
"message_tokens": prompt_tokens,
|
|
"answer_tokens": completion_tokens,
|
|
"total_tokens": total_tokens,
|
|
"total_price": str(usage.get("total_price") or metadata.get("total_price") or Decimal(0)),
|
|
"currency": str(usage.get("currency") or metadata.get("currency") or ""),
|
|
"latency": float(usage.get("latency") or node_execution.elapsed_time or 0),
|
|
"created_at": to_timestamp(node_execution.created_at),
|
|
"updated_at": to_timestamp(node_execution.finished_at or node_execution.created_at),
|
|
}
|
|
|
|
@staticmethod
|
|
def _json_mapping(value: object) -> Mapping[str, Any]:
|
|
if isinstance(value, Mapping):
|
|
return value
|
|
if not isinstance(value, str) or not value:
|
|
return {}
|
|
try:
|
|
parsed = json.loads(value)
|
|
except (TypeError, ValueError):
|
|
return {}
|
|
return parsed if isinstance(parsed, Mapping) else {}
|
|
|
|
@staticmethod
|
|
def _mapping_value(value: Mapping[str, Any], key: str) -> Mapping[str, Any]:
|
|
nested = value.get(key)
|
|
return nested if isinstance(nested, Mapping) else {}
|
|
|
|
@staticmethod
|
|
def _enum_value(value: object) -> str:
|
|
return str(value.value) if isinstance(value, Enum) else str(value)
|
|
|
|
@staticmethod
|
|
def _int_value(value: object) -> int:
|
|
if not isinstance(value, (str, int, float, Decimal)):
|
|
return 0
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
@classmethod
|
|
def _workflow_node_query(cls, inputs: Mapping[str, Any], *, fallback: str) -> str:
|
|
request_data = cls._mapping_value(inputs, "agent_backend_request")
|
|
composition = cls._mapping_value(request_data, "composition")
|
|
layers = composition.get("layers")
|
|
prompts: list[str] = []
|
|
if isinstance(layers, list):
|
|
for layer_name in ("workflow_node_job_prompt", "workflow_user_prompt"):
|
|
for layer in layers:
|
|
if not isinstance(layer, Mapping) or layer.get("name") != layer_name:
|
|
continue
|
|
config = cls._mapping_value(layer, "config")
|
|
user_prompt = config.get("user")
|
|
if isinstance(user_prompt, str) and user_prompt.strip():
|
|
prompts.append(user_prompt.strip())
|
|
return "\n\n".join(prompts) or fallback
|
|
|
|
@staticmethod
|
|
def _workflow_node_answer(outputs: Mapping[str, Any]) -> str:
|
|
for key in ("output", "text", "answer"):
|
|
value = outputs.get(key)
|
|
if isinstance(value, str):
|
|
return value
|
|
return json.dumps(outputs, ensure_ascii=False) if outputs else ""
|
|
|
|
@classmethod
|
|
def _workflow_node_status(cls, status: object) -> str:
|
|
value = cls._enum_value(status)
|
|
if value in {
|
|
WorkflowNodeExecutionStatus.FAILED.value,
|
|
WorkflowNodeExecutionStatus.EXCEPTION.value,
|
|
WorkflowNodeExecutionStatus.STOPPED.value,
|
|
}:
|
|
return "failed"
|
|
if value in {
|
|
WorkflowNodeExecutionStatus.PAUSED.value,
|
|
WorkflowNodeExecutionStatus.PENDING.value,
|
|
WorkflowNodeExecutionStatus.RUNNING.value,
|
|
WorkflowNodeExecutionStatus.RETRY.value,
|
|
}:
|
|
return "paused"
|
|
return "success"
|
|
|
|
@staticmethod
|
|
def _conversation_status(*, paused_count: int, failed_count: int) -> str:
|
|
if paused_count:
|
|
return "paused"
|
|
if failed_count:
|
|
return "failed"
|
|
return "success"
|
|
|
|
@staticmethod
|
|
def _serialize_webapp_source(app: App) -> dict[str, Any]:
|
|
icon_type = app.icon_type.value if app.icon_type else None
|
|
return {
|
|
"id": f"webapp:{app.id}",
|
|
"type": "webapp",
|
|
"app_id": app.id,
|
|
"app_name": app.name,
|
|
"app_icon_type": icon_type,
|
|
"app_icon": app.icon,
|
|
"app_icon_background": app.icon_background,
|
|
"workflow_id": None,
|
|
"workflow_version": None,
|
|
"node_id": None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _serialize_workflow_app_source(*, app: App) -> dict[str, Any]:
|
|
"""Serialize the app-level source used by log and monitoring filters."""
|
|
icon_type = app.icon_type.value if app.icon_type else None
|
|
return {
|
|
"id": f"workflow:{app.id}",
|
|
"type": "workflow",
|
|
"app_id": app.id,
|
|
"app_name": app.name,
|
|
"app_icon_type": icon_type,
|
|
"app_icon": app.icon,
|
|
"app_icon_background": app.icon_background,
|
|
"workflow_id": None,
|
|
"workflow_version": None,
|
|
"node_id": None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _serialize_workflow_source(
|
|
*,
|
|
app: App,
|
|
workflow_id: str,
|
|
workflow_version: str,
|
|
node_id: str,
|
|
) -> dict[str, Any]:
|
|
icon_type = app.icon_type.value if app.icon_type else None
|
|
return {
|
|
"id": f"workflow:{app.id}:{workflow_id}:{workflow_version}:{node_id}",
|
|
"type": "workflow",
|
|
"app_id": app.id,
|
|
"app_name": app.name,
|
|
"app_icon_type": icon_type,
|
|
"app_icon": app.icon,
|
|
"app_icon_background": app.icon_background,
|
|
"workflow_id": workflow_id,
|
|
"workflow_version": workflow_version,
|
|
"node_id": node_id,
|
|
}
|
|
|
|
def get_statistics_summary(self, *, app: App, agent_id: str, params: AgentStatisticsQueryParams) -> dict[str, Any]:
|
|
source_filter = self.resolve_source_filter(params.source)
|
|
rows = self._load_daily_statistics(app=app, agent_id=agent_id, params=params, source_filter=source_filter)
|
|
charts = self._build_charts(rows)
|
|
summary = self._build_summary(rows)
|
|
return {
|
|
"source": params.source or "all",
|
|
"summary": summary,
|
|
"charts": charts,
|
|
}
|
|
|
|
def _load_daily_statistics(
|
|
self, *, app: App, agent_id: str, params: AgentStatisticsQueryParams, source_filter: AgentSourceFilter
|
|
) -> list[dict[str, Any]]:
|
|
rows: list[dict[str, Any]] = []
|
|
if source_filter.kind in {"all", "webapp"}:
|
|
rows.extend(self._load_webapp_daily_statistics(app=app, params=params, source_filter=source_filter))
|
|
if source_filter.kind in {"all", "workflow"}:
|
|
rows.extend(
|
|
self._load_workflow_daily_statistics(
|
|
app=app,
|
|
agent_id=agent_id,
|
|
params=params,
|
|
source_filter=source_filter,
|
|
)
|
|
)
|
|
return self._merge_daily_statistics(rows)
|
|
|
|
def _load_webapp_daily_statistics(
|
|
self, *, app: App, params: AgentStatisticsQueryParams, source_filter: AgentSourceFilter
|
|
) -> list[dict[str, Any]]:
|
|
converted_created_at = convert_datetime_to_date("m.created_at")
|
|
message_scope = self._statistics_webapp_message_scope_sql(source_filter)
|
|
sql_query = f"""SELECT
|
|
{converted_created_at} AS date,
|
|
COUNT(m.id) AS message_count,
|
|
COUNT(DISTINCT m.conversation_id) AS conversation_count,
|
|
COUNT(DISTINCT m.from_end_user_id) AS end_user_count,
|
|
COALESCE(SUM(COALESCE(m.message_tokens, 0) + COALESCE(m.answer_tokens, 0)), 0) AS token_count,
|
|
COALESCE(SUM(COALESCE(m.total_price, 0)), 0) AS total_price,
|
|
COALESCE(AVG(m.provider_response_latency), 0) AS avg_latency,
|
|
COALESCE(SUM(m.provider_response_latency), 0) AS latency_sum,
|
|
COALESCE(SUM(m.answer_tokens), 0) AS answer_tokens,
|
|
COUNT(mf.id) AS like_count
|
|
FROM messages m
|
|
LEFT JOIN message_feedbacks mf
|
|
ON mf.message_id = m.id AND mf.rating = 'like'
|
|
WHERE
|
|
{message_scope}"""
|
|
args: dict[str, Any] = {
|
|
"tz": params.timezone,
|
|
"app_id": app.id,
|
|
}
|
|
if source_filter.invoke_from is not None:
|
|
args["source"] = source_filter.invoke_from
|
|
if params.start:
|
|
sql_query += " AND m.created_at >= :start"
|
|
args["start"] = params.start
|
|
if params.end:
|
|
sql_query += " AND m.created_at < :end"
|
|
args["end"] = params.end
|
|
sql_query += " GROUP BY date ORDER BY date"
|
|
|
|
return [dict(row._mapping) for row in self._session.execute(sa.text(sql_query), args).all()]
|
|
|
|
@staticmethod
|
|
def _statistics_webapp_message_scope_sql(source_filter: AgentSourceFilter) -> str:
|
|
app_scope = "m.app_id = :app_id"
|
|
if source_filter.invoke_from is not None:
|
|
app_scope += " AND m.invoke_from = :source"
|
|
return app_scope
|
|
|
|
def _load_workflow_daily_statistics(
|
|
self,
|
|
*,
|
|
app: App,
|
|
agent_id: str,
|
|
params: AgentStatisticsQueryParams,
|
|
source_filter: AgentSourceFilter,
|
|
) -> list[dict[str, Any]]:
|
|
converted_run_created_at = convert_datetime_to_date("aru.created_at")
|
|
total_tokens = self._workflow_execution_metadata_numeric_sql(("total_tokens",), "BIGINT")
|
|
nested_total_tokens = self._workflow_execution_metadata_numeric_sql(
|
|
("agent_log", "agent_backend", "usage", "total_tokens"), "BIGINT"
|
|
)
|
|
total_price = self._workflow_execution_metadata_numeric_sql(("total_price",), "DECIMAL(65, 30)")
|
|
nested_total_price = self._workflow_execution_metadata_numeric_sql(
|
|
("agent_log", "agent_backend", "usage", "total_price"), "DECIMAL(65, 30)"
|
|
)
|
|
completion_tokens = self._workflow_execution_metadata_numeric_sql(
|
|
("agent_log", "agent_backend", "usage", "completion_tokens"), "BIGINT"
|
|
)
|
|
binding_filters = self._statistics_workflow_binding_filters_sql(source_filter)
|
|
run_date_filters = ""
|
|
args: dict[str, Any] = {
|
|
"tz": params.timezone,
|
|
"tenant_id": app.tenant_id,
|
|
"agent_id": agent_id,
|
|
"chat_workflow_type": WorkflowType.CHAT,
|
|
"end_user_role": CreatorUserRole.END_USER,
|
|
}
|
|
if source_filter.app_id:
|
|
args["source_app_id"] = source_filter.app_id
|
|
if source_filter.workflow_id:
|
|
args["workflow_id"] = source_filter.workflow_id
|
|
if source_filter.workflow_version:
|
|
args["workflow_version"] = source_filter.workflow_version
|
|
if source_filter.node_id:
|
|
args["node_id"] = source_filter.node_id
|
|
if params.start:
|
|
run_date_filters += " AND wr.created_at >= :start"
|
|
args["start"] = params.start
|
|
if params.end:
|
|
run_date_filters += " AND wr.created_at < :end"
|
|
args["end"] = params.end
|
|
|
|
run_query = f"""WITH agent_run_usage AS (
|
|
SELECT
|
|
wr.id,
|
|
wr.created_by_role,
|
|
wr.created_by,
|
|
wr.created_at,
|
|
COALESCE(SUM(COALESCE({total_tokens}, {nested_total_tokens}, 0)), 0) AS token_count,
|
|
COALESCE(SUM(COALESCE({total_price}, {nested_total_price}, 0)), 0) AS total_price,
|
|
COALESCE(SUM(COALESCE(wne.elapsed_time, 0)), 0) AS latency,
|
|
COALESCE(SUM(COALESCE({completion_tokens}, 0)), 0) AS answer_tokens
|
|
FROM workflow_runs wr
|
|
JOIN workflow_agent_node_bindings wanb
|
|
ON wanb.tenant_id = :tenant_id
|
|
AND wanb.agent_id = :agent_id
|
|
AND wanb.app_id = wr.app_id
|
|
AND wanb.workflow_id = wr.workflow_id
|
|
AND wanb.workflow_version = wr.version
|
|
{binding_filters}
|
|
JOIN workflow_node_executions wne
|
|
ON wne.workflow_run_id = wr.id
|
|
AND wne.node_id = wanb.node_id
|
|
WHERE wr.type != :chat_workflow_type{run_date_filters}
|
|
GROUP BY wr.id, wr.created_by_role, wr.created_by, wr.created_at
|
|
)
|
|
SELECT
|
|
{converted_run_created_at} AS date,
|
|
COUNT(aru.id) AS message_count,
|
|
COUNT(aru.id) AS conversation_count,
|
|
COUNT(DISTINCT CASE
|
|
WHEN aru.created_by_role = :end_user_role THEN aru.created_by
|
|
ELSE NULL
|
|
END) AS end_user_count,
|
|
COALESCE(SUM(aru.token_count), 0) AS token_count,
|
|
COALESCE(SUM(aru.total_price), 0) AS total_price,
|
|
COALESCE(AVG(aru.latency), 0) AS avg_latency,
|
|
COALESCE(SUM(aru.latency), 0) AS latency_sum,
|
|
COALESCE(SUM(aru.answer_tokens), 0) AS answer_tokens,
|
|
0 AS like_count
|
|
FROM agent_run_usage aru
|
|
GROUP BY date
|
|
ORDER BY date"""
|
|
rows = [dict(row._mapping) for row in self._session.execute(sa.text(run_query), args).all()]
|
|
rows.extend(
|
|
self._load_workflow_chat_daily_context(
|
|
app=app,
|
|
agent_id=agent_id,
|
|
params=params,
|
|
source_filter=source_filter,
|
|
)
|
|
)
|
|
return self._merge_daily_statistics(rows)
|
|
|
|
def _load_workflow_chat_daily_context(
|
|
self,
|
|
*,
|
|
app: App,
|
|
agent_id: str,
|
|
params: AgentStatisticsQueryParams,
|
|
source_filter: AgentSourceFilter,
|
|
) -> list[dict[str, Any]]:
|
|
converted_created_at = convert_datetime_to_date("m.created_at")
|
|
workflow_scope = self._statistics_workflow_message_scope_sql(source_filter)
|
|
sql_query = f"""SELECT
|
|
{converted_created_at} AS date,
|
|
COUNT(m.id) AS message_count,
|
|
COUNT(DISTINCT m.conversation_id) AS conversation_count,
|
|
COUNT(DISTINCT m.from_end_user_id) AS end_user_count,
|
|
COALESCE(SUM(COALESCE(m.message_tokens, 0) + COALESCE(m.answer_tokens, 0)), 0) AS token_count,
|
|
COALESCE(SUM(COALESCE(m.total_price, 0)), 0) AS total_price,
|
|
COALESCE(AVG(m.provider_response_latency), 0) AS avg_latency,
|
|
COALESCE(SUM(m.provider_response_latency), 0) AS latency_sum,
|
|
COALESCE(SUM(m.answer_tokens), 0) AS answer_tokens,
|
|
COUNT(mf.id) AS like_count
|
|
FROM messages m
|
|
LEFT JOIN message_feedbacks mf
|
|
ON mf.message_id = m.id AND mf.rating = 'like'
|
|
WHERE
|
|
{workflow_scope}"""
|
|
args: dict[str, Any] = {
|
|
"tz": params.timezone,
|
|
"tenant_id": app.tenant_id,
|
|
"agent_id": agent_id,
|
|
"chat_workflow_type": WorkflowType.CHAT,
|
|
}
|
|
if source_filter.app_id:
|
|
args["source_app_id"] = source_filter.app_id
|
|
if source_filter.workflow_id:
|
|
args["workflow_id"] = source_filter.workflow_id
|
|
if source_filter.workflow_version:
|
|
args["workflow_version"] = source_filter.workflow_version
|
|
if source_filter.node_id:
|
|
args["node_id"] = source_filter.node_id
|
|
if params.start:
|
|
sql_query += " AND m.created_at >= :start"
|
|
args["start"] = params.start
|
|
if params.end:
|
|
sql_query += " AND m.created_at < :end"
|
|
args["end"] = params.end
|
|
sql_query += " GROUP BY date ORDER BY date"
|
|
|
|
return [dict(row._mapping) for row in self._session.execute(sa.text(sql_query), args).all()]
|
|
|
|
@staticmethod
|
|
def _statistics_workflow_binding_filters_sql(source_filter: AgentSourceFilter) -> str:
|
|
workflow_binding_filters = []
|
|
if source_filter.app_id:
|
|
workflow_binding_filters.append("wanb.app_id = :source_app_id")
|
|
if source_filter.workflow_id:
|
|
workflow_binding_filters.append("wanb.workflow_id = :workflow_id")
|
|
if source_filter.workflow_version:
|
|
workflow_binding_filters.append("wanb.workflow_version = :workflow_version")
|
|
if source_filter.node_id:
|
|
workflow_binding_filters.append("wanb.node_id = :node_id")
|
|
return f"AND {' AND '.join(workflow_binding_filters)}" if workflow_binding_filters else ""
|
|
|
|
@classmethod
|
|
def _statistics_workflow_message_scope_sql(cls, source_filter: AgentSourceFilter) -> str:
|
|
binding_filters = cls._statistics_workflow_binding_filters_sql(source_filter)
|
|
return f"""m.workflow_run_id IS NOT NULL
|
|
AND EXISTS (
|
|
SELECT 1
|
|
FROM workflow_runs wr
|
|
JOIN workflow_agent_node_bindings wanb
|
|
ON wanb.tenant_id = :tenant_id
|
|
AND wanb.agent_id = :agent_id
|
|
AND wanb.app_id = wr.app_id
|
|
AND wanb.workflow_id = wr.workflow_id
|
|
AND wanb.workflow_version = wr.version
|
|
{binding_filters}
|
|
JOIN workflow_node_executions wne
|
|
ON wne.workflow_run_id = wr.id
|
|
AND wne.node_id = wanb.node_id
|
|
WHERE wr.id = m.workflow_run_id
|
|
AND wr.type = :chat_workflow_type
|
|
)"""
|
|
|
|
@staticmethod
|
|
def _workflow_execution_metadata_numeric_sql(path: tuple[str, ...], numeric_type: str) -> str:
|
|
if dify_config.DB_TYPE == "postgresql":
|
|
json_path = ",".join(path)
|
|
value = f"CAST(wne.execution_metadata AS JSONB) #>> '{{{json_path}}}'"
|
|
return f"CAST(NULLIF({value}, '') AS {numeric_type})"
|
|
if dify_config.DB_TYPE in {"mysql", "oceanbase", "seekdb"}:
|
|
json_path = "$." + ".".join(path)
|
|
mysql_numeric_type = "UNSIGNED" if numeric_type == "BIGINT" else numeric_type
|
|
value = f"JSON_UNQUOTE(JSON_EXTRACT(wne.execution_metadata, '{json_path}'))"
|
|
return f"CAST(NULLIF(NULLIF({value}, ''), 'null') AS {mysql_numeric_type})"
|
|
raise NotImplementedError(f"Unsupported database type: {dify_config.DB_TYPE}")
|
|
|
|
@staticmethod
|
|
def _merge_daily_statistics(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
merged: dict[Any, dict[str, Any]] = {}
|
|
weighted_latency: dict[Any, float] = {}
|
|
for row in rows:
|
|
date = row["date"]
|
|
target = merged.setdefault(
|
|
date,
|
|
{
|
|
"date": date,
|
|
"message_count": 0,
|
|
"conversation_count": 0,
|
|
"end_user_count": 0,
|
|
"token_count": 0,
|
|
"total_price": Decimal(0),
|
|
"avg_latency": 0.0,
|
|
"latency_sum": 0.0,
|
|
"answer_tokens": 0,
|
|
"like_count": 0,
|
|
},
|
|
)
|
|
message_count = int(row.get("message_count") or 0)
|
|
target["message_count"] += message_count
|
|
target["conversation_count"] += int(row.get("conversation_count") or 0)
|
|
target["end_user_count"] += int(row.get("end_user_count") or 0)
|
|
target["token_count"] += int(row.get("token_count") or 0)
|
|
target["total_price"] += Decimal(str(row.get("total_price") or 0))
|
|
target["latency_sum"] += float(row.get("latency_sum") or 0)
|
|
target["answer_tokens"] += int(row.get("answer_tokens") or 0)
|
|
target["like_count"] += int(row.get("like_count") or 0)
|
|
weighted_latency[date] = (
|
|
weighted_latency.get(date, 0.0) + float(row.get("avg_latency") or 0) * message_count
|
|
)
|
|
|
|
for date, row in merged.items():
|
|
message_count = int(row["message_count"])
|
|
row["avg_latency"] = weighted_latency[date] / message_count if message_count else 0.0
|
|
return sorted(merged.values(), key=lambda row: str(row["date"]))
|
|
|
|
@staticmethod
|
|
def _build_charts(rows: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
|
|
messages = []
|
|
conversations = []
|
|
end_users = []
|
|
token_usage = []
|
|
average_session_interactions = []
|
|
average_response_time = []
|
|
tokens_per_second = []
|
|
user_satisfaction_rate = []
|
|
|
|
for row in rows:
|
|
date = str(row["date"])
|
|
message_count = int(row["message_count"] or 0)
|
|
conversation_count = int(row["conversation_count"] or 0)
|
|
token_count = int(row["token_count"] or 0)
|
|
total_price = row["total_price"] or Decimal(0)
|
|
avg_latency = float(row["avg_latency"] or 0)
|
|
latency_sum = float(row["latency_sum"] or 0)
|
|
answer_tokens = int(row["answer_tokens"] or 0)
|
|
like_count = int(row["like_count"] or 0)
|
|
|
|
messages.append({"date": date, "message_count": message_count})
|
|
conversations.append({"date": date, "conversation_count": conversation_count})
|
|
end_users.append({"date": date, "terminal_count": int(row["end_user_count"] or 0)})
|
|
token_usage.append(
|
|
{
|
|
"date": date,
|
|
"token_count": token_count,
|
|
"total_price": str(total_price),
|
|
"currency": "USD",
|
|
}
|
|
)
|
|
average_session_interactions.append(
|
|
{
|
|
"date": date,
|
|
"interactions": round(message_count / conversation_count, 2) if conversation_count else 0,
|
|
}
|
|
)
|
|
average_response_time.append({"date": date, "latency": round(avg_latency * 1000, 4)})
|
|
tokens_per_second.append({"date": date, "tps": round(answer_tokens / latency_sum, 4) if latency_sum else 0})
|
|
user_satisfaction_rate.append(
|
|
{"date": date, "rate": round(like_count * 100 / message_count, 2) if message_count else 0}
|
|
)
|
|
|
|
return {
|
|
"daily_messages": messages,
|
|
"daily_conversations": conversations,
|
|
"daily_end_users": end_users,
|
|
"token_usage": token_usage,
|
|
"average_session_interactions": average_session_interactions,
|
|
"average_response_time": average_response_time,
|
|
"tokens_per_second": tokens_per_second,
|
|
"user_satisfaction_rate": user_satisfaction_rate,
|
|
}
|
|
|
|
@staticmethod
|
|
def _build_summary(rows: list[dict[str, Any]]) -> dict[str, Any]:
|
|
total_messages = sum(int(row["message_count"] or 0) for row in rows)
|
|
total_conversations = sum(int(row["conversation_count"] or 0) for row in rows)
|
|
total_end_users = sum(int(row["end_user_count"] or 0) for row in rows)
|
|
total_tokens = sum(int(row["token_count"] or 0) for row in rows)
|
|
total_price = sum(Decimal(str(row["total_price"] or 0)) for row in rows)
|
|
total_answer_tokens = sum(int(row["answer_tokens"] or 0) for row in rows)
|
|
total_latency = sum(float(row["latency_sum"] or 0) for row in rows)
|
|
weighted_latency = sum(float(row["avg_latency"] or 0) * int(row["message_count"] or 0) for row in rows)
|
|
total_likes = sum(int(row["like_count"] or 0) for row in rows)
|
|
|
|
return {
|
|
"total_messages": total_messages,
|
|
"total_conversations": total_conversations,
|
|
"total_end_users": total_end_users,
|
|
"total_tokens": total_tokens,
|
|
"total_price": str(total_price),
|
|
"currency": "USD",
|
|
"average_session_interactions": round(total_messages / total_conversations, 2)
|
|
if total_conversations
|
|
else 0,
|
|
"average_response_time": round((weighted_latency / total_messages) * 1000, 4) if total_messages else 0,
|
|
"tokens_per_second": round(total_answer_tokens / total_latency, 4) if total_latency else 0,
|
|
"user_satisfaction_rate": round(total_likes * 100 / total_messages, 2) if total_messages else 0,
|
|
}
|