Анализ и извлечение структурированных данных

Извлечение структурированных данных из текста

Одна из самых практичных задач для LLM — превращение неструктурированного текста в структурированные данные для интеграции с другими системами (CRM, базы данных, аналитические дашборды). Ключ к успеху — чёткое определение схемы вывода и валидация результатов.

Извлечение сущностей с гарантированным форматом

Для интеграции с базами данных или API критически важен предсказуемый формат вывода:

# Системный промпт для извлечения данных
system_prompt = """
Ты — система извлечения структурированных данных из текста.

ЗАДАЧА: извлечь информацию и вернуть ТОЛЬКО валидный JSON.

СХЕМА ВЫВОДА:
{
  "entities": [
    {
      "type": "person" | "organization" | "location" | "date",
      "value": "string",
      "confidence": 0.0-1.0
    }
  ],
  "sentiment": "positive" | "negative" | "neutral",
  "key_topics": ["string"],
  "summary": "string (max 100 chars)"
}

ПРАВИЛА:
- Если информация не найдена, возвращай пустой массив
- Confidence < 0.7 для неоднозначных случаев
- Строго следуй схеме, не добавляй поля
"""

# Валидация схемы
from jsonschema import validate, ValidationError

output_schema = {
    "type": "object",
    "required": ["entities", "sentiment", "key_topics", "summary"],
    "properties": {
        "entities": {
            "type": "array",
            "items": {
                "type": "object",
                "required": ["type", "value", "confidence"],
                "properties": {
                    "type": {"enum": ["person", "organization", "location", "date"]},
                    "value": {"type": "string"},
                    "confidence": {"type": "number", "minimum": 0, "maximum": 1}
                }
            }
        },
        "sentiment": {"enum": ["positive", "negative", "neutral"]},
        "key_topics": {"type": "array", "items": {"type": "string"}},
        "summary": {"type": "string", "maxLength": 100}
    }
}

def extract_structured_data(text):
    response = llm.generate(
        system=system_prompt,
        user=user_prompt.format(text=text),
        temperature=0.1  # Минимальная для детерминированности
    )

    try:
        data = json.loads(response)
        validate(instance=data, schema=output_schema)
        return {'success': True, 'data': data}
    except (json.JSONDecodeError, ValidationError) as e:
        # Retry с более строгим промптом
        return retry_with_correction(text, error=str(e))

Sentiment Analysis и категоризация

Анализ тональности и категоризация особенно полезны для обработки customer feedback:

# Многоуровневая категоризация отзывов
system_prompt = """
Ты — система анализа отзывов клиентов.

Проанализируй отзыв и верни JSON:

{
  "sentiment": {
    "overall": "positive" | "negative" | "neutral" | "mixed",
    "score": -1.0 to 1.0,
    "confidence": 0.0 to 1.0
  },
  "categories": [
    {
      "category": "product_quality" | "delivery" | "customer_service" | "price" | "other",
      "sentiment": "positive" | "negative" | "neutral",
      "mentions": ["цитата из текста"]
    }
  ],
  "priority": "high" | "medium" | "low",
  "requires_response": boolean,
  "suggested_action": "string или null"
}

ЛОГИКА PRIORITY:
- high: негативный отзыв + упоминание конкретной проблемы
- medium: смешанный отзыв или нейтральный с вопросами
- low: позитивный отзыв без запросов

REQUIRES_RESPONSE = true если:
- Есть вопрос
- Негативный sentiment
- Упоминание проблемы
"""

def analyze_feedback(feedback_text):
    response = llm.generate(
        system=system_prompt,
        user=f"Отзыв: {feedback_text}",
        temperature=0.2
    )

    analysis = json.loads(response)

    # Автоматическая маршрутизация
    if analysis['priority'] == 'high':
        notify_support_team(analysis)

    if analysis['requires_response']:
        draft_response = generate_response_draft(feedback_text, analysis)
        analysis['response_draft'] = draft_response

    return analysis

Суммаризация с сохранением ключевой информации

Для длинных документов важна не просто краткость, а сохранение критических деталей:

# Многоуровневая суммаризация
system_prompt = """
Ты — система суммаризации документов.

Создай три уровня summary:

{
  "executive_summary": "1-2 предложения, главная суть",
  "key_points": [
    "bullet point 1",
    "bullet point 2",
    ...
  ],
  "detailed_summary": "2-3 параграфа с важными деталями",
  "action_items": [
    {
      "action": "что нужно сделать",
      "priority": "high" | "medium" | "low",
      "deadline": "если указан в тексте или null"
    }
  ],
  "key_metrics": {
    "metric_name": "value"
  }
}

ПРАВИЛА:
- Executive summary: только главное решение/вывод
- Key points: 3-7 пунктов, конкретные факты
- Detailed summary: контекст и детали
- Action items: только явные задачи из текста
- Key metrics: только числовые данные
"""

def summarize_document(document, summary_type='all'):
    response = llm.generate(
        system=system_prompt,
        user=f"Документ:\n\n{document}",
        temperature=0.2
    )

    summary = json.loads(response)

    # Возврат нужного уровня детализации
    if summary_type == 'executive':
        return summary['executive_summary']
    elif summary_type == 'brief':
        return {
            'summary': summary['executive_summary'],
            'key_points': summary['key_points']
        }
    else:
        return summary

Обеспечение консистентности формата

Для надёжной интеграции используйте несколько уровней защиты:

1. Строгая схема в промпте:

# Включите JSON schema прямо в системный промпт
system_prompt = f"""
Верни ТОЛЬКО валидный JSON следующей структуры:

{json.dumps(output_schema, indent=2)}

Не добавляй пояснений, только JSON.
"""

2. Валидация и retry логика:

def extract_with_retry(text, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = llm.generate(system=system_prompt, user=text, temperature=0.1)

            # Очистка от markdown форматирования
            cleaned = response.strip()
            if cleaned.startswith('```json'):
                cleaned = cleaned[7:]
            if cleaned.endswith('```'):
                cleaned = cleaned[:-3]

            data = json.loads(cleaned)
            validate(instance=data, schema=output_schema)

            return {'success': True, 'data': data}

        except Exception as e:
            if attempt == max_retries - 1:
                return {'success': False, 'error': str(e), 'raw_response': response}

            # Корректирующий промпт
            system_prompt_retry = system_prompt + f"\n\nПредыдущая попытка вызвала ошибку: {e}\nИсправь и верни корректный JSON."

3. Fallback стратегии:

def extract_with_fallback(text):
    # Попытка 1: полная схема
    result = extract_structured_data(text)

    if result['success']:
        return result

    # Попытка 2: упрощённая схема
    result = extract_simplified_data(text)

    if result['success']:
        return enrich_with_defaults(result['data'])

    # Fallback: базовое извлечение
    return {
        'success': True,
        'data': extract_basic_info(text),
        'warning': 'Used fallback extraction'
    }

Ключевые принципы

Для надёжного извлечения данных:

Определяйте JSON schema явно — включайте её в системный промпт
Используйте минимальную temperature (0.1-0.2) для детерминированности
Валидируйте каждый ответ — не доверяйте формату вслепую
Реализуйте retry логику — с корректирующими промптами при ошибках
Создавайте fallback стратегии — для случаев, когда полное извлечение невозможно
Тестируйте на edge cases — пустые поля, неоднозначные данные, нестандартные форматы
Логируйте проблемные случаи — для последующей оптимизации промптов