AI Engineering Academy · leksjon

Implementere sentrale RAG- og agentfunksjoner

Bygg pipelinen for dokumentinntak, indeksering i vektorlagring, gjenfinning med omrangering og integrasjoner med agentverktøy etter mønstrene du har lært gjennom kurset.

Leksjon 2 av 413 trinn

Implementere sentrale RAG- og agentfunksjoner er en gratis leksjon i AI Engineering Academy på CoddyKit. Dette er leksjon 2 av 4. Du kan lese hele leksjonen gratis nedenfor – og deretter øve praktisk i nettleseren med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt. Den er en del av læringsløpet i AI Engineering Academy, og fremdriften din synkroniseres mellom nettet og CoddyKit-appen. Kurset i AI Engineering Academy inneholder totalt 4 leksjoner.

Rekkefølge for implementering og avhengigheter

Bygg systemet nedenfra og opp: Start med komponentene som ikke har eksterne avhengigheter, og legg deretter til komponenter som er avhengige av dem. For et RAG- og agentsystem er rekkefølgen: (1) skjema for vektorlageret, (2) inntakspipeline, (3) gjenfinner, (4) grunnleggende spørsmål-og-svar-kjede, (5) strømmeendepunkt, (6) agent med verktøy, (7) hurtigbufferlag og (8) sporingsinstrumentering. Test hver komponent isolert før De integrerer den i pipelinen.

# Build order:
IMPL_ORDER = [
    'pgvector_schema',      # prerequisite for everything
    'document_ingestion',   # populate the vector store
    'hybrid_retriever',     # test retrieval in isolation
    'qa_chain_basic',       # integrate LLM with retrieval
    'streaming_endpoint',   # expose via API
    'function_calling',     # add agent tool calls
    'semantic_cache',       # reduce repeat API calls
    'langsmith_tracing',    # add after core works
    'injection_filter',     # harden before load testing
]

Sette opp vektorlageret

Opprett pgvector-tabellen med riktig skjema før De tar inn dokumenter. Ta med kolonner for vektoren, teksten i biten, alle metadatafeltene og et updated_at-tidsstempel for selektiv nyindeksering. Opprett en vektorindeks (HNSW eller IVFFlat) på embedding-kolonnen med én gang — det er langt tregere å legge til indeksen etter millioner av rader enn å opprette den på forhånd i en tom tabell.

-- PostgreSQL schema with pgvector
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE document_chunks (
    id          BIGSERIAL PRIMARY KEY,
    doc_id      TEXT NOT NULL,
    chunk_text  TEXT NOT NULL,
    embedding   VECTOR(1536) NOT NULL,
    source_file TEXT,
    page_number INT,
    section     TEXT,
    doc_type    TEXT,
    tenant_id   TEXT NOT NULL,  -- for data isolation
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_chunks_hnsw ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

CREATE INDEX idx_chunks_tenant ON document_chunks(tenant_id);

Bygge inntakspipelinen

Implementer inntak som én asynkron funksjon som tar imot en filbane eller URL og returnerer antallet indekserte biter. Bruk LangChains dokumentlastere for ulike filtyper og den rekursive tekstsplitteren for tegn til oppdeling. Grupper embedding-kall i batcher for å holde Dem innenfor grensen på 2048 token i inndata og redusere antallet API-kall fra tusenvis til noen titalls.

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from openai import AsyncOpenAI

async def ingest_document(file_path: str, doc_id: str, tenant_id: str) -> int:
    loader = PyPDFLoader(file_path)
    pages = loader.load()
    splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)
    chunks = splitter.split_documents(pages)
    # Batch embed
    texts = [c.page_content for c in chunks]
    client = AsyncOpenAI()
    embeddings_response = await client.embeddings.create(
        model='text-embedding-3-small',
        input=texts
    )
    embeddings = [e.embedding for e in embeddings_response.data]
    await insert_chunks_to_pgvector(chunks, embeddings, doc_id, tenant_id)
    return len(chunks)

Bygge en hybridgjenfinner

Kombiner tett vektorsøk med BM25-nøkkelordsøk, og slå sammen resultatene ved hjelp av reciprocal rank fusion. Implementer gjenfinneren som en klasse med én retrieve(query, tenant_id, top_k)-metode. Kjør begge søkene samtidig med asyncio.gather, slå sammen de rangerte listene med RRF, fjern duplikater basert på bit-ID, og returner de øverste top_k-resultatene med kildemetadata.

import asyncio
from rank_bm25 import BM25Okapi

class HybridRetriever:
    def __init__(self, pool, k_rrf: int = 60):
        self.pool = pool
        self.k_rrf = k_rrf

    async def retrieve(self, query: str, tenant_id: str, top_k: int = 10) -> list:
        dense_results, sparse_results = await asyncio.gather(
            self._dense_search(query, tenant_id, top_k * 3),
            self._bm25_search(query, tenant_id, top_k * 3)
        )
        merged = self._rrf_merge(dense_results, sparse_results)
        return merged[:top_k]

    def _rrf_score(self, rank: int) -> float:
        return 1.0 / (self.k_rrf + rank + 1)

Implementere den sentrale QA-kjeden

Bygg den sentrale spørsmål-og-svar-kjeden med LangChain LCEL. Kjeden tar imot et spørsmål og gjenfunne biter, formaterer en utvidet prompt med instruksjoner om å bruke bare den oppgitte konteksten, og strømmer svaret. Legg til uttrykkelige instruksjoner om at modellen skal oppgi kilder med dokumentnavn og sidetall, og si «I don't know» når svaret ikke finnes i den gjenfunne konteksten.

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser

RAG_PROMPT = ChatPromptTemplate.from_messages([
    ('system', 'You are a precise assistant. Answer ONLY using the provided context. '
               'Cite sources as [DocName, p.N]. If the answer is not in the context, say "I don\'t have information about that."'),
    ('user', 'Context:\n{context}\n\nQuestion: {question}')
])

llm = ChatOpenAI(model='gpt-4o', temperature=0, streaming=True)
qa_chain = RAG_PROMPT | llm | StrOutputParser()

async def answer_question(question: str, chunks: list) -> str:
    context = '\n\n'.join(f'[{c["source"]}]\n{c["text"]}' for c in chunks)
    return await qa_chain.ainvoke({'context': context, 'question': question})

Legge til funksjonskall i agenten

Utvid det grunnleggende spørsmål-og-svar-systemet med funksjonskall for å håndtere spørringer som krever sanntidsdata eller beregninger. Definer verktøy for å søke på nettet etter aktuell informasjon, kjøre SQL-spørringer mot en sikker skrivebeskyttet database og slå opp bestemte poster etter ID. Agenten avgjør hvilke verktøy som skal kalles basert på spørsmålet, utfører kallene og innarbeider resultatene i det endelige svaret.

from openai import AsyncOpenAI
import json

TOOLS = [
    {
        'type': 'function',
        'function': {
            'name': 'search_knowledge_base',
            'description': 'Search the internal document knowledge base for relevant information',
            'parameters': {
                'type': 'object',
                'properties': {
                    'query': {'type': 'string', 'description': 'Search query'},
                    'top_k': {'type': 'integer', 'default': 5}
                },
                'required': ['query']
            }
        }
    }
]

async def agent_with_tools(question: str, tenant_id: str) -> str:
    client = AsyncOpenAI()
    messages = [{'role': 'user', 'content': question}]
    while True:
        resp = await client.chat.completions.create(
            model='gpt-4o', messages=messages, tools=TOOLS)
        if resp.choices[0].finish_reason != 'tool_calls':
            return resp.choices[0].message.content
        tool_call = resp.choices[0].message.tool_calls[0]
        args = json.loads(tool_call.function.arguments)
        result = await dispatch_tool(tool_call.function.name, args, tenant_id)
        messages.append({'role': 'tool', 'tool_call_id': tool_call.id, 'content': result})

Strømme agentens svar

Pakk agenten inn i en FastAPI StreamingResponse ved hjelp av server-sendte hendelser, slik at frontend viser tokenene etter hvert som de kommer inn. For agentsvar med verktøykall strømmer De en fremdriftsindikator mens verktøyet kjører («Searching knowledge base...»), og strømmer deretter det endelige svaret token for token. Dette hindrer siden i å se ut som om den har fryst mens verktøyet bruker tid på å kjøre.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def event_stream(question: str, tenant_id: str):
    yield 'data: {"type": "start"}\n\n'
    chunks = await retriever.retrieve(question, tenant_id)
    yield 'data: {"type": "retrieving", "count": ' + str(len(chunks)) + '}\n\n'
    async for token in qa_chain.astream({'context': format_context(chunks), 'question': question}):
        yield f'data: {{"type": "token", "content": {repr(token)}}}\n\n'
    yield 'data: {"type": "done"}\n\n'

@app.post('/query/stream')
async def stream_query(question: str, tenant_id: str):
    return StreamingResponse(event_stream(question, tenant_id), media_type='text/event-stream')

Koble inn den semantiske cachen

Legg til den semantiske cachen som et trinn før gjenfinning i spørringspipen. Før De kontakter retrieveren og LLM-en, bygger De et embedding av brukerspørsmålet og sjekker cachen. Hvis det finnes et cache-treff med likhet over terskelverdien, returnerer De det bufrede svaret umiddelbart med flagget cached: true. Ved cache-miss fortsetter behandlingen gjennom hele pipelinen, og det resulterende svaret lagres i cachen for fremtidige, lignende spørringer.

async def query_pipeline(question: str, tenant_id: str) -> dict:
    # 1. Check semantic cache
    cache_hit = await semantic_cache.lookup(question, tenant_id, threshold=0.92)
    if cache_hit:
        return {'answer': cache_hit.answer, 'cached': True, 'sources': cache_hit.sources}

    # 2. Retrieve
    chunks = await retriever.retrieve(question, tenant_id, top_k=5)
    chunks = await reranker.rerank(question, chunks, top_n=3)

    # 3. Generate
    answer = await answer_question(question, chunks)
    sources = [c['source'] for c in chunks]

    # 4. Cache result
    await semantic_cache.store(question, tenant_id, answer, sources)

    return {'answer': answer, 'cached': False, 'sources': sources}

Legge til LangSmith-sporing

Instrumenter pipelinen med LangSmith ved å angi to miljøvariabler. Alle kall til LangChain-kjeder spores automatisk med antall token, latenstid, inndata, utdata og eventuelle feil. For egendefinert kode som ikke bruker LangChain (gjenfinning og omrangering) pakker De kallene inn i @traceable-dekoratorer, slik at de inkluderes i sporingen. Dette gir full oversikt fra ende til ende over hvert trinn i pipelinen.

import os
from langsmith import traceable

os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_API_KEY'] = os.environ['LANGSMITH_API_KEY']
os.environ['LANGCHAIN_PROJECT'] = 'document-qa-production'

# Wrap non-LangChain steps with @traceable
@traceable(name='hybrid_retrieval')
async def traced_retrieval(question: str, tenant_id: str, top_k: int) -> list:
    return await retriever.retrieve(question, tenant_id, top_k)

@traceable(name='cohere_reranking')
async def traced_reranking(question: str, chunks: list) -> list:
    return await reranker.rerank(question, chunks)

# LangChain LCEL chains are automatically traced — no extra code needed

Kjøre integrasjonstester

Skriv integrasjonstester som kjører gjennom hele spørringspipelinen, fra inndata for spørsmålet til det endelige svaret. Bruk et lite testvektorlager med kjente dokumenter, slik at De kan skrive deterministiske påstander om hvilke tekstbiter som skal hentes, og hva svaret skal inneholde. Kjør integrasjonstestene mot et staging-miljø som gjenspeiler produksjonsinfrastrukturen, men bruker et separat vektorlaget og en separat LLM-nøkkel.

import pytest

@pytest.mark.asyncio
async def test_full_pipeline_returns_grounded_answer():
    # Setup: ingest known document
    await ingest_document('tests/fixtures/policy.pdf', 'policy_v1', 'test_tenant')

    # Query with a question that has a known answer in the document
    result = await query_pipeline(
        question='What is the cancellation policy?',
        tenant_id='test_tenant'
    )

    assert result['answer'] is not None
    assert len(result['answer']) > 50
    assert '24 hours' in result['answer']  # known fact in document
    assert 'policy.pdf' in str(result['sources'])
    assert result['cached'] is False  # fresh query

Måle grunnleggende kvalitet på gjenfinningen

Før De optimaliserer noe, bør De etablere en grunnlinje for gjenfinning. Bruk evalueringsdatasettet til å måle treffrate (vises det riktige dokumentet blant de fem beste resultatene?) og MRR (hvor høyt rangeres det?). Kjør denne grunnlinjemålingen etter at det første vektorlaget er satt opp, men før De legger til omrangering eller hybridsøk. Grunnlinjen viser hvilke konkrete forbedringer hver optimalisering gir, slik at De har dokumentasjon på hvilke teknikker det er verdt å beholde.

async def measure_retrieval_baseline(test_cases: list) -> dict:
    hits = 0
    reciprocal_ranks = []
    for case in test_cases:
        results = await retriever.retrieve(case['question'], case['tenant_id'], top_k=5)
        result_docs = [r['doc_id'] for r in results]
        if case['relevant_doc'] in result_docs:
            hits += 1
            rank = result_docs.index(case['relevant_doc']) + 1
            reciprocal_ranks.append(1.0 / rank)
        else:
            reciprocal_ranks.append(0.0)
    return {
        'hit_rate_at_5': hits / len(test_cases),
        'mrr': sum(reciprocal_ranks) / len(reciprocal_ranks)
    }

Rask sjekk

Test forståelsen Deres av hvordan man bygger grunnleggende RAG- og agentfunksjoner.

Besvar den i den interaktive leksjonen

Oppsummering av leksjonen

I denne leksjonen lærte De at implementering nedenfra og opp sikrer at hver komponent testes isolert før integrasjon, at hybrid gjenfinning med samtidige tette søk og BM25-søk kombinert gjennom RRF gir best kvalitet på gjenfinningen, og at LangSmith-sporing med @traceable-dekoratorer gir full oversikt over hele pipelinen. Neste steg er å herde systemet med mønstre for sikkerhet, caching og pålitelighet.

Gratis å komme i gang

Lær deg Python med en AI-veileder – gratis

Skriv og kjør ekte kode i nettleseren, få umiddelbar hjelp fra en AI-veileder som er tilgjengelig døgnet rundt, og fortsett der du slapp – på nettet eller i appen.

Kurs: 30
Leksjoner: 120

Start læringen gratis Se hele kurset

Ofte stilte spørsmål

Er leksjonen «Implementere sentrale RAG- og agentfunksjoner» gratis?

Ja – hele teksten i «Implementere sentrale RAG- og agentfunksjoner» er gratis å lese her på nettet. For å øve interaktivt med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt, og for å låse opp resten av AI Engineering Academy-kurset, kan du oppgradere til CoddyKit PRO. Kurset i AI Engineering Academy inneholder totalt 4 leksjoner.

Hva lærer jeg i «Implementere sentrale RAG- og agentfunksjoner»?

Bygg pipelinen for dokumentinntak, indeksering i vektorlagring, gjenfinning med omrangering og integrasjoner med agentverktøy etter mønstrene du har lært gjennom kurset. Du øver på AI Engineering Academy med praktisk kode som du kjører direkte i nettleseren, mens en AI-veileder som er tilgjengelig døgnet rundt, svarer på spørsmålene dine mens du jobber deg gjennom leksjonen.

Trenger jeg erfaring for å begynne med AI Engineering Academy?

Ingen tidligere erfaring er nødvendig. AI Engineering Academy på CoddyKit er lagt opp for både nybegynnere og viderekomne, så De kan begynne her eller helt fra start og lære i Deres eget tempo. Dette er leksjon 2 av 4.

Hvor lang tid tar leksjonen «Implementere sentrale RAG- og agentfunksjoner»?

De fleste CoddyKit-leksjoner tar omtrent 5–10 minutter. Hver leksjon er kort og interaktiv, slik at De gjør jevne fremskritt og kan fortsette akkurat der De slapp – både på nettet og i appen.

Kan jeg skrive og kjøre kode i denne AI Engineering Academy-leksjonen?

Ja. Alle AI Engineering Academy-leksjoner har en innebygd kodeeditor, slik at De kan skrive og kjøre ekte kode direkte i nettleseren og få umiddelbar tilbakemelding fra AI – uten lokal konfigurering.

Alle leksjonene i dette kurset

Utforme produksjonsarkitekturen
Implementere sentrale RAG- og agentfunksjoner
Sikring: sikkerhet, hurtigbufring og pålitelighet
Evaluering, utrulling og etteranalyse

← Tilbake til AI Engineering Academy