---
title: "LangChain и LangGraph – агенти, графи и цялата инфраструктура около тях"
url: "https://romankryvolapov.com/bg/langchain-langgraph/"
description: "Как е устроен стекът LangChain 1.x и LangGraph 1.x: models и messages, tools, create_agent и middleware, StateGraph с reducers, checkpointers и store, streaming, human-in-the-loop, LangSmith, Agent Server и langgraph CLI, миграция от 0.x."
language: bg
published: 2026-08-06
updated: 2026-08-31
---
**Здравейте!**
Езиковият модел умее точно едно нещо: да получи текст и да върне текст. Нищо повече. Не помни за какво сте говорили преди минута и не може сам да отиде в база данни или в интернет.
Да вземем прост пример: потребителят пита асистента какво е времето в момента. Моделът не го знае. За да може да отговори, трябва да се направи следното:
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** – потокова доставка: токените и събитията идват по мере появяването им.
## От какво се състои екосистемата
Под една марка се издават няколко продукта с близки имена. Ще ги подредим по рафтовете.
| Какво | Какво представлява | С какво се занимава |
|---|---|---|
| **LangChain** | Фреймуърк | Models, messages, промптове, tools, агентен цикъл |
| **LangGraph** | Рантайм под него | State, nodes и edges, checkpoints, streaming, устойчиво изпълнение |
| **Deep Agents** | Обвивка отгоре | Планиране, виртуална файлова система, subagents, компресиране на контекста |
| **LangSmith** | Платформа | Трасиране на стартиранията, датасети, оценка на качеството, работа с промптове |
| **LangSmith Deployment** | Хостинг | Готов сървър за стартиране на графи, с threads, опашка от задачи и API |
Слоевете зависят отдолу нагоре:
Функцията `create_agent` от LangChain връща **компилиран graph на LangGraph** – тоест всичко, което рантаймът умее да прави с граф, го умее и с агент: да пази state, да стриймва, да го поставя на пауза, да го влага в друг граф като обикновен node. Не е нужно да избирате между agent и graph, това е едно и също на различна височина на абстракция.
### Пакети
Библиотеката е разделена на много пакети; не е нужно да се инсталира всичко.
| Пакет | Какво има вътре |
|---|---|
| `langchain-core` | Базови типове: messages, блокове със съдържание, tools, шаблони за промптове, интерфейсът `Runnable` |
| `langchain` | Agents, middleware и удобни пространства от имена над ядрото |
| `langgraph` | Рантайм: state, graph, изпълнение, streaming |
| `langgraph-checkpoint` | Интерфейси на checkpointers и реализация в паметта; идва заедно с рантайма |
| `langgraph-checkpoint-postgres` | Checkpointer и store върху Postgres |
| `langgraph-checkpoint-sqlite` | Checkpointer върху SQLite за локална разработка |
| `langgraph-prebuilt` | Готови компоненти на графа, включително `ToolNode` |
| `langgraph-cli` | Команден ред: шаблон на проект, локален сървър, билд на образ, внедряване |
| `langgraph-sdk` | Клиент към внедрен Agent Server |
| `langchain-text-splitters` | Нарязване на документи на парчета за търсене |
| `langchain-community` | Дълга опашка от интеграции, които нямат отделен пакет |
| `langchain-classic` | Наследство от предишната версия: старите chains, старите retrievers, `AgentExecutor`, indexing API, hub |
| `langsmith` | Трасиране, датасети, оценка |
| `deepagents` | Готова обвивка за дълги задачи |
### Провайдъри на модели
Самата библиотека не е вързана за нито един вендор: провайдърът се свързва с отделен пакет, а превключването между тях е смяна на реда за инициализация, а не пренаписване на код. Пакетите се версионират независимо от ядрото.
| Вендор | Пакет | Низ на модела |
|---|---|---|
| OpenAI | `langchain-openai` | `openai:gpt-5.5` |
| Anthropic | `langchain-anthropic` | `anthropic:claude-sonnet-4-6` |
| Google Gemini | `langchain-google-genai` | `google_genai:gemini-2.5-flash-lite` |
| AWS Bedrock | `langchain-aws` | `bedrock_converse:us.anthropic.claude-sonnet-4-6` |
| Azure AI | `langchain-azure-ai` | внедряване в Azure |
| Mistral | `langchain-mistralai` | `mistralai:...` |
| Groq | `langchain-groq` | `groq:...` |
| Cohere | `langchain-cohere` | `cohere:...` |
| Hugging Face | `langchain-huggingface` | идентификатор на модела в хъба |
| Ollama, локални модели | `langchain-ollama` | `ollama:...` |
Низ от вида `"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` **не следва** семантичното версиониране и обикновено се фиксира към конкретна минорна серия.
### Кой слой да изберете за задачата
Проверява се отгоре надолу и се спира на първото съвпадение:
Последната точка е приложима по-често, отколкото изглежда: агентният цикъл около едно извикване добавя и токени, и забавяне.
## Инсталиране и първо извикване
```bash
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"` се разрешава в конкретна интеграция, но самият той не я инсталира – това е търсене сред вече инсталираното.
```python
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
| Параметър | Какво прави |
|---|---|
| `temperature` | Разсейване на отговорите. При разсъждаващите модели често не е достъпен – част от тях го отхвърлят |
| `max_tokens` | Горна граница на дължината на отговора |
| `timeout` | Ограничение на времето за заявка. Без него продължителността на отговора определя провайдърът |
| `max_retries` | Повторни опити от страна на клиента, по подразбиране шест |
| `rate_limiter` | Ограничител на честотата на заявките от страна на клиента |
| `reasoning_effort` | Дълбочина на разсъждението при моделите, които умеят да разсъждават: ниска, средна, висока |
За `rate_limiter` има нюанс: вградената реализация `InMemoryRateLimiter` работи **в рамките на един процес**. При пет реплики на услугата реалната честота ще е пет пъти по-голяма от зададената, а за общ за клъстера лимит е нужен общ ограничител.
За `reasoning_effort`: при разсъждаващите модели дълбочината на размисъла се задава с този параметър. Инструкцията „мисли стъпка по стъпка“ в промпта работи успоредно със собствения механизъм на разсъждение на модела, а не го допълва. Темата за промптовете е голяма, за нея имам [отделна статия](/bg/prompt-engineering/).
### Fallbacks – резервни модели
```python
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`.
```python
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` ръчно, е вързан за конкретен вендор.
### Мултимодален вход
```python
from langchain.messages import HumanMessage
message = HumanMessage(content=[
{"type": "text", "text": "Какво има на тази картинка?"},
{"type": "image", "url": "https://example.com/photo.png"},
])
```
### Подрязване на историята
Диалогът расте, контекстният прозорец е краен. Прост начин – детерминирано подрязване:
```python
from langchain.messages import trim_messages
trimmed = trim_messages(messages, max_tokens=8000, strategy="last", token_counter=model)
```
По-умен начин е резюмирането на старата част от историята, за agents то е достъпно като готов middleware. Разликата е, че подрязването е предвидимо и губи старото изцяло, а резюмирането запазва смисъла, но пренаписва онова, което асистентът помни.
### Отчитане на токените
```python
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`. Това е основа за изчисляване на разходите и потребителските квоти.
## Промптове
```python
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "Помагаш да се разберат логовете от билда. Отговаряй кратко."),
("human", "{question}"),
])
chain = prompt | model
```
`ChatPromptTemplate` сглобява списък от messages с подставяне на променливи. Ролите при това се запазват поотделно – а това е съществено: провайдърът обработва системната инструкция и потребителския вход по различен начин, те имат различна тежест и различно ниво на доверие. Ако ги слепите в един низ, текстът на потребителя ще се окаже там, където моделът очаква вашите инструкции – това е и загуба на качество, и отворен път за инжектиране в промпта.
Вертикалната черта е композиция: резултатът от левия обект се предава на десния. Получилата се верига сама умее `ainvoke`, `astream` и `abatch` и може да бъде част от нещо по-голямо.
## Structured output – отговор по схема
Често от модела е нужен не текст, а данни: категория на запитването, извлечени полета, решение за маршрут.
```python
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 е функция от вашия код, чието описание отива към модела заедно с промпта. Моделът сам не я изпълнява: той връща заявка за извикване с аргументи, извикването го изпълнява вашият код, резултатът се връща в модела.
```python
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.** Част от провайдърите отхвърлят други форми.
Когато анотациите не стигат – нужен е изброим набор от стойности, описание за всяко поле или валидация – се задава явна схема на аргументите:
```python
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` – фреймуъркът ще го подаде сам, **и в схемата за модела той няма да попадне**.
```python
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
| Върната стойност | Резултат |
|---|---|
| Низ | Става съдържание на `ToolMessage` |
| Речник или обект | Сериализира се в message, моделът чете полетата |
| Списък от блокове със съдържание | Мултимодален резултат – например текст заедно с картинка |
| `Command` | Обновява state на агента заедно с отговора на извикването |
| Каквото и да е при `return_direct=True` | Отива към потребителя без още един ход на модела |
```python
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 може да съобщава за хода на работата – това ще попадне в потока от събития:
```python
@tool
def index_documents(folder: str, runtime: ToolRuntime) -> str:
"""Индексира папка с документи."""
writer = runtime.stream_writer
writer({"type": "progress", "done": 0, "total": 100})
...
```
### Грешки на tools
По подразбиране изключение от tool прекратява цялото стартиране. За част от случаите това не е онова, което е нужно: неуспешното търсене е информация, с която моделът може да работи по-нататък.
```python
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
```python
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
| Параметър | Какво приема |
|---|---|
| `model` | Низ `provider:model` или готов обект на модела |
| `tools` | Списък с tools. Празен списък е допустим – тогава agent се изражда в едно извикване на модела с обвивка |
| `system_prompt` | Низ или `SystemMessage`; стойността е статична |
| `response_format` | Схема на отговора; резултатът идва в `structured_response` |
| `middleware` | Списък с middleware |
| `context_schema` | Описание на неизменяемите данни за стартирането: потребител, наемател, флагове |
| `state_schema` | Допълнителни полета на state, общи за агента и неговите tools |
| `checkpointer` | Запазване на state; без него няма памет между ходовете |
| `store` | Дългосрочна памет над всички threads |
| `name` | Име, под което agent се вижда, когато стане node или subagent |
В примерите моделът се задава с низ, а в работния код обикновено се взема от конфигурацията заедно с таймаута, повторните опити и веригата от fallbacks.
### Резултатът е речник
```python
result["messages"][-1].content # текст на отговора
result["structured_response"] # обект по схемата, ако е зададен response_format
```
Връща се финалното state, а не message, затова достъп от вида `result.content` ще даде `AttributeError`.
### Response format – схема вътре в цикъла
```python
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 – възможности на модела
```python
model.profile
# {'max_input_tokens': 400000, 'tool_calling': True, 'reasoning_output': True, 'image_inputs': True, ...}
```
Описание на възможностите на конкретния модел. По него може да се провери поддръжката на tools, преди да се закачат, или да се сверява с размера на контекста – вместо разклонения по името на модела. Същия профил използва и механизмът за избор на стратегия за `response_format`.
### Context и state – различни неща
| | Context | State |
|---|---|---|
| Живее | Едно стартиране, неизменяем | Променя се в хода на работата, попада в checkpoint |
| Подава се | С параметъра `context` при извикване | Вътре във входния речник |
| Съхранява | Потребител, наемател, флагове, дескриптори | Messages, натрупани данни, броячи |
```python
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 се запазва и се преиграва, затова при възобновяване на стар диалог в новото стартиране би попаднал вчерашният потребител.
### Памет между ходовете
```python
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, ще успее да изгори хиляди извиквания на модела, преди да се вдигне изключение – затова разумен лимит за вашата задача е добре да се зададе явно.
```python
result = agent.invoke(payload, config={"recursion_limit": 40})
```
Това е защита от безкраен цикъл. Ако е нужен продуктов лимит – например не повече от пет търсения за диалог – за това има `ModelCallLimitMiddleware` и `ToolCallLimitMiddleware`: те приключват работата по нормалния ред, а не с изключение.
## Middleware – шест точки за настройка
Middleware е код, който се изпълнява около стъпките на агентния цикъл. Чрез него се настройва всичко: подмяна на модела в движение, ограничения, повторни опити, маскиране на лични данни, пауза за човек, сглобяване на динамичен промпт.
| Хук | Кога се изпълнява |
|---|---|
| `before_agent` | Веднъж, преди началото на цикъла |
| `before_model` | Преди всяко извикване на model |
| `wrap_model_call` | Около всяко извикване на model |
| `wrap_tool_call` | Около всяко извикване на tool |
| `after_model` | След всеки отговор на model |
| `after_agent` | Веднъж, след завършването на цикъла |
На схемата това изглежда като слоеве около цикъла:
Формите са две. Node-style хуковете (`before_*`, `after_*`) получават state и връщат негово обновяване – те наблюдават и допълват. Wrap-style (`wrap_*`) стоят директно в пътя на извикването, получават заявка и функция-продължение: могат да повторят, да подменят параметри, да върнат свой резултат вместо истинския.
### Ред на изпълнение
За списък от три middleware редът е такъв:
Хуковете `before_*` вървят отгоре надолу, `after_*` – отдолу нагоре, `wrap_*` се влагат един в друг и първото middleware в списъка се оказва най-външното.
Оттук две практически следствия. Повторният опит, който трябва да обгърне всичко останало, се слага първи. Проверката, която е длъжна да види финалния отговор, също се слага по-близо до началото на списъка – на обратния път тя ще сработи последна.
### Как се пише
Един хук – с декоратор:
```python
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 – това е клас:
```python
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 или намерени документи, се сглобява с отделен хук:
```python
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, после се планира следващият такт. Преди стартиране графът се компилира.
```python
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.
```python
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` дописва този списък към самия себе си – и историята мълчаливо се удвоява.
```python
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 се тества лесно отделно.
### Маршрутизиране
```python
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`:
```python
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 – паралелно стартиране
Когато броят на разклоненията е известен само в рантайма:
```python
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 на друг граф:
```python
builder.add_node("research", research_graph)
```
Общите ключове на state се прехвърлят автоматично. Ако схемите на state са различни, е нужен node-адаптер, който преобразува данните на входа и на изхода.
### Управление на ниво node
```python
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())
```
| Настройка | За какво | Какво да се има предвид |
|---|---|---|
| `retry_policy` | Временни сривове: мрежа, отговори 429 и 5xx | Повторният опит рестартира **целия node от началото**, затова node трябва да е идемпотентен |
| `cache_policy` | Пропускане на повторно изчисление при еднакъв вход | Кешът се подава при компилация |
| `timeout` | Ограничение по общо време и по престой | Появи се във версия 1.2 |
| `error_handler` | Изпълнява се, когато повторните опити са изчерпани: компенсация, отмяна | Появи се във версия 1.2 |
От същия набор е и `RunControl.request_drain()`: кооперативно спиране, което моли стартирането да спре в най-близката безопасна точка и оставя checkpoint за продължаване. Това е нормалният начин да се преживее внедряването на нова версия или спирането на контейнер по средата на работата.
`recursion_limit` при графа е същият механизъм като при агента и се задава в конфигурацията на стартирането; различава се само стойността по подразбиране. Има и специално поле на state `RemainingSteps` – node може да го погледне и да свие работата предварително, вместо да се удари в лимита.
## Checkpointer и store: памет и устойчивост
Паметите са две и решават различни задачи.
| | Checkpointer | Store |
|---|---|---|
| Област на действие | Един thread, един диалог | Всички threads наведнъж |
| Съхранява | State на графа на всеки super-step | Произволни документи във ваш формат |
| Дава | Памет между ходовете, продължаване след срив, interrupts, превъртане | Предпочитания на потребителя, натрупани факти, общо знание |
| Ключ | `thread_id` плюс идентификатор на checkpoint | Кортеж namespace плюс ключ |
Нито едното, нито другото не е включено по подразбиране.
### Checkpointers
```python
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) # вижда първия ход
```
| Реализация | Пакет | Предназначение |
|---|---|---|
| `InMemorySaver` | `langgraph-checkpoint` | Тестове и разработка; изчезва при рестарт |
| `SqliteSaver` | `langgraph-checkpoint-sqlite` | Локална разработка със запазване между стартиранията |
| `PostgresSaver` / `AsyncPostgresSaver` | `langgraph-checkpoint-postgres` | Продукция; в асинхронна услуга – асинхронната форма |
Две експлоатационни подробности. Методът `.setup()`, който създава таблиците, е операция на ниво внедряване, а не на старта на приложението: промяната на схемата при старт при няколко реплики се изпълнява паралелно със самата себе си. И `thread_id` за Postgres се държи по-къс от 255 символа – колоната е ограничена.
### Преглед на историята и превъртане
```python
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`:
```python
from langgraph.types import Overwrite
await graph.aupdate_state(config, {"items": Overwrite(["C"])})
```
### Съхранение на checkpoints
Checkpoints се създават по един на super-step за всеки thread и сами не се изтриват. За дългоживеещ продукт това означава постоянен растеж на таблицата, затова политиката за съхранение – колко време thread остава възобновяем и какво изтрива останалото – е добре да се определи предварително.
### Store
```python
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, а не през глобална променлива:
```python
from langgraph.runtime import Runtime
def recall(state, runtime: Runtime):
item = runtime.store.get(("users", runtime.context.user_id), "preferences")
...
```
### Durability – режими на запис
Настройка за това колко често state се записва на диска. Подава се при извикване:
| Режим | Поведение | Какво се губи при срив |
|---|---|---|
| `"sync"` | Записва се преди началото на следващата стъпка | Нищо; най-бавният |
| `"async"` | Записва се паралелно със следващата стъпка | Малък прозорец – последният checkpoint |
| `"exit"` | Само при завършване: успех, грешка или interrupt | Всичко междинно; най-бързият |
```python
await graph.ainvoke(payload, config=config, durability="async")
```
Изборът зависи от цената на загубена стъпка: ако стъпките изпращат писма или извършват плащания, подхожда `"sync"`; за аналитично прогонване върху дълъг диалог е достатъчно `"exit"`.
### Преиграване при възобновяване
Общо правило на рантайма: възобновяването **изпълнява node от самото начало**, пропуска се само завършената работа, попаднала в checkpoint. Оттук две следствия:
- Страничните ефекти, изпълнени преди точката на спиране, ще се случат още веднъж.
- Недетерминираните стойности – време, случайни идентификатори – при преиграването ще излязат други. Те се изчисляват в отделен node, за да попадне стойността в checkpoint и да се подава оттам.
## Human-in-the-loop: пауза за човек
Понякога стартирането трябва да спре и да изчака човек: да потвърди изпращане, да поправи чернова, да отговори на уточняващ въпрос. За това има `interrupt`.
```python
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:
```python
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` пропуска извикването без въпроси, речник настройва кои решения са достъпни за проверяващия.
| Решение | Какво се случва |
|---|---|
| `approve` | Извикването се изпълнява както е предложено |
| `edit` | Изпълнява се с променени аргументи; в `edited_action` се подават и `name`, и `args` |
| `reject` | Не се изпълнява, а текстът на отказа отива към модела като резултат от извикването |
| `respond` | Текстът на човека се връща вместо резултата на tool |
```python
await agent.ainvoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)
```
Когато моделът поиска за един ход няколко защитени tools, те се представят заедно и списъкът с решения трябва да върви **в същия ред** като представените действия.
Interrupt може да се включва по условие – с параметъра `when`: функцията гледа аргументите на извикването и решава дали е нужно потвърждение. Така се проверяват само рисковите случаи, например запис извън работната директория, а останалите минават без въпроси.
### Странични ефекти и точка на спиране
Тук е същото правило за преиграване, но с нагледни последствия:
```python
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`. Подходящ е за повечето задачи.
| `stream_mode` | Какво връща |
|---|---|
| `values` | Цялото state след всяка стъпка |
| `updates` | Само онова, което е променил всеки node |
| `messages` | Токени на модела по мере постъпването им заедно с метаданни |
| `custom` | Онова, което node е написал сам през stream writer |
| `checkpoints` | Събития за запазване на state (нужен е checkpointer) |
| `tasks` | Стартове и завършвания на задачи с резултати и грешки (нужен е checkpointer) |
| `debug` | Checkpoints, tasks и метаданни заедно |
Режимите могат да се искат по няколко наведнъж, със списък.
### Версия на формата на потока
Форматът има версия и по подразбиране действа `v1`, където формата на отговора зависи от извикването: един режим връща сурови данни, няколко режима – кортежи `(mode, data)`, а включените subgraphs добавят в кортежа namespace.
Версия `v2` връща всичко еднообразно – речник `StreamPart` с полета `type`, `ns` и `data`:
```python
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`:
```python
classifier = model.with_config({"tags": ["nostream"]})
```
Тогава токените му изобщо не се излъчват.
### Subgraphs в потока
```python
async for chunk in graph.astream(payload, stream_mode="messages", subgraphs=True, version="v2"):
...
```
Без `subgraphs=True` токените, породени вътре във вложен граф – включително agent, поставен като node – не попадат в потока. Грешка при това няма, потокът просто излиза непълен. Полето `ns` във всеки фрагмент показва източника: празен кортеж за коренния граф, име на node с идентификатор на задачата за вложения.
### Custom events
```python
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` в графа съобщава за старт, за фрагменти и за завършване.
```python
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.
```python
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 делегира работата, без да знае нищо за вътрешното устройство.
```python
@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 се прехвърлят в двете посоки |
| Кой решава | Моделът – кога да делегира | Графът – детерминирано |
| Цена | Пълен агентен цикъл при всяко делегиране | Работата на един node |
| Видимост | Само в трасирането | На нарисуваната схема |
Определящото свойство е изолацията на контекста. Делегирането към subagent е уместно, когато подзадачата иначе би запълнила родителския контекст с подробности, които нататък не са нужни: четене на дълъг документ, шумно търсене. Subgraph е уместен, когато стъпката е част от процеса и нейното state е важно за следващите стъпки.
За цената: всеки subagent изпълнява свой цикъл със свой системен промпт и свои описания на tools, затова стартирането на пет subagents струва приблизително колкото пет агента, а не колкото един.
## Deep Agents
Готов комплект от middleware върху обикновен агент, предназначен за дълги задачи, които не се събират в един контекстен прозорец.
```python
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 и извикването на функции](/bg/rag-and-function-calling/) и [статия за векторните бази](/bg/vector-databases/). Тук – онова, което се отнася до стека.
Пайплайнът се състои от три стадия и всеки се тества отделно:
Индексирането се изпълнява отделно и предварително; търсенето и генерирането – при всяка заявка на потребителя.
Архитектурите са две. Класическа двустъпкова: винаги се търси, после се отговаря – предвидимо забавяне, по-малко извиквания на модела, подходяща е за тесен корпус, където търсенето е нужно при всеки въпрос. Agentic RAG: търсенето е оформено като tool и моделът сам решава дали е нужно, кога и с каква заявка – променливо забавяне, повече ходове, подходящо е за отворени въпроси и няколко източника. Във версия 1 като основен вариант е описан agentic.
```python
@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:**
```python
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:**
```python
from langchain.embeddings import init_embeddings
embeddings = init_embeddings("openai:text-embedding-3-small")
```
Един и същ модел и една и съща размерност трябва да се използват и при индексирането, и при заявките. Ако се смесят, разстоянията в хранилището губят смисъл и грешка при това няма.
**Vector stores:** `InMemoryVectorStore` за тестове, Chroma за локална разработка, pgvector до съществуващ Postgres, а също Pinecone, Qdrant и Weaviate с отделни пакети.
**Retrievers:**
```python
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 се включва с променливи на средата и не изисква промени в кода:
```bash
LANGSMITH_TRACING=true
LANGSMITH_API_KEY=<ключ>
LANGSMITH_PROJECT=<проект>
```
След това всяко стартиране на LangChain и LangGraph попада в трасирането: nodes, извиквания на модели, tools, токени, забавяния. Предишните имена с префикс `LANGCHAIN_` продължават да работят: клиентът първо гледа `LANGSMITH_`, а после се връща към тях. Те са обявени за остарели и при нови инсталации е добре да се взема новият префикс, но ако бъдат махнати от работещо внедряване, трасирането мълчаливо ще се изключи.
Към стартиранията е добре да се добавят тагове и метаданни – по тях после се търсят трасетата:
```python
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** (оценяващи функции). Резултатът от прогона се нарича експеримент и може да се сравнява с предишните.
```python
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 и пътищата на грешките:
```python
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 без управляващ слой.
Конфигурацията на проекта се описва с един файл:
```json title="langgraph.json"
{
"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`:
| Команда | Какво прави |
|---|---|
| `langgraph new` | Създава проект от шаблон |
| `langgraph dev` | Локален сървър с горещо презареждане, без Docker, порт 2024 |
| `langgraph up` | Вдига всичко локално в Docker заедно с Postgres |
| `langgraph build` | Билдва образа на сървъра |
| `langgraph deploy` | Билдва образа, изпраща го в регистъра и създава или обновява внедряването |
| `langgraph dockerfile` | Извежда Dockerfile без билд |
```bash
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 отново се стартира от пренаписан импорт, а не от пренаписан код.
```python
# 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, отговор по нерелевантни парчета текст. Затова трасирането на стартиранията тук е базова част от работата, а не допълнителна възможност.