mirror of
https://github.com/langgenius/dify.git
synced 2026-11-11 22:02:23 +08:00
84 lines
3.7 KiB
Python
84 lines
3.7 KiB
Python
"""Attribute bounded native resource failures using their executed graph.
|
|
|
|
This is presentation evidence, never a grant or proof that earlier nodes had no
|
|
effects. Ambiguous failures and arbitrary Code exception text are not classified.
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
from collections.abc import Mapping
|
|
from typing import Literal
|
|
|
|
from core.dify_builder.models import Run
|
|
from services.dify_builder.native_launch_observation import FailureResource, _public_id
|
|
|
|
NativeResourceReason = Literal["service_unavailable", "authentication", "permission_denied"]
|
|
|
|
|
|
def native_resource_failure(
|
|
run: Run, graph: Mapping
|
|
) -> tuple[FailureResource, tuple[str, ...], NativeResourceReason] | None:
|
|
if run.status != "failed" or not run.dify_run_id:
|
|
return None
|
|
failed = [node for node in run.per_node if node.status in {"failed", "exception"}]
|
|
if len(failed) != 1:
|
|
return None
|
|
failure = failed[0]
|
|
if not _public_id(failure.node_id):
|
|
return None
|
|
graph_nodes = graph.get("nodes")
|
|
if not isinstance(graph_nodes, list):
|
|
return None
|
|
nodes = [node for node in graph_nodes if isinstance(node, dict) and node.get("id") == failure.node_id]
|
|
if len(nodes) != 1 or not isinstance(data := nodes[0].get("data"), dict):
|
|
return None
|
|
kind: Literal["model", "tool"]
|
|
if data.get("type") == "llm" and isinstance(model := data.get("model"), dict):
|
|
kind, provider, name = "model", model.get("provider"), model.get("name")
|
|
elif data.get("type") == "tool" and data.get("provider_type") in {"builtin", "plugin"}:
|
|
kind, provider, name = "tool", data.get("provider_id"), data.get("tool_name")
|
|
else:
|
|
return None
|
|
if not isinstance(provider, str) or not isinstance(name, str) or not _public_id(provider) or not _public_id(name):
|
|
return None
|
|
reason = _native_reason(failure.error, kind)
|
|
return ((kind, provider, name), (failure.node_id,), reason) if reason is not None else None
|
|
|
|
|
|
def _native_reason(error: str, kind: Literal["model", "tool"]) -> NativeResourceReason | None:
|
|
if re.fullmatch(r"Request to Plugin Daemon Service failed-(?:500|502|503|504)", error.strip()):
|
|
return "service_unavailable"
|
|
if len(error) > 32768:
|
|
return None
|
|
envelope = re.fullmatch(r"(?:Run failed: |req_id: [A-Za-z0-9-]{1,128} )?PluginInvokeError: (.+)", error, re.DOTALL)
|
|
if envelope is None:
|
|
return None
|
|
try:
|
|
payload = json.loads(envelope[1])
|
|
args = payload.get("args") if isinstance(payload, dict) else None
|
|
description = args.get("description") if isinstance(args, dict) else None
|
|
if not isinstance(description, str):
|
|
return None
|
|
prefix = "models" if kind == "model" else "tools"
|
|
match = re.fullmatch(
|
|
r"\[" + prefix + r"\] Error: [^\n]*?status code (401|403): (\{.+\})", description, re.DOTALL
|
|
)
|
|
if match is None:
|
|
return None
|
|
body = json.loads(match[2])
|
|
detail = body.get("error") if isinstance(body, dict) else None
|
|
category = detail.get("type") if isinstance(detail, dict) else None
|
|
except (ValueError, TypeError, RecursionError):
|
|
return None
|
|
if (match[1], category) == ("401", "authentication_error"):
|
|
return "authentication"
|
|
if (match[1], category) == ("403", "permission_denied"):
|
|
return "permission_denied"
|
|
return None
|
|
|
|
|
|
def plugin_service_failure(run: Run, graph: Mapping) -> tuple[FailureResource, tuple[str, ...]] | None:
|
|
"""Retain the outage-only adapter contract for existing consumers."""
|
|
failure = native_resource_failure(run, graph)
|
|
return (failure[0], failure[1]) if failure is not None and failure[2] == "service_unavailable" else None
|