【LangGraph】ReActエージェントの実装方法

2025年5月3日

当ページには広告が含まれています。

こんにちは、DXCEL WAVEの運営者(@dxcelwave)です！

こんな方におすすめ！

LangGraphを活用した開発に興味がある！
LangGraphでReActエージェントを実装したい！

【LangGraph】実装イメージ

当記事では上図のようなエージェントを実装します。

エージェントが利用できるツールにははそれぞれ以下を用意するものとします。

ツール名	入力	出力
get_bitcoin_price	データ取得期間	ビットコインの価格データ
analyze_trend	価格データ	データの考察結果

【Python】LangGraphを用いたReActの実装

それでは実際にPythonを記述しながらLangGraphでReActエージェントを構築していきます。

ライブラリ

はじめに、以下のライブラリを読み込みます。今回の例ではChatGPTのAPI_KEYを利用します。

# State
from typing_extensions       import TypedDict,Annotated

# LangChain
from IPython.display          import Image, display
from langchain_openai         import ChatOpenAI
from langchain.prompts        import PromptTemplate
from langchain.schema         import SystemMessage, HumanMessage,AIMessage
from langchain_core.tools     import tool
from langgraph.prebuilt       import ToolNode
from langchain.agents         import initialize_agent, AgentType

# LangGraph
from langgraph.graph.message  import add_messages
from langgraph.graph          import StateGraph, START, END
from langgraph.prebuilt       import create_react_agent

# その他ライブラリ
import matplotlib.pyplot as plt
import yfinance as yf
import pandas as pd

# ChatGPT API Key
API_KEY = "...."

Stateの作成

今回の例では、次のようなStateを用意します。

# ================================
# State
# ================================

class State(TypedDict,total=False):
    question: str                            # 質問
    tool_needed: bool                        # Agentのツール必要要否判断
    agent_action:str                         # Agent実行結果
    messages: Annotated[list, add_messages]  # 会話履歴

LLMの作成

後述のNodeで実行するLLMを構築します。以下を記述しましょう。

# ================================
# LLM
# ================================

llm = ChatOpenAI(
        model       = "gpt-4.1",   
        temperature = 0,
        api_key     = API_KEY
        )

# 実行
# llm.invoke("こんにちは").content

# 出力イメージ
# こんにちは！ 今日はどんなご用件でしょうか？お手伝いできることがあれば教えてください。

Toolの作成

後述するAgentが呼び出すToolを実装します。以下のコードを記述しましょう。

# ================================
# Tools
# ================================
@tool
def get_bitcoin_price(start_date:str, end_date:str)-> list[dict]: 
    """ある期間におけるビットコインの価格推移を取得"""
    # データ取得
    print("\n\nビットコインのデータを取得します...\n")    
    ticker = yf.Ticker("BTC-USD")
    data   = ticker.history(start=start_date, end=end_date, interval="1D")
    # 描画
    X = pd.to_datetime(data.index)
    Y = pd.to_numeric(data["Close"])
    plt.figure(figsize=(8, 5))
    plt.plot(X, Y, marker='o')
    plt.xlabel("date")
    plt.ylabel("close")
    plt.title("bitcoin close price over time")
    plt.grid(True)
    plt.tight_layout()
    plt.show()
    print(data.head())
    
    # JSON シリアライズ可能なリスト形式に変換
    records = data.reset_index().to_dict(orient="records")
    return records

@tool
def analyze_trend(analyze_title:str, data:list[dict])->str:
    """取得したデータのトレンドを考察"""
    # =============================
    # 直近のニュースを取得
    # =============================
    print(f"\n\n{analyze_title}の直近の経済情報を取得します...")
    news_prompt = f"直近の{analyze_title}のニュースの中で、今後の展望を示した内容のみ要約して教えて。また要約結果のみ出力すること"
    news_result = llm.invoke(news_prompt).content

    # =============================
    # 考察
    # =============================
    analyze_prompt = \
    f"""
    # 分析内容:
    {analyze_title}

    # 過去のデータ分析結果
    以下にデータ分析によって得られた結果があります。
    日付(Date)
    終値(Close)

    # 直近の経済予測
    {news_result}

    # 出力してほしい内容
    以下3点を考察して下さい。
    - 直近のヒストリカル値動き: (上昇/下落/横ばい)
    - 上記で注目すべき値動き: 考察内容を記載
    - 今後の値動き予測: 経済予測情報を参考にして結果を要約
    """
    analyze_result = llm.invoke(analyze_prompt).content
    print(f"今後の{analyze_title}の市場予測:\n\n: {analyze_result}")
    return analyze_result


# ツール
tools = [get_bitcoin_price, analyze_trend]

Nodeの作成

LangGraphの処理の中心となるNodeを作成していきます。

# ================================
# Node
# ================================

def agent(question):
    # =============================
    # プロンプト
    # =============================
    text_prompt = \
    """
    #　目的
    ユーザーの質問に対して適切なツールを実行して下さい。最終的に結果をサマリーして出力すること
    
    # ツール一覧
        - get_bitcoin_price
            descripition: ビットコインの価格推移データを取得。
            parameter:
                - start_date(開始日):yyyy-mm-dd (ユーザーの質問から開始日を推測)
                - end_date(終了日):yyyy-mm-dd (ユーザーの質問から終了日を推測)
        - analyze_trend
            description: 取得したデータを考察
            parameter:
                - analyze_title(考察する対象):ビットコイン、FX、株式
                - data:データフレーム型の取得データ
    
    # ユーザーの質問は以下です
    {input}
    """

    prompt = PromptTemplate(
        input_variables = ["input"],
        template        = (text_prompt)
        )

    # =============================
    # エージェント
    # =============================
    
    agent_executor = initialize_agent(
            tools                 = tools,
            llm                   = llm,
            agent                 = AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
            verbose               = True,
            handle_parsing_errors = True,
            agent_kwargs = {
                "prefix": prompt.template,
            })

    response = agent_executor.invoke({
                    "input": question,
                })
    
    return response
    

def agent_node(state:State)->dict:
    # 質問内容
    question = state["question"]

    # Agent実行
    response = agent(question)

    return {"messages":response.get("messages", []),"agent_action":response["output"]}


def tools_node(state: State) -> State:
    return state

def should_continue(state: State):
    return "tools" if state.get("tool_needed") else END

Graphの作成

前述で作成したState, Tool, Nodeを用いてGraphを構築します。以下のコードを実行しましょう。

# ================================
# Graph構築
# ================================
graph = StateGraph(State)
graph.set_entry_point("agent")
graph.add_node("agent", agent_node)
graph.add_node("tools", tools_node)
graph.add_edge("tools", "agent")
graph.add_conditional_edges("agent", should_continue)

app = graph.compile()

# ================================
# 可視化
# ================================
display(Image(app.get_graph().draw_mermaid_png()))

実行

最後に作成したGraphをもとに処理を実行してみましょう。

# ================================
# 実行
# ================================

initial_state = {"question":"2025年03月01から2025年05月01におけるビットコインの動向が知りたい"}

app.invoke(initial_state)


# ================================
# 出力イメージ
# ================================
# 直近のビットコイン価格は、2024年4月の半減期を経て大きく上昇した後、一時的な調整を挟みつつも、全体としては高値圏で推移しています。
# ETF承認以降、機関投資家の資金流入が続き、ボラティリティは高いものの、全体的には「上昇基調」と評価できます。

【まとめ】LangGraphのコードすべて

今回紹介したLangGraphのコードを以下にまとめて記載します。

コード

# State
from typing_extensions       import TypedDict,Annotated

# LangChain
from IPython.display          import Image, display
from langchain_openai         import ChatOpenAI
from langchain.prompts        import PromptTemplate
from langchain.schema         import SystemMessage, HumanMessage,AIMessage
from langchain_core.tools     import tool
from langgraph.prebuilt       import ToolNode
from langchain.agents         import initialize_agent, AgentType

# LangGraph
from langgraph.graph.message  import add_messages
from langgraph.graph          import StateGraph, START, END
from langgraph.prebuilt       import create_react_agent

# その他ライブラリ
import matplotlib.pyplot as plt
import yfinance as yf
import pandas as pd

# ChatGPT API Key
API_KEY = "...."

# ================================
# State
# ================================

class State(TypedDict,total=False):
    question: str                            # 質問
    tool_needed: bool                        # Agentのツール必要要否判断
    agent_action:str                         # Agent実行結果
    messages: Annotated[list, add_messages]  # 会話履歴

# ================================
# LLM
# ================================

llm = ChatOpenAI(
        model       = "gpt-4.1",   
        temperature = 0,
        api_key     = API_KEY
        )

# ================================
# Tools
# ================================
@tool
def get_bitcoin_price(start_date:str, end_date:str)-> list[dict]: 
    """ある期間におけるビットコインの価格推移を取得"""
    # データ取得
    print("\n\nビットコインのデータを取得します...\n")    
    ticker = yf.Ticker("BTC-USD")
    data   = ticker.history(start=start_date, end=end_date, interval="1D")
    # 描画
    X = pd.to_datetime(data.index)
    Y = pd.to_numeric(data["Close"])
    plt.figure(figsize=(8, 5))
    plt.plot(X, Y, marker='o')
    plt.xlabel("date")
    plt.ylabel("close")
    plt.title("bitcoin close price over time")
    plt.grid(True)
    plt.tight_layout()
    plt.show()
    print(data.head())
    
    # JSON シリアライズ可能なリスト形式に変換
    records = data.reset_index().to_dict(orient="records")
    return records

@tool
def analyze_trend(analyze_title:str, data:list[dict])->str:
    """取得したデータのトレンドを考察"""
    # =============================
    # 直近のニュースを取得
    # =============================
    print(f"\n\n{analyze_title}の直近の経済情報を取得します...")
    news_prompt = f"直近の{analyze_title}のニュースの中で、今後の展望を示した内容のみ要約して教えて。また要約結果のみ出力すること"
    news_result = llm.invoke(news_prompt).content

    # =============================
    # 考察
    # =============================
    analyze_prompt = \
    f"""
    # 分析内容:
    {analyze_title}

    # 過去のデータ分析結果
    以下にデータ分析によって得られた結果があります。
    日付(Date)
    終値(Close)

    # 直近の経済予測
    {news_result}

    # 出力してほしい内容
    以下3点を考察して下さい。
    - 直近のヒストリカル値動き: (上昇/下落/横ばい)
    - 上記で注目すべき値動き: 考察内容を記載
    - 今後の値動き予測: 経済予測情報を参考にして結果を要約
    """
    analyze_result = llm.invoke(analyze_prompt).content
    print(f"今後の{analyze_title}の市場予測:\n\n: {analyze_result}")
    return analyze_result


# ツール
tools = [get_bitcoin_price, analyze_trend]

# ================================
# Node
# ================================

def agent(question):
    # =============================
    # プロンプト
    # =============================
    text_prompt = \
    """
    #　目的
    ユーザーの質問に対して適切なツールを実行して下さい。最終的に結果をサマリーして出力すること
    
    # ツール一覧
        - get_bitcoin_price
            descripition: ビットコインの価格推移データを取得。
            parameter:
                - start_date(開始日):yyyy-mm-dd (ユーザーの質問から開始日を推測)
                - end_date(終了日):yyyy-mm-dd (ユーザーの質問から終了日を推測)
        - analyze_trend
            description: 取得したデータを考察
            parameter:
                - analyze_title(考察する対象):ビットコイン、FX、株式
                - data:データフレーム型の取得データ
    
    # ユーザーの質問は以下です
    {input}
    """

    prompt = PromptTemplate(
        input_variables = ["input"],
        template        = (text_prompt)
        )

    # =============================
    # エージェント
    # =============================
    
    # agent_executor = create_react_agent(
    #     model        = llm,
    #     tools        = tools,
    #     state_schema = State,
    #     prompt       = prompt.template,
    #     debug        = False,
    #     )

    # response = agent_executor.invoke({
    #                 "input": question,
    #                 "remaining_steps": 2,
    #                 "is_last_step": False,
    #             })
    
    agent_executor = initialize_agent(
            tools                 = tools,
            llm                   = llm,
            agent                 = AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
            verbose               = True,
            handle_parsing_errors = True,
            agent_kwargs = {
                "prefix": prompt.template,
            })

    response = agent_executor.invoke({
                    "input": question,
                })
    
    return response
    

def agent_node(state:State)->dict:
    # 質問内容
    question = state["question"]

    # Agent実行
    response = agent(question)

    return {"messages":response.get("messages", []),"agent_action":response["output"]}


def tools_node(state: State) -> State:
    return state

def should_continue(state: State):
    return "tools" if state.get("tool_needed") else END

# ================================
# Graph構築
# ================================
graph = StateGraph(State)
graph.set_entry_point("agent")
graph.add_node("agent", agent_node)
graph.add_node("tools", tools_node)
graph.add_edge("tools", "agent")
graph.add_conditional_edges("agent", should_continue)

app = graph.compile()

# ================================
# 可視化
# ================================
display(Image(app.get_graph().draw_mermaid_png()))

実行用

# ================================
# 実行
# ================================

initial_state = {"question":"2025年03月01から2025年05月01におけるビットコインの動向が知りたい"}

app.invoke(initial_state)

最後に

この記事が気に入ったら
フォローしてね！

Follow @dxcelwave

本記事をシェア！