Зачем останавливать агента перед публикацией? Потому что он тупит
Вы настроили автономного агента постить в Bluesky. Он работает день, два, неделю. А потом генерирует текст про "преимущества каннибализма для экологии" или делает орфографические ошибки в каждом слове. Классика.
Полная автономность агентов — это как дать пятилетнему ребенку доступ к вашему аккаунту в соцсетях. Звучит смешно, пока не случается катастрофа. Решение? Human-in-the-Loop (HITL) — архитектура, где критичные решения принимает человек, а рутина остаётся за ИИ.
Текущий контекст (25.03.2026): LangGraph 0.2.1 поддерживает нативные интерфейсы для прерываний. GPT-4.5 Turbo — последняя стабильная модель OpenAI. Bluesky полностью открыл API для ботов, но требует модерации контента.
Архитектура: где вставить человека в конвейер
Не нужно прерывать агента на каждом шаге. Только в точках, где ошибка стоит дорого. Для Bluesky-бота это:
- Генерация текста поста (модель может галлюцинировать)
- Добавление медиа (картинки должны быть релевантными)
- Финальная публикация (точка невозврата)
В LangGraph это реализуется через ноды с типом interrupt. Граф исполняется, доходит до такой ноды — и замирает, ждёт внешнего сигнала. Человек проверяет, нажимает "ок" или редактирует — граф продолжает работу.
Собираем конструктор: от идеи до работающего кода
1 Готовим инструменты
Установите актуальные версии. На 25.03.2026 вот что работает стабильно:
pip install langgraph==0.2.1 langchain-openai==0.3.0 atproto==0.1.8 python-dotenv==1.0.0LangGraph 0.2.1 принёс упрощённый API для human-in-the-loop. Раньше приходилось городить костыли с кастомными состояниями, теперь есть встроенные примитивы.
2 Определяем состояние графа
Сохраняем всё, что нужно между шагами. Особенно — текст для утверждения человеком.
from typing import TypedDict, Optional
from langgraph.graph import add_messages
class AgentState(TypedDict):
"""Состояние нашего агента"""
topic: str # исходная тема для поста
draft_text: Optional[str] # черновик, сгенерированный ИИ
approved_text: Optional[str] # текст, утверждённый человеком
media_path: Optional[str] # путь к медиафайлу
published: bool # опубликовано или нет
human_feedback: Optional[str] # если человек решил отредактировать
# Используем новое API LangGraph 0.2.1 для управления историей
StateGraph = StateGraph(AgentState).add_messages()add_messages() автоматически сериализует историю диалога, что критично для human-in-the-loop сценариев, где контекст должен сохраняться между прерываниями.3 Создаём ноды: где работает ИИ, где ждёт человека
Сначала нода генерации текста. Используем GPT-4.5 Turbo — на март 2026 это самая сбалансированная модель по цене/качеству для творческих задач.
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
def generate_post(state: AgentState) -> AgentState:
"""ИИ генерирует черновик поста на заданную тему"""
llm = ChatOpenAI(
model="gpt-4.5-turbo",
temperature=0.7,
max_tokens=500
)
prompt = ChatPromptTemplate.from_messages([
("system", "Ты — автор технического блога на Bluesky. Пиши кратко, по делу, с долей иронии. Максимум 280 символов."),
("human", "Напиши пост на тему: {topic}")
])
chain = prompt | llm
draft = chain.invoke({"topic": state["topic"]})
# Обновляем состояние
return {"draft_text": draft.content}Теперь ключевая часть — нода human review. В LangGraph 0.2.1 для этого есть специальный декоратор.
from langgraph.graph import interrupt
from langgraph.types import Command
@interrupt(before=["approve_post"])
async def human_review(state: AgentState) -> Command[AgentState]:
"""Останавливаем граф и ждём решения человека"""
# В реальном приложении здесь будет:
# 1. Отправка уведомления (Telegram, email, веб-интерфейс)
# 2. Ожидание ответа через callback или webhook
# 3. Получение решения: approve, reject, или edited_text
print(f"\n=== ТРЕБУЕТСЯ ПОДТВЕРЖДЕНИЕ ===")
print(f"Текст для публикации: {state['draft_text']}")
print("1. Одобрить и опубликовать")
print("2. Отклонить (граф завершится)")
print("3. Ввести исправленный текст вручную")
# Для демо — симуляция ввода из консоли
# В продакшене это будет асинхронный callback
choice = input("Ваш выбор (1-3): ")
if choice == "1":
return Command(
update={"approved_text": state["draft_text"]},
goto="publish_post"
)
elif choice == "2":
print("Пост отклонён человеком")
return Command(update={"published": False}, goto="__end__")
else:
# Человек ввёл исправленный текст
corrected = input("Введите исправленный текст: ")
return Command(
update={
"approved_text": corrected,
"human_feedback": "Текст отредактирован человеком"
},
goto="publish_post"
)Внимание на архитектуру: Нода human_review возвращает Command, а не изменённое состояние. Это новая парадигма в LangGraph 0.2.1 — граф получает команду «куда идти дальше» и «что обновить». Так human-in-the-loop интегрируется естественно в поток.
4 Публикация в Bluesky через atproto
Когда человек одобрил — публикуем. Используем официальную библиотеку atproto.
import os
from atproto import Client, client_utils
def publish_to_bluesky(state: AgentState) -> AgentState:
"""Публикуем утверждённый текст в Bluesky"""
# Авторизация (ключи в .env)
client = Client()
client.login(
os.getenv("BLUESKY_HANDLE"),
os.getenv("BLUESKY_APP_PASSWORD")
)
# Создаём пост
post_text = state["approved_text"]
# Если есть медиа — загружаем
if state.get("media_path"):
with open(state["media_path"], "rb") as f:
img_data = f.read()
upload = client.upload_blob(img_data)
embed = client_utils.EmbedImages(
images=[client_utils.Image(
alt="Иллюстрация к посту",
image=upload.blob
)]
)
client.send_post(text=post_text, embed=embed)
else:
client.send_post(text=post_text)
return {"published": True}5 Собираем граф и запускаем
from langgraph.graph import StateGraph, START, END
# Создаём граф
graph_builder = StateGraph(AgentState)
# Добавляем ноды
graph_builder.add_node("generate", generate_post)
graph_builder.add_node("human_review", human_review) # interrupt-нода
graph_builder.add_node("publish", publish_to_bluesky)
# Определяем поток
graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", "human_review")
# От human_review идём туда, куда решит человек
# Публикация или конец — определяется в Command
# Компилируем граф
graph = graph_builder.compile()
# Запускаем
initial_state = {"topic": "Новые фичи LangGraph 0.2.1 для human-in-the-loop"}
result = graph.invoke(initial_state)
print(f"Итоговое состояние: {result}")Ошибки, которые съедят ваше время (проверено на себе)
| Ошибка | Почему происходит | Как исправить |
|---|---|---|
| Граф зависает после interrupt | Не настроен callback или webhook для возврата управления | Использовать graph.update_state() с внешнего эндпоинта |
| Состояние сбрасывается между прерываниями | Неправильная конфигурация persistence в StateGraph | В LangGraph 0.2.1 обязательно использовать .add_messages() |
| Bluesky блокирует посты как спам | Слишком частые публикации или шаблонный текст | Добавить задержки и вариативность в генерацию |
Человек против машины: кто кого
Human-in-the-loop — не про то, чтобы заменить человека. И не про то, чтобы полностью довериться ИИ. Это про распределение работы:
- ИИ делает: генерация идей, черновики, поиск информации, рутинные проверки
- Человек делает: финальное одобрение, корректура, этические решения, креативный контроль
В продакшене такой агент экономит 80% времени на создание контента, но сохраняет 100% контроля над тем, что уходит в публичное пространство.
А что если нужно масштабировать на команду?
Один человек утверждает посты — скучно. Десять человек — уже нужна очередь задач и ролевая модель. Вот как расширять архитектуру:
- Заменяем консольный input на веб-интерфейс (FastAPI + WebSocket)
- Добавляем очередь заданий (Redis или PostgreSQL)
- Реализуем систему приоритетов: срочные посты идут быстрее
- Добавляем историю решений для обучения модели (почему человек поправил именно так?)
Кстати, для production-развёртывания таких агентов есть отличный инструмент — LangGraph Deploy CLI. Он упаковывает граф в Docker, настраивает CI/CD и даже мониторинг.
Будущее: когда ИИ научится не тупить
Human-in-the-loop — временное решение. Временное, растянутое на ближайшие 5-7 лет. Модели станут надёжнее, но полностью доверять им публичную коммуникацию мы не будем никогда. Или будем?
Самый вероятный сценарий: человеческий контроль сместится с проверки каждого поста на настройку "тона голоса" и долгосрочной стратегии. Вместо "поправь опечатку в третьем слове" — "следующую неделю пиши в более формальном стиле".
А пока — собирайте агентов с interrupt-нодами, настраивайте уведомления, и пусть ваши посты в Bluesky всегда будут адекватными. Хотя бы потому, что их проверил живой человек перед публикацией.
