AIエージェントをREST APIとして提供する：LangChainとLangGraphによる構築入門（第2回：LangChainとLangGraphによる処理の実装）

本記事では、前回構築したREST APIサーバーに、LangChainとLangGraphを用いてAIエージェントの処理を実装します。具体的には、以下の3つの処理をLangChainで実装し、LangGraphで繋げることで、一連のタスクとして実行できるようにします。

  1. 入力ワードをGPT-4o-miniモデルを用いてWEB検索キーワードに変換する。
  2. WEB検索キーワードを検索する。
  3. 検索結果を要約する。

1. 必要なライブラリのインストール (追加)

前回インストールしたライブラリに加えて、以下のライブラリをpipでインストールします。

pip install duckduckgo-search langchain-openai
  • duckduckgo-search: DuckDuckGo検索APIを利用するためのライブラリ
  • langchain-openai: OpenAIのモデルを利用するためのLangChain連携ライブラリ

2. LangChainによる各処理の実装

main.py を修正し、各処理を個別の関数として実装します。

agent.py としてAgentの各処理を実装します。

from traceback import print_exception

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.utilities import DuckDuckGoSearchAPIWrapper
from langchain_core.output_parsers import StrOutputParser

class AIAgent:
    def __init__(self, llm_model="gpt-4o-mini"): # デフォルトをgpt-4o-miniにする
        """
        AIエージェントの初期化。

        Args:
            llm_model (str): 使用するLLMモデルの名前。
        """
        self.llm_model = llm_model
        self.llm = self._select_llm()
        self.search = DuckDuckGoSearchAPIWrapper()

    def _select_llm(self):
        """
        LLMを選択する。
        """
        if self.llm_model == "gpt-4o-mini":
            # OpenAIのモデルを利用
            # APIキーの設定が必要です
            try:
                print(f"OPENAI_API_KEY:{os.getenv('OPENAI_API_KEY')}")
                llm = ChatOpenAI(model="gpt-4o", temperature=0.0)
                return llm
            except KeyError:
                raise KeyError("OPENAI_API_KEYが設定されていません。")
        else:
            raise ValueError(f"サポートされていないLLMモデル: {self.llm_model}")

    def create_web_search_keywords(self, input_text: str) -> dict:
        """
        入力されたテキストから、WEB検索に適したキーワードを生成する。

        Args:
            input_text (str): 入力テキスト。

        Returns:
            str: WEB検索キーワード。
        """
        prompt = ChatPromptTemplate.from_template("入力されたテキスト: {input_text} から、WEB検索に適したキーワードを３つ生成してください。")
        chain = prompt | self.llm | StrOutputParser()
        keywords = chain.invoke({"input_text": input_text})
        return {'keywords': keywords}

    def web_search(self, keywords: str) -> dict:
        """
        WEB検索キーワードを用いて、DuckDuckGoで検索を行う。

        Args:
            keywords (str): WEB検索キーワード。

        Returns:
            str: 検索結果。
        """
        results = self.search.run(keywords)
        return {'search_results': results}

    def summarize_results(self, search_results: str) -> str:
        """
        検索結果を要約する。

        Args:
            search_results (str): 検索結果。

        Returns:
            str: 要約された結果。
        """
        prompt = ChatPromptTemplate.from_template("以下の検索結果: {search_results} を要約してください。3文以内で。")
        chain = prompt | self.llm | StrOutputParser()
        summary = chain.invoke({"search_results": search_results})
        return {'summary': summary}

コードの説明:

  • ライブラリのインポート: 必要なLangChainのモジュール、DuckDuckGo検索ライブラリなどをインポートします。
  • create_web_search_keywords関数:
    • GPT-4o-miniモデルを使用して、検索キーワードを作成します。
    • ChatPromptTemplateを用いて、入力テキストからLLMに渡すプロンプトを定義します。
    • LLMChainを用いて、LLMとプロンプトを組み合わせたChainを作成し実行します。
      • LangChainはBaseで__run__を実装しており、 prompt | self.llm | StrOutputParserとすることでchainします。
  • web_search関数:
    • DuckDuckGo検索APIを用いて、指定されたキーワードでWEB検索を行います。
    • 検索結果から、テキスト部分を抽出してリストとして返します。
  • summarize_results関数:
    • GPT-4o-miniモデルを使用して、検索結果のテキストを要約します。

3. LangGraphによる処理のグラフ化

LangGraphを使用して、各処理をノードとして定義し、それらを繋げることで、処理のフローを明確に定義します。main.pyを修正します。

from langgraph.graph import StateGraph, END
from typing import TypedDict, Dict, Any

class AgentState(TypedDict):
    """
    LangGraphの状態を定義する。
    """
    input: str
    web_search_keywords: str
    search_results: str
    summary: str

def define_langraph_workflow(agent: AIAgent):
    """
    LangGraphのワークフローを定義する。

    Args:
        agent (AIAgent): AIエージェントのインスタンス。

    Returns:
        StateGraph: LangGraphのワークフロー。
    """
    builder = StateGraph(AgentState)

    builder.add_node("create_web_search_keywords", agent.create_web_search_keywords)
    builder.add_node("web_search", agent.web_search)
    builder.add_node("summarize_results", agent.summarize_results)

    builder.set_entry_point("create_web_search_keywords")

    builder.add_edge("create_web_search_keywords", "web_search")
    builder.add_edge("web_search", "summarize_results")
    builder.add_edge("summarize_results", END)

    graph = builder.compile()
    return graph

コードの説明:

  • GraphState: グラフの状態を定義する TypedDict です。ここでは、keys という辞書型のフィールドを持つことを定義しています。
  • 各ノードの処理関数:
    • 各関数は、GraphState 型の引数を受け取り、dict 型の戻り値を返します。
    • GraphState から必要な情報を取得し、処理を実行します。
    • 処理結果を dict に格納して返します。
  • construct_graph関数:
    • StateGraph を使用してグラフを構築します。
    • add_node で各ノードを追加します。
    • set_entry_point で開始ノードを設定します。
    • add_edge でノード間の接続を定義します。
    • compile でグラフをコンパイルします。

4. FastAPIのエンドポイントから呼び出し

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import os

app = FastAPI()

class Query(BaseModel):
    text: str
    max_length: Optional[int] = 50

@app.post("/generate")
async def generate_text(query: Query):
    """
    テキスト生成エンドポイント。
    """
    try:
        # AIエージェントの初期化
        agent = AIAgent(llm_model="gpt-4o-mini")

        # LangGraphワークフローの定義
        workflow = define_langraph_workflow(agent)

        # ワークフローの実行
        inputs = {"input": query.text}
        result = workflow.invoke(inputs)

        # 結果の整形
        summary = result['summary']  # 最終結果はsummaryに格納されていると仮定

        return {"result": summary} # 要約結果を返す

    except Exception as e:
        print_exception(e)
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """
    ヘルスチェックエンドポイント。
    """
    return {"status": "ok"}

コードの説明:

  • /generate エンドポイント:
    • construct_graph 関数を呼び出してグラフを構築します。
    • 入力データを辞書に格納します。
    • graph.ainvoke を使用してグラフを実行します (非同期実行)。
    • 結果から要約を取り出して返します。

5. APIのテスト

APIサーバーを再起動し、前回と同様に/generateエンドポイントをテストします。

uvicorn main:app --reload

curl コマンドなどのツールを使用して、POST リクエストを送信します。

curl -X POST -H "Content-Type: application/json" -d '{"text": "AIエージェントとは"}' http://127.0.0.1:8000/generate

APIキーが正しく設定されていれば、AIエージェントが入力テキストに基づいてWEB検索を行い、検索結果を要約したテキストがJSON形式で返されるはずです。

{"result":"AIエージェントは、自律的に環境を認識し、行動し、学習・適応するシステムであり、業務効率化やデータ分析に優れて います。生成AIとは異なり、複数のAIモデルやデバイスを組み合わせて高度なタスクを実行します。企業導入によりコスト削減や業務の自動化が期待されます。"}

まとめ

今回は、LangChainとLangGraphを使用して、AIエージェントの中核となる処理を実装しました。

  • 各処理をLangChainで実装し、関数として定義
  • LangGraphを用いて、処理のフローをグラフとして定義
  • APIエンドポイントからLangGraphを実行

次回は、今回作成したAPIを呼び出すクライアントアプリケーションを作成仕様と思いましたが、、、ソースコードが美しくない気がしますのでリファクタリングしたものを公開します。お楽しみに！

コメント

タイトルとURLをコピーしました