LangChain і LangGraph — агенти, графи і вся інфраструктура навколо них

Привіт!

Мовна модель уміє рівно одне: отримати текст і повернути текст. Більше нічого. Вона не пам’ятає, про що ви говорили хвилину тому, і не може сама сходити до бази даних чи в інтернет.

Візьмімо простий приклад: користувач питає в асистента, яка зараз погода. Модель цього не знає. Щоб вона змогла відповісти, треба зробити ось що:

1. Заздалегідь розповісти моделі, що в неї є функція «дізнатися погоду», і описати її аргументи.
2. Отримати відповідь, у якій модель просить цю функцію викликати, і розібрати її.
3. Викликати функцію своїм кодом.
4. Надіслати моделі результат і попросити відповісти ще раз.

Вийшов цикл, а не один запит. Додайте до нього пам’ять діалогу, щоб наступне питання розумілося в контексті попереднього; вивід у міру генерації, щоб користувач не дивився сорок секунд у порожній екран; повтор при збої провайдера; запис того, що модель бачила, — інакше неправильну відповідь неможливо розібрати. Коду навколо простої дії набирається чимало.

LangChain і LangGraph — бібліотеки, у яких усе це вже написано. Перша дає готові частини: моделі, tools, промпти, цикл агента. Друга виконує процес: зберігає state, розгалужує кроки, зберігає прогрес, віддає потік подій.

У статті — як цим користуватися: що за що відповідає, з чого збирається agent, коли замість агента потрібен graph, як працюють пам’ять і streaming, навіщо потрібні LangSmith і готовий сервер запуску та що змінилося на переході з попередньої мажорної версії.

Приклади мовою Python. Версії пакетів на момент написання: langchain 1.3.14, langchain-core 1.5.3, langgraph 1.2.10, langchain-classic 1.0.8. Мінімальний Python — 3.10, для JavaScript-версії — Node 20.

Словник термінів

Терміни далі йдуть в оригінальному написанні — так само, як у документації та в коді, щоб одне легко зіставлялося з іншим. Ось короткі визначення; кожне детально розбирається у своєму розділі.

Model — мовна модель провайдера: OpenAI, Anthropic, Google, локальна через Ollama. У коді це об’єкт із методами виклику та потокового виклику.

Message — одиниця діалогу. Чотири типи: SystemMessage (інструкції), HumanMessage (введення користувача), AIMessage (відповідь моделі разом із запитами на виклики), ToolMessage (результат виконання функції).

Tool — функція вашого коду, опис якої віддається моделі. Модель її не виконує: вона повертає запит «виклич ось це з такими аргументами», а виконує ваш код.

Agent — цикл: викликати model, виконати tools, які вона запросила, повернути їй результати, повторити. Цикл закінчується, коли модель відповідає без запиту tools.

Graph — опис процесу кроками та переходами. Потрібен, коли порядок дій визначаєте ви, а не модель.

Node — один крок графа, звичайна функція. Edge — перехід між кроками.

State — дані, які nodes читають і доповнюють у процесі роботи.

Reducer — правило, за яким оновлення від node зливається з тим, що вже лежить у state.

Super-step — один такт виконання графа: виконується все заплановане, потім оновлення потрапляють у state, потім планується наступний такт.

Checkpointer — механізм збереження state. Завдяки йому другий хід діалогу пам’ятає перший, а процес, що впав, можна продовжити з місця зупинки.

Thread — ідентифікатор діалогу, під яким зберігається state. Один користувач, одна розмова — один thread.

Store — довготривала пам’ять поверх усіх threads: уподобання, накопичені факти.

Context — незмінні дані запуску: хто користувач, який орендар, які прапорці.

Runtime — об’єкт, через який node або tool отримують доступ до context, store і потоку подій.

Middleware — код, що виконується навколо кроків циклу агента: до виклику моделі, після, навколо виклику tool.

Interrupt — пауза запуску, щоб дочекатися людини.

Streaming — потокова видача: токени та події надходять у міру появи.

З чого складається екосистема

Під одним брендом випускається кілька продуктів зі схожими назвами. Розкладімо по полицях.

Шари залежать знизу вгору:

Функція create_agent з LangChain повертає скомпільований graph LangGraph — тобто все, що рантайм уміє робити з графом, він уміє і з агентом: зберігати state, стримити, ставити на паузу, вкладати всередину іншого графа як звичайний node. Обирати між agent і graph не потрібно, це одне й те саме на різній висоті абстракції.

Пакети

Бібліотека розкладена на багато пакетів; ставити треба не все.

Провайдери моделей

Сама бібліотека до жодного вендора не прив’язана: провайдер підключається окремим пакетом, і перемикання між ними — це зміна рядка ініціалізації, а не переписування коду. Пакети версіонуються незалежно від ядра.

Рядок на кшталт "openai:gpt-5.5" лише обирає вже встановлену інтеграцію — пакет вендора має бути в залежностях.

Окрім моделей, окремими пакетами підключаються й інші зовнішні сервіси: векторні сховища (langchain-chroma, langchain-postgres для pgvector, langchain-pinecone, langchain-qdrant, langchain-weaviate), вебпошук (langchain-tavily), а все, що не має свого пакета, живе в langchain-community.

Версіонування

Про версіонування варто знати одну річ. Основні пакети — langchain, langchain-core, langgraph — дотримуються семантичного версіонування: зміни, що ламають сумісність, бувають лише на зміні мажорної версії, а застарілі можливості й далі працюють із попередженням на всій лінійці. Звідси робоча нижня межа >=1.0,<2.0. А langchain-community семантичного версіонування не дотримується, і його зазвичай закріплюють на конкретній мінорній серії.

Який шар брати під задачу

Перевіряти згори вниз і зупинятися на першому збігу:

Останній пункт застосовний частіше, ніж здається: цикл агента навколо одного виклику додає і токени, і затримку.

Встановлення та перший виклик

pip install "langchain>=1.0,<2.0" "langchain-core>=1.0,<2.0" "langgraph>=1.0,<2.0" langchain-openai

langchain-core ставиться явно. Він приїде й транзитивно, але тоді його версією керує не ваш файл залежностей.

Пакет провайдера обов’язковий: рядок на кшталт "openai:gpt-5.5" розв’язується в конкретну інтеграцію, але сам її не встановлює — це пошук серед уже встановленого.

from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-5.5", temperature=0, timeout=30)

response = await model.ainvoke("Чому в папуг яскраве пір’я?")
print(response.text)

Формат рядка — provider:model. Приклади з документації: openai:gpt-5.5, anthropic:claude-sonnet-4-6, google_genai:gemini-2.5-flash-lite. Замість рядка можна створити об’єкт провайдера напряму — так відкривається доступ до параметрів, яких немає в загальному інтерфейсі.

response.text — властивість, а не метод. У попередній версії це був виклик із дужками.

Синхронний і асинхронний виклики

Кожен метод має дві форми: invoke та ainvoke, stream та astream. Для скриптів і ноутбуків підійде будь-яка. Для вебсервісу різниця принципова: синхронний виклик блокує цикл подій цілком, і поки він чекає на відповідь моделі, усі паралельні запити стоять. У сервісі на шляху запиту використовуються лише асинхронні форми.

Є ще abatch — паралельна обробка кількох незалежних промптів. Вона корисна для офлайн-завдань на кшталт прогону датасета, а не для одного ходу користувача.

Параметри model

Про rate_limiter є нюанс: вбудована реалізація InMemoryRateLimiter працює в межах одного процесу. За п’яти реплік сервісу фактична частота буде вп’ятеро більшою за задану, і для спільного на кластер ліміту потрібен спільний обмежувач.

Про reasoning_effort: на моделях, що міркують, глибина роздумів задається цим параметром. Інструкція «думай крок за кроком» у промпті працює з власним механізмом міркування моделі паралельно, а не доповнює його. Тема промптів велика, у мене про неї є окрема стаття.

Fallbacks — запасні моделі

model = init_chat_model("openai:gpt-5.5").with_fallbacks([init_chat_model("openai:gpt-5-nano")])

Перемикання спрацьовує при винятку від основної моделі. Це спосіб пережити збій провайдера, не віддаючи користувачеві помилку.

Messages: що надходить у відповіді

Типів messages чотири: SystemMessage несе інструкції, HumanMessage — введення користувача, AIMessage — відповідь моделі разом із запитами на виклики tools і метаданими, ToolMessage — результат виклику, що йде назад у модель.

У відповіді моделі є два подання тих самих даних:

• content — сире навантаження провайдера. Рядок або його власний список словників, збережений як є.
• content_blocks — типізований розбір поверх того самого, однаковий для всіх провайдерів.

Навіщо друге. Провайдери повертають однакові за суттю речі в різній формі: в одного міркування приїжджає під ім’ям thinking, в іншого — reasoning. У content_blocks це в обох випадках блок типу reasoning.

for block in response.content_blocks:
    if block["type"] == "reasoning":
        log_reasoning(block)
    elif block["type"] == "text":
        show_to_user(block)
    elif block["type"] == "tool_call":
        schedule(block)

Типи блоків покривають текст і міркування; картинки, аудіо, відео та файли; виклики tools і їхні фрагменти під час streaming; виклики tools, що виконуються на боці провайдера; і окремий тип non_standard для всього, що в стандарт не вклалося.

Практичний наслідок: код, який читає content_blocks, переживає зміну провайдера. Код, що розбирає content вручну, прив’язаний до конкретного вендора.

Мультимодальне введення

from langchain.messages import HumanMessage

message = HumanMessage(content=[
    {"type": "text", "text": "Що на цій картинці?"},
    {"type": "image", "url": "https://example.com/photo.png"},
])

Обрізання історії

Діалог росте, контекстне вікно скінченне. Простий спосіб — детерміноване обрізання:

from langchain.messages import trim_messages

trimmed = trim_messages(messages, max_tokens=8000, strategy="last", token_counter=model)

Розумніший спосіб — сумаризація старої частини історії, для agents вона доступна готовим middleware. Різниця в тому, що обрізання передбачуване і втрачає старе цілком, а сумаризація зберігає зміст, але переписує те, що асистент пам’ятає.

Облік токенів

from langchain_core.callbacks import UsageMetadataCallbackHandler

callback = UsageMetadataCallbackHandler()
await model.ainvoke("Привіт", config={"callbacks": [callback]})
callback.usage_metadata      # {'input_tokens': 8, 'output_tokens': 10, ...}

Те саме доступне в кожній відповіді в полі usage_metadata. Це основа для підрахунку витрат і квот користувачів.

Промпти

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "Ти допомагаєш розібратися в логах збірки. Відповідай коротко."),
    ("human", "{question}"),
])

chain = prompt | model

ChatPromptTemplate збирає список messages з підстановкою змінних. Ролі при цьому зберігаються окремо — а це суттєво: провайдер обробляє системну інструкцію та введення користувача по-різному, у них різна вага й різний рівень довіри. Якщо склеїти їх в один рядок, текст користувача опиниться там, де модель очікує ваших інструкцій — це і втрата якості, і відкрита дорога для ін’єкції промпта.

Вертикальна риска — композиція: результат лівого об’єкта передається правому. Отриманий ланцюжок сам уміє ainvoke, astream і abatch та може бути частиною чогось більшого.

Structured output — відповідь за схемою

Часто від моделі потрібен не текст, а дані: категорія звернення, витягнуті поля, рішення про маршрут.

from pydantic import BaseModel, Field

class BuildFailure(BaseModel):
    module: str
    reason: str
    is_flaky: bool = Field(description="Чи схоже на плаваючий тест")

structured = model.with_structured_output(BuildFailure)
result = await structured.ainvoke(log_text)   # -> BuildFailure(...)

Приймаються Pydantic-моделі, TypedDict і гола JSON-схема. Яким саме способом це реалізується під капотом — нативний режим провайдера, виклик функції чи JSON-режим — обирається за можливостями конкретної моделі.

Підхід «попросити JSON у промпті й розібрати прозу» в першій версії прибрали як ненадійний, і повертати його руками не варто: розібрана схема дає помилку одразу, а розбір прози ламається тихо й виявляється далі по коду.

Tools

Tool — функція вашого коду, опис якої йде моделі разом із промптом. Модель сама її не виконує: вона повертає запит на виклик з аргументами, виклик виконує ваш код, результат іде назад у модель.

from langchain.tools import tool

@tool
def search_listings(city: str, max_price: int = 1_000_000) -> str:
    """Пошук активних оголошень про продаж нерухомості в місті.

    Args:
        city: Назва міста, наприклад «Одеса»
        max_price: Верхня межа ціни в доларах
    """
    return render(query(city, max_price))

Три елементи тут є несучими:

Анотації типів формують схему, яку побачить модель. Без них аргумент не має типу, і значення доведеться вгадувати.

Docstring — це частина промпта. Він єдиний пояснює моделі, у яких випадках tool застосовний. Від його формулювання прямо залежить, чи буде tool викликано вчасно.

Ім’я у snake_case. Частина провайдерів інші форми відхиляє.

Коли анотацій недостатньо — потрібен перелічуваний набір значень, опис на кожне поле або валідація — задається явна схема аргументів:

from pydantic import BaseModel, Field
from typing import Literal

class WeatherInput(BaseModel):
    location: str = Field(description="Місто або координати")
    units: Literal["celsius", "fahrenheit"] = "celsius"

@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius") -> str:
    """Поточна погода."""
    ...

ToolRuntime — доступ до середовища

Tool часто потрібні дані, яких немає в аргументах: хто зараз користувач, який state у діалогу, куди писати довготривалу пам’ять. Для цього в сигнатуру додається параметр ToolRuntime — фреймворк підставить його сам, і в схему для моделі він не потрапить.

from langchain.tools import tool, ToolRuntime

@tool
def get_account_info(runtime: ToolRuntime[UserContext]) -> str:
    """Повертає інформацію по рахунку поточного користувача."""
    user_id = runtime.context.user_id
    return describe(lookup(user_id))

Через нього доступні: runtime.state — поточний state агента разом з історією messages; runtime.context — незмінний context запуску; runtime.store — довготривала пам’ять; runtime.stream_writer — записувач у потік подій; runtime.tool_call_id — ідентифікатор поточного виклику; runtime.execution_info — ідентифікатори thread і запуску; плюс конфігурація запуску.

Момент, важливий для безпеки: особу користувача, ідентифікатор орендаря та ключі беруть із runtime.context, а не з аргументів. Аргументи заповнює модель, а на модель можна вплинути текстом запиту — ідентифікатор користувача в схемі tool означає, що підставити туди чужий можна, просто попросивши про це в чаті.

Що tool повертає

from langchain.messages import ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.types import Command

@tool
def set_language(language: str, runtime: ToolRuntime) -> Command:
    """Задає мову, якою відповідати далі."""
    return Command(update={
        "preferred_language": language,
        "messages": [ToolMessage(content=f"Мову перемкнено на {language}.",
                                 tool_call_id=runtime.tool_call_id)],
    })

Коли tool повертає Command, ToolMessage треба включити в оновлення самостійно. Без нього провайдер побачить виклик tool без результату й на наступному ході його відхилить.

Прапорець return_direct завершує цикл і віддає результат tool як фінальну відповідь. Він підходить, коли вивід tool і є відповіддю, і не підходить, коли модель має його прокоментувати або поєднати з іншими даними.

Прогрес із tool

Довгий tool може повідомляти про хід роботи — це потрапить у потік подій:

@tool
def index_documents(folder: str, runtime: ToolRuntime) -> str:
    """Індексує теку з документами."""
    writer = runtime.stream_writer
    writer({"type": "progress", "done": 0, "total": 100})
    ...

Помилки tools

За замовчуванням виняток із tool припиняє весь запуск. Для частини випадків це не те, що потрібно: невдалий пошук — інформація, з якою модель може працювати далі.

from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage

@wrap_tool_call
def handle_tool_errors(request, handler):
    try:
        return handler(request)
    except Exception as exc:
        return ToolMessage(content=f"Інструмент не спрацював: {exc}",
                           tool_call_id=request.tool_call["id"])

Є й готові middleware для цього — ToolErrorMiddleware перетворює виняток на message, ToolRetryMiddleware повторює із затримкою. Усередині зібраного вручну графа ту саму роль відіграє прапорець handle_tool_errors у ToolNode.

Випадки варто розділяти, бо лікуються вони по-різному: тимчасовий збій мережі або відповідь 429 — повтор; нічого не знайшлося чи хибний аргумент — повернути message, модель спробує інакше; бракує даних, які є лише в людини — interrupt; справжній дефект — виняток назовні.

Окремо: текст внутрішнього винятку не варто класти в ToolMessage на шляху до користувача. Він іде в модель, а звідти може потрапити у відповідь користувачеві разом зі стектрейсом і параметрами підключення.

Tools на боці провайдера

Частину tools виконує сам провайдер — вебпошук, інтерпретатор коду. У відповіді вони видно блоками server_tool_call і server_tool_result, тарифікуються провайдером і вашого коду не виконують. З цього випливає, що ваші таймаути, повтори та журналювання до них не застосовуються.

Є й зворотний варіант — headless tools: схема оголошена, реалізації немає. Запуск ставиться на паузу з описом потрібної дії, застосунок виконує її там, де належить (у браузері, в іншому сервісі) і відновлює роботу з результатом.

Agent

from langchain.agents import create_agent

agent = create_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[search_listings, send_email],
    system_prompt="Ти асистент ріелтора. Відповідай коротко і по суті.",
)

result = agent.invoke({"messages": [{"role": "user", "content": "Що є в Одесі до 200 тисяч?"}]})
print(result["messages"][-1].content)

Що тут зібралося:

Цикл завершується, коли модель відповідає без виклику tools. Якщо вона запросила кілька tools за один хід, вони виконуються паралельно, і кожен повертає свій ToolMessage.

Сам цикл невеликий і фіксований. Місця, де в нього можна втрутитися, винесені в middleware — тобто налаштування агента йде не через аргументи конструктора, а через хуки навколо циклу.

Параметри create_agent

У прикладах модель задають рядком, у робочому коді її зазвичай беруть із конфігурації разом із таймаутом, повторами та ланцюжком fallbacks.

Результат — словник

result["messages"][-1].content    # текст відповіді
result["structured_response"]     # розібрана схема, якщо задано response_format

Повертається фінальний state, а не message, тому звертання на кшталт result.content дасть AttributeError.

Response format — схема всередині циклу

from pydantic import BaseModel

class Weather(BaseModel):
    temperature: float
    condition: str

agent = create_agent("openai:gpt-5.5", tools=[weather_tool], response_format=Weather)

result = agent.invoke({"messages": [{"role": "user", "content": "Погода в Одесі?"}]})
result["structured_response"]   # Weather(temperature=27.0, condition='ясно')

Схема застосовується всередині циклу, а не другим викликом моделі поверх готової відповіді. Стратегій дві: ProviderStrategy використовує нативну структуровану видачу провайдера (надійніше, дешевше за токенами, доступна не всюди), ToolStrategy емулює її через виклик tool (працює на будь-якій моделі з підтримкою tools і дозволяє об’єднувати кілька альтернативних схем). Якщо передати схему без вказівки стратегії, LangChain подивиться на профіль моделі й обере сам.

У ToolStrategy налаштовується handle_errors — поведінка при помилці валідації: за замовчуванням запит повторюється, а моделі передається текст помилки. Можна задати своє повідомлення, обмежити повтори конкретними типами винятків, передати функцію-обробник або вимкнути повтори зовсім.

Profile — можливості моделі

model.profile
# {'max_input_tokens': 400000, 'tool_calling': True, 'reasoning_output': True, 'multimodal': True, ...}

Опис можливостей конкретної моделі. За ним можна перевірити підтримку tools перед їх прив’язкою або звіритися з розміром контексту — замість розгалужень за назвою моделі. Тим самим профілем користується механізм вибору стратегії для response_format.

Context і state — різні речі

from dataclasses import dataclass

@dataclass
class Context:
    user_id: str

agent.invoke(
    {"messages": [{"role": "user", "content": "Який у мене баланс?"}]},
    config={"configurable": {"thread_id": "t-1"}},
    context=Context(user_id="user-123"),
)

Особу користувача кладуть у context, а не в state: state зберігається та переграється, тому при відновленні старого діалогу у свіжий запуск потрапив би вчорашній користувач.

Пам’ять між ходами

from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(model=..., tools=tools, checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "conversation-42"}}

agent.invoke({"messages": [{"role": "user", "content": "Мене звати Аліса"}]}, config=config)
agent.invoke({"messages": [{"role": "user", "content": "Як мене звати?"}]}, config=config)

Потрібні обидві частини: checkpointer і thread_id. Checkpointer без thread_id нічого не зберігає й при цьому не видає помилки — проявляється це тим, що асистент не пам’ятає попереднього ходу.

InMemorySaver живе до перезапуску процесу й призначений для розробки; для продакшену є Postgres.

Recursion limit

Цикл обмежений кількістю super-steps, за замовчуванням їх 25. При вичерпанні здіймається виняток.

result = agent.invoke(payload, config={"recursion_limit": 40})

Це захист від нескінченного циклу. Якщо потрібна продуктова межа — наприклад, не більше п’яти пошуків за діалог — для цього є ModelCallLimitMiddleware і ToolCallLimitMiddleware: вони завершують роботу штатно, а не винятком.

Middleware — шість точок налаштування

Middleware — код, що виконується навколо кроків циклу агента. Через нього налаштовується все: підміна моделі на льоту, обмеження, повтори, маскування персональних даних, пауза на людину, збирання динамічного промпта.

На схемі це виглядає як шари навколо циклу:

Форм дві. Node-style хуки (before_*, after_*) отримують state і повертають його оновлення — вони спостерігають і доповнюють. Wrap-style (wrap_*) стоять прямо в шляху виклику, отримують запит і функцію-продовження: можуть повторити, підмінити параметри, повернути свій результат замість справжнього.

Порядок виконання

Для списку з трьох middleware порядок такий:

Хуки before_* ідуть згори вниз, after_* — знизу вгору, wrap_* вкладаються один в одного, і перше middleware в списку виявляється найзовнішнім.

Звідси два практичні наслідки. Повтор, який має обгорнути все інше, ставлять першим. Перевірка, що зобов’язана побачити фінальну відповідь, теж ставиться ближче до початку списку — на зворотному шляху вона спрацює останньою.

Як пишеться

Один хук — декоратором:

from langchain.agents.middleware import before_model, wrap_model_call
from langchain.messages import AIMessage

@before_model(can_jump_to=["end"])
def stop_long_conversation(state, runtime):
    if len(state["messages"]) >= 50:
        return {"messages": [AIMessage("Діалог став надто довгим.")], "jump_to": "end"}
    return None

@wrap_model_call
def upgrade_for_experts(request, handler):
    if request.runtime.context.tier == "expert":
        return handler(request.override(model=strong_model, tools=advanced_tools))
    return handler(request)

ModelRequest несе messages, model, набір tools, системне повідомлення, state і runtime. Він не мутується: request.override(...) повертає змінену копію, яку й передають далі.

Node-style хуки можуть не повертатися в цикл, а перейти одразу до кінця, до tools або до model. Такий перехід оголошується заздалегідь списком можливих цілей — параметром can_jump_to, як у прикладі вище; у класовій формі те саме задається декоратором @hook_config над методом.

Wrap-style хуки зобов’язані повертати значення: генератор із yield усередині дасть помилку в рантаймі.

Кілька хуків одразу, власний state або свої tools — це клас:

from langchain.agents.middleware import AgentMiddleware, AgentState
from typing_extensions import NotRequired

class UsageState(AgentState):
    model_calls: NotRequired[int]

class UsageMiddleware(AgentMiddleware):
    state_schema = UsageState
    tools = [reset_usage]

    async def aafter_model(self, state, runtime):
        return {"model_calls": state.get("model_calls", 0) + 1}

Node-style хуки мають асинхронні варіанти з префіксом a — на шляху запиту реалізують саме їх.

Dynamic prompt

system_prompt задається статично. Усе, що залежить від state, context або знайдених документів, збирається окремим хуком:

from langchain.agents.middleware import dynamic_prompt

@dynamic_prompt
def prompt_for_tier(request):
    tier = request.runtime.context.tier
    return f"Ти агент підтримки. У клієнта тариф «{tier}»."

agent = create_agent(model=..., tools=tools, middleware=[prompt_for_tier], context_schema=Context)

Каталог готових middleware

Більша частина типових вимог закрита готовими реалізаціями.

Контекст і пам’ять. SummarizationMiddleware — сумаризація історії при наближенні до ліміту токенів, із налаштуваннями порога, збереженого хвоста та способу підрахунку. ContextEditingMiddleware — очищення старих результатів tools заради вивільнення контексту.

Безпека та контроль. HumanInTheLoopMiddleware — пауза на схвалення людиною. PIIMiddleware — виявлення та маскування персональних даних за типом, із вибором стратегії й указівкою, до чого застосовувати: до введення, до виведення, до результатів tools. ModelCallLimitMiddleware і ToolCallLimitMiddleware — обмеження кількості викликів у межах thread або одного запуску, із вибором поведінки при досягненні межі.

Надійність. ModelRetryMiddleware і ToolRetryMiddleware — повтор з експоненційною затримкою. ToolErrorMiddleware — перетворення винятків на message, з яким модель може працювати. ModelFallbackMiddleware — перемикання на запасні моделі при відмові основної.

Можливості. TodoListMiddleware — список завдань для планування. LLMToolSelectorMiddleware — попередній відбір релевантних tools дешевою моделлю, коли їх стало багато. SubAgentMiddleware — делегування підзадач ізольованим subagents. FilesystemMiddleware і FilesystemFileSearchMiddleware — файлова система поверх підключуваного бекенда та пошук по файлах. ShellToolMiddleware — постійна сесія командної оболонки. RubricMiddleware — самоперевірка за заданими критеріями з ітераціями. LLMToolEmulator — емулятор виконання tools на моделі, для тестів.

Специфічне для провайдерів. Кешування промпта та вбудовані tools в Anthropic, кешування в Bedrock, модерація вмісту в OpenAI.

Два попередження щодо поєднання. Механізм fallbacks є і в самої моделі, і в middleware — працювати має один із них, інакше вийде два шари перемикання один поверх одного. І SummarizationMiddleware переписує історію, тобто змінює те, що асистент пам’ятає про розмову: вмикати її варто свідомо.

Middleware чи node

Обидві конструкції виконують код навколо виклику моделі. Межа така: middleware — коли поведінка наскрізна й не пов’язана зі змістом розмови (повтори, маскування, ліміти, трасування, збирання промпта). Node — коли крок є частиною процесу: рішення про маршрут, окрема фаза роботи, те, що ви намалювали б на схемі.

LangGraph: коли порядок кроків задаєте ви

Agent підходить, поки послідовність дій обирає модель. Інший клас завдань виглядає інакше: спочатку завжди завантажити профіль користувача, потім класифікувати запит, далі за результатом — або уточнювальне питання, або відповідь, а після відповіді — запис у журнал. Тут порядок відомий заздалегідь, і описувати його словами в промпті необов’язково: він описується кодом.

Graph — спосіб описати процес явно. Детерміновані кроки лишаються кодом, рішення лишаються за моделлю, і все разом утворює схему, яку можна намалювати й обговорити.

Ось як виглядає типовий чат-конвеєр, зібраний графом. Позначено, що робить кожен крок: звичайний код чи виклик моделі.

Тут видно головну відмінність від агента: порядок кроків заданий явно. Модель вирішує, що за питання їй поставили, а не в якому порядку далі виконувати роботу.

Три сутності

State — спільні дані, з якими працюють усі кроки. Nodes — самі кроки, звичайні функції, кожна повертає часткове оновлення state. Edges — переходи: що виконується наступним.

Виконання йде super-steps. Усе, що заплановано на поточний такт, виконується — паралельно, якщо граф це дозволяє; потім оновлення зливаються в state, потім планується наступний такт. Перед запуском граф компілюється.

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    question: str
    answer: str

def answer(state: State) -> dict:
    return {"answer": f"Відповідь на {state['question']}"}

graph = (
    StateGraph(State)
    .add_node("answer", answer)
    .add_edge(START, "answer")
    .add_edge("answer", END)
    .compile()
)

START і END — спеціальні позначки: перша означає точку входу, куди потрапляє введення користувача, друга — вихід. Edge назад у START неприпустимий; якщо потрібен цикл, він робиться через іменований node.

Схемою state може бути TypedDict, Pydantic-модель або dataclass. Частіше беруть перше як найлегше.

Reducers — як зливаються оновлення

У кожного ключа state є reducer: правило, за яким оновлення від node поєднується з тим, що вже лежить у state.

from typing import Annotated
import operator
from langgraph.graph.message import add_messages

class State(TypedDict):
    name: str                                     # за замовчуванням: перезапис
    findings: Annotated[list[str], operator.add]  # додавання в кінець
    messages: Annotated[list, add_messages]       # додавання з дедуплікацією за id

Наочно, що відбувається на одному super-step, коли два nodes відпрацювали паралельно:

Reducer за замовчуванням перезаписує значення. Тому список без явного reducer, у який пишуть два nodes, збереже лише значення одного з них — без помилки й без попередження. Будь-яке накопичувальне поле має мати reducer.

Для messages є готовий add_messages: він додає нові, відкидає дублікати за ідентифікатором і перетворює словники на об’єкти messages. Є й готовий клас MessagesState, який вмикає його одразу.

Другий спосіб обійти reducers — повернути з node увесь state цілком:

def good(state: State) -> dict:
    return {"answer": "..."}          # лише те, що змінилося

def bad(state: State) -> State:
    state["answer"] = "..."           # мутація проходить повз reducers
    return state

Ще в графа можна задати окремі схеми входу й виходу — тоді внутрішні робочі поля не будуть частиною публічного інтерфейсу.

Nodes

Node приймає state, а за потреби ще RunnableConfig (у ньому лежать thread_id, теги, налаштовувані значення) або Runtime (context, store, записувач у потік). Асинхронні nodes — звичайні async def із тими самими сигнатурами.

Зручний прийом для сервісу: node створюється фабричною функцією, яка замикає в собі залежності — клієнт бази, HTTP-сесію, конфігурацію — і повертає функцію node. Так залежності не перетворюються на глобальні змінні, а node легко тестувати окремо.

Маршрутизація

def route(state: State) -> Literal["clarify", "generate"]:
    return "clarify" if state["needs_input"] else "generate"

builder.add_conditional_edges("assess", route, {"clarify": "clarify", "generate": "generate"})

Третій аргумент — карта значень, що повертаються, в імена nodes. Він необов’язковий, але з ним граф коректно малюється, а множина досяжних nodes стає явною.

Коли треба одночасно оновити state й обрати наступний крок, node повертає Command:

from langgraph.types import Command
from typing import Literal

def triage(state: State) -> Command[Literal["escalate", "resolve"]]:
    if state["severity"] > 3:
        return Command(update={"assigned": "oncall"}, goto="escalate")
    return Command(update={"assigned": "bot"}, goto="resolve")

Анотація Literal з переліком досяжних nodes потрібна для відмальовування та перевірки цілей.

Важлива деталь поведінки: Command додає динамічний edge, а не замінює статичний. Якщо з того самого node є звичайний edge, виконаються обидва напрямки. Зовні це виглядає як дублювання роботи.

Із subgraph команда може перевести керування в батьківський граф — для цього є Command(goto=..., graph=Command.PARENT).

Send — паралельний запуск

Коли кількість гілок відома лише в рантаймі:

from langgraph.types import Send

class State(TypedDict):
    topics: list[str]
    drafts: Annotated[list[str], operator.add]     # накопичувач обов’язковий

def fan_out(state: State):
    return [Send("write_draft", {"topic": topic}) for topic in state["topics"]]

builder.add_conditional_edges(START, fan_out, ["write_draft"])
builder.add_edge("write_draft", "synthesise")

Кожен Send несе свій приватний вхід одному викликові обробника — так робиться схема «розіслати й зібрати»:

Результати збираються через reducer; якщо reducer на збиральному полі немає, збережеться результат останнього обробника, що завершився.

Subgraphs

Скомпільований граф — виконуваний об’єкт, тому його можна поставити node іншого графа:

builder.add_node("research", research_graph)

Спільні ключі state перетікають автоматично. Якщо схеми state різні, потрібен node-перехідник, який перетворює дані на вході й на виході.

Керування на рівні node

from langgraph.types import RetryPolicy, CachePolicy
from langgraph.cache.memory import InMemoryCache

builder.add_node("fetch", fetch, retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0))
builder.add_node("embed", embed, cache_policy=CachePolicy(ttl=300))
graph = builder.compile(cache=InMemoryCache())

Звідти ж — RunControl.request_drain(): кооперативна зупинка, яка просить запуск зупинитися в найближчій безпечній точці й лишає checkpoint для продовження. Це штатний спосіб пережити викочування нової версії або зупинку контейнера посеред роботи.

recursion_limit у графа такий самий, як в агента, і задається в конфігурації запуску. Є й спеціальне поле state RemainingSteps — node може подивитися в нього й згорнути роботу заздалегідь, замість того щоб упертися в межу.

Checkpointer і store: пам’ять і стійкість

Пам’яті дві, і вони розв’язують різні задачі.

Ні те ні інше за замовчуванням не ввімкнено.

Checkpointers

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "conversation-1"}}

await graph.ainvoke({"messages": ["Привіт"]}, config)
await graph.ainvoke({"messages": ["І ще раз"]}, config)   # бачить перший хід

Дві експлуатаційні деталі. Метод .setup(), що створює таблиці, — операція рівня викочування, а не старту застосунку: зміна схеми на старті за кількох реплік виконується паралельно сама із собою. І thread_id для Postgres тримають коротшим за 255 символів — стовпець обмежений.

Перегляд історії та перемотування

snapshot = await graph.aget_state(config)                        # поточний state і що далі
history = [s async for s in graph.aget_state_history(config)]    # свіжі записи першими

past = history[-2]
await graph.ainvoke(None, past.config)                           # переграти з тієї точки

fork = await graph.aupdate_state(past.config, {"messages": ["виправлено"]})
await graph.ainvoke(None, fork.config)                           # продовжити новою гілкою

Виклик із None замість вхідних даних означає «продовж із checkpoint». На цьому ж побудоване налагодження: можна відмотати до потрібного кроку, поправити state й подивитися, як піде далі.

Одна особливість: update_state проходить через reducers. На полі з додаванням воно додасть там, де передбачалася заміна. Для заміни є обгортка Overwrite:

from langgraph.types import Overwrite

await graph.aupdate_state(config, {"items": Overwrite(["C"])})

Зберігання checkpoints

Checkpoints створюються по одному на super-step для кожного thread і самі не видаляються. Для довговічного продукту це означає постійне зростання таблиці, тому політику зберігання — скільки часу thread лишається відновлюваним і що видаляє решту — варто визначити заздалегідь.

Store

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
graph = builder.compile(checkpointer=checkpointer, store=store)

Операції: put, get, search, delete. Namespace — ієрархічний кортеж на кшталт ("users", user_id, "preferences"), і він же слугує межею ізоляції. Збирають його з довіреного context, а не з тексту, що надійшов від моделі: namespace без ідентифікатора користувача чи орендаря означає, що дані видно всім.

Якщо задати store функцію ембедингів і розмірність через IndexConfig, пошук стане семантичним замість точного. Продакшен-реалізація — PostgresStore.

У графі до store звертаються через runtime, а не через глобальну змінну:

def recall(state, runtime: Runtime):
    item = runtime.store.get(("users", runtime.context.user_id), "preferences")
    ...

Durability — режими запису

Налаштування того, як часто state записується на диск. Передається під час виклику:

await graph.ainvoke(payload, config=config, durability="async")

Вибір залежить від ціни втраченого кроку: якщо кроки надсилають листи або проводять платежі, підходить "sync"; для аналітичного прогону довгим діалогом достатньо "exit".

Переграння при відновленні

Загальне правило рантайму: відновлення виконує node із самого початку, пропускається лише завершена робота, що потрапила в checkpoint. Звідси два наслідки:

• Побічні ефекти, виконані до точки зупинки, відбудуться ще раз.
• Недетерміновані значення — час, випадкові ідентифікатори — при переграванні вийдуть іншими. Їх обчислюють в окремому node, щоб значення потрапило в checkpoint і підставлялося звідти.

Human-in-the-loop: пауза на людину

Іноді запуск має зупинитися й дочекатися людини: підтвердити надсилання, поправити чернетку, відповісти на уточнювальне питання. Для цього є interrupt.

from langgraph.types import interrupt, Command

def review(state):
    decision = interrupt({"draft": state["draft"], "question": "Публікуємо?"})
    return {"approved": decision == "yes"}

Виклик зупиняє запуск, зберігає все в checkpoint і віддає назовні передані дані. Запуск лишається призупиненим — секунду чи тиждень — доки його не продовжать через Command(resume=...); тоді interrupt поверне передане значення.

Зверніть увагу на друге з’явлення review у схемі: node з паузою при відновленні виконується із самого початку. Це окрема тема, до неї повернемося нижче.

Checkpointer і thread_id для цього обов’язкові: без них паузу нікуди зберегти.

Є й статичний варіант — interrupt_before та interrupt_after при компіляції, зупинка на межі node. Він зручний для налагодження та покрокового проходу, але не несе корисного навантаження, тому для продуктових сценаріїв схвалення використовують звичайний interrupt.

Схвалення викликів tools

Для найчастішого випадку — підтвердити небезпечний виклик до виконання — є готове middleware:

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[write_file, execute_sql, read_data],
    checkpointer=checkpointer,
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={
            "write_file": {"allowed_decisions": ["approve", "edit", "reject"]},
            "execute_sql": {"allowed_decisions": ["approve", "reject"]},
            "read_data": False,
        },
        description_prefix="Очікує схвалення",
    )],
)

True вмикає interrupt із рішеннями за замовчуванням, False пропускає виклик без питань, словник налаштовує, які рішення доступні перевіряльнику.

await agent.ainvoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config, version="v2")

Коли модель за один хід запитує кілька захищених tools, вони пред’являються разом, і список рішень має йти в тому самому порядку, що й пред’явлені дії.

Interrupt можна вмикати за умовою — параметром when: функція дивиться на аргументи виклику й вирішує, чи потрібне підтвердження. Так перевіряються лише ризиковані випадки, наприклад запис за межі робочого каталогу, а решта проходить без питань.

Побічні ефекти й точка зупинки

Тут те саме правило переграння, але з наочними наслідками:

def send_and_confirm(state):
    send_email(state["draft"])              # надішлеться ще раз при відновленні
    interrupt({"sent": True})

def confirm_then_send(state):
    decision = interrupt({"draft": state["draft"]})
    if decision == "approve":
        send_email(state["draft"])          # після паузи — один раз
    return {"sent": decision == "approve"}

Звідси три правила: побічні ефекти розміщують після interrupt або роблять ідемпотентними; недетерміновані значення виносять в окремий node; node з interrupt тримають невеликим, бо повториться все, що в ньому є.

Той самий механізм використовується не лише для схвалень: ним редагують state, запитують дані, які є лише в людини, і виконують headless tools на боці клієнта.

Streaming

Потокова видача потрібна, щоб відповідь з’являлася в міру генерації, а на довгих операціях було видно, що відбувається.

Звідки що береться:

Тут два різні API.

Перший — stream / astream із вказівкою stream_mode. Підходить для більшості завдань.

Режими можна запитувати по кілька одразу, списком.

Версія формату потоку

У формату є версія, і за замовчуванням діє v1, де форма відповіді залежить від виклику: один режим віддає сирі дані, кілька режимів — кортежі (mode, data), а ввімкнені subgraphs додають у кортеж namespace.

Версія v2 віддає все однаково — словник StreamPart із полями type, ns і data:

async for chunk in graph.astream(payload, stream_mode="messages", version="v2"):
    if chunk["type"] == "messages":
        message, metadata = chunk["data"]
        if metadata["langgraph_node"] in {"generate", "answer_user"}:
            print(message.content, end="")

У новому коді версію вказують явно: замовчування лишилося колишнім заради сумісності.

Метадані несуть langgraph_node, теги та ідентифікатори запуску. Фільтрація за іменем node — спосіб не пустити в потік користувача токени внутрішніх викликів, наприклад класифікації. Можна вчинити й інакше — позначити таку модель тегом nostream:

classifier = model.with_config({"tags": ["nostream"]})

Тоді її токени не емітуються взагалі.

Subgraphs у потоці

async for chunk in graph.astream(payload, stream_mode="messages", subgraphs=True, version="v2"):
    ...

Без subgraphs=True токени, породжені всередині вкладеного графа — включно з agent, поставленим node, — у потік не потрапляють. Помилки при цьому немає, потік просто виявляється неповним. Поле ns у кожному фрагменті показує джерело: порожній кортеж для кореневого графа, ім’я node з ідентифікатором завдання для вкладеного.

Custom events

from langgraph.config import get_stream_writer

def index_documents(state):
    writer = get_stream_writer()
    writer({"type": "progress", "done": 0, "total": 100})

Читаються в режимі custom. Той самий механізм — спосіб віддати в потік модель, яка не має інтеграції з LangChain: викликати її самостійно й записувати фрагменти вручну.

astream_events

Другий API детальніший: кожен Runnable у графі повідомляє про старт, про фрагменти й про завершення.

async for event in graph.astream_events(state, version="v2"):
    kind = event["event"]          # on_chain_start, on_chat_model_stream, on_chain_end, on_tool_start…
    name = event["name"]
    data = event["data"]
    metadata = event.get("metadata", {})   # тут же langgraph_node

Це потрібно, коли важливо знати, який саме node породив токен: так межі nodes перетворюються на стадії для користувача («шукаю», «звіряю», «відповідаю»), а токени різних nodes ідуть у різні частини інтерфейсу. Плата — обсяг: потік подій великий, і споживачеві потрібна точна фільтрація, щоб внутрішній вміст не потрапив в інтерфейс.

Є й новіший stream_events(version="v3"), побудований навколо блоків вмісту, з типізованими проєкціями за каналами повідомлень, значень і життєвого циклу. Перехід на нього — переписування споживача, а не зміна значення прапорця.

Functional API

Другий фасад до того самого рантайму. Процес описується звичайним Python — із if, for і await, — але зберігаються checkpoints, відновлення, interrupts і streaming.

from langgraph.func import entrypoint, task

@task
def write_essay(topic: str) -> str:
    return generate(topic)

@entrypoint(checkpointer=checkpointer)
def workflow(topic: str) -> dict:
    essay = write_essay(topic).result()
    approved = interrupt({"essay": essay, "action": "схвалюємо?"})
    return {"essay": essay, "approved": approved}

@entrypoint приймає один позиційний аргумент — кілька значень передають словником. Вхід і вихід мають серіалізуватися, бо потрапляють у checkpoint. Схеми state тут немає: замість неї в наступний запуск підставляється результат попереднього через параметр previous. Є й entrypoint.final, коли назовні повертається одне значення, а для наступного запуску зберігається інше.

@task — одиниця роботи, результат якої записується в checkpoint. Виклик повертає future одразу; кілька завдань, запущених до того, як розв’язано перше, виконуються паралельно — тут це замінює Send.

Правило переграння те саме, але сформульоване для процедурного коду: при відновленні тіло entrypoint програється згори, а результати завершених завдань підставляються з checkpoint. Тому кожен побічний ефект і кожне недетерміноване значення розміщують усередині @task — звичайний код між завданнями виконається заново.

Вибір між двома API. Functional API коротший, коли потік здебільшого лінійний і спільного state, яке варто було б називати, немає. Graph API виграє, коли в процесу є структура, кілька кроків працюють з одним state і форму процесу треба обговорювати: graph можна відмалювати, функцію із завданнями — ні.

Multi-agent: кілька агентів

Слово «мультиагент» зазвичай означає одну з трьох конструкцій:

Graph як tool. Скомпільований граф виконуваний, тому його загортають у tool — і agent, що викликає, делегує роботу, нічого не знаючи про внутрішній устрій.

@tool
def research(topic: str) -> str:
    """Запускає дослідницький конвеєр за темою й повертає звіт."""
    return research_graph.invoke({"topic": topic})["report"]

Supervisor. Координатор розподіляє роботу між спеціалістами та збирає результати. У вигляді графа це node-маршрутизатор плюс по одному node на спеціаліста, маршрут обирається через Command. Така схема малюється й інспектується.

Swarm. Агенти передають керування один одному напряму, без центрального маршрутизатора. Гнучкіше, але кожен edge передачі — потенційний цикл, тому потрібен recursion_limit.

Ще варіант — subagents через SubAgentMiddleware: тоді у звичайного агента з’являється tool делегування, без решти обв’язки Deep Agents.

Subagent чи subgraph

Визначальна властивість — ізоляція контексту. Делегування subagent доречне, коли підзадача інакше заповнить батьківський контекст деталями, які далі не потрібні: читання довгого документа, шумний пошук. Subgraph доречний, коли крок — частина процесу і його state важливий для наступних кроків.

Про вартість: кожен subagent виконує свій цикл зі своїм системним промптом і своїми описами tools, тому запуск п’яти subagents витрачає приблизно як п’ять агентів, а не як один.

Deep Agents

Готова збірка middleware поверх звичайного агента, розрахована на довгі завдання, які не вміщуються в одне контекстне вікно.

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[get_weather],
    system_prompt="Ти дослідницький асистент.",
)

Що ввімкнено одразу:

• Файлова система з повним набором tools: перегляд, читання, запис, правка, видалення, пошук за маскою й за вмістом. Окремі tools можна виключити, сам шар — ні.
• Сумаризація та вивантаження контексту, щоб довге завдання не вперлося в розмір вікна.
• Кешування промпта для статичних частин у провайдерів, які це підтримують.
• Subagents через tool, який породжує тимчасового агента зі свіжим контекстом.

Планування завдань із версії 0.7 стало опційним — у матеріалах старіших воно описано як увімкнене за замовчуванням.

Файлова система підключувана: пам’ять, локальний диск, store графа, композиція з кількох джерел або власна реалізація. У бекендів є декларативні правила доступу за масками шляхів — вони визначають, що саме агентові дозволено читати й писати. Для виконання коду існують пісочниці з командною оболонкою та інтерпретатори, що виконують JavaScript.

Обв’язка підходить завданням, яким справді потрібні планування на багато кроків, файли як робоча пам’ять, делегування та пам’ять між сесіями. Для обмеженого циклу з кількома tools вона принесе можливості, які завдання не використовує, але які займають частину бюджету промпта.

Retrieval: пошук по документах

Тема окрема й велика, у мене є стаття про RAG і функціональні виклики та стаття про векторні бази. Тут — те, що стосується стека.

Конвеєр складається з трьох стадій, і кожна тестується окремо:

Індексація виконується окремо й заздалегідь; пошук і генерація — на кожен запит користувача.

Архітектур дві. Класична двокрокова: завжди шукати, потім відповідати — передбачувана затримка, менше викликів моделі, підходить для вузького корпусу, де пошук потрібен на кожне питання. Agentic RAG: пошук оформлений tool, і модель сама вирішує, чи потрібен він, коли і з яким запитом — змінна затримка, більше ходів, підходить для відкритих питань і кількох джерел. У першій версії agentic описаний як основний варіант.

@tool
def search_docs(query: str) -> str:
    """Пошук по документації продукту. Використовуй для питань про те, як продукт працює."""
    return "\n\n".join(d.page_content for d in retriever.invoke(query))

agent = create_agent(model="openai:gpt-5.5", tools=[search_docs])

Docstring тут працює як політика пошуку: він пояснює моделі, для яких питань цей корпус підходить.

Складові частини

Loaders повертають Document — текст плюс метадані. Метадані проставляються при завантаженні: джерело, розділ, час, орендар. Додати їх пізніше означає переіндексувати корпус.

Splitters:

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""],
)
chunks = splitter.split_documents(docs)

Розмір шматка — параметр якості пошуку. Надто дрібний втрачає контекст, через який шматок узагалі відповідав на питання; надто великий розмиває ембединг, і він починає слабко збігатися з усім підряд. Звичайний стартовий діапазон — 500–1500 символів із перекриттям 10–20%, далі добирається вимірюванням.

Embeddings:

from langchain.embeddings import init_embeddings

embeddings = init_embeddings("openai:text-embedding-3-small")

Одна й та сама модель і одна й та сама розмірність мають використовуватися і при індексації, і при запитах. Якщо змішати, відстані у сховищі втратять сенс, і помилки при цьому не буде.

Vector stores: InMemoryVectorStore для тестів, Chroma для локальної розробки, pgvector поруч із наявним Postgres, а також Pinecone, Qdrant і Weaviate окремими пакетами.

Retrievers:

retriever = store.as_retriever(search_kwargs={"k": 4})

# Різноманіття замість майже однакових шматків
retriever = store.as_retriever(search_type="mmr",
                               search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.5})

# Обмеження за метаданими — тут же фільтр за орендарем
docs = store.similarity_search(query, k=5, filter={"tenant_id": tenant})

Фільтр за орендарем і правами ставиться саме в retriever: прохання в промпті не показувати чуже механізмом ізоляції не є.

Заземлення відповіді

Пошук допомагає, якщо промпт змушує модель віддати перевагу знайденому тексту, а не власній пам’яті. Практика така: сказати явно, що відповідати треба лише за наданим контекстом, а якщо відповіді там немає — так і сказати; пронумерувати шматки й вимагати посилання на них, тоді непідтверджені твердження стають видимими; тримати знайдений текст у явно обмеженому блоці.

Останнє важливо і з погляду безпеки: знайдений текст — недовірене введення. Документ із фразою «ігноруй попередні інструкції» становить ін’єкцію промпта, тому його не вклеюють у системне повідомлення.

LangSmith: трасування, датасети та оцінка

Налагодження звичайного коду спирається на стектрейс. Неправильна відповідь моделі стектрейсу не лишає: виходить складний текст, за яким не видно, що модель бачила й чому вирішила саме так. Тому запис запусків — частина робочого процесу.

LangSmith вмикається змінними середовища й не потребує правок у коді:

LANGSMITH_TRACING=true
LANGSMITH_API_KEY=<ключ>
LANGSMITH_PROJECT=<проєкт>

Після цього кожен запуск LangChain і LangGraph потрапляє в трасування: nodes, виклики моделей, tools, токени, затримки. Імена змінних із колишнім префіксом LANGCHAIN_ більше не працюють.

До запусків варто додавати теги й метадані — за ними траси потім шукаються:

import langsmith as ls

with ls.tracing_context(enabled=True, project_name="quality-runs"):
    await agent.ainvoke(payload)

await agent.ainvoke(payload, config={"tags": ["production"], "metadata": {"user_id": user_id}})

Що ще є на платформі: панелі й сповіщення за метриками якості; правила, вебхуки та онлайн-оцінка просто на потоці продакшену; черги розмітки, де відповіді анотують вручну або збирають зворотний зв’язок користувачів; автоматичний аналіз повторюваних проблем у трасах.

Окремий блок — робота з промптами: playground для експериментів із промптами та конфігураціями моделей, версіонування з фіксацією змін, prompt hub із тегами й публічною частиною, доступ до промптів із коду. Сенс у тому, що промпт змінюється частіше за код і редагувати його можуть люди, які не працюють із репозиторієм.

Важливе міркування: траси містять промпти й відповіді, тобто вміст користувачів, який залишає ваш периметр. Передача його в сторонній сервіс — питання домовленостей і вимог до обробки даних.

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

Точка розширення для власного запису — callbacks: BaseCallbackHandler з подіями on_llm_start, on_llm_end, on_tool_start тощо. Готовий обробник для підрахунку токенів уже є.

Оцінка якості

Оцінка складається з трьох частин: dataset (входи й уявлення про правильний вихід), target (те, що запускається) та evaluators (оцінювальні функції). Результат прогону називається експериментом, і його можна порівнювати з попередніми.

from langsmith import Client

client = Client()
dataset = client.create_dataset(dataset_name="Sample dataset")

examples = [
    {"inputs": {"question": "У якій країні розташована Кіліманджаро?"},
     "outputs": {"answer": "Кіліманджаро розташована в Танзанії."}},
]
client.create_examples(dataset_id=dataset.id, examples=examples)

experiment_results = client.evaluate(
    target,
    data="Sample dataset",
    evaluators=[correctness_evaluator],
    experiment_prefix="first-eval",
    max_concurrency=2,
)

Evaluators бувають трьох видів. Детерміновані перевірки — схема валідна, є обов’язковий ідентифікатор, довжина в межах — застосовні всюди, де перевірка механічна. LLM-as-judge оцінює те, що механічно не перевіряється: достовірність, тон, дотримання заданого голосу; у судді є зміщення за позицією та за багатослівністю, і він сам по собі промпт, який може змінитися. Людська розмітка — найдорожча, її зазвичай лишають для вибірки, яку позначили автоматичні перевірки.

Прогоняти оцінку має сенс при зміні промпта, при зміні моделі та перед релізом — це три моменти, коли якість змінюється, не завалюючи жодного тесту. Dataset від прогону до прогону має бути один і той самий.

Є й оцінка траєкторії — перевірка того, який шлях agent пройшов, а не лише якою вийшла відповідь.

Тестування

Шарами, від дешевих до дорогих.

Без моделі. Розведення графа, маршрутизатори, reducers і логіка nodes — звичайні функції. Маршрутизатор є чистою функцією від state, і тестувати його зручніше напряму.

З підставною моделлю. Справжній граф, фальшива модель із заздалегідь заданими відповідями по порядку. Цього вистачає, щоб детерміновано прогнати маршрутизацію, streaming і шляхи помилок:

from langchain_core.language_models.fake_chat_models import FakeListChatModel

model = FakeListChatModel(responses=["CLARIFY", "фінальна відповідь"])

Для tools ту саму роль відіграє LLMToolEmulator.

Зі справжньою моделлю, за датасетом. Шар, який ловить регресії промптів — ситуації, коли всі тести зелені, а якість відповідей змінилася. Такі прогони зазвичай виносять із загального конвеєра збірки, бо вони повільні, платні й трохи плавають.

Чого підставні моделі не покажуть: чи справді промпт викликає потрібну поведінку і чи розбирається structured output за граматикою конкретного провайдера.

Запуск і викочування

Два способи довести graph до користувачів.

Готовий сервер

LangSmith Deployment — середовище виконання для агентів: воно бере на себе чергу завдань, стійке виконання, streaming і горизонтальне масштабування. Робоча частина називається Agent Server і побудована навколо трьох понять: assistant — граф із конкретною конфігурацією, thread — контекст стану, run — одне виконання. Є cron jobs для регулярних запусків, шар зберігання з checkpoints і store, API для створення runs, читання state і підключення до потоку, а також клієнтська бібліотека langgraph-sdk.

Розмістити це можна чотирма способами: повністю керована хмара; власний кластер Kubernetes разом із самостійно розгорнутим LangSmith; гібрид, де керівний шар у постачальника, а сервери й дані у вас; окремо розташований сервер у Docker або Kubernetes без керівного шару.

Конфігурація проєкту описується одним файлом:

{
  "dependencies": ["."],
  "graphs": { "agent": "./my_agent/agent.py:graph" },
  "env": "./.env",
  "python_version": "3.12"
}

Обов’язкових ключів два: dependencies і graphs — карта «ідентифікатор графа → файл і змінна». З необов’язкових є env, налаштування store (включно із семантичним пошуком і часом життя записів), налаштування checkpointer, параметри http на кшталт CORS і вимкнення маршрутів, webhooks, обробник auth, базовий образ і версія Python, а також закріплення версії серверного API.

Команди langgraph-cli:

pip install -U "langgraph-cli[inmem]"
langgraph new path/to/app --template new-langgraph-project-python
cd path/to/app && pip install -e .
langgraph dev

Після запуску доступні API на порту 2024, його документація та Studio — візуальний інтерфейс, який під’єднується до локально працюючого агента. У ньому видно кожен крок: які промпти пішли моделі, які виклики tools відбулися і з якими аргументами, що повернулося, скільки зайняло часу й токенів. Звідти ж можна перезапустити thread із будь-якого кроку й подивитися, як зміниться поведінка.

Різниця між langgraph dev і langgraph up: перший швидкий, працює без Docker і тримає стан у пам’яті — він для ітерацій; другий піднімає середовище, близьке до продакшену, зі справжнім Postgres.

Вбудований у свій сервіс

Другий спосіб: graph лишається бібліотечним викликом усередині вашого застосунку, API та сховище ваші, викочування йде вашим наявним конвеєром. Такий варіант підходить, коли agent — частина великої системи, у якій автентифікація, квоти та зовнішній контракт уже реалізовані в іншому місці.

Тоді чотири речі, які Agent Server дає сам, треба забезпечити самостійно:

• Компілювати графи один раз при старті застосунку. Компіляція на кожен запит додає затримку кожному ходу.
• Не блокувати цикл подій. Синхронний виклик усередині node зупиняє всі паралельні запити цього робочого процесу.
• Готувати схему checkpointer на викочуванні, а не при старті, і використовувати спільний із застосунком пул з’єднань.
• Зупинятися акуратно. При викочуванні контейнер гаситься, і запуск на середині обривається; request_drain() лишає checkpoint, з якого можна продовжити.

Перехід із версії 0.x

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

Видалено повністю: підтримка Python 3.9; прив’язка tools до моделі перед передаванням її агентові (тепер tools передаються параметром create_agent); structured output через прохання повернути JSON і розбір прози; .text() у вигляді методу; AgentExecutor, initialize_agent і набір готових типів агентів.

Переїхало в langchain-classic: старі chains, увесь колишній модуль retrievers, indexing API, hub, CacheBackedEmbeddings і реекспорти інтеграцій спільноти. Це не «застаріло, але працює» — імпорт з основного пакета завершується помилкою.

from langchain.chains import LLMChain
from langchain.retrievers import MultiQueryRetriever

# 1.x
from langchain_classic.chains import LLMChain
from langchain_classic.retrievers import MultiQueryRetriever

Оголошено застарілим, але працює: create_react_agent із langgraph.prebuilt; змінні середовища трасування з префіксом LANGCHAIN_; частина інтеграцій спільноти, у яких з’явилися окремі пакети. Застаріле й далі працює з попередженням на всій лінійці першої версії.

Відповідність старих аргументів новим зводиться до одного правила: те, що було аргументом конструктора, стало хуком middleware. prompt як функція став @dynamic_prompt, pre- і post-model хуки — before_model і after_model, вибір моделі на льоту — wrap_model_call, обробник помилок tool — wrap_tool_call, об’єкт ToolNode у списку — просто списком tools.

З боку LangGraph змін майже немає: state, nodes і edges, модель виконання, checkpoints, streaming та interrupts перейшли без правок.

Порядок міграції: спочатку Python 3.10, усе інше блокується цим; потім поставити langchain-classic і перенаправити імпорти — після цього код знову запускається, і у змінах стає видно, де «просто переїхало», а де потрібне перероблення; далі схеми state, агенти по одному, хуки, промпти, робота з messages і structured output разом із його шляхом помилки. Останнім кроком — прогін датасета: механічна міграція починає компілюватися раніше, ніж поведінка повертається до колишньої, а змінюється вона передусім в обробці промптів і structured output.

Усередині першої мажорної версії оновлення зазвичай зводяться до підняття нижньої межі та оновлення файлу блокування. Дивитися в примітки до випусків усе одно варто з двох причин: з’являються middleware, що замінюють написаний вручну код, і з’являються нові версії форматів — вони не ламають нічого, але й не дають нічого, доки споживача не переписано.

Поведінка, яку легко пропустити

Список речей, які не повідомляють про себе помилкою.

• Поле-список без reducer. Пишуть два nodes — збережеться значення одного з них.
• Node повернув state цілком замість часткового оновлення. Мутація проходить повз reducers.
• Command із goto плюс статичний edge з того самого node. Виконаються обидва напрямки.
• stream_mode="messages" без subgraphs=True. Токени вкладеного агента в потік не потраплять.
• Checkpointer без thread_id. Не зберігається нічого, помилки немає.
• Побічний ефект до interrupt. Node переграється з початку, дія повторюється.
• update_state пройшов через reducer. Додав там, де передбачалася заміна.
• InMemorySaver на продакшені. State зникає при перезапуску.
• Синхронний виклик моделі в асинхронному сервісі. Блокує всі паралельні запити процесу.
• Ідентифікатор користувача аргументом tool. Його значення обирає модель.
• Tool повернув Command без ToolMessage. Провайдер бачить виклик без результату.
• Різні моделі ембедингів при індексації та при пошуку. Відстані перестають щось означати.

Коротко

Що варто тримати в голові за підсумками.

Шари різняться областю відповідальності: LangChain — models, tools і цикл агента; LangGraph — state, виконання, пам’ять і streaming; Deep Agents — готова обв’язка для довгих завдань; LangSmith — трасування й оцінка; Agent Server — готове середовище запуску. Agent із LangChain є graph LangGraph, тому можливості рантайму доступні на будь-якому рівні.

Налаштування агента живе в middleware: шість хуків навколо циклу й каталог готових реалізацій.

У графі все тримається на reducers: накопичувальне поле потребує reducer, node повертає лише те, що змінив.

Пам’ять вмикається явно: checkpointer і thread_id — обидві частини одразу.

І загальна властивість стека: значна частина його відмов тиха. Токени, що не з’явилися, незбережений state, втрачені результати Send, відповідь за нерелевантними шматками тексту. Тому трасування запусків тут — базова частина роботи, а не додаткова можливість.