第十三章 Deep Research Agent¶

"""Deep Research Agent — 数据模型定义"""

from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime

class ResearchPhase(Enum):
    """研究阶段状态机"""
    PLANNING = "planning"
    SEARCHING = "searching"
    READING = "reading"
    SYNTHESIZING = "synthesizing"
    WRITING = "writing"
    REVIEWING = "reviewing"
    COMPLETE = "complete"

@dataclass
class SubQuestion:
    """子问题"""
    id: int
    question: str
    priority: int = 1               # 1-5, 5最高
    status: str = "pending"          # pending / searching / answered / failed
    search_queries: list[str] = field(default_factory=list)
    findings: list[str] = field(default_factory=list)
    sources: list[str] = field(default_factory=list)

@dataclass
class SearchResult:
    """搜索结果"""
    url: str
    title: str
    snippet: str
    score: float = 0.0
    fetched: bool = False

@dataclass
class ExtractedInfo:
    """从网页提取的信息"""
    source_url: str
    source_title: str
    content: str
    key_facts: list[str] = field(default_factory=list)
    relevance_score: float = 0.0
    extracted_at: str = field(default_factory=lambda: datetime.now().isoformat())  # lambda包装确保每次实例化取当前时间

@dataclass
class Citation:
    """引用"""
    id: int
    url: str
    title: str
    accessed_at: str
    snippet: str = ""

@dataclass
class ResearchState:
    """研究状态——贯穿整个研究流程的全局状态"""
    query: str
    phase: ResearchPhase = ResearchPhase.PLANNING
    sub_questions: list[SubQuestion] = field(default_factory=list)
    search_results: list[SearchResult] = field(default_factory=list)
    extracted_infos: list[ExtractedInfo] = field(default_factory=list)
    citations: list[Citation] = field(default_factory=list)
    visited_urls: set[str] = field(default_factory=set)
    report: str = ""
    iteration: int = 0
    max_iterations: int = 3
    token_budget: int = 100_000      # Token预算上限
    tokens_used: int = 0

"""Research Planner — 问题分解与计划生成"""

import json
from openai import AsyncOpenAI

client = AsyncOpenAI()

PLANNER_SYSTEM_PROMPT = """你是一位资深的研究规划师。你的任务是将用户的研究问题分解为3-6个具体的子问题。

每个子问题应该：
1. 足够具体，可以通过网络搜索找到答案
2. 覆盖问题的不同方面（定义、现状、技术细节、对比、趋势等）
3. 按逻辑顺序排列（基础→进阶）
4. 包含推荐的搜索查询词

输出JSON格式：
{
    "research_plan": {
        "summary": "研究计划简述",
        "sub_questions": [
            {
                "id": 1,
                "question": "子问题描述",
                "priority": 5,
                "search_queries": ["搜索词1", "搜索词2"]
            }
        ]
    }
}"""

async def create_research_plan(query: str) -> list[SubQuestion]:
    """将用户问题分解为子问题，返回研究计划"""
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": PLANNER_SYSTEM_PROMPT},
            {"role": "user", "content": f"请为以下研究问题制定计划：\n\n{query}"},
        ],
        response_format={"type": "json_object"},
        temperature=0.7,
    )

    plan_data = json.loads(response.choices[0].message.content)
    sub_questions = []
    for sq in plan_data["research_plan"]["sub_questions"]:
        sub_questions.append(SubQuestion(
            id=sq["id"],
            question=sq["question"],
            priority=sq.get("priority", 3),
            search_queries=sq.get("search_queries", [sq["question"]]),
        ))

    return sub_questions

REPLAN_PROMPT = """基于当前的研究进展，评估是否需要调整研究计划。

已完成的子问题及其发现：
{completed_findings}

尚未完成的子问题：
{pending_questions}

请判断：
1. 是否有重要方向被遗漏？
2. 已有发现是否暗示了新的值得探索的子问题？
3. 哪些未完成的子问题可以跳过（因为已被其他发现覆盖）？

输出JSON：
{{
    "add_questions": [...],
    "skip_questions": [子问题id列表],
    "reasoning": "调整原因"
}}"""

async def replan(
    state: ResearchState,
) -> tuple[list[SubQuestion], list[int]]:
    """根据已有发现动态调整研究计划

    Returns:
        (new_questions, skip_ids): 新增的子问题和要跳过的子问题ID
    """
    completed = "\n".join(
        f"- Q{sq.id}: {sq.question}\n  发现: {'; '.join(sq.findings[:3])}"
        for sq in state.sub_questions if sq.status == "answered"
    )
    pending = "\n".join(
        f"- Q{sq.id}: {sq.question}"
        for sq in state.sub_questions if sq.status == "pending"
    )

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是研究计划调整助手。"},
            {"role": "user", "content": REPLAN_PROMPT.format(
                completed_findings=completed or "暂无",
                pending_questions=pending or "暂无",
            )},
        ],
        response_format={"type": "json_object"},
        temperature=0.5,
    )

    result = json.loads(response.choices[0].message.content)

    new_questions = []
    max_id = max((sq.id for sq in state.sub_questions), default=0)
    for i, q in enumerate(result.get("add_questions", []), start=max_id + 1):
        new_questions.append(SubQuestion(
            id=i,
            question=q if isinstance(q, str) else q.get("question", ""),  # 三元+isinstance：兼容字符串和字典两种输入格式
            priority=3,
            search_queries=[q] if isinstance(q, str) else q.get("search_queries", []),
        ))

    skip_ids = result.get("skip_questions", [])
    return new_questions, skip_ids

"""Web Searcher — 搜索API集成与内容获取"""

import asyncio
import httpx
from tavily import AsyncTavilyClient

class WebSearcher:
    """Web搜索器：支持Tavily和备用的httpx直接搜索"""

    def __init__(self, tavily_api_key: str):
        self.tavily = AsyncTavilyClient(api_key=tavily_api_key)
        self.http_client = httpx.AsyncClient(
            timeout=15.0,
            follow_redirects=True,
            headers={"User-Agent": "DeepResearchAgent/1.0"},
        )

    async def search(
        self,
        query: str,
        max_results: int = 5,
        search_depth: str = "advanced",
    ) -> list[SearchResult]:
        """执行搜索，返回结构化结果

        注意：生产环境应添加速率限制（如 asyncio.Semaphore 或令牌桶），
        避免短时间内发送大量请求导致API配额耗尽或被封禁。
        """
        try:
            response = await self.tavily.search(
                query=query,
                max_results=max_results,
                search_depth=search_depth,     # "basic" 或 "advanced"
                include_raw_content=False,
            )
            results = []
            for item in response.get("results", []):
                results.append(SearchResult(
                    url=item["url"],
                    title=item.get("title", ""),
                    snippet=item.get("content", ""),
                    score=item.get("score", 0.0),
                ))
            return results
        except Exception as e:
            print(f"[搜索错误] {query}: {e}")
            return []

    async def search_multiple(
        self,
        queries: list[str],
        max_results_per_query: int = 3,
    ) -> list[SearchResult]:
        """并行执行多个搜索查询"""
        tasks = [
            self.search(q, max_results=max_results_per_query)
            for q in queries
        ]
        all_results_nested = await asyncio.gather(*tasks)

        # 合并并去重（按URL）
        seen_urls: set[str] = set()
        merged: list[SearchResult] = []
        for results in all_results_nested:
            for r in results:
                if r.url not in seen_urls:
                    seen_urls.add(r.url)
                    merged.append(r)

        # 按相关性排序
        merged.sort(key=lambda x: x.score, reverse=True)  # 合并去重后按分数降序排列
        return merged

    async def close(self):
        await self.http_client.aclose()

"""Content Reader — 网页内容提取与清洗"""

import re
import httpx

async def fetch_page_content(
    url: str,
    client: httpx.AsyncClient | None = None,
    max_chars: int = 15_000,
) -> str | None:
    """获取网页内容并清洗为纯文本

    Args:
        url: 目标URL
        client: 可复用的httpx客户端
        max_chars: 最大字符数（截断保护）
    """
    _client = client or httpx.AsyncClient(timeout=15.0, follow_redirects=True)
    try:
        resp = await _client.get(url)
        resp.raise_for_status()
        html = resp.text
    except Exception as e:
        print(f"[获取失败] {url}: {e}")
        return None
    finally:
        if client is None:
            await _client.aclose()

    return clean_html(html, max_chars)

def clean_html(html: str, max_chars: int = 15_000) -> str:
    """将HTML清洗为纯文本"""
    # 移除script和style标签及其内容（这些不包含有用信息）
    html = re.sub(r"<script[^>]*>.*?</script>", "", html, flags=re.DOTALL | re.IGNORECASE)
    html = re.sub(r"<style[^>]*>.*?</style>", "", html, flags=re.DOTALL | re.IGNORECASE)
    # 移除HTML注释
    html = re.sub(r"<!--.*?-->", "", html, flags=re.DOTALL)
    # 移除所有HTML标签，保留纯文本
    text = re.sub(r"<[^>]+>", " ", html)
    # 解码常见HTML实体（如 &amp; → &）
    text = text.replace("&amp;", "&").replace("&lt;", "<").replace("&gt;", ">")
    text = text.replace("&nbsp;", " ").replace("&quot;", '"')
    # 合并多余空白字符为单个空格
    text = re.sub(r"\s+", " ", text).strip()
    # 截断保护：防止过长文本耗尽Token预算
    if len(text) > max_chars:
        text = text[:max_chars] + "\n\n[内容已截断...]"
    return text

from urllib.parse import urlparse

# 低质量域名黑名单
LOW_QUALITY_DOMAINS = {
    "pinterest.com", "quora.com", "facebook.com",
    "instagram.com", "tiktok.com", "twitter.com",
}

def filter_search_results(
    results: list[SearchResult],
    visited_urls: set[str],
    max_results: int = 8,
) -> list[SearchResult]:
    """过滤低质量和已访问的搜索结果，提升后续阅读效率"""
    filtered = []
    for r in results:
        # 跳过已访问的URL，避免重复处理
        if r.url in visited_urls:
            continue
        # 跳过社交媒体等低信息密度站点
        domain = urlparse(r.url).netloc.replace("www.", "")
        if domain in LOW_QUALITY_DOMAINS:
            continue
        # 跳过二进制文件格式（难以解析为纯文本）
        if r.url.endswith((".pdf", ".ppt", ".pptx", ".doc", ".docx")):
            continue
        filtered.append(r)

    return filtered[:max_results]

def chunk_text(text: str, chunk_size: int = 4000, overlap: int = 200) -> list[str]:
    """将长文本分块

    Args:
        text: 原始文本
        chunk_size: 每块字符数
        overlap: 块之间的重叠字符数
    """
    if len(text) <= chunk_size:
        return [text]

    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        # 尝试在句号处断开
        if end < len(text):
            last_period = text.rfind(".", start + chunk_size // 2, end)
            if last_period > start:
                end = last_period + 1
        chunks.append(text[start:end])
        start = end - overlap

    return chunks

EXTRACT_PROMPT = """你是一位研究助手。请从以下网页内容中提取与研究问题相关的关键信息。

研究问题：{question}

网页内容：
{content}

请输出JSON格式：
{{
    "key_facts": ["事实1", "事实2", ...],
    "relevance_score": 0.0 到 1.0 之间的相关度评分,
    "summary": "50字以内的核心摘要"
}}

注意：只提取与研究问题直接相关的信息，忽略广告、导航等无关内容。"""

async def extract_info_from_content(
    content: str,
    question: str,
    source_url: str,
    source_title: str,
) -> ExtractedInfo:
    """使用LLM从网页内容中提取关键信息"""
    chunks = chunk_text(content, chunk_size=6000)

    all_facts: list[str] = []
    best_relevance = 0.0

    for chunk in chunks[:3]:  # 最多处理3个块（预算控制）
        response = await client.chat.completions.create(
            model="gpt-4o-mini",  # 使用较小模型降低成本
            messages=[
                {"role": "system", "content": "你是一位精确的信息提取助手。"},
                {"role": "user", "content": EXTRACT_PROMPT.format(
                    question=question,
                    content=chunk,
                )},
            ],
            response_format={"type": "json_object"},
            temperature=0.2,
        )

        result = json.loads(response.choices[0].message.content)
        all_facts.extend(result.get("key_facts", []))
        relevance = result.get("relevance_score", 0.0)
        best_relevance = max(best_relevance, relevance)

    return ExtractedInfo(
        source_url=source_url,
        source_title=source_title,
        content=content[:2000],  # 保留前2000字符作为参考
        key_facts=all_facts,
        relevance_score=best_relevance,
    )

FACT_CHECK_PROMPT = """你是一位事实核查专家。请对以下声明进行交叉验证。

声明：{claim}

来源1的说法：{source1}
来源2的说法：{source2}

请判断：
1. 两个来源是否一致？
2. 如果不一致，哪个更可信？为什么？
3. 置信度评分（0-1）

输出JSON：
{{
    "consistent": true/false,
    "confidence": 0.0-1.0,
    "reasoning": "判断理由",
    "verified_fact": "经核查后的事实陈述"
}}"""

async def cross_validate(
    claim: str,
    sources: list[ExtractedInfo],
) -> dict:
    """交叉验证一个事实声明"""
    if len(sources) < 2:
        return {"consistent": True, "confidence": 0.5, "verified_fact": claim}

    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是事实核查专家。"},
            {"role": "user", "content": FACT_CHECK_PROMPT.format(
                claim=claim,
                source1=sources[0].key_facts[:3],
                source2=sources[1].key_facts[:3],
            )},
        ],
        response_format={"type": "json_object"},
        temperature=0.2,
    )
    return json.loads(response.choices[0].message.content)

class CitationTracker:
    """引用追踪器——管理所有引用来源"""

    def __init__(self):
        self.citations: list[Citation] = []
        self._url_to_id: dict[str, int] = {}

    def add_citation(self, url: str, title: str, snippet: str = "") -> int:
        """添加引用，返回引用编号"""
        if url in self._url_to_id:
            return self._url_to_id[url]

        cid = len(self.citations) + 1
        self.citations.append(Citation(
            id=cid,
            url=url,
            title=title,
            accessed_at=datetime.now().strftime("%Y-%m-%d"),
            snippet=snippet,
        ))
        self._url_to_id[url] = cid
        return cid

    def format_reference_list(self) -> str:
        """生成引用列表（Markdown格式）"""
        lines = ["## 参考资料\n"]
        for c in self.citations:
            lines.append(f"[{c.id}] [{c.title}]({c.url}) — 访问日期: {c.accessed_at}")
        return "\n".join(lines)

    def inline_cite(self, citation_id: int) -> str:
        """生成行内引用标记"""
        return f"[{citation_id}]"

SYNTHESIZE_PROMPT = """你是一位研究分析师。请综合以下多个来源的信息，回答研究子问题。

子问题：{question}

来源信息：
{sources_text}

要求：
1. 综合所有来源的关键发现
2. 标注每个关键事实的来源编号（如 [1], [2]）
3. 如果来源之间存在矛盾，请明确指出并给出你的判断
4. 给出对该子问题的综合回答

输出JSON：
{{
    "answer": "综合回答（带引用标注）",
    "key_findings": ["发现1 [来源编号]", "发现2 [来源编号]"],
    "contradictions": ["来源A说...，但来源B说..."],
    "confidence": 0.0-1.0,
    "gaps": ["尚未覆盖的方面"]
}}"""

async def synthesize_for_question(
    question: str,
    infos: list[ExtractedInfo],
    citation_tracker: CitationTracker,
) -> dict:
    """为一个子问题综合多源信息"""
    # 构建来源文本，同时注册引用
    sources_parts = []
    for info in infos:
        cid = citation_tracker.add_citation(
            url=info.source_url,
            title=info.source_title,
            snippet="; ".join(info.key_facts[:2]),  # 取前两条关键事实作为引用摘要
        )
        facts_text = "\n".join(f"  - {f}" for f in info.key_facts)
        sources_parts.append(
            f"来源[{cid}] ({info.source_title}):\n{facts_text}"
        )

    sources_text = "\n\n".join(sources_parts)

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是一位严谨的研究分析师。"},
            {"role": "user", "content": SYNTHESIZE_PROMPT.format(
                question=question,
                sources_text=sources_text,
            )},
        ],
        response_format={"type": "json_object"},
        temperature=0.3,
    )
    return json.loads(response.choices[0].message.content)

async def detect_contradictions(
    findings: list[dict],
) -> list[dict]:
    """检测不同子问题的发现之间的矛盾"""
    all_facts = []
    for f in findings:
        all_facts.extend(f.get("key_findings", []))

    if len(all_facts) < 2:
        return []

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是矛盾检测专家。"},
            {"role": "user", "content": f"""请检查以下研究发现之间是否存在矛盾：

{chr(10).join(f'- {fact}' for fact in all_facts)}

输出JSON：
{{
    "contradictions": [
        {{
            "fact_a": "事实A",
            "fact_b": "与之矛盾的事实B",
            "resolution": "建议的解决方式",
            "keep": "A 或 B 或 both"
        }}
    ]
}}"""},
        ],
        response_format={"type": "json_object"},
        temperature=0.2,
    )
    result = json.loads(response.choices[0].message.content)
    return result.get("contradictions", [])

def compute_overall_confidence(findings: list[dict]) -> float:
    """计算研究的总体置信度

    综合考虑三个因素：
    - 各子问题的置信度均值（基础分）
    - 矛盾数量（扣分项：矛盾越多置信度越低）
    - 来源数量（加分项：来源越多越可靠）
    """
    if not findings:
        return 0.0

    # 基础分：各子问题置信度的算术平均
    confidences = [f.get("confidence", 0.5) for f in findings]
    avg_confidence = sum(confidences) / len(confidences)

    # 矛盾惩罚：每个矛盾扣分0.05，最多扣0.2
    total_contradictions = sum(
        len(f.get("contradictions", [])) for f in findings
    )
    contradiction_penalty = min(total_contradictions * 0.05, 0.2)

    # 来源加分：来源越多交叉验证越充分，最多加0.1
    total_sources = sum(
        len(f.get("key_findings", [])) for f in findings
    )
    source_bonus = min(total_sources * 0.01, 0.1)

    # 结果限制在[0, 1]范围内
    return max(0.0, min(1.0, avg_confidence - contradiction_penalty + source_bonus))

REPORT_PROMPT = """你是一位专业的研究报告撰写人。请基于以下研究发现，撰写一份结构化的研究报告。

原始研究问题：{query}

各子问题的研究发现：
{findings_text}

矛盾与冲突：
{contradictions_text}

总体置信度：{confidence:.0%}

要求：
1. 使用Markdown格式
2. 包含以下结构：
   - 摘要（Executive Summary）：3-5句话概述核心发现
   - 详细分析：按主题组织，非按子问题组织
   - 关键发现：列表形式
   - 局限性与不确定性
   - 结论
3. 所有关键事实必须标注引用来源（如 [1], [2]）
4. 语言专业、客观、简洁
5. 如果存在矛盾信息，在报告中明确说明

请直接输出Markdown格式的报告内容（不要包裹在JSON中）。"""

async def generate_report(
    query: str,
    findings: list[dict],
    contradictions: list[dict],
    confidence: float,
    citation_tracker: CitationTracker,
) -> str:
    """生成结构化研究报告"""
    findings_text = ""
    for i, f in enumerate(findings, 1):
        findings_text += f"\n### 子问题 {i}\n"
        findings_text += f"回答: {f.get('answer', 'N/A')}\n"
        findings_text += "关键发现:\n"
        for kf in f.get("key_findings", []):
            findings_text += f"  - {kf}\n"

    contradictions_text = "无明显矛盾" if not contradictions else "\n".join(
        f"- {c['fact_a']} vs {c['fact_b']}（建议: {c['resolution']}）"
        for c in contradictions
    )

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是一位专业研究报告撰写人。"},
            {"role": "user", "content": REPORT_PROMPT.format(
                query=query,
                findings_text=findings_text,
                contradictions_text=contradictions_text,
                confidence=confidence,
            )},
        ],
        temperature=0.4,
        max_tokens=4000,
    )

    report = response.choices[0].message.content
    # 追加引用列表
    report += "\n\n---\n\n" + citation_tracker.format_reference_list()
    return report

import json as json_lib

def export_report(report: str, format: str = "markdown") -> str:
    """将报告导出为不同格式

    Args:
        report: Markdown格式的报告内容
        format: 输出格式 - "markdown", "html", "json"
    """
    if format == "markdown":
        return report

    elif format == "html":
        # 简易Markdown→HTML转换（生产环境建议用markdown库）
        html = report
        # 标题转换：### → <h3>, ## → <h2>, # → <h1>
        html = re.sub(r"^### (.+)$", r"<h3>\1</h3>", html, flags=re.MULTILINE)
        html = re.sub(r"^## (.+)$", r"<h2>\1</h2>", html, flags=re.MULTILINE)
        html = re.sub(r"^# (.+)$", r"<h1>\1</h1>", html, flags=re.MULTILINE)
        # 粗体和斜体：**text** → <strong>, *text* → <em>
        html = re.sub(r"\*\*(.+?)\*\*", r"<strong>\1</strong>", html)
        html = re.sub(r"\*(.+?)\*", r"<em>\1</em>", html)
        # 列表项：- text → <li>text</li>
        html = re.sub(r"^- (.+)$", r"<li>\1</li>", html, flags=re.MULTILINE)
        # 段落分隔：双换行转为段落标签
        html = re.sub(r"\n\n", r"</p><p>", html)
        return f"<html><body><p>{html}</p></body></html>"

    elif format == "json":
        # 将报告按##标题拆分为结构化JSON
        sections = re.split(r"^## ", report, flags=re.MULTILINE)
        result = {"title": sections[0].strip() if sections else "", "sections": []}
        for s in sections[1:]:
            lines = s.strip().split("\n", 1)
            result["sections"].append({
                "heading": lines[0].strip(),
                "content": lines[1].strip() if len(lines) > 1 else "",
            })
        return json_lib.dumps(result, ensure_ascii=False, indent=2)

    else:
        raise ValueError(f"不支持的格式: {format}")

"""
Mini Deep Research Agent — 完整实现
====================================
一个能够自主规划、搜索、阅读、推理、撰写研究报告的Agent。

依赖：pip install openai tavily-python httpx
环境变量：OPENAI_API_KEY, TAVILY_API_KEY
"""

from __future__ import annotations

import asyncio
import json
import os
import re
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from urllib.parse import urlparse

import httpx
from openai import AsyncOpenAI
from tavily import AsyncTavilyClient

# ─────────────────────────── 数据模型 ───────────────────────────

class ResearchPhase(Enum):
    PLANNING = "planning"
    SEARCHING = "searching"
    READING = "reading"
    SYNTHESIZING = "synthesizing"
    WRITING = "writing"
    COMPLETE = "complete"

@dataclass
class SubQuestion:
    id: int
    question: str
    priority: int = 3
    status: str = "pending"
    search_queries: list[str] = field(default_factory=list)
    findings: list[str] = field(default_factory=list)
    sources: list[str] = field(default_factory=list)

@dataclass
class SearchResult:
    url: str
    title: str
    snippet: str
    score: float = 0.0

@dataclass
class ExtractedInfo:
    source_url: str
    source_title: str
    key_facts: list[str] = field(default_factory=list)
    relevance_score: float = 0.0

@dataclass
class Citation:
    id: int
    url: str
    title: str
    accessed_at: str

@dataclass
class ResearchState:
    query: str
    phase: ResearchPhase = ResearchPhase.PLANNING
    sub_questions: list[SubQuestion] = field(default_factory=list)
    extracted_infos: list[ExtractedInfo] = field(default_factory=list)
    findings: list[dict] = field(default_factory=list)
    visited_urls: set[str] = field(default_factory=set)
    report: str = ""
    iteration: int = 0
    max_iterations: int = 3

# ─────────────────────────── 引用追踪 ───────────────────────────

class CitationTracker:
    def __init__(self):
        self.citations: list[Citation] = []
        self._url_map: dict[str, int] = {}

    def add(self, url: str, title: str) -> int:
        if url in self._url_map:
            return self._url_map[url]
        cid = len(self.citations) + 1
        self.citations.append(Citation(
            id=cid, url=url, title=title,
            accessed_at=datetime.now().strftime("%Y-%m-%d"),
        ))
        self._url_map[url] = cid
        return cid

    def format_references(self) -> str:
        lines = ["\n## 参考资料\n"]
        for c in self.citations:
            lines.append(f"[{c.id}] [{c.title}]({c.url})")
        return "\n".join(lines)

# ─────────────────────────── 工具函数 ───────────────────────────

LOW_QUALITY_DOMAINS = {"pinterest.com", "quora.com", "facebook.com", "tiktok.com"}

def clean_html(html: str, max_chars: int = 12_000) -> str:
    html = re.sub(r"<script[^>]*>.*?</script>", "", html, flags=re.DOTALL | re.I)
    html = re.sub(r"<style[^>]*>.*?</style>", "", html, flags=re.DOTALL | re.I)
    text = re.sub(r"<[^>]+>", " ", html)
    text = text.replace("&amp;", "&").replace("&nbsp;", " ")
    text = re.sub(r"\s+", " ", text).strip()
    return text[:max_chars]

def chunk_text(text: str, size: int = 5000, overlap: int = 200) -> list[str]:
    if len(text) <= size:
        return [text]
    chunks, start = [], 0
    while start < len(text):
        end = min(start + size, len(text))
        chunks.append(text[start:end])
        start = end - overlap
    return chunks

# ─────────────────────────── 核心Agent ──────────────────────────

class DeepResearchAgent:
    """Mini Deep Research Agent"""

    def __init__(
        self,
        openai_api_key: str | None = None,
        tavily_api_key: str | None = None,
        model: str = "gpt-4o",
        fast_model: str = "gpt-4o-mini",
        max_iterations: int = 3,
    ):
        self.llm = AsyncOpenAI(api_key=openai_api_key)
        self.tavily = AsyncTavilyClient(
            api_key=tavily_api_key or os.getenv("TAVILY_API_KEY", ""),
        )
        self.http = httpx.AsyncClient(timeout=15, follow_redirects=True)
        self.model = model
        self.fast_model = fast_model
        self.max_iterations = max_iterations
        self.citations = CitationTracker()

    # ────────── 1. 规划 ──────────

    async def plan(self, query: str) -> list[SubQuestion]:
        self._log("📋 规划研究方案...")
        resp = await self.llm.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": (
                    "你是研究规划师。将用户问题分解为3-5个可搜索的子问题。"
                    "输出JSON: {\"sub_questions\": [{\"id\": 1, \"question\": \"...\", "
                    "\"priority\": 1-5, \"search_queries\": [\"...\"]}]}"
                )},
                {"role": "user", "content": query},
            ],
            response_format={"type": "json_object"},
            temperature=0.7,
        )
        data = json.loads(resp.choices[0].message.content)
        sqs = []
        for sq in data["sub_questions"]:
            sqs.append(SubQuestion(
                id=sq["id"],
                question=sq["question"],
                priority=sq.get("priority", 3),
                search_queries=sq.get("search_queries", [sq["question"]]),
            ))
        self._log(f"   → 生成 {len(sqs)} 个子问题")
        return sqs

    # ────────── 2. 搜索 ──────────

    async def search(
        self,
        queries: list[str],
        visited: set[str],
        per_query: int = 3,
    ) -> list[SearchResult]:
        self._log(f"🔍 并行搜索 {len(queries)} 个查询...")
        tasks = [
            self.tavily.search(q, max_results=per_query, search_depth="advanced")
            for q in queries
        ]
        raw_results = await asyncio.gather(*tasks, return_exceptions=True)

        seen: set[str] = set()
        results: list[SearchResult] = []
        for raw in raw_results:
            if isinstance(raw, Exception):  # gather(return_exceptions=True)会将异常作为结果返回，此处跳过
                continue
            for item in raw.get("results", []):
                url = item["url"]
                domain = urlparse(url).netloc.replace("www.", "")
                if (url in visited or url in seen
                        or domain in LOW_QUALITY_DOMAINS):
                    continue
                seen.add(url)
                results.append(SearchResult(
                    url=url,
                    title=item.get("title", ""),
                    snippet=item.get("content", ""),
                    score=item.get("score", 0.0),
                ))
        results.sort(key=lambda r: r.score, reverse=True)
        self._log(f"   → 获得 {len(results)} 个唯一结果")
        return results[:10]

    # ────────── 3. 阅读 ──────────

    async def read_and_extract(
        self,
        result: SearchResult,
        question: str,
    ) -> ExtractedInfo | None:
        try:
            resp = await self.http.get(result.url)
            resp.raise_for_status()
            text = clean_html(resp.text)
        except Exception:
            return None

        if len(text) < 100:
            return None

        chunks = chunk_text(text)
        all_facts: list[str] = []
        best_rel = 0.0

        for chunk in chunks[:2]:
            llm_resp = await self.llm.chat.completions.create(
                model=self.fast_model,
                messages=[
                    {"role": "system", "content": "从网页文本中提取与问题相关的关键事实。"
                     "输出JSON: {\"facts\": [\"...\"], \"relevance\": 0.0-1.0}"},
                    {"role": "user", "content": (
                        f"问题: {question}\n\n网页内容:\n{chunk}"
                    )},
                ],
                response_format={"type": "json_object"},
                temperature=0.2,
            )
            data = json.loads(llm_resp.choices[0].message.content)
            all_facts.extend(data.get("facts", []))
            best_rel = max(best_rel, data.get("relevance", 0.0))

        if not all_facts:
            return None

        return ExtractedInfo(
            source_url=result.url,
            source_title=result.title,
            key_facts=all_facts,
            relevance_score=best_rel,
        )

    async def read_multiple(
        self,
        results: list[SearchResult],
        question: str,
        max_pages: int = 4,
    ) -> list[ExtractedInfo]:
        self._log(f"📖 阅读 {min(len(results), max_pages)} 个网页...")
        tasks = [
            self.read_and_extract(r, question)
            for r in results[:max_pages]
        ]
        infos_raw = await asyncio.gather(*tasks)
        infos = [i for i in infos_raw if i is not None]  # 过滤提取失败（返回None）的结果
        self._log(f"   → 成功提取 {len(infos)} 个来源的信息")
        return infos

    # ────────── 4. 综合 ──────────

    async def synthesize(
        self,
        question: str,
        infos: list[ExtractedInfo],
    ) -> dict:
        sources_text = ""
        for info in infos:
            cid = self.citations.add(info.source_url, info.source_title)
            facts = "\n".join(f"  - {f}" for f in info.key_facts)
            sources_text += f"\n来源[{cid}] ({info.source_title}):\n{facts}\n"

        resp = await self.llm.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": (
                    "综合多个来源回答研究子问题。标注引用来源编号。"
                    "输出JSON: {\"answer\": \"...\", \"key_findings\": [\"...\"], "
                    "\"confidence\": 0.0-1.0, \"gaps\": [\"...\"]}"
                )},
                {"role": "user", "content": f"子问题: {question}\n\n{sources_text}"},
            ],
            response_format={"type": "json_object"},
            temperature=0.3,
        )
        return json.loads(resp.choices[0].message.content)

    # ────────── 5. 报告 ──────────

    async def write_report(self, query: str, findings: list[dict]) -> str:
        self._log("📝 撰写研究报告...")
        findings_text = ""
        for i, f in enumerate(findings, 1):
            findings_text += f"\n### 子问题 {i}\n"
            findings_text += f"回答: {f.get('answer', 'N/A')}\n"
            for kf in f.get("key_findings", []):
                findings_text += f"  - {kf}\n"

        resp = await self.llm.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": (
                    "你是专业研究报告撰写人。基于研究发现撰写Markdown格式报告。"
                    "结构：摘要→详细分析→关键发现→局限性→结论。"
                    "所有关键事实标注引用编号 [1], [2] 等。"
                )},
                {"role": "user", "content": (
                    f"研究问题：{query}\n\n研究发现：\n{findings_text}"
                )},
            ],
            temperature=0.4,
            max_tokens=4000,
        )
        report = resp.choices[0].message.content
        report += "\n\n---\n" + self.citations.format_references()
        return report

    # ────────── 主循环 ──────────

    async def research(self, query: str) -> str:
        """执行完整的深度研究流程，返回研究报告"""
        self._log(f"🚀 开始研究: {query}\n")
        state = ResearchState(query=query, max_iterations=self.max_iterations)

        # Phase 1: 规划
        state.sub_questions = await self.plan(query)
        state.phase = ResearchPhase.SEARCHING

        # Phase 2-4: 对每个子问题执行 搜索→阅读→综合
        for sq in state.sub_questions:
            self._log(f"\n── 子问题 {sq.id}: {sq.question}")
            sq.status = "searching"

            # 搜索
            results = await self.search(
                sq.search_queries, state.visited_urls,
            )
            for r in results:
                state.visited_urls.add(r.url)

            # 阅读
            infos = await self.read_multiple(results, sq.question)
            state.extracted_infos.extend(infos)

            # 综合
            if infos:
                self._log("🧠 综合分析...")
                finding = await self.synthesize(sq.question, infos)
                state.findings.append(finding)
                sq.findings = finding.get("key_findings", [])
                sq.status = "answered"
                self._log(f"   → 置信度: {finding.get('confidence', 0):.0%}")
            else:
                sq.status = "failed"
                self._log("   → 未能获取有效信息")

        # Phase 5: 撰写报告
        state.phase = ResearchPhase.WRITING
        state.report = await self.write_report(query, state.findings)
        state.phase = ResearchPhase.COMPLETE

        self._log("\n✅ 研究完成！")
        return state.report

    async def close(self):
        await self.http.aclose()

    @staticmethod
    def _log(msg: str):
        print(msg)

# ─────────────────────────── 运行入口 ───────────────────────────

async def main():
    agent = DeepResearchAgent(
        max_iterations=2,
    )
    try:
        report = await agent.research(
            "2025年大语言模型在代码生成领域的最新进展和主要挑战是什么？"
        )
        print("\n" + "=" * 60)
        print(report)

        # 保存报告
        with open("research_report.md", "w", encoding="utf-8") as f:
            f.write(report)
        print("\n📄 报告已保存至 research_report.md")
    finally:
        await agent.close()

if __name__ == "__main__":
    asyncio.run(main())