Agent w Pythonie z narzędziami do wyszukiwania i odczytu dokumentów

Do tej pory nasz agent miał do dyspozycji dwie zabawkowe funkcje: cenę i stan magazynu produktu. Prawdziwy agent robi coś użytecznego: odpowiada na pytania na podstawie Twoich danych. W tym artykule damy mu narzędzia do przeszukiwania i czytania dokumentów, czyli małego asystenta obsługi klienta, który zanim odpowie, sam sprawdza regulamin.

To trzecia część cyklu. Zakładam, że znasz już tool calling w OpenAI API i pętlę agenta w Pythonie (po angielsku agent loop). Dokumenty i indeks pochodzą z projektu wyszukiwarki semantycznej w Pythonie, więc jeśli go nie masz, zbuduj go najpierw albo użyj własnych plików Markdown z sekcjami ##.

Co zbudujemy i jakie narzędzia da się agentowi?

Agent dostaje trzy narzędzia. Wszystkie tylko czytają, żadne niczego nie zmienia, co jest ważne dla bezpieczeństwa (wrócimy do tego w części o uprawnieniach).

Dlaczego dwa osobne narzędzia do wyszukiwania i odczytu, a nie jedno, które zwraca wszystko? Bo wyszukiwanie zwraca tylko zajawkę (100 znaków), więc odpowiedź mieści się w niewielu tokenach. Model sam ocenia, czy zajawka wystarcza, czy trzeba poprosić o całą sekcję. W dużym zbiorze dokumentów zwracanie pełnych tekstów za każdym razem szybko zapełniłoby okno kontekstu.

Jak opisać narzędzia schematami Pydantic?

Używamy tego samego podejścia co w części o tool callingu: klasa Pydantic opisuje argumenty, jej docstring jest opisem narzędzia, a Field(description=...) opisuje pola. Opisy piszemy jak instrukcję dla modelu, łącznie ze wskazówką, kiedy narzędzia nie używać. Ten i kolejny kod zapisz w pliku docs_tools.py, obok plików wyszukiwarki:

from pydantic import BaseModel, Field


class SearchDocs(BaseModel):
    """Szuka w dokumentach serwisu fragmentów pasujących znaczeniem do zapytania.
    Zwraca źródło, sekcję i początek każdego fragmentu. Nie używaj do pytań spoza
    dokumentów serwisu."""

    query: str = Field(description="Pytanie lub fraza opisująca, czego szukasz, np. 'zwrot pieniędzy'")
    k: int = Field(default=3, ge=1, le=5, description="Ile fragmentów zwrócić")


class ReadSection(BaseModel):
    """Zwraca pełny tekst jednej sekcji dokumentu. Użyj po wyszukiwaniu, gdy początek
    fragmentu nie wystarcza do odpowiedzi."""

    source: str = Field(description="Nazwa pliku z wyniku wyszukiwania, np. 'regulamin.md'")
    section: str = Field(description="Nazwa sekcji z wyniku wyszukiwania, np. 'Reset hasła'")


class ListDocuments(BaseModel):
    """Wypisuje dostępne dokumenty i ich sekcje."""

Zwróć uwagę na ge=1, le=5 przy k. To ograniczenie trafia do schematu i jest sprawdzane przy walidacji, więc model nie zażąda 50 fragmentów naraz.

Jak napisać funkcje, które wykonają narzędzia?

Funkcje trzymamy w klasie DocsToolbox (w tym samym pliku docs_tools.py), która jednorazowo wczytuje indeks i model embeddingów. Dzięki temu model nie ładuje się przy każdym wywołaniu narzędzia. Korzystamy z load_index i search z projektu wyszukiwarki:

from pathlib import Path

from sentence_transformers import SentenceTransformer

from search import load_index, search

MIN_SCORE = 0.30


class DocsToolbox:
    def __init__(self, index_dir: Path = Path("index")):
        self.index = load_index(index_dir)
        self.model = SentenceTransformer(self.index.model_name)

    def _body(self, chunk: dict) -> str:
        return chunk["text"].split("\n\n", 1)[1]

    def _sections(self) -> dict[str, list[str]]:
        wynik: dict[str, list[str]] = {}
        for c in self.index.chunks:
            sekcje = wynik.setdefault(c["source"], [])
            if c["section"] not in sekcje:
                sekcje.append(c["section"])
        return wynik

    def search_docs(self, args: SearchDocs) -> str:
        q = self.model.encode(args.query, normalize_embeddings=True)
        wyniki = search(self.index, q, k=args.k, min_score=MIN_SCORE)
        if not wyniki:
            return "Brak wyników. W dokumentach nie ma informacji na ten temat."
        return "\n".join(
            f"[{r.score:.2f}] {r.chunk['source']} › {r.chunk['section']}: "
            f"{self._body(r.chunk)[:100]}…"
            for r in wyniki
        )

    def read_section(self, args: ReadSection) -> str:
        sekcje = self._sections()
        if args.source not in sekcje:
            return f"Nie ma dokumentu {args.source}. Dostępne: {', '.join(sekcje)}."
        if args.section not in sekcje[args.source]:
            return (
                f"Nie ma sekcji {args.section} w {args.source}. "
                f"Dostępne sekcje: {', '.join(sekcje[args.source])}."
            )
        return "\n\n".join(
            self._body(c)
            for c in self.index.chunks
            if c["source"] == args.source and c["section"] == args.section
        )

    def list_documents(self, args: ListDocuments) -> str:
        return "\n".join(f"{s}: {', '.join(sekcje)}" for s, sekcje in self._sections().items())

Dwie decyzje projektowe warto omówić osobno.

Dlaczego brak wyników to osobny komunikat?

Gdy żaden fragment nie przekracza progu podobieństwa, narzędzie nie zwraca pustego tekstu, tylko jasne zdanie: „W dokumentach nie ma informacji na ten temat.” Pusty wynik model może odczytać jako błąd narzędzia i próbować w kółko. Wyraźne „brak odpowiedzi” pozwala mu uczciwie napisać użytkownikowi, że dokumenty tego nie obejmują. Próg 0,30 pochodzi z pomiaru w projekcie wyszukiwarki, a jak go wyznaczyć dla własnych danych, wyjaśniamy właśnie tam.

Dlaczego odczyt nie przyjmuje ścieżki do pliku?

read_section nie dotyka dysku. Sekcję składa z fragmentów w indeksie, a nazwę dokumentu sprawdza na liście znanych dokumentów. Dzięki temu model nie może poprosić o plik spoza zbioru, nawet gdyby dostał w wyniku wyszukiwania podstępną instrukcję, żeby przeczytać coś innego. Zabezpieczenie wynika ze sposobu zbudowania narzędzia, a nie z filtrowania podejrzanych znaków.

To ogólna zasada: zamiast pozwalać modelowi podawać ścieżkę do pliku i potem ją filtrować, zbuduj narzędzie tak, by ścieżka w ogóle nie istniała, na przykład odczyt z indeksu albo z listy znanych identyfikatorów. Filtrowanie podejrzanych znaków łatwo obejść, brak ścieżki obejść się nie da. Jeśli musisz czytać z dysku, porównuj znormalizowaną ścieżkę z dozwolonym katalogiem.

Jak łączymy narzędzia z pętlą agenta?

Pętla z poprzedniej części miała narzędzia i prompt systemowy wpisane na stałe. Teraz chcemy ją wykorzystać z różnymi zestawami narzędzi, więc przekazujemy je jako parametry. Poniższe funkcje też trafiają do docs_tools.py. Funkcja build_tools buduje listę definicji i funkcję execute, która robi to samo co run_tool wcześniej: waliduje argumenty i zwraca błędy jako tekst:

from pydantic import ValidationError


def to_tool_definition(name: str, model: type[BaseModel]) -> dict:
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": " ".join(model.__doc__.split()),
            "parameters": model.model_json_schema(),
        },
    }


def build_tools(box: DocsToolbox):
    registry = {
        "search_docs": (SearchDocs, box.search_docs),
        "read_section": (ReadSection, box.read_section),
        "list_documents": (ListDocuments, box.list_documents),
    }
    tools = [to_tool_definition(name, model) for name, (model, _) in registry.items()]

    def execute(name: str, raw_arguments: str) -> str:
        if name not in registry:
            return f"Błąd: nie ma narzędzia {name}."
        model, function = registry[name]
        try:
            args = model.model_validate_json(raw_arguments or "{}")
        except ValidationError as error:
            return f"Błąd argumentów: {error}"
        return function(args)

    return tools, execute

Dwie drobne poprawki względem wcześniejszego kodu: " ".join(model.__doc__.split()) usuwa znaki nowej linii z wielowierszowych opisów, a raw_arguments or "{}" obsługuje narzędzie bez argumentów (list_documents), dla którego model może przysłać pusty tekst.

W pętli agenta (plik agent.py) zmieniają się tylko trzy rzeczy: run_agent przyjmuje tools, execute i system_prompt jako parametry, a finish dostaje tools. Reszta, czyli limity kroków i tokenów, wykrywanie powtórzeń i kontrolowane zakończenie, zostaje bez zmian:

from typing import Callable

SYSTEM_PROMPT = (
    "Jesteś asystentem obsługi serwisu Kursownia. Odpowiadaj wyłącznie na podstawie "
    "dokumentów dostępnych przez narzędzia. Najpierw wyszukaj, a gdy początek fragmentu "
    "nie wystarcza, przeczytaj całą sekcję. Na końcu podaj źródło w formacie (plik › sekcja). "
    "Jeśli dokumenty nie zawierają odpowiedzi, napisz o tym wprost i niczego nie zgaduj."
)


def run_agent(
    client: OpenAI,
    question: str,
    tools: list[dict],
    execute: Callable[[str, str], str],
    system_prompt: str = SYSTEM_PROMPT,
    limits: AgentLimits = AgentLimits(),
    model: str = "gpt-4o-mini",
) -> AgentResult:
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": question},
    ]
    tokens_used = 0
    seen_calls: set[tuple[str, str]] = set()

    for step in range(1, limits.max_steps + 1):
        if tokens_used >= limits.max_tokens:
            return finish(client, messages, tools, model, step - 1, tokens_used, "max_tokens")

        response = client.chat.completions.create(model=model, messages=messages, tools=tools)
        tokens_used += response.usage.total_tokens
        message = response.choices[0].message

        if not message.tool_calls:
            return AgentResult(message.content, step, tokens_used, "done")

        messages.append(message)
        for call in message.tool_calls:
            key = (call.function.name, call.function.arguments)
            if key in seen_calls:
                result = "To wywołanie już było. Użyj innych argumentów albo odpowiedz użytkownikowi."
            else:
                seen_calls.add(key)
                result = execute(call.function.name, call.function.arguments)
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})

    return finish(client, messages, tools, model, limits.max_steps, tokens_used, "max_steps")

Funkcja finish jest taka sama jak w poprzedniej części, tylko przekazuje dalej tools: client.chat.completions.create(model=model, messages=messages, tools=tools, tool_choice="none"). Prompt systemowy robi tu ważną robotę: każe odpowiadać wyłącznie z dokumentów, podawać źródło i nie zgadywać, gdy nic nie znaleziono.

Uruchomienie agenta to kilka linijek:

from openai import OpenAI

box = DocsToolbox()
tools, execute = build_tools(box)
client = OpenAI(timeout=30, max_retries=2)

wynik = run_agent(client, "Czy mogę dostać zwrot pieniędzy po tygodniu?", tools, execute)
print(wynik.answer)
print(wynik.steps, wynik.tokens, wynik.stop_reason)

Co zwracają narzędzia dla typowych pytań?

Nie mogliśmy tu pokazać „przykładowej rozmowy z agentem”, bo odpowiedzi prawdziwego modelu zmieniają się między uruchomieniami i zależą od jego wersji. Możemy za to pokazać coś pewniejszego: dokładnie to, co narzędzia zwracają modelowi. Poniżej wyniki search_docs dla zapytań, które agent mógłby wysłać na pytania użytkowników (indeks z 19 fragmentów, trzy dokumenty Kursowni):

Dla pytania o certyfikat narzędzie zwróciło też dwa dalsze fragmenty (rejestracja konta i ochrona danych) z wynikiem 0,31, czyli tuż nad progiem. To szum, a nie odpowiedź. Model zwykle potrafi go zignorować, bo różnica 0,60 do 0,31 jest wyraźna, ale to dobry powód, żeby w prompcie wymagać oparcia odpowiedzi na najtrafniejszym fragmencie. Jeśli szum Ci przeszkadza, podnieś MIN_SCORE albo zmniejsz domyślne k.

Odczyt sekcji zwraca pełny tekst, np. dla read_section(regulamin.md, "Rezygnacja i zwrot pieniędzy") dostajemy oba akapity: o anulowaniu subskrypcji i o zwrocie w ciągu 14 dni od pierwszej płatności. Dopiero tego drugiego zdania model potrzebuje, żeby odpowiedzieć „po tygodniu tak, jeśli nie rozpocząłeś żadnego kursu”.

Co się dzieje, gdy agent poda błędne argumenty?

Narzędzia nie wyrzucają wyjątków, tylko zwracają tekst, który podpowiada, jak poprawić wywołanie. Oto prawdziwe wyjścia dla trzech błędnych wywołań:

Drugi przypadek pokazuje, że próba wyjścia poza katalog dokumentów kończy się niczym: narzędzie nie ma pojęcia ścieżki, więc nie ma czego atakować.

Jak pisać komunikaty błędów, żeby model mógł je naprawić?

Porównaj dwa komunikaty: „Nie ma takiej sekcji” i „Nie ma sekcji Zwrot w regulamin.md. Dostępne sekcje: …”. Pierwszy zostawia modelowi zgadywanie, drugi daje mu gotową listę, z której wybierze następny krok. Wypisuj dostępne wartości w każdym błędzie, w którym zbiór poprawnych wartości jest mały.

Czym ten agent różni się od klasycznego RAG?

W klasycznym RAG zawsze robisz to samo: wyszukujesz fragmenty, wklejasz je do promptu i generujesz odpowiedź. W agencie model decyduje, czy w ogóle szukać, ile razy i czy doczytać całą sekcję.

Agent daje elastyczność kosztem przewidywalności. Jeśli każde pytanie wymaga dokładnie jednego wyszukiwania, prostszy RAG będzie tańszy i szybszy. Jak działa RAG z bazą wektorową, opisujemy w artykule o bazach wektorowych i RAG w Pythonie.

Jak przetestować agenta z narzędziami w pytest?

Testujesz dwie rzeczy osobno: narzędzia (bez modelu językowego) i pętlę (z atrapą klienta). Do testów narzędzi nie ładujemy prawdziwego modelu embeddingów, tylko podstawiamy małą atrapę i ręczny indeks z dwoma wektorami:

import json

import numpy as np
import pytest
from openai.types.chat import ChatCompletion

from agent import run_agent
from docs_tools import DocsToolbox, build_tools
from search import Index


class FakeModel:
    def encode(self, text, normalize_embeddings=True):
        return np.array([1.0, 0.0], dtype=np.float32)


@pytest.fixture
def tools_and_execute():
    box = DocsToolbox.__new__(DocsToolbox)  # pomijamy __init__, żeby nie ładować modelu
    box.index = Index(
        np.array([[1.0, 0.0], [0.0, 1.0]], dtype=np.float32),
        [
            {"id": "r#0", "text": "Zwrot\n\nZwrot w 14 dni.", "source": "regulamin.md", "section": "Zwrot"},
            {"id": "r#1", "text": "Hasło\n\nReset na stronie logowania.", "source": "regulamin.md", "section": "Hasło"},
        ],
        "fake",
    )
    box.model = FakeModel()
    return build_tools(box)


def test_wyszukiwanie_zwraca_zrodlo_i_poczatek_fragmentu(tools_and_execute):
    _, execute = tools_and_execute

    wynik = execute("search_docs", '{"query": "zwrot"}')

    assert "regulamin.md › Zwrot" in wynik


def test_nieznana_sekcja_zwraca_liste_dostepnych(tools_and_execute):
    _, execute = tools_and_execute

    wynik = execute("read_section", '{"source": "regulamin.md", "section": "Płatności"}')

    assert "Dostępne sekcje: Zwrot, Hasło" in wynik


def test_zle_argumenty_wracaja_jako_tekst_a_nie_wyjatek(tools_and_execute):
    _, execute = tools_and_execute

    assert execute("search_docs", '{"query": "x", "k": 50}').startswith("Błąd argumentów")

Test pętli korzysta z atrapy klienta, która zwraca kolejne, z góry ustalone odpowiedzi „modelu”. Dzięki temu bez kosztów i bez losowości sprawdzasz, że agent wyszukuje, czyta sekcję i kończy odpowiedzią:

def odpowiedz(wiadomosc: dict) -> ChatCompletion:
    return ChatCompletion.model_validate({
        "id": "x", "object": "chat.completion", "created": 0, "model": "fake",
        "usage": {"prompt_tokens": 90, "completion_tokens": 10, "total_tokens": 100},
        "choices": [{"index": 0, "finish_reason": "stop", "message": wiadomosc}],
    })


class FakeClient:
    """Zwraca kolejne, z góry ustalone odpowiedzi modelu."""

    def __init__(self, *wiadomosci):
        self._odpowiedzi = [odpowiedz(w) for w in wiadomosci]
        self.chat = self.completions = self

    def create(self, **kwargs):
        return self._odpowiedzi.pop(0)


def wywolanie(id_: str, nazwa: str, argumenty: dict) -> dict:
    return {"role": "assistant", "content": None, "tool_calls": [
        {"id": id_, "type": "function",
         "function": {"name": nazwa, "arguments": json.dumps(argumenty)}}
    ]}


def test_agent_wyszukuje_czyta_i_odpowiada(tools_and_execute):
    tools, execute = tools_and_execute
    client = FakeClient(
        wywolanie("c1", "search_docs", {"query": "zwrot"}),
        wywolanie("c2", "read_section", {"source": "regulamin.md", "section": "Zwrot"}),
        {"role": "assistant", "content": "Zwrot w 14 dni (regulamin.md › Zwrot)"},
    )

    wynik = run_agent(client, "Czy mogę dostać zwrot?", tools, execute)

    assert wynik.stop_reason == "done"
    assert wynik.steps == 3
    assert "regulamin.md" in wynik.answer

Taki test nie sprawdza, czy prawdziwy model wybierze dobre narzędzie. To można zbadać tylko na prawdziwych odpowiedziach, najlepiej na kilkunastu pytaniach napisanych przez kogoś, kto dokumentów nie czytał. Więcej o atrapach w testach znajdziesz w artykule o testowaniu kodu z API LLM.

Co dalej?

Mamy agenta, który czyta dokumenty i zawsze zostaje w bezpiecznym obszarze: wszystkie jego narzędzia tylko czytają. W następnej części zadamy trudniejsze pytanie: co się dzieje, gdy narzędzie coś zmienia, na przykład wysyła maila albo usuwa rekord. Omówimy uprawnienia i zatwierdzanie akcji agenta AI, czyli listę dozwolonych akcji i zgodę człowieka. Jeśli wolisz najpierw poznać inne zastosowanie tych samych bibliotek, zobacz jak Pydantic waliduje dane w FastAPI.