6
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@YokeSin

watsonx.ai と LangGraphで遊んでみる④ ～出力の制御(構造化出力)～

6
Last updated at Posted at 2025-11-07

◾️はじめに

前回までの記事ではwatsonx.aiとLangGraphで最小限のチャットボットを作り、LangSmithでトレーシングをして、外部ツールと接続してWeb検索機能を足して最新情報を参照できるようにしました。

前回の記事での課題は検索は出来るので、ハルシネーションは減ったもののURLなどの情報源の記載があったりなかったりと、出力の形式が安定しないという難点がありました。
今回の記事では構造化出力を使って、決まった形式でLLMが回答できるようにすることで、安定した出力ができるようにしていきたいと思います。

◾️前提

  • ①や②や③のシリーズとなる記事です。
  • この記事では前回の内容が出てくるので、上記を簡単に見ていただけると幸いです。
  • 初学者向けの記事です。所謂やってみた系の記事です。
  • 私自身が非エンジニアなので、間違い、アドバイスあれば、コメントで指摘お願いします！
  • 当記事ではプラットフォームはwatsonx.aiを使います。無償評価版などは①の記事から登録可能です。

今回も実装に入る前に構造化出力について簡単に整理してみましょう！

本題の会話履歴の要約につてはこちらから

■LLMの出力を構造化するモチベーション

LLMは本来、人が理解しやすい自然言語の生成を得意とします（要約や翻訳など）。しかし、多くのシステムは「機械が扱いやすい形式」を必要とするため、出力をそのまま利用するのは難しい場合があります。そこでStructured Output（構造化出力）が必要になるというわけです。※これらの形式の違いについては以下がわかりやすいかと思います。

生成結果をJSONのような構造化された形式にできれば、データベースへの登録やAPI呼び出し、ワークフローへの組み込みが容易になります。また事前に定めたJSON Schemaなどに沿って生成ができれば、出力の揺れも抑えられ、検証やエラーハンドリングもしやすくなります。その結果、自然言語で指示しながらシステムは安定した形で受け取り、自動化を柔軟に進められるようになります。

これは前回の記事で紹介したFunction Callingとも直結します。関数の引数に値を正しく渡すには、最終的に構造を持った出力が欠かせません。つまり、構造化出力は「人に最適化された言語」を「システムに最適化されたデータ」へと変換する基盤であり、LLMを実用システムに組み込むための前提技術なわけです。

■LLMの出力を構造化する手法

もともとNLP(自然言語処理)の分野ではルールベースや正規表現を使って、文章を構文解析して、後処理で成形したりといった構造化するための手法は行われてきましたが、ここではLLMの出力をいかに構造化するか、に絞ってみていきたいと思います。

  • ①プロンプトや後処理で整える
    ChatGPT登場直後は、プロンプトの工夫で生成結果をJSONなどの構造化データにして、正規表現やPydanticでで検証・整形する方法が一般的でした。
    ただし「有効なJSONを出す」にとどまり、出力の不安定さが課題でした。
    （※以下で解説されているLangChainのOutputParserの様な感じ）

  • ②事前に定義したJSON Schemaに沿って出力させる
    次の段階では、事前にJSONスキーマを定義し、モデルに従わせる方式が登場。これによりフォーマット崩れを防ぎ、安定した出力が可能になります。
    これはまずLangChainなどを筆頭に採用しはじめ、OpenAIのAPIでも2023年6月に「Function Calling」を発表。事前定義した関数の引数スキーマに従った出力を保証し、より機械処理しやすくなりました。

  • ③JSON出力の厳格化
    加えてLLMのJSON出力自体の正確さも上がってきました。例えば、2023年11月にOpenAIが発表したJSONモードでは、APIのresponse_format指定により「必ず構文的に正しいJSON」を生成させられるようになり、パースエラーが大幅に減少しました。
    さらに2024年8月にはOpen AIからStructured Outputsが発表。ユーザー定義のJSON Schemaに完全準拠した出力を保証。これは「制約付きデコード」により、型崩れや値の乱れを防ぎます。

このように、初期の「出力後のパース頼み」から、関数呼び出し → JSON保証 → スキーマ完全準拠　へと進化していくことで、構造化の安定度が変化していき、　LLM出力を実運用システムで安心して扱える水準に徐々になっていった、というわけです。

IBM watsonx.ai での構造化出力

IBM watsonx.aiでは、上記②のように Function Calling（関数呼び出し） に対応した仕組みを提供しています。この仕組みの中核となるのが、Toolkit（ツールキット）です。
Toolkitを使うことで、外部ツール（＝関数）を 定義・取得・実行 でき、LLMがそのツール仕様に沿った 構造化出力（スキーマに従ったJSON形式など） を生成できるようになります。

外部ツールの呼び出しがサポートされているLLMは、こちらのメソッドを使うことで一覧を取得できます。以下のサンプルコードのように確認できます。

from ibm_watsonx_ai import APIClient, Credentials

# 認証情報（APIキー,プロジェクトID,リージョンは自分の設定に合わせて変更してね）
client = APIClient(
    credentials=Credentials(
        api_key="YOUR_IBM_CLOUD_APIKEY",
        url="https://us-south.ml.cloud.ibm.com"),
    project_id="YOUR_PROJECT_ID")

# Function Calling対応チャットモデル一覧を取得
models = client.foundation_models.get_chat_function_calling_model_specs(get_all=True)

for m in models:
    print(m["model_id"], "-", m.get("name", ""))

■構造化出力はどうやって実装する？（イメージ）

具体的な実装に入る前に、ざっくりイメージを説明したいと思います。やり方は様々ありますが、本記事ではIBM watsonx.aiとLangGraphでの実装を考えているので、LangChainを使った以下の記事を参考に解説します。

と言ってもやることは簡単で、ユーザーとLLMの自然言語の文章でのやり取りに以下二つを加えるだけです。

① LLMに出力してほしい形式（スキーマ）をJSONなどで定義する
　→ どんなデータを出したいのか（項目名・型・制約など）をあらかじめ指定します。

② そのスキーマに沿った出力を返すように、モデルに指示する
　→ Function Calling や Structured Outputs 機能を使って、スキーマに沿ったJSONを必ず出力させます。

具体例を挙げて説明すると…

営業担当者が顧客へのメールから取引状況を整理・抽出し、顧客管理システム（以下CRMシステム）へ登録するタスクを考えてみましょう。

image.png

通常のLLMの出力では、図の上段のように文章で返ってきますよね。このままではシステムには登録できません。そこで、図の下段のように、プログラム側で事前に以下を導入しておきます。

① CRMシステムに登録する内容をJSONで事前に定義。
② Function Calling などの機能を使って、 ① に沿った形で回答を出力するようにモデルへ指示。

こうすることで、ユーザー入力に対して、事前定義スキーマに沿った出力がされ、システムへの連携が可能になるわけです。

◾️チャットボットの出力をナレッジDBで使えるような構造化データにしてみよう！

というわけで、以前作ったSPSSModelerQ＆Aチャットボットをベースに改築していきたいと思います。元のチャットボットについてざっくり説明すると、以下のようなものになっています。

  • SPSSModelerについてのQ＆Aをしてくれる。
  • 過去の会話内容を踏まえ、マルチターンで返答する。
  • Web検索をして参考にしたURL付きで返答してくれる。

もちろん回答は文章で返ってくる仕様になっているので、これをナレッジDBとして保存できるようなJSON形式のデータとして出力できるように手を加えていきましょう。

※初見の方にとって見にくいので、最初にコード全文を載せておきます。以下折りたたみから参照ください。

※ここをクリックでstructuredoutput_app.pyのコード全文表示
structuredoutput_app.py
import asyncio
import os
import uuid
import json
from typing import Annotated, List

from dotenv import load_dotenv
from pydantic import BaseModel, Field, constr

from ibm_watsonx_ai.foundation_models.schema import TextChatParameters
from langchain_core.messages import AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_ibm import ChatWatsonx
from langchain_mcp_adapters.client import MultiServerMCPClient
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition


# === Pydantic schema (構造化データ定義) ===
class KnowledgeSchema(BaseModel):
    question_summary: str = Field(..., description="ユーザーの質問を簡潔に要約した文。")
    answer_summary: str = Field(..., description="LLMが生成した回答の要約。")
    keywords: List[str] = Field(..., description="回答に関連するキーワード。")
    sources: List[constr(pattern=r"^https?://")] = Field(
        ..., description="参考URLリスト（http/https のみ）。"
    )
    date: constr(pattern=r"^\d{4}-\d{2}-\d{2}$") = Field(
        ..., description="生成日（YYYY-MM-DD）。"
    )

load_dotenv()
project_id = os.getenv("WATSONX_PROJECT_ID")

parameters = TextChatParameters(
    max_tokens=2000,
    temperature=0.3,
    frequency_penalty=0.3,
    time_limit=600000,
)

model = ChatWatsonx(
    model_id="meta-llama/llama-4-maverick-17b-128e-instruct-fp8",
    project_id=project_id,
    params=parameters,
)

prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        "あなたはIBM SPSS Modelerの専門家です。必要に応じてMCPツール(tavily)で検索し、"
        "根拠が必要な箇所はURLを示し、簡潔な日本語で答えてください。"
        "次の構造化スキーマ(KnowledgeSchema)を厳密に満たすように生成してください。"
        "sourcesは有効なURL(http/https)のみ、keywordsは5個以内。dateは今日の日付をYYYY-MM-DDで。",
    ),
    MessagesPlaceholder("messages"),
])

mcp_client = MultiServerMCPClient(
    {
        "docker_gateway": {
            "transport": "stdio",
            "command": "docker",
            "args": ["mcp", "gateway", "run"],
        }
    }
)

tools = asyncio.run(mcp_client.get_tools())
tool_node = ToolNode(tools)


class GraphState(TypedDict):
    messages: Annotated[list, add_messages]


def llm_node(state: GraphState) -> dict:
    structured_model = model.bind_tools(tools, tool_choice="auto").with_structured_output(KnowledgeSchema)
    chain = prompt | structured_model

    knowledge = chain.invoke({"messages": state["messages"]})

    # ユーザーへの表示はanswer_summaryのみ。
    if isinstance(knowledge, KnowledgeSchema):
        knowledge_dict = knowledge.model_dump(by_alias=True)
        answer = knowledge.answer_summary
    else:
        # 念のためのフォールバック
        knowledge_dict = dict(knowledge)
        answer = knowledge_dict.get("answer_summary", "")

    ai_msg = AIMessage(
        content=answer,
        additional_kwargs={"knowledge": knowledge_dict},
    )
    return {"messages": [ai_msg]}


memory = InMemorySaver()
builder = StateGraph(GraphState)
builder.add_node("llm", llm_node)
builder.add_node("tools", tool_node)
builder.add_conditional_edges("llm", tools_condition)
builder.add_edge("tools", "llm")
builder.add_edge(START, "llm")

graph = builder.compile(checkpointer=memory)


THREAD_ID = str(uuid.uuid4())


async def astream_updates(user_input: str, thread_id: str = None):
    config = {"configurable": {"thread_id": thread_id}} if thread_id else None

    async for event in graph.astream(
        {"messages": [{"role": "user", "content": user_input}]},
        config=config,
        stream_mode="updates",
    ):
        if "llm" in event:
            msgs = event["llm"].get("messages", [])
            if msgs and isinstance(msgs[-1], AIMessage):
                # 画面(対話)には要約だけ
                print(msgs[-1].content)
                # ナレッジDB向けに構造化JSONも出力(ログ等で取り込みやすいようにプレフィックス付与)
                knowledge = msgs[-1].additional_kwargs.get("knowledge")
                if knowledge:
                    print("KNOWLEDGE_JSON:", json.dumps(knowledge, ensure_ascii=False))


def main():
    while True:
        try:
            user_input = input("User: ")
        except (EOFError, KeyboardInterrupt):
            break
        if user_input.strip().lower() in {"quit", "exit", "q"}:
            break
        asyncio.run(astream_updates(user_input, THREAD_ID))


if __name__ == "__main__":
    main()

① 事前定義するスキーマの実装

さて今回想定される問答はFAQです。例えば、「SPSSModelerの具体的な使い方をWebで検索して３、４ 個教えてください。」という質問をするとしましょう。すると通常は以下のような回答が出力されると思います。

AIの出力_文章
SPSS Modelerの具体的な使い方をいくつか紹介します。

1. データインポート：**入力**パレットを使用して、さまざまな形式のデータをインポートできます。例えば、カンマ区切りのCSVファイルの場合、区切り文字、ストレージ、データ型などの主要な項目を指定するだけでインポートできます。
2. 特徴量選択：**特徴量選択**ノードを使用して、対象と相関/連関の高い変数だけを選別して予測モデル作成に流し込むフィルタリングができます。
3. 異常値検査：**異常値検査**ノードを使用して、クラスタリングにより異常レコードを識別してストリームから除外させることができ、予測モデル作成のフィールド選択を補助します。
4. 予測モデル作成：50種類近い手法がGUIで提供されており、教師あり学習の**ニューラルネットワーク**、**線型モデル**、**決定木分析**、**ベイジアンネットワーク**、**ランダムフォレスト**などのほか、教師なし学習として、**K-Means**や**Kohonen**などのクラスター分析、因子分析や主成分分析に対応します。

これらの機能を使用することで、データ分析や予測モデルの構築が容易になります。

参考URL：
- https://www.stats-guild.com/ibm-spss/modeler/modeler-use
- https://www.niandc.co.jp/tech/20241206_56852/

ですが、今回はナレッジDBとして再利用できるように、質問要約、回答要約、参考URLの他に、キーワード、日付なども加えてJSONで出せるようにしたいと思います。

AIの出力_構造化データ
{
  "question_summary": "SPSS Modelerの具体的な使い方を知りたい",
  "answer_summary": "SPSS ModelerではGUI上でデータのインポート、特徴量選択、異常値検査、モデル作成などをノードで直感的に操作できます。代表的な手法にはC&Rツリーやニューラルネットワークなどがあります。",
  "keywords": ["SPSS Modeler", "データインポート", "特徴量選択", "異常値検査", "予測モデル"],
  "sources": [
    "https://www.stats-guild.com/ibm-spss/modeler/modeler-use",
    "https://www.niandc.co.jp/tech/20241206_56852/"
  ],
  "date": "2025-10-19"
}

こちらJSONで直接定義しても良いのですが、今回は以下のPydanticを使いたいと思います。

Pydanticはこのような構造化データをPythonで扱うのに重宝するライブラリで、主にデータ型のチェックと妥当性検証などを行ってくれます。何よりクラスをBaseModelから継承して「フィールド名: 型」で書くだけで良いので、読み書きがしやすい…
※以下の解説が大変わかりやすいです。

このPydanticを使って先ほどの構造化データを定義したい場合、Pythonのコードに以下のように記述していきます。事前定義はこれで完了。楽ですね〜！

Pydanticによる構造化データの定義
from pydantic import BaseModel, Field, HttpUrl
from typing import List
from datetime import date

class KnowledgeSchema(BaseModel):
    question_summary: str = Field(..., description="ユーザーの質問を簡潔に要約した文。")
    answer_summary: str = Field(..., description="LLMが生成した回答の要約。")
    keywords: List[str] = Field(..., description="回答に関連するキーワード。")
    sources: List[constr(pattern=r"^https?://")] = Field(
        ..., description="参考URLリスト（http/https のみ）。"
    )
    date: constr(pattern=r"^\d{4}-\d{2}-\d{2}$") = Field(
        ..., description="生成日（YYYY-MM-DD）。"
    )

PydanticのHttpUrlを使ってURLを処理したり、Datetimeを使って日付けを処理しようとすると、LLM側が対応しておらずエラーになるケースが多くありました。

なので今回は、制約付きの文字列を扱えるconstrを使って「https~」や「YYYY-MM-DD」以外が来ない形にしました。

ここら辺は単に私の知識不足だと思うので良いやり方があったらコメントなどで教えて欲しいです。

② スキーマに沿った出力を返すように、モデルに指示する

こちらはもっと簡単で、LangChainにおいてユーザーとLLMのやり取りを対話形式で扱うためのChatmodelクラスの１メソッド、with_structured_output()に ① を渡せばいいだけです。

  • Chatmodelの概要

  • with_structured_output()の解説

以下実装していきます。LLMの出力をそのままテキストで返すのではなく、Pydanticの KnowledgeSchema に沿った構造化データ（Structured Output）として受け取り、以下の二層出力にします。

  • 画面（ユーザー向け）は answer_summary のみを表示
  • 実際のJSONは AIMessage.additional_kwargs["knowledge"] に格納（後段のログ収集／DB登録用）
def llm_node(state: GraphState) -> dict:
    structured_model = model.bind_tools(tools, tool_choice="auto").with_structured_output(KnowledgeSchema)
    chain = prompt | structured_model

    knowledge = chain.invoke({"messages": state["messages"]})

    # ユーザーへの表示はanswer_summaryのみ。
    if isinstance(knowledge, KnowledgeSchema):
        knowledge_dict = knowledge.model_dump(by_alias=True)
        answer = knowledge.answer_summary
    else:
        # 念のためのフォールバック
        knowledge_dict = dict(knowledge)
        answer = knowledge_dict.get("answer_summary", "")

    ai_msg = AIMessage(
        content=answer,
        additional_kwargs={"knowledge": knowledge_dict},
    )
    return {"messages": [ai_msg]}

最後にユーザーへの出力部分を少し変えて、ユーザーには「要約（answer_summary）だけを表示」、同時に「ナレッジDB向けの構造化JSON をログ出力」といったする二重出力に修正します。
※実際にターミナル上で回答とJSONが続けて二つ出てきます。

async def astream_updates(user_input: str, thread_id: str = None):
    config = {"configurable": {"thread_id": thread_id}} if thread_id else None
    
    async for event in graph.astream(
        {"messages": [{"role": "user", "content": user_input}]},
        config=config,
        stream_mode="updates",
        ):
            if "llm" in event:
                msgs = event["llm"].get("messages", [])
                if msgs and isinstance(msgs[-1], AIMessage):
                    # 画面(対話)には要約だけ
                    print(msgs[-1].content)
                    # ナレッジDB向けに構造化JSONも出力(ログ等で取り込みやすいようにプレフィックス付与)
                    knowledge = msgs[-1].additional_kwargs.get("knowledge")
                    if knowledge:
                        print("KNOWLEDGE_JSON:", json.dumps(knowledge, ensure_ascii=False))

本記事では実際にDBに登録するまでは行いませんが、KNOWLEDGE_JSONを使えば簡単にDBに吐き出すことができます。

■実際に動かしてみよう

出来上がった、コードを実行してみましょう。以下今回の環境を記載します。

筆者の環境（以下の想定で進めます。）
OS     :Mac
IDE    :VSCode
python :3.11

ライブラリ
 - ibm-watsonx-ai>=1.3.13
 - langchain-core>=0.3.58
 - langchain-ibm>=0.3.10
 - langgraph>=0.4.1
 - pydantic>=2

今回のコードでもMCPツールをDocker経由で使ってWeb検索をしているので、コードを実行する前に DockerDesktop を起動させておきましょう。

前回までと同じくuvでローカル実行しておりますので、ターミナルから以下のように起動します。

source .venv/bin/activate
uv run structuredoutput_app.py

「User:」と出てきて、ユーザー入力が求められたらうまくいっています。
早速、シンプルな質問を投げてみましょう

ユーザー入力
SPSS Modelerで「決定木」が何をするのか、簡単に教えてください。

AIからの返答としてユーザー向けの要約が出た後、ナレッジDB向けの構造化JSON が出力されます。

AIの出力
決定木は、データを質問と分岐で分類していくシンプルな予測モデルです。

KNOWLEDGE_JSON: {
  "question_summary": "SPSS Modelerで決定木が何をするのか簡単に知りたいという質問。",

  "answer_summary": "決定木は、データを質問と分岐で分類していくシンプルな予測モデルです。",
  "keywords": ["決定木", "分類", "SPSS Modeler", "予測モデル"],
  "sources": ["https://ai365.jp/columns/decision-tree-analysis/"],
  "date": "2025-11-07"
}

ちゃんと要約はコンパクトに表示され、詳細は構造化JSONとして残っていますね。このJSONは、そのままナレッジDBに入れて検索・可視化・レポート化に活用できます

■ 感想・まとめ

感想

  • 会話を楽しむだけでなく、システム連携の要になる技術だと実感しました。LLMの“返答”を構造化しておくと、ログ→検索→可視化→通知など、業務フローへの自然な組み込みもきたいできそうです。
  • Pydanticをもっと使いこなせば、色々できそう…
  • 中間ステップでの安定化にも有効だと思うのでやってみたいと感じました。たとえばワークフローを
    　① URLだけを抽出するノード
    　② URLの中身を収集・要約してスキーマ（Pydantic）に詰めるノード
    　③ スキーマをもとにユーザー向け回答を生成するノード
    と分解すれば、各段階の品質が見える化・再現可能になります。

まとめ

今回の記事では　with_structured_output() + Pydantic（KnowledgeSchema）を使って出力の構造化に挑戦しました。
JSONでの出力が可能になることで、回答の制御ができるようになったのが主な利点かと思います。
これまでは主にワークフローで組んできたましたが、次の機会があればエージェント（Agent）を使って、より柔軟な自律動作ができるような改善をしたいと思います。

■ 参考

6
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
6
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?