Posted inAI

線形チェーンを超えて：LangGraphによるプロダクション品質のAIエージェント構築

AI tutorial - IT technology blog
AI tutorial - IT technology blog

線形AIチェーンの罠

多くの開発者は、単純な線形チェーンの構築から始めます。プロンプトが入力され、LLMがそれを処理し、レスポンスが出力される。これは基本的なチャットボットや要約ツールには有効です。しかし、ドキュメントのスクレイピングや事実確認、あるいは本番環境のデータベースを操作する前に人間の許可を求めるような、プロフェッショナル仕様のエージェントを構築しようとすると、線形チェーンは破綻します。

すぐに「if-else」地獄に陥っている自分に気づくでしょう。標準的なPython関数で会話履歴やツールの出力を管理しようとすると、コードはスパゲッティ状態になり、悪夢のような保守性を生みます。適切な構造がなければ、エージェントは目的を見失ったり、壊れたAPIを呼び出し続ける無限ループに陥ったりする可能性があります。

なぜ従来のチェーンはプロダクションで失敗するのか

標準的なLLMフレームワークは、多くの場合、インタラクションを有向非巡回グラフ（DAG）として扱います。このモデルでは、ロジックは前方にしか進みません。しかし、現実世界のタスクは循環的です。ツールが429（Rate Limit）エラーを返した場合や、ユーザーが下書きに対してフィードバックを提供した場合には、エージェントはループして戻る必要があります。

中央集権的なステート（状態）がなければ、50KBものJSONオブジェクトを関数間で受け渡し、何も壊れないことを祈るしかありません。「メモリ」の管理は、リストに文字列を追加していく手作業の苦行になります。LangGraphは、エージェントのオーケストレーションを一方通行の道ではなく、正式な「状態マシン」として扱うことで、この問題を解決します。

LangGraphのメンタルモデル：ノード、エッジ、ステート

LangGraphを、エージェントの「脳」の設計図と考えてください。いくつかのアージェントをプロダクション環境にデプロイしてきた経験から、このフレームワークは素のスクリプトにはない安定性を提供してくれると感じています。LangGraphでは、以下の3つのコアコンポーネントを定義することが求められます。

  • ステート（State）: これは「唯一の真実のソース（Single Source of Truth）」です。通常は現在のデータを保持する TypedDict です。グラフ内のすべてのノードは、この共有メモリを読み取り、更新することができます。
  • ノード（Nodes）: これらは独立したPython関数です。あるノードはGPT-4oを呼び出し、別のノードはPostgreSQLデータベースにクエリを実行するかもしれません。
  • エッジ（Edges）: これらは経路を定義します。条件付きエッジは交通整理の役割を果たし、LLMの出力に基づいて、次のツールに進むか、タスクを終了するかを決定します。

実践：人間による監視を備えたステートフルなエージェントの構築

回答を確定させる前に、人間による「承認」を必要とするエージェントを構築してみましょう。このパターンは、100%の精度が求められる財務報告や医療アドバイスなど、リスクの高い環境では不可欠です。

1. 環境セットアップ

まず、コアライブラリをインストールします。langgraph と langchain-openai の最新バージョンが必要です。

pip install langgraph langchain_openai

2. 共有ステートの定義

ステートは会話を追跡します。ここでは reducer 関数を指定した Annotated 型を使用します。これにより、新しいLLMのレスポンスが前のメッセージを上書きするのではなく、履歴に追加されるようになります。

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

class AgentState(TypedDict):
    # 'add_messages' 関数は、新しいメッセージをマージするロジックを処理します
    messages: Annotated[list, add_messages]

3. ノードの設計

ノードはモジュール化されるべきです。この例では、1つのノードがロジックを処理し、もう1つが人間による確認のためのチェックポイントとして機能します。

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o", temperature=0)

def call_model(state: AgentState):
    response = model.invoke(state['messages'])
    return {"messages": [response]}

def human_approval_node(state: AgentState):
    # これはUIでの中断をシミュレートするプレースホルダーです
    print("--- 人間による承認待ち ---")
    return state

4. グラフの構築

次に、コンポーネントを接続します。ステートを永続化するために MemorySaver を使用します。interrupt_before パラメータがここでのポイントです。これにより実行が一時停止され、人間がエージェントの作業を検査できるようになります。

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("human_review", human_approval_node)

workflow.set_entry_point("agent")
workflow.add_edge("agent", "human_review")
workflow.add_edge("human_review", END)

memory = MemorySaver()
app = workflow.compile(checkpointer=memory, interrupt_before=["human_review"])

エラーハンドリングとリトライ

プロダクション環境のAPIは頻繁に失敗します。LangGraphを使用すると、こうしたトラブルをスマートに処理できます。すべてを巨大な try-except ブロックで囲む代わりに、フローを特定の「リトライ・ノード」にルーティングできます。

ツールが接続タイムアウトを返した場合、グラフは自動的にツール・ノードに戻ることができます。ステートに「クールダウン」メッセージを含めて、LLMに「データベースが混雑しています。5秒待ってから再試行してください」と伝えることさえ可能です。この循環的な能力により、システムは標準的なスクリプトよりも大幅に堅牢になります。

ステートの永続化管理

複数のユーザーセッションにわたってコンテキストを維持することは、よくある課題です。checkpointer を使用することで、LangGraphは各ノードの実行後にエージェントの進行状況を自動的に保存します。サーバーが再起動したり、ユーザーが2日後に戻ってきたりしても、thread_id を使用して、まったく同じスレッドから再開できます。

config = {"configurable": {"thread_id": "session_88"}}

# エージェントは 'human_review' の中断ポイントに達するまで実行されます
app.invoke({"messages": [("user", "法的概要の草案を作成してください")]}, config)

# その後、人間が草案を確認したら、Noneを渡して再開します
app.invoke(None, config)

ワークフローの洗練

複雑なエージェントの構築は反復的なプロセスです。まずはシンプルな2ノードのグラフから始めましょう。基本ロジックが固まったら、エラーハンドリングやHuman-in-the-loopのチェックポイントを追加していきます。初日から15ノードのグラフを作ろうとすると、通常、何時間もの苦痛なデバッグを強いられることになります。

LangGraphは、脆弱なデモを信頼性の高いソフトウェアへと変貌させるために必要なフレームワークを提供します。AIを状態マシンとして扱うことで、ロジックフローを完全に制御できるようになります。あなたのアプリケーションは予測可能でデバッグ可能になり、実際のユーザーの要求に応える準備が整います。

Share:
Tags: