Deep Research Agent 深度研究智能体

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

deep_research/
├── main.py          # 主入口，DeepResearchAgent 高层API + CLI
├── orchestrator.py  # 协调者（Lead Agent），核心调度逻辑
├── workers.py       # 工作者（Sub-Agents），执行搜索与分析
├── tools.py         # LLM客户端 + 搜索API客户端 + 工具函数
├── prompts.py       # 所有Prompt模板（7大类，30+个模板）
├── config.py        # 全局配置（API、迭代参数、质量阈值等）
├── api.py           # FastAPI REST服务接口
├── __init__.py      # 包初始化
├── __main__.py      # CLI入口
└── requirements.txt # 依赖包

 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

# orchestrator.py - 核心调度流程（简化版）
def run(self, task: str) -> str:
    # 1. 分解任务，确定子任务列表
    task_plan = self._decompose_task(task)

    # 2. 并行派发Worker（使用线程池）
    with ThreadPoolExecutor(max_workers=task_plan.worker_count) as executor:
        futures = [executor.submit(worker.execute, subtask)
                   for subtask in task_plan.subtasks]
        results = [f.result() for f in futures]

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

    # 4. 迭代质量控制（最少2次，最多5次）
    for iteration in range(self.config.max_iterations):
        score = self._quality_check(report)
        if score >= 80 and iteration >= 2:
            break
        if self._diminishing_returns_detected():
            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
4
5
6
7
8

查询："GPT-4的工作原理是什么？"
→ simple（单一问题，1-2个Worker即可）

查询："分析特斯拉的商业模式、竞争优势、财务状况和未来风险"
→ breadth_first（4个独立子话题，各派1个Worker）

查询："深度分析气候变化对农业的影响"
→ depth_first（同一主题，从经济/生态/政策/技术等多维度深挖）

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

┌─────────────────────────────────────────────────────┐
│                   OODA 一个循环                      │
│                                                     │
│  Observe  →  执行搜索查询，收集原始信息               │
│     ↓                                               │
│  Orient   →  分析结果，识别关键发现和信息缺口          │
│     ↓                                               │
│  Decide   →  判断是否需要补充搜索，选择下一步策略       │
│     ↓                                               │
│  Act      →  执行补充搜索 OR 结束本轮，提交结果        │
│     ↓                                               │
│  （若预算未耗尽且有缺口，进入下一个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

# workers.py - OODA循环（简化版）
def _execute_ooda_cycle(self, subtask, budget):
    cycle = OODACycle()

    # Observe: 执行初始搜索
    results = self.search_client.multi_search(subtask.search_queries)
    cycle.observations = results

    # Orient: 分析结果，识别缺口
    analysis = self.llm.chat(ORIENT_PROMPT.format(results=results))
    gaps = self._extract_gaps(analysis)
    cycle.orientation = analysis

    # Decide: 是否需要补充搜索？
    if gaps and budget.remaining_queries > 0:
        decision = "search_more"
    else:
        decision = "complete"
    cycle.decision = decision

    # Act: 执行决策
    if decision == "search_more":
        extra_results = self.search_client.multi_search(gaps)
        results.extend(extra_results)
        budget.consume(len(gaps))

    return cycle, results

1
2
3
4
5

达到质量阈值（≥80分）AND 已完成最少迭代（≥2次）
    OR
检测到边际收益递减（连续3次迭代质量提升 < 5分）
    OR
达到最大迭代次数（5次）

1
2
3
4
5
6
7
8

# 当前方案（同步线程）
with ThreadPoolExecutor(max_workers=n) as executor:
    futures = [executor.submit(worker.execute, task) for task in subtasks]

# 优化方案（异步）
async def run_workers_async(subtasks):
    tasks = [worker.execute_async(task) for task in subtasks]
    return await asyncio.gather(*tasks)

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

# 在 tools.py 中添加 Redis 缓存
import hashlib, redis

cache = redis.Redis()

def search_with_cache(query: str):
    key = hashlib.md5(query.encode()).hexdigest()
    cached = cache.get(key)
    if cached:
        return json.loads(cached)
    result = search_client.search(query)
    cache.setex(key, 3600, json.dumps(result))  # 缓存1小时
    return result

1
2
3
4
5

pip install -r requirements.txt

# 创建 .env 文件
DASHSCOPE_API_KEY=your_dashscope_key
BOCHA_API_KEY=your_bocha_key

1
2
3
4
5

from deep_research import DeepResearchAgent

agent = DeepResearchAgent()
report = agent.research("分析2024年中国新能源汽车市场格局")
print(report)

1

python -m deep_research "分析2024年中国新能源汽车市场格局" --workers 6 --output ./reports

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

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

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

# 查询结果
curl http://localhost:8000/research/{task_id}/result