+─────────────────────────────────────────────────────────────────────────+
|                  Deep Research Fleet Architecture                       |
|                                                                         |
|  [ User Research Query ] ──▶ [ Lead Orchestrator ]                      |
|                                     │                                   |
|                        ┌────────────┴────────────┐                      |
|                        ▼                         ▼                      |
|             [ Hypothesis Tree ]        [ Plan Decomposition ]           |
|                        │                                                |
|     ┌──────────────────┼──────────────────┐                             |
|     ▼                  ▼                  ▼                             |
| [ Worker Subagent A] [ Worker Subagent B] [ Worker Subagent C]          |
|  (Playwright/MCP)     (Semantic Search)    (Academic APIs)              |
|     │                  │                  │                             |
|     └──────────────────┼──────────────────┘                             |
|                        ▼                                                |
|           [ Citation & Evidence DAG ] ◀──┐ (Re-query on gaps)           |
|                        │                 │                              |
|                        ▼                 │                              |
|             [ Draft Synthesizer ]        │                              |
|                        │                 │                              |
|                        ▼                 │                              |
|             [ Adversarial Critic ] ──────┘                              |
|                        │                                                |
|                        ▼ (Approved)                                     |
|             [ Final Comprehensive Dossier ]                             |
+─────────────────────────────────────────────────────────────────────────+

                  [ Raíz: Consulta de Usuario ]
                      /          \
            [ Rama 1: Mercado ]   [ Rama 2: Técnica ]
               /         \                │
        [ B1.1 US ]   [ B1.2 EU ]    [ B2.1 Latencia ] (PODADO: Baja Ganancia)
            │              │
      (Puntuación Alta)(Puntuación Alta)
            \              /
        [ Síntesis de Evidencias ]

{
  "claim_id": "CLM-2026-0984",
  "assertion": "El proceso TSMC 2nm N2 reduce el consumo de energía un 15% a igual velocidad frente a N3E.",
  "confidence_score": 0.96,
  "sources": [
    {
      "url": "https://pr.tsmc.com/english/news/3124",
      "sha256_hash": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
      "timestamp": "2026-09-14T08:12:00Z",
      "primary_source": true
    }
  ],
  "verification_status": "corroborated_dual_source"
}

'''
Open Deep Research Multi-Agent Fleet
Ecosystem: Python 3.11+, LangGraph, Pydantic v2, DuckDuckGo / Tavily Search
'''

import os
import json
from typing import List, Dict, Any, Optional, Annotated
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
import operator

# =====================================================================
# 1. Esquemas de Estado y Evidencias con Pydantic
# =====================================================================

class EvidenceItem(BaseModel):
    url: str
    title: str
    snippet: str
    relevance_score: float = Field(ge=0.0, le=1.0)

class SubTopic(BaseModel):
    id: str
    query: str
    reasoning: str
    status: str = 'pending'  # pending, completed, pruned

class ResearchState(TypedDict):
    research_goal: str
    max_iterations: int
    current_iteration: int
    subtopics: List[SubTopic]
    evidences: Annotated[List[EvidenceItem], operator.add]
    intermediate_draft: str
    critic_approved: bool
    critic_feedback: str
    final_report: str

# =====================================================================
# 2. Implementación de Nodos de Agentes
# =====================================================================

def orchestrator_plan_node(state: ResearchState) -> Dict[str, Any]:
    '''
    Descompone el objetivo de investigación en consultas exploratorias ortogonales.
    '''
    print(f"\n[Orquestador] Planificando investigacion para: {state['research_goal']}")
    
    planned_subtopics = [
        SubTopic(id='sub_1', query=f"{state['research_goal']} arquitectura base y benchmarks", reasoning="Establecer linea base"),
        SubTopic(id='sub_2', query=f"{state['research_goal']} limitaciones empresariales y fallos", reasoning="Investigar casos limite"),
        SubTopic(id='sub_3', query=f"{state['research_goal']} costes de produccion en 2026", reasoning="Cuantificar inversion")
    ]
    
    return {
        'subtopics': planned_subtopics,
        'current_iteration': state.get('current_iteration', 0) + 1
    }

def worker_search_node(state: ResearchState) -> Dict[str, Any]:
    '''
    Simula la ejecucion paralela de subagentes recolectando hechos verificados.
    '''
    new_evidences = []
    for sub in state['subtopics']:
        if sub.status == 'pending':
            print(f"  [Flota Trabajadores] Desplegando agente para: '{sub.query}'")
            new_evidences.append(
                EvidenceItem(
                    url=f"https://fuente-autorizada.org/analisis/{sub.id}",
                    title=f"Analisis Verificado sobre {sub.query}",
                    snippet=f"Evidencia empirica confirma que {sub.query} logra 3.4x rendimiento bajo MCTS.",
                    relevance_score=0.92
                )
            )
            sub.status = 'completed'
            
    return {'evidences': new_evidences}

def synthesis_node(state: ResearchState) -> Dict[str, Any]:
    '''
    Sintetiza las evidencias recopiladas en un borrador documentado.
    '''
    print(f"[Sintetizador] Compilando {len(state['evidences'])} evidencias en borrador...")
    draft = f"# Dossier Tecnico Exhaustivo: {state['research_goal']}\n\n"
    draft += "## Hallazgos Arquitectonicos Clave\n"
    for i, ev in enumerate(state['evidences'], 1):
        draft += f"- {ev.snippet} [^{i}]\n"
        
    draft += "\n## Registro de Citas y Fuentes\n"
    for i, ev in enumerate(state['evidences'], 1):
        draft += f"[^{i}]: [{ev.title}]({ev.url}) (Relevancia: {ev.relevance_score})\n"
        
    return {'intermediate_draft': draft}

def critic_review_node(state: ResearchState) -> Dict[str, Any]:
    '''
    El Critico Adversario audita la solidez probatoria del informe.
    '''
    print("[Critico] Auditando rigor documental y cobertura...")
    iteration = state['current_iteration']
    
    if iteration < state['max_iterations'] and len(state['evidences']) < 5:
        print("  [Feedback Critico] Faltan datos cuantitativos. Se requieren mas fuentes.")
        return {
            'critic_approved': False,
            'critic_feedback': "Investigar latencias bajo alta concurrencia."
        }
    else:
        print("  [Feedback Critico] Criterio probatorio satisfecho. Borrador aprobado.")
        return {
            'critic_approved': True,
            'critic_feedback': "Aprobado con verificacion cruzada multifuente.",
            'final_report': state['intermediate_draft']
        }

# =====================================================================
# 3. Ensamblado del Grafo con LangGraph
# =====================================================================

from langgraph.graph import StateGraph, END

def route_critic_decision(state: ResearchState) -> str:
    if state['critic_approved']:
        return "approved"
    return "replan"

def build_research_graph():
    builder = StateGraph(ResearchState)
    
    builder.add_node("orchestrator", orchestrator_plan_node)
    builder.add_node("workers", worker_search_node)
    builder.add_node("synthesizer", synthesis_node)
    builder.add_node("critic", critic_review_node)
    
    builder.set_entry_point("orchestrator")
    builder.add_edge("orchestrator", "workers")
    builder.add_edge("workers", "synthesizer")
    builder.add_edge("synthesizer", "critic")
    
    builder.add_conditional_edges(
        "critic",
        route_critic_decision,
        {
            "approved": END,
            "replan": "orchestrator"
        }
    )
    
    return builder.compile()

# =====================================================================
# 4. Punto de Entrada de Ejecucion
# =====================================================================

if __name__ == '__main__':
    app = build_research_graph()
    initial_input: ResearchState = {
        "research_goal": "Arquitecturas de Ejecucion Duradera para Agentes IA",
        "max_iterations": 2,
        "current_iteration": 0,
        "subtopics": [],
        "evidences": [],
        "intermediate_draft": "",
        "critic_approved": False,
        "critic_feedback": "",
        "final_report": ""
    }
    
    final_output = app.invoke(initial_input)
    print("\n================ DOSSIER FINAL GENERADO ================\n")
    print(final_output["final_report"])

+─────────────────────────────────────────────────────────────────────────+
|                  Deep Research Fleet Architecture                       |
|                                                                         |
|  [ User Research Query ] ──▶ [ Lead Orchestrator ]                      |
|                                     │                                   |
|                        ┌────────────┴────────────┐                      |
|                        ▼                         ▼                      |
|             [ Hypothesis Tree ]        [ Plan Decomposition ]           |
|                        │                                                |
|     ┌──────────────────┼──────────────────┐                             |
|     ▼                  ▼                  ▼                             |
| [ Worker Subagent A] [ Worker Subagent B] [ Worker Subagent C]          |
|  (Playwright/MCP)     (Semantic Search)    (Academic APIs)              |
|     │                  │                  │                             |
|     └──────────────────┼──────────────────┘                             |
|                        ▼                                                |
|           [ Citation & Evidence DAG ] ◀──┐ (Re-query on gaps)           |
|                        │                 │                              |
|                        ▼                 │                              |
|             [ Draft Synthesizer ]        │                              |
|                        │                 │                              |
|                        ▼                 │                              |
|             [ Adversarial Critic ] ──────┘                              |
|                        │                                                |
|                        ▼ (Approved)                                     |
|             [ Final Comprehensive Dossier ]                             |
+─────────────────────────────────────────────────────────────────────────+

                  [ Wurzel: Benutzeranfrage ]
                      /          \
            [ Zweig 1: Markt ]    [ Zweig 2: Technik ]
               /         \                │
        [ Z1.1 US ]   [ Z1.2 EU ]    [ Z2.1 Latenz ] (BESCHNITTEN: Geringer Gewinn)
            │              │
      (Hoher Score)  (Hoher Score)
            \              /
        [ Evidenzsynthese ]

{
  "claim_id": "CLM-2026-0984",
  "assertion": "Der TSMC 2nm N2 Prozess erzielt 15% weniger Leistungsaufnahme bei gleicher Geschwindigkeit gegenüber N3E.",
  "confidence_score": 0.96,
  "sources": [
    {
      "url": "https://pr.tsmc.com/english/news/3124",
      "sha256_hash": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
      "timestamp": "2026-09-14T08:12:00Z",
      "primary_source": true
    }
  ],
  "verification_status": "corroborated_dual_source"
}

'''
Open Deep Research Multi-Agent Fleet
Ecosystem: Python 3.11+, LangGraph, Pydantic v2, DuckDuckGo / Tavily Search
'''

import os
import json
from typing import List, Dict, Any, Optional, Annotated
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
import operator

# =====================================================================
# 1. Pydantic-Zustandsschemata
# =====================================================================

class EvidenceItem(BaseModel):
    url: str
    title: str
    snippet: str
    relevance_score: float = Field(ge=0.0, le=1.0)

class SubTopic(BaseModel):
    id: str
    query: str
    reasoning: str
    status: str = 'pending'  # pending, completed, pruned

class ResearchState(TypedDict):
    research_goal: str
    max_iterations: int
    current_iteration: int
    subtopics: List[SubTopic]
    evidences: Annotated[List[EvidenceItem], operator.add]
    intermediate_draft: str
    critic_approved: bool
    critic_feedback: str
    final_report: str

# =====================================================================
# 2. Agentenknoten-Implementierung
# =====================================================================

def orchestrator_plan_node(state: ResearchState) -> Dict[str, Any]:
    '''
    Zerlegt das Rechercheziel in orthogonale Teilabfragen.
    '''
    print(f"\n[Orchestrator] Plane Recherche fuer: {state['research_goal']}")
    
    planned_subtopics = [
        SubTopic(id='sub_1', query=f"{state['research_goal']} Kernarchitektur und Benchmarks", reasoning="Technische Basis"),
        SubTopic(id='sub_2', query=f"{state['research_goal']} Einschraenkungen und Ausfaelle", reasoning="Grenzfaelle"),
        SubTopic(id='sub_3', query=f"{state['research_goal']} Betriebskosten 2026", reasoning="Kostenanalyse")
    ]
    
    return {
        'subtopics': planned_subtopics,
        'current_iteration': state.get('current_iteration', 0) + 1
    }

def worker_search_node(state: ResearchState) -> Dict[str, Any]:
    '''
    Simuliert parallele Subagenten, die verifizierte Fakten zusammentragen.
    '''
    new_evidences = []
    for sub in state['subtopics']:
        if sub.status == 'pending':
            print(f"  [Worker Fleet] Starte Agent fuer: '{sub.query}'")
            new_evidences.append(
                EvidenceItem(
                    url=f"https://vertrauenswuerdige-quelle.org/analyse/{sub.id}",
                    title=f"Verifizierte Analyse zu {sub.query}",
                    snippet=f"Empirische Daten bestaetigen: {sub.query} erzielt 3.4x Durchsatz mit MCTS-Routing.",
                    relevance_score=0.92
                )
            )
            sub.status = 'completed'
            
    return {'evidences': new_evidences}

def synthesis_node(state: ResearchState) -> Dict[str, Any]:
    '''
    Fuehrt gesammelte Evidenzen in einem zitierten Berichtsentwurf zusammen.
    '''
    print(f"[Synthesizer] Fuehre {len(state['evidences'])} Evidenzen zusammen...")
    draft = f"# Technisches Dossier: {state['research_goal']}\n\n"
    draft += "## Zentrale architektonische Erkenntnisse\n"
    for i, ev in enumerate(state['evidences'], 1):
        draft += f"- {ev.snippet} [^{i}]\n"
        
    draft += "\n## Zitationsverzeichnis\n"
    for i, ev in enumerate(state['evidences'], 1):
        draft += f"[^{i}]: [{ev.title}]({ev.url}) (Relevanz: {ev.relevance_score})\n"
        
    return {'intermediate_draft': draft}

def critic_review_node(state: ResearchState) -> Dict[str, Any]:
    '''
    Der Adversarial Critic prueft Nachweisqualitaet und Quellenabdeckung.
    '''
    print("[Critic] Pruefe Entwurf gegen Dokumentationsstandards...")
    iteration = state['current_iteration']
    
    if iteration < state['max_iterations'] and len(state['evidences']) < 5:
        print("  [Critic Feedback] Mehr quantitative Belege erforderlich.")
        return {
            'critic_approved': False,
            'critic_feedback': "Untersuche Latenz unter hoher Last."
        }
    else:
        print("  [Critic Feedback] Kriterien erfuellt. Entwurf genehmigt.")
        return {
            'critic_approved': True,
            'critic_feedback': "Genehmigt mit Multi-Quellen-Abgleich.",
            'final_report': state['intermediate_draft']
        }

# =====================================================================
# 3. LangGraph Graphen-Aufbau
# =====================================================================

from langgraph.graph import StateGraph, END

def route_critic_decision(state: ResearchState) -> str:
    if state['critic_approved']:
        return "approved"
    return "replan"

def build_research_graph():
    builder = StateGraph(ResearchState)
    
    builder.add_node("orchestrator", orchestrator_plan_node)
    builder.add_node("workers", worker_search_node)
    builder.add_node("synthesizer", synthesis_node)
    builder.add_node("critic", critic_review_node)
    
    builder.set_entry_point("orchestrator")
    builder.add_edge("orchestrator", "workers")
    builder.add_edge("workers", "synthesizer")
    builder.add_edge("synthesizer", "critic")
    
    builder.add_conditional_edges(
        "critic",
        route_critic_decision,
        {
            "approved": END,
            "replan": "orchestrator"
        }
    )
    
    return builder.compile()

# =====================================================================
# 4. Einstiegspunkt zur Ausfuehrung
# =====================================================================

if __name__ == '__main__':
    app = build_research_graph()
    initial_input: ResearchState = {
        "research_goal": "Architekturen fuer langlebige KI-Agenten",
        "max_iterations": 2,
        "current_iteration": 0,
        "subtopics": [],
        "evidences": [],
        "intermediate_draft": "",
        "critic_approved": False,
        "critic_feedback": "",
        "final_report": ""
    }
    
    final_output = app.invoke(initial_input)
    print("\n================ FINALER BERICHT ================\n")
    print(final_output["final_report"])

+─────────────────────────────────────────────────────────────────────────+
|                  Deep Research Fleet Architecture                       |
|                                                                         |
|  [ User Research Query ] ──▶ [ Lead Orchestrator ]                      |
|                                     │                                   |
|                        ┌────────────┴────────────┐                      |
|                        ▼                         ▼                      |
|             [ Hypothesis Tree ]        [ Plan Decomposition ]           |
|                        │                                                |
|     ┌──────────────────┼──────────────────┐                             |
|     ▼                  ▼                  ▼                             |
| [ Worker Subagent A] [ Worker Subagent B] [ Worker Subagent C]          |
|  (Playwright/MCP)     (Semantic Search)    (Academic APIs)              |
|     │                  │                  │                             |
|     └──────────────────┼──────────────────┘                             |
|                        ▼                                                |
|           [ Citation & Evidence DAG ] ◀──┐ (Re-query on gaps)           |
|                        │                 │                              |
|                        ▼                 │                              |
|             [ Draft Synthesizer ]        │                              |
|                        │                 │                              |
|                        ▼                 │                              |
|             [ Adversarial Critic ] ──────┘                              |
|                        │                                                |
|                        ▼ (Approved)                                     |
|             [ Final Comprehensive Dossier ]                             |
+─────────────────────────────────────────────────────────────────────────+

                  [ ルート: ユーザーの問い ]
                      /          \
            [ 分岐 1: 市場動向 ]  [ 分岐 2: 技術仕様 ]
               /         \                │
        [ B1.1 米国 ]  [ B1.2 欧州 ]  [ B2.1 レイテンシ ]（剪定: 利得低）
            │              │
      (高スコア)      (高スコア)
            \              /
        [ 証拠の多層合成 ]

{
  "claim_id": "CLM-2026-0984",
  "assertion": "TSMC 2nm N2プロセスはN3E比で同一速度時に電力を15%削減。",
  "confidence_score": 0.96,
  "sources": [
    {
      "url": "https://pr.tsmc.com/english/news/3124",
      "sha256_hash": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
      "timestamp": "2026-09-14T08:12:00Z",
      "primary_source": true
    }
  ],
  "verification_status": "corroborated_dual_source"
}

'''
Open Deep Research Multi-Agent Fleet
Ecosystem: Python 3.11+, LangGraph, Pydantic v2, DuckDuckGo / Tavily Search
'''

import os
import json
from typing import List, Dict, Any, Optional, Annotated
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
import operator

# =====================================================================
# 1. Pydantic 状態と証拠スキーマ
# =====================================================================

class EvidenceItem(BaseModel):
    url: str
    title: str
    snippet: str
    relevance_score: float = Field(ge=0.0, le=1.0)

class SubTopic(BaseModel):
    id: str
    query: str
    reasoning: str
    status: str = 'pending'  # pending, completed, pruned

class ResearchState(TypedDict):
    research_goal: str
    max_iterations: int
    current_iteration: int
    subtopics: List[SubTopic]
    evidences: Annotated[List[EvidenceItem], operator.add]
    intermediate_draft: str
    critic_approved: bool
    critic_feedback: str
    final_report: str

# =====================================================================
# 2. エージェントノードの実装
# =====================================================================

def orchestrator_plan_node(state: ResearchState) -> Dict[str, Any]:
    '''
    高レイヤの調査課題を直交するサブクエリへと計画分解
    '''
    print(f"\n[オーケストレーター] リサーチ計画を立案中: {state['research_goal']}")
    
    planned_subtopics = [
        SubTopic(id='sub_1', query=f"{state['research_goal']} 基本アーキテクチャとベンチマーク", reasoning="技術的ベースラインの確立"),
        SubTopic(id='sub_2', query=f"{state['research_goal']} 実運用の課題と障害モード", reasoning="エッジケースの調査"),
        SubTopic(id='sub_3', query=f"{state['research_goal']} 2026年の運用コスト試算", reasoning="デプロイ費用の定量化")
    ]
    
    return {
        'subtopics': planned_subtopics,
        'current_iteration': state.get('current_iteration', 0) + 1
    }

def worker_search_node(state: ResearchState) -> Dict[str, Any]:
    '''
    並列サブエージェントがウェブ検索と情報抽出を実行
    '''
    new_evidences = []
    for sub in state['subtopics']:
        if sub.status == 'pending':
            print(f"  [ワーカー艦隊] エージェント起動: '{sub.query}'")
            new_evidences.append(
                EvidenceItem(
                    url=f"https://信頼できる情報源.org/analysis/{sub.id}",
                    title=f"{sub.query} に関する検証レポート",
                    snippet=f"実証実験の結果、{sub.query} はMCTSルーティング下で3.4倍のスループットを達成。",
                    relevance_score=0.92
                )
            )
            sub.status = 'completed'
            
    return {'evidences': new_evidences}

def synthesis_node(state: ResearchState) -> Dict[str, Any]:
    '''
    収集されたエビデンスを統合し、引用付き草案を生成
    '''
    print(f"[シンセサイザー] {len(state['evidences'])} 件のエビデンスからレポート草案を合成中...")
    draft = f"# 技術詳細レポート: {state['research_goal']}\n\n"
    draft += "## 主要なアーキテクチャ上の知見\n"
    for i, ev in enumerate(state['evidences'], 1):
        draft += f"- {ev.snippet} [^{i}]\n"
        
    draft += "\n## 引用元リスト\n"
    for i, ev in enumerate(state['evidences'], 1):
        draft += f"[^{i}]: [{ev.title}]({ev.url}) (関連度: {ev.relevance_score})\n"
        
    return {'intermediate_draft': draft}

def critic_review_node(state: ResearchState) -> Dict[str, Any]:
    '''
    敵対的クリティックがレポートの根拠と完全性を審査
    '''
    print("[クリティック] 引用基準に照らし合わせて草案を監査中...")
    iteration = state['current_iteration']
    
    if iteration < state['max_iterations'] and len(state['evidences']) < 5:
        print("  [クリティック講評] 定量データが不足しています。追加調査を指示。")
        return {
            'critic_approved': False,
            'critic_feedback': "高負荷時における実レイテンシのベンチマークを追加調査してください。"
        }
    else:
        print("  [クリティック講評] 証拠基準を満たしました。草案を承認。")
        return {
            'critic_approved': True,
            'critic_feedback': "複数ソースによる相互検証を完了。",
            'final_report': state['intermediate_draft']
        }

# =====================================================================
# 3. LangGraph によるグラフ構築
# =====================================================================

from langgraph.graph import StateGraph, END

def route_critic_decision(state: ResearchState) -> str:
    if state['critic_approved']:
        return "approved"
    return "replan"

def build_research_graph():
    builder = StateGraph(ResearchState)
    
    builder.add_node("orchestrator", orchestrator_plan_node)
    builder.add_node("workers", worker_search_node)
    builder.add_node("synthesizer", synthesis_node)
    builder.add_node("critic", critic_review_node)
    
    builder.set_entry_point("orchestrator")
    builder.add_edge("orchestrator", "workers")
    builder.add_edge("workers", "synthesizer")
    builder.add_edge("synthesizer", "critic")
    
    builder.add_conditional_edges(
        "critic",
        route_critic_decision,
        {
            "approved": END,
            "replan": "orchestrator"
        }
    )
    
    return builder.compile()

# =====================================================================
# 4. 実行エントリーポイント
# =====================================================================

if __name__ == '__main__':
    app = build_research_graph()
    initial_input: ResearchState = {
        "research_goal": "次世代AIエージェントの永続化実行アーキテクチャ",
        "max_iterations": 2,
        "current_iteration": 0,
        "subtopics": [],
        "evidences": [],
        "intermediate_draft": "",
        "critic_approved": False,
        "critic_feedback": "",
        "final_report": ""
    }
    
    final_output = app.invoke(initial_input)
    print("\n================ 最終生成レポート ================\n")
    print(final_output["final_report"])