## LLMから構造化データを抽出する技術

大規模言語モデル（LLM）はユーザーの入力を受け取り、質問に対する文章を生成できます。しかし生成された文章は非構造化データであり、他のシステムと連携するには構造化データとして出力を受け取る必要があります。

本記事ではLLMの出力を構造化データとして取り扱う方法を整理します。この手法を理解することで、単純な文章生成の用途だけではなく、他のシステムと連携して柔軟にワークフローを自動化させることができます。

また実際のユースケースとして「顧客からの見積依頼・発注依頼などの問い合わせにおいて、住所の情報を自動で取得し、顧客データベースや各種書類に自動で転記する」というシナリオを想定し、自然言語のシステムへの連携をスムーズに実現するヒントを示します。

LangChainにおける構造化出力

LangChain では`.with_structured_output()` を用いて、生成AIの出力から構造化データを抽出できます。LangChainでサポートしている出力方法は下記の2通りです。

- Pydantic のBaseModelを継承してスキーマを定義したクラスを渡すことで、出力を与えられたクラスのインスタンスとして返すもの

- TypedDictもしくはJSONスキーマを定義して渡すことで、合致したdictを返すもの

LangChainにおける構造化データの抽出方法は、OpenAIのStructured Outputsなど、LLM側で構造化出力をサポートしている機能を利用するものと、LLMの返答をOutputParserを用いて指定されたスキーマにパースする方法があります。

LLM側でサポートされている機能の中身については把握が難しいため、LangChain側でスキーマをパースするOutputParserの実装を確認していきます。当該処理は langchain_core/output_parsers/json.py に定義されています。

実際にパースする部分を見てみると、インポートされている parse_json_markdown を呼び出しているだけのようです。

def parse_result(self, result: list[Generation], *, partial: bool = False) -> Any:
    """Parse the result of an LLM call to a JSON object.
    Args:
        result: The result of the LLM call.
        partial: Whether to parse partial JSON objects.
            If True, the output will be a JSON object containing
            all the keys that have been returned so far.
            If False, the output will be the full JSON object.
            Default is False.
    Returns:
        The parsed JSON object.
    Raises:
        OutputParserException: If the output is not valid JSON.
    """
    text = result[0].text
    text = text.strip()
    if partial:
        try:
            return parse_json_markdown(text)
        except JSONDecodeError:
            return None
    else:
        try:
            return parse_json_markdown(text)
        except JSONDecodeError as e:
            msg = f"Invalid json output: {text}"
            raise OutputParserException(msg, llm_output=text) from e

parse_json_markdown は langchain_core/utils/json.py に定義されています。

def parse_partial_json(s: str, *, strict: bool = False) -> Any:
    """Parse a JSON string that may be missing closing braces.
    Args:
        s: The JSON string to parse.
        strict: Whether to use strict parsing. Defaults to False.
    Returns:
        The parsed JSON object as a Python dictionary.
    """
    # Attempt to parse the string as-is.
    try:
        return json.loads(s, strict=strict)
    except json.JSONDecodeError:
        pass

    # Initialize variables.
    new_chars = []
    stack = []
    is_inside_string = False
    escaped = False

    # Process each character in the string one at a time.
    for char in s:
        if is_inside_string:
            if char == '"' and not escaped:
                is_inside_string = False
            elif char == "\n" and not escaped:
                char = "\\n"  # Replace the newline character with the escape sequence.
            elif char == "\\":
                escaped = not escaped
            else:
                escaped = False
        else:
            if char == '"':
                is_inside_string = True
                escaped = False
            elif char == "{":
                stack.append("}")
            elif char == "[":
                stack.append("]")
            elif char == "}" or char == "]":
                if stack and stack[-1] == char:
                    stack.pop()
                else:
                    # Mismatched closing character; the input is malformed.
                    return None
        # Append the processed character to the new string.
        new_chars.append(char)

    # If we're still inside a string at the end of processing,
    # we need to close the string.
    if is_inside_string:
        new_chars.append('"')
    # Reverse the stack to get the closing characters.
    stack.reverse()
    # Try to parse mods of string until we succeed or run out of characters.
    while new_chars:
        # Close any remaining open structures in the reverse
        # order that they were opened.
        # Attempt to parse the modified string as JSON.
        try:
            return json.loads("".join(new_chars + stack), strict=strict)
        except json.JSONDecodeError:
            # If we still can't parse the string as JSON,
            # try removing the last character
            new_chars.pop()
    # If we got here, we ran out of characters to remove
    # and still couldn't parse the string as JSON, so return the parse error
    # for the original string.
    return json.loads(s, strict=strict)


def parse_json_markdown(
    json_string: str, *, parser: Callable[[str], Any] = parse_partial_json
) -> dict:
    """Parse a JSON string from a Markdown string.
    Args:
        json_string: The Markdown string.
    Returns:
        The parsed JSON object as a Python dictionary.
    """
    try:
        return parsejson(json_string, parser=parser)
    except json.JSONDecodeError:
        # Try to find JSON string within triple backticks
        match = jsonmarkdown_re.search(json_string)
        # If no match found, assume the entire string is a JSON string
        # Else, use the content within the backticks
        json_str = json_string if match is None else match.group(2)
    return parsejson(json_str, parser=parser)


def parsejson(
    json_str: str, *, parser: Callable[[str], Any] = parse_partial_json
) -> dict:
    # Strip whitespace,newlines,backtick from the start and end
    json_str = json_str.strip(_json_strip_chars)
    # handle newlines and other special characters inside the returned value
    json_str = customparser(json_str)
    # Parse the JSON string into a Python dictionary
    return parser(json_str)

上記のコードを見るとそのまま json.loads で変換できるものはそのまま、そうでないものは文字列を見ていきながら、気合いでパースして json.loads を実行します。

次にこのOutput Parserがどのようなプロンプトで生成されたテキストを受け取っているのかを見ていきます。この処理はOutput Parserの get_format_instructions で行われます。

JSONをパースするためのプロンプトの例です。

https://github.com/langchain-ai/langchain/blob/master/libs/core/langchain_core/output_parsers/format_instructions.py

JSON_FORMAT_INSTRUCTIONS = """The output should be formatted as a JSON instance that conforms to the JSON schema below.
As an example, for the schema {{"properties": {{"foo": {{"title": "Foo", "description": "a list of strings", "type": "array", "items": {{"type": "string"}}}}}}, "required": ["foo"]}}
the object {{"foo": ["bar", "baz"]}} is a well-formatted instance of the schema. The object {{"properties": {{"foo": ["bar", "baz"]}}}} is not well-formatted.
Here is the output schema:
{schema}
"""

Pydanticをパースする場合は下記のプロンプトをベースにしています。同じプロンプトを使用していることが確認できます。

PYDANTICFORMAT_INSTRUCTIONS = """The output should be formatted as a JSON instance that conforms to the JSON schema below.
As an example, for the schema {{"properties": {{"foo": {{"title": "Foo", "description": "a list of strings", "type": "array", "items": {{"type": "string"}}}}}}, "required": ["foo"]}}
the object {{"foo": ["bar", "baz"]}} is a well-formatted instance of the schema. The object {{"properties": {{"foo": ["bar", "baz"]}}}} is not well-formatted.
Here is the output schema:
{schema}
"""

OpenAIによる構造化出力

OpenAIが提供するAPIでは、Structured Outputsとしてこの手法がまとめられています。

https://platform.openai.com/docs/guides/structured-outputs

OpenAIでは、function callingとresponse_formatの引数を与えるという、2つの構造化データを抽出する手法があります。また少し前に提供が始まったjson_modeという機能もありますが、出力をJSONであることは保証しても、そのスキーマが何であるかは保証しないため、Structured Outputsがある現在ではあえてこの方法を採用する必要はないでしょう。

構造化出力を得る2つの方法

Open AIのドキュメント上では、function callingと response_formatの使い分けについて、下記のように記載されています。

Function calling is useful when you are building an application that bridges the models and functionality of your application.

https://platform.openai.com/docs/guides/structured-outputs より。

Conversely, Structured Outputs via response_format are more suitable when you want to indicate a structured schema for use when the model responds to the user, rather than when the model calls a tool.

https://platform.openai.com/docs/guides/structured-outputs より。

function callingはシステム上で異なる処理を実現するための繋ぎとして、response_formatはシステム上の処理のためではなく、出力をユーザー向けに提供するため、それぞれ使用するよう推奨されています。

Function Calling

Function CallingのAPIから弊社住所の「兵庫県神戸市中央区浪花町64 三宮電電ビル5階」に対する都道府県と市区町村のデータをJSON形式で取得してみます。

import os

import openai

os.environ["OPENAI_API_KEY"] = "sk-proj-xxx"

if name == "__main__":
    client = openai.OpenAI()
    user_prompt: str = "「兵庫県神戸市中央区浪花町64 三宮電電ビル5階」の都道府県と市区町村における天気を教えてください。"
    messages = [
        {
            "role": "system",
            "content": "You are a helpful assistant. The current date is August 6, 2024. You help users query for the data they are looking for by calling the query function.",
        },
        {
            "role": "user",
            "content": user_prompt,
        },
    ]
    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "prefecture": {"type": "string"},
                        "city": {"type": "string"},
                    },
                    "required": ["prefecture", "city"],
                },
            },
        }
    ]

    completion = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools,
    )

    print(completion.choices[0].message.tool_calls)

この実行結果を確認してみます。

$ rye run python src/function_calling.py

[ChatCompletionMessageToolCall(id='call_CrrhyoPtVEIXrVTzAePlJImj', function=Function(arguments='{"prefecture": "兵庫県", "city": "神戸市"}', name='get_weather'), type='function'), ChatCompletionMessageToolCall(id='call_7wHDOEtcdak4W9IVXVDSFSLa', function=Function(arguments='{"prefecture": "兵庫県", "city": "中央区"}', name='get_weather'), type='function')]

結果としては2通りの出力が得られました。

{"prefecture": "兵庫県", "city": "神戸市"}

{"prefecture": "兵庫県", "city": "中央区"}

response_format

次に response_format にPydanticのクラスを渡す方法を試します。

import os

import openai
import pydantic

os.environ["OPENAI_API_KEY"] = "sk-proj-xxx"

class Address(pydantic.BaseModel):
    prefecture: str
    city: str

if name == "__main__":
    client = openai.OpenAI()
    user_prompt: str = "「兵庫県神戸市中央区浪花町64 三宮電電ビル5階」の都道府県と市区町村における天気を教えてください。"
    messages = [
        {
            "role": "system",
            "content": "You are a helpful assistant. The current date is August 6, 2024. You help users query for the data they are looking for by calling the query function.",
        },
        {
            "role": "user",
            "content": user_prompt,
        },
    ]

    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=messages,
        response_format=Address
    )

    print(completion.choices[0].message.parsed)

この実行結果を確認してみます。

$ rye run python src/response_format.py

prefecture='兵庫県' city='神戸市中央区'

completion.choices[0].message.parsed には、 response_format として与えた pydantic.BaseModel のインスタンスもしくは None が返ってきます。

Introducing Structured Outputs in the API によると、OpenAIの一部の最新のモデルでは、精度良くJSONを出力できるよう、いくつかの調整がされています。

他には Instructor や outlines といった構造化データを出力させることに特化したライブラリもあります。

JSONなどの構造化データの抽出 - 実務応用

自然言語のような非構造化データから構造化データを抽出するには、上記の手法を用いたとしても、意図した通りに動作しないケースは多々あります。本セクションでは、よくある課題とそれをシステム上でどう対処するかについて解説します。

実務上の課題

LLMから適切なJSONではない出力を受け取る

LLMはプロンプトで「JSON形式を返してください」と指示しても、必ずしも正しいJSON形式の返答ができるわけではありません。本記事のような都道府県と市区町村のみのような簡単なフォーマットであればまだ正確な動作が多いことを期待できます。しかし、スキーマが複雑になればなるほど、この問題は顕在化してきます。

出力が意図しないJSON形式として不正確なものであれば、それをチェックする処理を挟み、チェックした結果の情報を元に、適切なJSON形式を返すことのみを再度LLMに依頼することが対策として考えられます。

スキーマの不一致

PydanticやTypedDict、JSON Schemaを使ってスキーマを定義しても、LLM側が必ずしもそのスキーマ通りのフィールドや型を返す保証はありません。たとえば必須フィールドが抜けていたり、期待とは異なる型（文字列ではなく数値など）が返ってきたりします。

この場合は処理後にバリデーションを実施することで、この問題を検出し、ユーザーへ適切な次のアクションを依頼できます。もしくは、スキーマ不一致の情報をバリデーションで洗い出し、その情報をもとに再度LLMへ必要なJSONスキーマを出力するようリクエストするのも良いでしょう。

課題への主な対応方法

few-shot prompting

few-shot prompting は正確な回答を得るために、解答例をいくつかプロンプトとして渡す方法です。

この手法を採用することで、適切なJSON形式の出力を得ること、適切なスキーマを得ること、データのバリデーションに対する精度、を向上させることができます。

例えばLangChainでは日本語に翻訳すると下記のようなプロンプトを渡していました。

以下のJSONスキーマに準拠したJSONインスタンスとして出力してください。
例として、スキーマ
{{"properties": {{"foo": {{"title": "Foo", "description": "a list of strings", "type": "array", "items": {{"type": "string"}}}}}}, "required": ["foo"]}}
the object {{"foo": ["bar", "baz"]}}
は、スキーマの整形されたJSON形式です。

オブジェクト
{{"properties": {{"foo": ["bar", "baz"]}}}}
は正確にフォーマットされたJSON形式ではありません。

以下は出力スキーマです：
{スキーマ｝

これはあくまで汎用的なプロンプトのテンプレートなので、few-shotサンプルとしては、スキーマ自体を指定するよりは、JSON文字列であることを伝えるプロンプトになっています。このスキーマの下にスキーマのサンプルを追加し、下記のようなプロンプトを与えることで、意図した出力を得る確率を向上させることができます。

以下のJSONスキーマに準拠したJSONインスタンスとして出力してください。

例として、スキーマ
{{"properties": {{"foo": {{"title": "Foo", "description": "a list of strings", "type": "array", "items": {{"type": "string"}}}}}}, "required": ["foo"]}}
the object {{"foo": ["bar", "baz"]}}
は、スキーマの整形されたJSON形式です。

オブジェクト
{{"properties": {{"foo": ["bar", "baz"]}}}}
は正確にフォーマットされたJSON形式ではありません。

以下は出力スキーマです：
{スキーマ｝

以下は出力して欲しいJSON形式のサンプルです：
{"prefecture": "大阪府", "city": "高槻市"}
{"prefecture": "兵庫県", "city": "神戸市中央区"}
{"prefecture": "東京都", "city": "豊島区"}

先ほどは出力として、「兵庫県神戸市中央区」という出力に対して、下記の2パターンを出力として得ていました。

• 都道府県：兵庫県、市区町村：神戸市
• 都道府県：兵庫県、市区町村：中央区

しかし取得したいデータは"都道府県：兵庫県、市区町村：神戸市中央区"です。

その場合は上記のようなプロンプトを与え、「神戸市中央区」という市区町村があることを伝え、精度向上を目指します。

import os

import openai

os.environ["OPENAI_API_KEY"] = "sk-proj-xxx"

if name == "__main__":
    client = openai.OpenAI()
    user_prompt: str = "「兵庫県神戸市中央区浪花町64 三宮電電ビル5階」の都道府県と市区町村における天気を教えてください。"
    system_prompt: str = """
        あなたは、有用なアシスタントです。現在の日付は2024年8月6日です。query関数を呼び出すことで、ユーザーが探しているデータを問い合わせるのを手伝います。
        query関数には、下記のようなデータを渡すことができます。
        {"prefecture": "大阪府", "city": "高槻市"}
        {"prefecture": "兵庫県", "city": "神戸市中央区"}
        {"prefecture": "東京都", "city": "豊島区"}
    """

    messages = [
        {
            "role": "system",
            "content": system_prompt,
        },
        {
            "role": "user",
            "content": user_prompt,
        },
    ]

    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "prefecture": {"type": "string"},
                        "city": {"type": "string"},
                    },
                    "required": ["prefecture", "city"],
                },
            },
        }
    ]

    completion = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        tools=tools,
    )

    print(completion.choices[0].message.tool_calls)

上記のコードを実行することで、下記のような主力を得られました。1つ目の ChatCompletionMessageToolCall で、市区町村が神戸市中央区となっており、意図した結果を得られていることが分かります。

$ rye run python src/main.py

[ChatCompletionMessageToolCall(id='call_M4OKMlbAMcQRvLWEcM3u32nD', function=Function(arguments='{"prefecture": "兵庫県", "city": "神戸市中央区"}', name='get_weather'), type='function'), ChatCompletionMessageToolCall(id='call_mNrAhIWeqBEobSFRWGp3dojC', function=Function(arguments='{"prefecture": "兵庫県", "city": "神戸市"}', name='get_weather'), type='function')]

リトライ

任意の処理に対してミスが発生した場合、そのミスの情報とともにLLMにミスを訂正するリトライ処理を挟みます。

手順としてはまず、JSON形式が適切でないこと、意図したスキーマでないことの両方の場合で、適切にバリデーションを行い、エラーメッセージを取得します。そのエラーメッセージの内容とともに、改めてLLMに適切なJSON形式を返すこと、意図したスキーマで返すこと、の指示とともに、リクエストを送信します。

アプリケーション上で人間からのフィードバックを得る

例えば市区町村の情報が「神戸市中央区」を意図しているときに、「神戸市」としてデータを抽出してしまうと、システム上にはそのデータをそのままに登録できないという課題が生じます。もしくは仮に登録できたとして、顧客へ送る書類に誤りが含まれてしまうことは防がなければなりません。

アプリケーション上ではLLMの出力を構造化データとしてパースし、その結果をユーザーに提示して誤りがあれば修正を依頼し、誤りがなければそのまま登録する、といったインターフェースを提供します。これにより、業務効率化を実現しながらも、誤りが問題にならないように制御できます。

これらのエラーは、LLMが本質的に確率的生成モデルであることや、ユーザーが入力する自然言語の曖昧さに起因しています。しかし、プロンプトの工夫、エラー処理の強化、人間のフィードバックを利用するといった対策を組み合わせることで、実務応用においても安定した構造化データ抽出が可能になります。

まとめ

LLMからの構造化データの抽出は、自然言語をシステム上で扱いやすい形式へと変換する上で重要な技術です。

LangChainやOpenAIのFunction Calling、response_formatオプション、また独自のパース処理やそれ用のライブラリを活用することで、容易に構造化データを抽出できます。

本記事で紹介したLangChain, OpenAIのFunction Callingやresponse_format, Pydantic, TypedDictなどの手法・ツールを活用することで、LLM出力をスキーマ化したうえで自動処理するパイプラインを容易に構築できます。こうしたアプローチは、顧客データ管理の効率化やワークフロー自動化といった実務上の課題解決につながり、業務プロセスの高速化や品質向上、さらにはビジネス価値の最大化に寄与します。

当社では、最新のAI技術を活用したソリューション開発に注力しており、データ活用の課題解決や業務効率化を支援しています。生成AIを活用し、各種書類などのファイルからシステムにデータを連携するようなシステムにご興味がある方は、ぜひお気軽にお問い合わせください。

参考文献

• https://python.langchain.com/docs/how_to/structured_output/
• https://platform.openai.com/docs/guides/structured-outputs
• https://openai.com/index/introducing-structured-outputs-in-the-api/
• https://dottxt-ai.github.io/outlines/latest/
• https://python.useinstructor.com/