Posted in人工知能

LangGraphによるAIエージェント構築：複雑なワークフローと自己修復機能の実現

Artificial Intelligence tutorial - IT technology blog
Artificial Intelligence tutorial - IT technology blog

なぜLangChainだけでは不十分なのか？LangGraphが登場した理由

LangChainでチャットボットを構築したことがあるなら、逐次的なタスクには非常に適していると感じたはずです。しかし、AIにコードを自己修正させたり、テストプロセスを繰り返させたりする問題に直面すると、従来の「Chain」の限界が見えてきます。これらのChainは通常、有向非巡回グラフ（DAG）に従い、AからBへ進んで終了するだけで、前のステップに戻って間違いを修正することができません。

実務におけるAIエージェントには、思考し、試行し、繰り返す（ループ）能力が必要です。それが、私がLangGraphに移行した理由です。このライブラリを使用すると、ループ（サイクル）を持つグラフ形式のワークフローを作成できます。その結果、エージェントのState（状態）管理がより厳密かつ透明になります。

私は社内の大規模データパイプラインの処理にLangGraphを導入しました。その結果、システムはより安定し、開発者が AIの各ステップに直接介入できるようになりました。これは、これまでのデフォルトのエージェントではスムーズに実現するのが難しかった点です。

環境構築

まず、必要なライブラリをインストールします。バージョン管理を適切に行い、既存プロジェクトとの衝突を避けるために、仮想環境（venv）の使用をお勧めします。

pip install -U langgraph langchain_openai langchain_core

効率的なデバッグのために、LangSmithへの登録をお勧めします。このツールを使用すると、各トークンの詳細やLLMのレスポンス時間を追跡できます。環境変数を以下のように設定してください：

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="your-api-key"
export OPENAI_API_KEY="sk-..."

StateとNodeによるAIエージェントの設定

LangGraphでは、**State**（状態）、**Nodes**（ノード）、**Edges**（エッジ）という3つの概念に集中するだけで済みます。

1. State（状態）の定義

Stateは、エージェントが処理中のすべてのデータを保存する場所です。各ノードは実行後に情報をここに更新します。TypedDictを使用することで、データ構造が明確になり、メンテナンスが容易になります。

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # add_messagesは、メッセージを上書きするのではなく、履歴に新しいメッセージを追加します
    messages: Annotated[list, add_messages]

2. Nodes（ロジック処理）の構築

各ノードは、現在のStateを受け取り、更新された結果を返すPython関数です。以下は、GPT-4oモデルを使用した基本的なチャットボットノードの作成方法です：

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

def chatbot_node(state: AgentState):
    return {"messages": [llm.invoke(state["messages"])]}

3. グラフの設定とナビゲーション（Edges）

ここでエージェントの動作図を定義します。私の経験上、コードを書く前に紙にフローをスケッチしておくことをお勧めします。これにより、ワークフローの分岐が増えたときに混乱するのを防げます。

from langgraph.graph import StateGraph, START, END

# グラフの初期化
workflow = StateGraph(AgentState)

# ノードの追加
workflow.add_node("chatbot", chatbot_node)

# フローの接続: 開始 -> chatbot -> 終了
workflow.add_edge(START, "chatbot")
workflow.add_edge("chatbot", END)

app = workflow.compile()

条件付きエッジ（Conditional Edges）による多段階ワークフローの構築

真のエージェントは、いつツールを使うべきかを知っている必要があります。例えば、AIが株価に関する質問を受けた場合、適当に答えるのではなく、検索ツールを自律的に呼び出す必要があります。条件関数を使用してフローを制御します。

def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

workflow.add_conditional_edges(
    "chatbot",
    should_continue,
    {
        "tools": "tools_node",
        END: END
    }
)

この構造により、「LLMに尋ねる -> ツールを呼び出す -> Stateを更新する -> 再度LLMに尋ねる」というインテリジェントなループが生まれます。このプロセスは、エージェントがユーザーに対して最も正確な回答を見つけるまで繰り返されます。

コスト管理と実戦的なモニタリング

運用を開始する際、エージェントの監視は必須です。以前、早期停止メカニズムがなかったためにエージェントが無限ループに陥り、わずか15分で50ドル以上のAPI費用を消費したケースを目にしたことがあります。

このリスクを防ぐために、エージェントを呼び出す際は常にrecursion_limitを設定する必要があります：

config = {"configurable": {"thread_id": "1"}, "recursion_limit": 10}
for event in app.stream({"messages": [("user", "現在のビットコインの価格はいくらですか？")]}, config):
    for value in event.values():
        print("エージェントからの回答:", value["messages"][-1].content)

実行図を視覚的に確認するために、LangSmithを活用することを忘れないでください。グラフのロジックを変更するたびに、エッジが設計通りに遷移するか確認しましょう。LangGraphによるエージェント構築は、最初は難しく感じるかもしれませんが、複雑な自動化問題に対してもたらす柔軟性は、それに見合う価値が十分にあります。

Share:
Tags: