こんにちは、石原です。
はじめに
2025年は「AIエージェント元年」とも呼ばれ、生成AIアプリケーションの進化が目覚ましい情勢となっています。そんな中、langgraphを使用して、最も簡単にエージェントを作成する方法が create_react_agent です。本記事では、その実装にフォーカスし、ソースコードを読み解くことで、AIエージェント実装の基本や仕組みについて解説していきます。
そもそもAIエージェントって？
AIエージェントとは、生成AIを活用して半自律的に判断・行動を行うプログラムのことです。これらのエージェントは、ユーザーの指示や環境からのフィードバックに基づいて、データの解析、意思決定、タスクの自動実行などを行います。
従来のルールベースのシステムと異なり、より柔軟で高度な知識獲得と適応が可能となっており、人間のような複雑なタスク処理が実現されています。
さらに、さまざまなツールやAPIとの連携により、情報の取得から処理、最終的な結果の提示までをシームレスに行うことができ、ビジネスや日常生活の多岐にわたるシーンでその応用が期待されています。
その中でも今回実装を確認していくAIエージェントはReActエージェントと呼ばれるものです
ReActエージェント
ReActエージェントとは2022年に公開された「ReAct: Synergizing Reasoning and Acting in Language Models」によって提案された手法で、「Reason：推論」と「Act：行動」を繰り返すことで生成AIが実行可能なタスクを拡張することができます
これは、もともと提案されていた「Chain of Thought」に代表される推論をすることで性能を上げる手法と「RAG」に代表されるような生成AIと別アプリケーションの実行結果を追加することで性能を上げる手法を組み合わせた方法になっています。
現在、AIエージェントとして一般に用いられているのは≒ReActエージェントと考えても差し支えないと思います。
create_react_agentの基本となる呼び出し方
それではまず、動かし方を試してみましょう。LangGraphで用意されているcreate_react_agentを呼び出してAIエージェントを作っていきます。
ReActエージェント開発の上では「Act」となるツールと呼ばれるものは切っても切り離せない関係にあります。まずはツール作成から見てみましょう。
ツールの作成
もっとも簡単なツールを作成する方法はツールデコレーターを使用し、関数をツール化することです。
今回は簡単化のために下記のようにルールベースで都市名が与えられたときにその日付を返す関数とユーザの住所を返す関数を定義します。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 | from typing import Literalfrom langchain_core.tools import tool@tooldef get_cherry_blossom_forecast(city: Literal["札幌", "新潟", "仙台", "東京", "名古屋", "大阪", "福岡"]): """2025年のさくらの開花予想または実際の開花日を返します。""" if city == "札幌": return "札幌のさくら開花予想日は4/23日です。" elif city == "新潟": return "新潟のさくら開花予想日は4/3日です。" elif city == "仙台": return "仙台のさくら開花予想日は3/29日です。" elif city == "東京": return "東京のさくら開花日は3/24日です。" elif city == "名古屋": return "名古屋のさくら開花予想日は3/25日です。" elif city == "大阪": return "大阪のさくら開花予想日は3/25日です。" elif city == "福岡": return "福岡のさくら開花予想日は3/25日です。" else: return "情報がありません。"@tooldef get_myaddress(): """ユーザの居住地を返します""" return "東京" |
使用可能なツールはリストの形で定義します。これはツールが1つの場合でも同様です。
1 | tools = [get_cherry_blossom_forecast,get_myaddress] |
これで後はモデルと、ツールをcreate_react_agentに渡すだけです。
1 2 3 | from langgraph.prebuilt import create_react_agentmodel = ChatOpenAI(model="gpt-4o", temperature=0)graph = create_react_agent(model, tools=tools) |
invokeで呼び出すことが出来ます。
住所を与えずに呼び出してみましょう。
1 2 | input = {"messages":["この辺りの桜の開花日は？"]}graph.invoke(input)['messages'][-1].content |
1 | 東京のさくら開花日は3/24日です。 |
地方を変えて聞いてみます。
1 | graph.invoke({"messages":["大阪の桜の開花日は？"]})['messages'][-1].content |
1 | 大阪のさくらの開花予想日は3/25日です。 |
場所を聞いてみます。
1 | graph.invoke({"messages":["ここはどこ？"]})['messages'][-1].content |
1 | 東京です。 |
何も関係のない質問を投げてみます。
1 | graph.invoke({"messages":["昨日の天気は？"]})['messages'][-1].content |
1 2 | 過去の天気情報は提供していません。 現在の場所の桜の開花予想を見ますか？ |
処理の流れを見る
それではcreate_react_agentの関数を見ていきましょう。（今回の呼び出しの際に実行されていない部分はスキップしています。）
stateの指定が無い場合、AgentStateが使用されます。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | if state_schema is not None: required_keys = {"messages", "remaining_steps"} if response_format is not None: required_keys.add("structured_response") schema_keys = set(get_type_hints(state_schema)) if missing_keys := required_keys - set(schema_keys): raise ValueError(f"Missing required key(s) {missing_keys} in state_schema")if state_schema is None: state_schema = ( AgentStateWithStructuredResponse if response_format is not None else AgentState ) |
AgentStateクラスは以下のようになっています。
messagesは要素が次々と追加されていくようにadd_messagesが設定されています。
is_last_stepは最終ステップかどうかのbool、remaining_stepsは残りのステップ数が入ります。
1 2 3 4 5 6 7 8 | class AgentState(TypedDict): """The state of the agent.""" messages: Annotated[Sequence[BaseMessage], add_messages] is_last_step: IsLastStep remaining_steps: RemainingSteps |
例えば下記のようにrecursion_limitを設定している場合はここの数値がstepの各値に反映されます。
1 | config = {"recursion_limit": 10} |
はじめのagentのリターン
1 | {'is_last_step': False, 'messages': [HumanMessage(content='xxx', additional_kwargs={}, response_metadata={}, id='xxx')], 'remaining_steps': 9} |
それ以外の場合は基本的には使用されていないようです。
次の処理部分では入力されたツールリストをlanggraphに定義されているTool Node変換しています。Tool Nodeは、メッセージの最後の要素（tool_callが入っている）を取得し、対応するツールを実行、その結果となるToolMessage型に変換し、メッセージへappendするという処理が実装されています。
1 2 3 4 5 6 7 | if isinstance(tools, ToolNode): tool_classes = list(tools.tools_by_name.values()) tool_node = toolselse: tool_node = ToolNode(tools) # get the tool functions wrapped in a tool class from the ToolNode tool_classes = list(tool_node.tools_by_name.values()) |
最終的な出力の型を指定している場合は変換するためのノードが追加されます。
1 2 3 4 5 6 7 8 9 10 11 12 | # Add a structured output node if response_format is providedif response_format is not None: workflow.add_node( "generate_structured_response", RunnableCallable( generate_structured_response, agenerate_structured_response ), ) workflow.add_edge("generate_structured_response", END) should_continue_destinations = ["tools", "generate_structured_response"]else: should_continue_destinations = ["tools", END] |
エッジの追加部分ではshould_continue関数を使用しています。
1 2 3 4 5 6 7 8 9 | # We now add a conditional edgeworkflow.add_conditional_edges( # First, we define the start node. We use `agent`. # This means these are the edges taken after the `agent` node is called. "agent", # Next, we pass in the function that will determine which node is called next. should_continue, path_map=should_continue_destinations,) |
最後のメッセージクラスによって処理の分岐を実施し、AImessageでないかtool_callでなければ次のノード（先ほど指定したレスポンスフォーマットの指定によって最後か変換処理に分岐する）へ移ります。
それ以外の場合、バージョンの指定によって処理を分けています。v1の場合はToolNodeでの並列処理が行われ、v2の場合はToolNodeのインスタンスを複数立ち上げて実行する処理が立ち上がります。
要するにツールが呼び出されていたら実行し、それ以外なら終了させるだけですね。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 | # Define the function that determines whether to continue or notdef should_continue(state: StateSchema) -> Union[str, list]: messages = _get_state_value(state, "messages") last_message = messages[-1] # If there is no function call, then we finish if not isinstance(last_message, AIMessage) or not last_message.tool_calls: return END if response_format is None else "generate_structured_response" # Otherwise if there is, we continue else: if version == "v1": return "tools" elif version == "v2": tool_calls = [ tool_node.inject_tool_args(call, state, store) # type: ignore[arg-type] for call in last_message.tool_calls ] return [Send("tools", [tool_call]) for tool_call in tool_calls] |
toolによってreturn_directlyオプションが有効になっている場合はそのツール呼び出し後に実行が終了されるようになります。
1 2 3 4 | if should_return_direct: workflow.add_conditional_edges("tools", route_tool_responses)else: workflow.add_edge("tools", "agent") |
最後にlanggraphで作成したグラフをコンパイルすればエージェントの完成です！
1 2 3 4 5 6 7 8 9 10 11 | # Finally, we compile it!# This compiles it into a LangChain Runnable,# meaning you can use it as you would any other runnablereturn workflow.compile( checkpointer=checkpointer, store=store, interrupt_before=interrupt_before, interrupt_after=interrupt_after, debug=debug, name=name,) |
まとめ
この記事では、2025年を迎えた今、langgraphを使って手軽にReActエージェントを構築する方法を解説しました。ReActエージェントは「推論」と「行動」をうまく組み合わせ、スマートなタスク処理を実現します。実際の例として、桜の開花予想やユーザの住所を返す簡単なツールを作り、内部での状態管理やツール連携の仕組みを見てきました。
今回紹介したcreate_react_agentは驚くほど簡単にAIエージェントを作成できます。ツール側をメール送信のAPIやカレンダーにすることで業務サポートツールを作成したり、DBのAPIにして文章からSQLの実行までを試したり、複数のAIエージェントに仕事を割り振ることで精度の高いマルチエージェントシステムを作成することだってできます。
今後どのようなエージェントが構築されるのか自分も楽しみにしています。
