Agentic RAG - 自律型AIエージェントによる高度な情報検索

質問 → ベクトル検索 → 文書取得 → LLM生成 → 回答

質問 → エージェント判断
       ↓
    クエリ拡張・情報源選択
       ↓
    並列検索（データベース + Web + ナレッジグラフ）
       ↓
    情報統合・推論
       ↓
    不足あれば再検索（反復）
       ↓
    高精度な回答生成

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class AgenticRAGState(TypedDict):
    question: str
    search_queries: List[str]
    documents: List[str]
    answer: str
    needs_more_info: bool

# プランナー
def planner_node(state: AgenticRAGState):
    # 質問を分析し、検索クエリを生成
    queries = llm.invoke(f"""
質問を分析し、必要な検索クエリを3つ生成してください：
質問: {state['question']}

検索クエリ:
""")
    return {"search_queries": queries.split("\n")}

# リトリーバー
def retriever_node(state: AgenticRAGState):
    documents = []
    for query in state["search_queries"]:
        # ベクトル検索
        vector_docs = vector_store.similarity_search(query, k=3)
        # Web検索
        web_docs = web_search_tool(query)
        # ナレッジグラフ検索
        kg_docs = knowledge_graph.query(query)
        
        documents.extend(vector_docs + web_docs + kg_docs)
    
    return {"documents": documents}

# シンセサイザー
def synthesizer_node(state: AgenticRAGState):
    context = "\n\n".join(state["documents"])
    answer = llm.invoke(f"""
以下の文書を参照して質問に答えてください：

質問: {state['question']}

参照文書:
{context}

回答:
""")
    return {"answer": answer}

# リフレクション
def reflection_node(state: AgenticRAGState):
    evaluation = llm.invoke(f"""
質問: {state['question']}
回答: {state['answer']}

この回答は質問に十分答えていますか？(yes/no)
""")
    
    needs_more = "no" in evaluation.lower()
    return {"needs_more_info": needs_more}

# 条件分岐
def should_continue(state: AgenticRAGState):
    if state.get("needs_more_info", False):
        return "planner"  # 再検索
    return "end"

# ワークフロー定義
workflow = StateGraph(AgenticRAGState)

workflow.add_node("planner", planner_node)
workflow.add_node("retriever", retriever_node)
workflow.add_node("synthesizer", synthesizer_node)
workflow.add_node("reflection", reflection_node)

# フロー定義
workflow.add_edge("planner", "retriever")
workflow.add_edge("retriever", "synthesizer")
workflow.add_edge("synthesizer", "reflection")
workflow.add_conditional_edges(
    "reflection",
    should_continue,
    {"planner": "planner", "end": END}
)

workflow.set_entry_point("planner")
app = workflow.compile()

# 複雑な質問に対応
result = app.invoke({
    "question": "2023年から2025年にかけて、AI業界でどのようなパラダイムシフトが起こりましたか？また、それがビジネスに与えた影響を、具体的な企業事例とともに説明してください。"
})

print(result["answer"])

def hybrid_retriever_node(state: AgenticRAGState):
    documents = []
    
    for query in state["search_queries"]:
        # 1. ベクトルRAG（セマンティック類似度）
        vector_docs = vector_store.similarity_search(query, k=5)
        
        # 2. GraphRAG（エンティティ関係）
        entities = extract_entities(query)
        graph_docs = knowledge_graph.traverse(
            entities,
            max_depth=2,
            relationship_types=["RELATED_TO", "CAUSED_BY"]
        )
        
        # 3. Web検索（最新情報）
        web_docs = web_search_tool(query, time_range="last_month")
        
        # 重要度でスコアリング
        scored_docs = score_documents(
            vector_docs + graph_docs + web_docs,
            query
        )
        
        documents.extend(scored_docs[:10])
    
    return {"documents": documents}

query = """
テスラの2024年のバッテリー技術革新が、
電気自動車市場全体に与えた影響を、
競合他社（BYD、フォルクスワーゲン）の
対応戦略とともに分析してください。
"""

# Agentic RAGが実行する検索戦略：
# 1. テスラのバッテリー技術（技術DB + 論文）
# 2. 2024年のEV市場動向（市場レポート + ニュース）
# 3. BYD/VWの戦略（企業発表 + アナリスト分析）
# 4. 技術と市場の因果関係（ナレッジグラフ）

result = agentic_rag.invoke({"question": query})

query = """
気候変動に対するAI技術の貢献可能性を、
以下の観点から評価してください：
1. エネルギー効率化
2. 環境モニタリング
3. カーボンクレジット取引の最適化

各観点について、実際の導入事例と定量的な
インパクト（CO2削減量等）も含めてください。
"""

# Agentic RAGの動作：
# Step 1: 質問を3つのサブクエリに分解
# Step 2: 各サブクエリを並列検索
# Step 3: 事例検索（企業DB + 論文）
# Step 4: 定量データ検索（統計DB + レポート）
# Step 5: 情報統合と回答生成
# Step 6: リフレクション（不足情報があれば再検索）

result = agentic_rag.invoke({"question": query})