Deep Research Agent 完整学习教程

1
2
3
4

TABLE WITHOUT ID
  file.link as "章节",
  tags as "标签"
WHERE contains(file.name, "Deep_Research")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# orchestrator.py 核心流程（简化版）

class Orchestrator:
    def run(self, task: str) -> ResearchState:
        # 步骤1：任务分解
        task_plan = self.decompose_task(task)

        # 步骤2：并行派发Worker
        with ThreadPoolExecutor(max_workers=task_plan.worker_count) as executor:
            # 提交所有任务到线程池
            futures = []
            for subtask in task_plan.subtasks:
                worker = self.worker_factory(subtask.id)
                future = executor.submit(worker.execute, subtask)
                futures.append(future)

            # 等待所有Worker完成
            results = []
            for future in as_completed(futures):
                result = future.result()
                results.append(result)

        # 步骤3：合成报告
        report = self._synthesize_results(task, results)

        # 步骤4：质量控制循环
        for iteration in range(self.max_iterations):
            score = self._quality_check(report)
            if score >= 80 and iteration >= 2:
                break  # 质量达标且满足最少迭代次数

            # 补充研究
            gaps = self._gap_analysis(report)
            new_results = self._dispatch_supplementary_workers(gaps)
            report = self._refine_report(report, new_results)

        return report

1
2
3

❓ "什么是GPT-4？"
❓ "Python的最新版本是多少？"
❓ "比特币的当前价格"

1
2
3
4
5
6

❓ "深度分析特斯拉的商业模式"
   → Worker 1: 收入模式分析
   → Worker 2: 成本结构分析
   → Worker 3: 竞争优势分析
   → Worker 4: 风险因素分析
   → Worker 5: 未来战略分析

1
2
3
4
5
6

❓ "2024年科技行业发展趋势"
   → Worker 1: AI领域趋势
   → Worker 2: 云计算趋势
   → Worker 3: 物联网趋势
   → Worker 4: 区块链趋势
   → Worker 5: 量子计算趋势

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# orchestrator.py - 查询类型判断

def _classify_query_type(self, task: str) -> QueryTypeAnalysis:
    """判断查询类型"""

    prompt = f"""
    分析以下研究任务，判断其查询类型：

    任务：{task}

    请判断属于以下哪种类型：
    1. simple: 简单直接的查询
    2. depth_first: 需要对单一主题深入分析
    3. breadth_first: 需要覆盖多个独立子主题

    返回JSON格式：
    {{
        "query_type": "depth_first",
        "reasoning": "需要从多个维度深入分析特斯拉...",
        "recommended_worker_count": 5
    }}
    """

    response = self.llm.chat(prompt)
    result = extract_json(response)

    return QueryTypeAnalysis(**result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

┌─────────────────────────────────────────────┐
│           OODA 循环 (一次迭代)               │
├─────────────────────────────────────────────┤
│                                             │
│  🔍 Observe (观察)                          │
│     ↓                                       │
│     执行搜索查询，收集原始信息                │
│     • 搜索关键词                             │
│     • 获取搜索结果                           │
│     • 记录信息来源                           │
│                                             │
│  🧭 Orient (定向)                           │
│     ↓                                       │
│     分析结果，识别关键发现和信息缺口           │
│     • 提取关键信息                           │
│     • 评估来源质量                           │
│     • 识别信息缺口                           │
│                                             │
│  🤔 Decide (决策)                           │
│     ↓                                       │
│     判断是否需要补充搜索                      │
│     • 信息是否充分？                         │
│     • 预算是否充足？                         │
│     • 是否需要补充？                         │
│                                             │
│  ⚡ Act (行动)                              │
│     ↓                                       │
│     执行决策：补充搜索 OR 结束               │
│     • 如需补充：生成新查询                   │
│     • 如已充分：整理结果                     │
│                                             │
│  ↻ 循环直到满足终止条件                      │
└─────────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# config.py - 研究预算配置

AGENT_CONFIG = {
    "research_budget": {
        "simple": {
            "max_queries": 3,      # 最多3次搜索
            "max_cycles": 1        # 最多1次OODA循环
        },
        "medium": {
            "max_queries": 5,      # 最多5次搜索
            "max_cycles": 2        # 最多2次OODA循环
        },
        "complex": {
            "max_queries": 8,      # 最多8次搜索
            "max_cycles": 3        # 最多3次OODA循环
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

# workers.py - OODA循环实现（简化版）

class SearchWorker:
    def _execute_ooda_cycle(self, subtask, budget: ResearchBudget):
        """执行一次OODA循环"""

        cycle_record = OODACycle(cycle=budget.cycles_used + 1)

        # ===== Observe: 观察 =====
        search_results = self.search_client.multi_search(
            subtask.search_queries
        )
        cycle_record.observe = f"执行了{len(subtask.search_queries)}次搜索"
        budget.queries_used += len(subtask.search_queries)

        # ===== Orient: 定向 =====
        analysis_prompt = f"""
        分析以下搜索结果：
        {format_search_context(search_results)}

        请回答：
        1. 关键发现是什么？
        2. 还有哪些信息缺口？
        3. 来源质量如何？
        """

        analysis = self.llm.chat(analysis_prompt)
        cycle_record.orient = analysis

        # 提取信息缺口
        gaps = self._extract_information_gaps(analysis)

        # ===== Decide: 决策 =====
        if gaps and budget.queries_remaining > 0:
            decision = "continue_search"
            cycle_record.decide = f"发现{len(gaps)}个信息缺口，继续搜索"
        else:
            decision = "complete"
            cycle_record.decide = "信息已充分或预算耗尽，结束搜索"

        # ===== Act: 行动 =====
        if decision == "continue_search":
            # 生成补充查询
            supplementary_queries = self._generate_queries_for_gaps(gaps)
            cycle_record.act = f"生成{len(supplementary_queries)}个补充查询"

            # 执行补充搜索
            extra_results = self.search_client.multi_search(
                supplementary_queries
            )
            search_results.extend(extra_results)
            budget.queries_used += len(supplementary_queries)
        else:
            cycle_record.act = "整理结果，准备输出"

        self.ooda_cycles.append(cycle_record)
        return search_results, decision

1
2
3
4
5
6
7
8
9

迭代历史：
第1次: 72分
第2次: 78分 (+6分) ✅ 提升明显
第3次: 81分 (+3分) ⚠️ 提升变小
第4次: 82分 (+1分) ⚠️ 提升很小
第5次: 83分 (+1分) ⚠️ 提升很小

判断：连续3次(第3-5次)提升都<5分
结果：🛑 触发边际收益递减，提前终止

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

# orchestrator.py - 迭代优化（简化版）

def _iterative_refinement(self, state: ResearchState) -> ResearchState:
    """迭代优化报告质量"""

    current_report = state.final_report

    for iteration in range(1, self.max_iterations + 1):
        self._emit_progress("iteration_start", {
            "iteration": iteration,
            "max_iterations": self.max_iterations
        })

        # 步骤1：质量检查
        quality_result = self._quality_check(current_report, state.original_task)
        score = quality_result["score"]

        self._emit_progress("iteration_evaluated", {
            "iteration": iteration,
            "score": score,
            "threshold": self.quality_threshold
        })

        # 步骤2：判断是否达标
        if score >= self.quality_threshold and iteration >= self.min_iterations:
            self._emit_progress("quality_passed", {"score": score})
            break

        # 步骤3：检查边际收益递减
        if self._check_diminishing_returns(state.iteration_history):
            self._emit_progress("diminishing_returns_detected", {})
            break

        # 步骤4：差距分析
        gap_analysis = self._gap_analysis(
            current_report,
            state.original_task,
            quality_result
        )

        # 步骤5：派发补充Workers
        supplementary_results = self._dispatch_supplementary_workers(
            gap_analysis
        )

        # 步骤6：精炼报告
        current_report = self._refine_report(
            current_report,
            gap_analysis,
            supplementary_results
        )

        # 记录本次迭代
        state.iteration_history.append(IterationRecord(
            iteration=iteration,
            report=current_report,
            quality_score=score,
            quality_result=quality_result,
            gap_analysis=gap_analysis,
            supplementary_results=supplementary_results
        ))

    state.final_report = current_report
    state.quality_score = score
    state.iteration_count = iteration

    return state

def _check_diminishing_returns(self, history: list) -> bool:
    """检查边际收益递减"""
    if len(history) < 3:
        return False

    # 检查最近3次迭代的提升
    recent_improvements = []
    for i in range(-3, 0):
        improvement = history[i].quality_score - history[i-1].quality_score
        recent_improvements.append(improvement)

    # 如果连续3次提升都<5分，触发终止
    return all(imp < 5 for imp in recent_improvements)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# config.py - 来源质量配置

"source_quality": {
    # 高质量域名
    "high_quality_domains": [
        ".gov",           # 政府网站
        ".edu",           # 教育机构
        ".org",           # 非营利组织
        "reuters",        # 路透社
        "bloomberg",      # 彭博社
        "nature",         # 自然杂志
        "arxiv",          # 学术预印本
    ],

    # 低质量指标
    "low_quality_indicators": [
        "forum", "reddit", "quora", "blog", "medium.com"
    ],

    # 推测性语言
    "speculative_language": [
        "可能", "也许", "据说", "传闻", "预计",
        "may", "might", "possibly", "reportedly"
    ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

deep_research/
├── 📄 main.py              # 主入口，提供高层API
├── 🎯 orchestrator.py      # 协调者，核心调度逻辑
├── 👷 workers.py           # 工作者，执行具体任务
├── 🔧 tools.py             # 工具层：LLM客户端、搜索客户端
├── 💬 prompts.py           # 所有Prompt模板（30+个）
├── ⚙️ config.py            # 全局配置
├── 🌐 api.py               # FastAPI REST服务
├── 📦 __init__.py          # 包初始化
├── 🚀 __main__.py          # CLI入口
└── 📋 requirements.txt     # 依赖清单

1
2
3
4
5
6
7
8

class DeepResearchAgent:
    """用户友好的高层API"""

    def research(self, task: str) -> str:
        """执行研究，返回报告"""

    def research_stream(self, task: str):
        """流式执行，生成器模式"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

class Orchestrator:
    def run(self, task: str) -> ResearchState:
        """主流程"""

    def decompose_task(self, task: str) -> TaskPlan:
        """任务分解"""

    def _dispatch_workers(self, task_plan: TaskPlan) -> list:
        """派发Workers"""

    def _synthesize_results(self, task: str, results: list) -> str:
        """合成报告"""

    def _iterative_refinement(self, state: ResearchState):
        """迭代优化"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

class SearchWorker(BaseWorker):
    """搜索研究Worker"""

    def execute(self, subtask) -> dict:
        """执行任务"""

    def _execute_ooda_cycle(self, subtask, budget):
        """执行OODA循环"""

    def _assess_source_quality(self, results: list):
        """评估来源质量"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

class LLMClient:
    """LLM客户端"""
    def chat(self, messages, model=None) -> str:
        """对话接口"""

class SearchClient:
    """搜索客户端"""
    def search(self, query: str) -> list:
        """单次搜索"""

    def multi_search(self, queries: list) -> list:
        """批量搜索"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

OODA循环1:
  Observe: 搜索"2024中国电动汽车市场规模"
  Orient: 发现销量数据，但缺少增长率
  Decide: 需要补充搜索
  Act: 搜索"2024电动汽车同比增长"

OODA循环2:
  Observe: 获得增长率数据
  Orient: 信息已充分，来源可靠
  Decide: 完成
  Act: 整理结果

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 2024年中国电动汽车市场分析

## 执行摘要
2024年中国电动汽车市场继续保持高速增长...

## 市场规模与增长趋势
- 销量：XXX万辆
- 增长率：XX%
[来源: 中汽协, 2024]

## 主要厂商与市场份额
...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from deep_research import DeepResearchAgent

# 创建Agent
agent = DeepResearchAgent(
    verbose=True,        # 显示进度
    save_output=True,    # 保存输出
    output_dir="./reports"
)

# 执行研究
report = agent.research(
    task="分析2024年中国电动汽车市场",
    max_workers=5
)

print(report)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 基础用法
python -m deep_research "分析2024年中国电动汽车市场"

# 指定Worker数量
python -m deep_research "区块链技术应用" --workers 6

# 指定输出目录
python -m deep_research "新能源行业研究" --output ./my_reports

# 安静模式
python -m deep_research "医疗AI应用" --quiet

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 启动服务
uvicorn deep_research.api:app --port 8000

# 提交任务
curl -X POST http://localhost:8000/research \
  -H "Content-Type: application/json" \
  -d '{"task": "分析2024年中国电动汽车市场"}'

# 响应
{
  "task_id": "abc123",
  "status": "pending",
  "message": "任务已提交"
}

# 查询结果
curl http://localhost:8000/research/abc123/result

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 改为 asyncio + aiohttp
import asyncio
import aiohttp

async def dispatch_workers_async(self, subtasks):
    """异步派发Workers"""
    tasks = [
        worker.execute_async(subtask)
        for subtask in subtasks
    ]
    results = await asyncio.gather(*tasks)
    return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

import redis
import hashlib

cache = redis.Redis()

def search_with_cache(query: str):
    """带缓存的搜索"""
    # 生成缓存key
    key = f"search:{hashlib.md5(query.encode()).hexdigest()}"

    # 检查缓存
    cached = cache.get(key)
    if cached:
        return json.loads(cached)

    # 执行搜索
    result = search_client.search(query)

    # 缓存结果（1小时）
    cache.setex(key, 3600, json.dumps(result))

    return result

1
2
3
4
5
6
7
8

def select_model(task_complexity: str) -> str:
    """根据任务复杂度选择模型"""
    if task_complexity == "simple":
        return "qwen-turbo"    # 快速、便宜
    elif task_complexity == "medium":
        return "qwen-plus"     # 平衡
    else:
        return "qwen-max"      # 强大、准确

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from chromadb import Client

# 初始化向量数据库
chroma_client = Client()
collection = chroma_client.create_collection("search_results")

def semantic_dedup(results: list) -> list:
    """语义去重"""
    unique_results = []

    for result in results:
        # 查询相似内容
        similar = collection.query(
            query_texts=[result.content],
            n_results=1
        )

        # 如果相似度 < 0.9，认为是新内容
        if not similar or similar['distances'][0][0] > 0.1:
            unique_results.append(result)
            collection.add(
                documents=[result.content],
                ids=[result.id]
            )

    return unique_results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# api.py - SSE流式输出
@app.get("/research/{task_id}/stream")
async def stream_progress(task_id: str):
    """SSE流式进度"""
    async def event_generator():
        while True:
            # 获取最新进度
            progress = get_task_progress(task_id)

            # 发送事件
            yield {
                "event": "progress",
                "data": json.dumps(progress)
            }

            if progress["status"] == "completed":
                break

            await asyncio.sleep(1)

    return EventSourceResponse(event_generator())

1
2
3
4
5
6

# 克隆项目（如果有）
git clone <repository-url>
cd deep_research

# 安装依赖
pip install -r requirements.txt

1
2
3
4
5

# 阿里云通义千问 API Key
DASHSCOPE_API_KEY=sk-your-dashscope-key

# Bocha 搜索 API Key
BOCHA_API_KEY=sk-your-bocha-key

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from deep_research import DeepResearchAgent

# 创建Agent
agent = DeepResearchAgent()

# 执行研究
report = agent.research("什么是大语言模型？")

# 查看报告
print(report)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 方式1：修改 config.py
AGENT_CONFIG = {
    "max_workers": 8,           # 增加并行数
    "quality_threshold": 85,    # 提高质量要求
}

# 方式2：运行时指定
agent = DeepResearchAgent()
report = agent.research(
    task="你的任务",
    max_workers=8  # 覆盖默认配置
)