概要
AI Agent開発シリーズの最終章として、計算、検索、分析を自動実行し、プロフェッショナルなWebUIで操作できる本格システムを構築します。
従来のチャットボットの限界を超え、「考えて、ツールを使って、問題を解決する」真の自律型エージェントの実装方法を解説します。
この記事で実装するもの
- ✅ 高度な計算ツール（安全性考慮）
- ✅ Web検索ツール（SerpAPI + フォールバック）
- ✅ OpenAI Function Callingによるツール統合
- ✅ Streamlit製プロフェッショナルWebUI
🛠️ ツール統合エージェントの実装
🧮 高度な計算ツールの実装
セキュリティを重視した安全な計算ツールを実装します：
# tools/calculator.py
# import文は省略（完全版はpontanuki.comで公開）
class CalculatorInput(BaseModel):
"""計算ツールの入力スキーマ"""
expression: str = Field(
description="数学式（例：2 + 3 * 4, sin(30), sqrt(16), log(100)）"
)
class AdvancedCalculator(BaseTool):
"""高度な計算機能を提供するツール
基本的な四則演算から三角関数、対数、統計関数まで
幅広い数学的計算をサポートします。
セキュリティ対策として、安全な関数のみを許可します。
"""
name = "calculator"
description = """数学的計算を安全に実行します。
対応している計算：
- 基本演算: +, -, *, /, %, **（べき乗）
- 三角関数: sin, cos, tan（度数で入力）
- 逆三角関数: asin, acos, atan
- 対数: log（自然対数）, log10（常用対数）
- 平方根: sqrt
- 指数関数: exp
- 絶対値: abs
- 円周率: pi
- ネイピア数: e
使用例：
- 基本計算: "2 + 3 * 4"
- 三角関数: "sin(30)" （30度のサイン）
- 対数: "log10(100)"
- 平方根: "sqrt(16)"
- 複合計算: "sin(30) + cos(60)"
"""
args_schema: Type[BaseModel] = CalculatorInput
def __init__(self):
super().__init__()
# 安全な数学関数のマッピング
self.safe_functions = {
# 基本数学関数
'sin': lambda x: math.sin(math.radians(x)), # 度数入力
'cos': lambda x: math.cos(math.radians(x)),
'tan': lambda x: math.tan(math.radians(x)),
'asin': lambda x: math.degrees(math.asin(x)), # 度数出力
'acos': lambda x: math.degrees(math.acos(x)),
'atan': lambda x: math.degrees(math.atan(x)),
# 対数・指数関数
'sqrt': math.sqrt,
'log': math.log, # 自然対数
'log10': math.log10, # 常用対数
'log2': math.log2, # 二進対数
'exp': math.exp,
# その他の関数
'abs': abs,
'round': round,
'ceil': math.ceil,
'floor': math.floor,
'pow': pow,
# 定数
'pi': math.pi,
'e': math.e,
}
def _run(self, expression: str) -> str:
"""計算を安全に実行"""
try:
# 入力の前処理
expression = expression.strip()
# 危険な文字列をチェック
dangerous_patterns = [
'__', 'import', 'exec', 'eval', 'open', 'file',
'input', 'raw_input', 'compile', 'reload'
]
for pattern in dangerous_patterns:
if pattern in expression.lower():
return "エラー: 安全でない式が検出されました"
# 数学関数の名前空間を設定
namespace = {
"__builtins__": {}, # 組み込み関数を無効化
**self.safe_functions
}
# 安全な評価の実行
result = eval(expression, namespace)
# 結果の型チェックと整形
if isinstance(result, (int, float)):
# 無限大・NaNのチェック
if math.isinf(result):
return "エラー: 結果が無限大です"
elif math.isnan(result):
return "エラー: 結果が数値ではありません（NaN）"
# 整数の場合は小数点を表示しない
if isinstance(result, float) and result.is_integer():
return str(int(result))
elif isinstance(result, float):
# 小数点以下10桁まで表示、末尾のゼロは削除
return f"{result:.10f}".rstrip('0').rstrip('.')
else:
return str(result)
else:
return str(result)
except ZeroDivisionError:
return "エラー: ゼロ除算は実行できません"
except ValueError as e:
return f"エラー: 数学的エラー - {str(e)}"
except OverflowError:
return "エラー: 数値が大きすぎます"
except SyntaxError:
return "エラー: 数式の構文が正しくありません"
except NameError as e:
return f"エラー: 未定義の関数または変数 - {str(e)}"
except Exception as e:
return f"エラー: 計算できませんでした - {str(e)}"
async def _arun(self, expression: str) -> str:
"""非同期版計算実行"""
return self._run(expression)
🌐 Web検索ツールの実装
リアルタイム情報検索機能を提供するツールを実装します：
# tools/web_search.py
# import文は省略（完全版はpontanuki.comで公開）
class WebSearchInput(BaseModel):
"""Web検索ツールの入力スキーマ"""
query: str = Field(description="検索したいキーワードや質問")
num_results: int = Field(
default=5,
description="取得する検索結果数（1-10）",
ge=1,
le=10
)
class WebSearchTool(BaseTool):
"""Web検索機能を提供するツール
SerpAPIまたはフォールバック手法を使用して
インターネット上の最新情報を検索します。
"""
name = "web_search"
description = """インターネット上の最新情報を検索して取得します。
使用場面：
- 最新のニュースや情報
- 特定のトピックについての詳細情報
- 統計データや事実確認
- 技術的な情報やチュートリアル
使用例：
- "2024年 Python 最新トレンド"
- "ChatGPT 最新アップデート"
- "東京オリンピック 結果"
- "機械学習 入門 チュートリアル"
注意: 正確性は情報源に依存するため、重要な情報は複数のソースで確認することをお勧めします。
"""
args_schema: Type[BaseModel] = WebSearchInput
def __init__(self):
super().__init__()
self.serpapi_key = os.getenv("SERPAPI_API_KEY")
self.fallback_enabled = True
# リクエスト間の待機時間（レート制限対策）
self.request_delay = 1.0
self.last_request_time = 0
def _wait_for_rate_limit(self):
"""レート制限のための待機"""
current_time = time.time()
time_since_last = current_time - self.last_request_time
if time_since_last < self.request_delay:
time.sleep(self.request_delay - time_since_last)
self.last_request_time = time.time()
def _run(self, query: str, num_results: int = 5) -> str:
"""Web検索を実行"""
try:
# レート制限対策
self._wait_for_rate_limit()
if self.serpapi_key:
return self._search_with_serpapi(query, num_results)
elif self.fallback_enabled:
return self._search_with_fallback(query, num_results)
else:
return "エラー: 検索APIが設定されていません。SERPAPI_API_KEYを設定してください。"
except Exception as e:
return f"検索エラー: {str(e)}"
def _search_with_serpapi(self, query: str, num_results: int) -> str:
"""SerpAPIを使用した本格的な検索"""
try:
url = "https://serpapi.com/search.json"
params = {
"engine": "google",
"q": query,
"api_key": self.serpapi_key,
"num": min(num_results, 10),
"hl": "ja", # 日本語
"gl": "jp", # 日本の検索結果
"safe": "active" # セーフサーチ
}
response = requests.get(url, params=params, timeout=15)
response.raise_for_status()
data = response.json()
# エラーチェック
if "error" in data:
return f"検索APIエラー: {data['error']}"
organic_results = data.get("organic_results", [])
if not organic_results:
return f"「{query}」に関する検索結果が見つかりませんでした。"
# 結果の整形
results = []
for i, result in enumerate(organic_results[:num_results], 1):
title = result.get("title", "タイトルなし")
link = result.get("link", "")
snippet = result.get("snippet", "説明なし")
# 長すぎるスニペットは切り詰め
if len(snippet) > 200:
snippet = snippet[:200] + "..."
results.append(f"""
🔗 **{i}. {title}**
URL: {link}
📄 概要: {snippet}
""")
search_info = data.get("search_information", {})
total_results = search_info.get("total_results", "不明")
header = f"「{query}」の検索結果 (総件数: {total_results}):\n"
return header + "\n".join(results)
except requests.exceptions.Timeout:
return "検索がタイムアウトしました。しばらくしてから再試行してください。"
except requests.exceptions.RequestException as e:
return f"検索リクエストエラー: {str(e)}"
except Exception as e:
return f"SerpAPI検索エラー: {str(e)}"
🤖 ツール統合エージェントの実装
計算ツールと検索ツールを統合した高度なエージェントを作成します：
# agents/tool_enhanced_agent.py
# import文は省略（完全版はpontanuki.comで公開）
class ToolEnhancedAgent:
"""ツール統合エージェント
このクラスは複数のツールを統合し、問題解決に必要な
ツールを自動選択して実行する高度なエージェントです。
"""
def __init__(self, model_name: str = None, memory_window: int = 10):
"""ツール統合エージェントを初期化
Args:
model_name: 使用するOpenAIモデル名
memory_window: メモリに保持する会話数
"""
# 設定を取得
model_config = settings.get_model_config(model_name)
# LLMの初期化（ツール使用時は低Temperatureを推奨）
self.llm = ChatOpenAI(
model=model_config["model_name"],
temperature=0.1, # ツール使用時は一貫性を重視
api_key=model_config["api_key"]
)
# ツールの初期化
self.tools = self._initialize_tools()
# メモリの設定
self.memory = ConversationBufferWindowMemory(
k=memory_window,
return_messages=True,
memory_key="chat_history"
)
# システムプロンプトの定義
self.system_prompt = """あなたはツールを使いこなす高度なAIアシスタントです。
ユーザーの問題を解決するために、適切なツールを選択し、組み合わせて使用してください。
利用可能なツール：
1. calculator: 数学的計算（基本演算、三角関数、対数など）
2. web_search: インターネット検索（リアルタイム情報取得）
ツール使用のガイドライン：
- 計算が必要な場合は calculator を使用
- 最新情報や具体的な事実確認が必要な場合は web_search を使用
- 複数のツールが必要な場合は、論理的な順序で実行
- ツールの結果を基に、分かりやすい回答を提供
- ツールを使用しない場合でも、質の高い回答を心がける
重要な指針：
1. ユーザーの意図を正確に理解する
2. 必要に応じて複数のツールを組み合わせる
3. ツールの実行結果を適切に解釈・説明する
4. エラーが発生した場合は代替手段を提案する
5. 常に正確で有用な情報提供を目指す"""
# プロンプトテンプレートの作成
self.prompt = ChatPromptTemplate.from_messages([
SystemMessage(content=self.system_prompt),
MessagesPlaceholder(variable_name="chat_history"),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
# OpenAI Tools Agentの作成
self.agent = create_openai_tools_agent(
llm=self.llm,
tools=self.tools,
prompt=self.prompt
)
# AgentExecutorの作成（メモリ付き）
self.agent_executor = AgentExecutor(
agent=self.agent,
tools=self.tools,
memory=self.memory,
verbose=True, # デバッグ用
max_iterations=5, # 無限ループ防止
early_stopping_method="generate",
handle_parsing_errors=True
)
def _initialize_tools(self) -> List:
"""利用可能なツールを初期化"""
tools = []
# 計算ツール
calculator = AdvancedCalculator()
tools.append(calculator)
# Web検索ツール
web_search = WebSearchTool()
tools.append(web_search)
return tools
def chat_with_tools(self, user_input: str) -> str:
"""ツール統合チャット
Args:
user_input: ユーザーからの入力
Returns:
AI生成の回答（ツール実行結果を含む）
"""
try:
# エージェント実行
response = self.agent_executor.invoke({
"input": user_input,
"chat_history": self.memory.chat_memory.messages
})
return response["output"]
except Exception as e:
error_msg = f"申し訳ありません。処理中にエラーが発生しました: {str(e)}"
# エラーもメモリに記録
self.memory.chat_memory.add_user_message(user_input)
self.memory.chat_memory.add_ai_message(error_msg)
return error_msg
🎨 StreamlitによるWebUI実装
作成したエージェントを使いやすいWebインターフェースで操作できるようにしましょう：
# ui/streamlit_app.py
# import文は省略（完全版はpontanuki.comで公開）
class StreamlitChatApp:
"""Streamlit チャットアプリケーション"""
def __init__(self):
self.setup_page_config()
self.initialize_session_state()
def setup_page_config(self):
"""ページ設定"""
st.set_page_config(
page_title="AI Agent Chat Interface",
page_icon="🤖",
layout="wide",
initial_sidebar_state="expanded"
)
# カスタムCSS
st.markdown("""
<style>
.main-header {
font-size: 3rem;
color: #1f77b4;
text-align: center;
margin-bottom: 2rem;
text-shadow: 2px 2px 4px rgba(0,0,0,0.1);
}
.agent-info {
background: linear-gradient(90deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 1rem;
border-radius: 10px;
margin-bottom: 1rem;
}
.chat-message {
padding: 1rem;
border-radius: 10px;
margin: 0.5rem 0;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
.user-message {
background-color: #e3f2fd;
border-left: 4px solid #2196f3;
margin-left: 20px;
}
.ai-message {
background-color: #f3e5f5;
border-left: 4px solid #9c27b0;
margin-right: 20px;
}
</style>
""", unsafe_allow_html=True)
def initialize_session_state(self):
"""セッション状態の初期化"""
if "messages" not in st.session_state:
st.session_state.messages = []
if "agent_type" not in st.session_state:
st.session_state.agent_type = "basic"
if "agent" not in st.session_state:
st.session_state.agent = None
if "session_id" not in st.session_state:
st.session_state.session_id = f"streamlit_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
if "settings_validated" not in st.session_state:
st.session_state.settings_validated = False
def render_sidebar(self):
"""サイドバーの描画"""
with st.sidebar:
st.title("🤖 AI Agent 設定")
# エージェントタイプ選択
st.subheader("エージェントタイプ")
agent_type = st.selectbox(
"使用するエージェント",
["basic", "memory", "tool_enhanced"],
format_func=lambda x: {
"basic": "🤖 基本チャット",
"memory": "🧠 メモリ付きチャット",
"tool_enhanced": "🔧 ツール統合チャット"
}[x],
key="agent_selector"
)
# エージェントタイプが変更された場合
if agent_type != st.session_state.agent_type:
st.session_state.agent_type = agent_type
st.session_state.agent = None # エージェントをリセット
st.rerun()
# モデル設定
st.subheader("モデル設定")
model_name = st.selectbox(
"使用モデル",
["gpt-3.5-turbo", "gpt-4", "gpt-4-turbo-preview"],
index=0
)
temperature = st.slider(
"Temperature（創造性）",
min_value=0.0,
max_value=1.0,
value=0.7,
step=0.1
)
# メモリ設定（メモリエージェント用）
if agent_type in ["memory", "tool_enhanced"]:
st.subheader("メモリ設定")
memory_window = st.slider(
"メモリウィンドウ",
min_value=5,
max_value=50,
value=10,
help="保持する会話数"
)
else:
memory_window = 10
# セッション管理
st.subheader("セッション管理")
st.text(f"セッションID: {st.session_state.session_id}")
if st.button("🧹 チャット履歴をクリア"):
st.session_state.messages = []
if st.session_state.agent and hasattr(st.session_state.agent, 'clear_memory'):
st.session_state.agent.clear_memory()
st.rerun()
# ツール情報（ツール統合エージェント用）
if agent_type == "tool_enhanced" and st.session_state.agent:
st.subheader("🔧 利用可能なツール")
tools = st.session_state.agent.get_available_tools()
for tool in tools:
with st.expander(f"📋 {tool['name']}"):
st.write(tool['description'])
return {
"agent_type": agent_type,
"model_name": model_name,
"temperature": temperature,
"memory_window": memory_window
}
def render_chat_interface(self, config: Dict[str, Any]):
"""チャットインターフェースの描画"""
st.markdown('<h1 class="main-header">🤖 AI Agent Chat Interface</h1>',
unsafe_allow_html=True)
# エージェント情報表示
agent_info = {
"basic": "🤖 基本的な質問応答機能",
"memory": "🧠 会話履歴を記憶して継続的な対話が可能",
"tool_enhanced": "🔧 計算とWeb検索ツールを使用可能"
}
st.markdown(f'''
<div class="agent-info">
<h3>現在のエージェント</h3>
<p>{agent_info[config["agent_type"]]}</p>
</div>
''', unsafe_allow_html=True)
# チャット履歴表示
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.write(message["content"])
# チャット入力
if prompt := st.chat_input("メッセージを入力してください..."):
# ユーザーメッセージを追加
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.write(prompt)
# AI応答を生成
with st.chat_message("assistant"):
with st.spinner("考え中..."):
try:
if config["agent_type"] == "basic":
response = st.session_state.agent.chat(prompt)
elif config["agent_type"] == "memory":
response = st.session_state.agent.chat_with_memory(
prompt,
st.session_state.session_id
)
elif config["agent_type"] == "tool_enhanced":
response = st.session_state.agent.chat_with_tools(prompt)
st.write(response)
# レスポンスを履歴に追加
st.session_state.messages.append({
"role": "assistant",
"content": response
})
except Exception as e:
error_msg = f"エラーが発生しました: {str(e)}"
st.error(error_msg)
st.session_state.messages.append({
"role": "assistant",
"content": error_msg
})
def run(self):
"""アプリケーションの実行"""
# 環境設定の検証
if not self.validate_environment_ui():
st.stop()
# サイドバー設定
config = self.render_sidebar()
# エージェントの初期化
if not self.create_agent(config):
st.stop()
# メインのチャットインターフェース
self.render_chat_interface(config)
# サンプルプロンプト
self.render_sample_prompts()
# フッター
st.markdown("---")
st.markdown("**AI Agent Chat Interface** - Built with Streamlit & LangChain")
🚀 実行とテスト
実際の使用例
ツール統合エージェント:
あなた: 2の10乗を計算してください
AI: 計算ツールを使用して2の10乗を求めます。
> Entering new AgentExecutor chain...
I need to calculate 2 to the power of 10.
Action: calculator
Action Input: 2 ** 10
Observation: 1024
Thought: I now know the final answer.
Final Answer: 2の10乗は1024です。
あなた: ChatGPTの最新情報を調べて
AI: Web検索ツールを使用してChatGPTの最新情報を調べます。
> Entering new AgentExecutor chain...
I need to search for the latest information about ChatGPT.
Action: web_search
Action Input: ChatGPT 最新情報 2024
Observation: 「ChatGPT 最新情報 2024」の検索結果 (総件数: 約2,340,000件):
🔗 **1. ChatGPT最新アップデート - 2024年3月版**
URL: https://example.com/chatgpt-update-2024
📄 概要: 2024年3月にリリースされたChatGPTの新機能について...
Thought: I have found recent information about ChatGPT updates.
Final Answer: ChatGPTの最新情報をお調べしました。2024年3月に大きなアップデートがあり、新機能として...
🌐 WebUIでの操作
StreamlitアプリケーションをWebブラウザで開くと：
- エージェント選択: サイドバーで3種類のエージェントから選択
- モデル設定: GPTモデルとTemperatureの調整
- チャットインターフェース: 直感的な対話画面
- ツール情報: 利用可能なツールの詳細表示
- サンプルプロンプト: ワンクリックでテスト可能
📈 次のステップと応用
🔮 さらなる機能拡張
1. カスタムツールの追加
# tools/custom_tools.py
class WeatherTool(BaseTool):
name = "weather"
description = "天気情報を取得"
# 実装...
class EmailTool(BaseTool):
name = "email_sender"
description = "メール送信"
# 実装...
2. より高度なメモリ
# 長期記憶、ベクターサーチなど
from langchain.memory import ConversationSummaryBufferMemory
from langchain.vectorstores import Chroma
3. マルチモーダル対応
# 画像、音声の処理
from langchain.schema import HumanMessage, AIMessage
from langchain.schema.messages import ImageMessage
🏢 実用的なビジネス応用
1. カスタマーサポート
- FAQ検索機能
- チケット管理連携
- エスカレーション判定
2. データ分析アシスタント
- データベース連携
- グラフ生成
- レポート自動作成
3. コンテンツ生成
- SEO最適化
- 多言語対応
- ブランドガイドライン準拠
🎯 第3部のまとめ
📚 この記事で学んだこと
ツール開発面：
- ✅ 安全な計算ツールの実装
- ✅ Web検索APIの統合
- ✅ セキュリティを考慮したコード実行
エージェント統合面：
- ✅ OpenAI Function Callingの活用
- ✅ 複数ツールの動的選択
- ✅ エラーハンドリングと回復機能
WebUI開発面：
- ✅ Streamlitによる本格的なWebアプリ開発
- ✅ リアルタイム対話インターフェース
- ✅ プロダクションレディな機能実装
システム設計面：
- ✅ スケーラブルなアーキテクチャ
- ✅ 設定管理とデプロイ対応
- ✅ ユーザビリティとパフォーマンス最適化
🚀 シリーズ全体の成果物
AI Agent開発シリーズを通じて、以下が完成しました：
- 基本エージェント: 質問応答機能
- メモリエージェント: 会話履歴記憶機能
- ツール統合エージェント: 計算・検索機能
- WebUIシステム: プロフェッショナルなインターフェース
- データベース管理: 効率的なデータ永続化
- 設定管理システム: 環境変数の安全な管理
💡 重要なポイント
- 段階的な構築: 基礎→応用→実践の体系的アプローチ
- 実用性重視: 企業でも使える本格的な実装
- 拡張可能性: カスタマイズと機能追加の容易さ
- セキュリティ考慮: 安全性を重視した設計
このAI Agentシステムは、単なる学習教材ではなく、実際のビジネスで活用できる本格的なツールです。
🔗 完全なソースコードはpontanuki.comで公開中！
完全なソースコードと詳細な実装解説は、pontanuki.comで公開しています！
この記事では主要な実装部分のみを紹介しましたが、pontanuki.comでは以下も含めて公開中：
- 完全なimport文とセットアップ手順
- エラーハンドリングの詳細実装
- プロジェクト構造と設定ファイル
- テスト関数とデモコード
- requirements.txtと環境設定
- その他のユーティリティ関数
📖 AI Agent開発シリーズ（完結）
- 第1部: 基本概念と環境構築
- 第2部: メモリとデータ管理
- 第3部: ツール統合とWebUI ← この記事の完全版
AI Agentの世界はまだ始まったばかりです。このシステムを基盤に、さらに高度な自動化と効率化を実現していきましょう！