今日は、LangGraphを使ったAI Agentの状態機械（State Machine）開発について、丁寧な説明をしていきます。「状態機械ってなに？」「API呼び出しってどうするの？」という方から、「LangGraphを始めてみたいけど、どこから手を付けていいかわからない」という方まで対象に、スクリーンショット例穿插しながら学んでいきましょう。

前提条件：必要なものと下準備

まず、以下をご用意ください：

1. 状態機械（State Machine）とは？

状態機械とは、プログラムが「現在の状態」を持ち、その状態に応じて「次の行動」を決める仕組みのことです。日常に例えると、自动販売機を想像してください：

AI Agentの世界では、「ユーザーの質問を受け取る」→「回答を生成する」→「ユーザーに返す」という流れを、この状態機械で管理します。

2. LangGraphとは？

LangGraphは、LangChainファミリーの一つです。大きな利点は三点：第一に、状態を明確に管理できる設計になっていること。第二に、分岐やループが簡単に作れること。第三に、デバッグや可視化がしやすいことです。

ここではHolySheep AIをAPI基盤として使用します。HolySheep AIは、レートが¥1=$1（公式サイト¥7.3=$1相比85%節約）と非常にお得で、WeChat PayやAlipayにも対応しています。さらに、レイテンシが50ミリ秒未満と高速な点も嬉しいです。2026年現在の出力価格は、GPT-4.1が$8/MTok、Claude Sonnet 4.5が$15/MTok、Gemini 2.5 Flashが$2.50/MTok、DeepSeek V3.2が$0.42/MTokとなっており、用途に応じて柔軟に選び分けられます。

3. プロジェクトの準備

まず、必要なライブラリをインストールしましょう。ターミナルまたはコマンドプロンプトで以下を実行してください：

pip install langgraph langchain-openai python-dotenv requests

次に、環境変数を設定するためのファイルを作成します。プロジェクトのルートディレクトリに.envというファイルを作成し、以下の内容を記述してください：

HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY

スクリーンショット例：「.envファイルの保存場所と内容の例」プロジェクトフォルダ直下に.envファイルがある状态を確認してください。ファイル名は必ず.envとして、先頭にドットをつけてください。

4. HolySheep AI APIへの接続設定

LangChainを通じてHolySheep AIに接続するための設定を行います。holysheep_config.pyというファイル名で以下のコードを作成してください：

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()


HolySheep AI の設定

HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
BASE_URL = "https://api.holysheep.ai/v1"


ChatOpenAI互換のモデルを設定

llm = ChatOpenAI(
    api_key=HOLYSHEEP_API_KEY,
    base_url=BASE_URL,
    model="gpt-4.1",  # または "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"
    temperature=0.7,
)

print("HolySheep AI への接続設定完了")
print(f"接続先: {BASE_URL}")
print(f"モデル: gpt-4.1")

このコードを実行して、「HolySheep AI への接続設定完了」と表示されれば準備完了です。もしModuleNotFoundErrorが出た場合は、pip installのやり直しを確認してください。

5. 基本的な状態機械を実装する

ここからは、実際のLangGraphコードを作成していきます。シンプルな「質問→分析→回答」という三段階の状態機械を一緒に作ってしていきましょう。

5-1. 状態の定義

まず、Agentが管理すべき「状態」を定義します。PythonのTypedDictを使って、どのようなデータが状態に含まれるかを明示します：

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator

class AgentState(TypedDict):
    """LangGraphが管理する状態の内容"""
    messages: Annotated[Sequence[BaseMessage], operator.add]
    user_input: str
    intent: str
    response: str
    next_action: str

print("AgentState の定義完了")
print("状態に含まれる項目: messages, user_input, intent, response, next_action")

各項目の意味は次のとおりです：

5-2. ノード（処理単位）を作成する

状態機械における「ノード」は、一つの処理单元です。まずは三つの基本ノードを作成します：

from langchain_core.messages import HumanMessage, AIMessage

def receive_input(state: AgentState) -> AgentState:
    """ノード1: ユーザー入力を 받는"""
    user_input = state["user_input"]
    print(f"入力を受け取りました: {user_input[:50]}...")
    return state

def analyze_intent(state: AgentState) -> AgentState:
    """ノード2: ユーザーの意図を分析する"""
    user_input = state["user_input"]
    
    prompt = f"""ユーザーの以下の入力を分析し、最も適切なカテゴリを one_word で返してください。
    選択肢: question, booking, complaint, other
    
    入力: {user_input}
    
    one_word で返答:"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    intent = response.content.strip().lower()
    
    print(f"意図を分析しました: {intent}")
    return {"intent": intent}

def generate_response(state: AgentState) -> AgentState:
    """ノード3: 回答を生成する"""
    intent = state["intent"]
    user_input = state["user_input"]
    
    prompt = f"""ユーザーの入力と意図に基づいて、適切な回答を生成してください。
    
    意図: {intent}
    入力: {user_input}
    
    回答:"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    
    return {
        "response": response.content,
        "messages": [AIMessage(content=response.content)]
    }

print("三つのノード関数の定義完了")

5-3. グラフを構築して 연결する

StateGraphを使って、ノード同士を接続していきます。ここがLangGraphの核心적인部分です：

from langgraph.graph import StateGraph, START, END


グラフの構築

workflow = StateGraph(AgentState)


ノードの登録

workflow.add_node("receive_input", receive_input)
workflow.add_node("analyze_intent", analyze_intent)
workflow.add_node("generate_response", generate_response)


開始点から最初のノードへ

workflow.add_edge(START, "receive_input")


ノード間の接続

workflow.add_edge("receive_input", "analyze_intent")
workflow.add_edge("analyze_intent", "generate_response")


終了点へ

workflow.add_edge("generate_response", END)


グラフをコンパイル

app = workflow.compile()

print("LangGraph のコンパイル完了")
print("グラフの流れ: START → receive_input → analyze_intent → generate_response → END")

スクリーンショット例：「グラフ構造の概念図」四角形の箱が三つ（左からreceive_input、analyze_intent、generate_response）があり、矢印が左から右へ流れ、最左にSTART、最右にENDが表示されます。

5-4. 実際に実行してみる

完成したグラフを実行してみましょう：

# グラフの実行
result = app.invoke({
    "messages": [],
    "user_input": "明日の東京の天気を教えてください",
    "intent": "",
    "response": "",
    "next_action": ""
})

print("\n=== 実行結果 ===")
print(f"分析された意図: {result['intent']}")
print(f"生成された回答:\n{result['response']}")

実行結果として、意図が「question」と判定され、天気についての回答が生成されることを確認できるはずです。

6. 条件分岐を追加する

より実用的なAgentにするために、条件分岐を追加します。例えば、意図が「booking」（予約）だった場合は追加情報を求め、それ以外の場合は直接回答を返すという流れを作ってみましょう。

from langgraph.graph import StateGraph, START, END

def decide_next_step(state: AgentState) -> str:
    """条件分岐：意図に基づいて次のステップを決定"""
    intent = state["intent"]
    
    if intent == "booking":
        return "collect_info"
    elif intent == "complaint":
        return "escalate"
    else:
        return "generate_response"

def collect_booking_info(state: AgentState) -> AgentState:
    """予約情報の收集"""
    print("予約情報の收集を開始します...")
    return {"next_action": "collect_info"}

def escalate_complaint(state: AgentState) -> AgentState:
    """苦情のエスカレーション"""
    print("苦情をエスカレーションします...")
    response = "お問い合わせありがとうございます。ご意見を確認し、担当者より24時間以内にご連絡いたします。"
    return {
        "response": response,
        "messages": [AIMessage(content=response)]
    }


新しいグラフを構築

workflow = StateGraph(AgentState)


全ノードを追加

workflow.add_node("receive_input", receive_input)
workflow.add_node("analyze_intent", analyze_intent)
workflow.add_node("collect_info", collect_booking_info)
workflow.add_node("escalate", escalate_complaint)
workflow.add_node("generate_response", generate_response)


STARTから開始

workflow.add_edge(START, "receive_input")
workflow.add_edge("receive_input", "analyze_intent")


条件分岐を追加（重要！）

workflow.add_conditional_edges(
    "analyze_intent",
    decide_next_step,
    {
        "collect_info": "collect_info",
        "escalate": "escalate",
        "generate_response": "generate_response"
    }
)


各ブランチの終点到 END への接続

workflow.add_edge("collect_info", "generate_response")
workflow.add_edge("escalate", END)
workflow.add_edge("generate_response", END)


コンパイル

app = workflow.compile()

print("条件分岐を含むグラフのコンパイル完了")

このコードでは、add_conditional_edgesを使用して、分岐のロジックをdecide_next_step関数で管理しています。予約なら追加情報を收集し、苦情ならエスカレーションし、それ以外は普通に回答生成に進みます。

7. ループ構造を追加する

実際のアプリケーションでは、「情報不足なのでユーザーに追加質問する」→「回答を生成する」というループが必要なことがあります。LangGraphでは簡単にループを表現できます：

from langgraph.graph import StateGraph, START, END

def check_information(state: AgentState) -> AgentState:
    """情報が不足しているかチェック"""
    messages = state["messages"]
    
    # 最後の数件のメッセージで判断
    if len(messages) < 3:
        return {"next_action": "need_more_info"}
    else:
        return {"next_action": "ready"}

def ask_for_info(state: AgentState) -> AgentState:
    """不足情報を求める"""
    prompt = "もう少し情報が必要です。具体的にどの部分について知りたいですか？"
    return {
        "messages": [AIMessage(content=prompt)]
    }

def should_continue(state: AgentState) -> str:
    """ループを続けるか終了するかを決定"""
    if state["next_action"] == "need_more_info":
        return "ask_for_info"
    return END


ループを含むグラフ

workflow = StateGraph(AgentState)

workflow.add_node("receive_input", receive_input)
workflow.add_node("analyze_intent", analyze_intent)
workflow.add_node("check_info", check_information)
workflow.add_node("ask_for_info", ask_for_info)
workflow.add_node("generate_response", generate_response)

workflow.add_edge(START, "receive_input")
workflow.add_edge("receive_input", "analyze_intent")
workflow.add_edge("analyze_intent", "check_info")


check_infoから条件分岐

workflow.add_conditional_edges(
    "check_info",
    should_continue,
    {
        "ask_for_info": "ask_for_info",
        END: "generate_response"
    }
)


ask_for_infoからはcheck_infoに戻る（ループ！）

workflow.add_edge("ask_for_info", "check_info")
workflow.add_edge("generate_response", END)

app = workflow.compile()
print("ループ構造を含むグラフのコンパイル完了")

このグラフでは、情報が不足している間はask_for_infoとcheck_infoの間を行き来し情報が 충분になったらgenerate_responseに進んで終了します。実務ではここにユーザーの返信を受け取る处理も追加しますが、基本構造はこれで完成です。

8. 実践例：マルチツールAgent

最後に、複数のAPI呼び出しを組み合わせた実用的な例を紹介します。天気検索とイベント検索の二つのツールを使い分けるAgentを作成しましょう：

# ツールの定義
def weather_search(location: str, date: str) -> str:
    """天気検索APIを呼び出す（デモ用）"""
    # 実際にはここで HolySheep AI の別モデルや外部APIを呼び出します
    return f"{date}の{location}の天気は晴れです。気温は22度预计です。"

def event_search(location: str, date: str) -> str:
    """イベント検索APIを呼び出す（デモ用）"""
    return f"{date}の{location}での予定：午後2時から地元のMarketsがあります。"

def route_through_tools(state: AgentState) -> AgentState:
    """ツールの選択と実行"""
    user_input = state["user_input"].lower()
    intent = state["intent"]
    
    # 場所と日付の抽出（简易な実装）
    location = "東京"
    date = "明日"
    
    if "天気" in user_input or intent == "weather":
        result = weather_search(location, date)
        tool_used = "weather_search"
    elif "イベント" in user_input or "予定" in user_input or intent == "event":
        result = event_search(location, date)
        tool_used = "event_search"
    else:
        result = generate_response(state)["response"]
        tool_used = "llm_response"
    
    return {
        "response": result,
        "messages": [AIMessage(content=f"[使用ツール: {tool_used}]\n{result}")],
        "next_action": tool_used
    }


ツール選択グラフ

workflow = StateGraph(AgentState)

workflow.add_node("receive_input", receive_input)
workflow.add_node("analyze_intent", analyze_intent)
workflow.add_node("route_tools", route_through_tools)

workflow.add_edge(START, "receive_input")
workflow.add_edge("receive_input", "analyze_intent")
workflow.add_edge("analyze_intent", "route_tools")
workflow.add_edge("route_tools", END)

app = workflow.compile()


テスト実行

test_inputs = [
    "明後日の神戸の天気を教えて",
    "今週の週末に大阪であったイベントは何がありますか？"
]

for input_text in test_inputs:
    print(f"\n{'='*50}")
    print(f"入力: {input_text}")
    result = app.invoke({
        "messages": [],
        "user_input": input_text,
        "intent": "",
        "response": "",
        "next_action": ""
    })
    print(f"結果: {result['response']}")

この例では、ユーザーの質問内容に応じて適切なツール（天気検索またはイベント検索）を選択·実行し 결과를返すという流れを実現しています。実際のプロジェクトでは、各ツールの 部分をもっと详细に実装し、APIキー管理やエラーハンドリングも追加してください。

よくあるエラーと対処法

エラー1: AuthenticationError - APIキーが認識されない

エラーメッセージ例：

AuthenticationError: Incorrect API key provided: YOUR_HOLYSHEEP_API_KEY

原因と解決策：APIキーが正しく.envファイルから読み込まれていない、またはキーが無効です。以下の確認步骤を試してください：まず、.envファイルのキーが本当にHolySheep AIダッシュボードで表示されているものと同じか確認してください。次に、load_dotenv()の返し值がTrueになっているか確認し、キー自体をダッシュボードで再生成してみてください。

# デバッグ用の確認コード
import os
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv("HOLYSHEEP_API_KEY")
print(f"API Key loaded: {api_key[:10]}..." if api_key else "API Key is None")

エラー2: RateLimitError - API呼び出し回数制限を超えた

エラーメッセージ例：

RateLimitError: Rate limit exceeded. Please wait before making more requests.



原因と解決策：短時間に过多なAPIリクエストを送ってしまった場合に発生します。HolySheep AIの無料の月間枠を使い果たしたする可能性もあります。time.sleep()でリクエスト間に待機時間を入れ、批量処理の場合はバックオフ方式是定してください。また、アカウントの残额度をダッシュボードで確認し、必要であればクレジットの追加を検討してください。



import time
from tenacity import retry, wait_exponential, stop_after_attempt

@retry(wait=wait_exponential(multiplier=1, min=2, max=60), stop=stop_after_attempt(3))
def safe_api_call(prompt):
    try:
        response = llm.invoke([HumanMessage(content=prompt)])
        return response
    except RateLimitError:
        print("レート制限を検知。再試行します...")
        raise




エラー3: InvalidStateError - 状態に必須キーが足りない



エラーメッセージ例：



KeyError: 'intent'
During call to StateGraph.invoke()



原因と解決策：AgentStateの定義に存在するキーが、初期状態で渡されていない場合に発生します。app.invoke()呼ぶ際は、必ず全てのキーを含む辞書を渡すか、不足しているキーに的空文字列や空のリストを設定してください。



# 正しい初期状態の例
initial_state = {
    "messages": [],
    "user_input": "テスト入力",
    "intent": "",  # 空文字列で初期化
    "response": "",  # 空文字列で初期化
    "next_action": ""  # 空文字列で初期化
}

result = app.invoke(initial_state)




エラー4: LangGraph循環参照エラー



エラーメッセージ例：



ValueError: Nodes form cycle: node_a -> node_b -> node_a



原因と解決策：グラフの接続定義で意図しないループが発生しています。特にconditional_edgesの設定時に、終了条件のEND定数を忘れ全日制的に自身に戻ってしまう場合が多いです。should_continue関数でENDを返し忘れないか、グラフの構造を可視化して確認してください。



# グラフを画像として出力して確認
from langgraph.drawing import draw_graphviz


グラフの構造をDOTフォーマットで出力

dot_graph = app.get_graph().draw_mermaid()
with open("graph_structure.txt", "w") as f:
    f.write(dot_graph)

print("グラフ構造を graph_structure.txt に出力しました")
print("https://mermaid.live/ にアクセスして確認できます")




まとめ



今日は、LangGraphを使った状態機械の基本から、条件分岐やループを含む実践的なAgentの開発까지を学びました。ポイントとしては、第一に状態を明確に定義することが重要であること、第二にノードという処理単位で機能を分割すること、第三にconditional_edgesを使って柔軟な分岐を実現できること、这三つを意識していただければと思います。



HolySheep AIを組み合わせることで、GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash、DeepSeek V3.2など、主要なモデルを¥1=$1という破格のレートで利用できます。DeepSeek V3.2に至っては$0.42/MTokという非常に经济的な价格で提供服务しており、コストパフォマンスを重視する開発者には特におすすめです。



まずは今回作成したシンプルな状態機械から始めて、少しずつ機能を追加していけば、きっと理解が深まるはずです。Happy coding!



👉 HolySheep AI に登録して無料クレジットを獲得


関連リソース
関連記事