Trzy zabezpieczenia middleware LangChain, które zapobiegają wyciekom tajemnic przez agenty

Wklejanie czegoś wrażliwego do okna czatu jest łatwe i łatwo o to zapomnieć. Być może jakiś plik trafił do ChatGPT w celach debugowania, a po wysłaniu zamarłeś, zastanawiając się, czy nie znajduje się w nim klucz API, hasło do bazy danych lub token klienta. Przeczytasz wiadomość ponownie, zamkniesz kartę i zdecydujesz, że prawdopodobnie wszystko w porządku. Prawdopodobnie to nie stanowi naruszenia zasad.

Agenty pogarszają sytuację, ponieważ robią coś podobnego automatycznie, wiele razy podczas jednej sesji i bez uprzedzenia. Daj im dostęp do bazy kodu w celu przeglądania, a w końcu otworzą plik .env zawierający coś w rodzaju OPENAI_API_KEY=sk-proj-.... Bez żadnych zabezpieczeń ten klucz trafia dosłownie do modelu językowego, pojawia się w logach śledzenia i zostaje zapisany w rekordach dostawcy. Czy trafi dalej, do zbioru danych do dopasowania modelu lub usługi logowania, której nie widzisz, z twojej strony jest to nie do ustalenia.

Kilka rund później ten sam agent uruchamia narzędzie do czyszczenia i usuwa katalog, o którym nigdy nie wspomniałeś, cicho zapisując delete_directory(path="./tmp_backups"), a następnie przechodzi dalej. Odkrywasz to wtedy, gdy zniknął plik, którego potrzebujesz.

Są to dwa odrębne błędy, bez żadnego baku ani dramatycznych halucynacji. Agent po prostu nie miał wyraźnej granicy pomiędzy tym, co może zrobić, a tym, co powinien zrobić. Rozwiązaniem jest middleware, a te same zasady leżą u podstaw mechanizmów ochronnych w narzędziach takich jak Cursor i GitHub Copilot. Oto jak można je włączyć do własnego agenta.

Czym jest middleware

Wyobraź sobie agenta jako linię produkcyjną. Dane przychodzą od ciebie, z plików oraz z wyników wyszukiwania, trafiają do modelu, który podejmuje decyzje, które następnie przeradzają się w działania. Middleware stanowi punkt kontrolny na tej linii – przerywa przepływ danych, wykonuje weryfikację lub transformację, a następnie przekazuje je bez zmian, modyfikuje je lub je zatrzymuje.

Lokalizacja ma kluczowe znaczenie. Jeśli to zbyt późno, dane poufne już przekroczyły określoną granicę. Jeśli za wcześnie, sprawdzasz niewłaściwe dane.

Tajne skanowanie Copilota odbywa się na etapie wyjściowym. Zanim sugestia pojawi się w twoim edytorze, analizator wzorców ją sprawdza i wszystko, co przypomina klucz zaczynający się od sk-, ghp_ lub AKIA, jest odrzucane. Funkcja Review Changes w Cursorze działa przed każdym zapisem pliku: wykonywanie zatrzymuje się całkowicie, pokazany zostaje raport różnic, a agent nie może kontynuować, dopóki nie zatwierdzisz zmian. Jeśli je odrzucisz, zapis nie nastąpi i agent poszuka innego rozwiązania.

Jeden mechanizm chroni to, co wychodzi z modelu, a drugi to, co model może zrobić. LangChain umożliwia stworzenie obu takich mechanizmów, a także ustawienie ograniczeń zasobowych.

Złapanie tajemnic: PIIMiddleware

Twój agent nigdy nie dotyka bezpośrednio systemu plików. Widzi tylko to, co trafia do jego okna kontekstowego, czyli to, co wpisujesz oraz to, co zwracają jego narzędzia. Dlatego dane wrażliwe mogą trafić do niego dokładnie na dwa sposoby: możesz je wkleić lub narzędzie może je odczytać z dysku, np. gdy skanowanie katalogu otwiera plik .env. W obu przypadkach dane przechodzą przez warstwę middleware, zanim model cokolwiek zobaczy, i to właśnie w tym punkcie działa PIIMiddleware.

Istnieje flaga apply_to_tool_results, której większość tutoriali używa bez wyjaśnienia. Jej domyślna wartość to False. Włącz tylko apply_to_input=True, aby zabezpieczyć się przed sytuacją, gdy wklejasz dane wrażliwe, natomiast sytuacja, gdy agent odczytuje plik, pozostaje nieskutecznie zabezpieczona. Agent z dostępem do plików chętnie pobra całego pliku .env jako wynik działania narzędzia, a model otrzymuje go.

Minimalne poprawne ustawienie wygląda tak:

from langchain.agents.middleware import PIIMiddleware

PIIMiddleware(
    "email",
    strategy="redact",            # replaces match with [REDACTED_EMAIL]
    apply_to_input=True,          # scans what you type
    apply_to_tool_results=True,   # scans what tools return - never skip this
)

Klucze API to te, które powodują rzeczywiste incydenty, a żaden detektor zapasów nie zna formatu twojego klucza, dlatego musisz dostarczyć wyrażenie regularne. To wyrażenie obejmuje powszechne prefiksy:

import re
API_KEY_PATTERN = r"(?:sk-|ghp_|AKIA)[a-zA-Z0-9]{20,48}"
# sk-   → OpenAI and Anthropic keys
# ghp_  → GitHub personal access tokens
# AKIA  → AWS access key IDs
PIIMiddleware(
    "api_key",
    detector=API_KEY_PATTERN,
    strategy="redact",
    apply_to_input=True,
    apply_to_tool_results=True,
)

Middleware oferuje cztery strategie:

• redact zastępuje dopasowanie wartością [REDACTED_FIELDNAME]. Nadaje się do logów, śladów oraz jednokierunkowych przepływów danych.
• mask zastępuje znaki gwiazdkami, ale zachowuje długość tekstu, co jest przydatne, gdy chce się pokazać, że dana wartość istnieje, bez ujawniania jej treści.
• block wywołuje wyjątek i kończy całe działanie. Należy go używać, gdy tolerancja jest zerowa, w środowiskach o wysokim poziomie bezpieczeństwa.
• hash zastępuje wartość deterministycznym haszem, co jest przydatne, gdy trzeba śledzić daną entytetę bez przechowywania jej wartości.

Większość ludzi uważa, że block to najbezpieczniejszy wybór, ale w przypadku wyników narzędzi zazwyczaj jest to błąd. Załóżmy, że agent przeczyta 30 plików podczas skanowania, a w jednym z nich znajduje się adres e-mail w komentarzu. Użycie block zakończy całą sesję. Funkcja redact usuwa tę wartość i pozwala na kontynuację pracy, co prawie zawsze jest tym, czego chcemy. Zachowaj block tylko w sytuacjach, gdy sama obecność danych wrażliwych oznacza, że agent dotarł tam, gdzie nie powinien.

Przetestuj się: jeśli ustawiono apply_to_input=True, a apply_to_tool_results pozostaje na swojej domyślnej wartości, która konkretna sytuacja pozostaje niezabezpieczona? Powinieneś móc odpowiedzieć w jednym zdaniu.

Zatrzymywanie cichych usuwań: HumanInTheLoopMiddleware

Z powrotem do usuniętego katalogu. HumanInTheLoopMiddleware to wzorzec Review Changes z Cursora przekształcony w prymityw kodu. Zanim uruchomi się jakikolwiek narzędzie niszczące lub modyfikujące, agent zawiesza działanie i zapisuje swój stan. Widzisz dokładnie, co zamierza zrobić, i musisz to zatwierdzić, zanim cokolwiek się wydarzy.

Zawieszanie to łatwa część. Przywracanie do pracy to trudniejsza część. Powrót do dokładnego stanu, w którym znajdował się agent, być może po kilku godzinach lub dniach oraz po ponownym uruchomieniu procesu, wymaga przechowywania stanu rozmowy, planowanej akcji oraz miejsca, w którym agent znajduje się w swoim rozumowaniu, w jakimś trwałym miejscu. Takie przechowywanie nazywane jest checkpointerem. Bez niego następuje przerwa, stan znika, a kolejne wezwanie rozpoczyna się od zera, więc mechanizm zatwierdzania po prostu kończy sesję.

Pauza następuje po tym, jak model zdecyduje, co robić, ale przed uruchomieniem jakiegokolwiek narzędzia. To jedyne możliwe okno czasowe, ponieważ decyzja jest widoczna i nie doszło do niczego nieodwracalnego.

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.sqlite import SqliteSaver

# The checkpointer is not optional. Without it, resume is impossible.
with SqliteSaver.from_conn_string("./review_agent.db") as checkpointer:
    agent = create_agent(
        model="anthropic:claude-sonnet-4-20250514",
        tools=[read_file, list_directory, write_file, search_codebase],
        middleware=[
            HumanInTheLoopMiddleware(
                interrupt_on={
                    "write_file": True,       # always pause before writing
                    "read_file": False,        # reading is safe - no pause needed
                    "list_directory": False,
                    "search_codebase": False,
                }
            ),
        ],
        checkpointer=checkpointer,            # saved to SQLite, persists across restarts
    )

Kontynuacja po zatwierdzeniu wygląda tak:

# First call — agent hits the interrupt at write_file and pauses
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Review and fix the config files"}]},
    config={"configurable": {"thread_id": "session-001"}}
    # thread_id ties the saved state to this specific session
)


# result.interrupted == True
# result.pending_tool_call == {"name": "write_file", "args": {"path": "src/config.py", ...}}
# You show this to the user and wait for approval
# User approves - resume the same thread
final_result = agent.invoke(
    Command(resume=True),
    config={"configurable": {"thread_id": "session-001"}}
    # same thread_id - loads state from the checkpointer and picks up where it stopped
)

Jedna pułapka może pochłonąć całe popołudnie. interrupt_on={"write_file": True} dopasowuje się do nazwy narzędzia jako dokładnej ciągu znaków. Jeśli twoje narzędzie faktycznie nazywa się file_write lub update_file, przerwanie nigdy nie następuje, a wywołanie przechodzi bez echa. Sprawdź logi śledzenia, aby poznać rzeczywistą nazwę podczas wywoływania narzędzia, i dopasuj to do niej.

Dwa limity dla dwóch trybów awarii

Każdy krok polega na tym, że agent wywołuje model, który wybiera działanie, a czasami wykorzystuje narzędzia; jedno wywołanie modelu może jednocześnie uruchomić kilka wywołań narzędzi. Są to odrębne zasoby, przy czym jeden nie wpływa na drugi.

ModelCallLimitMiddleware liczy wywołania modelu, natomiast ToolCallLimitMiddleware liczy wykonywania narzędzi. Każde z tych wywołań modelu, które przechodzą dalej w trzy równoległe wywołania narzędzi, daje łącznie 30 wykonań, a sam limit wywołań modelu nigdy by tego nie zauważył. Działa to również w drugą stronę – agent, który ciągle korzysta z własnego rozumowania, nieustannie wywołuje model bez używania żadnych narzędzi, więc licznik narzędzi pozostaje na zera i nigdy nie zostanie przekroczony.

from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware

ModelCallLimitMiddleware(
    max_calls=30,
    on_limit="raise",   # raises MaxCallsExceeded - catch this in your application
)
ToolCallLimitMiddleware(
    max_calls=60,
    on_limit="raise",
)

Niech oba będą używane, ponieważ każde z nich obejmuje coś, czego drugie nie może.

Jeśli chodzi o on_limit="raise" w porównaniu z on_limit="return": wybierz "return" w sesji interaktywnej, gdzie pomocne jest częściowe wynik, np. podsumowanie tego, co zostało już przejrzane. Wybierz "raise", gdy niekompletny wynik mógłby zakłócić dalsze działanie.

Należy uważać również na liczby. Limit max_calls=10 wydaje się bezpiecznym pułapem dla szybkiego testu, jednak agent z planem wieloetapowym może sam wykorzystać aż 8 połączeń już w pierwszej turze. Wtedy limit przerywa mu zadanie i zwraca skrócony wynik, przez co można stracić dwadzieścia minut, zanim logi to wyjaśnią. Rozwiązaniem nie jest obniżenie limitu, lecz ustawienie go na wartość rzeczywiście potrzebną do wykonania zadania.

Porządek to gwarancja bezpieczeństwa

Middleware działa w kolejności, w jakiej go umieścisz, a to nie jest błahe szczegół. Jeśli jakiś middleware przepisujący lub streszczający treść wiadomości znajduje się przed PIIMiddleware, kontrola PII skanuje tekst, który został już zmodyfikowany. Gdy zmiana spowodowała skompresowanie, parafrazowanie lub przebudowę treści tak, że twój wzorzec już się do niej nie odnosi, funkcja redakcji jej przeoczy – tajemnica znajduje się już w kontekście modelu, a proces redakcji odbywa się po powstaniu szkody.

Dla tej pakiety bezpieczeństwa kolejność jest następująca:

middleware=[
    # PIIMiddleware always first — it must see raw, untransformed data
    PIIMiddleware("api_key", detector=API_KEY_PATTERN, strategy="redact",
                  apply_to_input=True, apply_to_tool_results=True),
    PIIMiddleware("email", strategy="redact",
                  apply_to_input=True, apply_to_tool_results=True),

# Limits next - exact position within the group is flexible
    ModelCallLimitMiddleware(max_calls=30, on_limit="raise"),
    ToolCallLimitMiddleware(max_calls=60, on_limit="raise"),
    # Human-in-the-loop last in the safety group
    # (it fires in the after_model hook regardless of list position,
    # but last is a readable convention)
    HumanInTheLoopMiddleware(interrupt_on={"write_file": True}),
]

Traktuj listę jako łańcuch przetwarzania, a nie rejestr pluginów. Czwarty element nie otrzymuje oryginalnej wiadomości – otrzymuje to, co pozostawiły elementy od pierwszego do trzeciego. Przenieś PIIMiddleware poniżej wszystkiego, co manipuluje treścią wiadomości, a ta gwarancja po prostu zniknie.

Kolejna weryfikacja: co się stanie, jeśli PIIMiddleware znajdzie się po middleware’u, który podsumowuje historię rozmów? Jeśli potrafisz odpowiedzieć bez przewijania w górę, oznacza to, że zasada kolejności została spełniona.

Tworzenie agenta do przeglądania kodu

Teraz połącz to wszystko w agenta, który zachowuje się jak narzędzia komercyjne. Składa się on z czterech narzędzi: read_file, list_directory, write_file oraz search_codebase. Otaczają je trzy warstwy bezpieczeństwa: tajne dane nigdy nie trafiają do modelu, zapisy wymagają twojej aprobaty, a zużycie zasobów ma sztywny limit.

from langchain.agents import create_agent
from langchain.agents.middleware import (
    PIIMiddleware,
    HumanInTheLoopMiddleware,
    ModelCallLimitMiddleware,
    ToolCallLimitMiddleware,
)
from langgraph.checkpoint.sqlite import SqliteSaver

# Custom pattern for secrets common in codebases
API_KEY_PATTERN = r"(?:sk-|ghp_|AKIA)[a-zA-Z0-9]{20,48}"
# sk-   → OpenAI / Anthropic keys
# ghp_  → GitHub personal access tokens
# AKIA  → AWS access key IDs
with SqliteSaver.from_conn_string("./review_agent.db") as checkpointer:
    agent = create_agent(
        model="anthropic:claude-sonnet-4-20250514",
        tools=[read_file, list_directory, write_file, search_codebase],
        middleware=[
            # 1. Catch secrets before they reach the model - on both surfaces
            PIIMiddleware(
                "api_key",
                detector=API_KEY_PATTERN,
                strategy="redact",
                apply_to_input=True,
                apply_to_tool_results=True,  # this is the one that catches .env reads
            ),
            PIIMiddleware(
                "email",
                strategy="redact",
                apply_to_input=True,
                apply_to_tool_results=True,
            ),
            # 2. Hard resource limits - stops infinite loops and runaway sessions
            ModelCallLimitMiddleware(max_calls=30, on_limit="raise"),
            ToolCallLimitMiddleware(max_calls=60, on_limit="raise"),
            # 3. Approval gate - nothing gets written without your explicit sign-off
            HumanInTheLoopMiddleware(
                interrupt_on={
                    "write_file": True,
                    "read_file": False,
                    "list_directory": False,
                    "search_codebase": False,
                }
            ),
        ],
        checkpointer=checkpointer,  # required for interrupt/resume to work
    )

Gdy agent chce zapisać plik, proces przebiega w następujący sposób:

Agent: I'd like to update src/config.py to fix the circular import.
       Here is what I plan to write:
--- src/config.py ---
       from typing import Optional
       from pydantic import BaseModel
       class Settings(BaseModel):
           debug: bool = False
           database_url: str = "sqlite:///app.db"
       ...
       Approve this change? (yes/no)
User: yes
Agent: Written. Moving to tests/test_config.py next.

The agent cannot touch anything silently. Every write surfaces for review before it happens. Secrets in any file the agent reads are replaced with [REDACTED_API_KEY] before reaching the model — which also means they never appear in the approval prompt you read. What you are reviewing is already clean.

Dwie proste rozszerzenia warto znać. Jeśli baza kodu zawiera dane użytkowników, na przykład numery telefonów w plikach testowych, dodaj wpis PIIMiddleware("phone", strategy="redact") przed tym dotyczącym e-maili. Każdy detektor PII działa niezależnie na tym samym wejściu, więc kolejność wewnątrz tej grupy nie ma znaczenia. A jeśli wolisz łagodne przerwanie działania zamiast gwałtownego zatrzymania, zmień on_limit="raise" na on_limit="return" w obu middleware’ach do zarządzania limitem – agent zwróci wtedy każdy dostępny, częściowy wynik.

Teraz agent nie może wyciekać tajemnic do modelu, nie może modyfikować plików bez twojej zgody i nie może przekroczyć ustalonego limitu zasobów. Pozostaje jedna ograniczenie: nie potrafi zachować spójności podczas długiej sesji. Przegląd setek plików przy użyciu dziesiątek wywołań narzędzi wypełni okno kontekstowe zanim agent skończy pracę, przez co traci orientację w tym, co już omówił, a jego sugestie stają się coraz mniej trafne. To jest odrębny problem od tego, który został rozwiązany tutaj, a mianowicie kontrolowania tego, do czego agent ma dostęp.