Триаж заявок о поддержке с участием нескольких агентов с использованием LangGraph и Ollama

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

LangGraph помогает создать такую команду. Один агент классифицирует каждый запрос, маршрутизатор определяет дальнейшие действия, а второй агент составляет ответ. Срочные случаи обходят автоматизацию и передаются человеку. Граф строится один раз и работает либо локально на Ollama, либо через облачный API, без изменений в агентах или механизмах маршрутизации.

Почему нужно более одного агента

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

Метод сортировки заявок соответствует этой идее, поскольку работа естественным образом делятся на части, а конвейер, обрабатывающий все сообщения одинаково, будет неправильным по крайней мере в одном из следующих случаев:

• Рутинный вопрос о счетах можно ответить автоматически.
• Неопределенная просьба о новой функции получает предварительный ответ, который позже дорабатывается человеком.

Сообщение вроде «мое приложение не работает» совершенно не должно обрабатываться автоматически и должно немедленно попасть к человеку.

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

1. Агент-классификатор, который читает заявку и присваивает ей категорию и приоритет.
2. Агент-ответчик, который составляет ответы на заявки, которые можно обработать автоматически.
3. Человеческий контроль — узел, который передает срочные заявки человеку вместо того, чтобы отвечать сам.

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

Как работает LangGraph

Представьте схему на доске. LangGraph состоит из четырех частей:

• State — это буфер обмена данными, передаваемый от одной станции к другой. В нем хранится всё, что известно на данный момент о билете: исходный текст, категория, приоритет, возможный проект ответа и статус.
• Node — это станция, функция, которая берёт состояние, выполняет задачу и возвращает обновления.
• Edge — это фиксированная стрелка от одной станции к другой.
• Conditional edge — это стрелка, выбираемая во время выполнения. Небольшая функция-маршрутизатор читает состояние и решает, куда двигаться, вместо того чтобы всегда направляться в одно и то же место.

Таким образом, состояние передаётся через узлы, стрелки их соединяют, а условные стрелки позволяют пути разветвляться в зависимости от текущего состояния. Если вы только начинаете знакомство с этим, состояние — это просто словарь на Python с определённой структурой. TypedDict описывает, какие ключи имеет этот словарь.

Технологическая стек

LangGraph является слоем оркестрации. Он определяет узлы, состояние и маршрутизацию между агентами.

Ollama с Llama 3 запускает всё локально, бесплатно, на вашем собственном компьютере. Заявки на поддержку часто содержат личные данные клиентов, поэтому в деловой среде имеется реальное преимущество в хранении этих данных на собственной инфраструктуре вместо отправки их на сторонний API.

Модель API, здесь gpt-4o-mini, является альтернативой, когда у вас нет дополнительного оборудования или требуется более высокая производительность, чем обеспечивает небольшая локальная модель. Поскольку граф строится один раз, вы можете начать с локальной реализации и позже перейти на API без переписывания логики.

Построение пошагово

1. Настройка

Установите три пакета: langgraph для работы с графами, langchain-ollama в качестве моста к Ollama и langchain-openai в качестве моста к API-модели.

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

Чистая виртуальная среда сохраняет пакеты этого проекта отдельно от ваших других проектов. В Windows её можно создать с помощью команды python -m venv yourname, а активировать — через папку Scripts, находящуюся внутри. На Mac или Linux используйте команду python3 -m venv venv, затем source venv/bin/activate. После активации команда установки будет влиять только на этот проект.

Если Ollama ещё не установлен, скачайте его с официального сайта ollama.com/download. Затем загрузите локальную модель из терминала:

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

2. Агенты

Далее всё будет сохраняться в одном файле с именем triage.py, добавляясь сверху вниз в следующем порядке.

Фабрика модели. Начните с элемента, позволяющего использовать оба варианта бэкенда — небольшой функции-фабрики get_llm, которая возвращает соответствующую модель на основе одного параметра. Единственная задача фабрики — вернуть готовый объект. Представьте кофемашину с кнопкой «локально» и другой с кнопкой «API»: нажмите на одну из них, получите соответствующую чашку, не узнавая, что происходит внутри.

Функция принимает параметр backend со значением по умолчанию "local", поэтому вызов без аргументов использует локальную модель.

# 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)

Цепочка операторов if/elif/else проверяет параметр и возвращает либо llama3, либо gpt-4o-mini, либо сообщение об ошибке. Импорты намеренно размещены внутри отдельных веток. Тем, кто работает исключительно локально, никогда не понадобится библиотека langchain-openai, и наоборот.

Параметр temperature=0 контролирует степень случайности ответов модели. Большие значения придают ответам разнообразие, тогда как значение 0 делает их максимально стабильными и предсказуемыми. Для классификации это идеально, поскольку один и тот же запрос должен надежно попадать в одну и ту же категорию. Значение BACKEND = "local" определяет способ обработки, а следующая строка вызывает соответствующий механизм.

Классификатор. Этот агент читает запрос и возвращает категорию с приоритетом. Чтобы обработать результат программно, от модели требуется предоставлять ответы в формате JSON.

Вспомогательная функция parse_json преобразует ответ в настоящий словарь Python. Она использует функцию find, чтобы сначала найти начальные и заключительные фигурные скобки, поскольку локальные модели не всегда возвращают чистый JSON. Они обычно окружают его лишним текстом вроде «Конечно, вот JSON: ... Надеюсь, это поможет».

Функция classify_ticket представляет собой сам агент. Она получает текущее состояние и возвращает обновления. В промпте четко определена роль модели как ассистента для сортировки заявок, ответ ограничивается четырьмя категориями и тремя приоритетами, а также требуется чистый JSON без дополнительных комментариев. Промпт составлен особенно строго, чтобы ответ был более структурированным и функции parse_json приходилось делать меньше работы. Вызов llm.invoke(prompt) отправляет его в модель.

Обратите внимание, что возвращается: только два поля, которые заполняет этот агент — категория и приоритет. Это основная особенность LangGraph. Узел не возвращает всё состояние, а только те его части, которые он хочет изменить, и LangGraph объединяет их в общее состояние. Это позволяет сохранять узлы небольшими, облегчать тестирование их по отдельности, а также предотвращает случайное перезапись одним агентом поля, принадлежащего другому.

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"]}

Ответчик. Второй агент составляет ответы на заявки, которые можно автоматизировать. Функция draft_reply получает информацию о состоянии заявки, формирует промпт, в котором модель представлена как дружелюбный и краткий сотрудник службы поддержки, и вводит ограничения: модель не может давать обещаний, которые не может выполнить. Она возвращает два поля — составленный текст в draft_reply и статус «auto-drafted», что позволяет позже увидеть, что заявка была обработана автоматически и не передана на более высокий уровень. Промпт составлен сознательно скромно. Ответ службы поддержки, написанный моделью, должен быть кратким и честным, а обещания относительно сроков возврата средств или даты выхода новой функции — это то, что должен решать человек.

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"}

Человеческий оператор. Третий агент никогда не обращается к модели. Всё, что он делает, — это помечает заявку как переданную на рассмотрение человека. Он возвращает пустой объект draft_reply, поскольку автоматического ответа не должно существовать, а также статус «передано на рассмотрение человека». Это ясно указывает на то, что заявкой занимается человек. Узел не обязан использовать LLM, чтобы считаться агентом в графе. Любой элемент, который берет на себя обработку заявки, выполняет задачу и возвращает обновления, может находиться в этом же графе, независимо от того, обращается ли он к модели, базе данных или вообще ни к чему.

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. Подключение графа

У каждого из трех агентов есть своя задача. Теперь их необходимо соединить, и это начинается с описания формы общего состояния. Класс TicketState является аналогом буфера обмена данных, описанным ранее; в нем указываются существующие поля и их типы.

Эти пять полей также отслеживают путь заявки внутри системы. Поле ticket хранит исходный текст. Классификатор заполняет поля category и priority. Ответчик заполняет поле draft_reply, или оставляет его пустым после передачи заявки на более высокий уровень обработки. Поле status фиксирует, где в конечном итоге оказалась заявка.

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

Далее идет маршрутизатор, то есть системный администратор. Функция route_by_priority считывает приоритет, установленный классификатором, и возвращает строку с указанием пути, который должен выбрать LangGraph. Если приоритет равен «высокий», заявка направляется к человеку-оператору. В противном случае функция возвращает значение «responder», и заявке присылается автоматический ответ.

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"

Теперь соберите граф:

1. Создайте пустой граф с помощью StateGraph(TicketState), передав форму состояния, чтобы граф знал, какие поля находятся в буфере обмена.
2. Зарегистрируйте три агента с помощью add_node. Каждый вызов связывает краткое имя с функцией. Имя «classifier» относится к classify_ticket, «responder» — к draft_reply, а «human_review» — к escalate_to_human. Во всех последующих шагах к узлам обращаются по этим именам, поэтому выбирайте их тщательно и соблюдайте последовательность.
3. Добавьте строку add_edge(START, „classifier“), чтобы каждый тикет начинался с классификатора. START — это встроенная точка входа.

Добавьте условную грань с помощью функции add_conditional_edges; она принимает узел, в котором происходит разветвление ("classifier"), функцию маршрутизации (route_by_priority) и таблицу поиска.

Отправьте оба пути в точку END, поскольку процесс завершается независимо от того, был ли запрос ответлен или передан на более высокий уровень обработки.

Вызовите функцию graph.compile(), которая преобразует определение в исполняемый код. Результат сохраняется в объекте app, и именно он обрабатывает запросы.

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()

Словарь, передаваемый в функцию add_conditional_edges, представляет собой таблицу поиска. Её ключи — это строки, которые может вернуть маршрутизатор, а значения — узлы, на которые указывают эти строки. Когда маршрутизатор возвращает значение "human_review", LangGraph переходит к узлу human_review. Именно такое соответствие является основой формирования ветвей в графе. Чтобы позже добавить новую категорию обработки, например специалиста по счётам, достаточно добавить один новый узел, один новый ключ в таблицу и соответствующую ветвь в маршрутизаторе.

4. Запуск: локально или через API

Сначала отправьте срочный запрос через граф:

# 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"])

Запустите скрипт с помощью python triage.py. Ожидайте, что точные формулировки выводимого ответа немного будут отличаться при разных запусках и у разных моделей, но категория, приоритет и статус должны оставаться неизменными — вот почему используется параметр temperature=0. Заявка имеет высокий приоритет, поэтому маршрутизатор направляет её на рассмотрение человека. Статус возвращается как «передано на рассмотрение человеком», и автоматический ответ не генерируется.

Попробуйте вторую, обычную заявку:

# 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"])

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

Переход на API-модель

Для выполнения той же операции через API вместо локальной обработки требуется всего один API-ключ и одна изменённая строка благодаря функции-фабрике.

Войдите на платформу API OpenAI, создайте новый ключ и сразу же скопируйте его, поскольку он отображается только один раз. В целях безопасности не вставляйте ключ в скрипт. Храните его как переменную окружения и пусть код читает её оттуда. Запустите команду для вашей системы в терминале, заменив your-key-here на ваш ключ:

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

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

Выполняйте их в терминале, а не в скрипте, и желательно в том же терминале, где активна виртуальная среда. Переменная задаётся на уровне сессии терминала, а не внутри venv. После этого ключ хранится в одном месте вне вашего кода, никогда не появляется в скрипте и не может случайно попасть на GitHub.

Затем измените одну строку в скрипте и снова запустите python triage.py:

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

Что это показывает и куда двигаться дальше

Пример отражает реальный ежедневный рабочий процесс. Цель не в замене команды поддержки, а в автоматизации механической сортировки запросов, пока важные и срочные случаи передаются человеку. Поскольку граф, агенты и механизм направления работают как с Ollama, так и с API, можно создать прототип с использованием Ollama, а затем перейти на модель API при необходимости в большей производительности.

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

Для дальнейшего изучения документация LangGraph и LangChain по адресу https://docs.langchain.com/oss/python/langgraph/overview, а также руководство по быстрому старту Ollama по адресу https://docs.ollama.com/quickstart описывают инструменты, используемые здесь.