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. За скриптове и notebook-ове става всяка. За уеб услуга разликата е принципна: синхронното извикване блокира цикъла на събитията изцяло и докато чака отговора на модела, всички паралелни заявки стоят. В услуга по пътя на заявката се използват само асинхронните форми.

Има още 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      # {'gpt-5.5': {'input_tokens': 8, 'output_tokens': 10, ...}}

Резултатите в callback са разложени по имена на модели, така че прогон през няколко модела ще се върне разбит по тях; числата за един отговор са в самия отговор, в полето му 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 в промпта и разбери прозата“ във версия 1 е премахнат като ненадежден и не бива да се връща обратно на ръка: парсирането по схема дава грешка веднага, а парсирането на проза се чупи тихо и проличава по-надолу в кода.

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, 'image_inputs': 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 не се проваля мълчаливо – още първото извикване вдига ValueError със списък на очакваните ключове на конфигурацията, така че подсказката е направо в текста на грешката.

InMemorySaver живее до рестарта на процеса и е предназначен за разработка; за продукция има Postgres.

Recursion limit

Цикълът е ограничен по броя на super-steps и таванът по подразбиране е висок: create_agent се сглобява с 9999, голият граф – с 10007. Това е застраховка срещу безкраен цикъл, а не бюджет: агент, заседнал на зле описан tool, ще успее да изгори хиляди извиквания на модела, преди да се вдигне изключение – затова разумен лимит за вашата задача е добре да се зададе явно.

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 от евтин модел, когато са станали много. FilesystemFileSearchMiddleware – търсене във файлове. ShellToolMiddleware – постоянна сесия на командната обвивка. LLMToolEmulator – емулатор на изпълнението на tools с модел, за тестове. Още три идват не със самия LangChain, а с пакета deepagents и се импортират оттам: SubAgentMiddleware – делегиране на подзадачи към изолирани subagents, FilesystemMiddleware – файлова система върху подключаем бекенд, RubricMiddleware – самопроверка по зададени критерии с итерации.

Специфични за провайдърите. Кеширане на промпта и вградени 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, които пишат в него в една паралелна стъпка, не разиграват тиха надпревара: правило за сливане на две обновявания канал няма и графът спира с InvalidUpdateError – „Can receive only one value per step“. Всяко натрупващо поле трябва да има reducer.

За messages има готов add_messages: той добавя новите, отхвърля дубликатите по идентификатор и превръща речниците в обекти messages. Има и готов клас MessagesState, който го включва веднага.

Отделна грешка е да се върне от node цялото state вместо онова, което се е променило. Никакви reducers не го заобикалят и точно в това е бедата: всичко върнато минава през тях още веднъж, натрупващото поле идва заедно с вече събрания списък, operator.add дописва този списък към самия себе си – и историята мълчаливо се удвоява.

def good(state: State) -> dict:
    return {"answer": "..."}          # само онова, което се е променило

def bad(state: State) -> State:
    state["answer"] = "..."           # цялото state пак минава през 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, обработчиците пишат в един super-step и графът спира с InvalidUpdateError, вместо да ги слее.

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)                                  # продължаване по новия клон

Извикване с 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) – и ключ вътре в него, който се подава с отделен аргумент. Namespace служи и за граница на изолация: сглобява се от доверения context, а не от текст, дошъл от модела, защото namespace без идентификатор на потребител или наемател означава, че данните се виждат от всички.

Ако на store се зададе функция за ембединги и размерност чрез IndexConfig, търсенето ще стане семантично вместо точно. Продукционната реализация е PostgresStore.

В графа до store се стига през runtime, а не през глобална променлива:

from langgraph.runtime import 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)

Когато моделът поиска за един ход няколко защитени 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 и моделът сам решава дали е нужно, кога и с каква заявка – променливо забавяне, повече ходове, подходящо е за отворени въпроси и няколко източника. Във версия 1 като основен вариант е описан 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_ продължават да работят: клиентът първо гледа LANGSMITH_, а после се връща към тях. Те са обявени за остарели и при нови инсталации е добре да се взема новият префикс, но ако бъдат махнати от работещо внедряване, трасирането мълчаливо ще се изключи.

Към стартиранията е добре да се добавят тагове и метаданни – по тях после се търсят трасетата:

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() като метод.

Преместено в langchain-classic: старите chains, целият предишен модул retrievers, indexing API, hub, CacheBackedEmbeddings, реекспортите на интеграции на общността – и AgentExecutor с initialize_agent и набора от готови типове агенти. Това не е „остаряло, но работи“ – импортът от основния пакет завършва с грешка, но самият код е жив, така че агент от версия 0.x отново се стартира от пренаписан импорт, а не от пренаписан код.

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.

Вътре в първата версия обновяванията обикновено се свеждат до повдигане на долната граница и обновяване на lock файла. Да се гледа в бележките към изданията все пак си струва по две причини: появяват се middleware, които заместват писан на ръка код, и се появяват нови версии на формати – те не чупят нищо, но и нищо не дават, докато консуматорът не е пренаписан.

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

Списък от неща, които не се обявяват с грешка.

• Поле-списък без reducer. Два nodes пишат на различни стъпки – втората стойност мълчаливо измества първата; вътре в една паралелна стъпка това, напротив, е шумен InvalidUpdateError.
• Node върна цялото state вместо частично обновяване. Цялото съдържание минава през reducers втори път и натрупващите полета се удвояват.
• Command с goto плюс статичен edge от същия node. Ще се изпълнят и двете посоки.
• stream_mode="messages" без subgraphs=True. Токените на вложения агент няма да попаднат в потока.
• InMemorySaver, останал в продукция. Threads продължават правилно до рестарта на процеса, след него историята просто я няма.
• Страничен ефект преди interrupt. Node се преиграва от началото, действието се повтаря.
• update_state е минало през reducer. Добавило е там, където се е предполагала замяна.
• Синхронно извикване на модел в асинхронна услуга. Блокира всички паралелни заявки на процеса.
• Идентификатор на потребител като аргумент на 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, мълчаливо удвоила се история заради връщане на цялото state, отговор по нерелевантни парчета текст. Затова трасирането на стартиранията тук е базова част от работата, а не допълнителна възможност.