はじめに
これまで耳にしたことがあったが、実態が何であるかあまり掴めていなかったLangChainとLangGraphを使ってAIエージェントを実装し、実際に動かしてみました。
率直な感想としては、こんなにも少ないコードで簡単なAIエージェントを実装できてしまっていいのかと驚くばかりでした。
LangChainについて
LangChainとはLLMを活用したアプリケーション開発用のフレームワークです。
特にAIエージェントとの橋渡し部分を担っており、アプリ内で使いたいAIモデルやモデルに渡すツールなどの管理を行うことができます。
LangGraphについて
LangChainのエコシステムの一つとしてLangGraphがあります。
エージェントのための低レベルオーケストレーションフレームワークという位置付けでエージェントの構築や管理を行うために利用されます。
実装してみる
ここからは実際にLangChainとLangGraphを用いて簡単なAIエージェントを実装した時の手順です。
実行環境
| 項目 | バージョン |
|---|---|
| 端末 | MacbookAir M4 |
| Python | 3.12 |
| uv | 0.9.18 |
| langChain | 1.2.15 |
| langGraph | 1.1.8 |
| モデル | Amazon Bedrock / Claude Haiku 4.5 |
プロジェクト作成
まず作業用のディレクトリを作成し、uv initでプロジェクトを作成します。
# ディレクトリ作成と移動
mkdir bedrock-langgraph-agent && cd bedrock-langgraph-agent
# プロジェクト初期化
uv init
フレームワークのインストール
開発に必要となるパッケージのインストールを行います。
uv add langchain langchain-aws langgraph boto3
エージェントの実装
プロジェクトディレクトリ内にagent.pyというファイルを新規作成し、エージェントの実装を行います。
今回はマルチエージェントで動作するコードとしたいため、researcherとwriterという二つのエージェント（ノード）を作成しています。
それぞれのエージェントにはシステムプロンプトとなるコンテキストや、利用するモデルを与えています。
利用するモデルはAmazon Bedrockで提供されているClaude Haikuを利用するため、ChatBedrockにて初期化処理を行なっています。
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_aws import ChatBedrock
from langchain_core.messages import HumanMessage, SystemMessage
# 状態 (State) の定義
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
# モデルの初期化
llm = ChatBedrock(
model_id="jp.anthropic.claude-haiku-4-5-20251001-v1:0",
model_kwargs={"temperature": 0.5}
)
# ノード（エージェント）の定義
def researcher_node(state: AgentState):
"""リサーチャー：情報を整理してポイントを出す"""
messages = state["messages"]
sys_msg = SystemMessage(content="あなたは優秀なリサーチャーです。与えられたテーマについて、重要なポイントを3つ箇条書きで挙げてください。")
response = llm.invoke([sys_msg] + messages)
return {"messages": [response]}
def writer_node(state: AgentState):
"""ライター：リサーチ結果を元に記事を書く"""
messages = state["messages"]
sys_msg = SystemMessage(content="あなたはプロのライターです。これまでの会話をもとに、一般読者向けの分かりやすく短いブログ記事（200文字程度）を作成してください。")
response = llm.invoke([sys_msg] + messages)
return {"messages": [response]}
グラフの作成
作成した2つのエージェントをグラフという形で接続し、マルチエージェントとして動作するようコードを追加します。
add_nodeを使い、workflowにノードを宣言し、add_edgeを使ってそれらのノードがどのような順番で動作するのかを記述します。
ここではresearcherからwriterへと処理が流れるように実装を行なっています。
# グラフの構築
workflow = StateGraph(AgentState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", END)
app = workflow.compile()
処理実行部の実装
作成したマルチエージェントを呼び出すための処理を実装します。
インプットとなるプロンプトはHumanMessageとしてここで与えるようにしています。
# 5. 実行
if __name__ == "__main__":
print("AIマルチエージェントを起動します...\n")
inputs = {"messages": [HumanMessage(content="生成AIが教育に与える影響について")]}
for output in app.stream(inputs):
for node_name, state_update in output.items():
print(f"=== Agent: {node_name} ===")
print(state_update["messages"][-1].content)
print("\n" + "-"*40 + "\n")
ここまで実装したところでuv runコマンドでagent.pyを実行するとエージェントが動作し、処理が帰ってくることが確認できます。
なお、動作させる前にAWSの認証情報を取得しプロファイルとして設定することをお忘れなく。
実際に帰ってきた結果がこちら
AIマルチエージェントを起動します。。。
=== Agent: researcher ===
# AIマルチエージェントの仕組み
## 重要なポイント
• **複数の独立したAIエージェントの協働**
- 各エージェントが特定のタスクや役割を担当し、自律的に行動
- エージェント間で情報交換・協調することで、単一エージェントより複雑な問題を解決
- 例：営業担当エージェント、分析担当エージェント、承認担当エージェントが連携
• **通信・調整メカニズムの構築**
- エージェント間のメッセージ交換やAPI連携により、タスクの依存関係を管理
- 中央調整役（オーケストレーター）が全体の流れを制御する方式と、分散型で自律的に調整する方式がある
- 競合や矛盾の解決、優先順位の決定ロジックが必要
• **段階的なタスク分解と結果統合**
- 大きな問題を複数の小さなサブタスクに分割し、各エージェントが並列実行
- 各エージェントの出力を集約・検証して最終結果を生成
- 反復的なフィードバックループにより精度を向上
====================
=== Agent: writer ===
## 実用的な応用例
顧客対応、データ分析、プロジェクト管理など、複数の視点や専門知識が必要な業務で活躍しています。
====================
それぞれのエージェントが指定した順に動き、結果を返してくれていることが確認できました！
条件分岐を実装してみる
LangGraphを使うことでエージェントの結果に応じて処理を分岐させることができます。
ここではwriterが書いた文章をチェックするレビュアーエージェントを追加し、NGであればwriterへ差し戻し、OKなら処理を終了するという処理を実装してみます。
上記と同様の実装方法でレビュアーノードを追加します。
def reviewer_node(state: AgentState):
"""編集長（レビュアー）：ライターの記事を審査する"""
messages = state["messages"]
sys_msg = SystemMessage(content="""あなたは厳しい編集長です。直前のメッセージ（ライターが書いた記事）を審査してください。
- 200文字前後で分かりやすく書かれていれば『【合格】』というキーワードを含めて褒めてください。
- 説明が難しすぎたり、文字数が極端に合っていない場合は『【修正指示】』というキーワードを含めて、具体的な修正点を指摘してください。""")
response = llm.invoke([sys_msg] + messages)
return {"messages": [response]}
グラフ部分は以下のように修正します。
ポイントとなるのはworkflow.add_conditional_edgesです。ここでレビュアーが返してきた結果を受けてその後の振る舞いをどうするかを実装します。ここではreview_routerというルーティング関数を宣言し、その中でレビュー結果が合格であるか判定する条件分岐を実装しています。
# ★ 条件分岐のルーティング関数
def review_router(state: AgentState) -> str:
"""編集長のメッセージを読み取って、次の行き先を決定する"""
last_message = state["messages"][-1].content
if "【合格】" in last_message:
return "end" # 合格なら終了へ
else:
return "rewrite" # それ以外（修正指示など）なら書き直しへ
workflow = StateGraph(AgentState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)
workflow.add_node("reviewer", reviewer_node) # 編集長ノードを追加
# エッジの定義
workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer") # ライターが書いたら編集長へ
# ★ 条件付きエッジの追加（ここが分岐の要）
workflow.add_conditional_edges(
"reviewer", # 分岐の出発点となるノード
review_router, # 行き先を決定する関数
{
"end": END, # 関数が"end"を返したらEND（終了）へ
"rewrite": "writer" # 関数が"rewrite"を返したらwriter（ライター）へ差し戻し
}
)
app = workflow.compile()
最後に実行部についても修正します。
writerとreviewerのやりとりが無限ループに陥らないよう、明示的にループの回数を設定します。
if __name__ == "__main__":
print("AIマルチエージェント（編集長チェック付き）を起動します...\n")
# 意図的に難しいお題を出して、修正ループを発生しやすくしてみます
inputs = {"messages": [HumanMessage(content="量子コンピュータが暗号技術に与える影響について、小学生でもわかるように書いて")]}
# 無限ループ防止のため、recursion_limit（最大反復回数）を明示的に設定
config = {"recursion_limit": 10}
for output in app.stream(inputs, config):
for node_name, state_update in output.items():
print(f"=== Agent: {node_name} ===")
print(state_update["messages"][-1].content)
print("\n" + "="*40 + "\n")
この状態で実行してみると確かにwriterとreviewerが複数回やり取りを行う様子が確認できたのですが、肝心のwriterの出力結果が見えませんでした。
調べてみるとこれはAnthropicの提供しているMessagesAPIがユーザー（人間）とアシスタント（AI）のメッセージが交互に来ることを想定しているために発生した挙動とのことでした。
Our models are trained to operate on alternating user and assistant conversational turns. When creating a new Message, you specify the prior conversational turns with the messages parameter, and the model then generates the next Message in the conversation. Consecutive user or assistant turns in your request will be combined into a single turn.
先ほどまでの実装だと、messageの中身がuser -> assistant -> assistantという順番担ってしまっていたため、正常に動作しない状態だったということですね。
対応策としてAIの出力内容を取得してHumanMessageとしてコンテキストを再生成し、次のエージェントのインプットとなるようにしました。
こうすることでmessageの順番がuserとassistantの交互に並ぶことになります。
def writer_node(state: AgentState):
"""ライター：リサーチ結果や指摘を元に記事を書く"""
messages = state["messages"]
# これまでの会話履歴（リサーチ結果や指摘）を一つのテキストにまとめる
history = "\n".join([f"[{m.type}]: {m.content}" for m in messages])
sys_msg = SystemMessage(content="あなたはプロのライターです。")
# まとめた履歴を「人間からの指示」としてLLMに渡す
prompt = HumanMessage(content=f"以下の情報（これまでの経緯）をもとに、一般読者向けの分かりやすく短いブログ記事（200文字程度）を作成・修正してください。\n\n【これまでの経緯】\n{history}")
response = llm.invoke([sys_msg, prompt])
return {"messages": [response]}
def reviewer_node(state: AgentState):
"""編集長（レビュアー）：ライターの記事を審査する"""
messages = state["messages"]
# 直前のメッセージ（ライターが書いたばかりの記事）を取得
last_article = messages[-1].content
sys_msg = SystemMessage(content="あなたは厳しい編集長です。")
# ライターの記事を「人間からの指示」としてLLMに渡して審査させる
prompt = HumanMessage(content=f"""以下のライターが書いた記事を審査してください。
- 200文字前後で小学生でも理解できるように書かれていれば『【合格】』というキーワードを含めて褒めてください。
- 説明が難しすぎたり、文字数が極端に合っていない場合は『【修正指示】』というキーワードを含めて、具体的な修正点を指摘してください。
【ライターの記事】
{last_article}""")
response = llm.invoke([sys_msg, prompt])
return {"messages": [response]}
修正したソースを動かしてみるとreviewerがレビューをし、その結果を受けて修正を行う様子が確認できました！
AIマルチエージェント（編集長チェック付き）を起動します...
=== Agent: researcher ===
# 量子コンピュータが暗号技術に与える影響
## 重要なポイント３つ
• **今の暗号が壊れる危険性**
量子コンピュータはとても強い力を持っているので、今のインターネット銀行やクレジットカードを守っている「暗号」を、簡単に解いてしまう可能性があります。それは、普通のコンピュータなら1000年かかる計算を、量子コンピュータなら数時間でできてしまうほど強いからです。
• **新しい暗号の開発が急務**
この危険から守るために、量子コンピュータでも解けない「新しい暗号」を作る研究が今、世界中で急いで進められています。これを「ポスト量子暗号」と呼びます。
• **社会全体の準備が必要**
銀行、病院、政府など、たくさんの場所が今の暗号を使っているので、新しい暗号に切り替えるには時間がかかります。だから今から準備を始めることが大切です。
**たとえるなら：** 今の暗号は「複雑な迷路」ですが、量子コンピュータは「空を飛ぶ力」を持っているので、迷路を飛び越えてしまう、ということです。
====================
=== Agent: writer ===
# 量子コンピュータと暗号の話
量子コンピュータは普通のコンピュータより**ものすごく速い**ため、今のインターネット銀行やクレジットカードを守っている暗号を簡単に解いてしまう可能性があります。普通なら1000年かかる計算が、数時間でできてしまうほどです。
そこで世界中の研究者が、量子コンピュータでも解けない「新しい暗号」を急いで開発しています。銀行や病院など多くの場所が新しい暗号に切り替える必要があるため、今から準備を始めることが大切です。
**簡単にいうと：** 現在の暗号は「複雑な迷路」ですが、量子コンピュータは「空を飛ぶ力」を持っているので、迷路を飛び越えてしまう、ということです。
---
**修正ポイント：**
- 3つのポイントを統合し、流れを整理
- より簡潔で読みやすい構成に
- 重要な情報を凝縮（約200文字）
====================
=== Agent: reviewer ===
【修正指示】
以下の点を修正してください：
**1. 文字数が多すぎます**
- 現在：約280文字
- 目標：約200文字
- 指示内容の「約200文字」を守ってください
**2. 段落構成を簡潔に**
- 現在は3段落に分かれていますが、冗長です
- 「1000年かかる計算が数時間で」という具体例は、小学生向けには不要な詳細情報です
- 「迷路と空を飛ぶ力」の比喩は良いので、これを活用して統合してください
**3. 修正案の方向性**
以下の要素を200文字前後に凝縮してください：
- 量子コンピュータは普通のコンピュータより速い
- 今の暗号が危ない理由
- 新しい暗号が必要という結論
- 迷路と空を飛ぶ力の比喩は活用可
**例：** 「量子コンピュータは普通のコンピュータより〇〇倍速いため、銀行やクレジットカードを守っている暗号を簡単に解いてしまう可能性があります。暗号は『複雑な迷路』ですが、量子コンピュータは『空を飛ぶ力』を持っているので飛び越えてしまうのです。そのため世界中で新しい暗号の開発が急がれています。」
**再提出をお願いします。**
====================
=== Agent: writer ===
# 量子コンピュータと暗号の話【修正版】
量子コンピュータは普通のコンピュータより圧倒的に速いため、銀行やクレジットカードを守っている暗号を簡単に解いてしまう可能性があります。暗号は「複雑な迷路」ですが、量子コンピュータは「空を飛ぶ力」を持っているので飛び越えてしまうのです。そのため世界中で、量子コンピュータでも解けない新しい暗号の開発が急がれています。
---
**修正内容：**
- 文字数：約200文字に削減
- 冗長な説明（「1000年かかる」など）を削除
- 3段落を1段落に統合
- 比喩を活かしながら、要点を明確化
- 流れ：問題提示→比喩説明→解決策、で簡潔に構成
====================
=== Agent: reviewer ===
【合格】
素晴らしい修正版です！この記事は合格水準に達しています。
**評価のポイント：**
✅ **文字数が適切**：約200文字で、指定範囲内に収まっています
✅ **小学生にも理解できる**：「複雑な迷路」「空を飛ぶ力」という比喩が効果的で、難しい概念をビジュアルで理解させる工夫が素晴らしい
✅ **構成が明確**：問題提示→比喩説明→解決策という流れが論理的で読みやすい
✅ **要点がシャープ**：冗長さを削除し、「量子コンピュータの脅威」と「新しい暗号開発」という核心だけを残した判断が的確
前版との比較でも、不要な説明を削ぎ落とし、比喩の力を最大限に活用する修正が見事です。このレベルの簡潔性と分かりやすさなら、実際の小学生向け記事として十分に通用します。
====================
まとめ
LangChainとLangGraphを用いたAIエージェントの実装を行いました。
実際に触ってみるまではこの二つにどのような違いがあるのか理解できていませんでしたが、エージェントそのものの開発とエージェント同士の振る舞いを制御について、明確に線引きされていることでそれぞれの開発に集中できるというメリットがあるように感じました。
今回はシンプルな条件分岐を実装するのみとなりましたが、ツールの組み込みや実行環境へのデプロイなどといった分野にも手を出してみたいと思います。