Реализация Prompt Chaining (цепочка промптов)

Реализация Prompt Chaining (цепочка промптов)

Prompt Chaining — последовательное выполнение нескольких LLM-вызовов, где вывод одного промпта становится входом следующего. Каждый шаг решает подзадачу специализированным промптом. Позволяет декомпозировать сложные задачи, улучшить качество каждого шага и добавить промежуточную валидацию.

Базовая цепочка

from openai import OpenAI
import json

client = OpenAI()

def llm_call(prompt: str, system: str = None, temperature: float = 0.1) -> str:
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        temperature=temperature,
    )
    return response.choices[0].message.content

# Пример: цепочка генерации статьи
def generate_article_pipeline(topic: str) -> dict:

    # Шаг 1: Исследование темы
    research = llm_call(
        f"Исследуй тему '{topic}'. Перечисли 5-7 ключевых фактов, актуальных данных и экспертных мнений.",
        system="Ты — исследователь. Предоставляй только проверяемые факты.",
    )

    # Шаг 2: Структура статьи
    outline = llm_call(
        f"На основе исследования создай структуру статьи (заголовок + 4-5 разделов).\n\nИсследование:\n{research}",
        system="Ты — редактор. Создавай логичные, читаемые структуры.",
    )

    # Шаг 3: Написание статьи
    article = llm_call(
        f"Напиши статью по структуре, используя материалы исследования.\n\nСтруктура:\n{outline}\n\nМатериалы:\n{research}",
        system="Ты — копирайтер. Пиши ясно, конкретно, без воды.",
        temperature=0.4,
    )

    # Шаг 4: Редактура
    edited_article = llm_call(
        f"Отредактируй статью: улучши связность, убери повторы, проверь логику.\n\n{article}",
        system="Ты — литературный редактор.",
    )

    return {
        "research": research,
        "outline": outline,
        "draft": article,
        "final": edited_article,
    }

Цепочки с ветвлением

def document_processing_chain(document_text: str) -> dict:

    # Шаг 1: Классификация
    doc_type = llm_call(
        f"Определи тип документа одним словом: invoice/contract/complaint/inquiry\n\nДокумент:\n{document_text[:500]}",
        temperature=0,
    ).strip().lower()

    # Шаг 2: Ветвление по типу
    if doc_type == "invoice":
        extracted = llm_call(
            f"Извлеки из счёта: поставщик, сумма, дата, номер. Верни JSON.\n\n{document_text}",
            temperature=0,
        )
    elif doc_type == "contract":
        extracted = llm_call(
            f"Извлеки ключевые условия договора: стороны, предмет, сумма, сроки. Верни JSON.\n\n{document_text}",
            temperature=0,
        )
    else:
        extracted = llm_call(
            f"Извлеки основную информацию из документа типа '{doc_type}'.\n\n{document_text}",
            temperature=0,
        )

    # Шаг 3: Валидация
    validation = llm_call(
        f"Проверь корректность извлечённых данных. Тип документа: {doc_type}\n\nДанные: {extracted}\n\nЕсть ли ошибки или пропуски? Верни JSON: {{\"valid\": bool, \"issues\": [...]}}",
        temperature=0,
    )

    # Шаг 4: Итоговое решение
    decision = llm_call(
        f"На основе данных документа: {extracted}\nВалидация: {validation}\nПредложи действие. Верни JSON: {{\"action\": \"...\", \"priority\": \"high|medium|low\"}}",
        temperature=0,
    )

    return {
        "document_type": doc_type,
        "extracted_data": json.loads(extracted),
        "validation": json.loads(validation),
        "decision": json.loads(decision),
    }

Параллельные цепочки

import asyncio

async def parallel_analysis_chain(content: str) -> dict:
    """Параллельный анализ по нескольким направлениям"""

    # Запускаем три анализа параллельно
    sentiment_task = asyncio.create_task(
        async_llm_call(f"Оцени тональность текста. Верни JSON: {{\"sentiment\": \"positive|negative|neutral\", \"score\": -1.0 to 1.0}}\n\n{content}")
    )
    keywords_task = asyncio.create_task(
        async_llm_call(f"Извлеки топ-10 ключевых слов. Верни JSON: {{\"keywords\": [...]}}\n\n{content}")
    )
    summary_task = asyncio.create_task(
        async_llm_call(f"Создай краткое резюме в 2-3 предложения.\n\n{content}")
    )

    sentiment, keywords, summary = await asyncio.gather(
        sentiment_task, keywords_task, summary_task
    )

    # Финальный агрегирующий шаг
    insights = await async_llm_call(
        f"На основе анализа создай бизнес-инсайты.\nТональность: {sentiment}\nКлючевые слова: {keywords}\nРезюме: {summary}"
    )

    return {
        "sentiment": json.loads(sentiment),
        "keywords": json.loads(keywords),
        "summary": summary,
        "insights": insights,
    }

Практический кейс: обработка входящей корреспонденции

Цепочка из 5 шагов:

1. Классификация типа документа (1 LLM-вызов, температура=0)
2. Извлечение данных по схеме для типа (специализированный промпт)
3. Валидация извлечённых данных
4. Определение бизнес-решения (одобрить/отклонить/эскалировать)
5. Генерация ответного письма

Результат: автономная обработка 71% документов, среднее время обработки 45 секунд.

Сроки

• Базовая цепочка из 3-4 шагов: 2–3 дня
• Цепочка с ветвлением и валидацией: 1 неделя
• Параллельные цепочки с агрегацией: 1–2 недели