Strona główna / Artykuły / Wbudowanie agenta LangChain w FastAPI: narzędzia, wyszukiwanie w podręczniku, transmisja danych w formie strumienia.

PythonFastAPILangGraphRAGAI AgentsChromaDB

Wbudowanie agenta LangChain w FastAPI: narzędzia, wyszukiwanie w podręczniku, transmisja danych w formie strumienia.

Stworzenie asystenta w aplikacji za pomocą FastAPI i LangChain: podręcznik w formacie PDF przechowywany w ChromaDB dostępny jako narzędzie, kontekst indywidualny dla każdego użytkownika, zapisywanie historii oraz odpowiedzi w formie strumienia.

6913 słów

Gdy aplikacja przekracza kilka ekranów, jej dokumentacja zaczyna się rozrastać, a użytkownicy przestają ją czytać. 20-stronicowy podręcznik wyjaśniający funkcje, konfigurację oraz zasady domeny jest cenny, ale tylko wtedy, gdy ludzie mogą znaleźć w nim odpowiedzi bez długich poszukiwań. Asystent wbudowany w produkt może zamknąć tę lukę: odpowiada na pytania typu „jak to działa?” z podręcznika oraz „co jest w moim projekcie?” na podstawie danych samej aplikacji.

To przewodnik opisuje zwięzłą, działającą wersję takiego asystenta. Połączysz agenta LangChain z usługą FastAPI, dostarczysz mu narzędzie do odczytu danych aplikacji oraz drugie narzędzie do wyszukiwania instrukcji w formacie PDF przechowywanych w ChromaDB, przekażesz autoryzowanego użytkownika do agenta w momencie żądania, będziesz przechowywał historię rozmów za pomocą checkpointera LangGraph oraz przesyłać odpowiedź z powrotem do klienta. Po drodze wskazujemy braki w minimalnym kodzie, które musisz uzupełnić, aby aplikacja działała poprawnie, oraz to, co należy zmienić przed wdrożeniem jej do produkcji.

Sytuacja i elementy ruchome

Załóżmy zespół, który tworzy narzędzie do projektowania instalacji fotowoltaicznych. Produkt zaczął się od prostych rozwiązań, a następnie powstały dodatkowo panele, inwertery, szacunki produkcji, plany dachów oraz długa lista zasad projektowych. Ich instrukcja użytkownika liczy teraz ponad 20 stron. Ciągle pojawiają się dwa rodzaje pytań:

  • Pytania dotyczące samego produktu: co robi dana funkcja, gdzie znajduje się ustawienie, która zasada ma zastosowanie. Odpowiedzi znajdują się w dokumentacji.
  • Pytania dotyczące pracy samego użytkownika: jaki inwerter wybrał, ile mocy wytwarza jego system, jakie dachy ma obsłużone. Odpowiedzi znajdują się w bazie danych aplikacji, a żaden model nie zna ich domyślnie.

Retrieval-Augmented Generation (RAG) zajmuje się pierwszym typem pytań: indeksuje instrukcję obsługi, wyszukuje odpowiednie fragmenty po zadaniu pytania i przekazuje je modelowi jako kontekst. Narzędzia obsługują drugi typ: małe funkcje, które model może wywołać w celu pobrania danych z aplikacji. Po połączeniu obu metod otrzymujemy asystenta, który potrafi zarówno wyjaśnić produkt, jak i analizować konkretny projekt.

Rzeczywisty system stojący za tym scenariuszem posiada o wiele więcej narzędzi oraz znacznie bardziej złożoną logikę domenową. To, co zostało przedstawione, zostało celowo sprowadzone do najważniejszych elementów, aby architektura pozostała widoczna:

  • FastAPI udostępnia API HTTP i określa, kto jest żądającym.
  • Agent LangChain (działający na LangGraph) zarządza pętlą rozumowania oraz stanem rozmowy.
  • LLM interpretuje każde żądanie i decyduje, czy potrzebne są informacje z zewnątrz.
  • Narzędzia umożliwiają agentowi kontrolowany dostęp do funkcjonalności aplikacji.
  • RAG pozwala agentowi wyszukiwać informacje w dokumentacji.
  • ChromaDB przechowuje fragmenty podręcznika i przeprowadza wyszukiwanie wektorowe.
  • Strumieniowanie dostarcza tokeny do klienta w trakcie generowania odpowiedzi.

Zaletą tego podejścia jest to, że asystent działa w ramach istniejącego aplikacji typu full-stack z rzeczywistymi użytkownikami i danymi, zamiast funkcjonować obok niej jako zwykły chatbot.

Struktura projektu

Każda funkcja ma swój własny pakiet: routowanie HTTP, autoryzacja, logika agenta, narzędzia oraz pipeline RAG. Dzięki temu każdy plik pozostaje niewielki, a jest jasne, gdzie powinna znaleźć się nowa funkcjonalność.

project/
│
├── main.py
├── .env
├── .gitignore
│
├── auth/
│   ├── __init__.py
│   └── dependencies.py
│
├── routers/
│   ├── __init__.py
│   └── chat.py
│
├── llm/
│   ├── __init__.py
│   ├── agent.py
│   ├── context.py
│   ├── orchestrator.py
│   ├── prompts.py
│   ├── provider.py
│   │
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── demo_tool.py
│   │   └── manual_tool.py
│   │
│   └── rag/
│       ├── __init__.py
│       ├── config.py
│       ├── context.py
│       │
│       ├── ingestion/
│       │   ├── __init__.py
│       │   ├── loader.py
│       │   ├── chunker.py
│       │   ├── chroma.py
│       │   └── indexer.py
│       │
│       └── retrieval/
│           ├── __init__.py
│           └── retriever.py
│
├── scripts/
│   ├── __init__.py
│   └── index_manual.py
│
├── docs/
│   └── manual.pdf
│
└── chroma_data/ (*generated locally, not commited or deployed)

Co obejmuje każda część:

  • main.py tworzy aplikację FastAPI.
  • auth/ zawiera symulowaną zależność dotyczącą autoryzacji.
  • routers/ zawiera punkty końcowe HTTP.
  • llm/ zawiera wszystko, co dotyczy agenta.
  • llm/tools/ zawiera funkcje, które agent może wywołać.
  • llm/rag/ zawiera pipeline do wyszukiwania danych, podzielony na ingestion/ (ładowanie, dzielenie na fragmenty i indeksowanie PDF) oraz retrieval/ (wykonywanie zapytań do ChromaDB).
  • scripts/ zawiera polecenia, które uruchamiasz ręcznie, takie jak indeksowanie dokumentacji.
  • docs/ zawiera oryginalny plik PDF.
  • chroma_data/ jest generowany lokalnie i nigdy nie powinien być komitowany ani wdrażany.

    • Nie będziesz tworzyć ich wszystkich jednocześnie. Kolejność budowy to: warstwa API, następnie model, potem narzędzia, dalej pipeline RAG, a na końcu kontekst, transmisja danych w czasie rzeczywistym i historia działań.

    Krok 1: Szkielet FastAPI z symulowanym użytkownikiem

    Najpierw zainstaluj wszystko, czego potrzebuje projekt. Lista obejmuje serwer internetowy, LangChain i LangGraph, klienta do czatowania kompatybilnego z OpenAI, ChromaDB, narzędzia do ładowania PDF i dzielenia tekstu oraz python-dotenv do konfiguracji.

    pip install fastapi uvicorn langchain langgraph langchain-openai chromadb langchain-community langchain-text-splitters pypdf python-dotenv

    Do modelu będzie można uzyskać dostęp przez OpenRouter, więc utwórz plik .env w korzeniu projektu, który będzie zawierał klucz API.

    OPENROUTER_API_KEY=your_api_key_here

    Niezwłocznie dodaj .env do pliku .gitignore. Klucz, który trafi do systemu kontroli wersji, powinien być uznawany za wyciekły.

    Punkt wejścia aplikacji

    main.py pozostaje bardzo prosty. Tworzy aplikację i rejestruje router do czatów; nic związanego z modelem czy agentem nie powinno się tu znajdować.

    from fastapi import FastAPI
from routers.chat import chat_router

app = FastAPI(
    title="AI Agent Demo",
)
app.include_router(chat_router)

    Pierwszy punkt końcowy czatu

    W pliku routers/chat.py zdefiniuj router pod prefiksem /chat z jedną trasą POST. Na razie będzie on tylko odbijał przychodzące wiadomości, co wystarczy, by potwierdzić poprawne działanie infrastruktury przed włączeniem sztucznej inteligencji.

    from fastapi import APIRouter

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
):
    return {
        "message": message,
    }

    Uwaga: w ścieżce POST bez modelu ciała, pole message: str sprawia, że FastAPI odczytuje je z ciągu zapytania. Jest to wygodne podczas testowania w Swagger UI, ale w przypadku rzeczywistego klienta zazwyczaj przyjmuje się ciało w formacie JSON zdefiniowane za pomocą modelu Pydantic, ponieważ ciągi zapytania trafiają do logów dostępu i mają praktyczne ograniczenia co do długości.

    Symulowana zależność autoryzacyjna

    Aplikacja produkcyjna weryfikowałaby plik cookie sesji lub token JWT oraz ładowała użytkownika z bazy danych. Tutaj wszystko to zastępuje specjalny nagłówek. Stwórz plik auth/dependencies.py zawierający małą klasę danych MockUser oraz funkcję get_current_user, która odczytuje nagłówek X-Demo-User i odrzuca żądanie z kodem 401, jeśli ten jest brakujący.

    from dataclasses import dataclass
from fastapi import Header, HTTPException

@dataclass
class MockUser:
    id: str
    name: str

def get_current_user(
    x_demo_user: str | None = Header(default=None),
) -> MockUser:
    if x_demo_user is None:
        raise HTTPException(
            status_code=401,
            detail="Missing X-Demo-User header",
        )
    return MockUser(
        id=x_demo_user,
        name=x_demo_user,
    )

    Iniekcja zależności w FastAPI teraz przekazuje użytkownika do punktu końcowego. Wystarczy zadeklarować parametr za pomocą Depends(get_current_user).

    from fastapi import APIRouter, Depends
from auth.dependencies import (
    MockUser,
    get_current_user,
)

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    return {
        "user": current_user.name,
        "message": message,
    }

    Klient identyfikuje się, wysyłając taki nagłówek:

    X-Demo-User: user-123

    Przy każdej prośbie FastAPI najpierw wywołuje get_current_user(), a następnie przekazuje uzyskany obiekt MockUser do funkcji ask_ai. Kluczowym aspektem projektu jest podział zadań: FastAPI zajmuje się autoryzacją, natomiast warstwa AI otrzymuje obiekt użytkownika, któremu może ufać. Później ten obiekt użytkownika umożliwia narzędziom zwracanie danych należących do właściwej osoby. Zastąpienie mocka prawdziwą autoryzacją później zmienia tylko tę jedną zależność.

    Krok 2: Połączenie modelu za pomocą OpenRouter

    Z funkcjonującym punktem końcowym i znanym wywołującym, usługa potrzebuje modelu. OpenRouter udostępnia API kompatybilne z OpenAI, więc klasa ChatOpenAI z LangChain może być używana z nim, jeśli skierujesz base_url na OpenRouter i przekażesz swój klucz OpenRouter.

    Umieść to w pliku llm/provider.py. Plik ten ładuje plik .env, szybko zgłasza wyraźny błąd w przypadku braku klucza oraz otacza go strukturą SecretStr z Pydantic, aby nie został przypadkowo wyświetlony w logach lub reprezentacjach kodu.

    import os
from dotenv import load_dotenv
from pydantic import SecretStr
from langchain_openai import ChatOpenAI

load_dotenv()

api_key = os.getenv(
    "OPENROUTER_API_KEY",
)
if not api_key:
    raise RuntimeError(
        "OPENROUTER_API_KEY environment variable is not set."
    )

model = ChatOpenAI(
    model="YOUR_MODEL",
    api_key=SecretStr(api_key),
    base_url="https://openrouter.ai/api/v1",
)

    Ponieważ klucz pochodzi z środowiska, nigdy nie pojawia się w kodzie źródłowym. Na tym etapie można już wysyłać zapytania do modelu i otrzymywać odpowiedzi, ale to jest zwykłe wezwanie LLM. Celem jest agent, który sam może zdecydować, kiedy potrzebuje narzędzia.

    Wybór modelu

    Argument model to po prostu identyfikator modelu OpenRouter, więc możesz zmieniać modele bez ingerencji w resztę kodu. Porównując różne opcje, sprawdź:

    • wspieranie wywoływania narzędzi, od którego zależy agent;
    • wspieranie transmisji strumieniowej;
    • rozmiar okna kontekstowego;
    • ograniczenia szybkości;
    • obecność darmowej wersji.

    OpenRouter oferuje niektóre modele za darmo, co jest przydatne podczas eksperymentowania. Katalog regularnie się zmienia, więc sprawdź aktualną listę i filtrowaj modele darmowe, zamiast polegać na stałych rekomendacjach. Bez względu na wybór, zostanie on bezpośrednio podany do konstruktora:

    model = ChatOpenAI(
    model="YOUR_MODEL_ID",
    api_key=SecretStr(api_key),
    base_url="https://openrouter.ai/api/v1",
)

    Na przykład, jeśli katalog zawiera identyfikator taki jak ten poniżej (przykład z czasu pisania tekstu; może już nie być dostępny), należy przekazać dokładnie tę ścieżkę jako model:

    google/gemma-4-26b-a4b-it:free

    Należy pamiętać, że modele darmowe korzystają z wspólnej przepustowości. Mogą mieć ograniczenia co do częstotliwości wykorzystania lub stać się niedostępne, często w najgorszym momencie podczas demonstracji. Zmiana na inny model lub użycie własnego klucza dostawcy przez OpenRouter zazwyczaj rozwiązuje ten problem. Do zastosowań produkcyjnych należy wybierać modele pod kątem niezawodności, możliwości, opóźnienia i kosztu, a nie tylko ceny.

    Krok 3: Od modelu do agenta

    Bezpośrednia próba połączenia z modelem polega na jednym kroku: tekst użytkownika trafia do modelu, a następnie otrzymujemy gotową odpowiedź. Agent wprowadza pętlę decyzyjną – model analizuje żądanie, decyduje, czy może od razu odpowiedzieć, czy potrzebuje najpierw jakichś informacji, w razie potrzeby wywołuje narzędzie, odczytuje wynik i dopiero wtedy generuje ostateczną odpowiedź. Mniej więcej:

    • prosta próba połączenia: użytkownik, następnie LLM, potem odpowiedź;
  • agent: użytkownik, następnie agent, potem LLM decyduje, co jest potrzebne, dalej wywołanie narzędzia lub pobranie danych w razie potrzeby, na koniec odpowiedź.

    • Moduł agenta

    W pliku llm/agent.py funkcja create_agent z LangChain buduje agenta na podstawie modelu i promptu systemowego. Wewnętrznie tworzy graf LangGraph, który uruchamia pętlę modelu i narzędzi za nas.

    from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT

agent = create_agent(
    model=model,
    system_prompt=SYSTEM_PROMPT,
)

    Ten agent jeszcze nie ma żadnych narzędzi, więc zachowuje się bardzo podobnie do samego modelu. Najpierw należy mu dać instrukcje.

    Prompt systemowy

    W pliku llm/prompts.py znajduje się krótki prompt, który informuje model o jego przeznaczeniu, zabrania mu wymyślania danych oraz wskazuje, jaki rodzaj pytania odpowiada danemu rodzajowi wyszukiwania.

    SYSTEM_PROMPT = """
You are an AI assistant for our demo application.
You help users understand the application and navigate the system.
Never invent data.
When information about the demo system
is required, use the available application tools.
When answering questions about the application,
use the documentation search tool.
Always answer in clear, conversational language.
""".strip()

    Prompt określa dwa źródła prawdy:

    • Dane aplikacji (to, co znajduje się w koncie użytkownika) pochodzą z narzędzi aplikacji;
    • wiedza o aplikacji (jak działa produkt) pochodzi z wyszukiwania w dokumentacji.

    Jedno wyjaśnienie przed kontynuacją. Narzędzia i RAG są przedstawione oddzielnie poniżej, ponieważ dzięki temu łatwiej je zrozumieć, ale w gotowym agencie samo wyszukiwanie w dokumentacji jest narzędziem. Nie ma drugiego mechanizmu: agent widzi listę funkcji, które można wywołać, a wyszukiwanie w instrukcji jest jedną z nich.

    Krok 4: Pierwsze narzędzie

    Narzędziem jest funkcja, którą agent ma prawo wywołać. To właśnie umożliwia skalowanie architektury: zamiast wkładać wszystkie dane aplikacji do promptu, eksponuje się konkretne operacje i pozwala modelowi żądać ich tylko wtedy, gdy jest to konieczne przy odpowiadaniu na pytanie.

    Dla demonstracji plik llm/tools/demo_tool.py definiuje narzędzie, które zwraca ustalony blok informacji o projekcie.

    from langchain.tools import tool

@tool
def get_my_demo_data() -> str:
    """
    Return information about the demonstration data.
    This is just for demo data. But in production, make a more detailed instruction.
    """

    return """
    Project: Aperture Analytics Dashboard
    Owner: Jordan Lee
    Status: In Progress
    Team size: 6
    Budget: $84,000
    Deadline: 2026-11-15
    Description: An internal dashboard for visualizing customer usage
    metrics, built with FastAPI and React, integrating with the
    company's data warehouse.
    """.strip()

    Tutaj istotne są dwa szczegóły. Dekorator @tool przekształca zwykłą funkcję w Pythonie w narzędzie LangChain, pobierając jego nazwę oraz schemat argumentów z sygnatury funkcji. Dokumentacja staje się opisem narzędzia, który model czyta podczas decydowania o jego wywołaniu. W rzeczywistym systemie taki opis wymaga szczególnej uwagi: należy dokładnie określić, co narzędzie zwraca, kiedy jest to odpowiednie, a kiedy nie. Niejasny opis jest jedną z najczęstszych przyczyn, dla których agent wywołuje niewłaściwe narzędzie lub w ogóle żadne.

    Zarejestruj narzędzie, przekazując je do funkcji create_agent:

    from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import get_my_demo_data

agent = create_agent(
    model=model,
    tools=[
        get_my_demo_data,
    ],
    system_prompt=SYSTEM_PROMPT,
)

    get_my_demo_data(). Jeśli użytkownik zapyta „Co to jest demo?”, nie ma potrzeby żadnego wyszukiwania, a model odpowiada bezpośrednio. Ta decyzja jest podejmowana przy każdym kroku, wewnątrz pętli agenta.

    Krok 5: Budowa pipeline RAG dla instrukcji

    Warto być precyzyjnym co do tego, czym jest RAG, a czym nie. Nic nie jest szkoleniowe ani dopracowywane na podstawie dokumentacji. Podczas wysyłania zapytania system przeszukuje podręcznik w poszukiwaniu fragmentów najbardziej istotnych dla pytania i dostarcza te fragmenty modelowi jako kontekst, a model odpowiada na ich podstawie.

    Proces składa się z dwóch faz:

    1. Przyjmowanie danych, które odbywa się oddzielnie od aplikacji internetowej: załaduj PDF, podziel go na fragmenty i przechowaj te fragmenty wraz z ich embeddingami w ChromaDB.
    2. Pobieranie informacji, które odbywa się w ramach pojedynczego żądania: weź pytanie, przeszukaj ChromaDB, zebraj najlepsze fragmenty i przekaż je modelowi.

    Jeśli chcesz szerszego spojrzenia na te koncepcje, przegląd bloga dotyczący odzyskiwania świeżej wiedzy na żądanie omawia je bardziej szczegółowo; tutaj skupiamy się na implementacji.

    Ładowanie PDF

    llm/rag/ingestion/loader.py wykorzystuje PyPDFLoader z LangChain, który przekształca każdą stronę PDF w obiekt Document.

    from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader

PDF_PATH = Path("docs/manual.pdf")

def load_manual():
    loader = PyPDFLoader(
        str(PDF_PATH),
    )
    documents = loader.load()
    return documents

    Obiekt Document zawiera dwie rzeczy: wydobyty tekst w polu page_content oraz słownik metadata opisujący źródło tego tekstu. To właśnie metadane umożliwiają późniejsze cytowanie konkretnej strony w odpowiedzi. Koncepcyjnie każda załadowana strona wygląda w ten sposób:

    Document
├── page_content
│   └── "To create a new demo data..."
│
└── metadata
    ├── source: docs/manual.pdf
    └── page: 12

    PyPDFLoader zazwyczaj rejestruje zarówno indeks page liczony od zera, jak i czytelny dla człowieka page_label. Poniższy budownik kontekstu wykorzystuje page_label, który odpowiada numerom stron widocznym u czytelników w pliku PDF.

    Dzielenie stron na fragmenty

    Szukanie w całych stronach, nie mówiąc już o całym dokumencie jako jednym bloku, daje niedokładne wyniki. llm/rag/ingestion/chunker.py dzieli dokumenty za pomocą RecursiveCharacterTextSplitter.

    from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
)
from langchain_core.documents import Document

def chunk_documents(
    documents: list[Document],
) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=150,
        separators=[
            "\n\n",
            "\n",
            ". ",
            " ",
            "",
        ],
    )
    return splitter.split_documents(
        documents,
    )

    Dzielnik ma na celu tworzenie fragmentów o długości około 1000 znaków z 150-znakowym nakładaniem się. Lista separatorów jest sprawdzana w określonej kolejności: najpierw preferuje podział na granicach akapitów, potem na liniach, następnie na końcach zdań, dalej na przestrzeniach, a dopiero w ostateczności pośrodku słowa. Nakładanie się istnieje dlatego, że jeden fakt może obejmować kilka punktów podziału; powtórzenie krótkiego tekstu po obu stronach zmniejsza szansę, że odpowiednie zdanie zostanie przecięte na pół.

    Ty liczby stanowią punkty wyjścia, a nie reguły. Odpowiednia wielkość zależy od sposobu pisania dokumentów oraz od tego, jak precyzyjne musi być wyszukiwanie, więc traktuj je jako wartości do dostosowania w zależności od rzeczywistych potrzeb. Artykuł na blogu o dzielzeniu na fragmenty zachowującym dowody szczegółowo omawia ten kompromis.

    Ciągła kolekcja Chroma

    llm/rag/ingestion/chroma.py otwiera PersistentClient, który przechowuje dane na dysku i zwraca kolekcję dokumentów, tworząc ją przy pierwszym użyciu.

    import chromadb
from llm.rag.config import (
    CHROMA_PATH,
    MANUAL_COLLECTION_NAME,
)

def get_chroma_client():
    return chromadb.PersistentClient(
        path=CHROMA_PATH,
    )

def get_manual_collection():
    client = get_chroma_client()
    return client.get_or_create_collection(
        name=MANUAL_COLLECTION_NAME,
    )

    Ścieżki i nazwy pochodzą z llm/rag/config.py, który odczytuje zmienne środowiskowe i używa rozsądnych domyślnych wartości w razie braku danych:

    import os

CHROMA_PATH = os.getenv(
    "CHROMA_PATH",
    "./chroma_data",
)
MANUAL_PATH = os.getenv(
    "MANUAL_PATH",
    "docs/manual.pdf",
)
MANUAL_COLLECTION_NAME = os.getenv(
    "MANUAL_COLLECTION_NAME",
    "manual",
)

    Należy dodać odpowiednie wpisy do pliku .env:

    CHROMA_PATH=./chroma_data
MANUAL_PATH=docs/manual.pdf
MANUAL_COLLECTION_NAME=manual

    Żaden model embedding nie jest skonfigurowany, co jest zamierzone w tym demo. Gdy kolekcja jest tworzona bez wyraźnej funkcji embedding, Chroma używa swojego wbudowanego domyślnego modelu: za każdym razem, gdy dodaje się dokumenty, Chroma sama oblicza ich embeddingi i przechowuje je obok tekstu oraz metadanych. Domyślny model działa lokalnie i jest pobierany przy pierwszym użyciu, więc pierwsze indeksowanie może zostać wstrzymane podczas jego pobierania.

    W rezultacie powstaje lokalne, trwałe magazynowanie wektorów w katalogu chroma_data/. Ponieważ jest ono w całości wywodzone z pliku PDF, należy dodać je do pliku .gitignore obok pliku .env.

    Zadanie indeksowania

    llm/rag/ingestion/indexer.py łączy ze sobą poszczególne kroki procesu indeksowania.

    from pathlib import Path
from llm.rag.config import MANUAL_PATH
from llm.rag.ingestion.loader import load_manual
from llm.rag.ingestion.chunker import chunk_documents
from llm.rag.ingestion.chroma import get_manual_collection

def index_manual():
    collection = get_manual_collection()
    if collection.count() > 0:
        print(
            f"Manual already indexed "
            f"({collection.count()} chunks)."
        )
        return
    manual_path = Path(
        MANUAL_PATH,
    )
    if not manual_path.exists():
        raise FileNotFoundError(
            f"Manual not found: {manual_path}"
        )
    documents = load_manual()
    print(
        f"Loaded {len(documents)} pages."
    )
    chunks = chunk_documents(
        documents,
    )
    print(
        f"Created {len(chunks)} chunks."
    )
    collection.add(
        ids=[
            f"manual-chunk-{i}"
            for i in range(len(chunks))
        ],
        documents=[
            chunk.page_content
            for chunk in chunks
        ],
        metadatas=[
            chunk.metadata
            for chunk in chunks
        ],
    )
    print(
        f"Stored {len(chunks)} chunks."
    )

    Przeanalizujmy, co robi ten kod. Otwiera kolekcję i zwraca się wcześniej, jeśli już zawiera fragmenty danych, co sprawia, że ponowne uruchomienie nie ma żadnych konsekwencji. Sprawdza, czy plik PDF istnieje, a w przeciwnym razie wywołuje jasny błąd. Następnie ładuje strony, dzieli je na fragmenty i dodaje wszystko do Chromy za pomocą jednego wywołania, używając spójnych identyfikatorów (manual-chunk-0, manual-chunk-1 itd.), tekstów fragmentów oraz ich metadanych. Komunikaty o postępach informują o liczbie przetworzonych stron i fragmentów.

    Jedna z pułapek wynika z tego wczesnego powrotu: jeśli edytujesz manual i uruchamiasz skrypt ponownie, nic się nie dzieje, ponieważ zbiór nie jest pusty. Aby uwzględnić zmiany, musisz usunąć zbiór (lub katalog chroma_data/) przed ponownym indeksowaniem albo zastąpić mechanizm ochronny logiką, która celowo dokonuje aktualizacji lub odbudowy.

    Wykaz nie pokazuje pliku scripts/index_manual.py; wystarczy, że zaimportuje on index_manual i go wywoła. Uruchom go raz jako moduł z korzenia projektu:

    python -m scripts.index_manual

    To pojedyncze uruchomienie odczytuje PDF, dzieli go na fragmenty, wstawia te fragmenty i przechowuje je. Wyniki z terminala pokazują ich liczbę. W uproszczonej wersji demonstracji podręcznik to jednostronicowy PDF zawierający jedną zasadę: „Dane demonstracyjne mogą być udostępniane wyłącznie użytkownikom administracyjnym”, co wystarcza, by sprawdzić, czy funkcja pobierania działa poprawnie. Od tego momentu uruchamianie FastAPI w ogóle nie wpływa na PDF, ponieważ wektory zostały już zapisane.

    Zapytanie do kolekcji

    Samo indeksowanie nic nie daje agencie; potrzebuje on sposobu na wyszukiwanie. Plik llm/rag/retrieval/retriever.py otacza API do zapytań Chromy.

    from dataclasses import dataclass
from typing import Any
from llm.rag.ingestion.chroma import (
    get_manual_collection,
)

@dataclass
class RetrievedChunk:
    content: str
    metadata: dict[str, Any]
    distance: float

def retrieve_manual(
    query: str,
    n_results: int = 5,
) -> list[RetrievedChunk]:
    collection = get_manual_collection()
    results = collection.query(
        query_texts=[query],
        n_results=n_results,
        include=[
            "documents",
            "metadatas",
            "distances",
        ],
    )
    documents = results["documents"] or []
    metadatas = results["metadatas"] or []
    distances = results["distances"] or []
    retrieved_chunks = []
    for document, metadata, distance in zip(
        documents[0],
        metadatas[0],
        distances[0],
    ):
        retrieved_chunks.append(
            RetrievedChunk(
                content=document,
                metadata=dict(metadata)
                if metadata else {},
                distance=distance,
            )
        )
    return retrieved_chunks

    Funkcja wysyła pytanie jako query_texts, prosi o maksymalnie pięć wyników oraz żąda dokumentów, ich metadanych i odległości między nimi. Chroma zwraca po jednej liście na każde zapytanie, dlatego kod odczytuje indeks [0] każdego pola; warunek awaryjny or [] chroni przed brakującymi polami. Każdy wynik jest pakowany jako klasa RetrievedChunk, dzięki czemu reszta kodu nie zależy od struktury odpowiedzi Chromy.

    Ponieważ zapytanie jest włączone za pomocą tego samego modelu co przechowywane fragmenty tekstu, dopasowanie ma charakter semantyczny. Pytanie takie jak „Jak dodać nowe dane demonstracyjne?” znajduje fragmenty dotyczące tworzenia lub udzielania dostępu do danych demonstracyjnych, nawet jeśli nigdy nie użyto słów „dodać nowe”. Odległość pokazuje, jak bliskie jest każde dopasowanie; im mniejsza wartość, tym większe podobieństwo. Ta informacja jest przydatna później, jeśli chcesz odrzucić słabe dopasowania zamiast zawsze przekazywać modelowi pięć fragmentów.

    Gdy to jest już ustawione, funkcjonuje część systemu RAG odpowiedzialna za wyszukiwanie. Pozostaje już tylko przekazanie wyników modelowi.

    Zmiana fragmentów w kontekst

    llm/rag/context.py formatuje znalezione wyniki w jeden ciąg znaków, który model może odczytać.

    from llm.rag.retrieval.retriever import (
    RetrievedChunk,
)

def build_context(
    chunks: list[RetrievedChunk],
) -> str:
    context_parts = []
    for chunk in chunks:
        page = chunk.metadata.get(
            "page_label",
        )
        context_parts.append(
            f"Source: User Guide, page {page}\n"
            f"{chunk.content}"
        )
    return "\n\n---\n\n".join(
        context_parts,
    )

    Każdy fragment jest poprzedzony linijką źródłową z nazwą przewodnika użytkownika oraz jego numerem strony, a fragmenty są oddzielone podziałnikiem. To właśnie linijka źródłowa pozwala modelowi wskazać, skąd pochodzi odpowiedź, a także daje użytkownikom możliwość jej weryfikacji.

    Krok 6: Udostępnianie podręcznika jako narzędzia

    Proces jest zakończony: strony są ładowane i dzielone na fragmenty, które przechowują się w ChromaDB; można je znaleźć i sformatować. Agent jednak nie ma pojęcia, że cokolwiek z tego istnieje. W tym momencie przydaje się wcześniejsza uwaga architektoniczna – wyszukiwanie dokumentacji staje się po prostu kolejnym narzędziem.

    llm/tools/manual_tool.py definiuje funkcję search_user_manual, która przyjmuje zapytanie, pobiera pięć fragmentów tekstu i zwraca je w uformatowanym kontekście. Jeśli nic nie zostanie znalezione, zwraca wyraźną wiadomość informującą, że podręcznik nie obejmuje tego pytania, co daje modelowi coś konkretnego do przekazania zamiast pustej strony.

    from langchain.tools import tool
from llm.rag.context import build_context
from llm.rag.retrieval.retriever import retrieve_manual

@tool
def search_user_manual(
    query: str,
) -> str:
    """
    Search the application user manual.
    Use this tool when the user asks about application
    behavior, instructions, rules, limitations, or
    how something works.
    """
    chunks = retrieve_manual(
        query=query,
        n_results=5,
    )
    if not chunks:
        return (
            "The manual does not contain enough "
            "information to answer this question."
        )
    return build_context(
        chunks,
    )

    Podobnie jak wcześniej, dokumentacja jest reklamą narzędzia dla modelu. Podaje ona, że należy używać tego narzędzia do pytań dotyczących zachowania, instrukcji, zasad, ograniczeń oraz sposobu działania różnych elementów, co jest zgodne z instrukcjami systemowymi.

    Teraz należy zarejestrować oba narzędzia u agenta:

    from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import (
    get_my_solar_system,
)
from llm.tools.manual_tool import (
    search_user_manual,
)

agent = create_agent(
    model=model,
    tools=[
        get_my_solar_system,
        search_user_manual,
    ],
    system_prompt=SYSTEM_PROMPT,
)

    Zwróć uwagę na import w tym wykazie: odnosi się on do get_my_solar_system, nazwy z całej aplikacji, podczas gdy moduł narzędzia demonstracyjnego definiuje get_my_demo_data. Użyj get_my_demo_data zarówno w instrukcji importu, jak i na liście tools, w przeciwnym razie import modułu nie powiedzie się.

    Dlaczego projekt oparty na narzędziach sprawdza się w miarę rozwoju aplikacji

    Agent nigdy nie potrzebuje jednego ogromnego polecenia opisującego wszystko, co wie aplikacja. Zamiast tego otrzymuje wąskie, kontrolowane możliwości. Dodanie nowej funkcji do asystenta oznacza napisanie nowego narzędzia i jego zarejestrowanie; warstwa HTTP nie ulega zmianie. W demonstracji zachowuje się dokładnie jedno narzędzie do przetwarzania danych i jedno narzędzie dokumentacyjne, dzięki czemu wzorzec jest łatwy do zrozumienia, ale ta sama struktura umożliwia obsługę znacznie większej liczby narzędzi w pełnej aplikacji.

    Krok 7: Przekazywanie zalogowanego użytkownika do agenta

    Narzędzie demonstracyjne nadal zwraca dane hard-kodowane. Prawdziwe narzędzie musi wiedzieć, kto zadaje pytanie, a aplikacja już to wie: FastAPI rozwiązało problem identyfikacji użytkownika dzięki zależności auth. Brakującym elementem jest przeniesienie tego użytkownika do procesu wykonywania agenta. LangChain nazywa to kontekstem czasu wykonywania.

    Zdefiniuj strukturę kontekstu w pliku llm/context.py:

    from dataclasses import dataclass
from auth.dependencies import MockUser

@dataclass
class AgentContext:
    user: MockUser

    Obiekt ten jest dostarczany podczas wywołania agenta, a ta różnica ma znaczenie. Użytkownik to informacja skojarzona z konkretnym żądaniem. Należy ona do bieżącego żądania HTTP, a nie do całej rozmowy, i nigdy nie powinna być przechowywana jako wiadomość, którą model mógłby odczytać lub przepisać. Unikanie jej w historii wiadomości oznacza również, że prompt nie może zmusić agenta do działania jako inny użytkownik. W pełnej aplikacji ten sam obiekt kontekstu zawiera również informacje takie jak sesja bazy danych oraz identyfikator edytowanego projektu.

    Kod demonstracyjny kończy się na zdefiniowaniu klasy, więc pozostają dwa połączenia do nawiązania, a warto sprawdzić aktualną dokumentację LangChain w celu pozyskania dokładnych informacji na temat API. Po pierwsze, należy zadeklarować schemat podczas tworzenia agenta, zwykle za pomocą argumentu context_schema=AgentContext w funkcji create_agent. Po drugie, narzędzia muszą go odczytać: w LangChain 1.x narzędzie może przyjmować parametr czasu wykonywania (np. oznaczony jako ToolRuntime[AgentContext]) i odczytywać dane użytkownika z jego atrybutu context, który jest ukryty przed modelem w kontekście argumentów narzędzia. To właśnie tam rzeczywiste funkcje typu get_my_demo_data wyszukałyby rekordy dla user.id.

    Krok 8: Warstwa orkiestracji do przesyłania danych w strumieniu

    Zamiast wywoływać agenta bezpośrednio z wnętrza routera, umieść tę interakcję w pliku llm/orchestrator.py. Dzięki temu router będzie skupiony wyłącznie na protokole HTTP, a orchestrator będzie odpowiadał za to, jak wiadomość zostanie przekształcona w uruchomienie agenta.

    from collections.abc import Iterator
from langchain_core.messages import (
    AIMessage,
    AIMessageChunk,
    BaseMessage,
    ToolMessage,
)
from langchain_core.runnables import RunnableConfig
from llm.agent import agent
from llm.context import AgentContext

def chat_stream(
    user_message: str,
    user,
) -> Iterator[str]:
    config: RunnableConfig = {
        "configurable": {
            "thread_id": f"user:{user.id}",
        }
    }
    context = AgentContext(
        user=user,
    )
    for chunk, metadata in agent.stream(
        {
            "messages": [
                {
                    "role": "user",
                    "content": user_message,
                }
            ]
        },
        config=config,
        context=context,
        stream_mode="messages",
    ):
        if not isinstance(
            chunk,
            BaseMessage,
        ):
            continue
        if isinstance(
            chunk,
            ToolMessage,
        ):
            continue
        if not isinstance(
            chunk,
            (
                AIMessage,
                AIMessageChunk,
            ),
        ):
            continue
        if isinstance(
            chunk.content,
            str,
        ):
            yield chunk.content

    W tej funkcji jest wiele elementów, więc przetwarzaj ją krok po kroku.

    ID wątku wybiera rozmowę

    Pierwszy blok tworzy konfigurację uruchomienia:

    config = {
    "configurable": {
        "thread_id": f"user:{user.id}",
    }
}

    Checkpointer LangGraph przechowuje stan rozmowy pod kluczem thread_id. Każde uruchomienie z tym samym identyfikatorem wątku kontynuuje tę samą rozmowę, a jej wiadomości można później odczytać. Tutaj identyfikator wątku jest wywodzony z ID użytkownika, co oznacza, że każdy użytkownik ma dokładnie jedną rozmowę. To wystarcza do demonstracji; w rzeczywistej aplikacji należałoby tworzyć właściwe identyfikatory rozmów, umożliwiać kilka rozmów na użytkownika oraz sprawdzać przy każdej prośbie, czy osoba żądająca ma dostęp do danego wątku.

    Opis zakłada istnienie punktu sprawdzającego, ale żaden z fragmentów agenta go nie przechodzi. Bez niego thread_id nie ma żadnego wpływu i nic nie jest przechowywane pomiędzy zapytaniami. Stwórz jeden obiekt InMemorySaver w pliku llm/agent.py i przekaż go funkcji create_agent poprzez argument checkpointer, aby zarówno agent, jak i funkcje obsługujące historię mogły importować tę samą instancję.

    Kontekst wykonywania towarzyszy procesowi

    Następnie orkiestrator otacza użytkownika obiektem kontekstu:

    context = AgentContext(
    user=user,
)

    Obiekt ten jest przekazywany jako context= do funkcji agent.stream(). To właśnie w ten sposób tożsamość ustalona w FastAPI dociera do agenta, a stamtąd do narzędzi.

    Filtrowanie strumienia

    agent.stream() jest wywoływany z parametrem stream_mode="messages", co sprawia, że w miarę generowania tokenów przez model otrzymujemy pary składowych wiadomości oraz metadanych. Nie wszystko z tego strumienia powinno trafić do użytkownika. Pętla pomija wszystko, co nie jest wiadomością typu LangChain, pomija obiekty ToolMessage (surowy wynik narzędzia, np. pobrany tekst ręczny), zachowuje jedynie wiadomości AI oraz ich składowe, a ich treść jest zwracana w formie zwykłej ciągu znaków. Treść, którą niektórzy dostawcy przekazują jako listę części, jest cicho odrzucana podczas tej ostatniej weryfikacji, więc jeśli po zmianie modelu otrzymujesz puste odpowiedzi, warto tam sprawdzić.

    Krok 9: Zwracanie odpowiedzi w formie strumienia

    Decyzja o uczynieniu z chat_stream() generatora była celowa. Czekanie na pełną odpowiedź przed wysłaniem jednego bajtu zmusza użytkownika do patrzenia na ikonę spinowania, a odpowiedzi LLM mogą trwać kilka sekund. Transmisja strumieniowa pokazuje pierwsze słowa niemal natychmiast, co sprawia, że asystent wydaje się znacznie bardziej responsywny.

    StreamingResponse z FastAPI przyjmuje bezpośrednio generator. Zaktualizuj plik routers/chat.py:

    from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
from auth.dependencies import (
    MockUser,
    get_current_user,
)
from llm.orchestrator import chat_stream

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    return StreamingResponse(
        chat_stream(
            user_message=message,
            user=current_user,
        ),
        media_type="text/plain",
    )

    Odpowiedź jest wysyłana w formacie text/plain, przy czym każdy wygenerowany fragment jest zapisywany do połączenia zaraz po jego utworzeniu. Ponieważ chat_stream jest zwykłym (synchronicznym) generatorem, Starlette iteruje nad nim w wątku roboczym, dzięki czemu nie blokuje pętli zdarzeń. Jeśli później potrzebujesz ustrukturyzowanych zdarzeń po stronie klienta (na przykład aby pokazać komunikat „szukanie w instrukcji...” podczas działania narzędzia), Server-Sent Events stanowią naturalny następny krok.

    Pełna ścieżka żądania wygląda teraz tak: klient wysyła dane do /chat, FastAPI uwierzytelnia żądającego, chat_stream() uruchamia agenta, model decyduje, czy wezwać narzędzie, każde narzędzie jest wykonywane i zwraca swój wynik, model zapisuje odpowiedź, a tokeny są przesyłane z powrotem do klienta.

    Kluczową cechą jest to, że FastAPI nigdy nie uruchamia samego modelu. Router komunikuje się przez HTTP, orkiestrator kieruje agentem, agent decyduje, jakie informacje są mu potrzebne, a narzędzia zajmują się pobieraniem danych. Każda warstwa może ulegać zmianie bez wpływu na pozostałe.

    Krok 10: Odczytywanie historii rozmowy

    Ponieważ agent jest zapisywany w postaci checkpointów, jego stan jest przechowywany po każdej turze. Dzięki temu możliwe jest pokazanie powracającemu użytkownikowi jego poprzedniej rozmowy. Do tego zadania służą dwie funkcje pomocnicze.

    Odczytywanie surowego checkpointu

    Pierwsza funkcja ładowa najnowszy punkt kontrolny dla wątku i zwraca kanał messages, lub pusty list, jeśli wątek nigdy nie był używany:

    def get_conversation_messages(
    thread_id: str,
) -> list[BaseMessage]:
config: RunnableConfig = {
        "configurable": {
            "thread_id": thread_id,
        }
    }
    checkpoint = checkpointer.get(
        config,
    )
    if checkpoint is None:
        return []
    return checkpoint[
        "channel_values"
    ].get(
        "messages",
        [],
    )

    Jeśli skopiujesz ten kod, popraw wcięcie przy przypisaniu config: musi być ono umieszczone wewnątrz ciała funkcji, w przeciwnym razie Python wywoła błąd. Funkcja wymaga również importu klas BaseMessage, RunnableConfig oraz wspólnej instancji checkpointer.

    Zwracany jest surowy stan agenta, który obejmuje więcej informacji niż te, które użytkownik pamięta z rozmowy. Gdy agent wywołuje narzędzie, LangGraph rejestruje wiadomość AI zawierającą wywołanie narzędzia oraz oddzielną wiadomość z wynikiem. Są to szczegóły implementacji, których interfejs użytkownika nie powinien musieć interpretować.

    Kształtowanie wiadomości do wyświetlenia

    Druga funkcja tworzy widok przeznaczony dla użytkownika:

    def get_conversation_messages_for_display(
    thread_id: str,
) -> list[dict[str, str]]:
    display = []
    for message in get_conversation_messages(
        thread_id,
    ):
        if isinstance(
            message,
            HumanMessage,
        ):
            content = _extract_text_content(
                message.content,
            )
            if content.strip():
                display.append(
                    {
                        "type": "human",
                        "content": content,
                    }
                )
            continue
        if isinstance(
            message,
            AIMessage,
        ):
            content = _extract_text_content(
                message.content,
            )
            if content.strip():
                display.append(
                    {
                        "type": "ai",
                        "content": content,
                    }
                )
    return display

    Zachowuje jedynie obiekty HumanMessage i AIMessage, wydobywa ich tekst, odrzuca te puste i zwraca proste słowniki zawierające type oraz content. Komunikaty narzędziowych nigdy się nie pojawiają, ponieważ nie należą do żadnego z dwóch akceptowanych typów. Odrzucane są również puste komunikaty AI, co ma znaczenie, ponieważ taki komunikat, który jedynie żąda wywołania narzędzia, zazwyczaj nie zawiera tekstu.

    To rozwiązanie opiera się na funkcji pomocniczej _extract_text_content, która nie jest pokazana. Jej zadaniem jest zwrócenie treści bez zmian, jeśli jest to ciąg znaków, a jeśli jest to lista części treści – połączenie tych części w jeden tekst. Należy również zaimportować klasy HumanMessage i AIMessage.

    Endpunkt historii

    Ujawnij widok wyświetlany za pomocą ścieżki GET w pliku routers/chat.py. Pobiera on ten sam identyfikator wątku, który używa punkt końcowy czatu, i zwraca go wraz z wiadomościami.

    @chat_router.get("/current")
def get_current_conversation(
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    thread_id = (
        f"user:{current_user.id}"
    )
    messages = (
        get_conversation_messages_for_display(
            thread_id,
        )
    )
    return {
        "thread_id": thread_id,
        "messages": messages,
    }

    Interfejs użytkownika czatu może wywołać tę funkcję po otwarciu strony i wyświetlić istniejącą rozmowę, zanim użytkownik cokolwiek wpisze:

    GET /chat/current

    Odpowiedź wygląda w ten sposób:

    {
    "thread_id": "user:user-123",
    "messages": [
        {
            "type": "human",
            "content": "How much demo data do I have?"
        },
        {
            "type": "ai",
            "content": "You currently have 18 demo data."
        }
    ]
}

    Dlaczego nie zwracać surowego stanu?

    Wewnętrzny stan agenta oraz rozmowa widoczna dla użytkownika to różne rzeczy. W miarę dodawania nowych funkcjonalności stan ten gromadzi wywołania narzędzi, wyniki tych narzędzi, kroki pośrednie, metadane modelu oraz inne informacje pomocnicze. Zwracanie ich wszystkich powiązałoby interfejs użytkownika z wewnętrznymi mechanizmami agenta i mogłoby doprowadzić do wycieku informacji generowanych przez narzędzia, których nie zamierzało się pokazać. Serwer powinien określić, jakie informacje stanowią publiczną historię rozmów, i zwracać tylko je.

    Jak pojedynczy żądanie przepływa od początku do końca

    Gdy wszystko jest już na swoim miejscu, warto jasno określić jedną ważną kwestię: model nigdy nie ma dostępu do twojej bazy danych ani pliku PDF. Może jedynie poprosić o uruchomienie określonego narzędzia. Narzędzie to, działające jako zwykły kod Pythona z standardowymi mechanizmami kontroli dostępu, wykonywać dane operacje i zwraca tekst, który model wykorzystuje do sformułowania odpowiedzi. To właśnie ta granica zapewnia bezpieczeństwo asystenta podczas włączania go do aplikacji z rzeczywistymi danymi.

    Próba w Swagger UI

    FastAPI automatycznie generuje interaktywną dokumentację, więc podczas testowania nie ma potrzeby używania oddzielnego klienta. Uruchom serwer:

    python -m uvicorn main:app --reload

    Następnie otwórz interaktywną dokumentację API dostępną pod adresem /docs, ustaw nagłówek X-Demo-User i spróbuj trzech interakcji:

    • Pytanie dotyczące danych, na przykład pytanie o to, jakie dane demonstracyjne posiadasz. Agent powinien zdecydować, że potrzebuje narzędzia demonstracyjnego, je wywołać i odpowiedzieć na podstawie powróconych szczegółów projektu. W pełnej aplikacji to samo narzędzie sprawdza rzeczywiste rekordy użytkownika.
    • Pytanie dotyczące dokumentacji, na przykład pytanie o to, kto może otrzymywać dane demonstracyjne. Agent powinien wykonać ręczne wyszukiwanie, odnaleźć odpowiednią zapisaną regułę i odpowiedzieć, że tylko użytkownicy administracyjni mogą to robić.
    • Koniec punktu historii, GET /chat/current, który powinien zwracać odpowiedzi człowieka i AI dotyczące poprzednich dwóch pytań, bez żadnych komunikatów z narzędzia.

    Jeśli pierwsze dwa pytania dają odpowiedzi oparte na wynikach narzędzia, a trzecie pokazuje czysty zapis rozmowy, wszystkie warstwy funkcjonują prawidłowo.

    Zanim wdrożysz to w produkcji

    Demo celowo upraszcza kilka komponentów. To one należy ponownie przejrzeć, zanim rzeczywiści użytkownicy będą polegać na usłudze.

    Trwały stan rozmowy

    InMemorySaver jest doskonały do celów rozwojowych, ale wszystko, co przechowuje, znika po ponownym uruchomieniu procesu, a także nie może być udostępniane między kilkoma instancjami API za load balancerem. Należy użyć checkpointera wspieranego bazą danych lub innym trwałym nośnikiem pamięci, aby rozmowy przetrwały aktualizacje, a każda instancja widziała ten sam stan. Aby dowiedzieć się, co dokładnie przechowuje InMemorySaver i w jaki sposób, zapoznaj się z przewodnikiem na blogu dotyczącym sposobu, w jaki InMemorySaver organizuje checkpointy, dane tekstowe i pliki binarne.

    Rzeczywiste wdrożenie magazynu wektorów

    Lokalny katalog Chroma nadaje się do demonstracji całego procesu, ale nie stanowi infrastruktury produkcyjnej. Uruchom Chroma jako usługę trwałą lub przenieś się do zarządzanej bazy danych wektorowych pasującej do twojej architektury. Cokolwiek wybierzesz, musi być trwałe, mieć zapewnioną kopię zapasową i być dostępne z każdej instancji aplikacji.

    Indeksowanie pozostaje poza API

    Demo już prawidłowo podjęło jedną ważną decyzję: indeksowanie to odrębny skrypt, który uruchamiasz osobno, a API jedynie pobiera dane.

    python -m scripts.index_manual

    Server nigdy nie ładuje pliku PDF, nie dzieli go na fragmenty ani nie oblicza embeddingów przy starcie. Indeksowanie odbywa się offline; pobieranie danych jest częścią obsługi żądania. Rozdzielenie tych procesów oznacza, że API nie musi wykrywać zmian w dokumentacji ani niczego odbudowywać.

    W środowisku produkcyjnym należy przejść do następnego kroku i uruchamiać ten sam kod indeksowania jako dedykowaną zadanie pobierania danych, uruchamiane z pipeline’u implementacji, według harmonogramu lub przez procesor w momencie przesłania nowej dokumentacji. Architektura pozostaje taka sama; zadanie staje się jedynie zautomatyzowane, powtarzalne i możliwe do samodzielnego uruchomienia. Wtedy obowiązki dzielą się wyraźnie:

    • Zadanie pobierania danych: ładowanie dokumentów, dzielenie ich na fragmenty, włączanie tych fragmentów do struktury oraz aktualizacja magazynu wektorowego.
    • Służba FastAPI: przyjmowanie zapytań, wyszukiwanie odpowiednich fragmentów i generowanie odpowiedzi.
    • Magazyn wektorowy: przechowywanie zindeksowanych reprezentacji używanych podczas wyszukiwań.

    Pamiętaj o mechanizmie wczesnego zakończenia działania indeksatora przy jego automatyzacji; zadanie, które po cichu pomija ponowne indeksowanie, jest gorsze niż brak żadnego zadania.

    Jawny model embeddingu

    Zastosowanie domyślnej funkcji embeddingu w Chromie pozwala uniknąć dodatkowej konfiguracji w demonstracji, ale system produkcyjny powinien wybrać i skonfigurować swój model embeddingu w sposób jawny. Dzięki temu wyniki są powtarzalne, a użytkownik ma kontrolę nad jakością, kosztami, opóźnieniami oraz miejscem obliczania embeddingów. Istnieje jedna niepodważalna zasada: ten sam model embeddingu musi być używany zarówno do indeksowania, jak i do wyszukiwania. Jego zmiana oznacza konieczność ponownego indeksowania wszystkich danych.

    Obserwowalność i obsługa błędów

    Gdy agent jest już w użyciu, równie ważne jest sprawdzenie, co zrobił, jak jego poprawne działanie. Jeden żądanie może obejmować kilka wywołań modeli, jedno lub więcej wywołań narzędzi oraz krok pobierania informacji przed otrzymaniem ostatecznej odpowiedzi. Rejestrowanie tylko tej ostatecznej odpowiedzi niewiele mówi, gdy coś idzie nie tak. Należy zinstrumentalizować cały proces:

    • Wywołania narzędzi: które narzędzia zostały użyte, z jakimi argumentami oraz ile czasu zajęło każde z nich.
  • Czas oczekiwania modelu: czas trwania każdej prośby do LLM.
  • Użytkowanie tokenów i koszty: istotne, gdy jedno wiadomość od użytkownika może wywołać kilka wywołań modelu.
  • Błędy narzędzi: narzędzia powinny zwracać kontrolowane komunikaty o błędach, zamiast powodować awarię prośby.
  • Błędy dostawcy: należy sprawnie radzić sobie z ograniczeniami szybkości, czasem wygaśnięcia i niedostępnymi modelami, najlepiej z użyciem modelu awaryjnego.
  • Śledzenie wykonywania: rejestrowanie uporządkowanej sekwencji wywołań modelu, narzędzi oraz odpowiedzi dla każdego uruchomienia.
  • Jakość pobierania: jeśli ręczne wyszukiwanie nadal zwraca nieistotne fragmenty, przyczyną jest częściej sposób dzielenia na fragmenty, embeddingi, zapytanie lub ustawienia pobierania, a nie sam LLM.

    • Celem jest to, aby agent nigdy nie był „czarną skrzynką”. W każdym przypadku powinieneś być w stanie określić, co zrobił, jakie narzędzia użył, ile czasu zajęło każdy krok oraz gdzie napotkał błąd. Konkretna lista narzędzi zależy od Twojej architektury, ale zasada pozostaje niezmienna.

    Główne wnioski

    • Nie potrzebujesz dużej platformy AI, aby dodać przydatnego asystenta do aplikacji wymagającej obsługi złożonych domen. Zacznij od najmniejszego zestawu elementów, które rozwiązują rzeczywisty problem.
    • Traktuj wyszukiwanie dokumentacji jako jedno z narzędzi wśród innych. Dzięki temu agent ma jeden, spójny sposób dostępu zarówno do wiedzy o produkcie, jak i do danych użytkownika.
    • Zachowuj informacje o tożsamości w kontekście działania aplikacji, a nie w wiadomościach. Użytkownik jest powiązany z żądaniem, a narzędzia powinny je odczytywać stamtąd.
    • Rozdzielaj komponenty HTTP, orkiestrację, agenta oraz narzędzia. Każda warstwa powinna być prosta, a dodanie nowej funkcjonalności oznacza dodanie odpowiedniego narzędzia.
  • Stan zapisywany w punktach kontrolnych dostarcza historię niemal za darmo, ale w formie uporządkowanej prezentacji, a nie surowego stanu agenta.
  • Przepływ odpowiedzi, indeksowanie offline, fiksacja modelu embedding oraz monitorowanie każdego kroku przed przybyciem rzeczywistych użytkowników.

    • Literatura pokrewna

    Wszystkie artykułyWyślij brief