every Tech Blog

株式会社エブリーのTech Blogです。

RAGでOpenAI APIのStructured Outputsを使い倒す

こんにちは。
開発本部のデータ&AIチームでデータサイエンティストをしている古濵です。

昨今の生成AIはどんどん新しいものが生まれ、日々キャッチアップを欠かせない日々を過ごしております。 9月もo1モデルが発表されましたが、今回はそちらではなく、8月に発表されたStructured Outputsを活用した取り組みについてご紹介します。

Structured Outputsとは

openai.com

Structured Outputsとは、モデルによって生成された出力が、開発者が提供する JSON スキーマと完全に一致するように設計された新機能です。
今までもJson ModeやFunction Callingを利用することで、構造化した出力を得ることはできていたのですが、精度がもう一息といったところでした。
しかし、Structured OutputsはOpenAI調べで精度100%を実現し、非構造な入力から構造化された出力を生成できることで、AIを中心としたアプリケーションが構築できるようになりました。

下記のコードは、簡単な数学の問題をその計算過程も踏まえて解かせてみる例です。
response_formatで出力形式(pydantic等)を指定すれば、簡単にStructured Outputsを試せます。
このコードはJson Modeのような使い方ですが、Function Callingを利用する方法もあります。 betaとありますが、確かに体感100%と言ってよいほどの高精度でした。

from pydantic import BaseModel

from openai import OpenAI


class Step(BaseModel):
    explanation: str
    output: str


class MathResponse(BaseModel):
    steps: list[Step]
    final_answer: str


client = OpenAI(api_key=OPENAI_API_KEY)

completion = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "You are a helpful math tutor."},
        {"role": "user", "content": "solve 8x + 31 = 2"},
    ],
    response_format=MathResponse,
)

message = completion.choices[0].message
if message.parsed:
    print(message.parsed.steps)
    print(message.parsed.final_answer)
else:
    print(message.refusal)

# Step(explanation='Start by isolating the term with x. We need to subtract 31 from both sides of the equation.', output='8x + 31 - 31 = 2 - 31'), Step(explanation='Simplify both sides. The 31s cancel out on the left, and we do the subtraction on the right.', output='8x = -29'), Step(explanation='Now, divide both sides by 8 to solve for x.', output='x = -29/8')]
# x = -29/8

プロンプトエンジニアリングにおいて、step by stepで考えさせるというテクニックがあります。
このような文言はプロンプトにないですが、数学の式変形の途中式を列挙するようなstep by stepな出力が実現できていることが確認できます。
個人的に、o1モデルと組み合わせれば、より高度なタスクにも対応できないだろうかと妄想が膨らみます。

RAGへの応用

Retrieval-Augmented Generation for Large Language Models: A Surveyにおいて、RAG研究の発展を3つの段階に分類しています。

  1. Naive RAG
  2. Advanced RAG
  3. Modular RAG

一般的にRAGの検証を始める際も似たような道を辿ることになると思います。
まずは自社のデータでNaive RAGを試して精度を確認し、精度が不十分であれば前処理や後処理と言った生成以外での工夫(Advanced)をしていき、最終的にはモジュール化していくという流れです。

Structured OutputsはAdvanced RAGにおいて、指示したプロンプトと出力フォーマット通りに前処理や後処理ができるため、非常に有用だと感じています。 処理の作成の容易さと柔軟性に優れており、前処理や後処理を手始めに作ってみるという気軽さがあります。 (ただし、トークン量や処理時間には注意が必要です)。

DELISH KITCHEN x RAG

「デリッシュAI」の紹介

DELISH KITCHENでは 「作りたい!が見つかる」をサービスのコンセプトとして、様々な機能を提供してきました。
一方、ユーザーひとりひとりの多様なニーズに合わせたレシピを提案していくには既存機能でのサポートだけでは難しさもある中で、AIによる料理アシスタントとして「デリッシュAI」ベータ版を一部ユーザーから提供し始めています。

AI/LLMでtoC向けサービスはどう変わるのか?『DELISH KITCHEN』は、「レシピ動画アプリ」から「AI料理アシスタント」へ

ユースケース

「デリッシュAI」でユーザが自然言語でレシピを検索するための機能を提供するために、RAGを活用しています。 RAGでは、ユーザの自然言語入力をベクトル化し、ベクトル検索(Retrieval)することで関連したレシピを先に選定します。 今回はベクトル検索前後にStructured Outputsを使って処理を挟む事例を紹介していきます。

Structured Outputsを使った処理は以下の箇所で「デリッシュAI」の一部に組み込まれています。

  • 前処理:よりベクトル検索しやすくするためにユーザのクエリを処理する
  • ベクトルデータベースのフィルター機能:ユーザのクエリにフィルタリング要素を含むか判定し、ベクトルデータベースの仕様に沿ったフォーマットで検索できるように出力する
  • 後処理:ベクトル検索後に、ユーザのクエリにマッチしたレシピを選定する

共通処理

共通処理を先に定義しておきます。

from enum import Enum
from typing import Union, List

from pydantic import BaseModel

from openai import OpenAI
import json

client = OpenAI(api_key=OPENAI_API_KEY)

1. フィルタリング

ユーザのクエリによっては特定の条件でフィルタリングすることで、より適切なレシピを提案できる可能性が高いです。 一般的なベクトル検索のアルゴリズムでは、「ダイエット中なので◯kcal以下のレシピを探して」といったような特定数値の範囲内での抽出は苦手な印象です。

以下は、ユーザのクエリにフィルタリング要素を含むか判定し、含む場合は指定したフォーマットで出力する例です。 ここでのフォーマットはベクトルデータベースでフィルタリングする想定しており、そこで利用できるような構造の出力を行います。
豊富なメタデータがあれば、ベクトルデータベースのフィルタリングはかなり強力になると思います。

class Column(str, Enum):
    calorie = "calorie_kcal"
    cooking_time = "cooking_time_min"
    cooking_cost = "cooking_cost_yen"
    protein = "protein_g"
    lipid = "lipid_g"
    carbohydrate = "carbohydrate_g"
    saccharide = "saccharide_g"
    salt = "salt_g"

class Operator(str, Enum):
    gt = ">"
    lt = "<"
    le = "<="
    ge = ">="

class Filter(BaseModel):
    columm: Column
    operator: Operator
    value: Union[str, int]

class QueryFilter(BaseModel):
    filters: list[Filter]

def create_query_filter(user_query: str) -> QueryFilter:
    system_prompt =  """
        あなたは料理の知識が豊富なレシピ検索AIです。
        ユーザーがレシピ検索のために入力したuser_queryを解読し、そこからユーザーが特定の条件でフィルタリングして検索したいかどうか判定してください。

        ## 出力形式
        * json形式で出力してください
        * 「ユーザーがフィルタリングして検索したい」の場合は、colummにカラム名、operatorに不等号、valueにフィルタリング対象を入れてください
        * 「ユーザーがフィルタリングして検索しない」の場合は、[]を入れてください
    """

    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_query},
        ],
        response_format=QueryFilter,
    )

    return json.loads(completion.choices[0].message.content)

簡単なユーザのクエリであれば問題なく抽出できます。
また、フィルタリングの要素がない場合は空が返ってくることが確認できます。

create_query_filter("塩分3グラム未満レシピ")
# {'filters': [{'columm': 'salt_g', 'operator': '<', 'value': 3}]}

create_query_filter("減塩を意識したレシピ")
# {'filters': []}

複数条件でも同様に抽出できることが確認できます。

create_query_filter("1000円以内で作れる1000kcal以上のレシピ")
# {'filters': [{'columm': 'cooking_cost_yen', 'operator': '<=', 'value': 1000}, {'columm': 'calorie_kcal', 'operator': '>=', 'value': 1000}]}

少し意地悪なクエリとして単位を意図的に変えてみます。
秒→分の変換は問題なく対応できましたが、カロリー→kcalの変換は対応できませんでした。

create_query_filter("1200秒以内に作れるデザート")
# {'filters': [{'columm': 'cooking_time_min', 'operator': '<=', 'value': 20}]}

create_query_filter("1000カロリー以上の和食")
# {'filters': [{'columm': 'calorie_kcal', 'operator': '>=', 'value': 1000}]}

2. 除外

ユーザのクエリに除外要素を含むか判定し、含む場合は指定したフォーマットで出力する例です。
ベクトル検索は曖昧な検索ができる点が強力ですが、ユーザのクエリに除外要素が含まれている場合でもそのレシピを抽出してしまうことがあります。 そのため、除外要素を抽出し、ベクトル検索の前にユーザのクエリから除外することで、より適切なレシピを提案できる可能性が高まります。
後処理で抽出した除外要素でフィルタリングすることも可能です。

class ExcludePreprocessedUserQuery(BaseModel):
    user_query: str
    user_query_preprocessed: str
    excluded_foods: List[str]


def exclude_preprocess_user_query(user_query: str) -> ExcludePreprocessedUserQuery:
    system_prompt =  """
        あなたはユーザーの調理ニーズを理解できるレシピ検索AIです。
        ユーザーがレシピ検索のために入力したuser_queryを解読し、そこから「ユーザーが検索で除外したい」かどうか判定してください。

        ## 出力形式
        * json形式で出力してください
        * 「ユーザーが検索で除外したい」場合は、excluded_foodsに除外したキーワード、user_query_preprocessedにuser_queryからexcluded_foodsを除外したキーワードを入れてください
        * 「ユーザーが検索で除外しない」の場合は、[]を入れてください
    """

    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_query},
        ],
        response_format=ExcludePreprocessedUserQuery,
    )

    return json.loads(completion.choices[0].message.content)

除外要素が含まれている場合は、除外要素を抽出し、ユーザのクエリから除外したクエリを生成できています。

exclude_preprocess_user_query("卵を使わない炒飯のレシピ教えて")
# {'user_query': '卵を使わない炒飯のレシピ教えて', 'user_query_preprocessed': '炒飯のレシピ教えて', 'excluded_foods': ['卵']}

除外はできるが文中に除外する言葉があるとuser_query_preprocessedがうまく生成できないこともあります。

exclude_preprocess_user_query("焼肉のタレが余ってるから焼肉以外で使いたい")
# {'user_query': '焼肉のタレが余ってるから焼肉以外で使いたい', 'user_query_preprocessed': '焼肉のタレが余ってるから焼肉以外で使いたい', 'excluded_foods': ['焼肉']}

3. 注視

同様に、除外の逆でユーザのクエリに注視要素を含むか判定し、含む場合は指定したフォーマットで出力する例です。
ベクトル検索後の後処理で、ユーザのクエリに注視要素が含まれている場合は、そのレシピを優先的に提案することができます。

class FocusPreprocessedUserQuery(BaseModel):
    user_query: str
    focused_foods: List[str]

def focus_preprocess_user_query(user_query:str) -> FocusPreprocessedUserQuery:
    system_prompt = """
        あなたはユーザーの調理ニーズを理解できるレシピ検索AIです。
        ユーザーがレシピ検索のために入力したuser_queryを解読し、そこから「検索する上で注視したいレシピ、食材、調味料」を含むかどうか判定してください。

        ## 出力形式
        * json形式で出力してください
        * 「検索する上で注視したい食材または調味料を含む」場合は、focused_foodsに注視したキーワードを入れてください
        * 「検索する上で注視したい食材または調味料を含まない」場合は、[]を入れてください
    """

    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_query},
        ],
        response_format=FocusPreprocessedUserQuery,
    )

    return json.loads(completion.choices[0].message.content)

以下のようなクエリでうまく生成できたことを確認できました。

focus_preprocess_user_query("ベーコンを使う炒飯のレシピ教えて")
# {'user_query': 'ベーコンを使う炒飯のレシピ教えて', 'focused_foods': ['ベーコン', '炒飯']}

focus_preprocess_user_query("ポン酢を大量に消費したい")
# {'user_query': 'ポン酢を大量に消費したい', 'focused_foods': ['ポン酢']

focus_preprocess_user_query("卵または玉ねぎを使ったレシピを教えて")
# {'user_query': '卵または玉ねぎを使ったレシピを教えて', 'focused_foods': ['卵', '玉ねぎ']}

focus_preprocess_user_query("シチューまたはカレーのレシピを教えて")
# {'user_query': 'シチューまたはカレーのレシピを教えて', 'focused_foods': ['シチュー', 'カレー']}

4. 名寄せ

除外や注視で抽出したキーワードはユーザのクエリをもとにしたものであるため、表記揺れがあります。
そこで、表記揺れを解消するために、同じ意味のキーワードを列挙する例です。 名寄せ用の辞書を用意する方が妥当ではありますが、今回はStructured Outputsを使い倒すコンセプトで進めます。

class FoodNameCollection(BaseModel):
    foods: List[str]

def merge_food_name_collection(user_query:str) -> FoodNameCollection:
    system_prompt = """
        あなたは食材、調味料の専門家です。
        入力された食材、調味料keywordの表記揺れとして思いつくものを最大5件生成し、その単語をListで出力してください。
        無理に5件作る必要はありません。妥当性を重視してください。
    """
    
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_query},
        ],
        response_format=FoodNameCollection,
    )

    return json.loads(completion.choices[0].message.content)
merge_food_name_collection("卵")
# {'foods': ['たまご', 'エッグ', '卵(たまご)', '卵白', '卵黄']}

merge_food_name_collection("生姜")
# {'foods': ['ショウガ', 'ジンジャー', '生姜根', '新生姜', '生姜粉']}

merge_food_name_collection("焼き肉")
# {'foods': ['焼き肉', '焼肉', 'やきにく', '焼きにく', 'ヤキニク']}

5. スクリーニング

ユーザのクエリがレシピの検索クエリとして妥当かどうか判定し、妥当な場合のみ指定したフォーマットで出力する例です。
ベクトル検索の前にこの処理を入れることで、不適切なクエリを排除することができ、無駄なリソースやAPIコストを削減できます。 しかし、スクリーニングが過剰に効きすぎた場合、ユーザは思うように検索ができない可能性もあります。

class ScreeningUserQuery(BaseModel):
    comment: str

def screening_user_query(user_query: str) -> ScreeningUserQuery:
    system_prompt = """
    あなたは料理の知識が豊富なレシピ検索AIです。

    ## 答えるべきでない入力について
    ~~ 中略:答えるべきでない内容を箇条書きで列挙 ~~

    ## 答えるべき入力について
    ~~ 中略: 答えるべき内容を箇条書きで列挙 ~~

    ## 出力形式
    * json形式で出力してください
    * 「## 答えるべきでない入力」に該当する質問の場合は、commentに回答できない理由と、どういう検索すると良いかの提案をユーザに寄り添ったフレンドリーな形で回答をしてください
    * 「## 答えるべきでない入力」に該当しない「料理を調べる」文脈の質問が来た場合は、commentに何も入力しないでください
    """
    
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_query},
        ],
        response_format=ScreeningUserQuery,
    )

    return json.loads(completion.choices[0].message.content)
screening_user_query("毒キノコを使ったレシピ")
# {'comment': '申し訳ありませんが、毒キノコを使用するレシピに関してはお答えできません。安全で美味しい料理を楽しむために、正しい食材を使うことが大切です。食材の選び方や健康的なレシピに関する質問がありましたら、ぜひお聞きください。'}

screening_user_query("暗殺者のパスタを作りたい")
# {'comment': '「暗殺者のパスタ」という具体的な料理は存在しないため、どのような料理を指しているのか明確ではありません。ただし、料理の種類やレシピを知りたい場合は、具体的な食材や風味の特徴を教えていただければ、それに合ったパスタ料理のレシピを提案できます。'}

暗殺者のパスタというレシピは世の中に実在しますが、モデルは存在しないレシピとして扱ってしまいます。
DELISH KITCHEN内でも暗殺者のパスタのレシピは公開しており、ベクトル検索すれば抽出できる可能性が高いです。
この場合は先にスクリーニングするのではなく、ベクトル検索後にスクリーニングするのが良いかもしれません。

おわりに

RAGでStructured Outputsを使い倒す例を紹介しました。 全てgpt-4o-miniモデルの性能とプロンプトエンジニアリングの範囲内での処理でしたが、非常に高い精度で処理できていることが確認できました。

トークンの使用量や処理時間には注意が必要ですが、Advanced RAGを短期間で構築する際には非常に有用だと感じています。

この記事が、RAGの検証を始める際の参考になれば幸いです。