Вбудування агента LangChain у FastAPI: інструменти, пошук у інструкціях, потоковий формат відповідей.

Коли додаток переростає кілька екранів, його документація починає розростатися, і користувачі перестають її читати. 20-сторінковий посібник, який пояснює функції, налаштування та правила домену, є корисним, але лише тоді, коли люди можуть знайти в ньому відповідь без пошуків. Допоміжний інструмент, вбудований у продукт, може подолати цю проблему: він відповідає на запитання «Як це працює?» з посібника та «Що є у моєму проекті?» з власних даних додатку.

Цей посібник детально описує компактну, функціональну версію такого асистента. Ви під’єднаєте агента LangChain до сервісу FastAPI, надасте йому один інструмент для читання даних програми та другий інструмент для пошуку PDF-інструкцій, збережених у ChromaDB, передасте автентифікованого користувача агенту за запитом, будете зберігати історію розмов за допомогою checkpointer від LangGraph та передаватимете відповідь назад клієнту. По дорозі ми вказуємо на недоліки мінімального коду, які потрібно усунути для його безперешкодної роботи, та на зміни, які необхідно внести перед впровадженням у продакшн.

Сценарій та складові

• Питання щодо самого продукту: що робить певна функція, де знаходиться налаштування, яке правило застосовується. Відповіді є в документації.
• Питання щодо власної роботи користувача: який інвертор він обрав, скільки енергії генерує його система, які дахи він обслуговує. Відповіді зберігаються в базі даних додатку, і жодна модель за замовчуванням їх не знає.

Технологія Retrieval-Augmented Generation (RAG) обробляє перший тип питань: індексується посібник, при надходженні запитання знаходяться відповідні уривки, які потім передаються моделі як контекст. Інструменти обробляють другий тип питань: це невеликі функції, які модель може викликати для отримання даних з додатку. Поєднавши ці два підходи, отримуємо асистента, який може як пояснювати продукт, так і аналізувати конкретний проект.

Справжня система, що стоїть за цим сценарієм, має набагато більше інструментів та значно складнішу логіку домену. Те, що наведено далі, свідомо спрощено до основних елементів, щоб архітектура залишалася зрозумілою:

• FastAPI надає HTTP API та визначає, хто є запитувачем.
• Агент LangChain (що працює на LangGraph) керує циклом міркувань та станом розмови.
• LLM інтерпретує кожен запит та вирішує, чи потрібна зовнішня інформація.
• Інструменти надають агенту контрольований доступ до функціоналу додатку.
• RAG дозволяє агенту шукати інформацію в документації.
• ChromaDB зберігає фрагменти посібника та виконує векторний пошук.
• Стрімінг передає токени клієнту під час генерації відповіді.

Перевага цієї структури полягає у тому, що асистент працює всередині існуючого фул-стек додатку з реальними користувачами та реальними даними, а не існує окремо як звичайний чат-бот.

Структура проекту

Кожна функція має власний пакет: маршрутизація HTTP, автентифікація, логіка агента, інструменти та конвеєр RAG. Це дозволяє зберігати кожен файл невеликим та чітко визначати, де має знаходитися нова функціональність.

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)

Для чого призначена кожна частина:

• main.py створює додаток FastAPI.
• auth/ містить імітовану залежність для автентифікації.
• routers/ містить кінцеві точки HTTP.
• llm/ містить усе, що стосується агента.
• llm/tools/ містить функції, які агент може викликати.

llm/rag/ містить конвеєр отримання даних, розділений на ingestion/ (завантаження, розділення на частини та індексування PDF) та retrieval/ (запит до ChromaDB).

scripts/ містить команди, які ви виконуєте вручну, наприклад, індексування посібника.

docs/ містить вихідний PDF.

chroma_data/ генерується локально та ніколи не повинен бути комітованим чи розгорнутим.

Ви не створюватимете все це одночасно. Порядок створення: шар API, потім модель, потім інструменти, потім конвеєр RAG, потім контекст, стрімінг та історія.

Крок 1: Шаблон FastAPI зі штучним користувачем

Почніть з встановлення всього, що потрібно проєкту. У списку передбачені веб-сервер, LangChain та LangGraph, клієнт чату сумісний з OpenAI, ChromaDB, завантажувач PDF та інструменти для розділення тексту, а також python-dotenv для налаштувань.

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

До моделі можна буде отриматися через OpenRouter, тому створіть файл .env у корені проекту, який міститиме ключ API.

OPENROUTER_API_KEY=your_api_key_here

Негайно додайте .env до файлу .gitignore. Ключ, який потрапляє під контроль версій, слід вважати витеклим.

Точка входу до додатку

main.py залишається дуже простим. Він створює додаток та реєструє маршрутизатор чату; тут не повинно бути нічого, що стосується моделі чи агента.

from fastapi import FastAPI
from routers.chat import chat_router

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

Перший кінцевий пункт чату

У файлі routers/chat.py визначте маршрутизатор під префіксом /chat із одним маршрутом POST. Наразі він лише відтворює надіслане повідомлення, що достатньо для підтвердження правильної роботи системи перед використанням ШІ.

from fastapi import APIRouter

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

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

Зауважте, що у маршруті POST без моделі тіла параметр message: str змушує FastAPI читати його з рядка запиту. Це зручно для тестування у Swagger UI, але для справжнього клієнта зазвичай використовується JSON-тіло, визначене за допомогою моделі Pydantic, оскільки рядки запиту потрапляють до журналів доступу та мають практичні обмеження за довжиною.

Штучна залежність автентифікації

У продакшн-додатку було б перевірено кукю сесії або JWT та завантажено користувача з бази даних. Тут усе це замінюється власним заголовком. Створіть файл auth/dependencies.py з невеликою дата-класом MockUser та функцією get_current_user, яка читає заголовок X-Demo-User та відхиляє запит із кодом 401, якщо його відсутньо.

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,
    )

Ін’єкція залежностей у FastAPI тепер передає користувача до кінцевої точки. Достатньо лише оголосити параметр за допомогою 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,
    }

Клієнт ідентифікує себе, надсилаючи заголовок у такому вигляді:

X-Demo-User: user-123

За кожним запитом FastAPI спочатку виконує get_current_user(), а потім передає отриманий об’єкт MockUser у функцію ask_ai. Важливою частиною проектування є розподіл обов’язків: FastAPI відповідає за автентифікацію, а шар ШІ просто отримує об’єкт користувача, якому може довіряти. Пізніше саме цей об’єкт користувача дозволяє інструментам повертати дані, що належать відповідній особі. Заміна мок-об’єкта на справжню систему автентифікації пізніше впливає лише на цю одну залежність.

Крок 2: Підключення моделі через OpenRouter

З наявністю функціонального кінцевої точки та відомого викликача сервісу потрібна модель. OpenRouter надає API, сумісне з OpenAI, тому клас ChatOpenAI від LangChain може працювати з ним, якщо вказати base_url на OpenRouter та передати свій ключ OpenRouter.

Розмістіть цей код у файлі llm/provider.py. Він завантажує файл .env, швидко повідомляє про помилку, якщо ключ відсутній, та обгортає ключ за допомогою класу SecretStr від Pydantic, щоб він випадково не потрапляв у журнали подій чи результати виведення.

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",
)

Оскільки ключ походить з середовища, він ніколи не з’являється у вихідному коді. На цьому етапі ви вже можете надсилати запити до моделі та отримувати відповіді, але це звичайний виклик LLM. Метою є агент, який може самостійно вирішувати, коли йому потрібен інструмент.

Вибір моделі

Аргумент model — це просто ідентифікатор моделі OpenRouter, тож ви можете замінювати моделі, не торкаючись решти коду. При порівнянні варіантів перевірте:

• підтримку викликів інструментів, від яких залежить агент;
• підтримку стрімінгу;
• розмір вікна контексту;
• обмеження швидкості обробки;
• наявність безкоштовного тарифу.

OpenRouter пропонує деякі моделі безкоштовно, що зручно під час експериментування. Каталог регулярно оновлюється, тому перегляньте поточний список та відфільтруйте безкоштовні моделі, замість того щоб покладатися на фіксовані рекомендації. Що б ви не обрали, це одразу потрапляє до конструктора:

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

Наприклад, якщо в каталозі вказаний ідентифікатор, подібний до наведеного нижче (приклад взятий на момент написання; він може бути недоступний), ви передасте саме цей рядок як model:

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

Пам’ятайте, що безкоштовні моделі використовують спільну пропускну здатність. Їх робота може бути обмежена чи вони можуть стати недоступними, зазвичай у найгірший момент під час демонстрації. Зміна на іншу модель або використання власного ключа постачальника через OpenRouter зазвичай допомагає вирішити цю проблему. Для продакшну обирайте моделі за надійністю, функціоналом, затримкою та вартістю, а не лише за ціною.

Крок 3: Від моделі до агента

Прямий виклик моделі — це один етап: текст користувача надходить, а потім отримується результат. Агент додає ланцюжок прийняття рішень. Модель аналізує запит, вирішує, чи може негайно відповісти, чи їй потрібні додаткові дані, за потреби викликає інструмент, читає результат тільки після цього формує остаточну відповідь. Приблизно:

• простий виклик: користувач → LLM → відповідь;

агент: користувач, потім агент, після чого LLM вирішує, що потрібно, далі — виклик інструменту або пошук за потреби, і нарешті — відповідь.

Модуль агента

У файлі llm/agent.py функція create_agent з LangChain створює агента на основі моделі та системного запиту. Під спудом вона генерує граф LangGraph, який автоматично виконує цикл модель-інструменти.

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,
)

Цей агент поки що не має інструментів, тому поводиться майже як проста модель. Спочатку потрібно дати йому інструкції.

Системний запит

У файлі llm/prompts.py зберігається короткий запит, який пояснює моделі її призначення, забороняє вигадувати дані та вказує, який тип запиту відповідає певному типу пошуку.

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()

Цей запит встановлює два джерела інформації:

• дані додатку (те, що знаходиться в обліковому записі користувача) надходять з інструментів додатку;
• Знання про програму (як працює продукт) отримуються шляхом пошуку в документації.

Одне уточнення перед тим, як рухатися далі. Інструменти та RAG розглядаються окремо, оскільки це полегшує їх розуміння, але у готовому агенті сам пошук у документації є інструментом. Іншого механізму немає: агент бачить список функцій, які можна викликати, а пошук у посібнику — це одна з них.

Крок 4: Перший інструмент

Інструмент — це функція, яку агент може викликати. Саме це дозволяє архітектурі масштабуватися: замість того, щоб вкладати всі дані програми у запит, ви надаєте доступ до конкретних операцій та дозволяєте моделі запитувати їх лише тоді, коли це необхідно для відповіді на запитання.

Для демонстрації llm/tools/demo_tool.py визначає інструмент, який повертає фіксований блок інформації про проект.

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()

Тут важливі два моменти. Декоратор @tool перетворює звичайну функцію Python на інструмент LangChain, беручи її назву та схему аргументів з підпису функції. А документація стає описом інструменту, який модель читає під час прийняття рішення про його виклик. У реальній системі такий опис потребує особливої уваги: необхідно точно описати, що повертає інструмент, коли це доцільно, а коли — ні. Нечіткий опис є однією з найпоширеніших причин того, чому агент викликає неправильний інструмент або взагалі жодного.

Щоб зареєструвати інструмент, передайте його до функції 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(). Якщо користувач запитує: «Що таке демо?», пошук не потрібен, і модель відповідає безпосередньо. Цей вибір здійснюється щоразу під час обробки запиту, у межах циклу агента.

Крок 5: Створення конвеєра RAG для посібника

Тепер агент може отримувати дані додатку, але все ще нічого не знає про те, як працює продукт. Вставка 20-сторінкового посібника у системний запит призведе до марної витрати токенів на кожен запит та ускладнить його оновлення. RAG допомагає уникнути обох проблем.

Варто бути точним щодо того, що таке RAG, а що ні. На документації нічого не тренується та не налаштовується. Під час запиту система шукає в посібнику уривки, які найбільш релевантні до запитання, та надає ці уривки моделі як контекст, а модель відповідає на основі них.

Процес складається з двох етапів:

1. Прийом даних, який виконується окремо від веб-додатку: завантажується PDF, розділяється на частини та зберігається разом із їхніми ембеддингами в ChromaDB.
2. Отримання інформації, яке виконується під час запиту: береться запитання, шукається інформація в ChromaDB, збираються найкращі частини та передаються моделі.

Якщо ви хочете отримати більш повне розуміння цих концепцій, огляд блогу про механізми отримання свіжих знань за запитом розглядає їх більш детально; тут ми зосередимося на самій реалізації.

Завантаження PDF

llm/rag/ingestion/loader.py використовує PyPDFLoader з бібліотеки LangChain, який перетворює кожну сторінку PDF на об’єкт 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

Об’єкт Document містить два елементи: витягнутий текст у полі page_content та словник metadata, який описує джерело цього тексту. Саме завдяки метаданим пізніше можливо посилатися на конкретну сторінку у відповіді. Концептуально кожна завантажена сторінка виглядає так:

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

PyPDFLoader зазвичай записує як індекс page, що йде від нуля, так і page_label, зрозумілий для людини. Будівник контексту нижче використовує page_label, який збігається з номерами сторінок, що відображаються у PDF.

Розділення сторінок на частини

Пошук усіх сторінок, не кажучи вже про всього документа як про один блок, дає грубі результати. llm/rag/ingestion/chunker.py розділяє документи за допомогою 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,
    )

Розділювач прагне створювати фрагменти по приблизно 1 000 символів із 150 символами перекриття. Список роздільників використовується у порядку: спочатку робиться перерву на межах абзаців, потім — на розділах рядків, далі — наприкінці речень, після цього — між пробілами, і лише у крайньому випадку — посеред слова. Перекриття існує тому, що один факт може охоплювати кілька точок розділення; повторення невеликої частини тексту з обох боків зменшує ймовірність того, що відповідне речення буде розрізане навпіл.

Ці цифри є вихідними точками, а не правилами. Оптимальний розмір залежить від того, як написані ваші документи та наскільки точним має бути пошук, тому сприймайте їх як значення, які потрібно налаштовувати відповідно до конкретних завдань. Стаття блогу про розділення на фрагменти для збереження доказів детальніше розглядає цей баланс.

Постійна колекція Chroma

llm/rag/ingestion/chroma.py відкриває об’єкт PersistentClient, який зберігає дані на диску та повертає колекцію документів, створюючи її при першому використанні.

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,
    )

Шляхи та назви походять з файлу llm/rag/config.py, який читає змінні середовища та використовує розумні стандартні значення у разі їх відсутності:

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",
)

Додайте відповідні записи до файлу .env:

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

Жодна модель ембеддингу не налаштована, і це є навмисним для демонстрації. Коли колекція створюється без явно вказаної функції ембеддингу, Chroma використовує свою вбудовану стандартну модель: щоразу, коли додаються документи, Chroma сама обчислює їхні ембеддинги та зберігає їх поруч із текстом та метаданими. Стандартна модель працює локально та завантажується при першому використанні, тому перший процес індексації може затриматися під час її завантаження.

Результатом є локальний, постійний сховище векторів у каталозі chroma_data/. Оскільки воно створюється повністю з PDF, додайте його до .gitignore разом із .env.

Завдання індексації

llm/rag/ingestion/indexer.py об’єднує кроки імпорту даних.

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."
    )

Розгляньмо, що він робить. Він відкриває колекцію та повертається раніше, якщо вона вже містить фрагменти даних, що робить повторні запуски безпечними. Він перевіряє наявність PDF, і в іншому випадку піднімає зрозумілу помилку. Потім він завантажує сторінки, ділить їх на фрагменти та додає все до Chroma за один виклик, використовуючи стабільні ідентифікатори (manual-chunk-0, manual-chunk-1 тощо), тексти фрагментів та їхню метадану. Повідомлення про прогрес відображають, скільки сторінок та фрагментів було оброблено.

Одна з проблем виникає через це передчасне повернення: якщо ви зміните посібник та знову запустите скрипт, нічого не станеться, оскільки колекція не є порожньою. Щоб врахувати зміни, потрібно спочатку видалити колекцію (або каталог chroma_data/) перед повторним індексуванням, або замінити захисний код на логіку, яка навмисно виконує операцію додавання чи перебудови.

У списку не відображається scripts/index_manual.py; йому достатньо імпортувати index_manual та викликати його. Запустіть його один раз як модуль з кореня проекту:

python -m scripts.index_manual

Цей єдиний запуск читає PDF, ділить його на фрагменти, вбудовує ці фрагменти та зберігає їх. Результати роботи терміналу відображають кількість оброблених фрагментів. У спрощеній демонстрації посібник — це односторінковий PDF із єдиним правилом: „Дані демонстрації можна надавати лише адміністраторам“, що достатньо, щоб показати, чи функція пошуку працює. Після цього запуск FastAPI зовсім не впливає на PDF, оскільки вектори вже збережені.

Запит до колекції

Одного лише індексування недостатньо для агента; йому потрібен спосіб пошуку. Файл llm/rag/retrieval/retriever.py обгортає API запитів Chroma.

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

Функція надсилає запит у вигляді query_texts, просить до п’яти результатів та запитує документи, їхню метадані та відстані. Chroma повертає по одному списку для кожного запиту, тому код читає індекс [0] кожного поля; умова or [] захищає від відсутності полів. Кожен результат пакується як клас RetrievedChunk, щоб решта коду не залежала від форми відповіді Chroma.

Оскільки запит інтегрований з тим самим моделлю, що й збережені фрагменти, пошук відбувається на семантичному рівні. Запит на кшталт „Як додати нові дані для демонстрації?“ знаходить уривки про створення чи надання таких даних, навіть якщо там ніколи не використовуються слова „додати нові“. Значення відстані показує, наскільки близьким є кожен результат; чим нижче значення, тим більша схожість. Цей показник корисний пізніше, якщо потрібно відкинути слабкі результати замість того, щоб завжди передавати моделі п’ять фрагментів.

З цим на місці функція пошуку в RAG працює. Залишається лише передати отримані результати моделі.

Перетворення фрагментів на контекст

llm/rag/context.py форматує знайдені результати у один рядок, який може прочитати модель.

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,
    )

Кожен фрагмент супроводжується рядком коду з назвою посібника для користувача та його міткою сторінки, причому фрагменти розділені разделювачем. Саме цей рядок коду дозволяє моделі вказати, звідки походить відповідь, та надає користувачам можливість перевірити це.

Крок 6: Використання посібника як інструменту

Процес завершено: сторінки завантажуються та діляться на фрагменти, фрагменти зберігаються в ChromaDB, можна знаходити та форматувати потрібні з них. Однак агент не має жодного уявлення про те, що все це існує. Саме тут корисна попередня архітектурна примітка: пошук документації стає ще одним інструментом.

llm/tools/manual_tool.py визначає функцію search_user_manual, яка приймає запит, отримує п’ять фрагментів інформації та повертає їх у форматованому вигляді. Якщо жодних даних не отримано, функція повертає чітке повідомлення про те, що посібник не охоплює це питання, що дає моделі можливість передати чесну інформацію замість порожнього рядка.

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,
    )

Як і раніше, документація є описом цього інструменту для моделі. У ній зазначено, що цей інструмент слід використовувати для запитань щодо функціонування, інструкцій, правил, обмежень та принципів роботи системи, що узгоджується зі спонукальним повідомленням системи.

Тепер необхідно зареєструвати обидва інструменти в агенті:

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,
)

Зверніть увагу на оператор імпорту в цьому описі: він посилається на get_my_solar_system, назву з повної версії додатку, тоді як модуль демонстраційного інструменту визначає get_my_demo_data. Використовуйте get_my_demo_data як у операторі імпорту, так і в списку tools, інакше модуль не буде імпортуватися.

Чому підхід, заснований на інструментах, залишається ефективним у міру зростання додатку

Агенту ніколи не потрібен один величезний запит, який описував би все, що знає додаток. Натомість він отримує обмежені, контрольовані можливості. Для додавання функції до асистента потрібно створити новий інструмент та зареєструвати його; шар HTTP не змінюється. У демонстраційній версії зберігається рівно один інструмент для обробки даних та один інструмент для документації, щоб патерн був зрозумілим, проте у повноцінному додатку така сама структура підтримує значно більшу кількість інструментів.

Крок 7: Передача автентифікованого користувача до агента

Інструмент демонстраційного типу все ще повертає жорстко закодовані дані. Справжній інструмент має знати, хто ставить запит, і додаток вже це знає: FastAPI ідентифікував користувача за допомогою залежності автентифікації. Відсутньою є лише можливість передати цього користувача під час виконання агента. LangChain називає це контекстом виконання.

Визначте структуру контексту у файлі llm/context.py:

from dataclasses import dataclass
from auth.dependencies import MockUser

@dataclass
class AgentContext:
    user: MockUser

Цей об’єкт надається під час виклику агента, і це має значення. Користувач — це інформація, пов’язана з конкретним запитом. Вона належить до поточного HTTP-запиту, а не до розмови, і її ніколи не слід зберігати у вигляді повідомлення, яке модель може прочитати чи переписати. Також це означає, що запит не може змусити агента діяти як інший користувач. У повноцінному додатку цей самий об’єкт контексту також містить такі елементи, як сесія бази даних та ідентифікатор проекту, який ведеться до змін.

Код демонстрації зупиняється на визначенні класу, тож залишається два з’єднання, які потрібно створити, і варто перевірити актуальну документацію LangChain для отримання точної інформації про API. По-перше, необхідно оголосити схему під час створення агента, зазвичай за допомогою параметра context_schema=AgentContext у функції create_agent. По-друге, інструменти мають її читати: у LangChain 1.x інструмент може приймати параметр часу виконання (наприклад, позначений як ToolRuntime[AgentContext]) та читати дані користувача з його атрибута context, який залишається непомітним для моделі серед аргументів інструменту. Саме тут справжня функція get_my_demo_data була б шукає записи за значенням user.id.

Крок 8: Шар оркестрації для стрімування

Замість того, щоб викликати агента зсередини роутера, перенесіть взаємодію до файлу llm/orchestrator.py. Таким чином роутер залишається зосередженим на обробці HTTP-запитів, а оркестратор керує процесом перетворення повідомлення на виконання агента.

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

У цій функції багато коду, тому розглядайте її по частинах.

ID потоку вибирає розмову

Перший блок створює конфігурацію виконання:

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

Чекпоїнтер LangGraph зберігає стан розмови за ключем thread_id. Кожен запуск із тим самим ID потоку продовжує ту саму розмову, а її повідомлення можна прочитати пізніше. Тут ID потоку отримується з ID користувача, що означає, що кожен користувач має рівно одну розмову. Це підходить для демонстрації; у справжньому додатку мали б створюватися належні ID розмов, дозволятися кілька розмов на одного користувача, а також перевірятися при кожному запиті, чи користувач є власником потоку, до якого намагаються отримати доступ.

У описі передбачається наявність об’єкта checkpointer, але жоден з фрагментів агента не проходить перевірку. Без нього thread_id не має жодного ефекту, і нічого не зберігається між запитами. Створіть єдиний об’єкт InMemorySaver у файлі llm/agent.py та передайте його функції create_agent через аргумент checkpointer, щоб як агент, так і функції обробки історії могли імпортувати один і той самий екземпляр.

Контекст виконання супроводжує процес

Далі оркестратор обгортає дані користувача в об’єкт контексту:

context = AgentContext(
    user=user,
)

Цей об’єкт передається як context= до функції agent.stream(). Саме через цей шлях ідентичність, встановлена у FastAPI, надходить до агента, а через нього — до інструментів.

Фільтрація потоку

agent.stream() викликається з параметром stream_mode="messages", що забезпечує отримання пар із фрагментом повідомлення та метаданими у міру генерації токенів моделлю. Не все з цього потоку має потрапляти до користувача. Цикл ігнорує все, що не є повідомленням LangChain, ігнорує об’єкти ToolMessage (сирі результати роботи інструментів, такі як отриманий ручний текст), зберігає лише повідомлення від ШІ та їхні фрагменти, а також виводить їхній вміст у вигляді звичайного рядка. Вміст, який деякі постачальники надсилають у вигляді списку частин, беззвучно видаляється під час останньої перевірки, тож якщо ви зміните модель і побачите порожні відповіді, саме тут варто шукати причину.

Крок 9: Повернення потокової відповіді

Рішення зробити chat_stream() генератором було свідомим. Очікування повної відповіді перед надсиланням байту змушує користувача чекати, а відповіді ШІ можуть займати кілька секунд. Стрімування дозволяє показати перші слова майже миттєво, що робить асистента значно більш реактивним.

StreamingResponse з FastAPI безпосередньо приймає генератор. Оновіть 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",
    )

Відповідь надсилається у форматі text/plain, причому кожен елемент, отриманий генератором, записується у з’єднання відразу після його створення. Оскільки chat_stream є звичайним (синхронним) генератором, Starlette обробляє його у робочій потоці, тож це не блокує цикл подій. Якщо згодом вам знадобляться структуровані події на клієнті (наприклад, для відображення повідомлення „шукаю інструкцію...“ під час виконання інструменту), Server-Sent Events є природним наступним кроком.

Повний шлях запиту тепер виглядає так: клієнт надсилає дані на /chat, FastAPI автентифікує користувача, chat_stream() запускає агента, модель вирішує, чи потрібно викликати інструмент, будь-який інструмент виконується та повертає свій результат, модель записує відповідь, а токени надсилаються назад до клієнта.

Ключовою особливістю є те, що FastAPI ніколи не запускає саму модель. Рутер використовує протокол HTTP, оркестратор керує агентом, агент визначає, яку інформацію йому потрібно, а інструменти виконують завдання зі збору даних. Кожен шар може змінюватися без впливу на інші.

Крок 10: Читання історії розмови

Оскільки стан агента зберігається після кожного кроку, можливо показати повертаючомуся користувачеві його попередню розмову. Для цього використовуються дві допоміжні функції.

Читання сирого файлу збереження стану

Перша функція завантажує найновіший пункт перевірки для потоку та повертає канал messages, або порожній список, якщо потоку ніколи не використовували:

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",
        [],
    )

Якщо ви скопіюєте цей код, виправте відступи при присвоєнню config: вони мають знаходитися всередині тіла функції, інакше Python видасть помилку. Функція також потребує імпорту класів BaseMessage, RunnableConfig та спільної інстанції checkpointer.

Те, що повертається, — це первинний стан агента, який включає більше інформації, ніж ту, що пам’ятає користувач у чаті. Коли агент викликає інструмент, LangGraph записує повідомлення від ШІ з описом цього виклику та окреме повідомлення з результатом. Це деталі реалізації, які фронтенд не повинен інтерпретувати.

Формування повідомлень для відображення

Друга функція створює інтерфейс для користувача:

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

Вона зберігає лише об’єкти HumanMessage та AIMessage, витягує їхній текст, видаляє порожні елементи та повертає прості словники з полями type та content. Повідомлення інструментів ніколи не з’являються, оскільки вони не належать до жодного з двох допустимих типів. Також видаляються порожні повідомлення AI, що має значення, адже повідомлення AI, яке лише запитує виконання інструменту, зазвичай не містить тексту.

Цей процес ґрунтується на допоміжній функції _extract_text_content, яка не показана. Її завдання — повертати вміст без змін, якщо це рядок, а якщо це список частин вмісту — об’єднувати ці частини в один текст. Також необхідно імпортувати класи HumanMessage та AIMessage.

Ендпоїнт історії

Відображіть переглядову версію через маршрут GET у файлі routers/chat.py. Він отримує той самий ідентифікатор потоку, що використовується кінцевою точкою чату, та повертає його разом із повідомленнями.

@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,
    }

Інтерфейс чату може викликати цей код під час завантаження сторінки та відобразити наявний діалог ще до того, як користувач щось введе:

GET /chat/current

Відповідь виглядає так:

{
    "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."
        }
    ]
}

Чому не повертати первинний стан?

Внутрішній стан агента та діалог, який бачить користувач, — це різні речі. З додаванням нових функцій стан накопичує виклики інструментів, результати їх роботи, проміжні кроки, метадані моделі та іншу інформацію для обліку. Повернення всього цього призведе до зв’язку фронтенду з внутрішньою структурою агента та може призвести до витоку даних, які ви не хотіли показувати. Бекенд має визначити, що саме є публічною історією діалогу, та повертати лише це.

Як працює один запит від початку до кінця

Коли всі елементи на своїх місцях, варто чітко зрозуміти одну особливість: модель ніколи не торкається вашої бази даних чи PDF-файлу. Вона може лише запитати про виклик певного інструменту. Цей інструмент, який працює як звичайний код Python із звичайними механізмами контролю доступу, виконує необхідну операцію та повертає текст, а модель використовує цей текст для формування своєї відповіді. Саме цей бар’єр забезпечує безпеку асистента під час вбудовування в додаток із реальними даними.

Спроба використання у Swagger UI

FastAPI автоматично генерує інтерактивну документацію, тому під час тестування немає потреби у окремому клієнті. Запустіть сервер:

python -m uvicorn main:app --reload

Потім відкрийте інтерактивну документацію API, яка доступна за адресою /docs, встановіть заголовок X-Demo-User та спробуйте три взаємодії:

• Запит щодо даних, наприклад, про те, які дані демо у вас є. Агент має вирішити, чи потрібен інструмент демо, скористатися ним та відповісти на основі отриманих деталей проекту. У повноцінному додатку подібний інструмент запитує реальні записи користувача.
• Запит щодо документації, наприклад, про те, хто може отримувати дані демо. Агент має здійснити ручний пошук, знайти відповідне правило та відповісти, що це можуть робити лише адміністратори.
• Ендпоїнт історії, GET /chat/current, який має повертати записи розмов людини та ШІ щодо попередніх двох запитань без будь-яких повідомлень від інструментів.

Якщо перші два запити дають відповіді, засновані на результатах роботи інструменту, а третій — чистий текст розмови, то кожен етап функціонує належним чином.

Перш ніж впровадити у продакшн

У демо-версії навмисно спрощено кілька компонентів. Саме їх потрібно переглянути знову, перш ніж справжні користувачі почнуть покладатися на цю послугу.

Стабільний стан розмови

InMemorySaver ідеально підходить для розробки, але все, що він зберігає, зникає після перезапуску процесу, і його неможливо поширювати між кількома інстанціями API за допомогою балансувальника навантаження. Використовуйте чекпоїнти, підтримувані базою даних або іншим стабільним сховищем, щоб розмови зберігалися під час оновлень та щоб кожна інстанція бачила один і той самий стан. Щоб дізнатися, що саме зберігає інструмент InMemorySaver та як це відбувається, перегляньте пост у блогу про те, як InMemorySaver організовує чекпоїнти, записи та блоки даних.

Реальне розгортання сховища векторів

Локальний каталог Chroma підходить для демонстрації процесу обробки даних, але це не інфраструктура для продакшену. Запускайте Chroma як постійну службу або переходьте на керовану базу даних векторних даних, яка підходить до вашої технологічної стеку. Що б ви не обрали, це має бути постійним ресурсом, який підлягає архівуванню та є доступним з кожної інстанції додатку.

Індексування не входить до складу API

У демонстрації вже правильно прийнято одне важливе рішення: індексування — це окремий скрипт, який викликається окремо, а API лише здійснює отримання даних.

python -m scripts.index_manual

Сервер ніколи не завантажує PDF, не розділяє його на частини та не обчислює ембеддинги під час запуску. Індексування відбувається офлайн; отримання даних — це частина обробки запиту. Розділення цих процесів означає, що API не потребує виявлення змін у документації чи перебудови чого-небудь.

У продакшені переходьте до наступного кроку та запускайте той самий код індексації як окрему задачу обробки даних, яка запускається з пайплайну розгортання, за графіком або автоматично коли завантажується нова документація. Архітектура залишається незмінною; задача просто стає автоматизованою, повторюваною та може розгортатися незалежно. Тоді обов’язки чітко розділяються:

• Задача обробки даних: завантажує документи, розділяє їх на частини, вбудовує ці частини та оновлює сховище векторів.
• Сервіс FastAPI: приймає запити, знаходить відповідні частини документів та генерує відповіді.
• Сховище векторів: зберігає індексовані представлення даних, які використовуються під час пошуку.

Пам’ятайте про механізм раннього повернення в індексаторі під час його автоматизації; задача, яка мовчки пропускає процес переіндексації, є гіршою, ніж взагалі жодна задача.

Явна модель вбудовування

Використання стандартної функції вбудовування Chroma дозволяє уникнути зайвої налаштуваності в демо-версії, але у продакшн-системі слід явно обирати та налаштовувати модель вбудовування. Це забезпечує відтворюваність результатів та дає можливість контролювати якість, витрати, затримки та місце обчислення ембеддингів. Є одне непереговорне правило: для індексування та запитів має використовуватися одна й та сама модель ембеддингів. Зміна її означає необхідність повторного індексування всього.

Спостережливість та обробка помилок

Як тільки агент починає працювати, важливо бачити, що він робив, настільки ж, як і забезпечити його функціонування. Один запит може включати кілька викликів моделей, один або кілька викликів інструментів та крок отримання даних перед остаточною відповіддю. Логування лише остаточної відповіді майже нічого не повідомляє, коли щось йде не так. Необхідно задокументувати весь процес:

• Виклики інструментів: які інструменти були використані, з якими аргументами та скільки часу кожен з них зайняв.

Затримка моделі: тривалість кожного запиту до LLM.

Використання токенів та витрати: це важливо, коли одне повідомлення користувача може спричинити кілька викликів моделі.

Помилки інструментів: інструменти повинні повертати контрольовані повідомлення про помилки замість того, щоб зруйнувати запит.

Помилки постачальника: необхідно грамотно керувати обмеженнями швидкості, таймаутами та недоступними моделями, бажано за допомогою резервної моделі.

Відстеження виконання: фіксувати послідовність викликів моделі, інструментів та відповідей для кожного запуску.

Якість отримання даних: якщо ручний пошук продовжує повертати нерелевантні фрагменти, причиною частіше є спосіб розділення на фрагменти, ембеддинги, запит або налаштування пошуку, а не сама LLM.

Мета полягає у тому, щоб агент ніколи не був «чорною скринькою». Для кожного запуску ви повинні мати можливість сказати, що він зробив, які інструменти використав, скільки часу зайняв кожен крок та де він зазнав невдачі. Конкретні інструменти залежать від вашої технологічної стеку; принцип же залишається незмінним.

Ключові висновки

• Вам не потрібна велика платформа ШІ, щоб додати корисного асистента до застосунку з великою кількістю доменних функцій. Почніть з найменшого набору елементів, які дозволяють вирішити реальну проблему.
• Розглядайте пошук документації як один із інструментів серед інших. Так агент матиме єдиний, уніфікований спосіб отримання як інформації про продукт, так і даних користувача.
• Зберігайте інформацію про ідентичність у контексті виконання, а не у повідомленнях. Користувач належить до запиту, і інструменти повинні отримувати цю інформацію звідти.
• Розділяйте HTTP, оркестрацію, агента та інструменти. Кожен шар залишається простим, а додавання нової функціональності означає додавання нового інструменту.

Стан у форматі чекпоїнтів забезпечує вам історію дій майже безкоштовно, але повертає відфільтрований перегляд, а не первинний стан агента.

Транслюйте відповіді, індексуйте дані офлайн, фіксуйте модель ембеддингів та аналізуйте кожен крок ще до приходу справжніх користувачів.

Пов’язана література

• Гібридна пам’ять агента: поєднання BM25 та пошуку векторів з RRF у Python — Дізнайтеся, чому чистий пошук векторів не підходить як пам’ять агента, як Reciprocal Rank Fusion об’єднує результати BM25 та щільних пошуків у Python, та коли корисні резюме GraphRAG.
• Проектування чотирирівневої пам’яті агента з використанням LangGraph та Amazon Bedrock — Дізнайтеся, як надати агентам на основі LLM функції епізодичної, семантичної та процедурної пам’яті на платформах Bedrock та LangGraph, а також як захистити її від отруєння даними, витоку конфіденційної інформації та небажаних впливів.
• Дозвіл Gemini обирати джерело: FAISS, Tavily та прямі відповіді в LangGraph — Створіть просту робочу схему LangGraph, у якій Gemini направлятиме кожне запитання до бази знань FAISS, веб-пошуку Tavily або надаватиме пряму відповідь із структурованим форматом результату.
• Надсилання запитів між інструментами SQL та веб-пошуком за допомогою агента Gemini — Як агент для виклику інструментів LangChain у Vertex AI обирає між трьома інструментами SQLite для перетворення тексту на SQL та прямим веб-пошуком, а також про можливі проблеми з даними, залежностями та автентифікацією.
• Стрімування агентів LangGraph у React без витоку внутрішніх даних інструментів — Як серіалізувати результати вихідних даних LangGraph astream_events для Server-Sent Events, маскувати конфіденційну інформацію викликів інструментів у FastAPI та відображати зручні індикатори інструментів у React.
• Оцінка поведінки агента: результати, траєкторії та зворотний зв’язок щодо продукції — Як оцінювати системи-агенти не лише за кінцевою відповіддю: прив’язка LangChain, LangGraph та LangSmith до ролей, оцінка траєкторій та контексту, а також врахування проблем під час створення продукції у процесі оцінки.