Deep Research Agent 测试题答案

1
2
3
4
5

执行搜索："特斯拉2024 Q4财报"
获得结果：
- 10条新闻报道
- 来源：CNBC, Reuters, 财经网站
- 内容：营收、利润、交付量等基本数据

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

关键发现：
- 营收：250亿美元（同比+15%）
- 净利润：30亿美元（同比-5%）
- 交付量：48万辆

信息缺口：
- 缺少详细的成本结构数据
- 缺少各业务线的收入拆分
- 缺少官方财报文件

来源质量：
- 中等（新闻报道，非官方文件）
- 需要更权威的来源

1
2
3
4
5

决定：需要补充搜索
原因：
1. 缺少成本结构等关键数据
2. 来源质量不够高（需要官方财报）
3. 预算充足（还剩6次查询）

1
2
3
4

生成补充查询：
- "Tesla Q4 2024 earnings report SEC filing"
- "特斯拉2024 Q4成本结构"
执行搜索

1
2
3
4
5

执行补充搜索
获得结果：
- Tesla官方投资者关系页面
- SEC 10-K文件
- 详细的财务数据表格

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

关键发现：
- 找到官方财报文件
- 获得完整的成本结构数据
- 各业务线收入详细拆分

信息缺口：
- 基本完整，无重大缺口

来源质量：
- 高（官方文件，SEC备案）

1
2
3
4
5

决定：信息已充分，可以结束
原因：
1. 关键信息已收集完整
2. 来源权威可靠
3. 满足研究目标

1
2
3

整理所有搜索结果
生成结构化的研究报告
返回给Orchestrator

1
2
3
4
5
6
7
8

场景1：第一次就达到82分
- 如果没有min_iterations：立即终止
- 有min_iterations=2：继续第2次迭代
- 第2次迭代后：可能提升到88分，质量更好

场景2：第一次只有75分
- 继续迭代，补充信息
- 第2次达到83分，满足条件，终止

1
2
3
4
5
6
7
8
9

迭代历史：
第1次: 70分 (基准)
第2次: 76分 (+6分) ✅ 提升明显
第3次: 79分 (+3分) ⚠️ 提升变小
第4次: 81分 (+2分) ⚠️ 提升很小
第5次: 82分 (+1分) ⚠️ 提升很小

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

1
2
3
4
5
6
7
8
9

迭代历史：
第1次: 70分
第2次: 76分 (+6分) ✅
第3次: 79分 (+3分) ⚠️
第4次: 85分 (+6分) ✅ 突然提升明显
第5次: 88分 (+3分) ⚠️

分析：第4次有明显提升，打破连续
结果：✅ 不触发，继续迭代

 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

# orchestrator.py - 改造前
def _dispatch_workers(self, subtasks):
    with ThreadPoolExecutor(max_workers=len(subtasks)) as executor:
        futures = [executor.submit(worker.execute, task)
                   for task in subtasks]
        results = [f.result() for f in as_completed(futures)]
    return results

# orchestrator.py - 改造后
async def _dispatch_workers(self, subtasks):
    tasks = [worker.execute_async(task) for task in subtasks]
    results = await asyncio.gather(*tasks)
    return results

# workers.py - 改造前
def execute(self, subtask):
    results = self.search_client.search(query)
    analysis = self.llm.chat(prompt)
    return result

# workers.py - 改造后
async def execute(self, subtask):
    results = await self.search_client.search_async(query)
    analysis = await self.llm.chat_async(prompt)
    return result

# tools.py - 改造前
def search(self, query):
    response = requests.post(url, json=data)
    return response.json()

# tools.py - 改造后
async def search_async(self, query):
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=data) as response:
            return await response.json()

1
2
3
4

场景：第一次就达到82分
- 无限制：立即终止，输出报告
- 问题：可能遗漏了重要信息
- 实际：第二次迭代可能发现缺口，提升到90分

1
2
3
4
5

场景：LLM评分波动
- 第一次评分：82分（偏高）
- 实际质量：75分
- 无限制：错误地认为达标
- 有限制：第二次评分可能修正为78分，继续优化

1
2
3
4

- 简单任务：1次迭代就结束
- 复杂任务：需要5次迭代
- 问题：用户无法预期系统行为
- 有限制：至少2次，行为更可预测

1
2
3

- 需要至少3次迭代才能判断趋势
- 如果第1次就终止，无法积累数据
- 无法启用智能终止机制

 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

# 1. 安装依赖
# pip install chromadb sentence-transformers

# 2. 初始化向量数据库
from chromadb import Client
from sentence_transformers import SentenceTransformer

class SemanticDeduplicator:
    def __init__(self, similarity_threshold=0.85):
        self.client = Client()
        self.collection = self.client.create_collection(
            name="search_results",
            metadata={"hnsw:space": "cosine"}
        )
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self.similarity_threshold = similarity_threshold

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

        for result in results:
            # 生成文本向量
            text = f"{result.title} {result.content}"

            # 查询相似内容
            similar = self.collection.query(
                query_texts=[text],
                n_results=1
            )

            # 判断是否重复
            if not similar['ids'] or similar['distances'][0][0] > (1 - self.similarity_threshold):
                # 不重复，添加到结果和数据库
                unique_results.append(result)
                self.collection.add(
                    documents=[text],
                    ids=[result.id],
                    metadatas=[{
                        "url": result.url,
                        "quality": result.quality_score
                    }]
                )
            else:
                # 重复，比较质量
                existing_quality = similar['metadatas'][0][0]['quality']
                if result.quality_score > existing_quality:
                    # 新结果质量更高，替换
                    self.collection.delete(ids=[similar['ids'][0][0]])
                    unique_results.append(result)
                    self.collection.add(
                        documents=[text],
                        ids=[result.id],
                        metadatas=[{
                            "url": result.url,
                            "quality": result.quality_score
                        }]
                    )

        return unique_results

# 3. 在 workers.py 中使用
class SearchWorker:
    def __init__(self):
        self.deduplicator = SemanticDeduplicator(
            similarity_threshold=0.85  # 85%相似度视为重复
        )

    def execute(self, subtask):
        # 执行搜索
        raw_results = self.search_client.multi_search(queries)

        # 语义去重
        unique_results = self.deduplicator.deduplicate(raw_results)

        # 继续处理
        return self._process_results(unique_results)

 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

用户任务
    ↓
[1] TASK_DECOMPOSITION_PROMPT
    ↓
子任务列表 + 查询类型
    ↓
[2] SEARCH_WORKER_TASK_PROMPT (并行)
    ↓
[3] OODA循环 Prompts
    - OBSERVE_PROMPT
    - ORIENT_PROMPT
    - DECIDE_PROMPT
    - ACT_PROMPT
    ↓
Worker结果集合
    ↓
[4] SYNTHESIS_PROMPT
    ↓
初步报告
    ↓
[5] QUALITY_CHECK_PROMPT
    ↓
质量评分 + 反馈
    ↓
如果未达标 ↓
[6] GAP_ANALYSIS_PROMPT
    ↓
信息缺口列表
    ↓
补充Worker执行
    ↓
[7] REPORT_REFINEMENT_PROMPT
    ↓
优化后报告
    ↓
循环回到 [5] 直到达标

1
2
3
4
5
6
7
8

USER_PREFERENCES = {
    "report_length": "medium",  # short/medium/long
    "detail_level": "balanced",  # concise/balanced/detailed
    "citation_format": "APA",    # APA/MLA/Chicago
    "language_style": "professional",  # casual/professional/academic
    "include_charts": True,
    "max_sources_per_section": 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

class UserPreferenceManager:
    """用户偏好管理器"""

    def __init__(self, user_id: str = "default"):
        self.user_id = user_id
        self.prefs_file = f"./.user_prefs/{user_id}.json"
        self.preferences = self._load_preferences()

    def _load_preferences(self) -> dict:
        """加载用户偏好"""
        if os.path.exists(self.prefs_file):
            with open(self.prefs_file) as f:
                return json.load(f)
        return USER_PREFERENCES.copy()

    def update_preference(self, key: str, value):
        """更新偏好"""
        self.preferences[key] = value
        self._save_preferences()

    def _save_preferences(self):
        """保存偏好"""
        os.makedirs(os.path.dirname(self.prefs_file), exist_ok=True)
        with open(self.prefs_file, 'w') as f:
            json.dump(self.preferences, f, indent=2)

    def get_preference(self, key: str, default=None):
        """获取偏好"""
        return self.preferences.get(key, default)

class DeepResearchAgent:
    def __init__(self, user_id: str = "default", **kwargs):
        self.pref_manager = UserPreferenceManager(user_id)
        # ... 其他初始化

    def research(self, task: str, **kwargs):
        # 应用用户偏好
        preferences = self.pref_manager.preferences

        orchestrator = Orchestrator(
            preferences=preferences,  # 传递偏好
            **kwargs
        )
        # ...

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

class Orchestrator:
    def __init__(self, preferences: dict = None, **kwargs):
        self.preferences = preferences or {}
        # ...

    def _synthesize_results(self, task, results):
        # 根据偏好调整报告生成
        report_length = self.preferences.get("report_length", "medium")
        detail_level = self.preferences.get("detail_level", "balanced")
        citation_format = self.preferences.get("citation_format", "APA")

        prompt = SYNTHESIS_PROMPT.format(
            task=task,
            results=results,
            length_preference=report_length,
            detail_preference=detail_level,
            citation_format=citation_format
        )
        # ...

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

SYNTHESIS_PROMPT_WITH_PREFERENCES = """
生成研究报告，遵循以下用户偏好：

报告长度：{length_preference}
- short: 1000-2000字
- medium: 2000-4000字
- long: 4000-8000字

详细程度：{detail_preference}
- concise: 只包含关键信息
- balanced: 平衡深度和可读性
- detailed: 深入详细的分析

引用格式：{citation_format}

任务：{task}
研究结果：{results}

请生成符合用户偏好的报告。
"""

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

# 创建Agent，指定用户ID
agent = DeepResearchAgent(user_id="user_123")

# 第一次使用，设置偏好
agent.pref_manager.update_preference("report_length", "long")
agent.pref_manager.update_preference("detail_level", "detailed")
agent.pref_manager.update_preference("citation_format", "MLA")

# 执行研究，自动应用偏好
report = agent.research("AI发展趋势")

# 后续使用，自动加载之前的偏好
agent2 = DeepResearchAgent(user_id="user_123")
report2 = agent2.research("区块链应用")  # 使用相同偏好