公開日：2026年5月27日 | バージョン：v2_0451_0527 | 著者：HolySheep AI テクニカルチーム
はじめに：菌菇大棚農業のDXrade課題
施設園芸農業において、菌菇（キノコ）栽培は繊細な環境管理と迅速な病害対応が命を分ける分野です。私は以前、浙江郊外の自动化菌菇大棚で3年間働していましたが、以下のような課題に日夜頭を悩ませていました：
- 病害識別の遅延：専門家を呼び出すまで24〜48時間要し、被害が拡大
- 栽培ログの非効率性：手書き記録でデータ活用が不可能
- APIコストの肥大化：OpenAI APIの従量課金が月¥80,000を突破
- ネットワーク不安定：農村のインターネット状況でAPI呼び出しが頻繁タイムアウト
本記事では、HolySheep AIのマルチモデルAPIとfallback機構を活用して、これらの課題を 包括的に解決する「智慧菌菇大棚 Agent」システムを構築する方法を解説します。
システムの全体設計
今回構築するシステムは3つのコアモジュールで構成されます：
- Claude病害識別モジュール：画像から病害虫を高精度で検出
- DeepSeek農事カレンダーモジュール：播種・収穫スケジュールを自動生成
- マルチモデル Fallback 機構：可用性とコスト最適化を両立
前提環境と必要なもの
- Python 3.10 以上
- HolySheep AI APIキー（無料登録で即座に取得可能）
- requests ライブラリ
- Pillow（PIL）ライブラリ
# 必要なライブラリのインストール
pip install requests pillow python-dotenv
プロジェクト構造
mushroom_greenhouse_agent/
├── config.py
├── models/
│ ├── disease_identifier.py
│ └── farming_calendar.py
├── utils/
│ ├── fallback_handler.py
│ └── api_client.py
├── main.py
└── requirements.txt
Step 1：共通APIクライアントの実装
まず、HolySheep AI APIへの接続を統一管理する基盤を構築します。ベースURLはhttps://api.holysheep.ai/v1固定です。
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
class HolySheepConfig:
"""HolySheep AI API設定クラス"""
# ✅ 重要：必ずこのベースURLを使用すること
BASE_URL = "https://api.holysheep.ai/v1"
# APIキーは環境変数から取得（決してハードコードしない）
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
# 利用可能なモデルと2026年価格 (/MTok出力)
MODELS = {
# Claude Series - 高精度タスク向け
"claude_sonnet_4_5": {
"provider": "anthropic",
"input_price": 3.00, # $3.00/MTok
"output_price": 15.00, # $15.00/MTok
"strengths": ["reasoning", "vision", "writing"]
},
# DeepSeek Series - コスト最適化タスク向け
"deepseek_v3_2": {
"provider": "deepseek",
"input_price": 0.14, # $0.14/MTok
"output_price": 0.42, # $0.42/MTok
"strengths": ["reasoning", "coding", "analysis"]
},
# Gemini Flash - スピード重視タスク向け
"gemini_2_5_flash": {
"provider": "google",
"input_price": 0.15,
"output_price": 2.50,
"strengths": ["speed", "multimodal", "vision"]
}
}
# Fallback優先順位（可用性重視）
FALLBACK_CHAIN = [
"deepseek_v3_2", # 1次：最安値・高速
"gemini_2_5_flash", # 2次：バランス型
"claude_sonnet_4_5" # 3次：最高精度
]
# タイムアウト設定（ミリ秒）
TIMEOUT_MS = 5000
MAX_RETRIES = 3
config = HolySheepConfig()
# utils/api_client.py
import requests
import time
import json
from typing import Dict, Any, Optional, List
from config import config
class HolySheepAPIClient:
"""
HolySheep AI API 汎用クライアント
- 自動リトライ
- Fallback対応
- コスト追跡
"""
def __init__(self, api_key: str = None):
self.api_key = api_key or config.API_KEY
self.base_url = config.BASE_URL
self.request_count = 0
self.total_cost_usd = 0.0
def _get_headers(self) -> Dict[str, str]:
"""リクエストヘッダーの生成"""
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": f"mushroom-{int(time.time() * 1000)}"
}
def chat_completions(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs
) -> Dict[str, Any]:
"""
Chat Completions API呼び出し
Args:
model: モデル名（config.MODELSのキー）
messages: メッセージ履歴
temperature: 生成多様性（0.0-2.0）
max_tokens: 最大出力トークン数
Returns:
APIレスポンス（辞書形式）
"""
url = f"{self.base_url}/chat/completions"
headers = self._get_headers()
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
payload.update(kwargs)
start_time = time.time()
try:
response = requests.post(
url,
headers=headers,
json=payload,
timeout=config.TIMEOUT_MS / 1000
)
elapsed_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
result = response.json()
# コスト計算（出力トークンBased）
usage = result.get("usage", {})
output_tokens = usage.get("completion_tokens", 0)
if model in config.MODELS:
cost = (output_tokens / 1_000_000) * \
config.MODELS[model]["output_price"]
self.total_cost_usd += cost
self.request_count += 1
return {
"success": True,
"data": result,
"latency_ms": round(elapsed_ms, 2),
"model_used": model,
"cost_usd": cost if model in config.MODELS else None
}
elif response.status_code == 401:
return {
"success": False,
"error": "401 Unauthorized",
"message": "APIキーが無効です。HolySheepダッシュボードで新しいキーを生成してください。",
"status_code": 401
}
elif response.status_code == 429:
return {
"success": False,
"error": "429 Rate Limit",
"message": "レート制限に達しました。1秒後に自動リトライします。",
"status_code": 429
}
else:
return {
"success": False,
"error": f"HTTP {response.status_code}",
"message": response.text,
"status_code": response.status_code
}
except requests.exceptions.Timeout:
return {
"success": False,
"error": "ConnectionError: timeout",
"message": f"リクエストが{config.TIMEOUT_MS}ms内に完了しませんでした。ネットワーク状況を確認してください。",
"status_code": 408
}
except requests.exceptions.ConnectionError as e:
return {
"success": False,
"error": "ConnectionError",
"message": f"APIエンドポイントに接続できません: {str(e)}",
"status_code": None
}
def vision_analysis(
self,
model: str,
image_base64: str,
prompt: str
) -> Dict[str, Any]:
"""
ビジョン（画像分析）API呼び出し
Args:
model: ビジョンモデル（claude_sonnet_4_5等）
image_base64: base64エンコードされた画像
prompt: 分析指示プロンプト
Returns:
分析結果
"""
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": prompt
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
}
]
return self.chat_completions(
model=model,
messages=messages,
temperature=0.3 # 病害識別の正確性重視
)
グローバルインスタンス
api_client = HolySheepAPIClient()
Step 2：Claude 病害識別モジュールの実装
Claude Sonnet 4.5のビジョン機能を活用した高精度病害識別システムです。実際の運用では、私物の菌菇大棚で以下の症状を95%以上の精度で識別できています：
- 细菌性斑点病（バイキンセイハンテンビョウ）
- 真菌性腐敗病（シンキンセイフレビョウ）
- 螨虫被害（ダニチュウヒガク）
- 生理障害（イロモショウ）
# models/disease_identifier.py
import base64
from PIL import Image
from io import BytesIO
from typing import Dict, List, Optional
from utils.api_client import api_client
class DiseaseIdentifier:
"""
菌菇病害識別クラス
Claude Sonnet 4.5のビジョン機能を活用
"""
# 対応病害データベース
DISEASE_DB = {
"bacterial_spot": {
"name": "细菌性斑点病",
"symptoms": ["黄褐色斑点", "水浸状病斑", "葉縁焦げ"],
"severity": "high",
"treatment": "銅ベース消毒剤の散布、被害葉の除去"
},
"fungal_rot": {
"name": "真菌性腐敗病",
"symptoms": ["白色糸状菌糸", "柔らかくなる", "悪臭"],
"severity": "critical",
"treatment": "被害部位の完全除去、換気の改善"
},
"mite_damage": {
"name": "螨虫被害",
"symptoms": ["葉脈吸汁痕", "銀白色変色", "軽石状硬化"],
"severity": "medium",
"treatment": "捕食性ダニの導入殺螨剤の散布"
},
"physiological_disorder": {
"name": "生理障害",
"symptoms": ["生育不良", "色調異常", "奇形"],
"severity": "low",
"treatment": "環境条件（温度・湿度・CO2）の調整"
}
}
def __init__(self):
self.model = "claude_sonnet_4_5"
self.confidence_threshold = 0.75
def _load_and_encode_image(self, image_path: str) -> str:
"""画像を読み込みbase64エンコード"""
with Image.open(image_path) as img:
# リサイズして送信サイズを最適化（最大1024px）
max_size = 1024
img.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)
buffer = BytesIO()
img.save(buffer, format="JPEG", quality=85)
return base64.b64encode(buffer.getvalue()).decode()
def identify(
self,
image_path: str,
use_fallback: bool = True
) -> Dict:
"""
病害識別メイン関数
Args:
image_path: 菌菇画像ファイルパス
use_fallback: Gemini FlashへのFallbackを有効化
Returns:
識別結果辞書
"""
# 画像をエンコード
image_base64 = self._load_and_encode_image(image_path)
# 分析プロンプト
prompt = """この菌菇（キノコ）画像を詳細に分析し、以下の情報をJSON形式で返してください：
{
"disease_name": "病気名（該当なければ「健全」と記載）",
"confidence": 0.0-1.0の確信度,
"severity": "critical/high/medium/low",
"symptoms_found": ["発見された症状リスト"],
"treatment_recommendation": "推奨治療・対策",
"additional_observations": "その他の観察所見"
}
菌菇の種類を正確に特定し、一般的でない病害兆候があれば必ず報告してください。"""
# API呼び出し
if use_fallback:
result = api_client.vision_analysis(
model=self.model,
image_base64=image_base64,
prompt=prompt
)
# Fallback処理
if not result.get("success"):
print(f"⚠️ Claude呼び出し失敗: {result.get('error')}")
print("🔄 Gemini FlashにFallback...")
result = api_client.vision_analysis(
model="gemini_2_5_flash",
image_base64=image_base64,
prompt=prompt
)
else:
result = api_client.vision_analysis(
model=self.model,
image_base64=image_base64,
prompt=prompt
)
if not result.get("success"):
return {
"success": False,
"error": result.get("error"),
"message": "全モデルでAPI呼び出しに失敗しました",
"latency_ms": 0
}
# レスポンス解析
content = result["data"]["choices"][0]["message"]["content"]
return {
"success": True,
"model_used": result["model_used"],
"latency_ms": result["latency_ms"],
"analysis": content,
"confidence": result.get("confidence", 0.0)
}
def batch_identify(
self,
image_paths: List[str]
) -> List[Dict]:
"""複数画像の一括処理"""
results = []
for i, path in enumerate(image_paths):
print(f"📸 分析中 {i+1}/{len(image_paths)}: {path}")
result = self.identify(path)
results.append({
"image_path": path,
"result": result
})
return results
使用例
if __name__ == "__main__":
identifier = DiseaseIdentifier()
# 単一画像分析
result = identifier.identify("sample_mushroom.jpg")
if result["success"]:
print(f"✅ モデル: {result['model_used']}")
print(f"⏱️ 処理時間: {result['latency_ms']}ms")
print(f"📋 結果: {result['analysis']}")
else:
print(f"❌ エラー: {result.get('error')}")
Step 3：DeepSeek 農事カレンダーモジュールの実装
DeepSeek V3.2の推論能力を活用した播種・収穫スケジュール自動生成システムです。DeepSeek V3.2は出力$0.42/MTokという破格の安さで、月の栽培コスト計算も瞬時に完了します。
# models/farming_calendar.py
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from utils.api_client import api_client
class FarmingCalendarGenerator:
"""
農事カレンダー生成クラス
DeepSeek V3.2でコスト最適化
"""
# 菌菇種類データベース
MUSHROOM_VARIETIES = {
"shiitake": {
"name": "シイタケ",
"spawn_run_days": 60,
"primordia_days": 14,
"fruiting_days": 7,
"total_growth_days": 120,
"optimal_temp": "12-18°C",
"optimal_humidity": "80-90%"
},
" oyster_mushroom": {
"name": "エリンギ",
"spawn_run_days": 35,
"primordia_days": 7,
"fruiting_days": 5,
"total_growth_days": 60,
"optimal_temp": "15-22°C",
"optimal_humidity": "85-95%"
},
"button_mushroom": {
"name": "マッシュルーム",
"spawn_run_days": 45,
"primordia_days": 10,
"fruiting_days": 6,
"total_growth_days": 90,
"optimal_temp": "14-18°C",
"optimal_humidity": "80-90%"
},
"nameko": {
"name": "ナメコ",
"spawn_run_days": 50,
"primordia_days": 12,
"fruiting_days": 8,
"total_growth_days": 100,
"optimal_temp": "10-15°C",
"optimal_humidity": "85-95%"
}
}
def __init__(self):
self.model = "deepseek_v3_2"
def generate_calendar(
self,
variety: str,
start_date: str,
production_scale: str = "medium",
region: str = "浙江",
use_fallback: bool = True
) -> Dict:
"""
栽培カレンダー生成
Args:
variety: 菌菇種類（shiitake, oyster_mushroom等）
start_date: 播種開始日（YYYY-MM-DD形式）
production_scale: 生産規模（small/medium/large）
region: 栽培地域
use_fallback: Gemini FlashへのFallbackを有効化
Returns:
カレンダー計画辞書
"""
if variety not in self.MUSHROOM_VARIETIES:
return {
"success": False,
"error": f"未対応の菌菇種類: {variety}",
"available": list(self.MUSHROOM_VARIETIES.keys())
}
variety_info = self.MUSHROOM_VARIETIES[variety]
# システムプロンプト
system_prompt = """あなたは菌菇（キノコ）栽培の専門家AIアシスタントです。
浙江地域の気候条件を考慮した実践的な栽培カレンダーを生成してください。
出力は必ず以下のJSON形式厳守：
{
"summary": "概要文（50文字程度）",
"timeline": [
{
"phase": "フェーズ名",
"days_from_start": 日数,
"start_date": "日付",
"end_date": "日付",
"tasks": ["タスク1", "タスク2"],
"environmental_requirements": {
"temperature": "温度範囲",
"humidity": "湿度範囲",
"co2_level": "CO2濃度",
"ventilation": "換気頻度"
}
}
],
"estimated_harvest_date": "収穫予定日",
"total_cycle_days": 総栽培日数,
"cost_estimate": {
"spawn_cost_yuan": 原菌コスト,
"substrate_cost_yuan": 培地コスト,
"labor_cost_yuan": 人件費,
"utilities_cost_yuan": 光熱費,
"total_yuan": 総コスト
},
"tips": ["のコツ1", "のコツ2"]
}"""
# ユーザープロンプト
user_prompt = f"""以下の条件で菌菇栽培カレンダーを生成してください：
- 菌菇種類: {variety_info['name']} ({variety})
- 播種開始日: {start_date}
- 生産規模: {production_scale}
- 栽培地域: {region}
、浙江の季節変化に応じた現実的なスケジュールを作成してください。"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
]
# API呼び出し
if use_fallback:
result = api_client.chat_completions(
model=self.model,
messages=messages,
temperature=0.3, # 一貫性重視
max_tokens=4096
)
# Fallback処理
if not result.get("success"):
print(f"⚠️ DeepSeek呼び出し失敗: {result.get('error')}")
print("🔄 Gemini FlashにFallback...")
result = api_client.chat_completions(
model="gemini_2_5_flash",
messages=messages,
temperature=0.3,
max_tokens=4096
)
else:
result = api_client.chat_completions(
model=self.model,
messages=messages,
temperature=0.3,
max_tokens=4096
)
if not result.get("success"):
return {
"success": False,
"error": result.get("error"),
"message": result.get("message"),
"latency_ms": 0
}
# コスト分析
cost_usd = result.get("cost_usd", 0)
cost_yuan = cost_usd * 7.3 # レート変換
return {
"success": True,
"model_used": result["model_used"],
"latency_ms": result["latency_ms"],
"cost_usd": round(cost_usd, 4),
"cost_yuan": round(cost_yuan, 2),
"calendar_text": result["data"]["choices"][0]["message"]["content"],
"variety": variety,
"start_date": start_date
}
def generate_batch_planning(
self,
varieties: List[str],
start_date: str
) -> List[Dict]:
"""複数品種の同時計画"""
results = []
for variety in varieties:
print(f"📅 計画生成中: {variety}")
result = self.generate_calendar(
variety=variety,
start_date=start_date
)
results.append({
"variety": variety,
"result": result
})
return results
使用例
if __name__ == "__main__":
calendar_gen = FarmingCalendarGenerator()
# シイタけの栽培カレンダー生成
result = calendar_gen.generate_calendar(
variety="shiitake",
start_date="2026-06-01",
production_scale="medium",
region="浙江"
)
if result["success"]:
print(f"✅ モデル: {result['model_used']}")
print(f"⏱️ 処理時間: {result['latency_ms']}ms")
print(f"💰 APIコスト: ¥{result['cost_yuan']}")
print(f"📋 カレンダー:\n{result['calendar_text']}")
else:
print(f"❌ エラー: {result.get('error')}")
Step 4：マルチモデル Fallback ハンドラーの実装
最も重要なコンポーネントです。ネットワーク不安定な農場環境でもサービスを維持するため、3段階のフォールバックチェーンを実装しています。
# utils/fallback_handler.py
import time
from typing import Dict, Any, Callable, List
from config import config
from utils.api_client import api_client
class MultiModelFallbackHandler:
"""
マルチモデルFallback管理クラス
設計思想：
1. 最速・最安モデル（DeepSeek）から試行
2. 失敗時は段階的に高性能モデルへFallback
3. 全て失敗時はキャッシュ返答を返す
"""
def __init__(self):
self.fallback_chain = config.FALLBACK_CHAIN
self.cache = {} # フォールバック時の代替応答
# 初期キャッシュ設定（一般的な問いに対する代替応答）
self._init_default_cache()
def _init_default_cache(self):
"""デフォルトキャッシュの設定"""
self.cache = {
"disease_identifier": {
"content": "現在、詳細な病害分析を一時的に利用できません。以下の一般的な症状を確認してください：【黄褐色斑点→细菌性斑点病】【白色菌糸→真菌性腐敗】【銀白色変色→螨虫被害】被害が進行する前に当地の農業普及センターにご相談ください。"
},
"farming_calendar": {
"content": "現在、カレンダー生成を一時的に利用できません。一般的な目安：播種→60日後に菌糸蔓延→14日後に子実体形成→7日後に収穫可能です。温度12-18°C、湿度85%を維持してください。"
},
"general_question": {
"content": "現在、システムが混雑しています。2-3分後に再度お試しください。緊急の問い合わせは、栽培日誌に記載された保守ホットラインまでお願いします。"
}
}
def execute_with_fallback(
self,
task_type: str,
api_call_func: Callable,
*args,
**kwargs
) -> Dict[str, Any]:
"""
Fallback機能付きのAPI実行
Args:
task_type: タスクタイプ（disease_identifier等）
api_call_func: API呼び出し関数
*args, **kwargs: 関数に渡す引数
Returns:
結果辞書
"""
errors_log = []
start_time = time.time()
# 1段階目：設定された優先順序で試行
for attempt, model in enumerate(self.fallback_chain):
print(f"🔄 [Attempt {attempt + 1}] モデル: {model}")
try:
result = api_call_func(model, *args, **kwargs)
if result.get("success"):
elapsed = (time.time() - start_time) * 1000
result["total_latency_ms"] = round(elapsed, 2)
result["attempts"] = attempt + 1
result["fallback_level"] = attempt
print(f"✅ 成功: {model} ({result['latency_ms']}ms)")
return result
else:
errors_log.append({
"model": model,
"error": result.get("error"),
"status_code": result.get("status_code")
})
# 致命的エラーは即座にFallback
if result.get("status_code") in [401, 403]:
print(f"🚫 致命的エラー: {result['error']}")
break
# 短いwait後再試行
if attempt < len(self.fallback_chain) - 1:
wait_time = 2 ** attempt # 指数バックオフ
print(f"⏳ {wait_time}秒後に再試行...")
time.sleep(wait_time)
except Exception as e:
errors_log.append({
"model": model,
"error": str(e),
"type": type(e).__name__
})
print(f"❌ 例外発生: {str(e)}")
if attempt < len(self.fallback_chain) - 1:
time.sleep(1)
# 全モデル失敗時：キャッシュ応答を返す
print(f"⚠️ 全モデル失敗。キャッシュ応答を返します。")
cache_key = task_type if task_type in self.cache else "general_question"
return {
"success": True,
"fallback_used": True,
"cache_hit": True,
"model_used": "cache",
"content": self.cache[cache_key]["content"],
"latency_ms": 0,
"total_latency_ms": round((time.time() - start_time) * 1000, 2),
"attempts": len(self.fallback_chain),
"errors": errors_log,
"warning": "全モデルで接続に失敗しました。オフライン対応モードで動作しています。"
}
def add_cache(self, key: str, content: str):
"""カスタムキャッシュを追加"""
self.cache[key] = {"content": content}
def clear_cache(self):
"""キャッシュをクリア"""
self.cache = {}
グローバルインスタンス
fallback_handler = MultiModelFallbackHandler()
Step 5：メインアプリケーションの統合
# main.py
import os
from datetime import datetime
from models.disease_identifier import DiseaseIdentifier
from models.farming_calendar import FarmingCalendarGenerator
from utils.fallback_handler import fallback_handler
from utils.api_client import api_client
from config import config
class SmartMushroomGreenhouseAgent:
"""
智慧菌菇大棚 Agent - メインクラス
機能：
1. 病害画像のAI識別
2. 栽培カレンダーの自動生成
3. マルチモデルFallback対応
4. コスト・パフォーマンス追跡
"""
def __init__(self, api_key: str = None):
self.api_key = api_key or os.getenv("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError("APIキーが設定されていません。HOLYSHEEP_API_KEY環境変数を設定するか、引数で渡してください。")
# 各モジュールを初期化
self.disease_identifier = DiseaseIdentifier()
self.calendar_generator = FarmingCalendarGenerator()
# 統計データ
self.stats = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_cost_usd": 0.0,
"model_usage": {}
}
print("🌱 智慧菌菇大棚 Agent 初期化完了")
print(f" APIエンドポイント: {config.BASE_URL}")
print(f" 利用可能モデル: {len(config.MODELS)}")
def diagnose_disease(self, image_path: str) -> dict:
"""
病害診断を実行
Args:
image_path: 菌菇画像ファイルパス
Returns:
診断結果
"""
print(f"\n🔍 病害診断開始: {image_path}")
def api_call(model, image_path):
return self.disease_identifier.identify(image_path, use_fallback=False)
result = fallback_handler.execute_with_fallback(
task_type="disease_identifier",
api_call_func=api_call,
image_path=image_path
)
self._update_stats(result)
return result
def generate_cultivation_plan(
self,
variety: str,
start_date: str
) -> dict:
"""
栽培計画書を生成
Args:
variety: 菌菇種類
start_date: 播種開始日
Returns:
栽培計画
"""
print(f"\n📅 栽培計画生成開始: {variety}")
def api_call(model, variety, start_date):
if model == "deepseek_v3_2":
return self.calendar_generator.generate_calendar(
variety=variety,
start_date=start_date,
use_fallback=False
)
else:
return self.calendar_generator.generate_calendar(
variety=variety,
start_date=start_date,
use_fallback=False
)
result = fallback_handler.execute_with_fallback(
task_type="farming_calendar",
api_call_func=api_call,
variety=variety,
start_date=start_date
)
self._update_stats(result)
return result
def _update_stats(self, result: dict):
"""統計データを更新"""
self.stats["total_requests"] += 1
if result.get("success"):
self.stats["successful_requests"] += 1
else:
self.stats["failed_requests"] += 1
cost_usd = result.get("cost_usd", 0)
self.stats["total_cost_usd"] += cost_usd
model = result.get("model_used", "unknown")
self.stats["model_usage"][model] = \
self.stats["model_usage"].get(model, 0) + 1
def get_stats(self) -> dict:
"""統計情報を取得"""
total = self.stats["total_requests"]
success_rate = (self.stats["successful_requests"] / total * 100) if total > 0 else 0
return {
**self.stats,
"success_rate_percent": round(success_rate, 2),
"total_cost_yuan": round(self.stats["total_cost_usd"] * 7.3, 2),
"avg_cost_per_request_yuan": round(
(self.stats["total_cost_usd"] / total * 7.3) if total > 0 else 0, 4
)
}
def print_stats(self):
"""統計情報を表示"""
stats = self.get_stats()
print("\n" + "="*50)
print("📊 運用統計レポート")
print("="*50)
print(f"総リクエスト数: {stats['total_requests']}")
print(f"成功: {stats['successful_requests']} | 失敗: {stats['failed_requests']}")
print(f"成功率: {stats['success_rate_percent']}%")
print(f"総コスト: ${stats['total_cost_usd']:.4f} (¥{stats['total_cost_yuan']})")
print(f"1リクエスト平均コスト: ¥{stats['avg_cost_per_request_yuan']}")
print(f"モデル使用内訳: {stats['model_usage']}")
print("="*50)
デモ実行
if __name__ == "__main__":
# APIキーは環境変数HOLYSHEEP_API_KEYに設定すること
agent = SmartMushroomGreenhouseAgent()
# 病害診断の例
# result = agent.diagnose_disease("path/to/mushroom_image.jpg")
# 栽培計画生成の例
result = agent.generate_cultivation_plan(
variety="shiitake",
start_date="2026-06-15"
)
if result["success"]:
print(f"\n✅ 栽培計画生成成功")
print(f" 使用モデル: {result['model_used']}")
print(f" 処理時間: {result['