AIエージェントの実践的導入ガイド - 業務自動化の第一歩

pip install openai python-dotenv

import os
import json
import logging
from typing import List, Dict, Any, Optional
from openai import OpenAI

# ロギングの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class Tool:
    """エージェントが利用できるツールの基底クラス"""
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    def run(self, **kwargs) -> str:
        raise NotImplementedError

class CalculatorTool(Tool):
    """計算を行うツール"""
    def __init__(self):
        super().__init__(
            name="calculator",
            description="数値のリストを受け取り、平均値を計算します。引数として 'numbers': [list] を必要とします。"
        )

    def run(self, numbers: List[float]) -> str:
        try:
            if not numbers:
                return "エラー: 数値リストが空です。"
            avg = sum(numbers) / len(numbers)
            logger.info(f"計算実行: 入力={numbers}, 平均={avg}")
            return json.dumps({"average": avg})
        except Exception as e:
            logger.error(f"計算ツールエラー: {e}")
            return f"エラー: 計算中に問題が発生しました ({e})"

class DatabaseTool(Tool):
    """擬似的なデータベースからデータを取得するツール"""
    def __init__(self):
        super().__init__(
            name="database",
            description="特定のIDのデータをデータベースから取得します。引数として 'id': int を必要とします。"
        )
        # 擬似データ
        self.mock_data = {
            1: {"id": 1, "sales": [100, 200, 150]},
            2: {"id": 2, "sales": [5000, 6000, 5500]},
            3: {"id": 3, "sales": [10, 20, 30]}
        }

    def run(self, id: int) -> str:
        try:
            data = self.mock_data.get(id)
            if data:
                logger.info(f"DB取得: ID={id}, データ={data}")
                return json.dumps(data)
            else:
                logger.warning(f"DB取得失敗: ID={id} は見つかりません")
                return f"エラー: ID {id} のデータは見つかりませんでした。"
        except Exception as e:
            logger.error(f"DBツールエラー: {e}")
            return f"エラー: データ取得中に問題が発生しました ({e})"

class Agent:
    """ReActパターンを実装したシンプルなエージェント"""
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.tools: Dict[str, Tool] = {
            "calculator": CalculatorTool(),
            "database": DatabaseTool()
        }
        self.system_prompt = self._build_system_prompt()

    def _build_system_prompt(self) -> str:
        tool_descriptions = "\n".join([
            f"- {tool.name}: {tool.description}" 
            for tool in self.tools.values()
        ])
        return f"""
あなたは役立つAIアシスタントです。
利用可能なツール:
{tool_descriptions}

ユーザーの質問に対して、以下のJSON形式で思考と行動を出力してください。
{{
  "thought": "次に何をすべきかの思考",
  "action": "ツール名 または 'final_answer'",
  "action_input": {{ツールへの入力パラメータ}} または "最終的な回答文字列"
}}

ルール:
1. 必ずツールを使って情報を確認してから回答してください。
2. 最終回答が決まったら、actionを 'final_answer' にしてください。
3. action_inputは必ず有効なJSON形式、または文字列にしてください。
"""

    def _call_llm(self, messages: List[Dict[str, str]]) -> Dict[str, Any]:
        """LLMを呼び出し、レスポンスをパースする"""
        try:
            response = self.client.chat.completions.create(
                model="gpt-4o-mini", # コスト効率の良いモデルを選択
                messages=messages,
                temperature=0
            )
            content = response.choices[0].message.content
            logger.info(f"LLM応答: {content}")
            return json.loads(content)
        except json.JSONDecodeError:
            logger.error("LLMの応答をJSONとしてパースできませんでした")
            return {
                "thought": "応答の解析に失敗しました",
                "action": "final_answer",
                "action_input": "申し訳ありません。内部処理エラーが発生しました。"
            }
        except Exception as e:
            logger.error(f"LLM APIエラー: {e}")
            raise

    def run(self, user_query: str, max_steps: int = 5) -> str:
        """エージェントの実行ループ"""
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_query}
        ]

        for step in range(max_steps):
            logger.info(f"--- ステップ {step + 1} ---")
            
            # LLMによる思考と行動の決定
            llm_response = self._call_llm(messages)
            
            action = llm_response.get("action")
            action_input = llm_response.get("action_input")
            thought = llm_response.get("thought", "")

            # 最終回答の場合
            if action == "final_answer":
                logger.info("最終回答を生成しました。")
                return action_input

            # ツール実行の場合
            if action in self.tools:
                tool = self.tools[action]
                observation = tool.run(**action_input)
                
                # 観察結果を会話履歴に追加
                messages.append({
                    "role": "assistant",
                    "content": json.dumps(llm_response)
                })
                messages.append({
                    "role": "user",
                    "content": f"観察結果: {observation}"
                })
            else:
                # 不明なアクションの場合
                error_msg = f"エラー: 不明なアクション '{action}' です。利用可能なツールから選択してください。"
                logger.warning(error_msg)
                messages.append({
                    "role": "user",
                    "content": error_msg
                })

        return "ステップ数の上限に達しました。タスクを完了できませんでした。"

if __name__ == "__main__":
    # 実行例
    # 環境変数からAPIキーを取得、または直接入力（実際には.envファイル等を使用推奨）
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        print("エラー: OPENAI_API_KEY環境変数が設定されていません。")
        exit(1)

    agent = Agent(api_key=api_key)
    
    # 複雑なタスク: ID=2のデータを取得し、平均売上を計算する
    query = "IDが2のデータをデータベースから取得し、その売上平均を計算してください。"
    print(f"ユーザー: {query}")
    
    try:
        result = agent.run(query)
        print(f"エージェント: {result}")
    except Exception as e:
        print(f"実行中に例外が発生しました: {e}")