Створіть свої перші робочі процеси LangGraph ще до систем багатьох агентів
Перші робочі процеси LangGraph у Colab: спільний стан, послідовні вузли, один крок LLM та багатоетапний потік для блогу, ще до використання багатоагентних систем.
Джерела: Build AI Workflows with LangGraph (Part 1): Your First Agentic AI Application
Що таке LangGraph?
Звичайні скрипти на Python виконуються зверху вниз. Агентні додатки працюють у форматі графу: кожен вузол виконує певну задачу, а потім передає керування через ребро. Мінімальна схема виглядає так:
Input
↓
Node A
↓
Node B
↓
Node C
↓
Output
Кожен вузол отримує спільний стан, оновлює його та передає далі. Ця схема може розширюватися за рахунок гілкування, використання пам’яті, планування, рефлексії та багатоагентних конфігурацій — але першою навичкою є створення простого потоку обробки даних.
Наведений нижче приклад залишається відтворюваним: Google Colab, моделі з відкритим кодом через Groq та короткий список залежностей. Не потрібні локальні GPU-системи. Педагогічна ідея полягає у тому, що агентні підходи впроваджуються швидше, коли кожен етап виконується на безкоштовній інфраструктурі, тож увага зосереджується на стані, вузлах та ребрах, а не на налаштуванні кластеру.
Встановлення залежностей
!pip install langchain langgraph langchain_groq
- LangGraph створює граф робочого процесу
Запуск інсталяцій усередині 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 змініть пороги категорій та перевірте оновлення стану. Після виконання графа для одного запитання до LLM замініть ідентифікатор моделі або шаблон запиту та спостерігайте лише за змінами в поведінці цього вузла. Після виконання пайплайну блогу додайте додатковий вузол між оглядом та чернеткою — можливо, „перевірку тону“ — щоб побачити, як разом зростають краї та поля стану.
На цьому етапі дисципліна у роботі зі спільним станом важливіша за кмітливі запити. Використовуйте невеликі, іменовані поля (outline, content, rating) замість одного величезного блоку messages, поки у вас не з’явиться потреба у історії чату. Допускається повертати часткове оновлення, у якому відсутні непов’язані ключі, якщо семантика об’єднання графів це дозволяє; важлива звичка — щоб вузли залишалися достатньо „чистими“ для тестування за допомогою вручну створеного словника.
У Colab обробка помилок може залишатися мінімальною, проте все одно свідомою: якщо виклик Groq зазнає невдачі, необхідно відобразити виниклу виняткову ситуацію, а не приховати її всередині вузла. Ранні навчальні матеріали, які приховують невдачі, формують хибну звичку для подальшого використання графів з викликами інструментів, де невдалий інструмент має стати окремою гілкою.
Нарешті, кожного разу, коли ви додаєте ребро, малюйте скомпільований граф. Візуальна перевірка дозволяє швидше виявити випадкові цикли та забуті посилання END, ніж аналізувати виклики add_edge. Як тільки зображення відповідає історії, яку ви хотіли розповісти, invoke стає кроком перевірки, а не сеансом виправлення помилок.
Що демонструють ці приклади
Три послідовні сценарії охоплюють основний словниковий запас:
- Чиста послідовна логіка Python з використанням спільного стану
- Один вузол LLM у межах того ж шаблону графа
Продовження: спільний стан, вузли, ребра, оркестрація, інтеграція з LLM та поетапне міркування. Ці ж ідеї згодом підтримують інструменти, пам’ять, розгалуження та агентів для співпраці.
Що далі
У наступних частинах проекту Agentic AI з використанням Python зазвичай додаються умовні ребра, паралельні гілки, механізми схвалення людиною, виклик інструментів, технологія RAG, структури з кількома агентами та пакування для продакшну — все це з метою створення конфігурацій, сумісних із Colab та використанням відкритих моделей.
Повний вихідний код
Супровідний ноутбук для Colab опублікований у проекті Agentic-AI на GitHub (файл Workflow_in_LangGraph.ipynb). Відкрийте цей ноутбук у Colab, виконайте комірки зверху вниз, а потім створіть власні вузли для ваших завдань.