Как создать надёжный инструмент проверки обоснованности с помощью Upstage API и LangChain

!pip install -qU langchain-core langchain-upstage
 
import os
import json
from typing import List, Dict, Any
from langchain_upstage import UpstageGroundednessCheck
 
os.environ["UPSTAGE_API_KEY"] = "Введите ваш API ключ здесь"

class AdvancedGroundednessChecker:
    """Расширенный обёртка для проверки обоснованности Upstage с поддержкой пакетной обработки и анализа"""
   
    def __init__(self):
        self.checker = UpstageGroundednessCheck()
        self.results = []
   
    def check_single(self, context: str, answer: str) -> Dict[str, Any]:
        """Проверка обоснованности одной пары контекст-ответ"""
        request = {"context": context, "answer": answer}
        response = self.checker.invoke(request)
       
        result = {
            "context": context,
            "answer": answer,
            "grounded": response,
            "confidence": self._extract_confidence(response)
        }
        self.results.append(result)
        return result
   
    def batch_check(self, test_cases: List[Dict[str, str]]) -> List[Dict[str, Any]]:
        """Обработка нескольких тестов"""
        batch_results = []
        for case in test_cases:
            result = self.check_single(case["context"], case["answer"])
            batch_results.append(result)
        return batch_results
   
    def _extract_confidence(self, response) -> str:
        """Извлечение уровня уверенности из ответа"""
        if hasattr(response, 'lower'):
            if 'grounded' in response.lower():
                return 'high'
            elif 'not grounded' in response.lower():
                return 'low'
        return 'medium'
   
    def analyze_results(self) -> Dict[str, Any]:
        """Анализ результатов пакетных проверок"""
        total = len(self.results)
        grounded = sum(1 for r in self.results if 'grounded' in str(r['grounded']).lower())
       
        return {
            "total_checks": total,
            "grounded_count": grounded,
            "not_grounded_count": total - grounded,
            "accuracy_rate": grounded / total if total > 0 else 0
        }
 
checker = AdvancedGroundednessChecker()

print("=== Тест 1: Несоответствие по высоте ===")
result1 = checker.check_single(
    context="Маунa-Кеа — неактивный вулкан на острове Гавайи.",
    answer="Маунa-Кеа имеет высоту 5207.3 метра."
)
print(f"Результат: {result1['grounded']}")
 
print("\n=== Тест 2: Правильная информация ===")
result2 = checker.check_single(
    context="Python — высокоуровневый язык программирования, созданный Гвидо ван Россумом в 1991 году. Он делает упор на читаемость и простоту кода.",
    answer="Python создан Гвидо ван Россумом и ориентирован на читаемость кода."
)
print(f"Результат: {result2['grounded']}")
 
print("\n=== Тест 3: Частичная информация ===")
result3 = checker.check_single(
    context="Великая Китайская стена примерно 13 000 миль в длину и строилась более 2000 лет.",
    answer="Великая Китайская стена очень длинная."
)
print(f"Результат: {result3['grounded']}")
 
print("\n=== Тест 4: Противоречивая информация ===")
result4 = checker.check_single(
    context="Вода кипит при 100 градусах Цельсия на уровне моря.",
    answer="Вода кипит при 90 градусах Цельсия на уровне моря."
)
print(f"Результат: {result4['grounded']}")

print("\n=== Пример пакетной обработки ===")
test_cases = [
    {
        "context": "Шекспир написал Ромео и Джульетту в конце XVI века.",
        "answer": "Ромео и Джульетта написаны Шекспиром."
    },
    {
        "context": "Скорость света примерно 299 792 458 метров в секунду.",
        "answer": "Свет движется примерно со скоростью 300 000 километров в секунду."
    },
    {
        "context": "У Земли есть один естественный спутник — Луна.",
        "answer": "У Земли два спутника."
    }
]
 
batch_results = checker.batch_check(test_cases)
for i, result in enumerate(batch_results, 1):
    print(f"Пакетный тест {i}: {result['grounded']}")
 
print("\n=== Анализ результатов ===")
analysis = checker.analyze_results()
print(f"Всего проверок: {analysis['total_checks']}")
print(f"Обоснованных ответов: {analysis['grounded_count']}")
print(f"Необоснованных ответов: {analysis['not_grounded_count']}")
print(f"Уровень обоснованности: {analysis['accuracy_rate']:.2%}")
 
print("\n=== Тестирование по доменам ===")
domains = {
    "Наука": {
        "context": "Фотосинтез — процесс, при котором растения преобразуют солнечный свет, углекислый газ и воду в глюкозу и кислород.",
        "answer": "Растения используют фотосинтез для получения пищи из солнечного света и CO2."
    },
    "История": {
        "context": "Вторая мировая война закончилась в 1945 году после капитуляции Японии после атомных бомбардировок.",
        "answer": "Вторая мировая война закончилась в 1944 году с капитуляцией Германии."
    },
    "География": {
        "context": "Гора Эверест — самая высокая вершина Земли, расположена в Гималаях на высоте 8848.86 метров.",
        "answer": "Эверест — самая высокая гора и находится в Гималаях."
    }
}
 
for domain, test_case in domains.items():
    result = checker.check_single(test_case["context"], test_case["answer"])
    print(f"{domain}: {result['grounded']}")

def create_test_report(checker_instance):
    """Создание подробного отчёта по тестам"""
    report = {
        "summary": checker_instance.analyze_results(),
        "detailed_results": checker_instance.results,
        "recommendations": []
    }
   
    accuracy = report["summary"]["accuracy_rate"]
    if accuracy < 0.7:
        report["recommendations"].append("Рекомендуется пересмотреть процесс генерации ответов")
    if accuracy > 0.9:
        report["recommendations"].append("Высокая точность — система работает хорошо")
   
    return report
 
print("\n=== Итоговый отчёт по тестам ===")
report = create_test_report(checker)
print(f"Общая производительность: {report['summary']['accuracy_rate']:.2%}")
print("Рекомендации:", report["recommendations"])