Блог об ИИ-агентах
СИСТЕМА КОМПЬЮТЕРНОГО ДОСТУПА

Конечные Автоматы и Агенты: Построение Надёжных Рабочих Процессов с LangGraph

agentsarchitecturetutoriallanggraphautonomous-systems

Большинство туториалов по агентам показывают простой цикл: спросить Claude, разобрать ответ, вызвать инструмент, повторить. Для демонстраций это работает. В продакшене нужны детерминизм, восстановление после ошибок, точки человеческого одобрения и возможность аудита.

LangGraph привносит конечные автоматы в рабочие процессы агентов. Вместо специального цикла из if-операторов вы получаете явный граф: именованные узлы (логические единицы), типизированные рёбра (переходы) и общую схему состояния, которая проходит через всё выполнение.

Почему Конечные Автоматы для Агентов?

Проблема Специального Цикла

Типичный цикл агента выглядит так:

messages = []
while True:
    response = claude.messages.create(model=..., messages=messages)
    if response.stop_reason == "end_turn":
        break
    for tool_call in get_tool_calls(response):
        result = execute_tool(tool_call)
        messages.append(tool_result(tool_call.id, result))

Для двух-трёх инструментов это читаемо. Добавьте пять инструментов, условные пути, шаг одобрения человеком и логику повторных попыток — и получите сотни строк запутанного потока управления.

Глубинная проблема — неявное состояние. На каком этапе находится агент? Какие данные он собрал? Всё живёт в messages — нетипизированном блобе без принудительной схемы.

Конечные Автоматы как Решение

Конечный автомат делает неявное явным. Вы определяете:

  • Узлы — дискретные логические единицы, получающие текущее состояние, выполняющие одну задачу и возвращающие обновления состояния.
  • Рёбра — переходы между узлами, безусловные (A → B всегда) или условные.
  • Состояние — типизированный словарь, проходящий через весь граф. Схема валидируется на каждом шаге.

Когда НЕ Использовать LangGraph

Для простых однократных задач прямой API-вызов быстрее и нагляднее. Используйте LangGraph, когда рабочий процесс включает:

  • Несколько различных этапов, выполняемых последовательно
  • Условное ветвление на основе промежуточных результатов
  • Шаги с участием человека
  • Логику восстановления после ошибок или повторных попыток
  • Требования аудитируемости

Основы LangGraph

Определение Состояния

from typing import TypedDict


class ResearchState(TypedDict):
    query: str
    research_notes: str
    draft: str
    review_feedback: str
    is_approved: bool

Узлы

import anthropic


client = anthropic.Anthropic()


def research_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Исследуй эту тему детально: {state['query']}"
        }]
    )
    return {"research_notes": response.content[0].text}


def draft_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Напиши черновик на основе этих заметок:\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

Рёбра и Условная Маршрутизация

from typing import Literal


def route_after_check(state: ResearchState) -> Literal["human_review", "draft"]:
    if state.get("review_feedback"):
        return "human_review"
    return "draft"


workflow.add_conditional_edges(
    "research",
    route_after_check,
    {"human_review": "review_node", "draft": "draft_node"}
)

Создание и Запуск Графа

from langgraph.graph import StateGraph, END


workflow = StateGraph(ResearchState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", END)


graph = workflow.compile()
result = graph.invoke({"query": "Что такое протокол MCP?"})
print(result["draft"])

Построение Рабочего Процесса Проверки Документов

Проектирование Состояния

class DocumentState(TypedDict):
    document: str
    key_terms: list[str]
    compliance_issues: list[str]
    summary: str
    human_feedback: str
    is_approved: bool

Реализация Узлов

import json


def extract_terms_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "Извлеки ровно 5 ключевых терминов из этого документа. "
                "Верни их как JSON-массив строк.\n\n"
                f"Документ: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        terms = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        terms = [t.strip() for t in response.content[0].text.split("\n") if t.strip()][:5]
    return {"key_terms": terms}




def check_compliance_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                "Проверь этот документ на проблемы соответствия. "
                "Ищи: персональные данные (имена, СНИЛС, email, телефоны), "
                "конфиденциальные пометки, непроверенные утверждения.\n\n"
                "Верни JSON-массив описаний проблем. "
                "Верни пустой массив [], если проблем не обнаружено.\n\n"
                f"Документ: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        issues = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        issues = []
    return {"compliance_issues": issues}




def human_review_node(state: DocumentState) -> dict:
    print("\n=== ТРЕБУЕТСЯ ПРОВЕРКА ЧЕЛОВЕКОМ ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "Проверено и одобрено после редактирования персональных данных",
        "is_approved": True
    }




def summarize_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"Кратко изложи этот документ ровно в двух предложениях:\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

Сборка Графа

def route_after_compliance(state: DocumentState) -> Literal["human_review", "summarize"]:
    if state.get("compliance_issues"):
        return "human_review"
    return "summarize"


workflow = StateGraph(DocumentState)
workflow.add_node("extract", extract_terms_node)
workflow.add_node("check", check_compliance_node)
workflow.add_node("review", human_review_node)
workflow.add_node("summarize", summarize_node)


workflow.set_entry_point("extract")
workflow.add_edge("extract", "check")
workflow.add_conditional_edges(
    "check", route_after_compliance,
    {"human_review": "review", "summarize": "summarize"}
)
workflow.add_edge("review", "summarize")
workflow.add_edge("summarize", END)


graph = workflow.compile()


result = graph.invoke({
    "document": "Пациент Иван Петров (СНИЛС: 123-456-789 00) страдает гипертонией.",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

Контрольные Точки Человека и Прерывания

from langgraph.checkpoint.memory import MemorySaver


checkpointer = MemorySaver()
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)


thread_config = {"configurable": {"thread_id": "doc-review-001"}}
result = graph.invoke(initial_state, config=thread_config)


current_state = graph.get_state(thread_config)
print("Проблемы:", current_state.values["compliance_issues"])


graph.update_state(
    thread_config,
    {"human_feedback": "Одобрено после ручного редактирования", "is_approved": True}
)


final_result = graph.invoke(None, config=thread_config)

Персистентность и Продакшн-развёртывание

from langgraph.checkpoint.postgres import PostgresSaver


with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    graph = workflow.compile(checkpointer=checkpointer)


import asyncio


async def batch_process(documents: list[str]) -> list[dict]:
    return await asyncio.gather(*[
        graph.ainvoke({"document": doc, "key_terms": [], "compliance_issues": [],
                       "summary": "", "human_feedback": "", "is_approved": False})
        for doc in documents
    ])

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

  • Разветвление / Объединение: Параллельный запуск нескольких узлов через Send с объединением результатов.
  • Чрезмерно сложная маршрутизация: При слишком многих ветвях — разбить на подграфы.
  • Раздутое состояние: Хранить ссылки (ID базы данных, ключи S3), а не большие артефакты.
  • Изолированное тестирование узлов: Поскольку узлы — обычные функции, их можно юнит-тестировать напрямую.
def test_extract_terms_node():
    state = {
        "document": "Искусственный интеллект быстро меняет современное общество.",
        "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    }
    result = extract_terms_node(state)
    assert "key_terms" in result
    assert isinstance(result["key_terms"], list)

LangGraph заменяет специальные циклы явными, отлаживаемыми и возобновляемыми рабочими процессами. Начните с линейного графа (A → B → C → КОНЕЦ). Добавьте одно условное ребро, когда нужно ветвление. Добавьте контрольную точку человека, когда нужно одобрение. Большинству продакшн-агентов нужны именно эти три паттерна.

Связанные статьи