ctcAdvent Calendar 2024

RAGコンペ参加記 (raggle)

Posted at 2024-12-03

はじめに

腕試しを兼ねて、プライベートでRAGのコンペティション「raggle」に参加しました。
この記事では、私が取り組んだ内容を共有します。
なお、プライベートでの参加であり、所属する組織とは無関係であることを明記しておきます。
結果として、リーダーボード上では86人中18位という成績で入賞までには至りませんでしたが様々な知見が得られたので紹介します。

参加したイベント: raggle

raggleは、株式会社Galirageが運営するサービスです。

RAGはLLMシステムにおいて中核的な機能を担っており、その精度改善は多くのLLMエンジニアにとって重要なテーマです。Raggleでは、RAGの精度を極めるエンジニア向けにコンペティションが開催されています。

<参考> RAG（検索拡張生成）とは？

RAG（Retrieval Augmented Generation）は、生成AIのビジネス活用において注目される技術です。
大規模言語モデル（LLM）にLLMが学習していない情報を提供し、より正確で信頼性のある回答を生成することを目的としています。具体的には、企業内の業務文書や外部の最新情報を検索し、それを情報抽出の基盤として活用します。ctcでも社内にRAGの展開を進めている最中です。

参加したコンペ: 法務RAGシステムの性能改善ハッカソン

このコンペティションでは、「法務RAGシステム」を構築し、その性能を競います。提供されたPDFデータを基に、ユーザーからの質問に対して適切な文書を取得し、回答を生成するRAGシステムを開発します。

データ

架空の契約について書かれた日本語の契約書PDFファイル(10ファイル程度)
資料中に画像や表はなし

ルール

生成モデルとして使用できるのはOpenAI API GPT-4o-miniのみ
利用可能なライブラリは制限あり（追加希望は可能）
処理時間の上限は5分
OpenAI APIの呼び出し回数は5回まで

進め方

レギュレーションにより、OpenAI APIの呼び出し回数が5回までに制限されていたため、一般的にRAGで使用されるベクトル検索は行わない方針としました。ベクトル化（埋め込み）を取得するEmbeddingAPIも呼び出し回数も制限を受ける関係ですべての文書(チャンク)をベクトル化するのは困難と判断しました。

そのため、キーワード検索であるBM25を用いたBM25Retrieverを中心に文章検索システムを構築しました。単純なBM25のみでは質問への回答も不十分だったため、以下の手法で精度向上を図りました。

テキスト化とチャンク分割
クエリ拡張とリランキング(RAG Fusion)
複数の検索エンジンの利用
回答生成のプロンプト修正
回答データの加工

最終的な構成

各手法について

テキスト化とチャンク分割

使用可能なライブラリとしてpypdfとpdfplumberがありましたが、pdfplumberを選定しました。提供されたGoogle DriveのURLからPDFPlumberLoaderでダウンロードおよびテキスト化。デフォルトですでにページ単位で分割されていたので、そのままチャンクとして利用しました。後から振り返ると、この部分はもう少し工夫できたと感じています。

テキスト化とチャンク分割

from langchain_community.document_loaders import PDFPlumberLoader

def create_documents_from_pdf_files(pdf_file_urls):
    documents = []
    for url in pdf_file_urls:
        loader = PDFPlumberLoader(url)
        docs = loader.load()
        documents.extend(docs)
    return documents

クエリ拡張とリランキング(RAG Fusion)

幅広いキーワードで文章をヒットさせるため、ユーザークエリをLLMを使って複数に分割しました。元のクエリに加えて追加した4つのクエリの計5クエリを検索エンジンに投げるようにしました。
また、複数クエリで検索して取得したチャンクの中から、最終的に採用するチャンクを決定するため、RRF（Reciprocal Rank Fusion）を使ったリランク実装しました。

これらの実装はこちらの記事を参考にさせていただきました。
修正点は元記事では改行でクエリを区切っていた部分をStructured Outputを使ってJsonで扱えるようにした程度でプロンプトやRRFのロジックはほぼそのまま利用させていただいています。

クエリ拡張とリランク

from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.documents import Document
from langchain_openai import ChatOpenAI

# structured outputのデータ構造の定義
class Query(BaseModel):
    """質問のリスト"""
    query: list[str] = Field(description="質問のリスト")

# クエリ拡張
def query_generator(original_query: dict) -> list[str]:
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant that generates multiple search queries based on a single input query."),
        ("user", "Generate multiple search queries related to: {question}. When creating queries, please refine or add closely related contextual information in Japanese, without significantly altering the original query's meaning"),
        ("user", "OUTPUT (4 queries):")
    ])

    user_query = original_query.get("question")

    llm = ChatOpenAI(
        model_name=model,
        temperature=0,
    )
    structured_llm = llm.with_structured_output(Query)

    def query_to_list(query):
        query_list= []
        query_list.append(user_query)
        query_list.extend(query.query)
        return query_list

    query_generator_chain = prompt | structured_llm | query_to_list
    queries = query_generator_chain.invoke({"question": user_query})
    return queries


# RRF
def reciprocal_rank_fusion(results: list[list], k=60):
    fused_scores = {}
    for docs in results:
        for rank, doc in enumerate(docs):
            # doc_str = dumps(doc)
            doc_str = json.dumps(Document.to_json(doc), ensure_ascii=False)
            if doc_str not in fused_scores:
                fused_scores[doc_str] = 0
            fused_scores[doc_str] += 1 / (rank + k)

    reranked_results = [
        (doc, score)
        for doc, score in sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    ]

    return [x[0] for x in reranked_results[:4]]

複数の検索モデルの利用

LangchainのBM25Retrieverはデフォルトでは日本語文章に対応していなかったため、sudachipyを使用して単語単位で分かち書きし、n-gramを作成するモデルを構築しました。単語単位のn-gramは文法や構文を考慮した検索が可能です。また追加で、文字単位のn-gramも実装も構築しました。文字単位のn-gramは表記ゆれやキーワードの部分一致に強いという特性があります。それぞれのモデルをEnsembleRetrieverを利用して併用しました。モデルの重みについては試行錯誤し最終的に単語0.7, 文字0.3としました。

実装はこちらの記事のものをほぼそのまま使わせていただきました。

EnsembleRetrieverの作成

from sudachipy import tokenizer
from sudachipy import dictionary
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# 単語単位のn-gramを作成
def generate_word_ngrams(text, i, j, binary=False):
    tokenizer_obj = dictionary.Dictionary(dict="core").create()
    mode = tokenizer.Tokenizer.SplitMode.A
    tokens = tokenizer_obj.tokenize(text ,mode)
    words = [token.surface() for token in tokens]

    ngrams = []
    
    for n in range(i, j + 1):
        for k in range(len(words) - n + 1):
            ngram = tuple(words[k:k + n])
            ngrams.append(ngram)
    
    if binary:
        ngrams = list(set(ngrams))  # 重複を削除
    
    return ngrams

def preprocess_func(text: str) -> List[str]:
    return generate_word_ngrams(text,1, 1, True)

# 文字単位のn-gramを作成
def generate_character_ngrams(text, i, j, binary=False):
    ngrams = []
    
    for n in range(i, j + 1):
        for k in range(len(text) - n + 1):
            ngram = text[k:k + n]
            ngrams.append(ngram)
    
    if binary:
        ngrams = list(set(ngrams))  # 重複を削除
    
    return ngrams

def preprocess_char_ngram_func(text: str) -> List[str]:
    i, j = 1, 3
    if len(text) < i:
        return [text]
    return generate_character_ngrams(text, i, j, True)


# 単語と文字のBM25Retrieverを作成
word_retriever = BM25Retriever.from_documents(docs, preprocess_func=preprocess_word_func)
char_retriever = BM25Retriever.from_documents(docs, preprocess_func=preprocess_char_func)
word_retriever.k = 4
char_retriever.k = 4

# EnsembleRetrieverを作成
ensemble_retriever = EnsembleRetriever(
    retrievers=[word_retriever, char_retriever], weights=[0.7, 0.3]
)

回答生成のプロンプト修正および回答生成部分

OpenAIのメタプロンプトを使用して回答生成用のプロンプトを作成。
生成されたプロンプトについては特に細かくチューニングはしておらず日本語で回答するように追加した程度です。

回答生成プロンプト

template = """
** Please be sure to answer in Japanese **
# Steps

1. Receive the question from the user.
2. Retrieve relevant context based on the question.
3. Generate a simple, concise answer using only the retrieved context.
4. Ensure the response does not include information beyond the provided context.
5. Answers must be faithful to the sentences in the context.
(When answering the amount, please check the context to see if it is tax-exclusive or tax-included.)

# Output Format

The output should be a single sentence or short paragraph that directly answers the question using only the given context.

# Examples

**Example 1:**
- **Input Question:** "What is the capital of France?"
- **Context:** "France is a country in Europe. The capital city is Paris."
- **Output Answer:** "The capital of France is Paris."

**Example 2:**
- **Input Question:** "Who is the author of '1984'?"
- **Context:** "'1984' is a novel written by George Orwell."
- **Output Answer:** "The author of '1984' is George Orwell."

(Real examples should include relevant complexity and contextual information retrieved from a source.)

# Notes

- Avoid providing any information not directly supported by the context.
- Ensure clarity and accuracy in the output.
- Handle ambiguous questions by stating that the context does not provide an answer.

# Context
{context}

# Question
{question}

"""

回答生成部分についてはLangchianを使ってクエリ拡張 -> EnsembleRetriever -> RRFでリランクしたものを上記のプロンプトのcontextに与えて最終的な回答を生成するようにしています。

from operator import itemgetter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template(template)

llm = ChatOpenAI(
    model=model,
    temperature=0
)

def format_docs(docs):
    format_docs = []
    for n in docs:
        docs_json = json.loads(n)
        format_docs.append(docs_json['kwargs']['page_content'])
    return '\n\n'.join(format_docs)

rag_fusion_retriever = (
    {"question": itemgetter("question")}
    | RunnableLambda(query_generator)
    | ensemble_retriever.map()
    | reciprocal_rank_fusion
)

chain = (
    {"context": rag_fusion_retriever | format_docs, "question": itemgetter('question')}
    | prompt
    | llm
    | StrOutputParser()
)
result = chain.invoke({'question': question})

回答データの加工

回答に余計な補足が多く含まれていたため、スコアが下がる原因となっていました。
そこで、最初の生成した回答に修正を加える処理を追加しました。
追加したプロンプトについては下記記事のものを参考にほぼそのまま使わせてもらいました。(RAG-1グランプリも出たかった…)

回答データの加工部分の抜粋

from langchain_core.output_parsers import StrOutputParser

def create_reanswer(question, answer, llm):
    reanswer_template = """
    あなたはプロの編集者です。
    以下に質問文に対する回答文があります。
    質問文に対して回答文の中から最も簡潔に重要な内容のみを抽出してください。
    質問文が何かわからないような回答は避けるようにしてください。

    # 質問文
    {question} 

    # 回答文
    {answer}
    """

    custom_reanswer_prompt = ChatPromptTemplate.from_template(reanswer_template)

    reanswer_chain = custom_reanswer_prompt | llm | StrOutputParser()

    return reanswer_chain.invoke({"question": question, "answer": answer})

まとめ

ベクトル検索が使えない制約下でも工夫次第で文章の揺らぎに対応できたことが最大の収穫でした。
チャンク分割・プロンプト・リランクの部分には改善の余地があると感じています。
このようなコンペを開催してくださり、ありがとうございました。次回もぜひ参加したいです。

おわりに

こちらの記事はctc Advent Calendar 2024の記事となります
この後もctc(中部テレコミュニケーション株式会社)のメンバーが技術にまつわる知見を投稿していきますのでご期待ください

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up