RAGとエージェントの中核機能の実装

実装の順序と依存関係

# 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
]

ベクトルストアのセットアップ

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

取り込みパイプラインの構築

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)

ハイブリッドretrieverの構築

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)

コアQAチェーンの実装

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

エージェントへのFunction Callingの追加

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

エージェント応答のストリーミング

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

Semantic Cache の組み込み

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}

LangSmith トレーシングの追加

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

統合テストの実行

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

検索品質のベースライン測定

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

クイックチェック

レッスンのまとめ