1
用户：帮我订一张明天去上海的机票
2
  ↓
3
LLM：决定调用 book_ticket 工具
4
  ↓
5
执行：book_ticket(destination="上海", date="明天")
6
  ↓
7
结果：订票失败——缺少出发地、航班偏好等信息

1
Thought（思考）→ Action（行动）→ Observation（观察）→ Thought（再思考）→ ...

1
任务：「分析某科技公司 2024 年的市场表现」
2

3
=== 第 1 轮 ===
4
Thought: 需要先了解这家公司的基本情况，我应该搜索公司信息
5
Action: search("某科技公司 2024 年财报")
6
Observation: 找到公司 2024 年 Q1-Q4 财报链接
7

8
=== 第 2 轮 ===
9
Thought: 有了财报链接，需要读取具体数据
10
Action: read_document("财报链接")
11
Observation: 营收 100 亿（+20%），净利润 15 亿（-5%），研发投入 20 亿（+30%）
12

13
=== 第 3 轮 ===
14
Thought: 数据有了，但需要了解行业对比情况
15
Action: search("2024 年科技行业平均增长率")
16
Observation: 行业平均增长 10%，净利润率平均 18%
17

18
=== 第 4 轮 ===
19
Thought: 信息足够了，可以总结分析
20
Action: finish("该公司 2024 年营收增长 20%，超过行业平均..."

1
REACT_PROMPT = """
2
你是一个智能助手，使用 Thought-Action-Observation 循环来完成任务。
3

4
可用工具：
5
{tools_description}
6

7
使用以下格式：
8

9
Thought: 思考下一步应该做什么
10
Action: 工具名称[参数]
11
Observation: 工具执行结果
12
... (重复 Thought/Action/Observation)
13
Thought: 我现在知道最终答案了
14
Final Answer: 最终回答
15

16
开始！
17

18
任务：{task}
19
"""

1
from typing import List, Dict, Any
2
import json
3

4
class ReActAgent:
5
    def __init__(self, llm_client, tools: Dict[str, callable]):
6
        self.llm = llm_client
7
        self.tools = tools
8
        self.max_iterations = 10
9

10
    def run(self, task: str) -> str:
11
        """执行 ReAct 循环"""
12
        history = []
13
        tools_desc = self._format_tools()
14

15
        for i in range(self.max_iterations):
16
            # 构建当前上下文
17
            context = self._build_context(task, tools_desc, history)
18

19
            # LLM 生成下一步
20
            response = self.llm.generate(context)
21

22
            # 解析 Thought 和 Action
23
            thought, action, args = self._parse_response(response)
24
            history.append({"thought": thought, "action": action, "args": args})
25

26
            # 检查是否完成
27
            if action == "finish":
28
                return args.get("answer", "")
29

30
            # 执行工具
31
            if action in self.tools:
32
                observation = self.tools[action](**args)
33
                history[-1]["observation"] = observation
34
            else:
35
                history[-1]["observation"] = f"错误：未知工具 {action}"
36

37
        return "达到最大迭代次数，任务未完成"
38

39
    def _parse_response(self, response: str):
40
        """解析 LLM 响应中的 Thought 和 Action"""
41
        lines = response.strip().split("\n")
42
        thought = ""
43
        action = None
44
        args = {}
45

46
        for line in lines:
47
            if line.startswith("Thought:"):
48
                thought = line.replace("Thought:", "").strip()
49
            elif line.startswith("Action:"):
50
                action_part = line.replace("Action:", "").strip()
51
                # 解析 Action[参数]
52
                if "[" in action_part:
53
                    action = action_part.split("[")[0]
54
                    args_str = action_part.split("[")[1].rstrip("]")
55
                    args = json.loads(args_str) if args_str else {}
56
                else:
57
                    action = action_part
58
            elif line.startswith("Final Answer:"):
59
                return thought, "finish", {"answer": line.replace("Final Answer:", "").strip()}
60

61
        return thought, action, args
62

63
    def _format_tools(self) -> str:
64
        """格式化工具描述"""
65
        return "\n".join([
66
            f"- {name}: {func.__doc__}"
67
            for name, func in self.tools.items()
68
        ])
69

70
    def _build_context(self, task: str, tools_desc: str, history: List[Dict]) -> str:
71
        """构建上下文"""
72
        context = f"任务：{task}\n\n可用工具：\n{tools_desc}\n\n"
73

74
        for h in history:
75
            context += f"Thought: {h['thought']}\n"
76
            context += f"Action: {h['action']}"
77
            if h.get('args'):
78
                context += f"[{json.dumps(h['args'])}]"
79
            context += "\n"
80
            if h.get('observation'):
81
                context += f"Observation: {h['observation']}\n"
82

83
        return context
84

85

86
# 使用示例
87
def search_tool(query: str) -> str:
88
    """搜索互联网信息"""
89
    return f"搜索结果：{query} 的相关信息..."
90

91
def calculate_tool(expression: str) -> str:
92
    """计算数学表达式"""
93
    try:
94
        return str(eval(expression))
95
    except:
96
        return "计算错误"
97

98
agent = ReActAgent(
99
    llm_client=openai_client,
100
    tools={"search": search_tool, "calculate": calculate_tool}
101
)
102

103
result = agent.run("某公司去年营收 1000 万，今年增长 20%，计算今年营收")

1
from pydantic import BaseModel
2
from typing import List, Optional
3

4
class PlanStep(BaseModel):
5
    """单个计划步骤"""
6
    step_id: int
7
    description: str
8
    tool: str
9
    args: dict
10
    depends_on: List[int] = []  # 依赖的前置步骤
11

12
class Plan(BaseModel):
13
    """完整执行计划"""
14
    steps: List[PlanStep]
15
    expected_output: str
16

17
class PlanAndExecuteAgent:
18
    def __init__(self, llm_client, tools: Dict[str, callable]):
19
        self.llm = llm_client
20
        self.tools = tools
21
        self.max_replans = 3
22

23
    def run(self, task: str) -> str:
24
        """执行 Plan-and-Execute 流程"""
25
        for attempt in range(self.max_replans):
26
            # 规划阶段
27
            plan = self._create_plan(task)
28
            print(f" 生成的计划：\n{self._format_plan(plan)}")
29

30
            # 执行阶段
31
            results = {}
32
            success = True
33

34
            for step in plan.steps:
35
                # 检查依赖是否满足
36
                if not self._check_dependencies(step, results):
37
                    success = False
38
                    break
39

40
                # 执行步骤
41
                try:
42
                    result = self.tools[step.tool](**step.args)
43
                    results[step.step_id] = result
44
                    print(f"步骤 {step.step_id}: {step.description} -> 成功")
45
                except Exception as e:
46
                    print(f"步骤 {step.step_id} 失败: {e}")
47
                    success = False
48
                    break
49

50
            if success:
51
                return self._synthesize_results(plan, results)
52

53
            print(f" 计划执行失败，尝试重新规划... ({attempt + 1}/{self.max_replans})")
54

55
        return "任务执行失败，已达到最大重试次数"
56

57
    def _create_plan(self, task: str) -> Plan:
58
        """使用 LLM 生成执行计划"""
59
        prompt = f"""
60
你需要为一个任务生成详细的执行计划。
61

62
任务：{task}
63

64
可用工具：
65
{self._format_tools()}
66

67
请以 JSON 格式输出计划，格式如下：
68
{{
69
    "steps": [
70
        {{"step_id": 1, "description": "步骤描述", "tool": "工具名", "args": {{}}, "depends_on": []}}
71
    ],
72
    "expected_output": "预期输出描述"
73
}}
74

75
注意：
76
1. 步骤要有明确的依赖关系
77
2. 每个步骤只调用一个工具
78
3. 参数要具体可执行
79
"""
80

81
        response = self.llm.generate(prompt)
82
        return Plan.parse_raw(response)
83

84
    def _check_dependencies(self, step: PlanStep, results: Dict) -> bool:
85
        """检查步骤依赖是否满足"""
86
        return all(dep_id in results for dep_id in step.depends_on)
87

88
    def _synthesize_results(self, plan: Plan, results: Dict) -> str:
89
        """综合所有结果生成最终答案"""
90
        prompt = f"""
91
任务：{plan.expected_output}
92

93
执行结果：
94
{json.dumps(results, ensure_ascii=False, indent=2)}
95

96
请根据以上结果，生成最终答案。
97
"""
98
        return self.llm.generate(prompt)
99

100
    def _format_plan(self, plan: Plan) -> str:
101
        """格式化计划输出"""
102
        lines = []
103
        for step in plan.steps:
104
            deps = f" (依赖: {step.depends_on})" if step.depends_on else ""
105
            lines.append(f"  {step.step_id}. {step.description} [{step.tool}]{deps}")
106
        return "\n".join(lines)
107

108
    def _format_tools(self) -> str:
109
        """格式化工具描述"""
110
        return "\n".join([
111
            f"- {name}: {func.__doc__}"
112
            for name, func in self.tools.items()
113
        ])
114

115

116
# 使用示例
117
def database_query(sql: str) -> str:
118
    """执行数据库查询"""
119
    return f"查询结果：{sql}"
120

121
def web_search(query: str) -> str:
122
    """搜索网络信息"""
123
    return f"搜索结果：{query}"
124

125
def generate_report(data: str) -> str:
126
    """生成分析报告"""
127
    return f"报告内容：{data}"
128

129
agent = PlanAndExecuteAgent(
130
    llm_client=openai_client,
131
    tools={
132
        "database_query": database_query,
133
        "web_search": web_search,
134
        "generate_report": generate_report
135
    }
136
)
137

138
result = agent.run("分析公司最近的销售趋势并生成报告")

1
 生成的计划：
2
  1. 查询最近 6 个月的销售数据 [database_query]
3
  2. 搜索行业销售趋势作为对比 [web_search]
4
  3. 生成综合分析报告 [generate_report] (依赖: [1, 2])
5

6
步骤 1: 查询最近 6 个月的销售数据 -> 成功
7
步骤 2: 搜索行业销售趋势作为对比 -> 成功
8
步骤 3: 生成综合分析报告 -> 成功
9

10
最终结果：根据数据分析，公司销售额...

1
执行 → 评估 → 反思 → 改进策略 → 重新执行

1
from dataclasses import dataclass
2
from typing import List, Optional
3

4
@dataclass
5
class Reflection:
6
    """反思结果"""
7
    attempt: int
8
    task: str
9
    trajectory: str  # 执行轨迹
10
    failure_reason: str  # 失败原因
11
    improvement_suggestion: str  # 改进建议
12

13
class ReflexionAgent:
14
    def __init__(self, llm_client, tools: Dict[str, callable]):
15
        self.llm = llm_client
16
        self.tools = tools
17
        self.max_attempts = 3
18
        self.reflections: List[Reflection] = []
19

20
    def run(self, task: str) -> str:
21
        """执行 Reflexion 循环"""
22
        for attempt in range(1, self.max_attempts + 1):
23
            print(f"\n 尝试 {attempt}/{self.max_attempts}")
24

25
            # 构建上下文（包含之前的反思）
26
            context = self._build_context(task)
27

28
            # 执行任务
29
            trajectory, result = self._execute(task, context)
30

31
            # 评估结果
32
            evaluation = self._evaluate(task, result)
33

34
            if evaluation["success"]:
35
                print(f"任务成功完成！")
36
                return result
37

38
            # 生成反思
39
            reflection = self._reflect(
40
                attempt=attempt,
41
                task=task,
42
                trajectory=trajectory,
43
                failure_reason=evaluation["reason"]
44
            )
45
            self.reflections.append(reflection)
46
            print(f" 反思：{reflection.improvement_suggestion}")
47

48
        print("达到最大尝试次数")
49
        return "任务执行失败"
50

51
    def _execute(self, task: str, context: str) -> tuple:
52
        """执行任务并返回轨迹和结果"""
53
        # 使用 ReAct 风格执行
54
        trajectory = []
55
        result = ""
56

57
        prompt = f"""
58
{context}
59

60
任务：{task}
61

62
请使用 Thought-Action-Observation 格式执行任务。
63
"""
64

65
        response = self.llm.generate(prompt)
66
        trajectory.append(response)
67

68
        # 解析并执行工具调用
69
        # ... (类似 ReAct 的执行逻辑)
70

71
        return "\n".join(trajectory), result
72

73
    def _evaluate(self, task: str, result: str) -> dict:
74
        """评估执行结果"""
75
        prompt = f"""
76
任务：{task}
77
执行结果：{result}
78

79
请评估结果是否成功完成了任务。
80
输出 JSON 格式：
81
{{"success": true/false, "reason": "失败原因（如果失败）", "score": 0-10}}
82
"""
83

84
        response = self.llm.generate(prompt)
85
        return eval(response)  # 实际应用中应使用 json.loads
86

87
    def _reflect(self, attempt: int, task: str, trajectory: str,
88
                 failure_reason: str) -> Reflection:
89
        """生成反思"""
90
        prompt = f"""
91
任务：{task}
92

93
执行轨迹：
94
{trajectory}
95

96
失败原因：{failure_reason}
97

98
请分析失败原因并提出改进建议：
99
1. 哪里出了问题？
100
2. 为什么会失败？
101
3. 下次应该如何改进？
102

103
输出 JSON 格式：
104
{{"failure_analysis": "...", "improvement_suggestion": "..."}}
105
"""
106

107
        response = self.llm.generate(prompt)
108
        analysis = eval(response)
109

110
        return Reflection(
111
            attempt=attempt,
112
            task=task,
113
            trajectory=trajectory,
114
            failure_reason=failure_reason,
115
            improvement_suggestion=analysis["improvement_suggestion"]
116
        )
117

118
    def _build_context(self, task: str) -> str:
119
        """构建包含反思的上下文"""
120
        context = f"任务：{task}\n\n"
121

122
        if self.reflections:
123
            context += "=== 之前的尝试和反思 ===\n"
124
            for r in self.reflections:
125
                context += f"尝试 {r.attempt}：\n"
126
                context += f"失败原因：{r.failure_reason}\n"
127
                context += f"改进建议：{r.improvement_suggestion}\n\n"
128

129
        return context
130

131

132
# 使用示例
133
agent = ReflexionAgent(
134
    llm_client=openai_client,
135
    tools={"search": search_tool, "calculate": calculate_tool}
136
)
137

138
result = agent.run("找出销售额增长最快的产品并分析原因")

1
 尝试 1/3
2
Thought: 需要查询销售数据
3
Action: search("公司销售数据 2024")
4
Observation: 找到部分数据，但缺少增长率
5
...
6
评估：失败 - 缺少完整数据
7

8
 反思：搜索查询不够精确，应该使用更具体的数据库查询工具而非网络搜索
9

10
 尝试 2/3
11
Thought: 根据反思，应该使用数据库查询获取精确数据
12
Action: database_query("SELECT product, growth_rate FROM sales_2024 ORDER BY growth_rate DESC")
13
Observation: 产品 A 增长率 45%，产品 B 增长率 32%...
14
...
15
任务成功完成！

1
使用 ReAct 当：
2
   - 任务不确定性高，需要灵活调整
3
   - 需要透明的决策过程
4
   - 资源消耗不是首要考虑
5

6
使用 Plan-and-Execute 当：
7
   - 任务步骤清晰可预见
8
   - 需要高效执行多个子任务
9
   - 任务可以并行化
10

11
使用 Reflexion 当：
12
   - 任务质量要求高
13
   - 允许多次迭代尝试
14
   - 可以接受更高的资源消耗

1
class HybridAgent:
2
    """混合架构：Plan-and-Execute + Reflexion"""
3

4
    def __init__(self, llm_client, tools):
5
        self.planner = PlanAndExecuteAgent(llm_client, tools)
6
        self.reflector = ReflexionAgent(llm_client, tools)
7

8
    def run(self, task: str) -> str:
9
        # 先用 Plan-and-Execute 执行
10
        plan = self.planner.create_plan(task)
11
        result = self.planner.execute_plan(plan)
12

13
        # 评估结果
14
        if self._is_satisfactory(result):
15
            return result
16

17
        # 如果不满意，切换到 Reflexion
18
        print("Plan-and-Execute 结果不满意，启用 Reflexion 优化...")
19
        return self.reflector.run(task)

1
from typing import Dict, List, Any
2
from dataclasses import dataclass
3
import json
4

5
@dataclass
6
class ResearchTask:
7
    """调研任务"""
8
    topic: str
9
    requirements: List[str]
10
    deadline: str = None
11

12
@dataclass
13
class ResearchResult:
14
    """调研结果"""
15
    summary: str
16
    key_findings: List[str]
17
    data_sources: List[str]
18
    confidence_score: float
19

20
class ResearchAgent:
21
    """智能调研 Agent"""
22

23
    def __init__(self, llm_client, tools: Dict[str, callable]):
24
        self.llm = llm_client
25
        self.tools = tools
26
        self.reflections = []
27
        self.max_iterations = 3
28

29
    def research(self, task: ResearchTask) -> ResearchResult:
30
        """执行调研任务"""
31
        for iteration in range(self.max_iterations):
32
            print(f"\n{'='*50}")
33
            print(f"第 {iteration + 1} 轮调研")
34
            print(f"{'='*50}")
35

36
            # 阶段 1：规划
37
            plan = self._plan_research(task)
38
            self._print_plan(plan)
39

40
            # 阶段 2：执行
41
            results = self._execute_plan(plan)
42

43
            # 阶段 3：综合
44
            draft = self._synthesize_results(task, results)
45

46
            # 阶段 4：质量检查
47
            evaluation = self._evaluate_quality(task, draft)
48

49
            if evaluation["score"] >= 8:
50
                print(f"\n调研完成！质量评分：{evaluation['score']}/10")
51
                return self._format_result(draft, results)
52

53
            # 阶段 5：反思并改进
54
            reflection = self._reflect_on_failure(
55
                task, plan, results, evaluation
56
            )
57
            self.reflections.append(reflection)
58
            print(f"\n 改进建议：{reflection['improvement']}")
59

60
        print("\n 达到最大迭代次数，返回最佳结果")
61
        return self._format_result(draft, results)
62

63
    def _plan_research(self, task: ResearchTask) -> List[Dict]:
64
        """规划调研步骤"""
65
        prompt = f"""
66
你是一个市场调研专家。请为以下任务制定详细的调研计划。
67

68
调研主题：{task.topic}
69
具体要求：{json.dumps(task.requirements, ensure_ascii=False)}
70

71
可用工具：
72
- search_web: 搜索网络信息
73
- query_database: 查询内部数据库
74
- analyze_data: 分析数据趋势
75
- fetch_report: 获取行业报告
76

77
请输出 JSON 格式的计划：
78
[
79
    {{
80
        "step": 1,
81
        "action": "工具名称",
82
        "params": {{"参数": "值"}},
83
        "purpose": "这步的目的"
84
    }}
85
]
86

87
注意：根据之前的反思调整计划。
88
当前反思历史：{json.dumps(self.reflections[-1] if self.reflections else {}, ensure_ascii=False)}
89
"""
90

91
        response = self.llm.generate(prompt)
92
        return json.loads(response)
93

94
    def _execute_plan(self, plan: List[Dict]) -> Dict:
95
        """执行调研计划"""
96
        results = {}
97

98
        for step in plan:
99
            step_num = step["step"]
100
            action = step["action"]
101
            params = step["params"]
102
            purpose = step["purpose"]
103

104
            print(f"\n 步骤 {step_num}: {purpose}")
105
            print(f"   工具: {action}, 参数: {params}")
106

107
            try:
108
                result = self.tools[action](**params)
109
                results[step_num] = {
110
                    "success": True,
111
                    "data": result,
112
                    "purpose": purpose
113
                }
114
                print(f"   成功")
115
            except Exception as e:
116
                results[step_num] = {
117
                    "success": False,
118
                    "error": str(e),
119
                    "purpose": purpose
120
                }
121
                print(f"   失败: {e}")
122

123
        return results
124

125
    def _synthesize_results(self, task: ResearchTask, results: Dict) -> str:
126
        """综合结果生成报告"""
127
        prompt = f"""
128
请根据调研结果生成市场调研报告。
129

130
调研主题：{task.topic}
131
具体要求：{json.dumps(task.requirements, ensure_ascii=False)}
132

133
调研数据：
134
{json.dumps(results, ensure_ascii=False, indent=2)}
135

136
请生成结构化的调研报告，包括：
137
1. 市场概况
138
2. 关键发现
139
3. 数据支撑
140
4. 建议结论
141
"""
142

143
        return self.llm.generate(prompt)
144

145
    def _evaluate_quality(self, task: ResearchTask, draft: str) -> Dict:
146
        """评估报告质量"""
147
        prompt = f"""
148
请评估以下调研报告的质量。
149

150
调研主题：{task.topic}
151
具体要求：{json.dumps(task.requirements, ensure_ascii=False)}
152

153
报告内容：
154
{draft}
155

156
请从以下维度评分（1-10 分）：
157
1. 完整性：是否覆盖所有要求？
158
2. 准确性：数据是否可靠？
159
3. 可读性：结构是否清晰？
160
4. 洞察深度：分析是否深入？
161

162
输出 JSON：
163
{{
164
    "score": 总分,
165
    "completeness": 分数,
166
    "accuracy": 分数,
167
    "readability": 分数,
168
    "insight": 分数,
169
    "issues": ["问题1", "问题2"]
170
}}
171
"""
172

173
        response = self.llm.generate(prompt)
174
        return json.loads(response)
175

176
    def _reflect_on_failure(self, task: ResearchTask, plan: List[Dict],
177
                           results: Dict, evaluation: Dict) -> Dict:
178
        """反思失败原因"""
179
        prompt = f"""
180
调研任务未达标，请分析原因并提出改进建议。
181

182
任务：{task.topic}
183
计划：{json.dumps(plan, ensure_ascii=False)}
184
结果问题：{json.dumps(evaluation["issues"], ensure_ascii=False)}
185

186
请分析：
187
1. 计划是否合理？
188
2. 数据是否充足？
189
3. 分析是否深入？
190

191
输出 JSON：
192
{{
193
    "root_cause": "根本原因",
194
    "improvement": "改进建议",
195
    "next_plan_adjustments": ["调整1", "调整2"]
196
}}
197
"""
198

199
        response = self.llm.generate(prompt)
200
        return json.loads(response)
201

202
    def _format_result(self, draft: str, results: Dict) -> ResearchResult:
203
        """格式化最终结果"""
204
        # 提取关键信息
205
        return ResearchResult(
206
            summary=draft[:200] + "...",
207
            key_findings=[],  # 从 draft 中提取
208
            data_sources=[str(k) for k in results.keys()],
209
            confidence_score=0.85
210
        )
211

212
    def _print_plan(self, plan: List[Dict]):
213
        """打印计划"""
214
        print("\n 调研计划：")
215
        for step in plan:
216
            print(f"  {step['step']}. {step['purpose']} ({step['action']})")
217

218

219
# 使用示例
220
def search_web(query: str) -> str:
221
    """搜索网络信息"""
222
    return f"搜索结果：{query} 的相关信息..."
223

224
def query_database(sql: str) -> str:
225
    """查询数据库"""
226
    return f"查询结果：{sql}"
227

228
def analyze_data(data: str) -> str:
229
    """分析数据"""
230
    return f"分析结果：{data}"
231

232
def fetch_report(industry: str) -> str:
233
    """获取行业报告"""
234
    return f"报告：{industry} 行业分析"
235

236
# 创建 Agent
237
agent = ResearchAgent(
238
    llm_client=openai_client,
239
    tools={
240
        "search_web": search_web,
241
        "query_database": query_database,
242
        "analyze_data": analyze_data,
243
        "fetch_report": fetch_report
244
    }
245
)
246

247
# 执行调研任务
248
task = ResearchTask(
249
    topic="新能源汽车市场趋势分析",
250
    requirements=[
251
        "市场规模及增长率",
252
        "主要竞争者分析",
253
        "技术发展趋势",
254
        "政策影响"
255
    ]
256
)
257

258
result = agent.research(task)

1
==================================================
2
第 1 轮调研
3
==================================================
4

5
 调研计划：
6
  1. 获取新能源汽车行业报告 (fetch_report)
7
  2. 搜索市场增长数据 (search_web)
8
  3. 分析竞争格局 (analyze_data)
9
  4. 查询政策信息 (query_database)
10

11
 步骤 1: 获取新能源汽车行业报告
12
   工具: fetch_report, 参数: {"industry": "新能源汽车"}
13
   成功
14

15
 步骤 2: 搜索市场增长数据
16
   工具: search_web, 参数: {"query": "2024年新能源汽车市场规模增长率"}
17
   成功
18

19
 步骤 3: 分析竞争格局
20
   工具: analyze_data, 参数: {"data": "前两步的数据"}
21
   成功
22

23
 步骤 4: 查询政策信息
24
   工具: query_database, 参数: {"sql": "SELECT * FROM policies WHERE industry='新能源汽车'"}
25
   失败: 数据库连接超时
26

27
 质量评分：6/10
28
- 缺少政策影响分析
29
- 数据来源单一
30

31
 改进建议：增加政策信息的搜索渠道，不依赖单一数据库
32

33
==================================================
34
第 2 轮调研
35
==================================================
36

37
 调研计划：
38
  1. 获取新能源汽车行业报告 (fetch_report)
39
  2. 搜索市场增长数据 (search_web)
40
  3. 搜索新能源汽车补贴政策 (search_web) ← 调整
41
  4. 分析竞争格局 (analyze_data)
42
  ...
43

44
调研完成！质量评分：8/10