Создайте свои первые рабочие процессы LangGraph до реализации многоподразделенческих систем

Что такое LangGraph?

Обычные скрипты на Python выполняются сверху вниз. Агентные приложения работают в рамках графа: каждый узел выполняет одну задачу, затем передает управление по ребру. Простая схема выглядит так:

Input
   ↓
Node A
   ↓
Node B
   ↓
Node C
   ↓
Output

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

Приведённый ниже пример легко воспроизводим: Google Colab, модели с открытым исходным кодом через Groq и короткий список зависимостей. Не требуется локальная ферма GPU. Педагогическая идея заключается в том, что концепции агентных систем лучше усваиваются, когда каждая операция выполняется на бесплатной инфраструктуре, что позволяет сосредоточиться на состоянии, узлах и ребрах, а не на настройке кластера.

Установка зависимостей

!pip install langchain langgraph langchain_groq

• LangGraph создаёт граф рабочего процесса

LangChain обеспечивает абстракции для LLM

LangChain Groq взаимодействует с API инференса Groq

Запуск установок внутри Colab позволяет сохранять среду временной и общедоступной.

Импорт необходимых библиотек

from langgraph.graph import StateGraph, START, END
from typing import TypedDict
from IPython.display import Image

• StateGraph формирует рабочий процесс
• START / END обозначают начало и конец выполнения
• TypedDict определяет типы общедоступного состояния
• Image отображает диаграмму скомпилированной структуры

Пример 1 — Простой последовательный рабочий процесс

Начните без использования LLM, чтобы путь обработки данных был очевиден.

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

class BMIState(TypedDict):
    weight_kg: float
    height_m: float
    bmi: float
    category: str

Состояние — это общий словарь, который передается вместе с выполнением. Узлы читают его, записывают данные и возвращают обновленную структуру. В этом примере хранятся вес, рост, ИМТ и метка категории.

Создание узла обработки

# define calculate_bmi node
def calculate_bmi(state: BMIState) -> BMIState:

    weight = state['weight_kg']
    height = state['height_m']

    bmi = weight/(height**2)

    state['bmi'] = round(bmi, 2)

    return state

# define label_bmi node
def label_bmi(state: BMIState) -> BMIState:
    bmi = state['bmi']

    if bmi < 18.5:
        state["category"] = "Underweight"
    elif 18.5 <= bmi < 25:
        state["category"] = "Normal"
    elif 25 <= bmi < 30:
        state["category"] = "Overweight"
    else:
        state["category"] = "Obese"

    return state

Функция считывает рост и вес, вычисляет ИМТ, присваивает категорию и возвращает новое состояние. Принцип работы LangGraph прост: он принимает текущее состояние и возвращает следующее.

Создание графа

# define graph
graph = StateGraph(BMIState)

# add nodes to your graph
graph.add_node('calculate_bmi', calculate_bmi)
graph.add_node('label_bmi', label_bmi)

# add edges to your graph
graph.add_edge(START, 'calculate_bmi')
graph.add_edge('calculate_bmi', 'label_bmi')
graph.add_edge('label_bmi', END)


# compile the graph
workflow = graph.compile()

# show the graph
Image(workflow.get_graph().draw_mermaid_png())

Регистрируются узлы, соединяется путь START → узел → END, после чего происходит компиляция. LangGraph может нарисовать полученный граф, чтобы вы могли проверить его топологию перед запуском.

Запуск рабочего процесса

# execute the graph
intial_state = {'weight_kg':80, 'height_m':1.73}
final_state = workflow.invoke(intial_state)
print(final_state)

Подается начальный словарь с данными о весе и росте. Функция invoke выполняет указанный путь и возвращает состояние, в котором теперь указаны ИМТ и категория.

Пример 2 — последовательный рабочий процесс с LLM

Заменяется узел для вычислений на вызов модели. Форма графа остается прежней.

Безопасным образом собирается ключ API Groq (по запросу, а не через жесткую привязку), затем загружается доступная открытая модель:

model = ChatGroq(model="openai/gpt-oss-120b")

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

# create a state
class LLMState(TypedDict):
    question: str
    answer: str

Поля сокращаются до вопроса и ответа — форма состояния определяется процессом работы, а не глобальной схемой.

Создание узла LLM

def llm_qa(state: LLMState) -> LLMState:

    # extract the question from state
    question = state['question']

    # form a prompt
    prompt = f'Answer the following question {question}'

    # ask that question to the LLM
    answer = model.invoke(prompt).content

    # update the answer in the state
    state['answer'] = answer

    return state

Читается вопрос, вызывается модель чата, сохраняется ответ, возвращается состояние. Структура идентична узлу BMI; меняется только внутренняя логика обработки.

Создание рабочего процесса LLM

# create our graph
graph = StateGraph(LLMState)

# add nodes
graph.add_node('llm_qa', llm_qa)

# add edges
graph.add_edge(START, 'llm_qa')
graph.add_edge('llm_qa', END)

# compile
workflow = graph.compile()

Добавляется узел, соединяются ребра, происходит компиляция. Структура остается знакомой намеренно.

Задание вопроса LLM

# execute
intial_state = {'question': 'How far is sun from the earth?'}
final_state = workflow.invoke(intial_state)
print(final_state['answer'])

Вызывается модель с строкой вопроса, после чего выводится поле с ответом. Модель находится внутри графа, как и любой другой шаг.

Пример 3 — многоэтапный рабочий процесс ИИ

В реальных системах последовательно выполняются несколько этапов обработки информации. Можно расширить эту схему.

Определение состояния блога

class BlogState(TypedDict):
    title: str
    outline: str
    content: str
    rating: int

По мере развития системы отслеживаются название, структура, содержимое и оценка качества.

Создание узлов рабочего процесса

def create_outline(state: BlogState) -> BlogState:
    # fetch title
    title = state['title']

    # call llm gen outline
    prompt = f'Generate a detailed outline for a blog on the topic - {title}'
    outline = model.invoke(prompt).content

    # update state
    state['outline'] = outline

    return state

def create_blog(state: BlogState) -> BlogState:
    title = state['title']
    outline = state['outline']

    prompt = f'Write a short blog (less than 20 lines) on the title - {title} using the follwing outline \n {outline}'
    content = model.invoke(prompt).content
    state['content'] = content
    return state

def create_rating(state: BlogState) -> BlogState:
    title = state["title"]
    content = state["content"]

    prompt = f"""
    Rate the following blog out of 10.
    Return ONLY a single integer between 1 and 10.

    Title: {title}

    Blog:
    {content}
    """

    rating = model.invoke(prompt).content.strip()

    state["rating"] = int(rating)

    return state

Отдельные функции занимаются генерацией структуры и последующими этапами. Каждый узел выполняет одну конкретную задачу, что обеспечивает читаемость графов.

Соберите полную цепочку обработки

graph = StateGraph(BlogState)

# nodes
graph.add_node('create_outline', create_outline)
graph.add_node('create_blog', create_blog)
graph.add_node('create_rating', create_rating)

# edges
graph.add_edge(START, 'create_outline')
graph.add_edge('create_outline', 'create_blog')
graph.add_edge('create_blog', 'create_rating')
graph.add_edge('create_rating', END)

workflow = graph.compile()
workflow

Соедините задачи ИИ в определенном порядке. Концептуально:

Title
   ↓
Create Outline
   ↓
Generate Blog
   ↓
Evaluate Blog
   ↓
END

Разделение работы на этапы лучше, чем включение всех инструкций в один огромный запрос: каждый шаг определяет условия для следующего.

Запустите рабочий процесс

intial_state = {'title': 'Rise of AI in India'}
final_state = workflow.invoke(intial_state)
print(final_state['rating'])

Начиная с заголовка, система создает структуру, составляет черновик контента, оценивает его качество и возвращает значение рейтинга. Это основа агентных цепочек обработки: координация работы нескольких моделей в рамках одного общего состояния.

Рассматривайте каждый пример как шаг калибровки, а не как временный демонстрационный код. После запуска графика BMI измените пороговые значения категорий и проверьте обновления состояния. После запуска графика для одного вопроса к ЯИ, замените идентификатор модели или шаблон запроса и наблюдайте только за изменением поведения соответствующего узла. После запуска процесса обработки блогов вставьте дополнительный узел между этапами составления плана и черновика — например, узел для проверки тона — чтобы понять, как вместе развиваются связи между узлами и поля состояния.

На этом этапе более важна дисциплина в управлении общим состоянием, чем изощренные шаблоны запросов. Используйте небольшие, именованные поля (outline, content, rating) вместо одного огромного поля messages, пока у вас не появится потребность в истории чатов. Возвращение частичного обновления без ненужных ключей допустимо, если семантика объединения графов это позволяет; важной привычкой является сохранение чистоты узлов, чтобы их можно было тестировать с помощью вручную созданного словаря.

В Colab обработка ошибок может оставаться минимальной, но при этом осознанной: если вызов Groq терпит неудачу, необходимо выявить исключение, а не поглощать его внутри узла. Ранние учебные материалы, скрывающие неудачи, формируют неверный подход к последующим схемам вызова инструментов, где неудачный инструмент должен стать отдельной ветвью.

Наконец, рисуйте скомпилированную схему каждый раз при добавлении ребра. Визуальная проверка позволяет быстрее обнаружить случайные циклы и забытые ссылки END, чем анализ вызовов add_edge. Как только изображение соответствует той истории, которую вы хотели рассказать, операция invoke превращается в шаг верификации, а не в сеанс отладки.

Что демонстрируют эти примеры

Три уровня сложности рабочих процессов охватывают основной набор элементов:

1. Чистая последовательная логика на Python с использованием общего состояния
2. Один узел LLM внутри той же схемы графа

Многоэтапная цепочка ИИ с несколькими взаимодействующими узлами

Основные элементы: общее состояние, узлы, рёбра, оркестрация, интеграция с LLM и поэтапное рассуждение. Эти же принципы позже используются для поддержки инструментов, памяти, разветвлённых структур и сотрудничающих агентов.

Что дальше

В последующих версиях агентного ИИ с использованием Python обычно добавляются условные рёбра, параллельные ветви, механизмы одобрения от человека, вызов инструментов, технология RAG, структуры с несколькими агентами и инструменты для производственного использования — всё это при сохранении цели создания удобных для Colab конфигураций с открытыми моделями.

Полный исходный код

Сопутствующий ноутбук для Colab опубликован в проекте Agentic-AI на GitHub (файл Workflow_in_LangGraph.ipynb). Откройте этот ноутбук в Colab, выполните ячейки сверху вниз, затем скопируйте узлы для своих задач.