template_selection_node.py
11.1 KB
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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
"""
模板选择节点
根据查询内容和可用模板选择最合适的报告模板
"""
import os
import json
from typing import Dict, Any, List, Optional
from .base_node import BaseNode
from ..prompts import SYSTEM_PROMPT_TEMPLATE_SELECTION
class TemplateSelectionNode(BaseNode):
"""模板选择处理节点"""
def __init__(self, llm_client, template_dir: str = "ReportEngine/report_template"):
"""
初始化模板选择节点
Args:
llm_client: LLM客户端
template_dir: 模板目录路径
"""
super().__init__(llm_client, "TemplateSelectionNode")
self.template_dir = template_dir
def run(self, input_data: Dict[str, Any], **kwargs) -> Dict[str, Any]:
"""
执行模板选择
Args:
input_data: 包含查询和报告内容的字典
- query: 原始查询
- reports: 三个子agent的报告列表
- forum_logs: 论坛日志内容
Returns:
选择的模板信息
"""
self.log_info("开始模板选择...")
query = input_data.get('query', '')
reports = input_data.get('reports', [])
forum_logs = input_data.get('forum_logs', '')
# 获取可用模板
available_templates = self._get_available_templates()
if not available_templates:
self.log_info("未找到预设模板，使用内置默认模板")
return self._get_fallback_template()
# 首先尝试简单关键词匹配
simple_match = self._simple_keyword_matching(query, available_templates)
if simple_match:
self.log_info(f"通过关键词匹配选择模板: {simple_match['template_name']}")
return simple_match
# 如果关键词匹配失败，尝试LLM选择
try:
llm_result = self._llm_template_selection(query, reports, forum_logs, available_templates)
if llm_result:
return llm_result
except Exception as e:
self.log_error(f"LLM模板选择失败: {str(e)}")
# 所有方法都失败，使用默认的社会热点事件模板
default_template = self._get_default_social_event_template(available_templates)
if default_template:
return default_template
# 最后备选方案
return self._get_fallback_template()
def _simple_keyword_matching(self, query: str, available_templates: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""基于关键词的简单模板匹配"""
query_lower = query.lower()
# 关键词映射
keyword_mapping = {
'企业': ['企业品牌'],
'品牌': ['企业品牌'],
'声誉': ['企业品牌'],
'市场': ['市场竞争'],
'竞争': ['市场竞争'],
'格局': ['市场竞争'],
'政策': ['政策', '行业'],
'行业': ['政策', '行业'],
'动态': ['政策', '行业'],
'突发': ['突发事件', '危机'],
'危机': ['突发事件', '危机'],
'公关': ['突发事件', '危机'],
'日常': ['日常', '定期'],
'定期': ['日常', '定期'],
'监测': ['日常', '定期'],
'热点': ['社会公共热点'],
'社会': ['社会公共热点'],
'事件': ['社会公共热点'],
}
# 检查查询中的关键词
for keyword, template_keywords in keyword_mapping.items():
if keyword in query_lower:
# 查找匹配的模板
for template in available_templates:
for template_keyword in template_keywords:
if template_keyword in template['name']:
return {
'template_name': template['name'],
'template_content': template['content'],
'selection_reason': f'基于关键词"{keyword}"匹配选择'
}
return None
def _llm_template_selection(self, query: str, reports: List[Any], forum_logs: str,
available_templates: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""使用LLM进行模板选择"""
self.log_info("尝试使用LLM进行模板选择...")
# 构建模板列表
template_list = "\n".join([f"- {t['name']}: {t['description']}" for t in available_templates])
user_message = f"""查询内容: {query}
报告数量: {len(reports)} 个分析引擎报告
论坛日志: {'有' if forum_logs else '无'}
可用模板:
{template_list}
请选择最合适的模板。"""
# 调用LLM
response = self.llm_client.invoke(SYSTEM_PROMPT_TEMPLATE_SELECTION, user_message)
# 检查响应是否为空
if not response or not response.strip():
self.log_error("LLM返回空响应")
return None
self.log_info(f"LLM原始响应: {response[:200]}...")
# 尝试解析JSON响应
try:
# 清理响应文本
cleaned_response = self._clean_llm_response(response)
result = json.loads(cleaned_response)
# 验证选择的模板是否存在
selected_template_name = result.get('template_name', '')
for template in available_templates:
if template['name'] == selected_template_name or selected_template_name in template['name']:
self.log_info(f"LLM选择模板: {selected_template_name}")
return {
'template_name': template['name'],
'template_content': template['content'],
'selection_reason': result.get('selection_reason', 'LLM智能选择')
}
self.log_error(f"LLM选择的模板不存在: {selected_template_name}")
return None
except json.JSONDecodeError as e:
self.log_error(f"JSON解析失败: {str(e)}")
# 尝试从文本响应中提取模板信息
return self._extract_template_from_text(response, available_templates)
def _clean_llm_response(self, response: str) -> str:
"""清理LLM响应"""
# 移除可能的markdown代码块标记
if '```json' in response:
response = response.split('```json')[1].split('```')[0]
elif '```' in response:
response = response.split('```')[1].split('```')[0]
# 移除前后空白
response = response.strip()
return response
def _extract_template_from_text(self, response: str, available_templates: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""从文本响应中提取模板信息"""
self.log_info("尝试从文本响应中提取模板信息")
# 查找响应中是否包含模板名称
for template in available_templates:
template_name_variants = [
template['name'],
template['name'].replace('.md', ''),
template['name'].replace('模板', ''),
]
for variant in template_name_variants:
if variant in response:
self.log_info(f"在响应中找到模板: {template['name']}")
return {
'template_name': template['name'],
'template_content': template['content'],
'selection_reason': '从文本响应中提取'
}
return None
def _get_available_templates(self) -> List[Dict[str, Any]]:
"""获取可用的模板列表"""
templates = []
if not os.path.exists(self.template_dir):
self.log_error(f"模板目录不存在: {self.template_dir}")
return templates
# 查找所有markdown模板文件
for filename in os.listdir(self.template_dir):
if filename.endswith('.md'):
template_path = os.path.join(self.template_dir, filename)
try:
with open(template_path, 'r', encoding='utf-8') as f:
content = f.read()
template_name = filename.replace('.md', '')
description = self._extract_template_description(template_name)
templates.append({
'name': template_name,
'path': template_path,
'content': content,
'description': description
})
except Exception as e:
self.log_error(f"读取模板文件失败 {filename}: {str(e)}")
return templates
def _extract_template_description(self, template_name: str) -> str:
"""根据模板名称生成描述"""
if '企业品牌' in template_name:
return "适用于企业品牌声誉和形象分析"
elif '市场竞争' in template_name:
return "适用于市场竞争格局和对手分析"
elif '日常' in template_name or '定期' in template_name:
return "适用于日常监测和定期汇报"
elif '政策' in template_name or '行业' in template_name:
return "适用于政策影响和行业动态分析"
elif '热点' in template_name or '社会' in template_name:
return "适用于社会热点和公共事件分析"
elif '突发' in template_name or '危机' in template_name:
return "适用于突发事件和危机公关"
return "通用报告模板"
def _get_default_social_event_template(self, available_templates: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""获取默认的社会热点事件分析模板"""
# 查找社会热点事件分析模板
for template in available_templates:
if '社会公共热点事件' in template['name'] or '热点' in template['name']:
self.log_info(f"使用默认模板: {template['name']}")
return {
'template_name': template['name'],
'template_content': template['content'],
'selection_reason': '默认使用社会热点事件分析模板'
}
return None
def _get_fallback_template(self) -> Dict[str, Any]:
"""获取备用默认模板（空模板，让LLM自行发挥）"""
self.log_info("未找到合适模板，使用空模板让LLM自行发挥")
return {
'template_name': '自由发挥模板',
'template_content': '',
'selection_reason': '未找到合适的预设模板，让LLM根据内容自行设计报告结构'
}