Wzmocnienie agenta LangChain w Pythonie za pomocą siedmiu wbudowanych middleware’ów
Dowiedz się, w jaki sposób middleware LangChain 1.0 dodaje funkcje podsumowywania, ograniczeń liczby wywołań, prób ponownych, alternatywnych modeli, usuwania danych osobowych oraz zatwierdzenia przez człowieka do agenta Gemini, nie wpływając na jego podstawową logikę.
Załadowanie agenta LangChain do odpowiadania na pytania w notatniku zajmuje zaledwie kilka minut. Stworzenie agenta, któremu można by zaufać w środowisku produkcyjnym, jest trudniejsze: nie może on wpadać w błędne pętle, które wyczerpią budżet API, przekazywać numer kart kredytowych klientów dostawcy modeli ani wysyłać e-maili bez zgody kogokolwiek. LangChain 1.0 rozwiązuje te problemy operacyjne za pomocą middleware – warstwy hooków wokół pętli agenta, którą konfiguruje się jako prosty list. Niniejszy przewodnik przedstawia model mentalny, następnie stosuje siedem różnych middleware do agenta w Pythonie opartego na Gemini, a na końcu pokazuje przykład własnego middleware, aby można było dokładnie zobaczyć, co zmienia każdy hook i kiedy należy go użyć.
Checkpointy wokół pętli agenta
Pomyśl o lotnisku. Celem jest przelecenie z jednego miasta do drugiego, ale wokół tej podstawowej czynności znajduje się szereg punktów kontrolnych: rejestracja potwierdza tożsamość, skanery bezpieczeństwa sprawdzają bagaże, brama kontrolna weryfikuje bilety, a odbiór bagażu następuje po lądowaniu. Żaden z tych elementów nie pilotuje samolotu, a pilot nie sprawdza bagaży. Każdy z tych elementów pełni jedną funkcję przed lub po głównej czynności.
Middleware stosuje tę samą zasadę w przypadku agentów. Istotą agenta jest pętla: wywołanie modelu, umożliwienie mu wyboru narzędzi, ich uruchomienie i powtórzenie tego procesu, aż model dostarczy ostateczną odpowiedź. Middleware wstawia punkty kontrolne wokół tej pętli, nie modyfikując jej:
- Usuwanie numerów kart przed dostarczeniem tekstu do LLM to skaner bezpieczeństwa.
- Wymaganie od osoby zatwierdzenia wysyłanego e-maila to brama kontrolna.
- Zatrzymanie działania po dziesięciu wywołaniach modelu w celu ograniczenia wydatków to wyłącznik bezpieczeństwa.
Jeśli pracujesz z TypeScript, towarzyszący artykuł na temat LangChain guardrails i middleware omawia te same koncepcje z perspektywy JavaScript; ten przewodnik koncentruje się natomiast na Pythonie i skupia się na konkretnych wbudowanych klasach.
Haki, których może używać middleware
Pętla udostępnia haki na każdym etapie, a middleware może być przypisany do jednego lub kilku z nich:
before_agentiafter_agentsą wykonywane raz – na samym początku i na końcu wezwania.before_modeliafter_modelsą uruchamiane za każdym razem, gdy pętla zamierza wywołać model lub właśnie go wywołała.wrap_model_calliwrap_tool_callotaczają sam wywołanie, dzięki czemu mogą je ponowić, zastąpić lub pobrać z pamięci podręcznej.
To jest cały model mentalny. Dołączasz middleware, przekazując listę do funkcji create_agent, jak w tym przykładzie łączącym redakcję e-maili, ograniczenie liczby połączeń oraz ponawianie prób używaniem narzędzi:
agent = create_agent(
model=model,
tools=[my_tool],
middleware=[
PIIMiddleware("email", strategy="redact"),
ModelCallLimitMiddleware(run_limit=5),
ToolRetryMiddleware(max_retries=3),
],
)
Kolejność w liście ma znaczenie. Middleware są stosowane sekwencyjnie, jak warstwy cebuli otaczające agenta, więc krok redakcji umieszczony na początku przetwarza surowy wprowadzony tekst przed wszystkimi innymi.
Ustawienia i podstawowy agent
Middleware wymagają wersji LangChain 1.0 lub nowszej, więc zainstaluj je za pomocą flagi -U w celu aktualizacji starszych wersji. Przykłady wykorzystują Gemini w ramach jego darmowej taryfy; możesz utworzyć klucz w Google AI Studio.
!pip install -qU langchain langchain-google-genai
Importuj klasę os oraz model rozmówowy Gemini:
import os
from langchain_google_genai import ChatGoogleGenerativeAI
Następnie konfiguruje się model. Ten fragment ustawia klucz bezpośrednio w kodzie wyłącznie w celach demonstracyjnych; w prawdziwym kodzie należy eksportować GOOGLE_API_KEY w środowisku lub załadować go z menedżera sekretów, zamiast umieszczać go w kodzie źródłowym. Temperatura ustawiona na zero zapewnia powtarzalność wyników:
os.environ["GOOGLE_API_KEY"] = "YOUR_GEMINI_API_KEY_HERE"
model = ChatGoogleGenerativeAI(model="gemini-3.5-flash-lite", temperature=0)
Przedmiotem każdego eksperymentu jest minimalny agent bez środowiska pośredniczącego. Potrzebuje on funkcji create_agent oraz dekoratora tool:
from langchain.agents import create_agent
from langchain_core.tools import tool
Zawiera jeden fałszywy narzędzie pogodowe, które zawsze informuje o słońcu, i jest wywoływane przy użyciu pojedynczej wiadomości od użytkownika:
@tool
def get_weather(city: str) -> str:
"""Get the current weather for a city."""
return f"The weather in {city} is 31°C and sunny."agent = create_agent(model=model, tools=[get_weather])result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the weather in Bengaluru?"}]}
)
print(result["messages"][-1].text)
Każdy poniższy rozdział dodaje kolejny punkt kontrolny do tego agenta.
1. SummarizationMiddleware dla ograniczonej pamięci
W długich rozmowach historia wiadomości stale się powiększa, aż przekroczy rozmiar okna kontekstowego, przy czym za każdy dodatkowy token pobierana jest opłata. SummarizationMiddleware sprawdza rozmiar tej historii w funkcji before_model; gdy przekroczy określony próg, sprowadza starsze wiadomości do streszczenia i zachowuje tylko najnowsze w niezmienionej formie.
from langchain.agents.middleware import SummarizationMiddleware
Poniższa konfiguracja wykorzystuje ten sam model do tworzenia streszczeń, aktywuje się po 10 wiadomościach i zachowuje ostatnie 4 bez zmian. Test tworzy fałszywą historię sześciu miast (dwanaście wiadomości), a następnie pyta, które miasto pojawiło się pierwsze:
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
SummarizationMiddleware(
model=model, # which LLM writes the summary
trigger=("messages", 10), # summarize when history hits 10 messages
keep=("messages", 4), # keep the 4 most recent messages intact
),
],
)# Simulate a long conversation
long_history = []
for city in ["Delhi", "Mumbai", "Chennai", "Kolkata", "Pune", "Jaipur"]:
long_history.append({"role": "user", "content": f"What's the weather in {city}?"})
long_history.append({"role": "assistant", "content": f"The weather in {city} is sunny."})
long_history.append({"role": "user", "content": "Which city did I ask about first?"})print("Messages passed IN:", len(long_history)) # 13result = agent.invoke({"messages": long_history})
print("Final answer:", result["messages"][-1].text)
print("Messages now in state:", len(result["messages"])) # 6
Wchodzi trzynaście wiadomości, a po ich przetworzeniu pozostaje sześć: streszczenie, cztery zachowane wiadomości oraz nowa odpowiedź. Model nadal odpowiada „Delhi”, ponieważ ta informacja znalazła się w streszczeniu. Liczenie wiadomości ułatwia zobaczenie tego zachowania w demonstracji, ale w środowisku produkcyjnym wyzwalacze oparte na tokenach, takie jak ("tokens", 3000), lub ułamek okna kontekstowego, np. ("fraction", 0.8), znacznie lepiej kontrolują rzeczywisty koszt i ustalają ograniczenia. Należy pamiętać, że streszczenia powodują straty informacji: dokładne liczby lub identyfikatory wspomniane na początku rozmowy mogą nie zostać zachowane.
2. Ograniczenia liczby wywołań jako mechanizmy ochronne przed wysokimi kosztami
Najdroższym błędem agenta jest niekontrolowany pętla, w której model i narzędzia ciągle do siebie wysyłają żądania, co skutkuje wydatkami trwającymi minuty, zanim ktoś to zauważy. Dwa middleware umożliwiają postawienie twardego ograniczenia w tym zakresie:
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
Tutaj model ma ograniczenie do trzech wywołań na jedną sesję, przy użyciu exit_behavior="end", dzięki czemu agent zatrzymuje się prawidłowo zamiast wywołać wyjątek, natomiast narzędzia mają ograniczenie do dwóch wywołań. Prompty celowo proszą o sześć miast po kolei:
agent = create_agent(
model=model,
tools=[get_weather],
middleware=[
# "end" = stop gracefully instead of raising an error
ModelCallLimitMiddleware(run_limit=3, exit_behavior="end"),
ToolCallLimitMiddleware(run_limit=2),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content":
"Get the weather for Delhi, Mumbai, Chennai, Kolkata, Pune and Jaipur one by one."}]}
)
print(result["messages"][-1].text)
Agent potrzebuje sześciu zapytań, osiąga swoje limity i kończy pracę w sposób uporządkowany, korzystając z uzyskanych częściowych wyników. Dostępny jest również parametr thread_limit, który umożliwia ograniczenie liczby wywołań w całym wątku rozmowy, a nie tylko w jednej sesji. Te dwa rozwiązania stanowią proste zabezpieczenie; należy ustawić limity wyżej niż te potrzebne do obsługi prawidłowych zapytań, aby aktywowały się jedynie w przypadku prawdziwych problemów.
3. ToolRetryMiddleware dla niestabilnych zależności
Prawdziwe narzędzia zawodzą: wywołania HTTP się timeoutują, a połączenia zostają przerwane. ToolRetryMiddleware wykorzystuje wrap_tool_call, aby wykrywać błędy i próbować je ponownie z eksponencjalnym opóźnieniem.
from langchain.agents.middleware import ToolRetryMiddleware
Aby to zademonstrować, narzędzie do obliczania cen akcji liczy swoje próby i wywołuje ConnectionError przy pierwszych dwóch, zanim uda mu się osiągnąć cel. Middleware umożliwia do trzech prób ponownych, zaczynając od jednosekundowego opóźnienia, które co raz się podwaja:
attempt_counter = {"count": 0}@tool
def flaky_stock_price(symbol: str) -> str:
"""Get the current stock price for a ticker symbol."""
attempt_counter["count"] += 1
print(f" [tool called — attempt #{attempt_counter['count']}]")
if attempt_counter["count"] < 3:
raise ConnectionError("API timeout — please retry")
return f"{symbol} is trading at ₹2,845.50"agent = create_agent(
model=model,
tools=[flaky_stock_price],
middleware=[
ToolRetryMiddleware(
max_retries=3, # retry a failed tool up to 3 times
initial_delay=1.0, # wait 1s before first retry
backoff_factor=2.0, # double the wait each time: 1s, 2s, 4s
),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content": "What's the price of RELIANCE stock?"}]}
)
print(result["messages"][-1].text)
Narzędzie zawodzi dwa razy, middleware czeka i próbuje ponownie, a trzecia próba się udaje. Z perspektywy agenta nic nie poszło nie tak. Ponawianie prób jest bezpieczne tylko w przypadku operacji idempotentnych, takich jak odczyty; ponawianie prób z narzędziem, które pobiera pieniądze z karty lub wysyła wiadomość, może powtórzyć efekt uboczny.
4. ModelFallbackMiddleware w przypadku awarii dostawcy
To samo podejście oparte na odporności może chronić wywołanie modelu. Gdy główny model zawodzi – na przykład z powodu ograniczeń szybkości lub awarii – ten middleware próbuje ponownie wysłać żądanie do modeli zapasowych w kolejności, w jakiej je zdefiniowano:
from langchain.agents.middleware import ModelFallbackMiddleware
W przykładzie jako główny model używany jest lżejszy model Gemini, a jako zapasowy dodany jest drugi model Gemini:
backup_model = ChatGoogleGenerativeAI(model="gemini-3.5-flash", temperature=0)agent = create_agent(
model=model, # primary: gemini-3.5-flash-lite
tools=[get_weather],
middleware=[ModelFallbackMiddleware(backup_model)],
)
Dopóki główny model funkcjonuje prawidłowo, nic nie zauważysz, co jest właśnie celem. Problem pojawia się dopiero wtedy, gdy u dostawcy wystąpią trudności. Aby uzyskać silniejszą ochronę, rozważ użycie modelu zapasowego od innego dostawcy, ponieważ awaria często wpływa na wszystkie modele obsługiwane przez tę samą API.
5. PIIMiddleware dla danych wrażliwych
Często w ogóle nie chcesz, aby e-maile, numery kart lub adresy IP były wysyłane do dostawcy modeli. PIIMiddleware skanuje tekst w funkcji before_model, zanim model go zobaczy, i stosuje jedną z czterech strategii: redact, mask, hash lub block.
from langchain.agents.middleware import PIIMiddleware
Ten agent nie posiada żadnych narzędzi. Zastępuje adresy e-mail całkowicie zmiennikiem i maskuje numery kart tak, aby pozostały tylko ostatnie cztery cyfry; obie te zasady są stosowane do danych wprowadzonych przez użytkownika:
agent = create_agent(
model=model,
tools=[],
middleware=[
# Replace emails entirely with [REDACTED_EMAIL]
PIIMiddleware("email", strategy="redact", apply_to_input=True),
# Mask credit cards — keeps last 4 digits
PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
],
)result = agent.invoke(
{"messages": [{"role": "user", "content":
"Draft a support reply to priya.sharma@example.com confirming her card "
"4111-1111-1111-1234 was not charged."}]}
)
print(result["messages"][-1].text)
Model tworzy swoją odpowiedź, nie otrzymując nigdy prawdziwego adresu ani pełnego numeru karty. Możesz również zarejestrować własny typ PII ze swoją regularną wyrażeniem, na przykład aby zablokować każdy tekst przypominający klucz API wewnętrzny:
PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block")
Rozpoznawanie oparte na wzorcach wykrywa poprawnie sformatowane wartości, a nie każdą kreatywną pisownię, dlatego należy je traktować jako jedną warstwę ochrony, a nie gwarancję zgodności.
6. HumanInTheLoopMiddleware do bram zatwierdzania
Część działań ma zbyt duże konsekwencje, by można je było wykonywać w pełnej autonomii: wysyłanie e-maili, usuwanie rekordów lub dokonywanie płatności. HumanInTheLoopMiddleware zatrzymuje agenta tuż przed uruchomieniem narzędzia wrażliwego, czeka na decyzję osoby i dopiero potem kontynuuje.
To działa inaczej niż poprzednie middleware. Zatrzymany agent nie jest zamykany. Checkpointer zachowuje jego pełny stan, a identyfikator wątku służy jako klucz do późniejszego odnalezienia i wznowienia tej pracy. Osoba sprawdzająca może zatwierdzić żądanie, edytować jego argumenty lub je odrzucić.
Importy dostarczają middleware, wewnętrznego kontrolera stanu z LangGraph oraz typu Command używanego do wznowienia działania:
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
Podany agent posiada narzędzie send_email, oznacza je jako do przerwania i przechowuje stan wstrzymania w InMemorySaver. Pierwsze wywołanie, na wątku demo-1, prosi agenta o wysłanie e-maila do menedżera:
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email to the given recipient."""
return f"Email sent to {to} with subject '{subject}'"agent = create_agent(
model=model,
tools=[send_email],
middleware=[
# Pause and ask a human whenever the agent wants to call send_email
HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),
],
checkpointer=InMemorySaver(), # where the paused state is saved
)config = {"configurable": {"thread_id": "demo-1"}}# Step 1: run — the agent PAUSES before sending
result = agent.invoke(
{"messages": [{"role": "user", "content":
"Send an email to boss@company.com saying the report is ready."}]},
config,
)
print("Agent paused! It wants to run:")
print(result["__interrupt__"])
Wykonywanie zostaje zatrzymane przed uruchomieniem narzędzia. Współrzędna __interrupt__ w wyniku pokazuje oczekujące zapytanie wraz z adresatem, tematem i treścią, przy czym nic nie zostało wysłane. Zatwierdzenie jest wysyłane jako Command na tym samym wątku, co wznowia wstrzymane działanie:
# Step 2: approve and resume
result = agent.invoke(
Command(resume={"decisions": [{"type": "approve"}]}),
config, # same thread_id -> resumes the paused run
)
print(result["messages"][-1].text)
Zamiast approve można wysłać reject wraz z powodem lub edit z zmienionymi argumentami. W rzeczywistej aplikacji w tym momencie wyświetlana byłaby strona z potwierdzeniem. Dwie praktyczne uwagi: InMemorySaver traci stan po ponownym uruchomieniu procesu, więc systemy produkcyjne wymagają trwałego wskaźnika stanu; ponadto format danych przy wznowieniu działania zmienił się pomiędzy wersjami LangChain, więc należy to sprawdzić w dokumencie referencyjnym middleware dla używanej wersji.
7. Tworzenie własnego middleware za pomocą dekoratora
Gdy żaden wbudowany rozwiązanie nie pasuje, prosty middleware jest łatwy do stworzenia, ponieważ każdy hook ma odpowiadający mu dekorator. Potrzebne są importy dekoratora before_model oraz typu AgentState:
from langchain.agents.middleware import before_model, AgentState
To przykład pokazuje, ile wiadomości ma zostać wysłanych przy każdym wywołaniu modelu, i jest dodawany do listy tak jak każdy gotowy middleware:
@before_model
def log_before_model(state: AgentState, runtime) -> None:
print(f" [middleware] Calling model with {len(state['messages'])} messages")
# Returning None = observe only.
# Returning a dict would UPDATE the agent's state (e.g., trim messages)
return Noneagent = create_agent(
model=model,
tools=[get_weather],
middleware=[log_before_model], # plugs in like any prebuilt middleware
)
Wartość zwracana jest kluczowym elementem projektu. Zwrócenie None oznacza, że middleware jedynie obserwuje. Zwrócenie słownika aktualizuje stan agenta, co umożliwia filtrowanie wiadomości, wstawianie kontekstu lub stosowanie niestandardowych ograniczeń. Dla innych punktów interakcji istnieją również dekoratory: @before_agent, @after_model, @wrap_model_call, @wrap_tool_call oraz @dynamic_prompt do tworzenia promptów systemowych w czasie wykonywania.
Główne wnioski
- Middleware oddziela kwestie operacyjne od logiki agenta: pętla pozostaje niezmieniona, podczas gdy punkty kontrolne są umieszczane wokół niej jako zwykła lista przekazywana do
create_agent.
TodoListMiddleware, LLMToolSelectorMiddleware i ContextEditingMiddleware, jest opisany w oficjalnej dokumentacji.Pozycje pokrewne
- Od chatbota z jednym węzłem do agenta opartego na MCP w LangGraph — Buduj aplikację LangGraph warstwa po warstwie: stan i reduktory, krawędzie, pętle narzędziowe, wątki z punktami kontrolnymi, trzy tryby strumieniowania oraz narzędzia dostarczane przez MCP.
- Hybrydowa pamięć agenta: fuzja BM25 i wyszukiwania wektorowego z RRF w Pythonie — Dowiedz się, dlaczego czyste wyszukiwanie wektorowe nie nadaje się jako pamięć agenta, jak Reciprocal Rank Fusion łączy wyniki BM25 i gęste w Pythonie oraz kiedy podsumowania GraphRAG są przydatne.