案例：多步骤深度研究 Agent

第七十一章：案例：多步骤深度研究 Agent

章节导语

信息时代最大的悖论是：数据越来越多，而真正的洞察越来越难得。一份高质量的竞品分析报告，过去需要分析师花费数天收集、阅读、归纳、交叉验证；而今天，这项工作可以由一个 Hermes Agent 在数十分钟内完成——它能自主搜索学术论文、新闻报道、技术文档，逐页阅读，提炼关键信息，追踪引用来源，最终输出一份结构完整、引用可追溯的研究报告。本章将从零构建这个多步骤深度研究 Agent，重点讲解它如何在保证质量的前提下自动化整个研究流程。

71.1 需求分析：自动化研究报告的核心挑战

研究报告的五大难点

手工研究流程的时间分布：
搜索相关资料          ██████ 30%
阅读筛选内容          ████████ 40%
交叉验证事实          ████ 20%
整理撰写报告          ██ 10%

自动化研究面临的核心挑战：

Agent 的目标功能

• 输入：研究主题 + 可选的深度/广度参数
• 输出：结构化 Markdown 研究报告，含引用列表
• 过程：搜索 → 筛选 → 阅读 → 提炼 → 综合 → 写报告
• 质量保证：引用追踪 + 相互印证 + 置信度标注

71.2 系统架构

研究流程状态机

┌─────────────────────────────────────────────────────────┐
│                   Research Agent Pipeline                │
│                                                         │
│  [INIT] 解析研究主题，制定搜索策略                       │
│     ↓                                                   │
│  [SEARCH] 多维度搜索（Tavily/SerpAPI）                  │
│     ↓                                                   │
│  [FILTER] 评估来源可靠性，过滤低质量结果                 │
│     ↓                                                   │
│  [READ] 深度阅读高价值页面（获取全文）                   │
│     ↓                                                   │
│  [EXTRACT] 从每个来源提炼关键信息                        │
│     ↓                                                   │
│  [VERIFY] 交叉核验：寻找印证或矛盾                       │
│     ↓                                                   │
│  [SYNTHESIZE] 综合所有信息，识别主要论点                 │
│     ↓                                                   │
│  [WRITE] 生成结构化研究报告                             │
│     ↓                                                   │
│  [QUALITY_CHECK] 验证引用完整性，标注置信度              │
└─────────────────────────────────────────────────────────┘

架构组件

┌─────────────────────────────────────────────────────────┐
│                  Deep Research Agent                     │
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │               Hermes LLM Core                   │   │
│  │  - 研究策略规划                                  │   │
│  │  - 信息提炼与综合                                │   │
│  │  - 报告撰写                                      │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
│  ┌──────────────┐  ┌──────────────┐  ┌─────────────┐  │
│  │  搜索工具    │  │  阅读工具    │  │  写作工具   │  │
│  │              │  │              │  │             │  │
│  │ - Tavily API │  │ - URL Fetch  │  │ - Markdown  │  │
│  │ - SerpAPI    │  │ - PDF解析    │  │ - 引用格式化│  │
│  │ - arXiv API  │  │ - 内容清洗   │  │ - 目录生成  │  │
│  └──────────────┘  └──────────────┘  └─────────────┘  │
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │           Research Memory（研究记忆）            │   │
│  │  - 已访问 URL 列表（防重复）                     │   │
│  │  - 提炼的关键事实库                              │   │
│  │  - 引用追踪表                                    │   │
│  └─────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────┘

工具集设计

71.3 完整实现代码

核心 Agent

# research_agent/agent.py
import os
import json
from datetime import datetime
from openai import OpenAI

client = OpenAI(
    base_url=os.getenv("HERMES_BASE_URL", "http://localhost:11434/v1"),
    api_key=os.getenv("HERMES_API_KEY", "ollama"),
)
MODEL = os.getenv("HERMES_MODEL", "nous-hermes-2-mixtral-8x7b-dpo")

SYSTEM_PROMPT = """你是一位专业的研究分析师，擅长从海量信息中提炼洞察，撰写高质量研究报告。

你的工作方法：
1. **广泛搜索**：从多个角度搜索，不只是表面问题
2. **深度阅读**：对关键来源进行全文阅读，而非仅看摘要
3. **交叉核验**：对重要论断，寻找至少2个独立来源印证
4. **引用追踪**：每个数据点都记录来源，不允许无来源的论断
5. **结构化输出**：报告必须有清晰的层级结构

置信度标注规则：
- 🟢 高置信度：2+ 个权威来源印证
- 🟡 中置信度：1个可靠来源，未能交叉验证
- 🔴 低置信度：仅有推测或单一来源，需标注

报告格式：
- 执行摘要（300字内）
- 主要发现（3-5个）
- 详细分析（分章节）
- 结论与建议
- 参考文献"""

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "使用 Tavily 搜索引擎搜索网络内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索查询词"},
                    "max_results": {"type": "integer", "default": 5, "description": "最多返回结果数"},
                    "include_domains": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "优先搜索的域名，如 ['arxiv.org', 'nature.com']"
                    },
                    "search_depth": {
                        "type": "string",
                        "enum": ["basic", "advanced"],
                        "default": "advanced"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_academic",
            "description": "搜索学术论文和研究报告",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "year_from": {"type": "integer", "description": "发表年份下限"},
                    "max_results": {"type": "integer", "default": 5}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "fetch_page_content",
            "description": "获取指定 URL 的页面全文内容",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "目标页面 URL"},
                    "max_chars": {"type": "integer", "default": 8000, "description": "最大字符数"}
                },
                "required": ["url"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "record_fact",
            "description": "记录一个已核验的事实到研究记忆库",
            "parameters": {
                "type": "object",
                "properties": {
                    "fact": {"type": "string", "description": "事实陈述"},
                    "source_url": {"type": "string"},
                    "source_title": {"type": "string"},
                    "confidence": {
                        "type": "string",
                        "enum": ["high", "medium", "low"]
                    },
                    "category": {"type": "string", "description": "事实分类，如 '市场规模'/'技术现状'"}
                },
                "required": ["fact", "source_url", "confidence", "category"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_recorded_facts",
            "description": "获取已记录的所有研究事实",
            "parameters": {
                "type": "object",
                "properties": {
                    "category": {"type": "string", "description": "按分类过滤，空则返回全部"}
                }
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write_final_report",
            "description": "将所有收集的信息整合成最终研究报告",
            "parameters": {
                "type": "object",
                "properties": {
                    "topic": {"type": "string"},
                    "facts": {"type": "array"},
                    "report_structure": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "报告章节列表"
                    }
                },
                "required": ["topic", "facts"]
            }
        }
    }
]


class ResearchMemory:
    """研究记忆：存储已访问URL和已提炼事实"""
    
    def __init__(self):
        self.visited_urls: set = set()
        self.facts: list = []
        self.citations: dict = {}  # url -> {title, date, content}
    
    def add_fact(self, fact: str, source_url: str, source_title: str,
                 confidence: str, category: str):
        self.facts.append({
            "id": len(self.facts) + 1,
            "fact": fact,
            "source_url": source_url,
            "source_title": source_title,
            "confidence": confidence,
            "category": category
        })
        return {"success": True, "fact_id": len(self.facts)}
    
    def get_facts(self, category: str = None) -> list:
        if category:
            return [f for f in self.facts if f["category"] == category]
        return self.facts
    
    def mark_visited(self, url: str):
        self.visited_urls.add(url)
    
    def is_visited(self, url: str) -> bool:
        return url in self.visited_urls


def run_research_agent(topic: str, depth: str = "comprehensive") -> dict:
    """运行深度研究 Agent"""
    memory = ResearchMemory()
    
    print(f"[研究Agent] 开始研究主题：{topic}")
    print(f"[研究Agent] 研究深度：{depth}")
    
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {
            "role": "user",
            "content": f"""请对以下主题进行深度研究，并输出完整的研究报告：

**研究主题：** {topic}
**研究深度：** {depth}（comprehensive=全面综合，focused=聚焦分析）
**当前时间：** {datetime.now().strftime('%Y年%m月')}

研究步骤：
1. 制定搜索策略（至少3个不同角度的搜索词）
2. 执行多轮搜索，收集来自不同来源的信息
3. 深度阅读最相关的5-10个来源的全文
4. 记录关键事实（使用 record_fact 工具）
5. 对重要数据进行交叉验证
6. 调用 write_final_report 生成报告

请开始研究。"""
        }
    ]
    
    max_iterations = 30  # 研究 Agent 需要更多迭代
    
    for iteration in range(max_iterations):
        print(f"[研究Agent] 第 {iteration + 1} 轮推理...")
        
        response = client.chat.completions.create(
            model=MODEL,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
            temperature=0.3,  # 研究需要适度创造性
            max_tokens=4000
        )
        
        message = response.choices[0].message
        messages.append(message)
        
        if not message.tool_calls:
            print("[研究Agent] 研究完成")
            return {
                "status": "completed",
                "report": message.content,
                "facts_collected": len(memory.facts),
                "sources_visited": len(memory.visited_urls),
                "iterations": iteration + 1
            }
        
        # 执行工具调用
        for tool_call in message.tool_calls:
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            print(f"  → 工具调用: {name}")
            
            result = _dispatch_tool(name, args, memory)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })
    
    return {"status": "max_iterations", "facts_collected": len(memory.facts)}


def _dispatch_tool(name: str, args: dict, memory: ResearchMemory) -> dict:
    from .tools import search_tools, content_tools
    
    if name == "search_web":
        return search_tools.search_tavily(**args)
    elif name == "search_academic":
        return search_tools.search_academic(**args)
    elif name == "fetch_page_content":
        url = args["url"]
        if memory.is_visited(url):
            return {"cached": True, "message": "该URL已访问过"}
        memory.mark_visited(url)
        return content_tools.fetch_page(url, args.get("max_chars", 8000))
    elif name == "record_fact":
        return memory.add_fact(**args)
    elif name == "get_recorded_facts":
        return {"facts": memory.get_facts(args.get("category"))}
    elif name == "write_final_report":
        return content_tools.write_report(
            args["topic"], 
            memory.get_facts(),
            args.get("report_structure", [])
        )
    return {"error": f"未知工具: {name}"}

搜索工具实现

# research_agent/tools/search_tools.py
import os
import requests
from typing import List, Optional

TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")

def search_tavily(
    query: str,
    max_results: int = 5,
    include_domains: List[str] = None,
    search_depth: str = "advanced"
) -> dict:
    """Tavily 搜索 API"""
    payload = {
        "api_key": TAVILY_API_KEY,
        "query": query,
        "max_results": max_results,
        "search_depth": search_depth,
        "include_answer": True,   # 获取 AI 摘要
        "include_raw_content": False,
    }
    
    if include_domains:
        payload["include_domains"] = include_domains
    
    resp = requests.post(
        "https://api.tavily.com/search",
        json=payload,
        timeout=30
    )
    resp.raise_for_status()
    data = resp.json()
    
    results = []
    for r in data.get("results", []):
        results.append({
            "title": r["title"],
            "url": r["url"],
            "snippet": r.get("content", "")[:500],
            "score": r.get("score", 0),
            "published_date": r.get("published_date", "")
        })
    
    return {
        "query": query,
        "total": len(results),
        "ai_answer": data.get("answer", ""),
        "results": results
    }


def search_academic(
    query: str,
    year_from: int = None,
    max_results: int = 5
) -> dict:
    """搜索 Semantic Scholar 学术数据库"""
    params = {
        "query": query,
        "limit": max_results,
        "fields": "title,abstract,year,authors,externalIds,citationCount,url"
    }
    if year_from:
        params["year"] = f"{year_from}-"
    
    resp = requests.get(
        "https://api.semanticscholar.org/graph/v1/paper/search",
        params=params,
        timeout=30
    )
    
    if resp.status_code != 200:
        return {"error": f"API错误: {resp.status_code}", "results": []}
    
    papers = []
    for p in resp.json().get("data", []):
        papers.append({
            "title": p.get("title", ""),
            "abstract": p.get("abstract", "")[:400],
            "year": p.get("year"),
            "citation_count": p.get("citationCount", 0),
            "url": p.get("url", ""),
            "authors": [a["name"] for a in p.get("authors", [])[:3]]
        })
    
    # 按引用数排序（引用多的通常更重要）
    papers.sort(key=lambda x: x["citation_count"], reverse=True)
    
    return {"query": query, "total": len(papers), "papers": papers}

内容处理工具

# research_agent/tools/content_tools.py
import re
import requests
from bs4 import BeautifulSoup
from typing import List

def fetch_page(url: str, max_chars: int = 8000) -> dict:
    """获取页面全文内容"""
    headers = {
        "User-Agent": "Mozilla/5.0 (Research Bot; +https://example.com/bot)"
    }
    
    try:
        resp = requests.get(url, headers=headers, timeout=15)
        resp.raise_for_status()
        
        soup = BeautifulSoup(resp.text, "html.parser")
        
        # 移除无用标签
        for tag in soup(["script", "style", "nav", "footer", "aside", "iframe"]):
            tag.decompose()
        
        # 优先提取文章主体
        article = (
            soup.find("article") or 
            soup.find("main") or 
            soup.find(class_=re.compile(r"(content|article|post|entry)")) or
            soup.find("body")
        )
        
        text = article.get_text(separator="\n", strip=True) if article else ""
        
        # 清理多余空行
        text = re.sub(r"\n{3,}", "\n\n", text)
        
        return {
            "url": url,
            "title": soup.title.string if soup.title else "",
            "content": text[:max_chars],
            "total_chars": len(text),
            "truncated": len(text) > max_chars
        }
    
    except requests.RequestException as e:
        return {"url": url, "error": str(e), "content": ""}


def write_report(topic: str, facts: List[dict], structure: List[str]) -> dict:
    """生成最终研究报告的模板"""
    if not structure:
        structure = [
            "执行摘要",
            "研究背景",
            "主要发现",
            "详细分析",
            "结论与建议",
            "参考文献"
        ]
    
    # 按类别整理事实
    facts_by_category = {}
    for fact in facts:
        cat = fact.get("category", "其他")
        facts_by_category.setdefault(cat, []).append(fact)
    
    # 生成引用列表
    citations = []
    seen_urls = set()
    for fact in facts:
        url = fact.get("source_url", "")
        if url and url not in seen_urls:
            seen_urls.add(url)
            citations.append({
                "title": fact.get("source_title", url),
                "url": url
            })
    
    return {
        "topic": topic,
        "total_facts": len(facts),
        "total_sources": len(citations),
        "facts_by_category": facts_by_category,
        "suggested_structure": structure,
        "citations": citations,
        "ready_for_writing": True
    }

71.4 质量控制机制

引用追踪系统

class CitationTracker:
    """引用追踪：确保每个论断都有来源"""
    
    def __init__(self):
        self._citations = {}  # id -> citation
        self._counter = 0
    
    def add(self, url: str, title: str, accessed_date: str = None) -> int:
        """添加引用，返回引用编号"""
        self._counter += 1
        self._citations[self._counter] = {
            "id": self._counter,
            "url": url,
            "title": title,
            "accessed": accessed_date or "2026-04"
        }
        return self._counter
    
    def format_bibliography(self, style: str = "apa") -> str:
        """格式化参考文献列表"""
        lines = ["## 参考文献\n"]
        for cid, c in self._citations.items():
            if style == "apa":
                lines.append(f"[{cid}] {c['title']}. 检索自 {c['url']} ({c['accessed']})")
            else:
                lines.append(f"[{cid}] [{c['title']}]({c['url']})")
        return "\n".join(lines)
    
    def inject_citations(self, text: str) -> str:
        """在报告文本中注入引用标记"""
        # 查找类似 [cite:url] 的占位符，替换为编号引用
        import re
        def replace_cite(match):
            url = match.group(1)
            for cid, c in self._citations.items():
                if c["url"] == url:
                    return f"[{cid}]"
            return "[?]"  # 未知来源
        return re.sub(r"\[cite:(https?://[^\]]+)\]", replace_cite, text)

事实核查流程

def cross_verify_fact(
    claim: str, 
    supporting_sources: list, 
    contradicting_sources: list
) -> dict:
    """对一个论断进行置信度评估"""
    
    confidence_score = 0
    
    # 权威来源加权
    authoritative_domains = [
        "nature.com", "science.org", "arxiv.org", 
        "gov", ".edu", "reuters.com", "bloomberg.com"
    ]
    
    for source in supporting_sources:
        url = source.get("url", "")
        # 权威来源得分更高
        is_authoritative = any(d in url for d in authoritative_domains)
        confidence_score += 2 if is_authoritative else 1
    
    for source in contradicting_sources:
        confidence_score -= 1
    
    if confidence_score >= 3:
        level = "high"
        emoji = "🟢"
    elif confidence_score >= 1:
        level = "medium"
        emoji = "🟡"
    else:
        level = "low"
        emoji = "🔴"
    
    return {
        "claim": claim,
        "confidence": level,
        "confidence_emoji": emoji,
        "score": confidence_score,
        "supporting_count": len(supporting_sources),
        "contradicting_count": len(contradicting_sources),
        "recommendation": (
            "可直接引用" if level == "high" else
            "需标注来源有限" if level == "medium" else
            "建议进一步核实或标注为推测"
        )
    }

71.5 报告输出示例

# AI 大模型市场竞争格局研究报告

**研究日期：** 2026年4月
**数据来源：** 15个网络来源，4篇学术论文
**研究置信度：** 综合评分 82/100

---

## 执行摘要

大型语言模型市场在2025年经历了快速整合，从数十家竞争者收缩至形成三大阵营：
闭源商业模型（OpenAI/Anthropic/Google）、开源社区模型（Meta/MistralAI/NousResearch），
以及专业垂直领域模型。🟢 市场总规模预计2026年达到 1,200亿美元 [1][2]。

## 主要发现

1. **开源追赶效应明显** 🟢：开源模型在代码生成、指令遵循等基准测试中
   已接近或超过2024年初的商业模型水平 [3]

2. **推理成本持续下降** 🟡：主流模型推理成本年均下降约60%，
   部分厂商单独发布"轻量推理版本"以应对成本竞争 [4]

3. **Agent 能力成为新战场** 🟢：2025年下半年，各主要厂商均推出
   针对 Agent 场景优化的模型版本 [5][6]

---

## 参考文献

[1] Goldman Sachs AI Report 2026. https://gs.com/ai-report-2026
[2] Gartner Emerging Tech Hype Cycle 2025. https://gartner.com/...
...

71.6 耗时与成本分析

不同研究深度的资源消耗对比

注：成本基于 Hermes 自托管推理场景，商业 API 成本约为 5-10 倍

成本优化策略

class CostOptimizer:
    """研究成本优化器"""
    
    def __init__(self, budget_usd: float = 1.0):
        self.budget = budget_usd
        self.spent = 0.0
        self.token_cost_per_1k = 0.002  # 自托管估算
    
    def should_continue(self, facts_count: int, min_facts: int = 10) -> bool:
        """判断是否应继续研究"""
        # 已超预算：停止
        if self.spent >= self.budget:
            return False
        # 事实数量已足够：可以停止
        if facts_count >= min_facts and self.spent >= self.budget * 0.5:
            return False
        return True
    
    def should_deep_read(self, relevance_score: float, threshold: float = 0.7) -> bool:
        """判断是否值得深度阅读该页面"""
        return relevance_score >= threshold and self.spent < self.budget * 0.8
    
    def record_tokens(self, input_tokens: int, output_tokens: int):
        cost = (input_tokens + output_tokens) / 1000 * self.token_cost_per_1k
        self.spent += cost

本章小结

本章构建了一个具备生产可用性的深度研究 Agent：

• 流程设计：搜索→筛选→阅读→提炼→核验→综合→写报告的完整链路
• 质量机制：引用追踪系统 + 置信度评分 + 交叉验证
• 工具栈：Tavily 搜索 + Semantic Scholar + BeautifulSoup 内容提取
• 成本控制：按研究深度分级，内置预算控制器

这个 Agent 的核心价值在于将研究员的方法论（系统搜索、严格核验、引用完整）编码为 Agent 的行为模式，而不仅仅是"让 AI 回答问题"。

思考题

1. 如何检测并处理研究中发现的相互矛盾的事实？
2. 研究 Agent 如何避免"确认偏误"——只收集支持某个预设结论的信息？
3. 对于中文研究主题，如何平衡中英文来源的比例和权威性？
4. 如何评估一份 AI 生成研究报告的质量？可以设计哪些自动化质量指标？