Triaż zgłoszeń wsparcia wieloagentowego przy użyciu LangGraph i Ollama

Każdy, kto zajmował się obsługą skrzynki pocztowej z zapytaniami klientów, zna tę poranną rutynę. Ktoś czyta każdą nową wiadomość, sortuje prośby o sformatowanie haseł oraz pytania dotyczące fakturowania, a także szuka tych nielicznych wiadomości, które są naprawdę pilne. Jest to czysto mechaniczny proces, który doskonale nadaje się do realizacji przez mały zespół współpracujących agentów.

Następnie LangGraph pomaga stworzyć taki zespół. Jeden agent klasyfikuje każdą wiadomość, router decyduje, co robić dalej, a drugi agent przygotowuje odpowiedź. Przypadki pilne omijają automatyzację i trafiają bezpośrednio do osoby. Graf jest tworzony raz i działa albo lokalnie na Ollama, albo poprzez API chmurowe, bez konieczności zmian w agentach czy mechanizmach routingu.

Dlaczego potrzebny jest więcej niż jeden agent

Jedyny agent przypomina pracownika, który nigdy z nikim nie rozmawia, a w prawdziwej firmie takie podejście szybko hamuje rozwój. Gdy dodaje się więcej zadań, pojedynczy polecenie staje się długie, kruche i trudne do debugowania, ponieważ jeden model musi jednocześnie klasyfikować, decydować, pisać i oceniać ryzyko. Podział pracy pozwala, by każda część pozostała mała, poddawalna testom i wymienna. System wielu agentów przypomina zespół specjalistów, z których każdy ma określoną rolę i koordynuje kolejne kroki.

Tryaż zleceń pasuje do tej koncepcji, ponieważ praca naturalnie się dzieli, a pipeline traktujący każdą wiadomość w ten sam sposób byłby błędny przynajmniej w jednym z poniższych przypadków:

• Rutynowe pytanie dotyczące fakturacji można odpowiedzieć automatycznie.
• Niejasna prośba o nową funkcję otrzymuje wstępną odpowiedź, którą później udoskonala człowiek.

Wiadomość typu „moja aplikacja nie działa” w ogóle nie powinna być automatyzowana i musi natychmiast trafić do człowieka.

LangGraph obsługuje dokładnie takie sytuacje, gdy kolejny krok zależy od aktualnego stanu. Zespół składa się z trzech osób:

1. Agent klasyfikator, który czyta zgłoszenie i przypisuje mu kategorię oraz priorytet.
2. Agent odpowiadający, który przygotowuje odpowiedzi na zgłoszenia, które można bezpiecznie przetworzyć automatycznie.
3. Człowiek jako brama, węzeł, który przekazuje pilne zgłoszenia osobie zamiast samodzielnie odpowiadać.

Pomiędzy nimi znajduje się logika routingu, która pełni rolę nadzorcy. Analizuje priorytet i wybiera odpowiednią ścieżkę.

Jak myśli LangGraph

Wyobraź sobie schemat przepływów na tablicy. LangGraph składa się z czterech części:

• Stan to pamięć przenoszona z jednej stacji na drugą. Zawiera wszystko, co dotychczas wiadomo o bilecie: oryginalny tekst, kategorię, priorytet, ewentualną zaprojektowaną odpowiedź oraz status.
• Węzeł to stacja, funkcja, która bierze stan, wykonuje zadanie i zwraca aktualizacje.
• Krawędź to stała strzała od jednej stacji do drugiej.
• Krawędź warunkowa to strzała wybierana w czasie wykonywania programu. Mała funkcja routera odczytuje stan i decyduje, dokąd iść, zamiast zawsze kierować się w to samo miejsce.

Stan przepływa przez węzły, krawędzie łączą je ze sobą, a krawędzie warunkowe umożliwiają rozgałęzianie ścieżki w zależności od aktualnego stanu. Dla początkujących stan to po prostu słownik w Pythonie o określonej strukturze. TypedDict określa, jakie klucze posiada ten słownik.

Stack technologiczny

LangGraph stanowi warstwę orkiestracji. Określa węzły, stan oraz ścieżki komunikacyjne pomiędzy agentami.

Ollama z Llama 3 uruchamia wszystko lokalnie, bezpłatnie, na Twoim własnym komputerze. Zgłoszenia obsługi klienta często zawierają osobiste dane klientów, dlatego w środowisku biznesowym istnieje prawdziwa korzyść z przechowywania tych danych na własnej infrastrukturze zamiast wysyłania ich do zewnętrznego API.

Model API, tutaj gpt-4o-mini, jest alternatywą, gdy nie masz wolnego sprzętu lub potrzebujesz wyższej wydajności niż zapewnia mały lokalny model. Ponieważ graf jest tworzony tylko raz, możesz zacząć od wersji lokalnej, a później przenieść się na API bez konieczności przepisywania logiki.

Budowanie krok po kroku

1. Konfiguracja

Zainstaluj trzy pakiety: langgraph do obsługi grafów, langchain-ollama jako most do Ollamy oraz langchain-openai jako most do modelu API.

# Install LangGraph and the two model integrations
pip install langgraph langchain-ollama langchain-openai

Czyste środowisko wirtualne chroni pakiety tego projektu przed innymi. W systemie Windows utwórz je za pomocą python -m venv twojaNazwa, a następnie aktywuj je poprzez folder Scripts znajdujący się w jego wnętrzu. W systemach Mac lub Linux użyj python3 -m venv venv, a potem wywołaj source venv/bin/activate. Gdy zostanie ono aktywne, powyższy polecenie instalacji będzie dotyczyć wyłącznie tego projektu.

Jeśli Ollama jeszcze nie jest zainstalowane, pobierz je ze strony ollama.com/download. Następnie ściągnij lokalny model z terminala:

# Download the Llama 3 model so we can run it locally
ollama pull llama3

2. Agenci

Od tego momentu wszystko trafia do jednego pliku o nazwie triage.py, dodawanego od góry do dołu w poniższej kolejności.

Fabryka modelu. Zacznij od elementu, który umożliwia działanie obu backendów – małej funkcji-fabryki get_llm, która zwraca odpowiedni model na podstawie jednego ustawienia. Jedynym zadaniem fabryki jest zwrócenie gotowego obiektu. Wyobraź sobie maszynkę do kawy z przyciskiem oznaczonym jako „local” i innym oznaczonym jako „API”: naciśnij jeden z nich, aby otrzymać odpowiedni kubek, nie musząc wiedzieć, co dzieje się w środku.

Funkcja przyjmuje parametr backend, którego domyślna wartość to „local”, więc wywołanie jej bez argumentów używa lokalnego modelu.

# A factory that returns the model we want, so the rest of the code
# never has to care which backend we are using.
def get_llm(backend: str = "local"):
    if backend == "local":
        from langchain_ollama import ChatOllama
        # temperature=0 keeps classification stable and repeatable
        return ChatOllama(model="llama3", temperature=0)
    elif backend == "api":
        from langchain_openai import ChatOpenAI
        return ChatOpenAI(model="gpt-4o-mini", temperature=0)
    else:
        raise ValueError("backend must be 'local' or 'api'")


# Choose your backend here: "local" or "api"
BACKEND = "local"
llm = get_llm(BACKEND)

Łańcuch if/elif/else sprawdza parametr i zwraca albo llama3, albo gpt-4o-mini, albo komunikat o błędzie. Importy znajdują się celowo wewnątrz poszczególnych gałęzi. Osoba pracująca wyłącznie lokalnie nigdy nie potrzebuje langchain-openai, i odwrotnie.

temperature=0 kontroluje stopień losowości odpowiedzi modelu. Wyższe wartości dodają różnorodności, natomiast 0 sprawia, że odpowiedzi są jak najbardziej stabilne i przewidywalne. To idealne rozwiązanie do klasyfikacji, ponieważ ten sam zgłoszenie powinno niezawodnie trafiać do tej samej kategorii. BACKEND = „local” określa wybór, a następna linia wywołuje odpowiedni mechanizm.

Klasyfikator. Ten agent czyta zgłoszenie i zwraca kategorię oraz priorytet. Aby móc przetworzyć wynik w sposób programowy, od modelu wymaga się odpowiedzi w formacie JSON.

Pomocnik parse_json przekształca odpowiedź w prawdziwy słownik Pythona. Wykorzystuje funkcję find, aby najpierw zlokalizować otwierające i zamykające nawiasy, ponieważ lokalne modele nie zawsze zwracają czysty JSON. Często otaczają go dodatkowymi informacjami, takimi jak „Oczywiście, oto JSON: ... Mam nadzieję, że to pomoże.”

Funkcja classify_ticket to sam agent. Otrzymuje aktualny stan i zwraca aktualizacje. Zadane jej instrukcje określają jasną rolę asystenta do triażu zgłoszeń, ograniczają odpowiedź do czterech kategorii i trzech priorytetów oraz wymagają czystego JSON bez dodatkowych komentarzy. Instrukcje są celowo ścisłe, aby odpowiedź była bardziej uporządkowana i parse_json miał mniej pracy. Wywołanie llm.invoke(prompt) wysyła je do modelu.

Zwróć uwagę na to, co jest zwracane: tylko dwa pola, które wypełnia ten agent – kategoria i priorytet. To kluczowa cecha LangGraph. Węzeł nie zwraca całego stanu, lecz tylko tych części, które chce zmienić, a LangGraph łączy je w wspólny stan. Dzięki temu węzły pozostają małe, łatwe do testowania osobno, a także zapobiega to sytuacji, w której jeden agent przypadkowo nadpisze pole należące do innego.

import json

def parse_json(text: str) -> dict:
    # Local models sometimes wrap their JSON in extra text,
    # so we grab everything between the first { and the last }.
    start = text.find("{")
    end = text.rfind("}") + 1
    return json.loads(text[start:end])

def classify_ticket(state: dict) -> dict:
    prompt = f"""You are a support triage assistant.
Classify the following support ticket.
Respond with ONLY a JSON object with two keys:
"category": one of ["billing", "technical", "account", "general"]
"priority": one of ["low", "medium", "high"]

Ticket:
{state['ticket']}"""

    response = llm.invoke(prompt)
    result = parse_json(response.content)

    # We return only the keys we want to update in the shared state
    return {"category": result["category"], "priority": result["priority"]}

Odpowiadający. Drugi agent przygotowuje odpowiedzi do zgłoszeń, które można bezpiecznie zautomatyzować. Funkcja draft_reply otrzymuje aktualny stan zgłoszenia, tworzy instrukcję skierowaną do modelu w roli przyjaznego i zwięzłego agenta obsługi klienta, a także wprowadza ograniczenia – model nie może składać obietnic, których nie może dotrzymać. Funkcja ta zwraca dwa pola: tekst przygotowany w draft_reply oraz status „auto-drafted”, dzięki czemu później można sprawdzić, że zgłoszenie zostało obsłużone automatycznie i nie trafiło do wyższej instancji. Instrukcja skierowana do modelu jest celowo umiarkowana. Odpowiedź obsługi klienta napisana przez model powinna być krótka i szczera, a decyzje dotyczące terminu zwrotu pieniędzy lub daty wprowadzenia nowej funkcji powinny podejmować osoby.

def draft_reply(state: dict) -> dict:
    prompt = f"""You are a friendly and concise support agent.
Write a short, helpful reply to this {state['category']} ticket.
Do not make promises you cannot keep.

Ticket:
{state['ticket']}"""

    response = llm.invoke(prompt)

    return {"draft_reply": response.content, "status": "auto-drafted"}

Ludzki bramkarz. Trzeci agent nigdy nie wywołuje modelu. Jego jedynym zadaniem jest oznaczenie zgłoszenia jako eskalowanego, aby mogła je obsłużyć osoba. Zwraca pusty draft_reply, ponieważ nie powinna istnieć żadna automatyczna odpowiedź, oraz status „eskalowane do ludzkiej oceny”. Dzięki temu jasno widać w stanie, że zgłoszenie znajduje się w rękach człowieka. Węzeł nie musi korzystać z LLM, aby być uznawany za agenta w grafie. Wszystko, co przyjmuje stan, wykona zadanie i zwraca aktualizacje, może znajdować się na tym samym obrazie, niezależnie od tego, czy wywołuje model, bazę danych, czy w ogóle nic.

def escalate_to_human(state: dict) -> dict:
    # No auto-reply for urgent tickets. We flag them for a person.
    return {
        "draft_reply": "",
        "status": "escalated to human review",
    }

3. Podłączanie grafu

Każdy z trzech agentów ma swoje zadanie. Teraz muszą zostać połączone, a to zaczyna się od opisania kształtu wspólnego stanu. Klasa TicketState to ta sama kopia z wcześniejszego etapu, która określa, jakie pola istnieją i jaki jest ich typ.

Tych pięć pól odzwierciedla również ścieżkę zgłoszenia w systemie. ticket przechowuje oryginalny tekst. Klasyfikator wypełnia pola category i priority. Osoba odpowiadająca wypełnia pole draft_reply, albo pozostawia je puste po eskalacji. status rejestruje, gdzie zgłoszenie ostatecznie się znalazło.

from typing import TypedDict


class TicketState(TypedDict):
    ticket: str        # the original ticket text
    category: str      # filled in by the classifier
    priority: str      # filled in by the classifier
    draft_reply: str   # filled in by the responder (or cleared on escalation)
    status: str        # where the ticket ended up

Następnie pojawia się router, czyli nadzorca. Funkcja route_by_priority odczytuje priorytet przypisany przez klasyfikator i zwraca ciąg znaków wskazujący LangGraph, którą ścieżkę obrać. Jeśli priorytet wynosi „wysoki”, zgłoszenie trafia do człowieka. W przeciwnym razie funkcja zwraca „responder”, a zgłoszenie otrzymuje automatyczną odpowiedź.

def route_by_priority(state: TicketState) -> str:
    # Urgent tickets skip the automatic reply and go to a human.
    if state["priority"] == "high":
        return "human_review"
    return "responder"

Teraz skonstruuj graf:

1. Stwórz pusty graf za pomocą StateGraph(TicketState), przekazując kształt stanu, aby graf wiedział, które pola znajdują się na skrawku papieru.
2. Zarejestruj trzy agenty za pomocą add_node. Każde wezwanie łączy krótką nazwę z funkcją. Nazwa „classifier” należy do classify_ticket, „responder” do draft_reply, a „human_review” do escalate_to_human. Każdy kolejny krok odnosi się do węzłów pod tymi nazwami, więc wybieraj je starannie i zachowuj spójność.
3. Dodaj add_edge(START, „classifier”), aby każdy ticket rozpoczynał się od klasyfikatora. START to wbudowany punkt wejścia.

Dodaj warunkową krawędź za pomocą add_conditional_edges – funkcja ta przyjmuje węzeł, w którym następuje rozgałęzienie ("classifier"), funkcję routingu (route_by_priority) oraz tabelę wyszukiwania.

Wyślij obie ścieżki do END, ponieważ proces jest zakończony niezależnie od tego, czy zgłoszenie zostało rozwiązane, czy przeniesione na wyższy poziom.

Wezwij graph.compile(), która przekształca definicję w coś wykonalnego. Wynik jest przechowywany w app, a to właśnie ono obsługuje zgłoszenia.

from langgraph.graph import StateGraph, START, END

# Create the graph and tell it the shape of our state
graph = StateGraph(TicketState)

# Register each agent as a node
graph.add_node("classifier", classify_ticket)
graph.add_node("responder", draft_reply)
graph.add_node("human_review", escalate_to_human)

# Every ticket starts at the classifier
graph.add_edge(START, "classifier")

# After classifying, the router decides where to go next
graph.add_conditional_edges(
    "classifier",
    route_by_priority,
    {
        "responder": "responder",       # normal tickets get an auto-draft
        "human_review": "human_review",  # urgent tickets get escalated
    },
)

# Both paths finish the run
graph.add_edge("responder", END)
graph.add_edge("human_review", END)

# Compile turns our definition into something runnable
app = graph.compile()

Słownik przekazywany do funkcji add_conditional_edges stanowi tabelę wyszukiwania. Jej klucze to ciągi znaków, które router może zwrócić, a wartości to węzły, do których te ciągi odnoszą się. Gdy router zwraca wartość „human_review”, LangGraph przechodzi do węzła human_review. To właśnie ta mapa stanowi istotę sposobu, w jaki graf się rozgałęzia. Później można dodać nową kategorię obsługi, na przykład specjalistę od fakturowania – zmiana polega wtedy na dodaniu jednego nowego węzła, jednego nowego klucza w tabeli oraz nowej gałęzi w routerze.

4. Uruchamianie, lokalnie lub przez API

Najpierw należy wysłać pilną zgłoszenie przez graf:

# A clearly urgent billing issue
urgent_ticket = (
    "I was charged twice for my subscription this month "
    "and I need a refund today. This is unacceptable."
)

result = app.invoke({"ticket": urgent_ticket})

print("Category:", result["category"])
print("Priority:", result["priority"])
print("Status:  ", result["status"])
print("Reply:   ", result["draft_reply"])

Zrób test skryptu za pomocą python triage.py. Oczekuj, że dokładne sformułowanie wyświetlanego odpowiedzi może nieco się różnić pomiędzy poszczególnymi uruchomieniami oraz pomiędzy różnymi modelami, ale kategoria, priorytet i status powinny być spójne – to właśnie jest powód użycia temperature=0. Ticket ma wysoki priorytet, więc router kieruje go do ludzkiego pracownika. Status powraca jako „escalated to human review”, a żadna automatyczna odpowiedź nie jest generowana.

Spróbuj drugiego, rutynowego ticketa:

# A routine, low-stakes question
routine_ticket = "How do I change the email address on my account?"

result = app.invoke({"ticket": routine_ticket})

print("Category:", result["category"])
print("Priority:", result["priority"])
print("Status:  ", result["status"])
print("Reply:   ", result["draft_reply"])

To rozwiązanie kieruje wiadomość do osoby odpowiadającej i zwraca wstępnie przygotowaną odpowiedź, którą osoba fizyczna może przejrzeć przed wysłaniem. Uruchamianie obu zadań równolegle ułatwia zrozumienie różnic: identyczny graf, identyczny kod, dwa różne ścieżki wybrane wyłącznie na podstawie priorytetu przypisanego przez klasyfikator.

Przejście na model API

Uruchamianie tej samej funkcji za pośrednictwem API zamiast lokalnie wymaga tylko jednego klucza API oraz jednej zmienionej linii kodu, dzięki funkcji fabrycznej.

Zaloguj się na platformę API OpenAI, utwórz nowy klucz i natychmiast go skopiuj, ponieważ jest pokazywany tylko raz. Dla bezpieczeństwa nie wklejaj klucza do skryptu – przechowuj go jako zmienną środowiskową i pozwól kodowi odczytywać go stamtąd. Uruchom w terminalu odpowiednią komendę dla twojego systemu, zastępując twoj-klucz-tutaj swoim kluczem:

# On Windows (PowerShell)
setx OPENAI_API_KEY "your-key-here"

# On Mac/Linux
export OPENAI_API_KEY="your-key-here"

Wykonaj je w terminalu, a nie w skrypcie, najlepiej w tym samym terminalu, w którym aktywne jest środowisko wirtualne. Zmienna jest ustawiana na poziomie sesji terminala, a nie wewnątrz venv. Po tym klucz znajduje się w jednym miejscu poza twoim kodem, nigdy nie pojawia się w skrypcie i nie może przypadkowo trafić na GitHuba.

Następnie zmień jedną linię w skrypcie i ponownie uruchom python triage.py:

# Switch from local to API without touching the graph or the agents
BACKEND = "api"
llm = get_llm(BACKEND)

Co pokazuje i dokąd dalej

Przykład odzwierciedla rzeczywisty codzienny proces pracy. Celem nie jest zastąpienie zespołu wsparcia, lecz przejęcie rutynowej sortacji, podczas gdy ważne i pilne przypadki trafiają do konkretnego pracownika. Ponieważ wykres, agenci oraz mechanizm kierowania działają zarówno z Ollamą, jak i z API, można stworzyć prototyp przy użyciu Ollamy, a następnie przejść na model oparty na API, gdy potrzebna będzie lepsza wydajność.

Istnieje wiele możliwości rozszerzenia tego systemu. Agenci mogliby mieć dostęp do większej ilości pamięci, dzięki czemu kolejne zgłoszenia od tego samego klienta zawierałyby odpowiedni kontekst. Mogliby dołączyć również inni specjaliści, np. pracownik ds. fakturowania z dostępem do rzeczywistych danych konta. Wykres mógłby łączyć się z rzeczywistym systemem obsługi zgłoszeń za pośrednictwem API, czyniąc skrypt częścią prawdziwego procesu przetwarzania. Ponadto możliwe byłoby włączenie kroku z udziałem człowieka, który umieściłby zgłoszenia o wyższym priorytecie w kolejce do przeglądu z już przydzieloną kategoryzacją.

Dla dalszej lektury dokumentacja LangGraph i LangChain dostępna pod adresem https://docs.langchain.com/oss/python/langgraph/overview oraz szybki start z Ollamą pod adresem https://docs.ollama.com/quickstart opisują narzędzia użyte w tym przykładzie.