Головна / Статті / RAG: від первинних принципів до трьох основних агентів

rag embeddings bedrock langgraph retrieval aws

RAG: від первинних принципів до трьох основних агентів

Чому ваги не працюють з приватними даними, як функціонують ембеддинги та метод косинусного пошуку, а також три етапи розвитку агентів від керованої бази знань до самокерованого отримання даних.

4464 слів

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

Чому ваги самі по собі не допомагають

Три структурні обмеження заважають роботі систем Q&A в корпораціях:

Приватні корпуси даних ніколи не використовувалися під час навчання. Правила зберігання та макроси не є публічним веб-текстом.
Час зупиняється на момент заморожування. Останні зміни в законодавстві недоступні для замороженого знімка.
Мовчання не є стандартним станом. Прогнозування наступного токена віддає перевагу правдоподібному продовженню перед чесним невідомим.

Три способи внесення знань — та які з них ефективні

Варіант 1: Доопрацювання на корпусі даних

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

Варіант 2: Вставка всього в інструкції системи

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

Варіант 3: Вставка лише важливої інформації

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

Що таке RAG з початку до кінця

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

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

Вектори та ембеддинги без містичності

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

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

Арифметика схожості

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

Створення трьох повноцінних агентів на основі керованого хмарного інференсу

Решта цього посібника реалізує ці ідеї через три прогресивні агента на керованих моделях у стилі AWS Bedrock: відібрану базу знань, оркестратор LangGraph та самокерований канал вхідних даних. Кожен опис є повним скетчем агента — сприймайте його як виконувану архітектуру, а не як уривки коду.

Агент 1 — шлях до керованої бази знань

"""
AGENT 1: Managed Knowledge Base. The 30-line version.
Bedrock does the chunking, embedding, indexing and searching. You wire it up.
"""
import boto3
from langchain_aws import AmazonKnowledgeBasesRetriever, ChatBedrockConverse
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

REGION = "eu-west-1"
KB_ID = "ABCD1234EF"

retriever = AmazonKnowledgeBasesRetriever(
    knowledge_base_id=KB_ID,
    region_name=REGION,
    retrieval_config={
        "vectorSearchConfiguration": {
            "numberOfResults": 5,
            "overrideSearchType": "HYBRID",
        }
    },
)

llm = ChatBedrockConverse(
    model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name=REGION,
    temperature=0,
    max_tokens=1024,
)

PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "Answer only from the numbered context. Cite every claim as [n]. "
     "If the context does not contain the answer, say so plainly."),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])


def format_docs(docs) -> str:
    return "\n\n".join(
        f"[{i}] source={d.metadata.get('location', {}).get('s3Location', {}).get('uri', '?')}\n"
        f"{d.page_content}"
        for i, d in enumerate(docs, start=1)
    )


chain = (
    RunnableParallel(context=retriever | format_docs, question=RunnablePassthrough())
    | PROMPT
    | llm
    | StrOutputParser()
)

if __name__ == "__main__":
    print(chain.invoke("How fast must we submit a shortlist for a niche engineering role?"))

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

Агент 2 — оркестрація за допомогою LangGraph

"""
AGENT 2: LangGraph agent that decides WHETHER and WHICH index to search.
Retrieval is a tool. Identity travels in typed context, never in the prompt.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any

import boto3
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
from langchain.tools import ToolRuntime
from langchain_aws import ChatBedrockConverse
from langchain_core.tools import tool

REGION = "eu-west-1"
POLICY_KB = "ABCD1234EF"
ENGINEERING_KB = "WXYZ5678GH"

agent_runtime = boto3.client("bedrock-agent-runtime", region_name=REGION)


@dataclass
class RequestContext:
    """Built from validated JWT claims. The model cannot read or forge it."""
    subject: str
    roles: tuple[str, ...]
    region: str


def _retrieve(kb_id: str, query: str, ctx: RequestContext, k: int = 5) -> str:
    """Bedrock Retrieve API with a server-side metadata filter for tenant + residency."""
    resp = agent_runtime.retrieve(
        knowledgeBaseId=kb_id,
        retrievalQuery={"text": query},
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                "numberOfResults": k,
                "overrideSearchType": "HYBRID",
                "filter": {
                    "andAll": [
                        {"in": {"key": "acl_roles", "value": list(ctx.roles)}},
                        {"in": {"key": "region", "value": [ctx.region, "GLOBAL"]}},
                    ]
                },
            }
        },
    )
    results: list[dict[str, Any]] = resp.get("retrievalResults", [])
    if not results:
        return "NO_AUTHORISED_RESULTS: nothing in this index that you may read."
    return "\n\n".join(
        f"[{i}] score={r.get('score'):.3f} "
        f"source={r.get('location', {}).get('s3Location', {}).get('uri', '?')}\n"
        f"{r['content']['text']}"
        for i, r in enumerate(results, start=1)
    )


@tool
def search_policy_docs(query: str, runtime: ToolRuntime[RequestContext]) -> str:
    """Search HR, compliance and commercial policy.

    Use for shortlist SLAs and service credits, candidate data retention and consent,
    GDPR erasure, statement-of-work approval thresholds, background check rules and
    automated decision governance.
    """
    return _retrieve(POLICY_KB, query, runtime.context)


@tool
def search_engineering_docs(query: str, runtime: ToolRuntime[RequestContext]) -> str:
    """Search platform runbooks and integration documentation.

    Use for the requisition sync service, Workday and SuccessFactors webhooks, Kafka
    topics and event schemas, idempotency and retry behaviour, and vector index refresh.
    """
    return _retrieve(ENGINEERING_KB, query, runtime.context)


agent = create_agent(
    model=ChatBedrockConverse(
        model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
        region_name=REGION,
        temperature=0,
    ),
    tools=[search_policy_docs, search_engineering_docs],
    system_prompt=(
        "You are an enterprise knowledge assistant.\n"
        "- Answer only from tool results and cite the [n] markers they return.\n"
        "- If a tool returns NO_AUTHORISED_RESULTS, say you cannot access a source and "
        "name the document owner. Never fill the gap from memory.\n"
        "- For questions spanning policy and platform, call both tools before answering."
    ),
    context_schema=RequestContext,
    middleware=[
        ModelCallLimitMiddleware(thread_limit=8, run_limit=6, exit_behavior="end"),
        ToolCallLimitMiddleware(thread_limit=6, run_limit=5, exit_behavior="end"),
    ],
)

if __name__ == "__main__":
    claims = {"sub": "asha@example.com", "roles": ["recruiter"], "region": "EU"}
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "How is the vector index refreshed?"}]},
        context=RequestContext(
            subject=claims["sub"],
            roles=tuple(claims["roles"]),
            region=claims["region"],
        ),
    )
    print(result["messages"][-1].content)

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

Агент 3 — самокерований процес отримання та збору даних

"""
AGENT 3: Self-managed pipeline: your chunking, your index, your fusion, your rerank.
Chosen when Bedrock KB's fixed options are not enough (custom fusion, per-stage timing,
an abstention gate, or a chunking rule the managed service will not express).
"""
from __future__ import annotations
import time
from typing import Annotated, Any, TypedDict

import boto3
from langchain_aws import BedrockEmbeddings, ChatBedrockConverse
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from langgraph.graph import END, START, StateGraph

REGION = "eu-west-1"
CONFIDENCE_FLOOR = 0.34
CANDIDATE_K, CONTEXT_K = 20, 4

embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0",
    region_name=REGION,
    model_kwargs={"dimensions": 1024, "normalize": True},
)
llm = ChatBedrockConverse(
    model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name=REGION,
    temperature=0,
)
rerank_client = boto3.client("bedrock-agent-runtime", region_name=REGION)


# ---------------------------------------------------------------- ingest
def chunk_policy_document(markdown: str, source: str, acl: list[str], region: str):
    """Structure-aware first, size-capped second. Headings are the natural unit for
    policy text; the recursive splitter only intervenes on oversized sections."""
    by_heading = MarkdownHeaderTextSplitter(
        headers_to_split_on=[("#", "title"), ("##", "section"), ("###", "clause")]
    ).split_text(markdown)

    capper = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)
    chunks: list[Document] = []
    for part in capper.split_documents(by_heading):
        heading = " > ".join(
            v for k, v in part.metadata.items() if k in ("title", "section", "clause")
        )
        chunks.append(
            Document(
                # contextual retrieval: the chunk carries its own scope
                page_content=f"{heading}\n{part.page_content}",
                metadata={**part.metadata, "source": source, "acl": acl, "region": region},
            )
        )
    return chunks


# ---------------------------------------------------------------- state
class State(TypedDict, total=False):
    question: str
    roles: list[str]
    region: str
    candidates: list[Document]
    context_docs: list[Document]
    confidence: float
    answer: str
    timings: Annotated[dict[str, float], lambda a, b: {**(a or {}), **(b or {})}]


def _ms(name: str, t0: float) -> dict[str, float]:
    return {name: round((time.perf_counter() - t0) * 1000, 1)}


# ---------------------------------------------------------------- nodes
def retrieve(state: State, *, store) -> State:
    """Hybrid search with an ACL pre-filter pushed into the index query."""
    t0 = time.perf_counter()
    docs = store.similarity_search(
        state["question"],
        k=CANDIDATE_K,
        filter={"acl": {"$in": state["roles"]}, "region": {"$in": [state["region"], "GLOBAL"]}},
    )
    return {"candidates": docs, "timings": _ms("retrieve", t0)}


def rerank(state: State) -> State:
    """Bedrock Rerank: a cross-encoder scoring each (query, chunk) pair jointly."""
    t0 = time.perf_counter()
    cands = state["candidates"]
    if not cands:
        return {"context_docs": [], "confidence": 0.0, "timings": _ms("rerank", t0)}

    resp = rerank_client.rerank(
        queries=[{"type": "TEXT", "textQuery": {"text": state["question"]}}],
        sources=[
            {"type": "INLINE", "inlineDocumentSource": {
                "type": "TEXT", "textDocument": {"text": d.page_content}}}
            for d in cands
        ],
        rerankingConfiguration={
            "type": "BEDROCK_RERANKING_MODEL",
            "bedrockRerankingConfiguration": {
                "numberOfResults": CONTEXT_K,
                "modelConfiguration": {
                    "modelArn": f"arn:aws:bedrock:{REGION}::foundation-model/amazon.rerank-v1:0"
                },
            },
        },
    )
    ranked = resp["results"]
    top = [cands[r["index"]] for r in ranked]
    return {
        "context_docs": top,
        "confidence": round(float(ranked[0]["relevanceScore"]), 3),
        "timings": _ms("rerank", t0),
    }


def generate(state: State) -> State:
    """Abstain below the floor rather than answering from weak context."""
    t0 = time.perf_counter()
    if not state["context_docs"] or state["confidence"] < CONFIDENCE_FLOOR:
        return {
            "answer": "I cannot ground an answer for that in a source you are permitted "
                      "to read. Routing to the document owner.",
            "timings": _ms("generate", t0),
        }
    context = "\n\n".join(
        f"[{i}] {d.metadata.get('source')}\n{d.page_content}"
        for i, d in enumerate(state["context_docs"], start=1)
    )
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Answer only from the numbered context and cite each claim as [n]."),
        ("human", "Context:\n{context}\n\nQuestion: {question}"),
    ])
    reply = (prompt | llm).invoke({"context": context, "question": state["question"]})
    return {"answer": reply.content, "timings": _ms("generate", t0)}


def build_graph(store):
    g = StateGraph(State)
    g.add_node("retrieve", lambda s: retrieve(s, store=store))
    g.add_node("rerank", rerank)
    g.add_node("generate", generate)
    g.add_edge(START, "retrieve")
    g.add_edge("retrieve", "rerank")
    g.add_edge("rerank", "generate")
    g.add_edge("generate", END)
    return g.compile()

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

Вибір між трьома варіантами

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

Оцінка, яка дійсно покращує якість

Відстежуйте Recall@k та Precision@k на замороженому наборі запитань, а також точність відповідей до отриманих фрагментів. Попарюйте кожну зміну розміру фрагментів чи алгоритму вбудовування з показниками до та після змін. Відмовляйтесь від використання, якщо найкращі фрагменти є слабкими — калібрована тиша краща за бездумну вигадку.

Коли не варто використовувати RAG

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

Підсумок

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

Чек-лист для експлуатації: (1) один інструмент для вбудовування даних для індексу та запитів, (2) метадані для ACL та дати з першого дня, (3) гібридний пошук у разі наявності ідентифікаторів, (4) переранжування даних, коли важлива точність, (5) ключові запитання в процесі CI, (6) тригери для перебудови індексу при змінах джерел. Сприймайте простір знань як продукт із власниками, SLO та механізмами скасування змін — а не як індекс на вихідних, за яким ніхто не стежить. Коли графи агентів зростають, вимірюйте вартість токенів разом із якістю відповідей; оркестрація, яка споживає у п’ятнадцять разів більше ресурсів, ніж простий чат, не підвищуючи точності, є регресією під виглядом архітектури. Тримайте посібники для ситуацій з порожніми результатами, проблем із застарілим індексом та неузгодженостями в інструментах вбудовування даних, щоб спеціалісти могли діяти за планом, а не імпровізувати на основі журналів чатів.

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

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

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

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

Бюджет на використання API для індексації слід виділяти окремо від бюджету на інтерактивне обслуговування, щоб фінансовий відділ бачив обидві статті витрат.

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

Для систем з кількома користувачами перевіряйте, чи не можна обійти фільтри шляхом вставки запитів під приводом виконання обов’язків адміністратора.

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

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

Навчіть персонал підтримки читати інформацію про посилання перед тим, як передавати заявки про «неправильного бота» на вищий рівень.

Документуйте кожне джерело корпусу з вказанням власника та частоти оновлення; безпідтримувані контейнери стають причиною прихованого зіпсування даних.

Віддавайте перевагу ранньому перекриттю сегментів; використовуйте семантичні розділювачі лише тоді, коли оцінки покажуть, що фіксовані вікна не ефективні.

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

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

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

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

Навчіть персонал підтримки читати панелі посилань перед тим, як передавати заявки про «неправильного бота» на вищий рівень.

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

Віддавайте перевагу ранньому перекриттю сегментів; використовуйте семантичні розділювачі лише тоді, коли оцінки покажуть, що фіксовані інтервали не ефективні.

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

Бюджетуйте витрати на API для індексації окремо від витрат на інтерактивне обслуговування, щоб фінансовий відділ бачив обидві статті витрат.

Шаблони запитів версії поруч із індексними збірками; зміна запиту без повторної оцінки — це спосіб появи регресій.

Для систем з кількома користувачами перевірте, чи не можна обійти фільтри шляхом введення спеціальних запитів під приводом виконання обов’язків адміністратора.

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

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

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

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

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

Усі статті Надіслати бриф