スキップしてメイン コンテンツに移動

【実践コード付き】LangGraphで「人間と対話する」スマートエージェントを作る

【実践コード付き】LangGraphで「人間と対話する」スマートエージェントを作る

はじめに

前回はLangGraphの概念について解説しました。今回は手を動かして、実際に動作するエージェントを構築します。

作るものは「ユーザーの質問に答えるチャットボット」ですが、ただ答えるだけでなく、必要であればWeb検索ツールを呼び出し、その結果を踏まえて回答するという、条件分岐を含んだフローを実装します。

StateGraphの定義(バケツリレーの箱)

まずは、ノード間で受け渡す「状態(State)」の型を定義します。

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]

Annotated[..., operator.add] は、「新しいメッセージが来たら、リストを上書きするのではなく追記(append)する」というLangGraph特有の記法です。

Nodeの実装(AIの思考プロセス)

次に、実際に処理を行う関数(ノード)を作ります。

def chatbot(state: AgentState):
    # LLMを呼び出して回答を生成
    response = llm.invoke(state['messages'])
    return {"messages": [response]}

ここではシンプルにしていますが、実際にはここで「Toolを呼ぶべきか?」を判断したりします。

グラフの構築とコンパイル

定義したノードをグラフに追加し、線(エッジ)で繋ぎます。

graph = StateGraph(AgentState)

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

# エントリーポイントの設定
graph.set_entry_point("chatbot")

# 終了条件(ここでは常に終了)
graph.add_edge("chatbot", END)

# コンパイル(実行可能なアプリにする)
app = graph.compile()

実行とストリーミング

作成したアプリを実行してみましょう。

inputs = {"messages": [HumanMessage(content="こんにちは")]}
for event in app.stream(inputs):
    for key, value in event.items():
        print(f"Node '{key}':")
        print(value)

グラフ構造がイメージ通りに動いているか、Asciiアートで表示する機能もあります。print(app.get_graph().draw_ascii()) で確認できます。

まとめ

これがLangGraphの「Hello World」です。コード量は増えますが、処理の流れが明示的であるため、複雑なロジックになればなるほど、この構造化されたアプローチが力を発揮します。

次回は、ベクトル検索のアルゴリズム(ANN、コサイン類似度)について、数学的な側面から少し深く掘り下げて解説します。

このブログ記事はAIを利用して自動生成されました。

コメント