Gemini Deep Research：Google 的自主研究智能体

意图理解 → 研究规划 → 并行搜索 → 综合分析 → 报告生成

# 伪代码：Deep Research 意图理解
import google.generativeai as genai

genai.configure(api_key="YOUR_API_KEY")

model = genai.GenerativeModel(
    model_name="gemini-2.5-pro",
    tools=[{
        "google_search_retrieval": {
            "dynamic_retrieval_config": {
                "mode": "MODE_DYNAMIC",
                "dynamic_threshold": 0.3,
            }
        }
    }],
)

# 启动 Deep Research
response = model.start_deep_research(
    query="比较 2025-2026 年主流大语言模型在代码生成领域的性能差异",
    config={
        "depth": "comprehensive",      # quick | standard | comprehensive
        "max_sources": 50,             # 最大信息源数量
        "language": "zh-CN",           # 输出语言
        "format": "academic",          # academic | business | casual
        "citation_style": "inline",    # inline | footnote | endnote
        "max_duration": 300,           # 最大研究时长（秒）
    }
)

# 并行搜索的简化模型
import asyncio
import google.generativeai as genai

async def parallel_search(sub_queries: list[str]):
    """并行执行多个子查询"""
    tasks = []
    for query in sub_queries:
        tasks.append(search_single(query))
    
    results = await asyncio.gather(*tasks)
    
    # 去重和排序
    deduplicated = deduplicate_results(results)
    ranked = rank_by_relevance(deduplicated)
    
    return ranked

async def search_single(query: str):
    model = genai.GenerativeModel("gemini-2.5-flash")
    response = model.generate_content(
        query,
        tools=[{"google_search_retrieval": {}}],
    )
    return {
        "query": query,
        "sources": extract_sources(response),
        "content": response.text,
        "timestamp": get_current_time(),
    }

def synthesize_findings(search_results: list[dict]):
    """综合分析搜索结果"""
    model = genai.GenerativeModel("gemini-2.5-pro")
    
    # 将所有搜索结果拼接到上下文中
    context = "\n\n".join([
        f"[来源 {i+1}] {r['query']}\n{r['content']}"
        for i, r in enumerate(search_results)
    ])
    
    prompt = f"""
    基于以下 {len(search_results)} 个信息源的研究发现，
    请进行综合分析：
    
    1. 识别关键发现和共识
    2. 标记矛盾观点和争议
    3. 评估各信息源的可信度
    4. 指出信息空白和局限
    5. 提出进一步研究方向
    
    信息源内容：
    {context}
    """
    
    response = model.generate_content(prompt)
    return response.text

def generate_report(analysis: str, config: dict):
    """生成结构化研究报告"""
    model = genai.GenerativeModel("gemini-2.5-pro")
    
    format_instructions = {
        "academic": "学术论文格式，包含摘要、引言、方法、结果、讨论、结论",
        "business": "商业报告格式，包含执行摘要、关键发现、建议、附录",
        "casual": "通俗易懂的文章格式，适合大众阅读",
    }
    
    prompt = f"""
    将以下分析内容整理为{format_instructions[config['format']]}。
    
    要求：
    - 所有观点必须标注来源
    - 使用{config['citation_style']}引用格式
    - 语言：{config['language']}
    - 长度：3000-8000 字
    
    分析内容：
    {analysis}
    """
    
    return model.generate_content(prompt)

import google.generativeai as genai
import json

genai.configure(api_key="YOUR_API_KEY")

# 方法一：一键式 Deep Research
model = genai.GenerativeModel("gemini-2.5-pro")
result = model.start_deep_research(
    query="分析 2026 年全球 AI 芯片供应链格局",
    config={
        "depth": "comprehensive",
        "max_sources": 50,
        "language": "zh-CN",
        "format": "academic",
        "stream": True,
    }
)

# 流式接收研究过程
for event in result.stream():
    if event.type == "planning":
        print(f"📋 研究计划: {event.plan}")
    elif event.type == "searching":
        print(f"🔍 搜索中: {event.query} (第{event.index}个/共{event.total}个)")
    elif event.type == "analyzing":
        print(f"🧠 分析中: 已处理 {event.sources_processed} 个信息源")
    elif event.type == "writing":
        print(f"✍️ 生成中: {event.section}")
    elif event.type == "complete":
        print(f"✅ 完成！报告长度: {len(event.report)} 字")

# 方法二：分步控制
# Step 1: 生成研究计划
plan = model.plan_research("分析 2026 年全球 AI 芯片供应链格局")
print(json.dumps(plan, indent=2, ensure_ascii=False))

# Step 2: 修改计划（可选）
plan["sub_queries"].append("台积电产能扩张计划")
plan["sources"].append("https://www.tsmc.com/investor-relations")

# Step 3: 执行研究
results = model.execute_research(plan)

# Step 4: 生成报告
report = model.generate_research_report(results, format="academic")

# Deep Research 成本估算
cost_breakdown = {
    "quick": {
        "duration": "30-60秒",
        "search_calls": 5-10,
        "token_usage": "~50K",
        "estimated_cost": "$0.10-0.20",
    },
    "standard": {
        "duration": "2-5分钟",
        "search_calls": 15-25,
        "token_usage": "~200K",
        "estimated_cost": "$0.40-0.80",
    },
    "comprehensive": {
        "duration": "5-15分钟",
        "search_calls": 30-50,
        "token_usage": "~500K-1M",
        "estimated_cost": "$1.00-3.00",
    },
}

class EnterpriseResearchAgent:
    def __init__(self, api_key: str, knowledge_base=None):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel("gemini-2.5-pro")
        self.knowledge_base = knowledge_base  # 企业内部知识库
    
    def research(self, topic: str, depth: str = "standard"):
        """执行企业研究任务"""
        # 结合内部知识库
        context = ""
        if self.knowledge_base:
            context = f"\n企业内部资料：\n{self.knowledge_base.search(topic)}"
        
        result = self.model.start_deep_research(
            query=topic + context,
            config={
                "depth": depth,
                "format": "business",
                "language": "zh-CN",
                "max_sources": 30,
            }
        )
        return result
    
    def batch_research(self, topics: list[str]):
        """批量研究"""
        import concurrent.futures
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            futures = [executor.submit(self.research, t) for t in topics]
            results = [f.result() for f in futures]
        
        return results