Stwórz swoje pierwsze przepływy pracy LangGraph przed systemami wielu agentów
Pierwsze prace z LangGraph w Colab: stan współdzielony, węzły sekwencyjne, jeden krok LLM oraz pipeline bloga wieloetapowego, zanim przystąpiono do systemów wieloagentowych.
Źródła: Build AI Workflows with LangGraph (Part 1): Your First Agentic AI Application
Czym jest LangGraph?
Zwykłe skrypty w Pythonie działają od góry do dołu. Aplikacje agentowe poruszają się po grafie: każdy węzeł wykona jedno zadanie, a następnie przekazuje kontrolę dalej wzdłuż krawędzi. Minimalny przykład wygląda tak:
Input
↓
Node A
↓
Node B
↓
Node C
↓
Output
Każdy węzeł otrzymuje wspólny stan, aktualizuje go i przekazuje dalej. Ten wzorzec pozwala na rozbudowę do struktur z gałęziami, zarządzania pamięcią, planowania, refleksji oraz konfiguracji wielu agentów – ale pierwszą umiejętnością jest stworzenie prostego przepływu danych.
Poniższy przewodnik pozostaje powtarzalny: Google Colab, modele o otwartym kodzie dostępne przez Groq oraz krótka lista zależności. Nie wymaga lokalnej farmy GPU. Założeniem edukacyjnym jest to, że pomysły związane z agentami realizują się szybciej, gdy każda komórka działa na bezpłatnej infrastrukturze, dzięki czemu uwaga skupia się na stanie, węzłach i krawędziach, a nie na konfiguracji klastra.
Zainstaluj zależności
!pip install langchain langgraph langchain_groq
- LangGraph tworzy graf przepływu pracy
Uruchamianie instalacji wewnątrz Colab zapewnia, że środowisko pozostaje jednorazowe i udostępnialne.
Importowanie niezbędnych bibliotek
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
from IPython.display import Image
StateGraphkonstruuje schemat pracySTART/ENDoznaczają początek i koniec procesuTypedDictdefiniuje typy współdzielonego stanuImagewyświetla diagram złożonego grafu
Przykład 1 — Prosty sekwencyjny schemat pracy
Rozpocznij bez użycia LLM, aby ścieżka danych była oczywista.
Zdefiniuj stan
class BMIState(TypedDict):
weight_kg: float
height_m: float
bmi: float
category: str
Stan to współdzielona tabela, która towarzyszy procesowi. Węzły odczytują z niej dane, edytują pola i zwracają zaktualizowaną wersję. W tym przykładzie przechowywane są wartości masy, wzrostu, BMI oraz etykieta kategorii.
Stwórz węzeł przetwarzania
# 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
Funkcja odczytuje wzrost i wagę, oblicza BMI, przypisuje kategorię i zwraca nowy stan. Zasada działania LangGraph jest prosta: przyjmuje bieżący stan i zwraca następny.
Budowanie grafu
# 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())
Zarejestruj węzeł, połącz START → węzeł → END i skompiluj. LangGraph może narysować powstały graf, dzięki czemu można sprawdzić topologię przed jego uruchomieniem.
Wykonywanie procesu
# execute the graph
intial_state = {'weight_kg':80, 'height_m':1.73}
final_state = workflow.invoke(intial_state)
print(final_state)
Dostarcz początkowy słownik zawierający wagę i wzrost. Funkcja invoke uruchamia określony szlak i zwraca stan, który teraz zawiera wartość BMI oraz kategorię.
Przykład 2 — Proces sekwencyjny z użyciem LLM
Zamień węzeł matematyczny na wywołanie modelu. Zachowaj tę samą strukturę grafu.
Bezpiecznie pobierz klucz API Groq (na podstawie prośby użytkownika, a nie jako wartość hardkodowana), a następnie załaduj dostępny model otwarty:
model = ChatGroq(model="openai/gpt-oss-120b")
Definicja stanu LLM
# create a state
class LLMState(TypedDict):
question: str
answer: str
Pola skracają się do pytania i odpowiedzi – forma stanu odpowiada schematowi pracy, a nie globalnemu schematowi.
Stwórz węzeł 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
Przeczytaj pytanie, wezwij model do rozmowy, zapisz odpowiedź i zwróć stan. Strukturalnie identyczny jak węzeł BMI; zmienia się tylko sposób przetwarzania danych.
Zbuduj proces pracy 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()
Dodaj węzeł, połącz krawędziami i skompiluj. Ramy pozostają celowo znajome.
Zadaj pytanie LLM
# execute
intial_state = {'question': 'How far is sun from the earth?'}
final_state = workflow.invoke(intial_state)
print(final_state['answer'])
Wezwij model za pomocą ciągu znaków pytania i wyświetl pole z odpowiedzią. Model znajduje się w grafie jak każdy inny krok.
Przykład 3 — Wieloetapowy proces pracy AI
Rzeczywiste systemy łączą ze sobą kilka etapów rozumowania. Rozszerz ten wzorzec.
Zdefiniuj stan bloga
class BlogState(TypedDict):
title: str
outline: str
content: str
rating: int
Nadzoruj tytuł, strukturę, treść oraz ocenę jakości w miarę rozwoju procesu.
Stwórz węzły procesu pracy
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
Odrębne funkcje zajmują się tworzeniem szkicu oraz późniejszymi etapami. Każdy węzeł ma jedną konkretne rolę, dzięki czemu wykresy pozostają czytelne.
Zbuduj kompletny pipeline
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
Połącz zadania AI w odpowiedniej kolejności. Koncepcyjnie:
Title
↓
Create Outline
↓
Generate Blog
↓
Evaluate Blog
↓
END
Rozdzielenie pracy na poszczególne etapy jest lepsze niż umieszczanie wszystkich instrukcji w jednym mega-prompt: każdy krok wpływa na następny.
Zrealizuj proces pracy
intial_state = {'title': 'Rise of AI in India'}
final_state = workflow.invoke(intial_state)
print(final_state['rating'])
Począwszy od tytułu, proces tworzy szkic, przygotowuje treść, ocenia jej jakość i zwraca pole z oceną. To jest szkielet pipeline agentowych: koordynacja wielu możliwości modeli pod jednym wspólnym stanem.
Traktuj każdy przykład jako krok kalibracji, a nie jako kod demonstracyjny do szybkiego wyrzucenia. Po uruchomieniu wykresu BMI zmień progi kategorii i sprawdź aktualizacje stanu. Po uruchomieniu wykresu LLM z jednym pytaniem zamień identyfikator modelu lub szablon promptu i obserwuj tylko zmianę zachowania tego węzła. Po uruchomieniu procesu tworzenia bloga wstaw dodatkowy węzeł pomiędzy zarysem a szkicem – na przykład „sprawdzenie tonu” – aby zobaczyć, jak łącza i pola stanu rozwijają się razem.
Na tym etapie ważniejsza jest dyscyplina w zarządzaniu wspólnym stanem niż pomysłowe prompty. Woląć małe, nazwane pola (outline, content, rating) zamiast jednego ogromnego obiektu messages, dopóki nie będziesz potrzebował historii rozmów. Wracanie z częściową aktualizacją, która pomija nieskorelowane klucze, jest w porządku, jeśli semantyka łączenia wykresów na to pozwalają; ważnym nawykiem jest utrzymywanie węzłów na tyle czystych, by można je było testować za pomocą ręcznie stworzonego słownika.
Obsługa błędów w Colab może pozostać minimalna, ale nadal musi być celowa: jeśli wywołanie Groq się nie powiedzie, należy pokazać wyjątek zamiast pochłaniać go wewnątrz węzła. Wczesne tutoriale, które ukrywają błędy, wpajają niewłaściwe nawyki przy późniejszym tworzeniu grafów wywołań narzędzi, gdzie nieudane narzędzie musi stać się wyraźną gałęzią.
Na koniec rysuj graf po skompilowaniu za każdym razem, gdy dodajesz krawędź. Wizualna weryfikacja szybciej wykrywa przypadkowe cykle i zapomniane łącza END niż analiza wywołań funkcji add_edge. Gdy obraz odpowiada historii, którą chciałeś przekazać, funkcja invoke staje się krokiem weryfikacyjnym, a nie sesją debugowania.
Co pokazują te przykłady
Trzy stopniowo złożone schematy pracy obejmują podstawową terminologię:
- Czysta logika sekwencyjna w Pythonie działająca na wspólnym stanie
- Jeden węzeł LLM w ramach tego samego wzoru grafu
Kluczowe elementy to wspólny stan, węzły, krawędzie, orkiestracja, integracja z modelami LLM oraz stopniowe rozumowanie. Te same koncepcje później umożliwiają wykorzystanie narzędzi, pamięci, rozgałęzień oraz współpracujących agentów.
Co dalej
Późniejsze elementy systemu Agentic AI w Pythonie zazwyczaj obejmują warunkowe krawędzie, równoległe rozgałęzienia, mechanizmy zatwierdzania przez człowieka, wywoływanie narzędzi, technologię RAG, struktury wieloagentowe oraz pakietowanie do użycia w produkcji – wszystko przy zachowaniu możliwości pracy z Colab i otwartych modeli.
Pierwotny tekst
Towarzyszący notatnik Colab został opublikowany w ramach projektu Agentic-AI na GitHubie (plik Workflow_in_LangGraph.ipynb). Otwórz ten notatnik w Colab, uruchom komórki od góry do dołu, a następnie utwórz własne węzły do realizacji swoich zadań.
Powiązane materiały
- Twoje pierwsze narzędzie – agent AI z LangChain — Stwórz małego agenta matematycznego z LangChain w Pythonie: zainicjuj model do rozmów, zarejestruj narzędzia, pozwól LLM wybrać funkcję do wywołania i zobacz wieloetapowe używanie narzędzi.
- Orkiestracja agentów badacza i programisty z LangGraph — Stwórz aplikację wieloagentową z LangGraph wykorzystującą prompty typu soul, modele Ollama, wyszukiwanie w Tavily, narzędzia do przekazywania zadań oraz punkty kontrolne Postgres dla prac prowadzonych przez orkiestratora.