Gemini Deep Research：AI深度研究

import asynciofrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optionalfrom enum import Enum class SearchPhase(Enum):    SEARCH = "search"    READ = "read"    SYNTHESIZE = "synthesize" @dataclassclass KnowledgeNode:    entity: str    info: str    sources: List[str]    confidence: float = 0.0    relations: Dict[str, str] = field(default_factory=dict) @dataclassclass SearchResult:    query: str    url: str    title: str    snippet: str    relevance: float = 0.0 class PlanningAgent:    def __init__(self, research_topic: str):        self.topic = research_topic        self.sub_tasks: List[str] = []        self.search_plan: List[str] = []        self.knowledge_gaps: List[str] = []     def decompose_task(self) -> List[str]:        self.sub_tasks = [            f"{self.topic} - 核心概念研究",            f"{self.topic} - 技术方案分析",            f"{self.topic} - 应用场景调研",            f"{self.topic} - 行业趋势追踪",        ]        return self.sub_tasks     def plan_search_paths(self) -> List[str]:        self.search_plan = [            f"query: {t} site:arxiv.org OR site:scholar.google.com"            for t in self.sub_tasks        ]        return self.search_plan     def identify_gaps(self, knowledge_network: Dict[str, KnowledgeNode]) -> List[str]:        self.knowledge_gaps = [            t for t in self.sub_tasks            if t not in knowledge_network            or knowledge_network[t].confidence < 0.7        ]        return self.knowledge_gaps class ExecutionAgent:    def __init__(self, agent_id: str):        self.agent_id = agent_id        self.current_phase: SearchPhase = SearchPhase.SEARCH        self.results: List[SearchResult] = []     async def execute_search(self, query: str) -> List[SearchResult]:        self.current_phase = SearchPhase.SEARCH        self.results = [            SearchResult(query=query, url=f"https://source-{i}.com",                        title=f"Result {i}", snippet=f"Content {i}",                        relevance=0.9 - i * 0.05)            for i in range(5)        ]        return self.results     async def read_documents(self, results: List[SearchResult]) -> str:        self.current_phase = SearchPhase.READ        extracted = " | ".join(r.snippet for r in results if r.relevance > 0.5)        return extracted     async def synthesize(self, new_info: str,                         existing: Dict[str, KnowledgeNode]) -> Dict[str, KnowledgeNode]:        self.current_phase = SearchPhase.SYNTHESIZE        node = KnowledgeNode(            entity=self.agent_id, info=new_info,            sources=[r.url for r in self.results],            confidence=min(0.95, 0.5 + len(self.results) * 0.1)        )        existing[self.agent_id] = node        return existing class SharedState:    def __init__(self):        self.knowledge_network: Dict[str, KnowledgeNode] = {}        self.messages: List[Dict] = []        self.completed_tasks: List[str] = []     def update_knowledge(self, node_id: str, node: KnowledgeNode):        self.knowledge_network[node_id] = node     def add_message(self, sender: str, content: str):        self.messages.append({"sender": sender, "content": content}) class DeepResearchOrchestrator:    def __init__(self, topic: str, max_rounds: int = 10):        self.planner = PlanningAgent(topic)        self.executors = [ExecutionAgent(f"executor-{i}") for i in range(3)]        self.shared = SharedState()        self.max_rounds = max_rounds        self.current_round = 0     async def run_research(self) -> Dict:        self.planner.decompose_task()        self.planner.plan_search_paths()        for round_num in range(self.max_rounds):            self.current_round = round_num            gaps = self.planner.identify_gaps(self.shared.knowledge_network)            if not gaps:                break            tasks = gaps[:len(self.executors)]            coros = [                self._run_executor_round(exec, task)                for exec, task in zip(self.executors, tasks)            ]            await asyncio.gather(*coros)            self.shared.add_message("planner",                f"Round {round_num} completed, "                f"knowledge size: {len(self.shared.knowledge_network)}")        return self._generate_report()     async def _run_executor_round(self, executor: ExecutionAgent,                                   task: str):        results = await executor.execute_search(task)        info = await executor.read_documents(results)        self.shared.knowledge_network = await executor.synthesize(            info, self.shared.knowledge_network)        self.shared.update_knowledge(task,            self.shared.knowledge_network[executor.agent_id])     def _generate_report(self) -> Dict:        nodes = self.shared.knowledge_network        avg_conf = sum(n.confidence for n in nodes.values()) / max(len(nodes), 1)        return {            "topic": self.planner.topic,            "rounds_completed": self.current_round + 1,            "knowledge_nodes": len(nodes),            "avg_confidence": round(avg_conf, 3),            "sources_count": sum(len(n.sources) for n in nodes.values()),            "status": "completed"        } async def main():    orchestrator = DeepResearchOrchestrator(        topic="量子计算技术发展趋势", max_rounds=5)    report = await orchestrator.run_research()    print(json.dumps(report, indent=2, ensure_ascii=False)) if __name__ == "__main__":    import json    asyncio.run(main())