Введение: почему этот разговор важен именно сейчас
Несколько лет назад разработчик с десятилетним опытом в бэкенде мог позволить себе смотреть на машинное обучение как на экзотику - интересную, но далёкую от реальной инженерной работы. Сегодня ситуация принципиально изменилась. Крупные технологические компании переориентируют продуктовые дорожные карты, стартапы строят целые платформы поверх языковых моделей, а в описаниях вакансий для старших инженеров появляются требования, которых три года назад попросту не существовало.
Речь идёт не об очередной волне хайпа, которую можно переждать. Это структурное изменение в том, как строятся программные продукты. И здесь возникает закономерный вопрос: что делать опытному разработчику, который строил надёжные системы на протяжении многих лет, но никогда не обучал нейронные сети и не работал с векторными базами данных?
Ответ, как это ни парадоксально, гораздо менее пугающий, чем кажется на первый взгляд. AI Engineering - это не новая профессия с нуля. Это расширение классической инженерной практики новым инструментальным слоем. Большая часть того, что вы уже умеете - проектирование API, работа с базами данных, управление состоянием, асинхронная обработка, паттерны проектирования - остаётся не просто релевантной, а критически важной.
Эта статья написана именно для тех, у кого есть крепкая инженерная база и кто хочет понять, как органично встроить AI-инструменты в свой профессиональный арсенал.
Часть первая: Деконструкция иллюзий
Почему опытные разработчики избегают AI-тематики
Парадокс состоит в следующем: именно те инженеры, которым проще всего войти в AI Engineering, нередко сопротивляются этому дольше всех.
Причин несколько, и они вполне понятны:
Первая - ощущение, что нужно заново начинать с нуля. Когда в разговоре мелькают слова "трансформеры", "градиентный спуск", "функция потерь", у разработчика, который никогда не занимался машинным обучением, возникает стойкое чувство, что перед ним - закрытый клуб со своим языком и своими многолетними правилами входа. Это ощущение понятно, но оно вводит в заблуждение.
Вторая - профессиональная идентичность. Инженер, который годами гордился умением строить надёжные, предсказуемые системы, сталкивается с инструментами, которые по самой своей природе являются вероятностными. Это создаёт когнитивный дискомфорт: как строить систему, если один из её ключевых компонентов может давать разные ответы на один и тот же вопрос?
Третья - профессиональный снобизм, который иногда принимает форму скептицизма.
"Это просто автодополнение текста в промышленных масштабах".
Это отчасти верно с технической точки зрения, но совершенно не отменяет практической ценности инструмента.
Ни одна из этих причин не является достаточным основанием для того, чтобы игнорировать происходящие изменения.
Часть вторая: Технические основы - архитектурный взгляд
Шаг 1. Понимание механики генеративного ИИ
Первый шаг - избавиться от ощущения "магии". Большая языковая модель (LLM) - это не разум и не оракул. Это вероятностная математическая функция, которая принимает последовательность токенов на входе и возвращает распределение вероятностей для следующего токена. Из этого простого механизма, повторённого миллиарды раз на огромных данных, возникает то, что мы воспринимаем как "понимание текста".
Для практикующего бэкенд-инженера взаимодействие с LLM через API - это, по существу, вызов внешнего микросервиса. Разница в одном: этот сервис не детерминирован. На один и тот же запрос он может вернуть разные ответы. Это не баг - это фундаментальное свойство, которым нужно научиться управлять.
Несколько концепций, которые необходимо понять на базовом уровне:
Токенизация. Токен - это не слово и не символ. Это единица текста, которую модель обрабатывает за один шаг. В зависимости от языка и модели один токен приблизительно соответствует четырём символам в английском тексте или двум-трём символам в русском. Это имеет прямые практические следствия: вы платите за токены, контекстное окно измеряется в токенах, и именно токены определяют, сколько информации модель может "удержать в голове" за один вызов.
Эмбеддинги. Представьте, что любой текст - слово, предложение, абзац - можно описать точкой в многомерном пространстве так, что семантически близкие тексты окажутся рядом друг с другом. Именно это делают эмбеддинги. Если вы работали с индексами в базах данных или с алгоритмами хеширования, думайте об эмбеддингах как о семантическом индексе смысла. Расстояние между двумя векторами - это мера смысловой близости двух текстов.
Температура. Этот параметр управляет тем, насколько предсказуемы ответы модели. При температуре, близкой к нулю, модель будет воспроизводить наиболее статистически вероятный ответ - это полезно для задач, где требуется точность и повторяемость. При высокой температуре модель "рискует" и выбирает менее очевидные токены - это нужно для творческих задач. Выбор температуры - это инженерное решение, которое зависит от конкретной задачи.
Контекстное окно. Это максимальный объём информации, который модель может обработать за один вызов. Современные модели имеют контекстные окна от нескольких десятков тысяч до нескольких миллионов токенов. Управление контекстом - одна из ключевых инженерных задач при построении AI-приложений.
Вот базовый пример взаимодействия с языковой моделью через API:
import openai
client = openai.Client(api_key="your_api_key")
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": (
"Ты — старший инженер-программист. "
"Отвечай точно и по существу, без лишних слов."
)
},
{
"role": "user",
"content": "Объясни разницу между эмбеддингами и хэш-функциями."
}
],
temperature=0.0, # Нулевая температура для точных, воспроизводимых ответов
max_tokens=500
)
print(response.choices[0].message.content)
# Не забывайте обрабатывать метаданные вызова:
usage = response.usage
print(f"Использовано токенов: {usage.total_tokens} "
f"(запрос: {usage.prompt_tokens}, ответ: {usage.completion_tokens})")
Обратите внимание на параметр temperature=0.0. Когда вы строите систему, которая должна анализировать логи, парсить документы или генерировать структурированные данные, детерминированность важнее разнообразия. Это инженерное решение, а не "магический" параметр.
Шаг 2. Структурирование хаоса: от текста к типизированным объектам
Одна из главных болей при интеграции языковых моделей в классический бэкенд - это работа с неструктурированным выводом. Ваша база данных ожидает конкретные типы, фронтенд - строгий JSON-контракт, а оркестрирующий код - предсказуемые объекты с известными полями. Языковая модель по умолчанию возвращает свободный текст.
Решение этой проблемы - принудительная типизация вывода через валидационные схемы. Концептуально это аналог паттерна DTO (Data Transfer Object) в классической архитектуре, только применённый к вероятностному компоненту системы.
В экосистеме Python стандартом де-факто стал Pydantic в связке с соответствующими инструментами LangChain.
Принцип прост: вы описываете схему ожидаемого ответа, передаёте её инструкции в промпт, и парсер принудительно приводит вывод модели к заданному типу.
from pydantic import BaseModel, Field
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
# Описываем схему так же, как описывали бы DTO или схему базы данных
class LogAnalysis(BaseModel):
severity: Literal["INFO", "WARNING", "ERROR", "FATAL"] = Field(
description="Уровень критичности события"
)
affected_service: str = Field(
description="Название микросервиса, в котором произошла ошибка"
)
root_cause: str = Field(
description="Краткое описание предполагаемой причины проблемы"
)
recommended_action: str = Field(
description="Рекомендуемое первоочередное действие для устранения проблемы"
)
is_production_affecting: bool = Field(
description="True, если ошибка влияет на продакшен прямо сейчас"
)
parser = PydanticOutputParser(pydantic_object=LogAnalysis)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = PromptTemplate(
template=(
"Ты — старший SRE-инженер. "
"Проанализируй следующую строку лога и верни структурированный анализ.\n\n"
"{format_instructions}\n\n"
"Строка лога:\n{log_entry}\n"
),
input_variables=["log_entry"],
partial_variables={"format_instructions": parser.get_format_instructions()},
)
chain = prompt | llm | parser
log_line = "[2024-01-15 03:47:22] FATAL: Connection pool exhausted in payment-gateway-service. Active connections: 512/512. Queue depth: 847."
result: LogAnalysis = chain.invoke({"log_entry": log_line})
# Теперь result — это строго типизированный объект, а не свободный текст
print(f"Сервис: {result.affected_service}")
print(f"Критичность: {result.severity}")
print(f"Влияет на продакшен: {result.is_production_affecting}")
print(f"Рекомендация: {result.recommended_action}")
Практическая ценность этого подхода становится очевидна, когда вы встраиваете подобный анализатор в реальный пайплайн: результат можно сразу записывать в базу данных, отправлять в систему управления инцидентами или использовать как условие для маршрутизации алертов.
Важно учитывать, что Pydantic-парсер внутри себя делает дополнительный вызов модели при ошибке формата. Если надёжность важнее скорости, используйте with_structured_output напрямую через OpenAI - это использует нативную поддержку JSON Schema на уровне API и более устойчиво к ошибкам:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(LogAnalysis)
prompt = ChatPromptTemplate.from_messages([
("system", "Ты — старший SRE-инженер. Проанализируй строку лога."),
("user", "{log_entry}")
])
chain = prompt | structured_llm
result = chain.invoke({"log_entry": log_line})
Шаг 3. От простых вызовов к агентам
На определённом этапе простая схема "передали запрос - получили ответ" перестаёт быть достаточной. Реальные задачи требуют многошаговых рассуждений, обращений к внешним источникам данных и принятия решений о том, какой следующий шаг предпринять.
Именно здесь появляется концепция агента. Агент - это языковая модель, которой дан набор инструментов (Tools) и право самостоятельно решать, в каком порядке и как их использовать для решения поставленной задачи.
С точки зрения классической архитектуры это напоминает паттерн Strategy: у вас есть интерфейс "инструмент", конкретные реализации (поиск в базе данных, вызов внешнего API, расчёт), и оркестратор, который выбирает нужную стратегию. Только в данном случае оркестратором выступает языковая модель.
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
import requests
# Определяем инструменты как обычные Python-функции
@tool
def get_service_metrics(service_name: str) -> dict:
"""
Возвращает текущие метрики сервиса: CPU, память, количество запросов в секунду.
Используй этот инструмент, когда нужно проверить состояние конкретного сервиса.
"""
# В реальном коде здесь был бы запрос к Prometheus или DataDog
return {
"service": service_name,
"cpu_usage": 87.3,
"memory_mb": 1842,
"rps": 1243,
"error_rate": 0.023,
"p99_latency_ms": 342
}
@tool
def create_incident(severity: str, title: str, description: str) -> str:
"""
Создаёт инцидент в системе управления инцидентами (например, PagerDuty или Opsgenie).
Используй только при severity ERROR или FATAL.
"""
# В реальном коде здесь был бы вызов API системы инцидентов
incident_id = f"INC-{hash(title) % 10000:04d}"
return f"Инцидент создан: {incident_id}"
@tool
def get_recent_deployments(service_name: str, hours: int = 24) -> list:
"""
Возвращает список деплойментов сервиса за последние N часов.
Полезно для корреляции проблем с недавними изменениями в коде.
"""
return [
{"time": "2024-01-15 02:30:00", "version": "v2.4.1", "author": "deploy-bot"},
{"time": "2024-01-15 00:15:00", "version": "v2.4.0", "author": "deploy-bot"}
]
tools = [get_service_metrics, create_incident, get_recent_deployments]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", (
"Ты — автоматизированный SRE-ассистент. "
"Анализируй проблемы, собирай метрики и при необходимости создавай инциденты. "
"Всегда проверяй метрики перед созданием инцидента."
)),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
response = executor.invoke({
"input": "Получили алерт: payment-gateway-service начал возвращать 503 ошибки. "
"Проверь метрики, посмотри недавние деплойменты и реши, нужно ли создавать инцидент."
})
print(response["output"])
В этом примере агент самостоятельно решает, в каком порядке вызывать инструменты: сначала собирает метрики, затем проверяет деплойменты, и только после этого принимает решение о создании инцидента. Вы не пишете эту логику явно - модель выводит её из описаний инструментов и контекста задачи.
Шаг 4. Управление состоянием сложных агентных систем
Когда число шагов в агентном пайплайне растёт, а логика ветвления становится нетривиальной, код на базе простого AgentExecutor начинает деградировать в запутанный клубок условных операторов. Это знакомая проблема: достаточно вспомнить любой крупный сервис, где бизнес-логика постепенно превратилась в лабиринт if-else.
Индустрия решает эту проблему с помощью LangGraph - библиотеки для построения агентных систем на основе ориентированных графов. С архитектурной точки зрения LangGraph реализует конечный автомат (Finite State Machine), где узлы - это функции или сервисы обработки, а рёбра - это правила переходов между ними.
Ключевое отличие от классической FSM заключается в том, что решение о переходе из одного состояния в другое может принимать языковая модель, а не жёстко заданное условие.
from typing import TypedDict, Annotated, Optional
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
import operator
# Состояние, которое передаётся между узлами графа
class SupportTicketState(TypedDict):
ticket_id: str
user_message: str
category: Optional[str] # Категория обращения (billing, technical, general)
sentiment: Optional[str] # Тональность (positive, neutral, negative, angry)
response: Optional[str] # Сгенерированный ответ
needs_human: bool # Флаг эскалации на живого оператора
history: Annotated[list, operator.add] # Лог обработки
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Узел 1: классификация обращения
def classify_ticket(state: SupportTicketState) -> SupportTicketState:
classification_prompt = f"""
Классифицируй обращение пользователя. Верни ТОЛЬКО JSON с двумя ключами:
- "category": одно из ["billing", "technical", "general", "complaint"]
- "sentiment": одно из ["positive", "neutral", "negative", "angry"]
Обращение: {state['user_message']}
"""
response = llm.invoke([HumanMessage(content=classification_prompt)])
import json, re
data = json.loads(re.search(r'\{.*\}', response.content, re.DOTALL).group())
return {
"category": data["category"],
"sentiment": data["sentiment"],
"history": [f"Классифицировано: {data['category']}, тональность: {data['sentiment']}"]
}
# Узел 2: генерация ответа
def generate_response(state: SupportTicketState) -> SupportTicketState:
system_context = {
"billing": "Ты — специалист по биллингу. Будь чётким и предоставь конкретные цифры.",
"technical": "Ты — технический специалист поддержки. Давай пошаговые инструкции.",
"general": "Ты — специалист общей поддержки. Будь дружелюбным и полезным.",
"complaint": "Ты — менеджер по работе с клиентами. Проявляй эмпатию и предлагай решения."
}
context = system_context.get(state["category"], system_context["general"])
response = llm.invoke([
HumanMessage(content=f"Системный контекст: {context}\n\nОбращение: {state['user_message']}")
])
return {
"response": response.content,
"history": ["Ответ сгенерирован"]
}
# Узел 3: эскалация на человека
def escalate_to_human(state: SupportTicketState) -> SupportTicketState:
return {
"needs_human": True,
"response": "Ваше обращение передано старшему специалисту. Ожидайте ответа в течение 2 часов.",
"history": ["Эскалировано на живого оператора"]
}
# Маршрутизатор: определяет, нужна ли эскалация
def route_after_classification(state: SupportTicketState) -> str:
# Гневные обращения или жалобы сразу идут к живому оператору
if state["sentiment"] == "angry" or state["category"] == "complaint":
return "escalate"
return "generate_response"
# Сборка графа
workflow = StateGraph(SupportTicketState)
workflow.add_node("classify", classify_ticket)
workflow.add_node("generate_response", generate_response)
workflow.add_node("escalate", escalate_to_human)
workflow.set_entry_point("classify")
workflow.add_conditional_edges(
"classify",
route_after_classification,
{
"escalate": "escalate",
"generate_response": "generate_response"
}
)
workflow.add_edge("generate_response", END)
workflow.add_edge("escalate", END)
app = workflow.compile()
# Запуск
result = app.invoke({
"ticket_id": "TKT-001",
"user_message": "Вы списали деньги дважды! Это уже второй раз за месяц!",
"category": None,
"sentiment": None,
"response": None,
"needs_human": False,
"history": []
})
print(f"Нужен оператор: {result['needs_human']}")
print(f"Ответ: {result['response']}")
print(f"Лог обработки: {result['history']}")
Граф явно описывает все возможные пути обработки. Любой инженер, пришедший в проект после вас, может за несколько минут понять, какие состояния существуют и как осуществляются переходы между ними. Это и есть инженерная культура, применённая к AI-системам.
Шаг 5. Архитектура RAG: дать модели доступ к вашим данным
У любой языковой модели есть фундаментальное ограничение: она знает только то, на чём была обучена. Ваша внутренняя документация, база знаний, архивы тикетов, технические спецификации - всё это для неё terra incognita. Паттерн RAG (Retrieval-Augmented Generation) решает именно эту проблему.
Для опытного инженера RAG - это хорошо известная архитектурная связка: поисковый индекс плюс кэш контента плюс сервис обработки запросов. Новизна заключается в том, что поиск осуществляется не по ключевым словам, а по семантической близости - с использованием эмбеддингов.
Схема работы следующая: документы разбиваются на фрагменты, каждый фрагмент превращается в вектор (эмбеддинг) и сохраняется в специализированной векторной базе данных. При поступлении пользовательского запроса система находит семантически близкие фрагменты и передаёт их языковой модели вместе с исходным вопросом. Модель отвечает, опираясь на предоставленный контекст.
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain.schema import Document
# --- Этап 1: Подготовка индекса (выполняется один раз при инициализации) ---
# Ваша документация или база знаний
documents = [
Document(
page_content="""
API аутентификации использует JWT-токены с временем жизни 15 минут.
Refresh-токены хранятся в Redis с TTL 30 дней.
При истечении access-токена клиент должен вызвать endpoint /auth/refresh.
Endpoint /auth/refresh принимает только HTTP POST запросы.
При невалидном refresh-токене возвращается 401 с кодом ошибки TOKEN_EXPIRED.
""",
metadata={"source": "auth-service-docs.md", "section": "authentication"}
),
Document(
page_content="""
Сервис платежей поддерживает следующие провайдеры: Stripe, PayPal, YooKassa.
Все транзакции логируются в таблицу payment_transactions с retention 7 лет.
При сумме транзакции свыше 100 000 рублей требуется дополнительная верификация.
Верификация осуществляется через SMS-код или подтверждение по email.
Максимальное число попыток верификации — 3, после чего транзакция блокируется.
""",
metadata={"source": "payment-service-docs.md", "section": "payments"}
),
Document(
page_content="""
Деплойменты в продакшен осуществляются через GitLab CI/CD.
Каждый деплоймент проходит 4 стадии: test, build, staging, production.
Стадия production требует ручного подтверждения от tech lead или CTO.
Rollback выполняется командой: kubectl rollout undo deployment/<service-name>.
Среднее время полного деплоймента — 12 минут при отсутствии проблем.
""",
metadata={"source": "deployment-runbook.md", "section": "deployments"}
)
]
# Разбиваем на чанки с перекрытием для сохранения контекста
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # Символов в одном фрагменте
chunk_overlap=50, # Перекрытие между фрагментами — важно для сохранения контекста
separators=["\n\n", "\n", ". ", " "]
)
split_docs = text_splitter.split_documents(documents)
# Создаём векторный индекс
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(split_docs, embeddings)
# --- Этап 2: Построение цепочки RAG ---
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3} # Возвращаем 3 наиболее релевантных фрагмента
)
def format_docs(docs):
"""Форматируем найденные документы для передачи в промпт"""
return "\n\n---\n\n".join(
f"Источник: {doc.metadata.get('source', 'неизвестен')}\n{doc.page_content}"
for doc in docs
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", (
"Ты — внутренний технический ассистент компании. "
"Отвечай ТОЛЬКО на основе предоставленной документации. "
"Если ответ не содержится в документации, прямо скажи об этом. "
"Всегда указывай источник информации.\n\n"
"Документация:\n{context}"
)),
("human", "{question}")
])
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# --- Этап 3: Использование ---
questions = [
"Как долго живёт refresh-токен?",
"Что происходит при попытке перевести 150 000 рублей?",
"Сколько длится деплоймент в продакшен?"
]
for question in questions:
print(f"\nВопрос: {question}")
print(f"Ответ: {rag_chain.invoke(question)}")
print("-" * 60)
Здесь важно понять одну вещь: 80% качества RAG-системы определяется не выбором языковой модели, а качеством инженерии вокруг неё. Как вы разбиваете документы на фрагменты? Насколько правильно структурированы метаданные? Как настроены параметры поиска? Это классические инженерные вопросы, а не задачи машинного обучения.
Ключевые решения при построении RAG:
- Размер чанка определяет компромисс между точностью поиска и полнотой контекста. Слишком маленькие чанки - модель не получает достаточного контекста; слишком большие - снижается точность семантического поиска. Хорошая отправная точка для технической документации: 400-600 символов с перекрытием 10-15%.
- Метаданные критически важны для фильтрации. Добавляйте к каждому фрагменту источник, дату обновления, раздел, тип документа. Это позволяет строить гибридный поиск: сначала фильтрация по метаданным, затем семантический поиск внутри отфильтрованного подмножества.
- Оценка качества - это инженерная задача. Стройте набор тестовых пар "вопрос - ожидаемый ответ" и измеряйте качество системы автоматически при каждом изменении параметров.
Шаг 6. Потоковая передача и асинхронность
Вызов языковой модели - это долгая операция. Типичный ответ на сложный запрос занимает от 3 до 30 секунд. Если вы строите API, которое синхронно ждёт полного ответа модели, прежде чем вернуть что-либо клиенту, пользовательский опыт будет неприемлемым.
Решения те же, что и в классическом бэкенде: стриминг и асинхронность.
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3, streaming=True)
prompt = ChatPromptTemplate.from_messages([
("system", "Ты — технический писатель. Пиши чётко и структурированно."),
("human", "{topic}")
])
chain = prompt | llm | StrOutputParser()
# Асинхронный стриминг — идеально для API на FastAPI
async def stream_response(topic: str):
async for chunk in chain.astream({"topic": topic}):
print(chunk, end="", flush=True)
# В реальном API здесь был бы yield chunk для SSE или WebSocket
print() # Новая строка в конце
# Параллельная обработка нескольких запросов
async def process_multiple_queries(queries: list[str]) -> list[str]:
tasks = [chain.ainvoke({"topic": q}) for q in queries]
return await asyncio.gather(*tasks)
# Запуск
asyncio.run(stream_response(
"Объясни разницу между RAG и fine-tuning языковых моделей"
))
```
Пример интеграции со стримингом в FastAPI:
```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import asyncio
app = FastAPI()
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
@app.get("/generate")
async def generate_stream(query: str):
prompt = ChatPromptTemplate.from_messages([
("system", "Ты — полезный технический ассистент."),
("human", "{query}")
])
chain = prompt | llm | StrOutputParser()
async def event_generator():
async for chunk in chain.astream({"query": query}):
yield f"data: {chunk}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream"
)
Обратите внимание: astream не просто ускоряет ответ - он фундаментально меняет пользовательский опыт. Первые токены появляются уже через 200-500 миллисекунд, и пользователь видит, что система работает, даже если полный ответ формируется ещё несколько секунд.
Часть третья: Инженерный фундамент как конкурентное преимущество
Почему опыт классической разработки - это преимущество, а не балласт
Здесь возникает парадокс, который многие не сразу замечают. Рынок AI-инженеров сегодня переполнен людьми, которые умеют вызывать API языковых моделей и получать ответы. Но построить из этого надёжную, масштабируемую, поддерживаемую систему - совсем другая история.
Разработчик с десятью годами опыта проектирования распределённых систем принесёт в AI Engineering то, чего у энтузиастов-самоучек нет: понимание того, как системы ломаются, как они масштабируются, как ими управляют в продакшене.
Observability.
Классический бэкенд-инженер знает, что без нормального логирования, трейсинга и метрик система неуправляема. В AI-системах это особенно важно, потому что вероятностный компонент делает отладку значительно сложнее. Трассировка LLM-вызовов с захватом промптов, токенов и времени ответа - это не опциональная функция, это базовая good-practice.
import time
import logging
from functools import wraps
from dataclasses import dataclass
from typing import Any
logger = logging.getLogger(__name__)
@dataclass
class LLMCallMetrics:
model: str
prompt_tokens: int
completion_tokens: int
latency_ms: float
success: bool
error: str | None = None
def track_llm_call(func):
"""Декоратор для автоматического трекинга вызовов LLM"""
@wraps(func)
async def wrapper(*args, **kwargs):
start = time.time()
try:
result = await func(*args, **kwargs)
latency = (time.time() - start) * 1000
# В реальной системе здесь — отправка в Prometheus/DataDog
logger.info(
"LLM call completed",
extra={
"latency_ms": latency,
"function": func.__name__,
}
)
return result
except Exception as e:
latency = (time.time() - start) * 1000
logger.error(
"LLM call failed",
extra={
"latency_ms": latency,
"error": str(e),
"function": func.__name__,
}
)
raise
return wrapper
Управление стоимостью.
Каждый вызов языковой модели стоит денег. Классический инженер немедленно задаётся вопросами кэширования: какие запросы повторяются? Какой TTL у кэша разумен для данного типа запросов? Можно ли батчевать вызовы?
Обработка ошибок.
LLM API возвращает ошибки: rate limits, timeout, временная недоступность сервиса. Стратегия retry с экспоненциальным backoff, circuit breaker, fallback на альтернативную модель - всё это стандартная практика надёжного бэкенда, перенесённая в новый контекст.
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import RateLimitError, APITimeoutError
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10),
retry=retry_if_exception_type((RateLimitError, APITimeoutError))
)
async def resilient_llm_call(messages: list, model: str = "gpt-4o-mini"):
"""Вызов LLM с автоматическим retry при временных ошибках"""
response = await client.chat.completions.acreate(
model=model,
messages=messages,
timeout=30
)
return response
Тестирование.
Как тестировать систему, компонент которой вероятностен? Классический инженер инстинктивно понимает, что нужно изолировать вероятностный компонент через моки и тестировать остальную логику детерминированно. Тесты на интеграцию с реальной моделью - это отдельный, более редкий уровень тестирования, который запускается не при каждом коммите.
Инструментальный стек AI-инженера в 2026 году
Несколько слов о выборе инструментов. Экосистема AI-разработки меняется быстрее, чем большинство других областей, и здесь важно понимать, что именно находится под капотом инструментов, которые вы используете.
Оркестрация и цепочки: LangChain остаётся наиболее распространённым фреймворком для построения LLM-приложений. LangGraph - его расширение для сложных агентных систем. Для более лаконичного кода существует LlamaIndex, особенно популярный в RAG-сценариях.
Векторные базы данных: Chroma подходит для прототипов и небольших объёмов данных; Pinecone - для облачных продакшен-решений; Qdrant и Weaviate - для самостоятельного развёртывания с расширенными возможностями фильтрации; pgvector - если вы уже используете PostgreSQL и не хотите вводить дополнительную инфраструктуру.
Observability: LangSmith (от команды LangChain) - специализированный инструмент для трассировки LLM-вызовов; Weights & Biases - для более глубокого мониторинга и экспериментов; Phoenix от Arize - для оценки качества RAG-систем.
Оценка качества: RAGAS - фреймворк для автоматической оценки RAG-систем по метрикам faithfulness (верность источнику), answer relevancy (релевантность ответа) и context precision (точность контекста).
Заключение: правильный взгляд на трансформацию
Переход в AI Engineering - это не смена профессии. Это расширение профессионального контекста. Если переформулировать это в инженерных терминах: вы не мигрируете на новую платформу, вы добавляете новый слой абстракции поверх существующего стека.
Технические концепции, которые мы рассмотрели в этой статье: типизированный вывод, управление состоянием через конечные автоматы, поиск по векторным индексам, асинхронная обработка, стриминг - это не революционно новые идеи. Это переосмысление хорошо известных инженерных принципов применительно к новому классу компонентов.
Несколько практических рекомендаций для тех, кто готов начать:
Начните с задачи, которую вы хорошо понимаете. Возьмите реальную проблему из своей области: анализ логов, автоматизация code review, обработка технической документации, и решите её с помощью языковой модели. Знание предметной области поможет вам объективно оценить качество результата.
Измеряйте всё с самого начала. Добавляйте логирование LLM-вызовов до того, как система уйдёт в продакшен. Стоимость, задержка, частота ошибок - эти данные необходимы для принятия архитектурных решений.
Читайте исходный код инструментов, которыми пользуетесь. LangChain, LangGraph - это Python-библиотеки с открытым кодом. Понимание того, что происходит под капотом, помогает избежать многих неожиданных проблем в продакшене.
Не переоценивайте роль модели. Подавляющее большинство проблем в AI-приложениях - это инженерные проблемы: качество данных, архитектура системы, обработка ошибок, управление состоянием. Смена модели к примеру с GPT-4o на Claude или наоборот редко кардинально меняет ситуацию. Инженерные решения - меняют.
Искусственный интеллект не заменит инженеров-программистов. Но инженеры, которые поняли, как встроить AI в архитектуру систем, как управлять его нестабильностью, масштабировать его вызовы, тестировать его поведение и обеспечивать его надёжность, станут теми, кто будет проектировать следующее поколение программных систем.
У вас уже есть для этого фундамент. Теперь есть и карта.
