mirror of
https://github.com/langgenius/dify.git
synced 2026-02-10 04:54:30 +08:00
4a9b74f86b
**Problem:**
The telemetry system had unnecessary abstraction layers and bad practices
from the last 3 commits introducing the gateway implementation:
- TelemetryFacade class wrapper around emit() function
- String literals instead of SignalType enum
- Dictionary mapping enum → string instead of enum → enum
- Unnecessary ENTERPRISE_TELEMETRY_GATEWAY_ENABLED feature flag
- Duplicate guard checks scattered across files
- Non-thread-safe TelemetryGateway singleton pattern
- Missing guard in ops_trace_task.py causing RuntimeError spam
**Solution:**
1. Deleted TelemetryFacade - replaced with thin emit() function in core/telemetry/__init__.py
2. Added SignalType enum ('trace' | 'metric_log') to enterprise/telemetry/contracts.py
3. Replaced CASE_TO_TRACE_TASK_NAME dict with CASE_TO_TRACE_TASK: dict[TelemetryCase, TraceTaskName]
4. Deleted is_gateway_enabled() and _emit_legacy() - using existing ENTERPRISE_ENABLED + ENTERPRISE_TELEMETRY_ENABLED instead
5. Extracted _should_drop_ee_only_event() helper to eliminate duplicate checks
6. Moved TelemetryGateway singleton to ext_enterprise_telemetry.py:
- Init once in init_app() for thread-safety
- Access via get_gateway() function
7. Re-added guard to ops_trace_task.py to prevent RuntimeError when EE=OFF but CE tracing enabled
8. Updated 11 caller files to import 'emit as telemetry_emit' instead of 'TelemetryFacade'
**Result:**
- 322 net lines deleted (533 removed, 211 added)
- All 91 tests pass
- Thread-safe singleton pattern
- Cleaner API surface: from TelemetryFacade.emit() to telemetry_emit()
- Proper enum usage throughout
- No RuntimeError spam in EE=OFF + CE=ON scenario
77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
import logging
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
from core.app.app_config.entities import AppConfig
|
|
from core.moderation.base import ModerationAction, ModerationError
|
|
from core.moderation.factory import ModerationFactory
|
|
from core.ops.ops_trace_manager import TraceQueueManager
|
|
from core.ops.utils import measure_time
|
|
from core.telemetry import TelemetryContext, TelemetryEvent, TraceTaskName
|
|
from core.telemetry import emit as telemetry_emit
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class InputModeration:
|
|
def check(
|
|
self,
|
|
app_id: str,
|
|
tenant_id: str,
|
|
app_config: AppConfig,
|
|
inputs: Mapping[str, Any],
|
|
query: str,
|
|
message_id: str,
|
|
trace_manager: TraceQueueManager | None = None,
|
|
) -> tuple[bool, Mapping[str, Any], str]:
|
|
"""
|
|
Process sensitive_word_avoidance.
|
|
:param app_id: app id
|
|
:param tenant_id: tenant id
|
|
:param app_config: app config
|
|
:param inputs: inputs
|
|
:param query: query
|
|
:param message_id: message id
|
|
:param trace_manager: trace manager
|
|
:return:
|
|
"""
|
|
inputs = dict(inputs)
|
|
if not app_config.sensitive_word_avoidance:
|
|
return False, inputs, query
|
|
|
|
sensitive_word_avoidance_config = app_config.sensitive_word_avoidance
|
|
moderation_type = sensitive_word_avoidance_config.type
|
|
|
|
moderation_factory = ModerationFactory(
|
|
name=moderation_type, app_id=app_id, tenant_id=tenant_id, config=sensitive_word_avoidance_config.config
|
|
)
|
|
|
|
with measure_time() as timer:
|
|
moderation_result = moderation_factory.moderation_for_inputs(inputs, query)
|
|
|
|
if trace_manager:
|
|
telemetry_emit(
|
|
TelemetryEvent(
|
|
name=TraceTaskName.MODERATION_TRACE,
|
|
context=TelemetryContext(tenant_id=tenant_id, app_id=app_id),
|
|
payload={
|
|
"message_id": message_id,
|
|
"moderation_result": moderation_result,
|
|
"inputs": inputs,
|
|
"timer": timer,
|
|
},
|
|
),
|
|
trace_manager=trace_manager,
|
|
)
|
|
|
|
if not moderation_result.flagged:
|
|
return False, inputs, query
|
|
|
|
if moderation_result.action == ModerationAction.DIRECT_OUTPUT:
|
|
raise ModerationError(moderation_result.preset_response)
|
|
elif moderation_result.action == ModerationAction.OVERRIDDEN:
|
|
inputs = moderation_result.inputs
|
|
query = moderation_result.query
|
|
|
|
return True, inputs, query
|