実践LESSON 04
この教材で学ぶこと Tool Use・RAG・エージェントをAPIで実装します。Claude API 活用 — Tool Use・RAG・エージェント構築
参照コース: https://anthropic.skilljar.com/claude-with-the-anthropic-api
前提知識: API呼び出しとデータモデルの基礎。Python + Anthropic SDK で実装します。
1. Tool Use（Claude に関数を呼ばせる）
Tool Use は Kotlin の interface + DI に近い考え方：
Claude がどの関数を呼ぶか決め → アプリ側が実行 → 結果を返す
import anthropic
import json
import subprocess
client = anthropic.Anthropic()
# ── ツール定義（JSON Schema）────────────────────────────
tools = [
{
"name": "execute_bash",
"description": "Linux コマンドを実行して結果を返す（読み取り専用コマンドのみ）",
"input_schema": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "実行する bash コマンド"
}
},
"required": ["command"]
}
},
{
"name": "get_aws_resource",
"description": "AWS リソースの情報を取得する",
"input_schema": {
"type": "object",
"properties": {
"service": {
"type": "string",
"enum": ["ec2", "ecs", "rds", "s3"],
"description": "AWSサービス名"
},
"region": {
"type": "string",
"default": "ap-northeast-1"
}
},
"required": ["service"]
}
}
]
# ── ツール実行関数 ────────────────────────────────────────
SAFE_COMMANDS = {"df", "free", "ps", "top", "ls", "cat", "grep", "awk", "sed", "curl"}
def execute_tool(name: str, inputs: dict) -> str:
if name == "execute_bash":
cmd = inputs["command"]
base_cmd = cmd.split()[0]
if base_cmd not in SAFE_COMMANDS:
return f"Error: '{base_cmd}' は許可されていないコマンドです"
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True, timeout=10
)
return result.stdout or result.stderr
elif name == "get_aws_resource":
import boto3
service = inputs["service"]
region = inputs.get("region", "ap-northeast-1")
# ... boto3 で取得
return json.dumps({"service": service, "region": region, "status": "ok"})
return f"Unknown tool: {name}"
# ── エージェントループ ────────────────────────────────────
def run_agent(user_message: str) -> str:
messages = [{"role": "user", "content": user_message}]
while True:
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=4096,
tools=tools,
messages=messages
)
# 通常の回答で終了
if response.stop_reason == "end_turn":
return next(
b.text for b in response.content if hasattr(b, "text")
)
# ツール呼び出しがある場合
if response.stop_reason == "tool_use":
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f" 🔧 Tool: {block.name}({block.input})")
result = execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
messages.append({"role": "user", "content": tool_results})
# 使用例
answer = run_agent("本番ECSサービスの状態を確認して、問題があれば教えてください")
print(answer)2. RAG（Retrieval-Augmented Generation）
用途: 社内ドキュメント・手順書を Claude に検索させる
from anthropic import Anthropic
import numpy as np
client = Anthropic()
# ── テキストをベクトル化 ─────────────────────────────────
def get_embedding(text: str) -> list[float]:
"""テキストを埋め込みベクトルに変換（OpenAI or Cohere の埋め込みを使用）"""
# 実際は OpenAI text-embedding-3-small や Cohere Embed を使う
# ここでは概念を示す疑似コード
...
# ── ドキュメントの準備 ───────────────────────────────────
documents = [
{"id": "1", "text": "RDSの接続が切れた場合は、まずセキュリティグループを確認してください..."},
{"id": "2", "text": "ECSタスクがOOMKillされる場合、タスク定義のメモリ制限を確認してください..."},
{"id": "3", "text": "ALBの503エラーが出る場合、ターゲットグループのヘルスチェックを確認してください..."},
]
# ── コサイン類似度で検索 ─────────────────────────────────
def cosine_similarity(a: list, b: list) -> float:
a, b = np.array(a), np.array(b)
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
def search_documents(query: str, top_k: int = 3) -> list[dict]:
query_embedding = get_embedding(query)
scored = []
for doc in documents:
doc_embedding = get_embedding(doc["text"])
score = cosine_similarity(query_embedding, doc_embedding)
scored.append({"doc": doc, "score": score})
return [x["doc"] for x in sorted(scored, key=lambda x: x["score"], reverse=True)[:top_k]]
# ── RAG で回答 ───────────────────────────────────────────
def answer_with_rag(question: str) -> str:
relevant_docs = search_documents(question)
context = "\n\n".join(f"【文書{i+1}】\n{d['text']}" for i, d in enumerate(relevant_docs))
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=1024,
system="社内のインフラ運用ドキュメントを参照して回答するアシスタントです。",
messages=[{
"role": "user",
"content": f"""以下の社内ドキュメントを参照して質問に答えてください。
{context}
質問: {question}
ドキュメントに記載がない場合はその旨を伝えてください。"""
}]
)
return response.content[0].text
result = answer_with_rag("ECSタスクが起動しない場合の対処法は？")3. 構造化出力（Pydantic + Tool Use）
from pydantic import BaseModel
from typing import Literal
import json
class InfraIssue(BaseModel):
severity: Literal["critical", "high", "medium", "low"]
component: str
description: str
immediate_action: str
root_cause: str
prevention: list[str]
def analyze_incident(log_text: str) -> InfraIssue:
"""インシデントログを分析して構造化データを返す"""
# ツールとして定義することで確実にJSON形式で受け取れる
tools = [{
"name": "report_issue",
"description": "インシデント分析結果を報告する",
"input_schema": InfraIssue.model_json_schema()
}]
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=1024,
tools=tools,
tool_choice={"type": "tool", "name": "report_issue"}, # 強制的にツール使用
messages=[{
"role": "user",
"content": f"以下のインシデントログを分析してください:\n{log_text}"
}]
)
tool_use = next(b for b in response.content if b.type == "tool_use")
return InfraIssue(**tool_use.input)
issue = analyze_incident("[CRITICAL] DB connection pool exhausted. 500 errors spike.")
print(f"重大度: {issue.severity}")
print(f"即時対応: {issue.immediate_action}")4. プロンプトキャッシュ — 大きなコンテキストを繰り返す場合
# 大きなシステムプロンプトやドキュメントを何度も使う場合
# 最大 90% のコスト削減（キャッシュヒット時は入力コストが 10% に）
INFRA_RUNBOOK = open("runbook.md").read() # 数万トークンの運用手順書
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=1024,
system=[
{
"type": "text",
"text": f"あなたはインフラ運用アシスタントです。\n\n運用手順書:\n{INFRA_RUNBOOK}",
"cache_control": {"type": "ephemeral"} # ← このブロックをキャッシュ
}
],
messages=[{"role": "user", "content": "RDSの接続エラーが出ています"}]
)
# 2回目以降: キャッシュヒット → 大幅コスト削減5. Batch API — 大量リクエストの非同期処理
# 1000件のログを一括解析（通常より50%安い）
import anthropic
client = anthropic.Anthropic()
requests = [
{
"custom_id": f"log-{i}",
"params": {
"model": "claude-haiku-4-5",
"max_tokens": 256,
"messages": [{"role": "user", "content": f"このログの深刻度を答えてください: {log}"}]
}
}
for i, log in enumerate(log_lines)
]
# バッチ作成（非同期で処理される）
batch = client.messages.batches.create(requests=requests)
print(f"Batch ID: {batch.id}")
# 後で結果を取得
for result in client.messages.batches.results(batch.id):
print(f"{result.custom_id}: {result.result.message.content[0].text}")🔗 次のステップ
- ../../04_AI/MCP/01_MCP入門.md — API の Tool Use を MCP に統合
- ../../01_Programming/Python/01_Python実践_Kotlinエンジニア向け.md — Python 実装力を上げる