SEあれこれ

プログラミングからSE、PM色々なこと書いていくブログ

LangGraphでマルチエージェントを構築する

Python 生成AI

以前MCPやFunction Calling、自前実装のRAGを扱ってきましたが、1つのLLM呼び出しで完結しない「複数の役割を持つエージェントが連携して処理を進める」構成を組みたい場面も出てきます。今回はそうしたワークフローをグラフ構造で定義できるLangGraphを紹介します。

LangGraphとは

　(1) LangGraphは、LangChainのエコシステム上に位置するライブラリで、LLMを使った処理の流れを「状態(State)」と「ノード(Node)」「エッジ(Edge)」からなるグラフとして定義できます。

　(2) 単純な直列のチェーンと違い、条件分岐・ループ・複数エージェント間の受け渡しといった、より複雑な制御フローを表現しやすいのが特徴です。以前扱ったStep Functionsのステートマシンに近い発想で、LLMの処理フローを組み立てるイメージです。

　(3) 各ノードは「状態を受け取り、処理をして、更新した状態を返す」という単純な関数として書けるため、ノードごとに独立してテストしやすい設計にもなります。

セットアップ

　(1) パッケージをインストールします。

pip install langgraph langchain-core

　(2) グラフ全体で受け渡す状態の型を、`TypedDict`で定義しておきます。

from typing import TypedDict

class AgentState(TypedDict):
    question: str
    research_result: str
    answer: str

ノードの定義

　(1) 各ノードは、状態を受け取って一部を更新した辞書を返す関数として定義します。

def research_node(state: AgentState) -> dict:
    question = state["question"]
    # 実際にはツール呼び出しやRAG検索などを行う
    result = search_documents(question)
    return {"research_result": result}

def answer_node(state: AgentState) -> dict:
    prompt = f"以下の情報をもとに回答してください。\n{state['research_result']}\n\n質問: {state['question']}"
    answer = llm_client.generate(prompt)
    return {"answer": answer}

　(2) 「調べる役割」と「回答をまとめる役割」を別ノードに分けておくことで、それぞれの責務がはっきりし、後から片方だけを差し替えることも容易になります。

グラフの組み立て

　(1) `StateGraph`にノードを登録し、エッジで繋いでいきます。

from langgraph.graph import StateGraph, END

graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("answer", answer_node)

graph.set_entry_point("research")
graph.add_edge("research", "answer")
graph.add_edge("answer", END)

app = graph.compile()

　(2) 実行はシンプルに、初期状態を渡すだけです。

result = app.invoke({"question": "解約方法を教えてください", "research_result": "", "answer": ""})
print(result["answer"])

条件分岐と役割分担

　(1) 質問の内容によって処理を振り分けたい場合は、`add_conditional_edges`で分岐先を関数で指定できます。

def route_by_category(state: AgentState) -> str:
    if "料金" in state["question"]:
        return "billing"
    return "general"

graph.add_conditional_edges(
    "classify",
    route_by_category,
    {"billing": "billing_node", "general": "research"},
)

　(2) これにより「分類エージェントが質問の種類を判定し、種類に応じて専門のエージェントに処理を委譲する」といった、役割分担のあるマルチエージェント構成を組めます。

ループによる自己修正

　(1) 生成した回答を別のエージェントに評価させ、不十分であれば再生成させる、というループもエッジの張り方だけで表現できます。

def review_node(state: AgentState) -> dict:
    review_prompt = f"次の回答は質問に十分答えていますか？「はい」か「いいえ」で答えてください。\n質問:{state['question']}\n回答:{state['answer']}"
    verdict = llm_client.generate(review_prompt)
    return {"review_verdict": verdict}

def should_retry(state: AgentState) -> str:
    return "answer" if "いいえ" in state.get("review_verdict", "") else END

graph.add_node("review", review_node)
graph.add_edge("answer", "review")
graph.add_conditional_edges("review", should_retry, {"answer": "answer", END: END})

　(2) ただし無限ループを避けるため、実運用では再試行回数の上限を状態に持たせ、一定回数を超えたら強制的に終了させる制御を必ず入れておく必要があります。

単純なチェーンとの使い分け

　(1) 「質問を投げて1回で回答が返ればいい」程度の用途では、LangGraphを使わずシンプルな関数呼び出しの連鎖で十分です。

　(2) 条件分岐・ループ・複数の役割を持つエージェント間の連携が必要になってきたタイミングで、LangGraphのようなグラフベースの表現に切り替えるのが妥当な判断基準だと思います。最初から複雑な構成にする必要はありません。

まとめ

複数の役割を持つエージェントを連携させる処理は、if文をネストさせて自前で制御フローを書くよりも、LangGraphでノードとエッジとして表現したほうが、後から見て処理の流れを把握しやすくなります。以前のRAG自前実装や関数呼び出しの仕組みと組み合わせれば、「調査→回答→レビュー」のような一連の流れを持つエージェントも構築しやすくなるはずです。

次回のテーマはまだ模索中なので、相談しながら決めましょう。