【実践コード付き】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、コサイン類似度)について、数学的な側面から少し深く掘り下げて解説します。
コメント
コメントを投稿