検索だけのRAGはもう古い？Agentic RAGで解決する複雑な推論タスク

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

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

class AgenticRAG:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.tools = self._define_tools()
        
    def _define_tools(self) -> List[Dict[str, Any]]:
        """エージェントが使用可能なツールの定義"""
        return [
            {
                "type": "function",
                "function": {
                    "name": "search_internal_knowledge_base",
                    "description": "社内の技術ドキュメントやマニュアルを検索します。製品仕様や内部手順に関する質問に使用してください。",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "検索キーワード"
                            }
                        },
                        "required": ["query"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "search_web",
                    "description": "インターネット上の最新情報を検索します。市場動向や最新のニュース、外部の技術情報に使用してください。",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "検索キーワード"
                            }
                        },
                        "required": ["query"]
                    }
                }
            }
        ]

    def _execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str:
        """ツールの実行ロジック（モック実装）"""
        logger.info(f"ツール実行中: {tool_name} 引数: {arguments}")
        
        try:
            if tool_name == "search_internal_knowledge_base":
                # 実際にはベクトルDB等へのクエリが入ります
                query = arguments.get("query", "")
                if "API" in query:
                    return "社内APIドキュメントによると、エンドポイントは /api/v1/resource で、認証にはBearerトークンが必要です。"
                else:
                    return "該当する社内ドキュメントは見つかりませんでした。"
            
            elif tool_name == "search_web":
                # 実際にはGoogle Search APIやBing API等が入ります
                query = arguments.get("query", "")
                if "Python" in query:
                    return "Python 3.12の最新情報では、パフォーマンスの向上とエラーメッセージの改善が図られています。"
                else:
                    return "ウェブ上に関連する最新情報は見つかりませんでした。"
            
            else:
                return f"未知のツールが呼び出されました: {tool_name}"
                
        except Exception as e:
            logger.error(f"ツール実行エラー: {e}")
            return f"エラーが発生しました: {str(e)}"

    def run(self, user_query: str, max_turns: int = 5) -> str:
        """Agentic RAGのメインループ"""
        messages = [
            {"role": "system", "content": "あなたは有用なAIアシスタントです。利用可能なツールを使ってユーザーの質問に答えてください。"},
            {"role": "user", "content": user_query}
        ]

        for turn in range(max_turns):
            logger.info(f"--- ターン {turn + 1} ---")
            
            try:
                # LLMによる応答生成（ツール呼び出し判断含む）
                response = self.client.chat.completions.create(
                    model="gpt-4o", # または利用可能なモデル
                    messages=messages,
                    tools=self.tools,
                    tool_choice="auto" # モデルにツールを使うか自動判断させる
                )
                
                response_message = response.choices[0].message
                tool_calls = response_message.tool_calls

                # ツール呼び出しがなければ最終回答とみなす
                if not tool_calls:
                    logger.info("最終回答が生成されました。")
                    return response_message.content
                
                # ツール呼び出しの結果を処理
                messages.append(response_message) # モデルのツール呼び出し要求を履歴に追加
                
                for tool_call in tool_calls:
                    function_name = tool_call.function.name
                    function_args = json.loads(tool_call.function.arguments)
                    
                    # ツール実行
                    function_response = self._execute_tool(function_name, function_args)
                    
                    # 実行結果を履歴に追加
                    messages.append(
                        {
                            "tool_call_id": tool_call.id,
                            "role": "tool",
                            "name": function_name,
                            "content": function_response,
                        }
                    )
            
            except Exception as e:
                logger.error(f"LLM推論中に例外発生: {e}")
                messages.append({
                    "role": "system", 
                    "content": f"前回の処理でエラーが発生しました。エラー内容: {str(e)}。別のアプローチを試してください。"
                })

        return "最大ターン数に達しましたが、回答を導き出せませんでした。"

# 実行例
if __name__ == "__main__":
    # 環境変数等からAPIキーを取得してください
    import os
    api_key = os.getenv("OPENAI_API_KEY")
    
    if not api_key:
        print("OPENAI_API_KEYが設定されていません。")
    else:
        agent = AgenticRAG(api_key=api_key)
        
        # 複雑な質問：社内情報と外部情報の両方が必要なケース
        query = "自社のAPI仕様を確認した上で、Pythonの最新バージョンでそれを実装する際の注意点を教えてください。"
        answer = agent.run(query)
        print(f"\n最終回答:\n{answer}")