AIエージェント入門：自律的なAIシステムの設計と実装
AIエージェントは、2024年から2025年にかけて最も注目を集めているAI技術の一つです。OpenAIのGPTs、AnthropicのClaude、GoogleのGeminiなど、主要なLLMプロバイダーがエージェント機能を強化し、企業での実用化も急速に進んでいます。
本記事では、LangChain v1.x 系をベースに、AIエージェントの基本概念から実装まで解説します。記事を読み終える頃には、自分でAIエージェントを設計・実装できるようになることを目指します。
AIエージェントとは
AIエージェントは、LLM（大規模言語モデル）を「頭脳」として、外部ツールを「手足」として使い、自律的にタスクを実行するシステムです。従来のチャットボットが「質問に答える」だけだったのに対し、AIエージェントは「目標を達成するために自ら考え、行動する」ことができます。
従来のLLMとの違い
従来のLLMとAIエージェントの違いを、具体例で見てみましょう。
質問: 「東京の明日の天気に合わせた服装を提案して」
従来のLLM:
- 学習データに基づいた一般的な回答を返す
- 「明日の天気」をリアルタイムで取得できない
- 結果: 「東京の天気は季節によって異なりますが...」という曖昧な回答
AIエージェント:
- 「天気を調べる必要がある」と判断（推論）
- 天気APIを呼び出して明日の東京の天気を取得（ツール使用）
- 取得した天気情報（例: 最高気温15℃、曇り時々雨）を分析
- 具体的な服装を提案（最終回答）
この違いを表にまとめると以下のようになります。
| 特性 | 従来のLLM | AIエージェント | |------|----------|--------------| | 応答形式 | 単発の回答 | 複数ステップの実行 | | ツール使用 | なし | Web検索、API呼び出し、コード実行など | | 計画能力 | 限定的 | タスク分解と計画立案 | | 自己修正 | なし | エラーからの回復・再試行 | | リアルタイム情報 | 取得不可 | ツール経由で取得可能 |
エージェントの基本アーキテクチャ
AIエージェントは、以下の4つのコンポーネントで構成されます。
┌─────────────────────────────────────────────────────────┐
│ AIエージェント │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ LLM │ │ メモリ │ │
│ │ (推論エンジン) │◄────►│ (コンテキスト) │ │
│ │ │ │ │ │
│ │ ・思考の生成 │ │ ・短期メモリ │ │
│ │ ・次の行動決定 │ │ ・長期メモリ │ │
│ │ ・結果の解釈 │ │ ・作業メモリ │ │
│ └────────┬────────┘ └─────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────┐ │
│ │ ツールキット │ │
│ │ │ │
│ │ [Web検索] [計算機] [コード実行] [API] │ │
│ │ [ファイル操作] [データベース] [外部サービス] │ │
│ └─────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
開発環境のセットアップ
実際にAIエージェントを実装するための環境を構築しましょう。
LangChain v1.x系を使用します。また、より高度な制御が必要な場合は langgraph も使用しますが、今回は基礎として langchain のコア機能を中心に解説します。
必要なもの
- Python 3.10以上
- OpenAI APIキーまたはAnthropic APIキー
- pip（Pythonパッケージマネージャー）
環境構築
まず、プロジェクトディレクトリを作成し、仮想環境をセットアップします。
# プロジェクトディレクトリの作成
mkdir ai-agent-tutorial
cd ai-agent-tutorial
# 仮想環境の作成と有効化
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 必要なパッケージのインストール
# LangChain v1.x系と、最新のエージェント構築ライブラリLangGraphをインストール
pip install langchain langchain-openai langchain-anthropic langgraph
pip install python-dotenv requests
APIキーの設定
プロジェクトルートに .env ファイルを作成し、APIキーを設定します。
# .env
OPENAI_API_KEY=sk-your-openai-api-key
ANTHROPIC_API_KEY=sk-ant-your-anthropic-api-key
Pythonコードで環境変数を読み込むには以下のようにします。
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
最初のエージェントを作る：ReActパターン
AIエージェントの最も基本的なパターンが「ReAct（Reasoning + Acting）」です。このパターンでは、エージェントが以下のサイクルを繰り返します。
思考 (Thought) → 行動 (Action) → 観察 (Observation) → 繰り返し...
【コラム】AgentExecutor vs LangGraph
LangChain v1.x時代において、エージェントを構築する方法は主に2つあります。
- AgentExecutor: 従来からある高レベルAPI。手軽にエージェントを作れますが、複雑なループや条件分岐のカスタマイズは難しいです。
- LangGraph: v0.1以降で導入され、v1.xで標準となったグラフベースの構築手法。ステートフルで複雑なフロー制御が可能ですが、記述量は増えます。
本記事では、概念の理解を優先し、コードがシンプルになる AgentExecutor （create_react_agent）を使用して解説します。実務で複雑なエージェントを開発する際は、LangGraph への移行をお勧めします。
ステップ1: シンプルなツールの定義
まず、エージェントが使用するツールを定義します。ここでは、計算機と現在時刻取得の2つのツールを作成します。
@tool デコレータを使用することで、簡単に関数をツール化できます。
# tools.py
from datetime import datetime
from langchain_core.tools import tool
@tool
def calculator(expression: str) -> str:
"""
数学的な計算を実行します。
四則演算や累乗などの計算式を文字列で受け取り、結果を返します。
Args:
expression: 計算式（例: "2 + 3 * 4", "10 ** 2"）
Returns:
計算結果の文字列
"""
try:
# 安全な評価のため、許可する文字を制限
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return "エラー: 無効な文字が含まれています"
result = eval(expression)
return f"計算結果: {result}"
except Exception as e:
return f"計算エラー: {str(e)}"
@tool
def get_current_time() -> str:
"""
現在の日時を取得します。
引数は不要です。
Returns:
現在の日時（日本時間）
"""
now = datetime.now()
return f"現在時刻: {now.strftime('%Y年%m月%d日 %H時%M分%S秒')}"
ステップ2: エージェントの作成
LangChainを使ってReActエージェントを作成します。
# agent.py
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.prompts import PromptTemplate
from tools import calculator, get_current_time
# LLMの初期化（OpenAIを使用する場合）
llm = ChatOpenAI(
model="gpt-4o",
temperature=0, # 決定的な出力のため0に設定
)
# 使用するツールのリスト
tools = [calculator, get_current_time]
# ReActプロンプトテンプレート
# v1.xではHubからプロンプトを取得することも一般的ですが、ここでは明示的に定義します
react_prompt = PromptTemplate.from_template("""
あなたは与えられたツールを使って質問に答えるAIアシスタントです。
利用可能なツール:
{tools}
ツールの使い方:
- ツールを使う場合は、以下の形式で出力してください
- 最終的な回答が分かったら、Final Answerで回答してください
質問に答えるために、以下の形式を使ってください:
Question: 答えるべき質問
Thought: 何をすべきか考える
Action: 使用するツール名（{tool_names}のいずれか）
Action Input: ツールへの入力
Observation: ツールの実行結果
... (Thought/Action/Action Input/Observationを必要なだけ繰り返す)
Thought: 最終的な答えが分かった
Final Answer: 質問への最終的な回答
それでは始めましょう！
Question: {input}
{agent_scratchpad}
""")
# エージェントの作成
agent = create_react_agent(llm, tools, react_prompt)
# AgentExecutorでラップ（実行環境を提供）
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 実行過程を表示
handle_parsing_errors=True, # パースエラーを処理
max_iterations=10, # 最大反復回数
)
# エージェントの実行
if __name__ == "__main__":
# テスト実行
result = agent_executor.invoke({
"input": "今何時ですか？また、123 × 456 の計算結果を教えてください。"
})
print("\n=== 最終回答 ===")
print(result["output"])
ステップ3: 実行と動作確認
上記のコードを実行すると、以下のような出力が得られます。
> Entering new AgentExecutor chain...
Thought: ユーザーは現在時刻と計算結果の2つを求めています。まず現在時刻を取得します。
Action: get_current_time
Action Input:
Observation: 現在時刻: 2025年1月20日 14時30分25秒
Thought: 現在時刻が分かりました。次に計算を実行します。
Action: calculator
Action Input: 123 * 456
Observation: 計算結果: 56088
Thought: 両方の情報が得られました。回答をまとめます。
Final Answer: 現在時刻は2025年1月20日 14時30分25秒です。また、123 × 456 の計算結果は 56,088 です。
> Finished chain.
=== 最終回答 ===
現在時刻は2025年1月20日 14時30分25秒です。また、123 × 456 の計算結果は 56,088 です。
実践的なツールの実装
より実用的なエージェントを作るために、Web検索やAPI呼び出しなどのツールを追加していきましょう。
Web検索ツール
DuckDuckGoの検索APIを使った無料のWeb検索ツールを実装します。
# tools/web_search.py
import requests
from langchain_core.tools import tool
@tool
def web_search(query: str) -> str:
"""
DuckDuckGoを使ってWeb検索を実行します。
Args:
query: 検索クエリ
Returns:
検索結果の要約（上位5件）
"""
try:
# DuckDuckGo Instant Answer API
url = "https://api.duckduckgo.com/"
params = {
"q": query,
"format": "json",
"no_html": 1,
"skip_disambig": 1,
}
response = requests.get(url, params=params, timeout=10)
data = response.json()
results = []
# Abstract（概要）があれば追加
if data.get("Abstract"):
results.append(f"概要: {data['Abstract']}")
# RelatedTopics（関連トピック）を追加
for topic in data.get("RelatedTopics", [])[:5]:
if isinstance(topic, dict) and topic.get("Text"):
results.append(f"- {topic['Text']}")
if results:
return "\n".join(results)
else:
return "検索結果が見つかりませんでした。"
except requests.Timeout:
return "検索がタイムアウトしました。"
except Exception as e:
return f"検索エラー: {str(e)}"
天気取得ツール
OpenWeatherMap APIを使った天気取得ツールです（無料プランで利用可能）。
# tools/weather.py
import os
import requests
from langchain_core.tools import tool
OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY")
@tool
def get_weather(city: str) -> str:
"""
指定した都市の現在の天気を取得します。
Args:
city: 都市名（英語、例: "Tokyo", "Osaka"）
Returns:
天気情報（気温、天気、湿度など）
"""
if not OPENWEATHER_API_KEY:
return "エラー: OpenWeatherMap APIキーが設定されていません"
try:
url = "https://api.openweathermap.org/data/2.5/weather"
params = {
"q": city,
"appid": OPENWEATHER_API_KEY,
"units": "metric", # 摂氏
"lang": "ja", # 日本語
}
response = requests.get(url, params=params, timeout=10)
if response.status_code == 404:
return f"エラー: 都市 '{city}' が見つかりません"
response.raise_for_status()
data = response.json()
weather_info = f"""
【{data['name']}の天気】
- 天気: {data['weather'][0]['description']}
- 気温: {data['main']['temp']}°C
- 体感温度: {data['main']['feels_like']}°C
- 湿度: {data['main']['humidity']}%
- 風速: {data['wind']['speed']} m/s
"""
return weather_info.strip()
except requests.RequestException as e:
return f"天気の取得に失敗しました: {str(e)}"
Pythonコード実行ツール
安全にPythonコードを実行するツールです。本番環境ではサンドボックス化が必須ですが、学習用に簡易版を実装します。
# tools/python_executor.py
import sys
from io import StringIO
from langchain_core.tools import tool
@tool
def execute_python(code: str) -> str:
"""
Pythonコードを実行して結果を返します。
Args:
code: 実行するPythonコード
Returns:
実行結果または標準出力
注意: このツールは学習用です。本番環境では適切なサンドボックスを使用してください。
"""
# 危険なモジュールのインポートを禁止
forbidden = ["os", "subprocess", "sys", "shutil", "pathlib"]
for module in forbidden:
if f"import {module}" in code or f"from {module}" in code:
return f"セキュリティエラー: {module}のインポートは禁止されています"
# 標準出力をキャプチャ
old_stdout = sys.stdout
sys.stdout = StringIO()
try:
# グローバル変数を制限して実行
exec_globals = {"__builtins__": {
"print": print,
"len": len,
"range": range,
"sum": sum,
"min": min,
"max": max,
"sorted": sorted,
"list": list,
"dict": dict,
"str": str,
"int": int,
"float": float,
"bool": bool,
"abs": abs,
"round": round,
}}
exec(code, exec_globals)
output = sys.stdout.getvalue()
if output:
return f"実行結果:\n{output}"
else:
return "コードは正常に実行されました（出力なし）"
except Exception as e:
return f"実行エラー: {type(e).__name__}: {str(e)}"
finally:
sys.stdout = old_stdout
メモリの実装
エージェントに「記憶」を持たせることで、会話の文脈を理解したり、過去の経験を活用したりできるようになります。
会話メモリ（短期メモリ）
直近の会話履歴を保持するメモリです。
# memory/conversation_memory.py
from langchain.memory import ConversationBufferWindowMemory
# 直近5回の会話を記憶
memory = ConversationBufferWindowMemory(
k=5,
memory_key="chat_history",
return_messages=True,
)
# 使用例
memory.save_context(
{"input": "私の名前は田中です"},
{"output": "田中さん、よろしくお願いします！"}
)
memory.save_context(
{"input": "好きな食べ物は寿司です"},
{"output": "寿司が好きなんですね！おすすめのお店はありますか？"}
)
# 記憶の確認
print(memory.load_memory_variables({}))
サマリーメモリ（長期メモリ）
長い会話を要約して保持するメモリです。トークン数を節約しながら重要な情報を保持できます。
# memory/summary_memory.py
from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 2000トークンを超えたら要約
memory = ConversationSummaryBufferMemory(
llm=llm,
max_token_limit=2000,
memory_key="chat_history",
return_messages=True,
)
エージェントにメモリを追加
# agent_with_memory.py
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain.memory import ConversationBufferWindowMemory
from langchain_core.prompts import PromptTemplate
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# メモリの初期化
memory = ConversationBufferWindowMemory(
k=10,
memory_key="chat_history",
return_messages=True,
)
# メモリを含むプロンプト
prompt_with_memory = PromptTemplate.from_template("""
あなたは会話の文脈を理解するAIアシスタントです。
これまでの会話:
{chat_history}
利用可能なツール:
{tools}
以下の形式で回答してください:
Question: {input}
Thought: 何をすべきか考える（過去の会話も参考に）
Action: ツール名
Action Input: ツールへの入力
Observation: 結果
... (繰り返し)
Final Answer: 最終回答
{agent_scratchpad}
""")
# エージェントの作成
agent = create_react_agent(llm, tools, prompt_with_memory)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory, # メモリを追加
verbose=True,
)
高度なパターン：Plan-and-Execute
複雑なタスクを処理するための「Plan-and-Execute」パターンを実装します。このパターンでは、まずタスクを細分化する「計画」フェーズと、各サブタスクを実行する「実行」フェーズに分けます。
# plan_and_execute.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# 計画のスキーマ定義（Pydantic v2準拠）
class Step(BaseModel):
"""タスクの1ステップ"""
step_number: int = Field(description="ステップ番号")
description: str = Field(description="このステップで行うこと")
tools_needed: List[str] = Field(description="必要なツール")
class Plan(BaseModel):
"""タスク全体の計画"""
goal: str = Field(description="最終目標")
steps: List[Step] = Field(description="実行ステップのリスト")
# 計画生成プロンプト
planning_prompt = ChatPromptTemplate.from_messages([
("system", """あなたはタスクプランナーです。
与えられたタスクを達成するための詳細な計画を立ててください。
利用可能なツール:
- calculator: 数学計算
- web_search: Web検索
- get_weather: 天気取得
- execute_python: Pythonコード実行
計画は具体的で実行可能なステップに分解してください。"""),
("human", "タスク: {task}"),
])
# 構造化出力で計画を生成（with_structured_outputはv0.1.14以降推奨）
planner = planning_prompt | llm.with_structured_output(Plan)
def plan_and_execute(task: str):
"""タスクを計画し、各ステップを実行する"""
# 1. 計画フェーズ
print("=== 計画フェーズ ===")
plan = planner.invoke({"task": task})
print(f"目標: {plan.goal}")
print("\n計画:")
for step in plan.steps:
print(f" {step.step_number}. {step.description}")
print(f" 使用ツール: {', '.join(step.tools_needed)}")
# 2. 実行フェーズ
print("\n=== 実行フェーズ ===")
results = []
for step in plan.steps:
print(f"\nステップ {step.step_number} を実行中...")
# ここで各ステップをエージェントで実行
result = agent_executor.invoke({"input": step.description})
results.append({
"step": step.step_number,
"description": step.description,
"result": result["output"]
})
print(f"結果: {result['output']}")
return results
# 使用例
if __name__ == "__main__":
task = "東京と大阪の今日の天気を比較して、どちらが外出に適しているか教えてください"
results = plan_and_execute(task)
RAG（検索拡張生成）との統合
エージェントにRAGを組み込むことで、社内ドキュメントやナレッジベースを活用した回答が可能になります。
ベクトルストアのセットアップ
# rag/vector_store.py
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
# サンプルドキュメント
documents = [
Document(
page_content="""
当社の勤怠システムの使い方：
1. 出勤時は「出勤」ボタンを押してください
2. 退勤時は「退勤」ボタンを押してください
3. 休暇申請は「休暇申請」メニューから行えます
4. 申請は上長の承認が必要です
""",
metadata={"source": "勤怠マニュアル", "category": "人事"}
),
Document(
page_content="""
経費精算の手順：
1. 経費精算システムにログイン
2. 「新規申請」をクリック
3. 経費の種類を選択（交通費、接待費、消耗品など）
4. 金額と日付を入力
5. 領収書の画像をアップロード
6. 申請ボタンを押して完了
""",
metadata={"source": "経費精算マニュアル", "category": "経理"}
),
]
# テキストを分割
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
)
splits = text_splitter.split_documents(documents)
# ベクトルストアの作成
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
documents=splits,
embedding=embeddings,
persist_directory="./chroma_db" # 永続化
)
# 検索用リトリーバー
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3} # 上位3件を取得
)
RAGツールの作成
# rag/rag_tool.py
from langchain_core.tools import tool
from rag.vector_store import retriever
@tool
def search_knowledge_base(query: str) -> str:
"""
社内ナレッジベースを検索します。
勤怠、経費精算、その他社内手続きに関する情報を検索できます。
Args:
query: 検索クエリ
Returns:
関連するドキュメントの内容
"""
docs = retriever.invoke(query)
if not docs:
return "関連する情報が見つかりませんでした。"
results = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "不明")
results.append(f"【{i}. {source}】\n{doc.page_content}")
return "\n\n".join(results)
エラーハンドリングとデバッグ
本番環境で動作するエージェントには、適切なエラーハンドリングが不可欠です。
カスタムエラーハンドラー
# utils/error_handler.py
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AgentErrorHandler(BaseCallbackHandler):
"""エージェントのエラーを処理するコールバック"""
def on_tool_error(self, error: Exception, **kwargs: Any) -> None:
"""ツールエラー時の処理"""
logger.error(f"ツールエラー: {str(error)}")
# Slackやメールで通知するなどの処理を追加可能
def on_agent_action(self, action: Any, **kwargs: Any) -> None:
"""エージェントのアクションをログ"""
logger.info(f"アクション: {action.tool} - 入力: {action.tool_input}")
def on_agent_finish(self, finish: Any, **kwargs: Any) -> None:
"""エージェント完了時のログ"""
logger.info(f"完了: {finish.return_values}")
# 使用方法
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
callbacks=[AgentErrorHandler()],
handle_parsing_errors=True,
max_iterations=10,
)
リトライ機構
# utils/retry.py
import time
from functools import wraps
def retry_on_error(max_retries: int = 3, delay: float = 1.0):
"""エラー時にリトライするデコレーター"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_error = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_error = e
if attempt < max_retries - 1:
print(f"エラー発生、リトライします ({attempt + 1}/{max_retries})")
time.sleep(delay * (attempt + 1)) # 指数バックオフ
raise last_error
return wrapper
return decorator
# 使用例
@retry_on_error(max_retries=3, delay=1.0)
def run_agent(query: str):
return agent_executor.invoke({"input": query})
本番環境でのベストプラクティス
1. レート制限の実装
# utils/rate_limiter.py
import time
from collections import deque
from threading import Lock
class RateLimiter:
"""APIコールのレート制限"""
def __init__(self, max_calls: int, period: float):
self.max_calls = max_calls
self.period = period
self.calls = deque()
self.lock = Lock()
def acquire(self):
with self.lock:
now = time.time()
# 古いコール記録を削除
while self.calls and self.calls[0] < now - self.period:
self.calls.popleft()
# 制限に達している場合は待機
if len(self.calls) >= self.max_calls:
sleep_time = self.calls[0] - (now - self.period)
time.sleep(sleep_time)
self.calls.append(time.time())
# 1分間に60回までに制限
rate_limiter = RateLimiter(max_calls=60, period=60.0)
2. コスト管理
# utils/cost_tracker.py
from langchain_community.callbacks import get_openai_callback
def run_with_cost_tracking(agent_executor, query: str):
"""コストを追跡しながらエージェントを実行"""
with get_openai_callback() as cb:
result = agent_executor.invoke({"input": query})
print(f"\n--- コスト情報 ---")
print(f"合計トークン: {cb.total_tokens}")
print(f"プロンプトトークン: {cb.prompt_tokens}")
print(f"完了トークン: {cb.completion_tokens}")
print(f"合計コスト: ${cb.total_cost:.4f}")
return result
3. セキュリティ考慮事項
# security/input_validator.py
import re
from typing import Optional
class InputValidator:
"""ユーザー入力のバリデーション"""
# 禁止パターン（SQLインジェクション、プロンプトインジェクションなど）
FORBIDDEN_PATTERNS = [
r"(?i)ignore\s+previous\s+instructions",
r"(?i)system\s*:\s*",
r"(?i)drop\s+table",
r"(?i)delete\s+from",
]
@classmethod
def validate(cls, input_text: str) -> tuple[bool, Optional[str]]:
"""入力をバリデート"""
for pattern in cls.FORBIDDEN_PATTERNS:
if re.search(pattern, input_text):
return False, "不正な入力が検出されました"
# 長さチェック
if len(input_text) > 10000:
return False, "入力が長すぎます（最大10000文字）"
return True, None
# 使用例
def safe_run_agent(query: str):
is_valid, error_msg = InputValidator.validate(query)
if not is_valid:
return {"error": error_msg}
return agent_executor.invoke({"input": query})
まとめ
本記事では、AIエージェントの基本概念から実装まで、実際に動作するコードを交えて解説しました。
学んだこと
- AIエージェントの基本構造: LLM、メモリ、ツールの3要素
- ReActパターン: 思考→行動→観察のサイクル
- 実践的なツール実装: 計算、Web検索、天気取得、コード実行
- メモリの種類: 短期メモリ、長期メモリ、サマリーメモリ
- Plan-and-Execute: 複雑なタスクの分解と実行
- RAGとの統合: ナレッジベースの活用
- 本番運用のベストプラクティス: エラーハンドリング、レート制限、セキュリティ
次のステップ
- 本記事のコードを実際に動かしてみる
- 自分のユースケースに合わせたカスタムツールを作成
- LangGraph を学んで、より複雑でステートフルなエージェントを構築する
- ストリーミング応答の実装
- 評価・モニタリングの仕組み構築
AIエージェントは急速に進化している分野です。LangChainやLangGraphなどのフレームワークも頻繁にアップデートされるため、公式ドキュメントを定期的にチェックすることをおすすめします。
この記事が、皆さんのAIエージェント開発の第一歩となれば幸いです。