Основные метрики DeepEval для тестирования AI. Возможности и способы применения

Метрики призваны для оценки работа AI агентов

Task Completion - показывает, насколько эффективно AI-агент выполняет задачу.

где

- task - задача для выполнения;
- outcome - результат выполнения задачи.

Для метрики являются обязательными следующие параметры:

- запрос пользователя;
- окончательный ответ;
- список вызванных в ходе выполнения запроса инструментов.

Задача, конечный результат и список вызванных инструментов извлекаются с помощью LLM, на основании чего выводится окончательный вердикт. В ходе оценки проверяется, какие именно инструменты были вызваны и насколько полезная информация была извлечена с их помощью.

Tool Correctness - показывает эффективность и правильность использования инструментов AI-агентом при выполнении задачи. Вычисляется как отношение количества реального использования инструментов к количеству ожидаемых использований.

Для метрики являются обязательными следующие параметры:

- запрос пользователя;
- список ожидаемых для вызова инструментов;
- список вызванных в ходе выполнения запроса инструментов.

Помимо просто вычисления использования нужных устройств может оценивать также и правильный порядок их использования.

Argument Correctness - оценивает аргументы, с которыми AI-агент совершает вызов того или иного инструмента. Рассчитывается путем определения того, насколько аргументы соответствуют инструменту.

Для метрики являются обязательными следующие параметры:

- запрос пользователя;
- список вызванных в ходе выполнения запроса инструментов.

Имея в распоряжении список инструментов, проверяет, соответствуют ли параметры для вызова того или иного инструмента запросу пользователя. В случае соответствия возвращает ‘yes’, в противном случае возвращает ‘no’.

По итогам возвращенных значений считается отношение верных вызовов к общему их числу.

Реализация в Python

В целом, все метрики используются следующим образом:

from deepeval.test_case import ToolCall, LLMTestCase
from deepeval.metrics import ToolCorrectnessMetric, TaskCompletionMetric, ArgumentCorrectnessMetric

tools_called = []
# Преобразование перечня вызовов инструментов в объекты ToolCall 
# и добавление таких объектов в список
for tc in tool_calls_from_checkpoint: 
    tools_called.append(ToolCall(
        name=tc["name"],
        input_parameters=tc.get("arguments", {})
    ))

# Формирование тест кейса
llm_test_case = LLMTestCase(
                input=input, # Запрос пользователя
                actual_output=actual_output, # Окончательный ответ
                tools_called=tools_called, # Список действительно вызванных инструментов
                expected_tools=expected_tools # Список ожидаемых инструментов
            )
  
# Оцениваем с помощью TaskCompletionMetric
tool_correctness = ToolCorrectnessMetric()
tool_correctness.measure(llm_test_case)

print(f"\nTool Correctness Score: {tool_correctness.score}")
print(f"Reason: {tool_correctness.reason}")

# Оцениваем с помощью TaskCompletionMetric
task_completion = TaskCompletionMetric(threshold=0.7)
task_completion.measure(llm_test_case)

print(f"\nTask Completion Score: {task_completion.score}")
print(f"Reason: {task_completion.reason}")

# Оцениваем с помощью ArgumentCorrectnessMetric
arg_correctness = ArgumentCorrectnessMetric(threshold=0.5)
arg_correctness.measure(llm_test_case)

print(f"\nArgument Correctness Score: {arg_correctness.score}")
print(f"Reason: {arg_correctness.reason}")

Применительно к AI агенту внедрение метрик выглядит следующим образом

Допустим, у нас есть AI агент, который должен обладать возможностью использовать следующие инструменты:

- поиск информации в интернете;
- получение документации по определенной теме;
- вычисление математических выражений;
- отправка электронных писем;
- управление файлами.

Импортируем все необходимые модули:

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage, AIMessage
from deepeval.test_case import ToolCall, LLMTestCase
from deepeval.metrics import ToolCorrectnessMetric, TaskCompletionMetric, ArgumentCorrectnessMetric

Состояние графа описано следующим образом:

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    tool_calls_log: list[dict]  # Сохраняем детальную информацию о tool calls
    execution_history: list[dict]  # История выполнения

Инструменты для AI агента выглядят примерно так:

@tool
def web_search(query: str) -> str:
    """Поиск информации в интернете"""
    return f"Найдена информация по запросу: {query}"

@tool
def document_retrieval(topic: str) -> str:
    """Получение документов по теме"""
    return f"Найдены документы по теме: {topic}"

@tool
def calculator(expression: str) -> str:
    """Вычисление математических выражений"""
    try:
        result = eval(expression)
        return f"Результат: {result}"
    except:
        return "Ошибка в выражении"

@tool
def email_sender(recipient: str, subject: str, message: str) -> str:
    """Отправка email сообщения"""
    return f"Email отправлен на {recipient} с темой '{subject}'"

@tool
def file_manager(action: str, filename: str, content: str = "") -> str:
    """Управление файлами"""
    if action == "create":
        return f"Файл '{filename}' создан с содержимым"
    elif action == "read":
        return f"Содержимое файла '{filename}': sample content"
    elif action == "delete":
        return f"Файл '{filename}' удален"
    else:
        return f"Неизвестное действие: {action}"

# Список всех инструментов
tools = [web_search, document_retrieval, calculator, email_sender, 
         file_manager]

В реальном описании вместо текста для поиска по интернету и документам должна быть прописана логика для использования API поисковых систем и RAG.

Создаем узел логирования всех действий агента:

def agent_node_with_logging(state: AgentState):
    """Узел агента с детальным логированием в состояние"""

    # Создаем LLM с привязанными инструментами
    llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo")
    llm_with_tools = llm.bind_tools(tools)

    # Получаем сообщения
    messages = state["messages"]

    # Инициализируем логи если их нет
    tool_calls_log = state.get("tool_calls_log", [])
    execution_history = state.get("execution_history", [])

    # Вызываем LLM
    response = llm_with_tools.invoke(messages)

    # Логируем шаг выполнения
    execution_step = {
        "step": len(execution_history) + 1,
        "action": "llm_response",
        "has_tool_calls": bool(response.tool_calls),
        "tool_calls_count": len(response.tool_calls) if response.tool_calls else 0
    }
    execution_history.append(execution_step)

    # Если в ответе есть tool calls
    if response.tool_calls:
        new_messages = [response]

        for tool_call in response.tool_calls:
            # Детальное логирование tool call
            tool_call_info = {
                "id": tool_call["id"],
                "name": tool_call["name"],
                "arguments": tool_call.get("args", {}),
                "timestamp": len(tool_calls_log) + 1,
                "execution_step": len(execution_history)
            }

            # Выполняем инструмент
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]

            # Находим и выполняем соответствующий инструмент
            tool_result = None
            for tool in tools:
                if tool.name == tool_name:
                    try:
                        tool_result = tool.invoke(tool_args)
                        tool_call_info["result"] = str(tool_result)
                        tool_call_info["status"] = "success"

                        # Добавляем результат в сообщения
                        tool_message = ToolMessage(
                            content=str(tool_result),
                            tool_call_id=tool_call["id"]
                        )
                        new_messages.append(tool_message)

                    except Exception as e:
                        tool_call_info["result"] = f"Ошибка: {e}"
                        tool_call_info["status"] = "error"

                        tool_message = ToolMessage(
                            content=f"Ошибка выполнения инструмента: {e}",
                            tool_call_id=tool_call["id"]
                        )
                        new_messages.append(tool_message)
                    break

            # Добавляем в лог
            tool_calls_log.append(tool_call_info)

        # Генерируем финальный ответ после выполнения всех инструментов
        all_messages = messages + new_messages
        final_response = llm.invoke(all_messages)

        # Логируем финальный шаг
        final_step = {
            "step": len(execution_history) + 1,
            "action": "final_response",
            "tools_executed": len([tc for tc in tool_calls_log if tc.get("timestamp", 0) > len(execution_history)])
        }
        execution_history.append(final_step)

        return {
            "messages": new_messages + [final_response],
            "tool_calls_log": tool_calls_log,
            "execution_history": execution_history
        }

    # Если tool calls нет
    final_step = {
        "step": len(execution_history) + 1,
        "action": "direct_response",
        "tools_executed": 0
    }
    execution_history.append(final_step)

    return {
        "messages": [response],
        "tool_calls_log": tool_calls_log,
        "execution_history": execution_history
    }

Далее создаем граф с checkpointer:

def create_checkpointed_graph():

    # Создаем checkpointer
    checkpointer = MemorySaver()

    # Создаем граф
    workflow = StateGraph(AgentState)

    # Добавляем узел агента
    workflow.add_node("agent", agent_node_with_logging)

    # Устанавливаем точку входа
    workflow.set_entry_point("agent")

    # Добавляем условную логику для завершения
    def should_continue(state: AgentState):
        last_message = state["messages"][-1]
        # Если последнее сообщение от агента и без tool_calls - завершаем
        if isinstance(last_message, AIMessage) and last_message.tool_calls:
            return "agent"
        return END

    workflow.add_conditional_edges("agent", should_continue)

    # Компилируем с checkpointer
    return workflow.compile(checkpointer=checkpointer), checkpointer

Извлекаем информацию о вызовах инструментов:

def extract_tool_calls_from_checkpoint(checkpointer, config):

    # Получаем состояние из checkpoint
    checkpoint_data = checkpointer.get(config)
    
    if checkpoint_data and "channel_values" in checkpoint_data:
        state = checkpoint_data["channel_values"]
    
        # Извлекаем tool calls
        tool_calls_log = state.get("tool_calls_log", [])
        execution_history = state.get("execution_history", [])
    
        return tool_calls_log, execution_history
    
    return [], []

Оцениваем качество работы агента:

def tool_tracking_with_checkpointer():
    # Тестовые случаи
    test_cases = [
        {
            "input": "Найди информацию о машинном обучении и создай файл с результатами",
            "expected_tools": [ToolCall(name="web_search"), ToolCall(name="file_manager")],
        },
        {
            "input": "Вычисли 15 * 8 + 25 и отправь результат на email [email protected]",
            "expected_tools": [ToolCall(name="calculator"), ToolCall(name="email_sender")],
        },
        {
            "input": "Найди документы о Python и удали ненужный файл temp.txt",
            "expected_tools": [ToolCall(name="document_retrieval"), ToolCall(name="file_manager")],
        }
    ]

    for i, test_case in enumerate(test_cases, 1):

        # Создаем граф и checkpointer
        app, checkpointer = create_checkpointed_graph()

        # Уникальная конфигурация для каждого теста
        config = {"configurable": {"thread_id": f"test_{i}"}}

        # Начальное состояние
        initial_state = {
            "messages": [HumanMessage(content=test_case["input"])],
            "tool_calls_log": [],
            "execution_history": []
        }

        # Выполняем с сохранением в checkpoint
        result = app.invoke(initial_state, config=config)

        # Получаем финальный ответ
        final_message = result["messages"][-1]
        actual_output = final_message.content if hasattr(final_message, 'content') else str(final_message)

        # Извлекаем tool calls из checkpoint
        tool_calls_from_checkpoint, execution_history = extract_tool_calls_from_checkpoint(checkpointer, config)

        # Преобразуем в формат для deepeval
        tools_called = []
        for tc in tool_calls_from_checkpoint:
            tools_called.append(ToolCall(
                name=tc["name"],
                input_parameters=tc.get("arguments", {})
            ))

        # Создаем test case для deepeval
        llm_test_case = LLMTestCase(
                input=test_case["input"],
                actual_output=actual_output,
                tools_called=tools_called,
                expected_tools=test_case["expected_tools"]
        )

        # Оцениваем с помощью ToolCorrectnessMetric
        tool_correctness = ToolCorrectnessMetric()
        tool_correctness.measure(llm_test_case)

        print(f"\nTool Correctness Score: {tool_correctness.score}")
        print(f"Reason: {tool_correctness.reason}")

        # Оцениваем с помощью TaskCompletionMetric
        task_completion = TaskCompletionMetric(threshold=0.7)
        task_completion.measure(llm_test_case)

        print(f"\nTask Completion Score: {task_completion.score}")
        print(f"Reason: {task_completion.reason}")

        # Оцениваем с помощью ArgumentCorrectnessMetric
        arg_correctness = ArgumentCorrectnessMetric(threshold=0.5)
        arg_correctness.measure(llm_test_case)

        print(f"\nArgument Correctness Score: {arg_correctness.score}")
        print(f"Reason: {arg_correctness.reason}")

Пример вывода по итогам оценки:

Tool Correctness Score: 0.5
Reason: Incomplete tool usage: missing tools [ToolCall(
name="file_manager"
), ToolCall(
name="web_search"
)]; expected ['web_search', 'file_manager'], called ['web_search'].
See more details above.

Task Completion Score: 1.0
Reason: The system correctly computed the expression 15 * 8 + 25,
obtaining 145, and prepared the result for sending to the specified email
address, fully aligning with the task requirements.

Argument Correctness Score: 1.0
Reason: Great job! The score is 1.00 because all tool calls were correct
and there were no issues with the input or process.

Для оценки работы агента не обязательно использовать checkpointer. Это лишь один из способов фиксации выполнения задач.
Например, результаты оценки выполнения задач AI агентом успешно передаются в LangSmith и выглядят следующим образом:

Turn Relevancy - показывает, может ли ваш чат-бот генерировать релевантные ответы постоянно на протяжении диалога.

Метрика оценивает каждое сообщение чат-бота на предмет релевантности созданному в ходе диалога контексту. Если ответ релевантен предыдущему диалогу, возвращает ‘yes’, в противном случае возвращает ‘no’.

По итогам всех сообщений бота вычисляет окончательный результат.

Role Adherence - показывает, может ли ваш чат-бот придерживаться заданной ему роли на протяжении диалога.

Метрика оценивает все сообщения чат-бота на предмет соответствия заданной ему роли. Каждому сообщению по итогам проверки присваивается значение от 0 до 1 и выносится вердикт, соответствует ли оно заданной роли.

Итоговый результат рассчитывается как отношение соответствующих заданной роли сообщений общему числу таких сообщений.

Knowledge Retention - показывает, насколько чат-бот способен сохранять полученную фактическую информацию на протяжении диалога.

Метрика на основании полученного диалога оценивает каждое сообщение ассистента на предмет того, утрачен ли контекст диалога или нет. Каждому сообщению присваивается значение ‘yes’ в случае, если контекст утрачен, и ‘no’, если контекст сохранен.

Результат рассчитывается как отношение числа сообщений без утраты контекста к общему числу сообщений.

Conversation Completeness - показывает, может ли чат-бот закончить диалог, удовлетворив при этом запрос пользователя.

Метрика подсчитывает количество запросов пользователей в ходе диалога и затем оценивает ответа чат-бота на предмет удовлетворения таких запросов. Каждому сообщению чат-бота присваивается значение ‘yes’ в случае, если запрос пользователя удовлетворен, и ‘no’ в обратном случае.

Все указанные выше метрики предназначены для оценки качества работы чат-бота в ходе диалога с пользователем.
Только метрика Turn Relevancy рассчитывается по итогам завершения диалога. Все остальные метрики рассчитываются во время диалога с чат-ботом.

Реализация в Python выглядит следующим образом:

from openai import OpenAI
from deepeval.test_case import ConversationalTestCase, LLMTestCase
from deepeval.metrics import RoleAdherenceMetric, KnowledgeRetentionMetric, ConversationCompletenessMetric, TurnRelevancyMetric

class Chatbot:
    def __init__(self, role_description):
        # Описание роли
        self.role_description = role_description
        # История диалога
        self.dialog_history = [{"role": "system", "content": role_description}]
        self.turns_list = []
        self.client = OpenAI()
        self.role_metric = RoleAdherenceMetric(verbose_mode=True)
        self.memory_metric = KnowledgeRetentionMetric(threshold=0.5)
        self.completeness_metric = ConversationCompletenessMetric(threshold=0.5)
        self.turn_relevancy_metric = TurnRelevancyMetric(threshold=0.5)

    def generate_response(self, user_input):
        try:
            # Добавляем пользовательский ввод в историю
            self.dialog_history.append({"role": "user", "content": user_input})

            # Ограничиваем историю до последних 20 сообщений + системное сообщение
            if len(self.dialog_history) > 20:
                self.dialog_history = [self.dialog_history[0]] + self.dialog_history[-19:]

            # Генерируем ответ с учетом всей истории диалога
            response = self.client.chat.completions.create(
                model="gpt-4o",
                messages=self.dialog_history
            )

            # Получаем ответ модели
            result = response.choices[0].message.content

            # Добавляем ответ чат-бота в историю
            self.dialog_history.append({"role": "assistant", "content": result})

            # Создаем LLMTestCase для метрик
            llm_test_case = LLMTestCase(input=user_input, actual_output=result)

            # Добавляем turns
            self.turns_list.append({"role": "user", "content": user_input})
            self.turns_list.append({"role": "assistant", "content": result})

            # Создаем ConversationalTestCase
            convo_test_case = ConversationalTestCase(chatbot_role=self.role_description, turns=self.turns_list)

            # Оценка метрик в реальном времени
            self.role_metric.measure(convo_test_case)
            self.memory_metric.measure(convo_test_case)
            self.completeness_metric.measure(convo_test_case)

            print(f"\nRole Adherence Score: {self.role_metric.score}, Reason: {self.role_metric.reason}")
            print(f"Knowledge Retention Score: {self.memory_metric.score}, Reason: {self.memory_metric.reason}")
            print(f"Conversation Completeness Score: {self.completeness_metric.score}, Reason: {self.completeness_metric.reason}\n")

            return result

        except Exception as e:
            print(f"Ошибка при генерации ответа: {e}")
            return "Извините, произошла ошибка при обработке вашего запроса."

    def evaluate_conversation(self):
        """Итоговая оценка диалога после завершения разговора"""
        # Создаем ConversationalTestCase для финальной оценки
        convo_test_case = ConversationalTestCase(chatbot_role=self.role_description, turns=self.turns_list)

        # Оценка Turn Relevancy метрики для всего диалога
        self.turn_relevancy_metric.measure(convo_test_case)

        print("\n" + "="*50)
        print("ИТОГОВАЯ ОЦЕНКА ДИАЛОГА")
        print("="*50)
        print(f"Role Adherence Score: {self.role_metric.score}")
        print(f"Reason: {self.role_metric.reason}\n")
        print(f"Knowledge Retention Score: {self.memory_metric.score}")
        print(f"Reason: {self.memory_metric.reason}\n")
        print(f"Conversation Completeness Score: {self.completeness_metric.score}")
        print(f"Reason: {self.completeness_metric.reason}\n")
        print(f"Turn Relevancy Score: {self.turn_relevancy_metric.score}")
        print(f"Reason: {self.turn_relevancy_metric.reason}")
        print("="*50)

chatbot = Chatbot(role_description="AI assistant for banking assistance")

while True:
    user_input = input("Вы: ")
    if user_input.lower() == "выход":
        # Оценка диалога при выходе
        chatbot.evaluate_conversation()
        break
    bot_response = chatbot.generate_response(user_input)
    print(f"Бот: {bot_response}")

Вывод в ходе выполнения кода выглядит следующим образом:

Вы: привет! можешь помочь рассчитать сумму ежемесячных платежей по кредиту?

Role Adherence Score: 1.0, Reason: The score is 1.0 because there are no out of character responses from the LLM chatbot; all responses adhere to the specified role of an AI assistant for banking assistance.
Knowledge Retention Score: 1.0, Reason: The score is 1.00 because there are no attritions, indicating no instances of forgetfulness or loss of previously established knowledge.
Conversation Completeness Score: 0.0, Reason: The score is 0.0 because the LLM did not provide any calculation or direct assistance with the monthly loan payment as requested by the user, instead only asking for more information. The user's intention to receive help with the calculation was not met at all.

Итоговый вывод выглядит так:

Role Adherence Score: 1.0
Reason: The score is 1.0 because there are no out of character responses from the LLM chatbot. All responses adhered to the specified role of an AI assistant for banking assistance throughout the conversation.
Knowledge Retention Score: 1.0
Reason: The score is 1.00 because there are no attritions, indicating no instances of forgetfulness or loss of previously established knowledge.
Conversation Completeness Score: 1.0
Reason: The score is 1.0 because the LLM response fully addressed the user's intention to calculate the monthly payment for a loan, with no incompletenesses identified.
Turn Relevancy Score: 1.0
Reason: The score is 1.0 because there are no irrelevancies listed, indicating that all assistant messages were fully relevant to the user messages.

MCP-use - показывает, насколько эффективно AI-агент c использованием MCP использует mcp серверы, к которым имеет доступ. Рассчитывается, исходя из оценки вызываемых инструментов (примитивов), а также аргументов, сгенерированных LLM приложением.

Метрика проверяет, соответствует ли причина вызова того или иного инструмента ожидаемой причине. Оценка ведется не строго, допускаются значения частичного соответствия (например, 0.25, 0.5, 0.75). Итоговый результат расчета сопровождается указанием причины, по которой дана та или иная оценка.

Multi-Turn MCP-Use - показывает, насколько эффективно AI-агент c использованием MCP использует mcp серверы, к которым имеет доступ. Рассчитывается как отношение вызываемых инструментов (примитивов), а также аргументов, сгенерированных LLM приложением, к общему числу взаимодействий с MCP.

Метрика показывает, насколько соответствует использование mcp сервером имеющихся устройств ожидаемым использованиям. Рассчитывается как общее значение соответствия по отношению ко всем запросам MCP сервера.

MCP Task Completion - показывает, насколько эффективно AI-агент с использованием MCP выполняет задачу. Рассчитывается как отношение корректно выполненных запросов в каждом взаимодействии к общему числу взаимодействий.

Метрика показывает, насколько каждое взаимодействие mcp сервера с устройствами было оптимальным с точки зрения вызова нужного устройства из списка имеющихся, а также с позиции правильно переданного промта для устройства. Допускаются частичные соответствия (например, 0.25, 0.5, 0.75). Итоговый результат расчета сопровождается указанием причины, по которой дана та или иная оценка.

Реализация в Python

Сами метрики рассчитываются следующим образом:

from deepeval.metrics import MCPUseMetric, MCPTaskCompletionMetric, MultiTurnMCPUseMetric
from deepeval.test_case import LLMTestCase, MCPServer, ConversationalTestCase, Turn
from deepeval.test_case.mcp import MCPToolCall

# Создание ConversationalTestCase
convo_test = ConversationalTestCase(
                turns=turns, # Список взаимодействий в ходе использования агента
                mcp_servers=[mcp_server] # Список MCP серверов
            )
# Расчет метрики MCPTaskCompletionMetric            
task_metric = MCPTaskCompletionMetric(threshold=0.7)
task_metric.measure(convo_test)
print(f"MCP Task Completion Score: {task_metric.score:.2f}")
print(f" Reason:    {task_metric.reason}")

# Расчет метрики MultiTurnMCPUseMetric
multiturn_metric = MultiTurnMCPUseMetric(threshold=0.5)
multiturn_metric.measure(convo_test)
print(f"Multi-turn MCP-use Score: {multiturn_metric.score:.2f}")
print(f" Reason:    {multiturn_metric.reason}")

# Создание LLMTestCase
llm_test_case = LLMTestCase(
    input=scenario['input'], # Запрос пользователя
    actual_output=response, # Итоговый ответ
    mcp_servers=[mcp_server], # Список MCP серверов
    mcp_tools_called=mcp_tool_calls # Список использованных инструментов
)

mcp_use_metric = MCPUseMetric(threshold=0.5)
mcp_use_metric.measure(llm_test_case)
print(f"MCP Use Score: {mcp_use_metric.score:.2f}")
print(f" Reason:    {mcp_use_metric.reason}")

Реализация AI агента с MCP сервером выглядит следующим образом:

Импортируем все необходимые модули. На момент написания статьи в метриках MCPTaskCompletionMetric и MultiTurnMCPUseMetric были обнаружены неисправности в коде (ошибка с делением на 0 при расчете результата оценки), поэтому были созданы собственные версии метрик с исправлениями.

import asyncio
import logging
import json
from typing import Dict, List, Any, Optional, Annotated
from pathlib import Path
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, AIMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from deepeval.metrics import MCPUseMetric, MCPTaskCompletionMetric, MultiTurnMCPUseMetric
from deepeval.test_case import LLMTestCase, MCPServer, ConversationalTestCase, Turn
from deepeval.test_case.mcp import MCPToolCall
from mcp.types import Tool, CallToolResult, TextContent

Настраиваем логирование и определяем состояние агента

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-agent-checkpointer")

class MCPAgentState(TypedDict):
    """Состояние MCP агента с полной информацией о взаимодействиях"""
    messages: Annotated[List[BaseMessage], add_messages]
    mcp_interactions: List[Dict[str, Any]]  # История вызовов MCP инструментов
    execution_history: List[Dict[str, Any]]  # История выполнения шагов
    current_task: Optional[str]  # Текущая задача

Создаем класс со всеми необходимыми методами для обеспечения работы MCP агента

class MCPAgentWithCheckpointer:
    """MCP агент с LangGraph Checkpointer для сохранения состояния"""

    def __init__(self, server_command: List[str], model_name: str = "gpt-3.5-turbo"):
        self.server_command = server_command # Сохраняет список строк для MCP сервера 
        self.llm = ChatOpenAI(temperature=0, model=model_name) # Экземпляр OpenAI
        self.session: Optional[ClientSession] = None # Атрибут для MCP сессии
        self.transport_context = None # Хранит контекстный менеджер для stdio_client
        self.available_tools: List[Tool] = [] # список доступных MCP инструментов
        self.checkpointer = MemorySaver()  # Создаем checkpointer
        self.graph = None # Атрибут для скомпилированного графа

    async def connect_to_server(self):
        """Подключение к MCP серверу"""
        logger.info(f"Подключение к MCP серверу: {' '.join(self.server_command)}")
        # Создание параметров для запуска MCP сервера
        server_params = StdioServerParameters(
            command=self.server_command[0], # Исполняемый файл
            args=self.server_command[1:] if len(self.server_command) > 1 else [] # Аргументы команды
        )

        # Создаем контекстный менеджер для транспорта
        self.transport_context = stdio_client(server_params)
        read_stream, write_stream = await self.transport_context.__aenter__()

        # Создаем сессию из потоков
        self.session = ClientSession(read_stream, write_stream)
        await self.session.__aenter__()

        # Инициализация
        await self.session.initialize()

        # Получение списка инструментов
        tools_result = await self.session.list_tools()
        self.available_tools = tools_result.tools

        logger.info(f"Подключение установлено. Доступно инструментов: {len(self.available_tools)}")
        for tool in self.available_tools:
            logger.info(f"{tool.name}: {tool.description}")

        # Создаем граф после подключения
        self._build_graph()

        return True


    async def disconnect(self):
        """Отключение от MCP сервера"""
        if self.session:
            await self.session.__aexit__(None, None, None)
        if hasattr(self, 'transport_context'):
            await self.transport_context.__aexit__(None, None, None)
        logger.info("Отключение от MCP сервера")

    async def call_mcp_tool(self, tool_name: str, arguments: Dict[str, Any]) -> tuple[CallToolResult, str]:
        """Вызов инструмента MCP сервера"""

        logger.info(f"Вызов инструмента {tool_name} с аргументами: {arguments}")
        result = await self.session.call_tool(tool_name, arguments)

        # Извлекаем текстовое содержимое результата
        result_text = ""
        if result.content:
            for content in result.content:
                if isinstance(content, TextContent):
                    result_text += content.text + "\n"

        logger.info(f"Инструмент {tool_name} выполнен успешно")
        return result, result_text

    def _create_tools_description(self) -> str:
        """Создает описание доступных инструментов для LLM"""
        description = "Доступные MCP инструменты:\n\n"
        for tool in self.available_tools:
            description += f" **{tool.name}**: {tool.description}\n"
            if tool.inputSchema and 'properties' in tool.inputSchema:
                description += "   Параметры:\n"
                for param, details in tool.inputSchema['properties'].items():
                    required = param in tool.inputSchema.get('required', [])
                    req_marker = "*" if required else ""
                    description += f"   - {param}{req_marker}: {details.get('description', 'Нет описания')}\n"
            description += "\n"

        return description

    def _build_graph(self):
        """Построение LangGraph с checkpointer"""

        # Создаем граф
        workflow = StateGraph(MCPAgentState)

        # Добавляем узел агента
        workflow.add_node("agent", self._agent_node)

        # Устанавливаем точку входа
        workflow.set_entry_point("agent")

        # Добавляем условную логику для завершения
        def should_continue(state: MCPAgentState):
            """Определяет, нужно ли продолжать выполнение"""
            messages = state["messages"]
            if not messages:
                return END

            last_message = messages[-1]

            # Если в последнем сообщении есть вызовы инструментов - продолжаем
            if isinstance(last_message, AIMessage) and hasattr(last_message, 'content'):
                if "TOOL_CALL:" in last_message.content:
                    return "agent"

            return END

        workflow.add_conditional_edges("agent", should_continue)

        # Компилируем с checkpointer
        self.graph = workflow.compile(checkpointer=self.checkpointer)

    async def _agent_node(self, state: MCPAgentState) -> Dict[str, Any]:
        """Узел агента с обработкой MCP инструментов"""

        messages = state["messages"]
        mcp_interactions = state.get("mcp_interactions", [])
        execution_history = state.get("execution_history", [])

        # Создаем системное сообщение с описанием инструментов
        tools_description = self._create_tools_description()

        # ПРОМПТ
        system_message = SystemMessage(content=f"""
Промпт для работы AI агента с доступом к MCP инструментам. """)

        # Формируем контекст для LLM
        all_messages = [system_message] + messages

        # Получаем ответ от LLM
        response = await self.llm.ainvoke(all_messages)
        response_content = response.content

        # Логируем шаг выполнения
        execution_step = {
            "step": len(execution_history) + 1,
            "action": "llm_response",
            "has_tool_calls": "TOOL_CALL:" in response_content,
            "timestamp": datetime.now().isoformat()
        }
        execution_history.append(execution_step)

        # Обрабатываем вызовы инструментов
        if "TOOL_CALL:" in response_content:
            new_messages = [response]

            # Извлекаем все вызовы инструментов из ответа
            lines = response_content.split('\n')
            for line in lines:
                if "TOOL_CALL:" in line:
                    # Парсим вызов инструмента
                    parts = line.strip().split(":", 2)
                    if len(parts) != 3 or parts[0] != "TOOL_CALL":
                        continue

                    tool_name = parts[1]
                    arguments_json = parts[2]
                    arguments = json.loads(arguments_json)

                    # Вызываем инструмент
                    result, result_text = await self.call_mcp_tool(tool_name, arguments)

                    # Записываем взаимодействие
                    mcp_interaction = {
                        "name": tool_name,
                        "args": arguments,
                        "result": result,
                        "result_text": result_text,
                        "timestamp": datetime.now().isoformat()
                    }
                    mcp_interactions.append(mcp_interaction)

                    # Добавляем результат в сообщения
                    result_message = AIMessage(content=f"Результат {tool_name}: {result_text}")
                    new_messages.append(result_message)

                    # Логируем выполнение инструмента
                    tool_step = {
                        "step": len(execution_history) + 1,
                        "action": f"tool_executed_{tool_name}",
                        "status": "success",
                        "timestamp": datetime.now().isoformat()
                    }
                    execution_history.append(tool_step)

            tools_results_summary = "\n\nВЫПОЛНЕНЫЕ ДЕЙСТВИЯ И ИХ РЕЗУЛЬТАТЫ:\n"
            for idx, interaction in enumerate(mcp_interactions, 1):
                tools_results_summary += f"{idx}. Инструмент '{interaction['name']}' с аргументами {interaction['args']}\n"
                tools_results_summary += f"   Результат: {interaction['result_text']}\n"

            continuation_prompt = SystemMessage(content=f"""Промпт для продолжения или окончания работы AI агента с доступом к MCP инструментам""")

            final_messages = all_messages + new_messages + [continuation_prompt]
            final_response = await self.llm.ainvoke(final_messages)
            new_messages.append(final_response)

            return {
                "messages": new_messages,
                "mcp_interactions": mcp_interactions,
                "execution_history": execution_history,
                "current_task": state.get("current_task")
            }

        # Если вызовов инструментов нет - просто возвращаем ответ
        return {
            "messages": [response],
            "mcp_interactions": mcp_interactions,
            "execution_history": execution_history,
            "current_task": state.get("current_task"),
        }

    async def process_request(self, user_input: str, thread_id: str = "default",) -> tuple[str, MCPAgentState]:
        """Обработка запроса пользователя с сохранением в checkpoint"""

        # Конфигурация для checkpoint
        config = {"configurable": {"thread_id": thread_id}}

        # Начальное состояние
        initial_state: MCPAgentState = {
            "messages": [HumanMessage(content=user_input)],
            "mcp_interactions": [],
            "execution_history": [],
            "current_task": user_input,
        }

        try:
            # Выполняем граф с сохранением в checkpoint
            result = await self.graph.ainvoke(initial_state, config=config)

            # Получаем финальный ответ
            final_message = result["messages"][-1]
            final_response = final_message.content if hasattr(final_message, 'content') else str(final_message)

            return final_response, result

        except Exception as e:
            logger.error(f"Ошибка обработки запроса: {e}")
            return f"Произошла ошибка: {str(e)}", initial_state

    def get_checkpoint_state(self, thread_id: str = "default") -> Optional[MCPAgentState]:
        """Получение сохраненного состояния из checkpoint"""
        config = {"configurable": {"thread_id": thread_id}}

        checkpoint_data = self.checkpointer.get(config)

        if checkpoint_data and "channel_values" in checkpoint_data:
            return checkpoint_data["channel_values"]

    def get_mcp_server_definition(self) -> MCPServer:
        """Создаем объект MCPServer для метрик DeepEval"""
        return MCPServer(
            server_name="mcp-tools-server",
            available_tools=self.available_tools
        )

Тестируем MCP агент

async def test_mcp_with_checkpointer():
    """Тестирование MCP агента с Checkpointer"""

    # Путь к серверу
    server_path = Path("путь к MCP серверу")
    server_command = ["python", str(server_path)]

    # Инициализация MCP агента
    agent = MCPAgentWithCheckpointer(server_command)

    # Пробуем прогнать тесты
    try:
        # Проверяем, есть ли подключение к серверу
        if not await agent.connect_to_server():
            print("Не удалось подключиться к MCP серверу")
            return

        # Тестовые сценарии
        test_scenarios = [
            {
                "name": "Файловые операции (базовый промпт)",
                "input": "Создай файл test.txt с содержимым 'Hello MCP!' и прочитай его",
                "description": "Тест базовых файловых операций с базовым промптом",
                "thread_id": "test_1"
            },
            {
                "name": "Калькулятор (базовый промпт)",
                "input": "Вычисли (25 * 4 + 10) / 3 и сохрани результат в файл calc_result.txt",
                "description": "Тест калькулятора и сохранения с базовым промптом",
                "thread_id": "test_2"
            },
            {
                "name": "Файловые операции (улучшенный промпт)",
                "input": "Создай файл test.txt с содержимым 'Hello MCP!' и прочитай его",
                "description": "Тест базовых файловых операций с улучшенным промптом",
                "thread_id": "test_3"
            }
        ]
        # Последовательно запускаем все указанные выше тест кейсы
        for i, scenario in enumerate(test_scenarios, 1):
            print(f"\n--- Тест {i}: {scenario['name']} ---")
            print(f"Описание: {scenario['description']}")
            print(f"Ввод: {scenario['input']}")

            # Выполняем запрос
            response, state = await agent.process_request(
                scenario['input'],
                thread_id=scenario['thread_id'],
            )
            print(f"Вывод: {response}")

            # Получаем состояние из checkpoint
            checkpoint_state = agent.get_checkpoint_state(scenario['thread_id'])
            if checkpoint_state:
                # Показываем детали MCP вызовов
                mcp_interactions = checkpoint_state.get('mcp_interactions', [])
                if mcp_interactions:
                    print(f"\nДетали MCP вызовов:")
                    for idx, interaction in enumerate(mcp_interactions, 1):
                        print(f"   {idx}. {interaction['name']} с аргументами {interaction['args']}")

            # Создаем тестовый случай для метрик
            mcp_server = agent.get_mcp_server_definition()

            # Преобразуем mcp_interactions в MCPToolCall
            mcp_tool_calls = []
            for interaction in state.get('mcp_interactions', []):
                mcp_call = MCPToolCall(
                    name=interaction['name'],
                    args=interaction['args'],
                    result=interaction['result']
                )
                mcp_tool_calls.append(mcp_call)

            # Готовим данные для передачи в ConversationalTestCase
            if len(mcp_tool_calls) > 1:
                # Multi-turn
                turns = [Turn(role="user", content=scenario['input'])]

                for idx, mcp_call in enumerate(mcp_tool_calls):
                    result_preview = ""
                    if mcp_call.result.content:
                        for content_item in mcp_call.result.content:
                            if hasattr(content_item, 'text'):
                                result_preview = content_item.text[:100]
                                break

                    assistant_response = f"Вызываю инструмент {mcp_call.name} с аргументами: {mcp_call.args}\nРезультат: {result_preview}"

                    turns.append(Turn(
                        role="assistant",
                        content=assistant_response,
                        mcp_tools_called=[mcp_call]
                    ))

                    if idx < len(mcp_tool_calls) - 1:
                        turns.append(Turn(role="user", content="Продолжай выполнение задачи"))

                turns.append(Turn(role="assistant", content=response))
            else:
                # Single-turn
                turns = [
                    Turn(role="user", content=scenario['input']),
                    Turn(
                        role="assistant",
                        content=response,
                        mcp_tools_called=mcp_tool_calls
                    )
                ]

            # Создаем ConversationalTestCase
            convo_test = ConversationalTestCase(
                turns=turns,
                mcp_servers=[mcp_server]
            )

            print(f"\nМЕТРИКИ DEEPEVAL:")
            print("-" * 50)

            # 1. MCPTaskCompletionMetric
            if MCPTaskCompletionMetric:
                task_metric = MCPTaskCompletionMetric(threshold=0.7)
                task_metric.measure(convo_test)
                print(f"MCP Task Completion Score: {task_metric.score:.2f}")
                print(f" Reason:    {task_metric.reason}")

            # 2. MultiTurnMCPUseMetric
            if MultiTurnMCPUseMetric:
                multiturn_metric = MultiTurnMCPUseMetric(threshold=0.5)
                multiturn_metric.measure(convo_test)
                print(f"Multi-turn MCP-use Score: {multiturn_metric.score:.2f}")
                print(f" Reason:    {multiturn_metric.reason}")

            # 3. MCPUseMetric
            llm_test_case = LLMTestCase(
                input=scenario['input'],
                actual_output=response,
                mcp_servers=[mcp_server],
                mcp_tools_called=mcp_tool_calls
            )

            mcp_use_metric = MCPUseMetric(threshold=0.5)
            mcp_use_metric.measure(llm_test_case)
            print(f"MCP Use Score: {mcp_use_metric.score:.2f}")
            print(f" Reason:    {mcp_use_metric.reason}")

    except Exception as e:
        logger.error(f"Ошибка тестирования: {e}")
        print(f"Ошибка тестирования: {e}")
        import traceback
        traceback.print_exc()

    finally:
        await agent.disconnect()

Для оценки работы MCP агента не обязательно использовать checkpointer. Это лишь один из способов фиксации выполнения задач.
Например, результаты оценки выполнения задач MCP агентом успешно передаются в LangSmith и выглядят следующим образом:

BIAS - показывает, есть ли в вашей AI модели гендерные, расовые или политические предубеждения

Метрика показывает, какое количество содержащихся в ответе модели мнений относятся к предубеждениям. Сначала из ответа извлекаются все содержащиеся в нем мнения, потом каждое из мнений оценивается на предмет отношения к предубеждениям. Принимает значения от 0 до 1.

Значение 1 - ответ содержит предубеждения
Значение 0 - ответ без предубеждений

Реализация в Python

import pytest
from deepeval.metrics import BiasMetric
from deepeval.test_case import LLMTestCase
from deepeval import assert_test
from deepeval.dataset import EvaluationDataset

# Инициализируем набор данных
dataset = EvaluationDataset()

question = question # Вопрос пользователя
answer = answer # Ответ модели

# Составляем тест кейс
test_case = LLMTestCase(input=question, actual_output=answer)

# Дополняем набор данных тест кейсом
dataset.add_test_case(test_case)

# Передаем параметры и вычисляем результат
@pytest.mark.asyncio
@pytest.mark.parametrize(
    "test_case",
    dataset.test_cases,
)
async def test_bias_metric(test_case: LLMTestCase):
    metric = BiasMetric(threshold=0.5) # Пороговое значение метрики
    assert_test(test_case, [metric])

Toxicity - призвана определять наличие токсичности в ответах AI-модели.

Метрика показывает, какое количество содержащихся в ответе модели мнений относятся к токсичным. Сначала из ответа извлекаются все содержащиеся в нем мнения, потом каждое из мнений оценивается на предмет отношения к токсичным мнениям. Принимает значения от 0 до 1.

Значение 1 - ответ содержит токсичные мнения
Значение 0 - ответ без токсичных мнений

Реализация в Python

import pytest
from deepeval.metrics import ToxicityMetric
from deepeval.test_case import LLMTestCase
from deepeval import assert_test
from deepeval.dataset import EvaluationDataset

# Инициализируем набор данных
dataset = EvaluationDataset()

question = question # Вопрос пользователя
answer = answer # Ответ модели

# Составляем тест кейс
test_case = LLMTestCase(input=question, actual_output=answer)

# Дополняем набор данных тест кейсом
dataset.add_test_case(test_case)

# Передаем параметры и вычисляем результат
@pytest.mark.asyncio
@pytest.mark.parametrize(
    "test_case",
    dataset.test_cases,
)
async def test_toxicity_metric(test_case: LLMTestCase):
    metric = ToxicityMetric(threshold=0.5) # Пороговое значение метрики
    assert_test(test_case, [metric])

Non-Advice - показывает, содержатся ли в ответах модели неподходящие профессиональные рекомендации, которых следует избегать.

Метрика показывает, какое количество содержащихся в ответе советов является допустимыми. Сначала из ответа извлекаются все содержащиеся в нем советы, потом каждый из советов оценивается на предмет нарушения ограничения на дачу рекомендаций. Список ограничений по даче рекомендаций указан в промте для оценки, при необходимости может быть изменен.
Принимает значения от 0 до 1.

Значение 1 - ответ не содержит нарушений ограничений по даче рекомендаций
Значение 0 - ответ содержит нарушения ограничений по даче рекомендаций

Реализация в Python

import pytest
from deepeval.metrics import NonAdviceMetric
from deepeval.test_case import LLMTestCase
from deepeval import assert_test
from deepeval.dataset import EvaluationDataset

# Инициализируем набор данных
dataset = EvaluationDataset()

question = question # Вопрос пользователя
answer = answer # Ответ модели

# Составляем тест кейс
test_case = LLMTestCase(input=question, actual_output=answer)

# Дополняем набор данных тест кейсом
dataset.add_test_case(test_case)

# Передаем параметры и вычисляем результат
@pytest.mark.asyncio
@pytest.mark.parametrize(
    "test_case",
    dataset.test_cases,
)
async def test_non_advice_metric(test_case: LLMTestCase):
    metric = NonAdviceMetric(advice_types = [advice], # Определение сферы деятельности для ограничения рекомендаций 
                             threshold=0.5) # Пороговое значение метрики
    assert_test(test_case, [metric])

Misuse - определяет, есть ли случаи использования вашего чат-бота не по назначению.

Метрика показывает количество допустимых использований чат-бота пользователями.

Сначала из ответа формируется список потенциальных использований чат-бота не по назначению, затем происходит анализ каждого потенциального случая с целью выявить, действительно ли такое использование было. Если есть факт использования не по назначению, такому случаю присваивается значение ‘yes’, в противном случае присваивается ‘no’.

Итоговое значение рассчитывается как отношение неверных использований чат-бота к общему числу использований.

Принимает значения от 0 до 1.

Значение 1 - ответ содержит использования не по назначению
Значение 0 - ответ содержит использований не по назначению

Реализация в Python

import pytest
from deepeval.metrics import MisuseMetric
from deepeval.test_case import LLMTestCase
from deepeval import assert_test
from deepeval.dataset import EvaluationDataset

# Инициализируем набор данных
dataset = EvaluationDataset()

question = question # Вопрос пользователя
answer = answer # Ответ чат-бота

# Составляем тест кейс
test_case = LLMTestCase(input=question, actual_output=answer)

# Дополняем набор данных тест кейсом
dataset.add_test_case(test_case)

# Передаем параметры и вычисляем результат
@pytest.mark.asyncio
@pytest.mark.parametrize(
    "test_case",
    dataset.test_cases,
)
async def test_misuse_metric(test_case: LLMTestCase):
    metric = MisuseMetric(domain = "domain", # Определение сферы компетенции чат-бота
                             threshold=0.5) # Пороговое значение метрики
    assert_test(test_case, [metric])

PII Leakage - определяет, есть ли случаи использования конфиденциальной информации в ответах AI модели.

Метрика показывает процент утверждений без использования или упоминания персональных данных.

Сначала из ответа формируется список потенциальных использований или упоминаний персональных данных, затем происходит анализ каждого потенциального случая с целью выявить, действительно ли такое использование было. Если есть факт использования данных, такому случаю присваивается значение ‘yes’, в противном случае присваивается ‘no’.

Итоговое значение рассчитывается как отношение использований чат-бота без указания персональной информации к общему числу потенциальных использований персональных данных.

Принимает значения от 0 до 1.

Значение 1 - ответ не содержит персональных или конфиденциальных данных
Значение 0 - ответ содержит персональные или конфиденциальные данные

Реализация в Python

import pytest
from deepeval.metrics import PIILeakageMetric
from deepeval.test_case import LLMTestCase
from deepeval import assert_test
from deepeval.dataset import EvaluationDataset

# Инициализируем набор данных
dataset = EvaluationDataset()

question = question # Вопрос пользователя
answer = answer # Ответ чат-бота

# Составляем тест кейс
test_case = LLMTestCase(input=question, actual_output=answer)

# Дополняем набор данных тест кейсом
dataset.add_test_case(test_case)

# Передаем параметры и вычисляем результат
@pytest.mark.asyncio
@pytest.mark.parametrize(
    "test_case",
    dataset.test_cases,
)
async def test_pii_leakage_metric(test_case: LLMTestCase):
    metric = PIILeakageMetric(threshold=0.5) # Пороговое значение метрики
    assert_test(test_case, [metric])

Role Violation - определяет, есть ли случаи нарушения AI-моделью прописанных в настройках роли или образа.

Метрика показывает, были ли случаи выхода чат-бота из прописанной для него роли.

Если был хотя бы один такой случай, возвращается значение 1, в остальных случаях возвращает 0.

Реализация в Python

import pytest
from deepeval.metrics import RoleViolationMetric
from deepeval.test_case import LLMTestCase
from deepeval import assert_test
from deepeval.dataset import EvaluationDataset

# Инициализируем набор данных
dataset = EvaluationDataset()

question = question # Вопрос пользователя
answer = answer # Ответ чат-бота
bot_role = role # Роль чат-бота

# Составляем тест кейс
test_case = LLMTestCase(input=question, actual_output=answer)

# Дополняем набор данных тест кейсом
dataset.add_test_case(test_case)

# Передаем параметры и вычисляем результат
@pytest.mark.asyncio
@pytest.mark.parametrize(
    "test_case",
    dataset.test_cases,
)
async def test_role_violation_metric(test_case: LLMTestCase):
    metric = RoleViolationMetric(role=bot_role, # Указание роли бота 
                                 threshold=0.5) # Пороговое значение метрики
    assert_test(test_case, [metric])

Итоговый вывод выглядит так:

MCP Task Completion Score: 0.75
Reason: [
Score: 0.5
Reason: The agent confirmed that the file 'test.txt' was created with the correct content, but did not display the contents of the file as requested by the user.
]

Multi-turn MCP-use Score: 1.00
Reason: []

MCP Use Score: 1.00
Reason: [
The user asked to create a file 'test.txt' with specific content and then read it. The agent used 'file_writer' to create and write the file, and 'file_reader' to read its contents. These are the correct and most appropriate tools for the task, with no unnecessary or missing tool calls.
All arguments passed to the tools were correct and well-formed. For 'file_writer', 'file_path' was set to 'test.txt', 'content' to 'Hello MCP!', and 'mode' to 'write', matching the tool's schema and the user's request to create the file with specific content. For 'file_reader', 'file_path' was set to 'test.txt', which is the correct required argument to read the file just created. No required arguments were missing or malformed, and all values were appropriate for the intended actions.
]

Summarization - определяет, может ли AI-модель генерировать фактически краткое содержание материала с упоминанием всех необходимых фактов.

Метрика показывает, насколько каждое утверждение из краткого содержания соотносится с исходным текстом и покрывает всю необходимую информацию.

На основании исходного текста генерируется список вопросов с возможными ответами да или нет.
Затем формируется список утверждений из краткого содержания и происходит проверка качества формирования путем оценки ответов на ранее сгенерированные вопросы.
Итоговое значение находится в диапазоне от 0 до 1.

Значение 1 - краткое содержание соответствует критериям оценки
Значение 0 - краткое содержание не соответствует критериям оценки.

Реализация в Python

import pytest
from deepeval.metrics import SummarizationMetric
from deepeval.test_case import LLMTestCase
from deepeval import assert_test
from deepeval.dataset import EvaluationDataset

# Инициализируем набор данных
dataset = EvaluationDataset()

answer = answer # Ответ модели
summary = summary # Краткое содержание ответа

# Составляем тест кейс
test_case = LLMTestCase(input=answer, actual_output=summary)

# Дополняем набор данных тест кейсом
dataset.add_test_case(test_case)

# Передаем параметры и вычисляем результат
@pytest.mark.asyncio
@pytest.mark.parametrize(
    "test_case",
    dataset.test_cases,
)
async def test_summarization_metric(test_case: LLMTestCase):
    metric = SummarizationMetric(
                                 threshold=0.5 # Пороговое значение метрики
                                 # Список критериев оценки краткого содержания
                                 assessment_questions=[assessment_questions])  
    assert_test(test_case, [metric])

Prompt Alignment - определяет, может ли ваша AI-модель генерировать ответы, соответствующие прописанным инструкциям.

Метрика показывает, насколько точно ваша модель следует прописанным для нее инструкциям. Проводится проверка на предмет соответствия ответа всем прописанным инструкциям. Если хотя бы один пункт инструкций не соблюден, такому ответу присваивается значение ‘no’, в противном случае присваивается значение ‘yes’.

Итоговое значение находится в диапазоне от 0 до 1.

Значение 1 - ответ соответствует промту
Значение 0 - ответ не соответствует промту

Реализация в Python

import pytest
from deepeval.metrics import PromptAlignmentMetric
from deepeval.test_case import LLMTestCase
from deepeval import assert_test
from deepeval.dataset import EvaluationDataset

# Инициализируем набор данных
dataset = EvaluationDataset()

prompt = prompt # Промт
answer = answer # Ответ модели

# Составляем тест кейс
test_case = LLMTestCase(input=prompt, actual_output=answer)

# Дополняем набор данных тест кейсом
dataset.add_test_case(test_case)

# Передаем параметры и вычисляем результат
@pytest.mark.asyncio
@pytest.mark.parametrize(
    "test_case",
    dataset.test_cases,
)
async def test_prompt_alignment_metric(test_case: LLMTestCase):
    metric = PromptAlignmentMetric(
                                 threshold=0.7 # Пороговое значение метрики
                                 # Список инструкций промта 
                                 prompt_instructions=[prompt_instructions])  
    assert_test(test_case, [metric])

Hallucination - проверяет, может ли ваша AI-модель генерировать фактически верную информацию на основании представленного контекста.

Метрика показывает, насколько ответ модели соответствует контекстам.

Для списка контекстов проводится оценка, соответствует ли ответ модели каждому из предоставленных контекстов. Если есть соответствие, такому ответу присваивается значение ‘yes’, в противном случае присваивается значение ‘no’.

Итоговое значение находится в диапазоне от 0 до 1.

Значение 1 - ответ не соответствует контексту
Значение 0 - ответ соответствует контексту

Реализация в Python

import pytest
from deepeval.metrics import HallucinationMetric
from deepeval.test_case import LLMTestCase
from deepeval import assert_test
from deepeval.dataset import EvaluationDataset

# Инициализируем набор данных
dataset = EvaluationDataset()

question = question # Вопрос
answer = answer # Ответ модели
context = context # Сопровождающий контекст

# Составляем тест кейс
test_case = LLMTestCase(input=prompt, actual_output=answer, context=context)

# Дополняем набор данных тест кейсом
dataset.add_test_case(test_case)

# Передаем параметры и вычисляем результат
@pytest.mark.asyncio
@pytest.mark.parametrize(
    "test_case",
    dataset.test_cases,
)
async def test_hallucination_metric(test_case: LLMTestCase):
    metric = HallucinationMetric(
                                 threshold=0.7 # Пороговое значение метрики
                                 )  
    assert_test(test_case, [metric])

Json Correctness - определяет, способна ли ваша AI-модель генерировать ответ в заданном формате JSON-схемы.

Метрика оценивает, соответствует ли заданной схеме сгенерированный JSON. Возвращает 1, если ответ соответствует схеме, и 0, если соответствия нет.

Реализация в Python

import pytest
from deepeval.metrics import JsonCorrectnessMetric
from deepeval.test_case import LLMTestCase
from deepeval import assert_test
from deepeval.dataset import EvaluationDataset

# Инициализируем набор данных
dataset = EvaluationDataset()

question = question # Вопрос
answer = answer # Ответ модели
json_schema = json_schema # Ожидаемая схема ответа в форме Pydantic модели

# Составляем тест кейс
test_case = LLMTestCase(input=prompt, actual_output=answer)

# Дополняем набор данных тест кейсом
dataset.add_test_case(test_case)

# Передаем параметры и вычисляем результат
@pytest.mark.asyncio
@pytest.mark.parametrize(
    "test_case",
    dataset.test_cases,
)
async def test_json_correctness_metric(test_case: LLMTestCase):
    metric = JsonCorrectnessMetric(
                                 threshold=0.7,  # Пороговое значение метрики
                                 expected_schema=json_schema
                                 )  
    assert_test(test_case, [metric])

Image Coherence - показывает, насколько визуальный контент соответствует текстовому описанию.

Individual Image Coherence

Final Score

Для оценки представляется изображение и его текстовое описание. Далее происходит оценка по 10-балльной шкале в следующем диапазоне:

• 0-3 - отсутствие или минимальное соответствие;

• 4-6 - изображение содержит элементы описания, но также содержит не относящиеся к описанию элементы;

• 7-9 - означает высокую степень соответствия изображения описанию

• 10 - означает полное соответствие описанию.

Итоговое значение находится в диапазоне от 0 до 1.

Значение 1 - визуальный контент соответствует текстовому описанию
Значение 0 - визуальный контент не соответствует текстовому описанию

Реализация в Python

from deepeval.metrics import ImageCoherenceMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question] # Запрос пользователя
answer = [
            "Text", # Текст ответа
            # Передаем данные о местонахождении картинки в MLLMImage
            MLLMImage(url="./image", # Местонахождение картинки  
                      local=True) # Указание о локальном расположении
        ]

def test_image_coherence_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, actual_output=answer)
    # Определяем метрику и ее пороговое значение
    metric = ImageCoherenceMetric(threshold=0.5)
    # Проводим проверку
    assert_test(test_case, [metric])

Image Helpfulness - показывает, насколько визуальные изображения помогают пользователям понять текст.

Individual Image Helpfulness

Final Score

Для оценки представляется изображение и текст, который такое изображение должно сопровождать. Далее происходит оценка по 10-балльной шкале в следующем диапазоне:

• 0-3 - отсутствие или минимальный уровень помощи в понимании текста;

• 4-6 - изображение содержит полезный контекст, но также содержит не относящиеся к тексту или менее важные детали;

• 7-9 - означает высокую степень помощи описанию

• 10 - означает идеальное дополнение и пояснение сопровождаемого текста.

Итоговое значение находится в диапазоне от 0 до 1.

Значение 1 - визуальный контент полезен для понимания текста
Значение 0 - визуальный контент не способствует пониманию текста

Реализация в Python

from deepeval.metrics import ImageHelpfulnessMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question] # Запрос пользователя
answer = [
            "Text", # Текст ответа
            # Передаем данные о местонахождении картинки в MLLMImage
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

def test_image_helpfulness_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, actual_output=answer)
    # Определяем метрику и ее пороговое значение
    metric = ImageHelpfulnessMetric(threshold=0.5)
    # Проводим проверку
    assert_test(test_case, [metric])

Image Reference - показывает, насколько точно изображения относятся к или объясняются текстом.

Individual Image Reference

Final Score

Для оценки представляется изображение и текст, который такое изображение должно сопровождать. Далее происходит оценка по 10-балльной шкале в следующем диапазоне:

• 0 - отсутствие упоминания или сопровождения изображения текстом;

• 1-3 - есть неявная ссылка на изображение, а также неправильное или некорректное сопровождение текстом

• 4-6 - явная, но неправильная ссылка на изображение, а также неявное сопровождение текстом;

• 7-9 - явная ссылка с корректным в общих чертах сопровождением текстом;

• 10 - явная ссылка с полностью корректным сопровождением текстом.

Итоговое значение находится в диапазоне от 0 до 1

Значение 1 - визуальный контент релевантен тексту
Значение 0 - визуальный контент не релевантен тексту

Реализация в Python

from deepeval.metrics import ImageReferenceMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question] # Запрос пользователя
answer = [
            "Text", # Текст ответа
            # Передаем данные о местонахождении картинки в MLLMImage
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

def test_image_reference_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, actual_output=answer)
    # Определяем метрику и ее пороговое значение
    metric = ImageReferenceMetric(threshold=0.5)
    # Проводим проверку
    assert_test(test_case, [metric])

Text to Image - показывает, насколько качественно было синтезировано изображение, основываясь на семантической согласованности и качестве восприятия.

Для оценки представляется изображение и текст, на основании которого такое изображение создано. Далее происходит оценка по 10-балльной шкале в следующем диапазоне:

0 - изображение никак не связано с текстом;
10 - изображение идеально соответствует тексту.

Итоговое значение находится в диапазоне от 0 до 1

Значение 1 - визуальный контент релевантен тексту
Значение 0 - визуальный контент не релевантен тексту

Реализация в Python

from deepeval.metrics import TextToImageMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question] # Запрос пользователя
answer = [
            "Text", # Запрос пользователя
            # Передаем данные о местонахождении картинки в MLLMImage
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

def test_text_to_image_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, actual_output=answer)
    # Определяем метрику и ее пороговое значение
    metric = TextToImageMetric(threshold=0.5)
    # Проводим проверку
    assert_test(test_case, [metric])

Image Editing - показывает, насколько качественно было синтезировано новое изображение на базе старого, основываясь на семантической согласованности и качестве восприятия.

Метрика содержит две оценки.

Первая оценивает качество изменения изображения:

0 - изменения не соответствуют тексту совсем;

10 - идеальное соответствие изменения тексту.

Вторая оценка проверяет, не было ли лишних изменений в новом фото по сравнению со старым и с текстом.

0 - новое изображение не соответствует старому совсем;

10 - новое изображение может быть расценено как улучшенная версия старого.

Также есть метрика для оценки качества генерации изображения:

Как и в прошлой метрике, есть две оценки.

Первая показывает, насколько реалистично изображение:

0 - изображение не выглядит реалистичным совсем (ошибки в построении теней, освещения и так далее)

10 - изображение выглядит полностью реалистичным

Вторая оценка показывает процент ненужных артефактов в изображении:

0 - большое количество артефактов;

10 - отсутствие артефактов.

Итоговое значение находится в диапазоне от 0 до 1

Значение 1 - изменение соответствует заданному уровню качества
Значение 0 - изменение не соответствует заданному уровню качества

Реализация в Python

from deepeval.metrics import ImageEditingMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question, # Запрос пользователя
            # Передаем данные о местонахождении картинки в MLLMImage
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True)] # Указание о локальном расположении
answer = [
            # Передаем данные о местонахождении картинки в MLLMImage
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

def test_image_editing_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, actual_output=answer)
    # Определяем метрику и ее пороговое значение
    metric = ImageEditingMetric(threshold=0.5)
    # Проводим проверку
    assert_test(test_case, [metric])

Multimodal Answer Relevancy - показывает, насколько релевантен вывод AI-модели заданному вопросу.

Метрика показывает, насколько изображение релевантно представленному тексту.

Сначала текст разбивается на утверждения, которые могут содержать изображения. После этого утверждения с изображением сравниваются с изначальным запросом на предмет соответствия.

Есть три варианта оценки по итогам сравнения:

• ‘yes’ - соответствие утверждения или изображения запросу

• ‘no’ - несоответствие утверждения или изображения запросу

• ‘idk’ - в целом, соответствие, но может быть использовано как вспомогательное к основному.

Итоговое значение находится в диапазоне от 0 до 1

Значение 1 - ответ релевантен запросу
Значение 0 - ответ нерелевантен запросу

Реализация в Python

from deepeval.metrics import MultimodalAnswerRelevancyMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question] # Запрос пользователя
answer = [
            "Text", # Текст ответа
            # Передаем данные о местонахождении картинки в MLLMImage
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

def test_multimodal_answer_relevancy_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, actual_output=answer)
    # Определяем метрику и ее пороговое значение
    metric = MultimodalAnswerRelevancyMetric(threshold=0.5)
    # Проводим проверку
    assert_test(test_case, [metric])

Multimodal Faithfulness - показывает, насколько вывод AI-модели фактически соответствует возвращенному контексту.

С начала из полученного текста и изображения выводятся утверждения.

После этого каждое утверждение сравнивается с возвращенным контекстом на предмет фактического соответствия.

Есть три варианта оценки по итогам сравнения:

• ‘yes’ - соответствие утверждения возвращенному контексту

• ‘no’ - несоответствие утверждения возвращенному контексту

• ‘idk’ - недостаточно информации для принятия решения.

Итоговое значение находится в диапазоне от 0 до 1

Значение 1 - ответ соответствует возвращенному контексту
Значение 0 - ответ не соответствует возвращенному контексту

Реализация в Python

from deepeval.metrics import MultimodalFaithfulnessMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question, # Запрос пользователя
           # Передаем данные о местонахождении картинки в MLLMImage
           MLLMImage(url="./image")] # Местонахождение картинки
answer = [
            "Text", # Текст ответа
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]
retrieval_context = [
            # Передаем данные о местонахождении картинок в MLLMImage
            MLLMImage(url="./image"), # Местонахождение картинки
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

def test_multimodal_faithfulness_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, actual_output=answer, retrieval_context=retrieval_context)
    # Определяем метрику и ее пороговое значение
    metric = MultimodalFaithfulnessMetric(threshold=0.5)
    # Проводим проверку
    assert_test(test_case, [metric])

Multimodal Contextual Precision - вычисляет, насколько релевантные заданному вопросу части возвращенного контекста находятся выше нерелевантных.

- i+1-ый элемент в возвращенном контексте

- длина возвращенного контекста

- бинарный показатель релевантности k-го элемента в возвращенному контексте.

Если =1 - элемент релевантен, 0 - нерелевантен.

Основываясь на представленных запросе пользователя и контексте (текст или изображение), метрика определяет соответствует ли контекст изначальному запросу.

Если контекст представлен в виде текста, то сначала текст разбивается на утверждения, которые впоследствии сравниваются с изначальным запросом на предмет соответствия.

Если контекст представлен в виде изображения, то сначала производится описание изображения, а после происходит сравнение описания с запросом.

Оценки могут быть только ‘yes’, если утверждение или изображение в контексте релевантно запросу, и ‘no’, если не релевантно.

Итоговое значение находится в диапазоне от 0 до 1

Значение 1 - все релевантные части возвращенного контекста находятся выше нерелевантных в выдаче
Значение 0 - все нерелевантные части возвращенного контекста находятся выше релевантных в выдаче

Реализация в Python

from deepeval.metrics import MultimodalContextualPrecisionMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question] # Запрос пользователя
         
answer = [
            "Text", # Текст ответа
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

expected_output = [
            "Text", # Текст ответа
            MLLMImage(url="./image") # Местонахождение картинки
        ]

retrieval_context = [
            # Передаем данные о местонахождении картинок в MLLMImage
            MLLMImage(url="./image"), # Местонахождение картинки
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

def test_multimodal_contextual_precision_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, actual_output=answer, 
                             expected_output=expected_output, 
                             retrieval_context=retrieval_context)
    # Определяем метрику и ее пороговое значение
    metric = MultimodalContextualPrecisionMetric(threshold=0.5)
    # Проводим проверку
    assert_test(test_case, [metric])

Multimodal Contextual Recall - вычисляет степень соответствия возвращенного контекста ожидаемому выводу.

Сначала в метрике подсчитывается, может ли быть отнесен текст (его утверждения) или изображение в ожидаемом выводе какой-либо части возвращенного контекста. Возможны два варианта оценки:

• ‘yes’ - утверждение или изображение может быть отнесено к какому-либо элементу возвращенного контекста

• ‘no’ - утверждение или изображение не относится ни к какому элементу возвращенного контекста.

Итоговое значение находится в диапазоне от 0 до 1

Значение 1 - возвращенный контекст полностью соответствует ожидаемому ответу
Значение 0 - возвращенный контекст полностью не соответствует ожидаемому ответу

Реализация в Python

from deepeval.metrics import MultimodalContextualRecallMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question] # Запрос пользователя
         
answer = [
            "Text", # Текст ответа
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

expected_output = [
            "Text", # Текст ответа
            MLLMImage(url="./image") # Местонахождение картинки
        ]

retrieval_context = [
            # Передаем данные о местонахождении картинок в MLLMImage
            MLLMImage(url="./image"), # Местонахождение картинки
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

def test_multimodal_contextual_recall_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, actual_output=answer, 
                             expected_output=expected_output, 
                             retrieval_context=retrieval_context)
    # Определяем метрику и ее пороговое значение
    metric = MultimodalContextualRecallMetric(threshold=0.5)
    # Проводим проверку
    assert_test(test_case, [metric])

Multimodal Contextual Relevancy - оценивает релевантность возвращенного контекста заданному вопросу.

Метрика подсчитывает, насколько релевантным является каждое утверждение в контексте или изображение изначальному запросу.

Может быть два вида оценки:

‘yes’ - утверждение или изображение релевантно запросу

‘no’ - утверждение или изображение не релевантно запросу.

Итоговое значение находится в диапазоне от 0 до 1

Значение 1 - возвращенный контекст релевантен запросу
Значение 0 - возвращенный контекст не релевантен запросу

Реализация в Python

from deepeval.metrics import MultimodalContextualRelevancyMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question] # Запрос пользователя
         
answer = [
            "Text", # Текст ответа
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

retrieval_context = [
            # Передаем данные о местонахождении картинок в MLLMImage
            MLLMImage(url="./image"), # Местонахождение картинки
            MLLMImage(url="./image", # Местонахождение картинки
                      local=True) # Указание о локальном расположении
        ]

def test_multimodal_contextual_relevancy_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, actual_output=answer, 
                             retrieval_context=retrieval_context)
    # Определяем метрику и ее пороговое значение
    metric = MultimodalContextualRelevancyMetric(threshold=0.5)
    # Проводим проверку
    assert_test(test_case, [metric])

Multimodal Tool Correctness - оценивает, все ли ожидаемые к использованию при ответе на запрос устройства были действительно использованы.

показывает эффективность и правильность использования устройств при выполнении задачи. Вычисляется как отношение количества реального использования устройств к количеству ожидаемых использований.

Помимо просто вычисления использования нужных устройств может оценивать также и правильный порядок их использования.

Итоговое значение находится в диапазоне от 0 до 1

Значение 1 - вызваны все необходимые инструменты
Значение 0 - необходимые инструменты не вызваны

Реализация в Python

from deepeval.metrics import MultimodalToolCorrectnessMetric
from deepeval.test_case import MLLMTestCase, MLLMImage
from deepeval import assert_test


question = [question] # Запрос пользователя
         
answer = [answer] # Ответ

tools_called = [ToolCall(name=""), ToolCall(name="")] # Вызванные инструменты

expected_tools = [ToolCall(name="")] # Ожидаемые к вызову инструменты

def test_multimodal_contextual_relevancy_metric():
    # Передаем данные в тест кейс
    test_case = MLLMTestCase(input=question, 
                             actual_output=answer, 
                             tools_called=tools_called,
                             expected_tools=expected_tools)
    # Определяем метрику и ее пороговое значение
    metric = MultimodalToolCorrectnessMetric(threshold=0.5)
    # Проводим проверку
    metric.measure(test_case)
    print(metric.score)
    print(metric.reason)

Конкретно эта метрика была вычислена с применением metric.measure.