wehub-resource-sync
354 lines
12 KiB
Python
354 lines
12 KiB
Python
import re
|
|
import warnings
|
|
from pydantic import (
|
|
BaseModel,
|
|
Field,
|
|
PrivateAttr,
|
|
model_validator,
|
|
AliasChoices,
|
|
)
|
|
from typing import List, Optional, Dict, Literal, Union
|
|
from copy import deepcopy
|
|
from enum import Enum
|
|
|
|
from deepeval.test_case import (
|
|
ToolCall,
|
|
ToolCallType,
|
|
MLLMImage,
|
|
RetrievedContextData,
|
|
)
|
|
from deepeval.test_case.mcp import (
|
|
MCPServer,
|
|
MCPPromptCall,
|
|
MCPResourceCall,
|
|
MCPToolCall,
|
|
validate_mcp_servers,
|
|
)
|
|
from deepeval.test_case.llm_test_case import _MLLM_IMAGE_REGISTRY
|
|
|
|
|
|
class MultiTurnParams(Enum):
|
|
ROLE = "role"
|
|
CONTENT = "content"
|
|
METADATA = "metadata"
|
|
TAGS = "tags"
|
|
SCENARIO = "scenario"
|
|
EXPECTED_OUTCOME = "expected_outcome"
|
|
CONTEXT = "context"
|
|
USER_DESCRIPTION = "user_description"
|
|
RETRIEVAL_CONTEXT = "retrieval_context"
|
|
CHATBOT_ROLE = "chatbot_role"
|
|
TOOLS_CALLED = "tools_called"
|
|
MCP_TOOLS = "mcp_tools_called"
|
|
MCP_RESOURCES = "mcp_resources_called"
|
|
MCP_PROMPTS = "mcp_prompts_called"
|
|
|
|
|
|
def __getattr__(name: str):
|
|
if name == "TurnParams":
|
|
warnings.warn(
|
|
"'TurnParams' is deprecated and will be removed in a future "
|
|
"release. Use 'MultiTurnParams' instead.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
return MultiTurnParams
|
|
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
|
|
|
|
|
class Turn(BaseModel):
|
|
role: Literal["user", "assistant"]
|
|
content: str
|
|
user_id: Optional[str] = Field(
|
|
default=None, validation_alias=AliasChoices("userId", "user_id")
|
|
)
|
|
retrieval_context: Optional[List[Union[str, RetrievedContextData]]] = Field(
|
|
default=None,
|
|
validation_alias=AliasChoices("retrievalContext", "retrieval_context"),
|
|
)
|
|
tools_called: Optional[List[ToolCall]] = Field(
|
|
default=None,
|
|
validation_alias=AliasChoices("toolsCalled", "tools_called"),
|
|
)
|
|
mcp_tools_called: Optional[List[MCPToolCall]] = Field(default=None)
|
|
mcp_resources_called: Optional[List[MCPResourceCall]] = Field(default=None)
|
|
mcp_prompts_called: Optional[List[MCPPromptCall]] = Field(default=None)
|
|
metadata: Optional[Dict] = Field(
|
|
default=None,
|
|
validation_alias=AliasChoices(
|
|
"metadata", "additionalMetadata", "additional_metadata"
|
|
),
|
|
)
|
|
|
|
@property
|
|
def additional_metadata(self) -> Optional[Dict]:
|
|
warnings.warn(
|
|
"'additional_metadata' is deprecated. Use 'metadata' instead.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
return self.metadata
|
|
|
|
@additional_metadata.setter
|
|
def additional_metadata(self, value: Optional[Dict]):
|
|
warnings.warn(
|
|
"'additional_metadata' is deprecated. Use 'metadata' instead.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
self.metadata = value
|
|
|
|
@property
|
|
def _mcp_tool_calls(self) -> List:
|
|
if self.mcp_tools_called is not None:
|
|
return self.mcp_tools_called
|
|
return [
|
|
tool
|
|
for tool in (self.tools_called or [])
|
|
if tool.type == ToolCallType.MCP
|
|
]
|
|
|
|
@property
|
|
def _mcp_interaction(self) -> bool:
|
|
has_mcp_tools = self.mcp_tools_called is not None or any(
|
|
tool.type == ToolCallType.MCP for tool in (self.tools_called or [])
|
|
)
|
|
return (
|
|
has_mcp_tools
|
|
or self.mcp_resources_called is not None
|
|
or self.mcp_prompts_called is not None
|
|
)
|
|
|
|
def __repr__(self):
|
|
attrs = [f"role={self.role!r}", f"content={self.content!r}"]
|
|
if self.user_id is not None:
|
|
attrs.append(f"user_id={self.user_id!r}")
|
|
if self.retrieval_context is not None:
|
|
attrs.append(f"retrieval_context={self.retrieval_context!r}")
|
|
if self.tools_called is not None:
|
|
attrs.append(f"tools_called={self.tools_called!r}")
|
|
if self.mcp_tools_called is not None:
|
|
attrs.append(f"mcp_tools_called={self.mcp_tools_called!r}")
|
|
if self.mcp_resources_called is not None:
|
|
attrs.append(f"mcp_resources_called={self.mcp_resources_called!r}")
|
|
if self.mcp_prompts_called is not None:
|
|
attrs.append(f"mcp_prompts_called={self.mcp_prompts_called!r}")
|
|
if self.metadata is not None:
|
|
attrs.append(f"metadata={self.metadata!r}")
|
|
return f"Turn({', '.join(attrs)})"
|
|
|
|
@model_validator(mode="before")
|
|
def validate_input(cls, data):
|
|
mcp_tools_called = data.get("mcp_tools_called")
|
|
mcp_prompts_called = data.get("mcp_prompts_called")
|
|
mcp_resources_called = data.get("mcp_resources_called")
|
|
|
|
if (
|
|
mcp_tools_called is not None
|
|
or mcp_prompts_called is not None
|
|
or mcp_resources_called is not None
|
|
):
|
|
from mcp.types import (
|
|
CallToolResult,
|
|
ReadResourceResult,
|
|
GetPromptResult,
|
|
)
|
|
|
|
if mcp_tools_called is not None:
|
|
if not isinstance(mcp_tools_called, list) or not all(
|
|
isinstance(tool_called, MCPToolCall)
|
|
and isinstance(tool_called.result, CallToolResult)
|
|
for tool_called in mcp_tools_called
|
|
):
|
|
raise TypeError(
|
|
"The 'tools_called' must be a list of 'MCPToolCall' with result of type 'CallToolResult' from mcp.types"
|
|
)
|
|
|
|
if mcp_resources_called is not None:
|
|
if not isinstance(mcp_resources_called, list) or not all(
|
|
isinstance(resource_called, MCPResourceCall)
|
|
and isinstance(resource_called.result, ReadResourceResult)
|
|
for resource_called in mcp_resources_called
|
|
):
|
|
raise TypeError(
|
|
"The 'resources_called' must be a list of 'MCPResourceCall' with result of type 'ReadResourceResult' from mcp.types"
|
|
)
|
|
|
|
if mcp_prompts_called is not None:
|
|
if not isinstance(mcp_prompts_called, list) or not all(
|
|
isinstance(prompt_called, MCPPromptCall)
|
|
and isinstance(prompt_called.result, GetPromptResult)
|
|
for prompt_called in mcp_prompts_called
|
|
):
|
|
raise TypeError(
|
|
"The 'prompts_called' must be a list of 'MCPPromptCall' with result of type 'GetPromptResult' from mcp.types"
|
|
)
|
|
|
|
return data
|
|
|
|
|
|
class ConversationalTestCase(BaseModel):
|
|
turns: List[Turn]
|
|
scenario: Optional[str] = Field(default=None)
|
|
context: Optional[List[str]] = Field(default=None)
|
|
name: Optional[str] = Field(default=None)
|
|
user_description: Optional[str] = Field(
|
|
default=None,
|
|
serialization_alias="userDescription",
|
|
validation_alias=AliasChoices("userDescription", "user_description"),
|
|
)
|
|
expected_outcome: Optional[str] = Field(
|
|
default=None,
|
|
serialization_alias="expectedOutcome",
|
|
validation_alias=AliasChoices("expectedOutcome", "expected_outcome"),
|
|
)
|
|
chatbot_role: Optional[str] = Field(
|
|
default=None,
|
|
serialization_alias="chatbotRole",
|
|
validation_alias=AliasChoices("chatbotRole", "chatbot_role"),
|
|
)
|
|
metadata: Optional[Dict] = Field(
|
|
default=None,
|
|
validation_alias=AliasChoices(
|
|
"metadata", "additionalMetadata", "additional_metadata"
|
|
),
|
|
)
|
|
comments: Optional[str] = Field(default=None)
|
|
tags: Optional[List[str]] = Field(default=None)
|
|
mcp_servers: Optional[List[MCPServer]] = Field(default=None)
|
|
multimodal: bool = False
|
|
|
|
_dataset_rank: Optional[int] = PrivateAttr(default=None)
|
|
_dataset_alias: Optional[str] = PrivateAttr(default=None)
|
|
_dataset_id: Optional[str] = PrivateAttr(default=None)
|
|
|
|
@property
|
|
def additional_metadata(self) -> Optional[Dict]:
|
|
warnings.warn(
|
|
"'additional_metadata' is deprecated. Use 'metadata' instead.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
return self.metadata
|
|
|
|
@additional_metadata.setter
|
|
def additional_metadata(self, value: Optional[Dict]):
|
|
warnings.warn(
|
|
"'additional_metadata' is deprecated. Use 'metadata' instead.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
self.metadata = value
|
|
|
|
@model_validator(mode="after")
|
|
def set_is_multimodal(self):
|
|
import re
|
|
|
|
if self.multimodal is True:
|
|
return self
|
|
|
|
pattern = r"\[DEEPEVAL:(?:IMAGE|PDF):(.*?)\]"
|
|
if self.scenario:
|
|
if re.search(pattern, self.scenario) is not None:
|
|
self.multimodal = True
|
|
return self
|
|
if self.expected_outcome:
|
|
if re.search(pattern, self.expected_outcome) is not None:
|
|
self.multimodal = True
|
|
return self
|
|
if self.user_description:
|
|
if re.search(pattern, self.user_description) is not None:
|
|
self.multimodal = True
|
|
return self
|
|
if self.turns:
|
|
for turn in self.turns:
|
|
if re.search(pattern, turn.content) is not None:
|
|
self.multimodal = True
|
|
return self
|
|
if turn.retrieval_context is not None:
|
|
self.multimodal = self.multimodal or any(
|
|
re.search(
|
|
pattern,
|
|
(
|
|
c.context
|
|
if isinstance(c, RetrievedContextData)
|
|
else c
|
|
),
|
|
)
|
|
for c in turn.retrieval_context
|
|
if isinstance(c, (RetrievedContextData, str))
|
|
)
|
|
|
|
return self
|
|
|
|
@model_validator(mode="before")
|
|
def validate_input(cls, data):
|
|
turns = data.get("turns")
|
|
context = data.get("context")
|
|
mcp_servers = data.get("mcp_servers")
|
|
|
|
if len(turns) == 0:
|
|
raise TypeError("'turns' must not be empty")
|
|
|
|
# Ensure `context` is None or a list of strings
|
|
if context is not None:
|
|
if not isinstance(context, list) or not all(
|
|
isinstance(item, (str, RetrievedContextData))
|
|
for item in context
|
|
):
|
|
raise TypeError(
|
|
"'context' must be None or a list of strings or RetrievedContextData"
|
|
)
|
|
|
|
if mcp_servers is not None:
|
|
validate_mcp_servers(mcp_servers)
|
|
|
|
copied_turns = []
|
|
for turn in turns:
|
|
if isinstance(turn, Turn):
|
|
copied_turns.append(deepcopy(turn))
|
|
elif isinstance(turn, dict):
|
|
try:
|
|
copied_turns.append(Turn.model_validate(turn))
|
|
except Exception as e:
|
|
raise TypeError(f"Invalid dict for Turn: {turn} ({e})")
|
|
else:
|
|
raise TypeError(
|
|
f"'turns' must be a list of Turn or dict, got {type(turn)}"
|
|
)
|
|
|
|
data["turns"] = copied_turns
|
|
|
|
return data
|
|
|
|
def _get_images_mapping(self) -> Dict[str, MLLMImage]:
|
|
pattern = r"\[DEEPEVAL:(?:IMAGE|PDF):(.*?)\]"
|
|
image_ids = set()
|
|
|
|
def extract_ids_from_string(s: Optional[str]) -> None:
|
|
"""Helper to extract image IDs from a string."""
|
|
if s is not None and isinstance(s, str):
|
|
matches = re.findall(pattern, s)
|
|
image_ids.update(matches)
|
|
|
|
def extract_ids_from_list(lst: Optional[List[str]]) -> None:
|
|
"""Helper to extract image IDs from a list of strings."""
|
|
if lst is not None:
|
|
for item in lst:
|
|
extract_ids_from_string(item)
|
|
|
|
extract_ids_from_string(self.scenario)
|
|
extract_ids_from_string(self.expected_outcome)
|
|
extract_ids_from_list(self.context)
|
|
extract_ids_from_string(self.user_description)
|
|
for turn in self.turns:
|
|
extract_ids_from_string(turn.content)
|
|
extract_ids_from_list(turn.retrieval_context)
|
|
|
|
images_mapping = {}
|
|
for img_id in image_ids:
|
|
if img_id in _MLLM_IMAGE_REGISTRY:
|
|
images_mapping[img_id] = _MLLM_IMAGE_REGISTRY[img_id]
|
|
|
|
return images_mapping if len(images_mapping) > 0 else None
|