实战：深度研究助手

from dataclasses import dataclass, field

@dataclass
class Evidence:
    id: str                 # E1, E2...
    claim: str              # 这条证据支持的结论
    quote: str              # 原文摘录（短）
    source_url: str
    source_title: str
    published_at: str | None
    retrieved_at: str
    confidence: float

@dataclass
class ResearchState:
    question: str
    sub_questions: list[str] = field(default_factory=list)
    evidence: list[Evidence] = field(default_factory=list)
    conflicts: list[dict] = field(default_factory=list)
    outline: list[str] = field(default_factory=list)
    draft: str = ""
    citations_ok: bool = False
    budget: dict = field(default_factory=dict)

你是研究规划器。把问题拆成 3–6 个可检索的子问题。

要求：
1. 每个子问题能独立搜索，避免相互依赖。
2. 覆盖：定义/背景、关键数据、不同立场、时间线。
3. 标注每个子问题需要的来源类型（论文/官方/新闻/数据库）。
4. 如果问题本身含糊，先列出需要澄清的点，不要硬编。

问题：{question}
输出 JSON：{"clarifications": [], "sub_questions": [{"q": "...", "source_type": "..."}]}

import asyncio

async def retrieve_all(sub_questions, per_query=6):
    sem = asyncio.Semaphore(4)  # 控制并发

    async def one(sq):
        async with sem:
            results = await search(sq["q"], limit=per_query)
            pages = await asyncio.gather(*[fetch_text(r.url) for r in results[:3]])
            return list(zip(results[:3], pages))

    batches = await asyncio.gather(*[one(sq) for sq in sub_questions])
    return dedupe_by_url([item for batch in batches for item in batch])

从下面的文档中抽取与问题相关的事实。

规则：
1. 每条事实必须附一段原文摘录（≤ 50 字），不得改写。
2. 不要推断文档没有的内容。
3. 数字、日期、机构名必须原样保留。
4. 如果文档与问题无关，返回空数组。
5. 标注这条事实支持哪个子问题。

文档：{text}
子问题：{sub_question}
输出：{"evidence": [{"claim": "...", "quote": "...", "confidence": 0.0-1.0}]}

async def extract_all(pages, sub_questions):
    tasks = [extract_one(text, sq) for (meta, text), sq in pair(pages, sub_questions)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    evidence = []
    for meta, result in zip(pages, results):
        if isinstance(result, Exception):
            continue
        for i, e in enumerate(result["evidence"]):
            if e["confidence"] < 0.5:
                continue
            evidence.append(Evidence(
                id=f"E{len(evidence)+1}",
                claim=e["claim"],
                quote=e["quote"],
                source_url=meta.url,
                source_title=meta.title,
                published_at=meta.published_at,
                retrieved_at=now_iso(),
                confidence=e["confidence"],
            ))
    return evidence

证据表：
[SQ1] E1: 2025 营收 12.4 亿（来源 A）
      E2: 2025 营收 12.1 亿（来源 B）
      → 冲突：口径不同（含/不含汇率影响），在报告中说明

根据证据表写一份研究报告。

结构：
1. 结论（3 条以内）
2. 关键发现（按子问题组织）
3. 风险与不确定性
4. 来源列表

规则：
- 每个事实性句子后标注 [E1] 这样的证据 ID。
- 只使用证据表中的内容；不足时写「资料不足」。
- 遇到冲突，说明两种说法与可能原因，不要擅自选一个。
- 不要写投资建议。
- 800–1500 字。

def verify_citations(draft, evidence_by_id):
    problems = []
    for sent, ids in extract_sentences_with_citations(draft):
        if not ids:
            if is_factual(sent):
                problems.append(("missing_citation", sent))
            continue
        for eid in ids:
            ev = evidence_by_id.get(eid)
            if ev is None:
                problems.append(("unknown_citation", eid))
                continue
            if not supports(ev.quote, sent):
                problems.append(("quote_mismatch", eid, sent))
    return problems