Prompt Templates и парсинг вывода в LangChain | Курс по LangChain урок 3

Цель урока

Вы научитесь строить сложные шаблоны промпта, добавлять few-shot примеры и получать от модели структурированные данные: словари, списки и объекты Pydantic.

Необходимые знания:

Урок 2 (LCEL, цепочки, базовый вызов модели)
Pydantic на базовом уровне (BaseModel, Field)

Ключевые концепции:

ChatPromptTemplate и его варианты
Few-shot prompting
Partial templates
StrOutputParser
JsonOutputParser
PydanticOutputParser
Метод .with_structured_output()

Зачем нужны шаблоны и парсеры

Во втором уроке мы передавали в промпт простые строки. В реальных задачах промпты сложнее, они включают динамические данные, примеры, форматирующие инструкции. А вывод модели нужен не как текст, а как структура: словарь, список, объект.

Две проблемы, которые решает этот урок:

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

ChatPromptTemplate

Мы уже использовали ChatPromptTemplate.from_messages() со строками. Рассмотрим все варианты.

Оба варианта эквивалентны. Первый короче, второй явнее.

from langchain_core.prompts import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
)

# Вариант 1: кортежи (role, template), краткий
prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты {role}. Отвечай на языке {language}."),
    ("human", "{question}"),
])

# Вариант 2: явные объекты сообщений
prompt = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template("Ты {role}."),
    HumanMessagePromptTemplate.from_template("{question}"),
])

Посмотрим, что внутри шаблона:

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты {role}."),
    ("human", "{question}"),
])

# Какие переменные ожидает шаблон
print(prompt.input_variables)
# ['role', 'question']

# Как выглядит результат подстановки (без вызова модели)
messages = prompt.invoke({"role": "эксперт по Python", "question": "Что такое GIL?"})
print(messages)
# messages=[SystemMessage(content='Ты эксперт по Python.', additional_kwargs={}, response_metadata={}), HumanMessage(content='Что такое GIL?', additional_kwargs={}, response_metadata={})]

.invoke() возвращает ChatPromptValue объект, который содержит список сообщений. Модель принимает его напрямую.

Включение истории диалога в шаблон

Задача вставить историю переписки в промпт. Для этого есть специальный плейсхолдер. MessagesPlaceholder - стандартный способ передать историю в шаблон. В уроке 4 он станет основой для реализации памяти.

Модель получит полный контекст диалога.

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage


prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты разработчик помощник."),
    MessagesPlaceholder(variable_name="history"),  # сюда подставится список сообщений
    ("human", "{question}"),
])

history = [
    HumanMessage(content="Как называется функция для сортировки в Python?"),
    AIMessage(content="sorted() для нового списка, list.sort() для сортировки на месте."),
]

messages = prompt.invoke({
    "history": history,
    "question": "А как отсортировать по ключу?",
})

print(messages)

Few-shot промптинг

Few-shot - это техника, когда в промпт включают примеры "вопрос-ответ", чтобы показать модели желаемый формат или поведение. Особенно полезно, когда нужен специфичный стиль ответа или нестандартный формат.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)

# Примеры для few-shot
examples = [
    {
        "input": "Найди баг: def add(a, b): return a - b",
        "output": "Баг: оператор `-` должен быть `+`. Исправление: return a + b"
    },
    {
        "input": "Найди баг: for i in range(10): print(i) if i == 5: break",
        "output": "Баг: отсутствует отступ у if. Исправление: перенести if внутрь тела цикла."
    },
]

# Шаблон для одного примера
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

# Few-shot шаблон
few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

# Итоговый промпт: система + примеры + вопрос пользователя
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты эксперт по отладке кода на Python. Находи баги кратко и точно."),
    few_shot_prompt,
    ("human", "{input}"),
])

chain = final_prompt | model

response = chain.invoke({"input": "Найди баг: def factorial(n): if n == 0: return 1 return n * factorial(n)"})
print(response.content)

Few-shot эффективен, когда инструкций в системном промпте недостаточно. Конкретные примеры показывают модели паттерн лучше, чем абстрактное описание.

Partial templates

Иногда часть переменных шаблона известна заранее, а часть приходит позже. partial() позволяет зафиксировать известные значения.

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

import os
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты {role}. Отвечай на {language}."),
    ("human", "{question}"),
])

# Фиксируем role и language, они одинаковы для всего приложения
partial_prompt = prompt.partial(role="senior Python developer", language="русском")

# Теперь нужно передать только question
chain = partial_prompt | model
response = chain.invoke({"question": "Как работает asyncio?"})
print(response.content)

Output Parsers: от текста к структуре

Модель возвращает AIMessage с текстом. Для использования в коде этот текст нужно преобразовать в нужный тип. Для этого мы используем output parsers, третье звено в цепочке.

prompt | model | parser

StrOutputParser

Самый простой парсер, извлекает строку из AIMessage. Используйте StrOutputParser, когда вам нужна строка и вы планируете передать её дальше по цепочке.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Объясняй кратко, в 2-3 предложениях."),
    ("human", "{question}"),
])

chain = prompt | model | StrOutputParser()

# Теперь invoke возвращает str, а не AIMessage
result = chain.invoke({"question": "Что такое list comprehension?"})
print(type(result))   # <class 'str'>
print(result)

JsonOutputParser

Когда нужен словарь или список, то указываем модели вернуть JSON. JsonOutputParser ждет от модели валидный JSON и парсит его в объект Python. Если модель вернет невалидный JSON, выбросит исключение. Поэтому важна инструкция в системном промпте.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)

parser = JsonOutputParser()

prompt = ChatPromptTemplate.from_messages([
    ("system", "Отвечай только валидным JSON, без markdown-блоков и пояснений."),
    ("human", """Проанализируй функцию и верни JSON со следующими полями:
    - name: название функции (str)
    - complexity: сложность O-нотация (str)
    - issues: список проблем (list of str)
    - refactored: улучшенная версия (str)

    Функция: {code}"""),
])

chain = prompt | model | parser

result = chain.invoke({"code": """
def find_duplicates(lst):
    duplicates = []
    for i in range(len(lst)):
        for j in range(i+1, len(lst)):
            if lst[i] == lst[j] and lst[i] not in duplicates:
                duplicates.append(lst[i])
    return duplicates
"""})

print(type(result))          # <class 'dict'>
print(result["name"])        # find_duplicates
print(result["complexity"])  # O(n^3)
print(result["issues"])      # ['...', '...']

PydanticOutputParser

Еще как вариант это описать ожидаемую структуру с помощью Pydantic. Парсер автоматически добавит инструкции по формату в промпт.

parser.get_format_instructions() возвращает строку с инструкцией для модели с описанием схемы JSON. Мы передаем её в системный промпт через .partial().

В примере получим валидный объект CodeReview с автодополнением и проверкой типов.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List

load_dotenv()
model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)

# Описываем структуру вывода
class CodeReview(BaseModel):
    function_name: str = Field(description="название функции")
    complexity: str = Field(description="сложность в O-нотации, например O(n^2)")
    issues: List[str] = Field(description="список найденных проблем")
    refactored_code: str = Field(description="улучшенная версия кода")

parser = PydanticOutputParser(pydantic_object=CodeReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты эксперт по code review. {format_instructions}"),
    ("human", "Проведи code review функции:\n{code}"),
]).partial(format_instructions=parser.get_format_instructions())

chain = prompt | model | parser

result = chain.invoke({"code": """
def find_duplicates(lst):
    duplicates = []
    for i in range(len(lst)):
        for j in range(i+1, len(lst)):
            if lst[i] == lst[j] and lst[i] not in duplicates:
                duplicates.append(lst[i])
    return duplicates
"""})

print(type(result))              # <class 'CodeReview'>
print(result.function_name)      # find_duplicates
print(result.complexity)         # O(n^3)
print(result.issues)             # ['...']
print(result.refactored_code)    # def find_duplicates(lst): ...

.with_structured_output()

Современные модели (OpenAI, Claude, Gemini и другие) нативно поддерживают структурированный вывод через tool use / function calling. LangChain предоставляет единый интерфейс для всех.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List

load_dotenv()

class CodeReview(BaseModel):
    function_name: str = Field(description="название функции")
    complexity: str = Field(description="сложность в O-нотации")
    issues: List[str] = Field(description="список найденных проблем")
    refactored_code: str = Field(description="улучшенная версия кода")

model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)

# Привязываем структуру к модели
structured_model = model.with_structured_output(CodeReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты эксперт по code review."),
    ("human", "Проведи code review функции:\n{code}"),
])

chain = prompt | structured_model

result = chain.invoke({"code": """
def find_duplicates(lst):
    duplicates = []
    for i in range(len(lst)):
        for j in range(i+1, len(lst)):
            if lst[i] == lst[j] and lst[i] not in duplicates:
                duplicates.append(lst[i])
    return duplicates
"""})

print(type(result))           # <class 'CodeReview'>
print(result.function_name)   # find_duplicates

Чем .with_structured_output() лучше PydanticOutputParser:

Использует нативный function calling API - модель не "пишет JSON", а заполняет схему напрямую
Меньше шансов получить невалидный вывод
Не нужно добавлять инструкции по формату в промпт

Для большинства задач используйте .with_structured_output(). PydanticOutputParser нужен, когда работаете с моделями без нативной поддержки structured output.

Распространенные ошибки

1. OutputParserException - модель вернула невалидный JSON

langchain_core.exceptions.OutputParserException: Invalid json output

Причина: модель добавила markdown ```json ... ``` или текст вокруг JSON.

Решение: явно укажите в системном промпте "Отвечай только валидным JSON, без markdown и пояснений". Или переходите на .with_structured_output().

2. ValidationError - JSON валидный, но не соответствует схеме

pydantic.ValidationError: 1 validation error for CodeReview

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

Решение: добавьте подробные description в Field(). Чем точнее описание, тем точнее вывод модели.

3. Забытый {format_instructions} в промпте

# Парсер создан, но инструкции не переданы в промпт
parser = PydanticOutputParser(pydantic_object=CodeReview)
prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты эксперт."),  # нет {format_instructions}
    ("human", "{code}"),
])

Решение: всегда добавляйте {format_instructions} в системный промпт и передавай через .partial().

4. with_structured_output со сложной вложенностью

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

Полный пример кода из урока

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
    MessagesPlaceholder,
    FewShotChatMessagePromptTemplate,
)
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser, PydanticOutputParser
from langchain_core.messages import HumanMessage, AIMessage
from pydantic import BaseModel, Field
from typing import List

load_dotenv()

model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)


# --- 1. ChatPromptTemplate в деталях ---

# Вариант 1: кортежи (role, template), краткий
prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты {role}. Отвечай на языке {language}."),
    ("human", "{question}"),
])

# Вариант 2: явные объекты сообщений
prompt = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template("Ты {role}."),
    HumanMessagePromptTemplate.from_template("{question}"),
])

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты {role}."),
    ("human", "{question}"),
])

# Какие переменные ожидает шаблон
print(prompt.input_variables)
# ['role', 'question']

# Как выглядит результат подстановки (без вызова модели)
messages = prompt.invoke({"role": "эксперт по Python", "question": "Что такое GIL?"})
print(messages)
# messages=[SystemMessage(content='Ты эксперт по Python.', additional_kwargs={}, response_metadata={}), HumanMessage(content='Что такое GIL?', additional_kwargs={}, response_metadata={})]


# --- 2. MessagesPlaceholder ---

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты разработчик помощник."),
    MessagesPlaceholder(variable_name="history"),  # сюда подставится список сообщений
    ("human", "{question}"),
])

history = [
    HumanMessage(content="Как называется функция для сортировки в Python?"),
    AIMessage(content="sorted() для нового списка, list.sort() для сортировки на месте."),
]

messages = prompt.invoke({
    "history": history,
    "question": "А как отсортировать по ключу?",
})

print(messages)


# --- 3. Few-shot ---

examples = [
    {
        "input": "Найди баг: def add(a, b): return a - b",
        "output": "Баг: оператор `-` должен быть `+`. Исправление: return a + b"
    },
    {
        "input": "Найди баг: for i in range(10): print(i) if i == 5: break",
        "output": "Баг: отсутствует отступ у if. Исправление: перенести if внутрь тела цикла."
    },
]

# Шаблон для одного примера
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

# Few-shot шаблон
few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

# Итоговый промпт: система + примеры + вопрос пользователя
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты эксперт по отладке кода на Python. Находи баги кратко и точно."),
    few_shot_prompt,
    ("human", "{input}"),
])

chain = final_prompt | model

response = chain.invoke({"input": "Найди баг: def factorial(n): if n == 0: return 1 return n * factorial(n)"})
print(response.content)


# --- 4. Partial templates ---

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты {role}. Отвечай на {language}."),
    ("human", "{question}"),
])

# Фиксируем role и language - они одинаковы для всего приложения
partial_prompt = prompt.partial(role="senior Python developer", language="русском")

# Теперь нужно передать только question
chain = partial_prompt | model
response = chain.invoke({"question": "Как работает asyncio?"})
print(response.content)


# --- 5. StrOutputParser ---

prompt = ChatPromptTemplate.from_messages([
    ("system", "Объясняй кратко, в 2-3 предложениях."),
    ("human", "{question}"),
])

chain = prompt | model | StrOutputParser()

# Теперь invoke возвращает str, а не AIMessage
result = chain.invoke({"question": "Что такое list comprehension?"})
print(type(result))   # <class 'str'>
print(result)


# --- 6. JsonOutputParser ---

parser = JsonOutputParser()

prompt = ChatPromptTemplate.from_messages([
    ("system", "Отвечай только валидным JSON, без markdown-блоков и пояснений."),
    ("human", """Проанализируй функцию и верни JSON со следующими полями:
    - name: название функции (str)
    - complexity: сложность O-нотация (str)
    - issues: список проблем (list of str)
    - refactored: улучшенная версия (str)

    Функция: {code}"""),
])

chain = prompt | model | parser

result = chain.invoke({"code": """
def find_duplicates(lst):
    duplicates = []
    for i in range(len(lst)):
        for j in range(i+1, len(lst)):
            if lst[i] == lst[j] and lst[i] not in duplicates:
                duplicates.append(lst[i])
    return duplicates
"""})

print(type(result))          # <class 'dict'>
print(result["name"])        # find_duplicates
print(result["complexity"])  # O(n^3)
print(result["issues"])      # ['...', '...']


# --- 7. PydanticOutputParser ---

class CodeReview(BaseModel):
    function_name: str = Field(description="название функции")
    complexity: str = Field(description="сложность в O-нотации, например O(n^2)")
    issues: List[str] = Field(description="список найденных проблем")
    refactored_code: str = Field(description="улучшенная версия кода")

parser = PydanticOutputParser(pydantic_object=CodeReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты эксперт по code review. {format_instructions}"),
    ("human", "Проведи code review функции:\n{code}"),
]).partial(format_instructions=parser.get_format_instructions())

chain = prompt | model | parser

result = chain.invoke({"code": """
def find_duplicates(lst):
    duplicates = []
    for i in range(len(lst)):
        for j in range(i+1, len(lst)):
            if lst[i] == lst[j] and lst[i] not in duplicates:
                duplicates.append(lst[i])
    return duplicates
"""})

print(type(result))              # <class 'CodeReview'>
print(result.function_name)      # find_duplicates
print(result.complexity)         # O(n^3)
print(result.issues)             # ['...']
print(result.refactored_code)    # def find_duplicates(lst): ...


# --- 8. with_structured_output ---

class CodeReview(BaseModel):
    function_name: str = Field(description="название функции")
    complexity: str = Field(description="сложность в O-нотации")
    issues: List[str] = Field(description="список найденных проблем")
    refactored_code: str = Field(description="улучшенная версия кода")

model = ChatOpenAI(model=os.getenv("MODEL_NAME", "gpt-4o"), temperature=0)

# Привязываем структуру к модели
structured_model = model.with_structured_output(CodeReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ты эксперт по code review."),
    ("human", "Проведи code review функции:\n{code}"),
])

chain = prompt | structured_model

result = chain.invoke({"code": """
def find_duplicates(lst):
    duplicates = []
    for i in range(len(lst)):
        for j in range(i+1, len(lst)):
            if lst[i] == lst[j] and lst[i] not in duplicates:
                duplicates.append(lst[i])
    return duplicates
"""})

print(type(result))           # <class 'CodeReview'>
print(result.function_name)   # find_duplicates

Практическое задание

Создайте систему анализа вакансий.

Требования:

1) Создайте модель Pydantic JobAnalysis с полями:

position - название должности (str)
required_skills - обязательные навыки (List[str])
nice_to_have_skills - желательные навыки (List[str])
experience_years - минимальный опыт в годах (int)
seniority_level - уровень: "junior", "middle", "senior" (str)
is_remote - удаленная работа (bool)

2) Используйте .with_structured_output() для получения результата

3) Добавьте few-shot примеры (минимум 2) чтобы показать модели, как определять уровень позиции

4) Протестируйте на трех разных вакансиях - junior, middle и senior

Пример входных данных:

"Ищем Python разработчика. Требования: Django, PostgreSQL, 
REST API. Опыт от 3 лет. Знание Docker будет плюсом. 
Офис в Москве, возможна частичная удаленка."

Итоги урока

Мы научились строить промпты и получать структурированный вывод. Но модель всё ещё не помнит контекст между вызовами, каждый .invoke() начинается с начала. В следующем уроке разберем, как добавить память, сохранять историю диалога и сделать чат-бота который ведет связный диалог.

<< Урок 2

Урок 4 >>

Подписывайтесь на мой Telegram канал

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