Головна / Статті

Створіть свої перші робочі процеси LangGraph ще до систем багатьох агентів

Перші робочі процеси LangGraph у Colab: спільний стан, послідовні вузли, один крок LLM та багатоетапний потік для блогу, ще до використання багатоагентних систем.

Видавець: REACTAPP.TOP8 хв читання1782 слів

Джерела: 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 створює граф робочого процесу
  • 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 змініть пороги категорій та перевірте оновлення стану. Після виконання графа для одного запитання до LLM замініть ідентифікатор моделі або шаблон запиту та спостерігайте лише за змінами в поведінці цього вузла. Після виконання пайплайну блогу додайте додатковий вузол між оглядом та чернеткою — можливо, „перевірку тону“ — щоб побачити, як разом зростають краї та поля стану.

    На цьому етапі дисципліна у роботі зі спільним станом важливіша за кмітливі запити. Використовуйте невеликі, іменовані поля (outline, content, rating) замість одного величезного блоку messages, поки у вас не з’явиться потреба у історії чату. Допускається повертати часткове оновлення, у якому відсутні непов’язані ключі, якщо семантика об’єднання графів це дозволяє; важлива звичка — щоб вузли залишалися достатньо „чистими“ для тестування за допомогою вручну створеного словника.

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

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

    Що демонструють ці приклади

    Три послідовні сценарії охоплюють основний словниковий запас:

    1. Чиста послідовна логіка Python з використанням спільного стану
    2. Один вузол LLM у межах того ж шаблону графа
  • Багатокрокова ланцюгова структура ШІ з кількома взаємодіючими вузлами
  • Продовження: спільний стан, вузли, ребра, оркестрація, інтеграція з LLM та поетапне міркування. Ці ж ідеї згодом підтримують інструменти, пам’ять, розгалуження та агентів для співпраці.

    Що далі

    У наступних частинах проекту Agentic AI з використанням Python зазвичай додаються умовні ребра, паралельні гілки, механізми схвалення людиною, виклик інструментів, технологія RAG, структури з кількома агентами та пакування для продакшну — все це з метою створення конфігурацій, сумісних із Colab та використанням відкритих моделей.

    Повний вихідний код

    Супровідний ноутбук для Colab опублікований у проекті Agentic-AI на GitHub (файл Workflow_in_LangGraph.ipynb). Відкрийте цей ноутбук у Colab, виконайте комірки зверху вниз, а потім створіть власні вузли для ваших завдань.