Work Documentation

Appearance

第5章 LangGraphによる高度なエージェント構築 ​

5.1 ReActエージェント（推論と行動のループ）の作成 ​

ReAct（Reasoning and Acting）は、AIが「推論（考えること）」と「行動（ツールを動かすこと）」を交互に繰り返しながら目標を達成する強力な設計パターンです。LLMに単に回答を考えさせるだけでなく、必要に応じて外部の検索エンジンや電卓などの「ツール」を実行し、その結果をもとにさらに考えを進めるというループを作成します。LangGraphでは、このReActエージェントを非常に直感的かつ安全に実装できます。

実装する際は、まずLLMが「次にどのツールを使うべきか（あるいはユーザーに最終回答を返すか）」を判断する思考ノードを作成します。次に、LLMが選んだツールを実際に動かす実行ノードを作成します。思考ノードから実行ノードへは、LLMの判断結果によって進路が変わる「条件付きエッジ」で接続します。ツールが実行されたら、その実行結果を状態（State）に保存した上で、再び思考ノードへと戻るループ（エッジ）を張ります。

以下は、LangGraphを用いた基本的なReActエージェントの実装コードです。

python
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

# 1. 状態（State）の定義
# add_messagesを指定することで、古いメッセージを消さずに末尾に追加します
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]

# 2. ツールとLLMの準備
model = ChatOpenAI(model="gpt-4o")
tools = [search_web]  # search_webは事前に定義された外部ツール
model_with_tools = model.bind_tools(tools)

# 3. ノード（処理を行う関数）の定義
def call_model(state: AgentState):
    # LLMを呼び出し、結果をメッセージリストに加えます
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# 4. グラフの構築
workflow = StateGraph(AgentState)

# ノードを登録
workflow.add_node("agent", call_model)
workflow.add_node("action", ToolNode(tools))  # ツールの実行を行う標準ノード

# エッジを定義
workflow.add_edge(START, "agent")

# 条件付きエッジで分岐を定義する判定関数
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    # LLMがツール呼び出しを求めている場合はツール実行ノードへ進む
    if last_message.tool_calls:
        return "action"
    # そうでなければ処理を終了する
    return END

workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("action", "agent")  # ツール実行後は再びモデル呼び出しに戻る

# コンパイルして実行可能なオブジェクトにする
app = workflow.compile()

このアプローチにより、開発者が事前にすべての手順をプログラムしなくても、AIが自分で考えて最適な手順で問題を解決する柔軟なシステムが構築できます。

5.2 Human-in-the-loop（人の承認や介入）の組み込み ​

AIエージェントが自律的に動くことは便利ですが、企業の重要なデータを書き換える処理や、外部へメールを送信する処理など、リスクの高い操作をAIだけに任せるのは危険です。そこで、処理の途中で一度プログラムを一時停止し、人間の承認や修正を得てから処理を再開させる設計を「Human-in-the-loop（ヒューマン・イン・ザ・ループ）」と呼びます。LangGraphは、この人間との連携処理を最初からフレームワークの機能としてサポートしています。

具体的には、特定のノード（例えば「メール送信ノード」）を実行する直前で、グラフの処理を自動的に「一時停止（Interrupt）」する設定を追加できます。一時停止したグラフは、その時点までの状態（State）を保持したまま待機状態になります。人間は画面上で「AIが作成した送信メールの下書き」を確認し、問題がなければ「承認」ボタンを押してグラフを再開させます。もし内容に誤りがあれば、人間がテキストを修正して状態（State）を書き換えた上で再開させることも可能です。

以下は、ツールノード実行の直前に一時停止を設定し、承認を挟んで再開させるコード例です。

python
# グラフをコンパイルする際、特定のノードの直前で一時停止する設定を行います
app = workflow.compile(
    interrupt_before=["action"]  # ツールを実行する前に一時停止して人間の承認を待つ
)

# 実行時
config = {"configurable": {"thread_id": "human-check-123"}}
events = app.stream({"messages": [("user", "メール送信のツールを動かして")]}, config)

# 実行すると、actionノードの手前で処理が自動的に一時停止します
for event in events:
    print(event)

# 一時停止中、人間が内容を確認して「OK」と判断したら、Noneを渡してそのまま処理を再開させます
app.invoke(None, config)

これにより、AIの自律性と人間による管理（セキュリティや品質管理）を両立させることができます。重大な失敗を防ぐためのこの安全弁のような機能は、実用的なエンタープライズアプリケーションを運用する上で極めて重要です。

5.3 状態の永続化（Persistence）とチェックポインタの仕組み ​

複雑な業務プロセスを動かすエージェントは、処理が数分から数日間に及ぶことがあります。また、処理の途中でサーバーが再起動したりエラーが発生したりした際に、最初からやり直すのは大きな時間とコストのロスになります。LangGraphでは、処理の各ステップが実行されるたびに、その時点の状態（State）をデータベースに自動で保存する「永続化（Persistence）」と「チェックポインタ」の仕組みが備わっています。

チェックポインタを有効にすると、ノードが一つ実行し終わるたびに、現在の状態のスナップショット（その瞬間の記録）が保存されます。これにより、プログラムが予期せず途中で停止してしまっても、最後に保存されたチェックポイントから何事もなかったかのように処理を再開（レジューム）できます。また、スナップショットは「スレッドID（会話やタスクごとの識別子）」に関連付けて管理されるため、何千人ものユーザーが同時に異なる会話を行っていても、それぞれの状態が混ざることなく正確に記憶されます。

以下は、メモリ上に状態を保存するチェックポインタを使用して、会話の文脈を記憶するコード例です。

python
from langgraph.checkpoint.memory import MemorySaver

# メモリに状態を保存するチェックポインタを準備（本番ではデータベースを指定可能）
memory = MemorySaver()

# コンパイル時にチェックポインタを登録
app = workflow.compile(checkpointer=memory)

# 会話ごとに一意のID（スレッドID）を設定して実行
config = {"configurable": {"thread_id": "user-session-123"}}

# 初回の質問
app.invoke({"messages": [("user", "こんにちは、私の名前は太郎です。")]}, config)

# 次の質問（同じスレッドIDを使うことで、前回の会話「太郎」を自動で覚えています）
response = app.invoke({"messages": [("user", "私の名前が分かりますか？")]}, config)
print(response["messages"][-1].content)  # 「はい、太郎さんですね」と答えます

さらに、この仕組みは次に説明する「タイムトラベル」機能の基盤にもなっており、信頼性が高く堅牢なシステムを開発するための要となっています。

5.4 タイムトラベル（過去の状態への巻き戻しと再実行） ​

LangGraphのチェックポインタによる状態の永続化を応用した非常にユニークな機能が「タイムトラベル（Time Travel）」です。これは、実行中のエージェントの歴史（過去の特定のステップ）に遡って、その時点の状態を確認したり、そこから異なる指示を与えて処理をやり直したりできる機能です。

開発や運用の現場において、「エージェントが10ステップ目で誤ったツールを選択してしまい、結果がおかしくなった」というようなバグに直面することがあります。従来のシステムでは、原因究明のために1ステップ目から再現を実行する必要がありましたが、タイムトラベルを使えば、問題の起きた9ステップ目の状態へ一瞬で巻き戻すことができます。そして、その時点のプロンプトを修正し、そこから別ルートで処理を再スタートさせることが可能です。

以下は、スレッドの過去の状態履歴を取得し、過去の時点から別の入力を与えて処理を分岐させるコード例です。

python
# 1. これまでのすべての実行履歴（チェックポイント）を取得する
history = list(app.get_state_history(config))

# 2. 過去の特定のステップ（例: 2つ前の状態）のIDを取得
checkpoint_id = history[-2].config["configurable"]["checkpoint_id"]

# 3. 過去の時点の構成オブジェクトを作成
historical_config = {
    "configurable": {
        "thread_id": "user-session-123",
        "checkpoint_id": checkpoint_id
    }
}

# 4. 過去の時点（歴史）から別の指示を与えて、新しい世界線で処理を再スタート
app.invoke(
    {"messages": [("user", "今の指示は忘れて、別の話題に変えます。")]},
    historical_config
)

また、本番環境のユーザーにとっても、「数ステップ前に戻って会話をやり直す」という便利な操作を提供できるようになります。このデバッグとユーザー体験の両面における強力な柔軟性は、状態の履歴をすべて記録しているLangGraphならではの大きな特徴です。

5.5 マルチエージェント（連携・監督パターン）の設計 ​

一つの巨大なプロンプトや単一のLLMエージェントにすべての業務を任せようとすると、エージェントの「頭脳」が混乱し、指示を忘れたり誤った判断をしたりしやすくなります。そこで、特定の専門分野に特化した複数の小さなエージェントを作り、それらを連携させて大きなタスクを解決させる設計が「マルチエージェント」システムです。

LangGraphは、このマルチエージェントを構築するのに最適なアーキテクチャを持っています。主な設計パターンとして、エージェント同士が対等に話し合いながら作業を渡していく「コラボレーションパターン」や、全体の進行役（スーパーバイザー）となるエージェントが、他の作業者エージェントに仕事を割り振って進捗を管理する「監督（スーパーバイザー）パターン」があります。各エージェントはそれぞれ独自のノードとして定義され、自分が必要な処理を終えると、マージされた状態（State）を介して次のエージェントにバトンを渡します。

以下は、スーパーバイザー（監督者）を配置し、次に実行すべき専門エージェントを振り分ける設計のコード例です。

python
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# マルチエージェント用の状態の定義
class RouterState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    next_agent: str  # 次に呼び出すエージェントの名前

# 監督者ノードの定義（LLMが次に誰が動くべきかを判断する）
def supervisor(state: RouterState):
    # LLMに会話の履歴を渡し、「研究者」か「ライター」のどちらを動かすか、または終了（FINISH）かを決定させる
    response = supervisor_chain.invoke(state["messages"])
    return {"next_agent": response.next_agent}

# 専門エージェントノードの定義
def researcher_agent(state: RouterState):
    response = researcher_chain.invoke(state["messages"])
    return {"messages": [response]}

def writer_agent(state: RouterState):
    response = writer_chain.invoke(state["messages"])
    return {"messages": [response]}

# グラフの定義
workflow = StateGraph(RouterState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("researcher", researcher_agent)
workflow.add_node("writer", writer_agent)

workflow.add_edge(START, "supervisor")

# 監督者のnext_agentの指示に基づいて遷移先を決定する条件付きエッジ
workflow.add_conditional_edges(
    "supervisor",
    lambda state: state["next_agent"],
    {
        "researcher": "researcher",
        "writer": "writer",
        "FINISH": END
    }
)

# 各専門エージェントは、作業が終わったら一度監督者に報告（戻る）する
workflow.add_edge("researcher", "supervisor")
workflow.add_edge("writer", "supervisor")

app = workflow.compile()

これにより、システムの役割分担が明確になり、各エージェントのプロンプトをシンプルに保つことができるため、全体の精度向上と開発の効率化（モジュール化）が同時に達成できます。