Azure OpenAI Service と LangChain を用いて、会話をしながら自社サービスに API を実行してくれるチャットボットを作った

この記事は、 NTT Communications Advent Calendar 2023 16日目の記事です。

こんにちは! クラウド & ネットワークサービス部の外村です。普段は VxF 基盤という社内サービス用クラウドの開発・運用をしつつ、ソフトウェアエンジニア育成研修である twada 塾の研修運営をしています。今回は自己研鑽と業務効率化を目的として大規模言語モデル (以下、LLM) を用いたチャットボットの開発に挑戦しました。 LLM を用いたアプリケーション開発に興味がある方や、LLM の選択肢として Azure OpenAI Service を検討されている方へ参考になればと思います。

本記事では以下の技術を中心に取り扱います。

Azure OpenAI Service
LangChain を用いた Function calling (非同期処理) の実装

開発のきっかけと開発したもの
Azure OpenAI Service
Azure OpenAI Service API
- API の利用
- Azure OpenAI Service API におけるリージョンごとのレスポンス時間の比較
Function calling
LangChain
LangChain の Tool と Agent を用いた Function calling の実装
おわりに

開発のきっかけと開発したもの

私が担当する VxF 基盤は Flexible InterConnect (以下、FIC) に直結しているクラウドなのですが、開発や試験をするなかで FIC 上に仮想ルータを作成したりコネクションを作成したりすることがしばしばあります。こういった作業を楽にできたらなという思いでチャットボットを作成しました。 API を通じて必要な情報を教えてあげると、FIC に対して API を実行し、リソース作成してくれます。 LLM として Azure OpenAI Service の GPT-4、アプリケーションフレームワークとして FastAPI と LangChain を用いました。アプリケーションの全体像は以下のようなイメージです。

アプリケーションの動作結果は以下のようになりました。

$ curl -X POST -H "Content-Type: application/json" -d '{"request_message":"こんにちは"}' http://127.0.0.1/v1/chat
{
    "response_message": "こんにちは！何かお手伝いできることがありますか？"
}

$ curl -X POST -H "Content-Type: application/json" -d '{"request_message":"FIC でルータを作成するために必要なことを教えてください。"}' http://127.0.0.1/v1/chat
{
    "response_message": "FICルータを作成するためには以下の情報が必要です：

    1. **名前（name）**：ルータの名前を指定します。これは一意である必要があります。
    2. **ユーザーIPアドレス（userIpAddress）**：ルータのユーザーIPアドレスを指定します。例えば、\"192.168.0.0/27\"のような形式で指定します。
    3. **エリア（area）**：ルータが配置されるエリアを指定します。\"JPEAST\"または\"JPWEST\"から選択します。
    4. **冗長化（redundant）**：ルータの冗長化を指定します。デフォルトはTrue（冗長化あり）です。
    5. **パブリックサービスエンドポイント（isPublicServiceEndpoint）**：パブリックサービスエンドポイントを使用するかどうかを指定します。デフォルトはFalse（使用しない）です。

    これらの情報を元に、FICルータの作成リクエストを送信します。ただし、ルータの作成は非同期で行われ、完了までに時間がかかることをご了承ください。"
}

$ curl -X POST -H "Content-Type: application/json" -d '{"request_message":"それでは FIC ルータを作成してください"}' http://127.0.0.1/v1/chat
{
    "response_message": "もちろんです。FICルータの作成に必要な情報を教えていただけますか？

    1. ルータの名前
    2. ユーザーのIPアドレス
    3. エリア（JPEAST または JPWEST）
    4. 冗長化の有無（デフォルトはTrue）
    5. パブリックサービスエンドポイントの有無（デフォルトはFalse）"
}

$ curl -X POST -H "Content-Type: application/json" -d '{"request_message":"はい、以下の情報で FIC のルータを作成してください。名前は tono-test-router, エリアはJPEAST、ユーザIPは192.168.0.0/27 でお願いします。"}' http://127.0.0.1/v1/chat
{
    "response_message": "FICルータの作成を開始しました。ルータのIDはF022300001179です。作成が完了するまで少々お待ちください。"
}

以降の章では、それぞれの技術要素の詳細や実装の内容について説明していきたいと思います。

Azure OpenAI Service

Azure OpenAI Service は、Microsoft Azure 上で OpenAI が提供する LLM を利用できるサービスです。 ChatGPT でも用いられている GPT-4 や GPT-3.5 (以下、まとめて GPT と呼びます) を Azure 上でデプロイし、 Azure OpenAI Studio や OpenAI API を用いて利用できます。 Azure OpenAI Service には次のような特徴があります。

Azure が提供する他のサービスとの連携が容易
Azure が提供するセキュリティ機能を利用可能
複数のリージョンにデプロイすることが可能

ChatGPT で発表された機能は順次 Azure 側でも利用可能になり、Function calling 機能や Fine-Tuning も利用可能です。機能ごとに利用できるモデルバージョンや API バージョンがあるため、公式ドキュメントをチェックしましょう。また、2023年11月時点では利用するリージョンによってはデプロイ可能なモデルに差分があったり、需要増にともない一部機能が制限されていました。

Azure OpenAI Service API

Azure OpenAI Service を利用した実装にあたり、 API call を同期処理にするか非同期処理にするか少し検討しておきたいと思います。

API の利用

Azure OpenAI Service の API は、OpenAI 公式のモジュールを使うことで利用できます。下記のコードで実行可能です。 Azure ではない OpenAI API との違いは openai.api_type や openai.api_base、engine といった指定が必要なことでしょうか。api_version も Azure で提供されたバージョンにする必要があります。

import openai

openai.api_type = "azure" # プラットフォームの宣言
openai.api_base = "https://test-azure-openai-deployment01.openai.azure.com/" # 作成した Azure OpenAI Service のエンドポイント
openai.api_version = "2023-07-01-preview" # API バージョン
openai.api_key = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" # api-key

    message_text = [
        {
            "role": "system",
            "content": "You are an AI assistant that helps people find information.",
        },
        {"role": "user", "content": "こんにちは、API とはなにか教えてください。"},
    ]

    completion = openai.ChatCompletion.create(
        engine="test-gpt35-turbo-deploy01", # デプロイした GPT
        messages=message_text,
        temperature=0,
        max_tokens=800,
        top_p=0.95,
        frequency_penalty=0,
        presence_penalty=0,
        stop=None,
    )

    print(completion["choices"][0]["message"]["content"]) # 回答結果

Azure OpenAI Service API におけるリージョンごとのレスポンス時間の比較

上記のコードを用いて、各リージョンあるいは ChatGPT で提供されている GPT で API レスポンス時間に差があるのか計測してみました。比較対象として、Azure OpenAI Service 上の 3 リージョン(Japan East、US East、UK South) にデプロイした GPT と、 OpenAI で提供されている GPT を用いました。全てのモデルは gpt-3.5-turbo (0613) を利用し、「こんにちは、API とはなにか教えてください。」というプロンプト (LLM に入力する文章) を5回なげたときのレスポンス時間 (秒) を表にまとめています。

	Azure Japan East	Azure US East	Azure UK South	OpenAI
1 回目	5.5	10.2	9.0	19.7
2 回目	4.1	4.3	9.9	18.1
3 回目	10.5	5.5	5.4	16.9
4 回目	5.3	3.7	4.8	13.0
5 回目	8.0	4.0	5.8	18.6
平均	6.7	5.5	7.0	17.3

上記の結果から以下のことが言えそうです。

どの GPT でも毎回の実行結果にばらつきがある
Azure OpenAI Service のリージョン間ではおおきな差分はない
Azure OpenAI Service では、どのリージョンでも 10秒程度レスポンスにかかることがある
Azure OpenAI Service と ChatGPT では ChatGPT のほうがレスポンスに時間がかかっている

上記の結果から、GPT との通信期間中も他の処理が止まらないように、チャットボットは非同期処理を前提に開発する必要がありそうですね。

Function calling

Function calling は OpenAI API で提供されている、 GPT を外部システムと連携させることができる機能です。事前に実行可能な関数を実装しておき、GPT に「きみは必要に応じてこんな関数を実行できるよ」というように関数の情報を渡すことで GPT が必要に応じてその関数の実行判断をしてくれる機能です。定義した関数をどのような引数を用いて実行すればよいかという判断も行ってくれます。Function calling を用いたアプリケーションのシーケンス図は以下のようになります。

Function calling は強力な機能ですが、GPT からのレスポンスに関数実行の指示がある場合は関数を実行してまた GPT に結果を渡す... といった処理は開発者で書く必要があります。しかし、LangChain を利用することでこの処理を書く必要はなくなります。詳しくは後述します。

LangChain

LangChain は LLM を用いたアプリケーション開発を円滑に進めるための OSS ライブラリです。 ChatGPT の普及とともに勢いを伸ばしている OSS であり、GitHub での Issue 数は 2023年11月時点で 1.7K にも上ります。 Python に加え、最近では TypeScript でも実装されたようです。

LLM を用いたアプリケーションでは、多くの場合下記のような実装が必要になります。

LLM 単体では会話履歴を保持できないため、アプリケーション側で管理する必要がある
LLM が未知な領域に対して、情報収集機能を持たせる必要がある等

LangChain はこのような実装をライブラリとして提供する OSS という理解をしておけば良さそうです。

LangChain の Tool と Agent を用いた Function calling の実装

Tool とは、Web 検索する関数、データベースを検索する関数、slack にメッセージを送る関数など、外部システムと連携するための関数です。

Agent とは、プロンプトに対して Tool をどの順番でどのように実行すればよいかを判断し実行してくれる仕組みです。例えば「ある作家の出版履歴を slack に通知してほしい。」というプロンプトを受け付けた場合、 Web 検索する Tool を実行し出版履歴を取得、その後 slack にメッセージを送る Tool を実行する、といった処理を実行してくれます。

LangChain では自作の関数を Tool として定義し、Agent に読み込ませることで Function calling を実現できます。詳しく見ていきましょう。

以下は、アプリケーションで用いた Python とモジュールのバージョン情報です。

Python    : 3.11.7
openai    : 0.28.0
langchain : 0.0.348
fastapi   : 0.104.1
httpx     : 0.25.2
pydantic  : 2.5.2

Azure OpenAI Service の GPT は以下の設定で用いています。

Reagion     : Japan East
Model       : gpt4
API version : 2023-07-01-preview

Function calling で実行させる関数の定義

ここでは FIC ルータ作成 API を一例にとり、Tool を作成します。まずは FIC ルータ作成のための関数 createFicRouter を実装しましょう。今回は非同期で動作させたいので、非同期実行をサポートする httpx を用いて作成しています。また、返り値として FIC から返ってくるレスポンス (json) が返却されるようにします。

import httpx
from fastapi import HTTPException
import os


async def createFicRouter(
    name, userIpAddress, area, redundant=True, isPublicServiceEndpoint=False
):
    token = await getToken() # FIC 操作のための token 取得 (詳細は省略)
    url = "https://api.ntt.com/fic-eri/v1/routers"
    headers = {
        "X-Auth-Token": token,
        "Content-Type": "application/json",
    }

    body = {
        "router": {
            "name": name,
            "area": area,
            "redundant": redundant,
            "userIpAddress": userIpAddress,
            "isPublicServiceEndpoint": isPublicServiceEndpoint,
            "tenantId": os.environ["FIC_TENANT_ID"],
        }
    }

    # POST API の実行
    async with httpx.AsyncClient(timeout=None) as client:
        r = await client.post(url, json=body, headers=headers)

    if r.status_code not in [202]:
        raise HTTPException(
            status_code=503, detail="Could not available: CreateFicRouter"
        )

    return r.json()

Tool の作成

Tool のパラメータには以下の3つがあります。

Tool としての名前
関数の説明
関数の引数

関数の引数を定義するためには Pydantic を用います。関数の引数に関して、LangChain では引数を 1 つまでに限定しているようでした。そこで、関数実行に必要な 5 引数を args という 1引数のオブジェクトとして内包させる仕組みにしました。

from pydantic import BaseModel


# 関数を実行するために必要な引数
class SchemaCreateRouter(BaseModel):
    name: str
    userIpAddress: str
    area: str
    redundant: bool
    isPublicServiceEndpoint: bool


# schemaCreateRouter で定義された5つの引数を args という一つの引数のオブジェクトにする
class SchemaArgsCreateRouter(BaseModel):
    args: SchemaCreateRouter

最後に Tool を作成します。 langchain.tools にある BaseTool を拡張することで作成可能です。下記のコードのポイントは 3点です。

description には関数の情報を詳細に記載する。この記述をもとに GPT が適切なタイミング・引数で実行判断を行う
非同期処理させたい場合は async def _arun(): で関数を実行するように記載する
args_schema には run または _arun を実行するために必要な引数を定義する

from langchain.tools import BaseTool
from pydantic import BaseModel
from typing import Type


# 作成した関数を Tool にする
class ToolCreateFicRouter(BaseTool):
    name = "createFicRouter"
    description = """
            FIC に対して Fic-Router を作成する。
            入力: name, userIpAddress (例: 192.168.0.0/27), area (JPEAST または JPWEST), redundant(Default:  True), isPublicServiceEndpoint (Default: False)
            備考: 非同期作成のため、作成完了までに時間がかかる。
        """
    args_schema: Type[BaseModel] = SchemaArgsCreateRouter

    # 同期処理
    def _run(self, args):
        raise NotImplementedError("CreateFicRouter does not support sync")

    # 非同期処理
    async def _arun(self, args):
        r = await createFicRouter(**args)
        return r

以上で Tool が完成しました。

Fuction calling を実行できる Agent の作成

上記の Tool を agent 初期化時に引数として渡すことで、 Agent が Function calling を実行できるようになります。今回は Agent を非同期で動作させたいので、initialize_agent の引数 async_mode に True を指定しましょう。

from langchain.agents import initialize_agent
from langchain.agents import AgentType
from langchain.chat_models import AzureChatOpenAI
import os


class ChatAgent:
    def __init__(self):
        # LLM の定義
        llm = AzureChatOpenAI(
            deployment_name=os.environ["DEPLOYMENT_NAME"], # デプロイ名
            model_name=os.environ["MODEL_NAME"], # gpt4
            temperature=0,
            max_tokens=2000,
        )

        # Agent の定義
        self.agent = initialize_agent(
            [ToolCreateFicRouter()], # 自作した Tool を配列に含める。Agent はこのなかで定義された Tool を使ってプロンプトに回答する
            llm,
            agent=AgentType.OPENAI_FUNCTIONS,
            verbose=True,
            return_intermediate_steps=False,
            async_mode=True,
        )

    # Agent 実行用の関数定義
    async def chat(self, message):
        r = await self.agent.arun({"input": message})
        return r

チャット用 API の作成

ここまで来れば後もう一歩です。 main.py という名前でファイルを作成し、ChatAgent class を用いて回答を生成する API を定義してコードは完成です。

from fastapi import FastAPI
from pydantic import BaseModel


# リクエストボディのスキーマを定義
class PostChatRequestBody(BaseModel):
    request_message: str

# レスポンスボディのスキーマを定義
class PostChatResponseBody(BaseModel):
    response_message: str


app = FastAPI()

# POST: /v1/chat で API を受け付ける
@app.post("/v1/chat")
async def create_chat(body: PostChatRequestBody):
    # ChatAgent class のインスタンス化
    chat_agent = ChatAgent()

    # chat() を実行
    response_message = await chat_agent.chat(body.request_message)

    # レスポンスボディの返却
    return PostChatResponseBody(response_message=response_message)

アプリケーションの起動

環境変数を設定して FastAPI のサーバを起動します。実行に必要な環境変数は以下のとおりです。記載したコードのなかで現れたものとそうでないものがありますが、LangChain が自動で読み込むものもあります。すべて設定してから FastAPI を起動しましょう。

# Azure OpenAI Service で OpenAI API を扱うために必要な環境変数
export OPENAI_API_TYPE="azure" # プラットフォームの宣言
export OPENAI_API_VERSION="2023-07-01-preview" # Function calling は 2023-07-01-preview から利用可能
export OPENAI_API_BASE="https://test-azure-openai-deployment01.openai.azure.com/" # 作成した Azure OpenAI Service のエンドポイント
export OPENAI_API_KEY="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" # API Key
export DEPLOYMENT_NAME="test-gpt-4-turbo01" # デプロイ名
export MODEL_NAME="gpt4" # モデル名

# 以下、FIC API を扱うために必要な環境変数
export FIC_TENANT_ID="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
︙

# サーバの起動
uvicorn main:app --reload

おわりに

今回は Azure OpenAI Service を用いてチャットボットを作成しました。チャットボットが必要に応じて外部 API を利用できるように、LangChain による Function calling も実装しました。今回は Function calling に焦点を当てましたが、会話管理やアプリケーションのテストなどについても、また別の機会に取り組んでいきたいと思います。

それでは、明日の記事もお楽しみに！