diff --git a/src/bidmaster/nodes/base.py b/src/bidmaster/nodes/base.py index cd6aa44..a2fe9c6 100644 --- a/src/bidmaster/nodes/base.py +++ b/src/bidmaster/nodes/base.py @@ -102,6 +102,53 @@ class BaseNode(ABC): """记录节点执行失败""" logger.error(f"节点执行失败: {self.name} - {error}") + def _update_state(self, state: Dict[str, Any], **kwargs) -> Dict[str, Any]: + """统一的状态更新方法 + + Args: + state: 当前状态字典 + **kwargs: 要更新的状态键值对 + + Returns: + 更新后的状态字典 + """ + state["current_step"] = self.name + state.update(kwargs) + return state + + def _handle_execution_error(self, state: Dict[str, Any], error: Exception) -> Dict[str, Any]: + """统一的执行错误处理 + + Args: + state: 当前状态字典 + error: 异常对象 + + Returns: + 更新后的状态字典 + """ + self._log_error(error) + state["error"] = str(error) + state["should_continue"] = False + return state + + def execute_with_error_handling(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: + """带统一错误处理的执行方法 + + Args: + state: 当前状态字典 + context: 节点执行上下文 + + Returns: + 更新后的状态字典 + """ + self._log_start() + try: + result = self.execute(state, context) + self._log_success() + return result + except Exception as e: + return self._handle_execution_error(state, e) + class ConditionalNode(BaseNode): """条件节点基类 diff --git a/src/bidmaster/nodes/toc/__init__.py b/src/bidmaster/nodes/toc/__init__.py index 2972644..387093c 100644 --- a/src/bidmaster/nodes/toc/__init__.py +++ b/src/bidmaster/nodes/toc/__init__.py @@ -1,18 +1,30 @@ """目录生成相关节点 -包含目录生成Agent的所有节点实现。 +包含目录生成Agent的所有节点实现和相关组件。 """ +# 核心节点 from .group_criteria import GroupCriteriaNode from .generate_first_level import GenerateFirstLevelNode from .generate_sub_chapters import GenerateSubChaptersNode from .review_structure import ReviewStructureNode from .finalize_chapters import FinalizeChaptersNode +# 辅助组件 +from .factories import ChapterFactory +from .category_manager import CategoryManager +from .llm_helper import LLMHelper + __all__ = [ + # 核心节点 "GroupCriteriaNode", "GenerateFirstLevelNode", "GenerateSubChaptersNode", "ReviewStructureNode", - "FinalizeChaptersNode" + "FinalizeChaptersNode", + + # 辅助组件 + "ChapterFactory", + "CategoryManager", + "LLMHelper" ] \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/category_manager.py b/src/bidmaster/nodes/toc/category_manager.py new file mode 100644 index 0000000..49f788e --- /dev/null +++ b/src/bidmaster/nodes/toc/category_manager.py @@ -0,0 +1,166 @@ +"""类别管理器 + +统一管理评分项类别相关的所有逻辑，包括分组、排序、匹配等。 +""" + +import logging +from typing import Dict, List, Any + +from ...tools.parser import ScoringCriteria, DocumentChapter +from .constants import CATEGORY_NAMES, CATEGORY_ORDER + +logger = logging.getLogger(__name__) + + +class CategoryManager: + """类别管理器 + + 提供统一的类别相关操作方法。 + """ + + @staticmethod + def group_criteria_by_category(criteria: List[ScoringCriteria]) -> Dict[str, List[ScoringCriteria]]: + """按类别分组评分项 + + Args: + criteria: 评分项列表 + + Returns: + 分组后的字典，key为类别，value为该类别下的评分项列表 + """ + # 初始化所有预定义类别 + category_groups = {category: [] for category in CATEGORY_ORDER} + + # 分组 + for criterion in criteria: + category_key = criterion.category.value + if category_key in category_groups: + category_groups[category_key].append(criterion) + else: + # 未知类别默认归入技术方案 + category_groups["technical_solution"].append(criterion) + logger.warning(f"未知类别 {category_key}，归入技术方案类别") + + # 只保留有评分项的类别 + filtered_groups = {k: v for k, v in category_groups.items() if v} + + # 记录分组统计 + for category, items in filtered_groups.items(): + total_score = sum(item.max_score for item in items) + logger.info(f"类别 {category}: {len(items)}项，总分 {total_score}") + + return filtered_groups + + @staticmethod + def get_category_first_index(category: str, technical_criteria: List[ScoringCriteria]) -> int: + """获取类别在原始评分项中的首次出现位置 + + Args: + category: 类别名称 + technical_criteria: 技术评分项列表 + + Returns: + 首次出现的索引位置，未找到时返回999 + """ + for i, criteria in enumerate(technical_criteria): + if criteria.category.value == category: + return i + return 999 + + @staticmethod + def sort_categories_by_order(category_groups: Dict[str, List[ScoringCriteria]], + technical_criteria: List[ScoringCriteria]) -> List[str]: + """按原始出现顺序排序类别 + + Args: + category_groups: 类别分组字典 + technical_criteria: 原始技术评分项列表 + + Returns: + 排序后的类别键名列表 + """ + return sorted( + category_groups.keys(), + key=lambda cat: CategoryManager.get_category_first_index(cat, technical_criteria) + ) + + @staticmethod + def extract_category_from_chapter_id(chapter_id: str) -> str: + """从章节ID中提取类别信息 + + Args: + chapter_id: 章节ID（格式：chapter_XX_category） + + Returns: + 类别名称，如果解析失败返回空字符串 + """ + if "_" not in chapter_id: + return "" + + parts = chapter_id.split("_") + if len(parts) < 3: + return "" + + return "_".join(parts[2:]) + + @staticmethod + def find_corresponding_criteria(chapter: DocumentChapter, + technical_criteria: List[ScoringCriteria]) -> List[ScoringCriteria]: + """查找章节对应的评分项 + + Args: + chapter: 章节对象 + technical_criteria: 技术评分项列表 + + Returns: + 对应的评分项列表 + """ + category = CategoryManager.extract_category_from_chapter_id(chapter.id) + if not category: + return [] + + return [ + c for c in technical_criteria + if c.category.value == category + ] + + @staticmethod + def format_criteria_summary(technical_criteria: List[ScoringCriteria]) -> str: + """格式化评分项摘要用于显示 + + Args: + technical_criteria: 技术评分项列表 + + Returns: + 格式化后的字符串 + """ + lines = [] + + # 按类别分组 + category_groups = {} + for criteria in technical_criteria: + category = criteria.category.value + if category not in category_groups: + category_groups[category] = [] + category_groups[category].append(criteria) + + # 格式化输出 + for category, items in category_groups.items(): + category_name = CATEGORY_NAMES.get(category, category) + lines.append(f"【{category_name}】({len(items)}项):") + for item in items: + lines.append(f" - {item.item_name} ({item.max_score}分)") + + return "\n".join(lines) + + @staticmethod + def calculate_category_score(criteria_list: List[ScoringCriteria]) -> int: + """计算类别总分 + + Args: + criteria_list: 该类别下的评分项列表 + + Returns: + 总分值 + """ + return sum(c.max_score for c in criteria_list) \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/constants.py b/src/bidmaster/nodes/toc/constants.py new file mode 100644 index 0000000..10e3ebc --- /dev/null +++ b/src/bidmaster/nodes/toc/constants.py @@ -0,0 +1,31 @@ +"""目录生成相关常量定义 + +统一管理TOC生成过程中使用的所有常量。 +""" + +# 类别名称映射 +CATEGORY_NAMES = { + "compliance": "合规响应", + "technical_solution": "技术方案", + "equipment_spec": "设备规格", + "quality_safety": "质量安全", + "after_sales": "售后服务", + "implementation": "实施方案" +} + +# 预定义类别顺序 +CATEGORY_ORDER = [ + "technical_solution", + "equipment_spec", + "implementation", + "quality_safety", + "after_sales", + "compliance" +] + +# 默认子章节模板 +DEFAULT_SUB_CHAPTERS_TEMPLATE = [ + {"title": "方案概述", "level": 2}, + {"title": "具体实施", "level": 2}, + {"title": "保障措施", "level": 2} +] \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/factories.py b/src/bidmaster/nodes/toc/factories.py new file mode 100644 index 0000000..00cf94e --- /dev/null +++ b/src/bidmaster/nodes/toc/factories.py @@ -0,0 +1,160 @@ +"""章节创建工厂类 + +统一管理DocumentChapter对象的创建逻辑，避免重复代码。 +""" + +import logging +from typing import List, Dict, Any, Optional + +from ...tools.parser import DocumentChapter, ScoringCriteria +from .constants import CATEGORY_NAMES + +logger = logging.getLogger(__name__) + + +class ChapterFactory: + """章节创建工厂类 + + 提供统一的章节创建、ID生成、占位符生成方法。 + """ + + @staticmethod + def create_main_chapter(category: str, + chapter_index: int, + total_score: int = 0) -> DocumentChapter: + """创建一级主章节 + + Args: + category: 类别键名 + chapter_index: 章节索引号 + total_score: 该类别总分 + + Returns: + DocumentChapter对象 + """ + chapter_id = f"chapter_{chapter_index:02d}_{category}" + category_name = CATEGORY_NAMES.get(category, category) + + # 生成标题，包含分值信息 + title = category_name + if total_score > 0: + title += f" ({total_score}分)" + + return DocumentChapter( + id=chapter_id, + title=title, + level=1, + score=total_score, + template_placeholder=f"{{{{{chapter_id}_content}}}}" + ) + + @staticmethod + def create_sub_chapter(parent_id: str, + sub_index: int, + title: str, + level: int = 2, + score: int = 0) -> DocumentChapter: + """创建子章节 + + Args: + parent_id: 父章节ID + sub_index: 子章节索引 + title: 章节标题 + level: 章节级别（2或3） + score: 章节分值 + + Returns: + DocumentChapter对象 + """ + chapter_id = f"{parent_id}_sub_{sub_index:02d}" + + return DocumentChapter( + id=chapter_id, + title=title, + level=level, + score=score, + template_placeholder=f"{{{{{chapter_id}_content}}}}" + ) + + @staticmethod + def create_third_level_chapter(parent_id: str, + sub_index: int, + child_index: int, + title: str) -> DocumentChapter: + """创建三级章节 + + Args: + parent_id: 一级章节ID + sub_index: 二级章节索引 + child_index: 三级章节索引 + title: 章节标题 + + Returns: + DocumentChapter对象 + """ + chapter_id = f"{parent_id}_sub_{sub_index:02d}_{child_index:02d}" + + return DocumentChapter( + id=chapter_id, + title=title, + level=3, + template_placeholder=f"{{{{{chapter_id}_content}}}}" + ) + + @staticmethod + def create_standard_chapter(chapter_id: str, + title: str, + level: int = 1) -> DocumentChapter: + """创建标准章节（如评标索引表等） + + Args: + chapter_id: 章节ID + title: 章节标题 + level: 章节级别 + + Returns: + DocumentChapter对象 + """ + return DocumentChapter( + id=chapter_id, + title=title, + level=level, + template_placeholder=f"{{{{{chapter_id}_content}}}}" + ) + + @staticmethod + def create_chapters_from_ai_response(parent_chapter: DocumentChapter, + sub_chapters_data: List[Dict[str, Any]]) -> List[DocumentChapter]: + """从AI响应数据创建子章节列表 + + Args: + parent_chapter: 父章节对象 + sub_chapters_data: AI返回的子章节数据列表 + + Returns: + 子章节对象列表 + """ + sub_chapters = [] + + for i, sub_data in enumerate(sub_chapters_data, 1): + title = sub_data.get("title", f"子标题{i}") + level = sub_data.get("level", 2) + score = sub_data.get("score", 0) + + # 创建二级章节 + sub_chapter = ChapterFactory.create_sub_chapter( + parent_chapter.id, i, title, level, score + ) + + # 处理三级章节 + for j, child_data in enumerate(sub_data.get("children", []), 1): + child_title = child_data.get("title", f"三级标题{j}") + + child_chapter = ChapterFactory.create_third_level_chapter( + parent_chapter.id, i, j, child_title + ) + sub_chapter.children.append(child_chapter) + + sub_chapters.append(sub_chapter) + + return sub_chapters \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/finalize_chapters.py b/src/bidmaster/nodes/toc/finalize_chapters.py index b2cb126..7b7035b 100644 --- a/src/bidmaster/nodes/toc/finalize_chapters.py +++ b/src/bidmaster/nodes/toc/finalize_chapters.py @@ -8,6 +8,7 @@ from typing import Dict, List, Any from ..base import BaseNode, NodeContext from ...tools.parser import DocumentChapter +from .factories import ChapterFactory logger = logging.getLogger(__name__) @@ -25,36 +26,23 @@ class FinalizeChaptersNode(BaseNode): def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: """执行章节最终确定""" - self._log_start() + preliminary_chapters = state.get("preliminary_chapters", []) + structure_review = state.get("structure_review", {}) - try: - preliminary_chapters = state.get("preliminary_chapters", []) - structure_review = state.get("structure_review", {}) + if not preliminary_chapters: + raise ValueError("缺少初步章节数据") - if not preliminary_chapters: - raise ValueError("缺少初步章节数据") + # 应用审查建议并最终确定 + final_chapters = self._finalize_with_review(preliminary_chapters, structure_review) - # 应用审查建议并最终确定 - final_chapters = self._finalize_with_review(preliminary_chapters, structure_review) + # 确保标准章节存在 + final_chapters = self._ensure_standard_chapters(final_chapters) - # 确保标准章节存在 - final_chapters = self._ensure_standard_chapters(final_chapters) + logger.info(f"最终生成{len(final_chapters)}个章节") - logger.info(f"最终生成{len(final_chapters)}个章节") - - # 更新状态 - state["final_chapters"] = final_chapters - state["current_step"] = self.name - state["should_continue"] = False # 完成流程 - - self._log_success() - return state - - except Exception as e: - self._log_error(e) - state["error"] = str(e) - state["should_continue"] = False - raise + return self._update_state(state, + final_chapters=final_chapters, + should_continue=False) def _finalize_with_review(self, preliminary_chapters: List[DocumentChapter], @@ -112,11 +100,10 @@ class FinalizeChaptersNode(BaseNode): # 确保评标索引表存在（作为第一章） has_index = any("评标索引" in ch.title for ch in final_chapters) if not has_index: - index_chapter = DocumentChapter( - id="evaluation_index", - title="评标索引表（技术评分完全对应）", # 不包含编号 - level=1, - template_placeholder="{{evaluation_index_content}}" + index_chapter = ChapterFactory.create_standard_chapter( + "evaluation_index", + "评标索引表（技术评分完全对应）", + 1 ) final_chapters.insert(0, index_chapter) logger.info("添加评标索引表章节") @@ -124,18 +111,4 @@ class FinalizeChaptersNode(BaseNode): # 可以在此添加其他必要的标准章节 # 例如：封面、目录、声明等 - return final_chapters - - def _log_chapter_structure(self, chapters: List[DocumentChapter]) -> None: - """记录最终章节结构 - - Args: - chapters: 章节列表 - """ - logger.info("最终章节结构:") - for chapter in chapters: - logger.info(f" {chapter.title}") - for sub in chapter.children: - logger.info(f" {sub.title}") - for child in sub.children: - logger.info(f" {child.title}") \ No newline at end of file + return final_chapters \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/generate_first_level.py b/src/bidmaster/nodes/toc/generate_first_level.py index 93962ca..ac41470 100644 --- a/src/bidmaster/nodes/toc/generate_first_level.py +++ b/src/bidmaster/nodes/toc/generate_first_level.py @@ -8,19 +8,11 @@ from typing import Dict, List, Any from ..base import BaseNode, NodeContext from ...tools.parser import ScoringCriteria, DocumentChapter +from .category_manager import CategoryManager +from .factories import ChapterFactory logger = logging.getLogger(__name__) -# 统一的类别名称映射 -CATEGORY_NAMES = { - "compliance": "合规响应", - "technical_solution": "技术方案", - "equipment_spec": "设备规格", - "quality_safety": "质量安全", - "after_sales": "售后服务", - "implementation": "实施方案" -} - class GenerateFirstLevelNode(BaseNode): """生成一级章节的节点""" @@ -35,32 +27,18 @@ class GenerateFirstLevelNode(BaseNode): def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: """执行一级章节生成""" - self._log_start() + category_groups = state.get("category_groups", {}) + technical_criteria = state.get("technical_criteria", []) - try: - category_groups = state.get("category_groups", {}) - technical_criteria = state.get("technical_criteria", []) + if not category_groups: + raise ValueError("缺少类别分组数据") - if not category_groups: - raise ValueError("缺少类别分组数据") + # 生成一级章节 + chapters = self._generate_first_level_chapters(category_groups, technical_criteria) - # 生成一级章节 - chapters = self._generate_first_level_chapters(category_groups, technical_criteria) + logger.info(f"生成{len(chapters)}个一级章节") - logger.info(f"生成{len(chapters)}个一级章节") - - # 更新状态 - state["preliminary_chapters"] = chapters - state["current_step"] = self.name - - self._log_success() - return state - - except Exception as e: - self._log_error(e) - state["error"] = str(e) - state["should_continue"] = False - raise + return self._update_state(state, preliminary_chapters=chapters) def _generate_first_level_chapters(self, category_groups: Dict[str, List[ScoringCriteria]], @@ -77,16 +55,8 @@ class GenerateFirstLevelNode(BaseNode): chapters = [] chapter_index = 1 - # 按原始顺序确定章节顺序 - def get_category_first_index(category: str) -> int: - """获取类别在原始评分项中的首次出现位置""" - for i, criteria in enumerate(technical_criteria): - if criteria.category.value == category: - return i - return 999 # 未找到时排在最后 - # 按原始出现顺序排序类别 - sorted_categories = sorted(category_groups.keys(), key=get_category_first_index) + sorted_categories = CategoryManager.sort_categories_by_order(category_groups, technical_criteria) # 为每个类别创建一级章节 for category in sorted_categories: @@ -95,29 +65,13 @@ class GenerateFirstLevelNode(BaseNode): continue # 计算类别总分 - total_score = sum(c.max_score for c in criteria_list) - - # 生成章节ID和标题 - chapter_id = f"chapter_{chapter_index:02d}_{category}" - category_name = CATEGORY_NAMES.get(category, category) - chapter_title = category_name - - # 添加分值信息（如果有分值） - if total_score > 0: - chapter_title += f" ({total_score}分)" - - # 创建章节对象 - chapter = DocumentChapter( - id=chapter_id, - title=chapter_title, - level=1, - score=total_score, - template_placeholder=f"{{{{{chapter_id}_content}}}}" - ) + total_score = CategoryManager.calculate_category_score(criteria_list) + # 使用工厂方法创建章节 + chapter = ChapterFactory.create_main_chapter(category, chapter_index, total_score) chapters.append(chapter) chapter_index += 1 - logger.debug(f"创建一级章节: {chapter_title}") + logger.debug(f"创建一级章节: {chapter.title}") return chapters \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/generate_sub_chapters.py b/src/bidmaster/nodes/toc/generate_sub_chapters.py index 55b9141..1040888 100644 --- a/src/bidmaster/nodes/toc/generate_sub_chapters.py +++ b/src/bidmaster/nodes/toc/generate_sub_chapters.py @@ -3,13 +3,14 @@ 通过AI智能生成或模板生成二三级子标题。 """ -import json import logging -from typing import Dict, List, Any, Optional +from typing import Dict, List, Any from ..base import BaseNode, NodeContext from ...tools.parser import ScoringCriteria, DocumentChapter -from ...tools.llm import LLMService +from .category_manager import CategoryManager +from .factories import ChapterFactory +from .llm_helper import LLMHelper logger = logging.getLogger(__name__) @@ -27,248 +28,49 @@ class GenerateSubChaptersNode(BaseNode): def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: """执行子标题生成""" - self._log_start() + preliminary_chapters = state.get("preliminary_chapters", []) + technical_criteria = state.get("technical_criteria", []) - try: - preliminary_chapters = state.get("preliminary_chapters", []) - technical_criteria = state.get("technical_criteria", []) + if not preliminary_chapters: + raise ValueError("缺少初步章节") - if not preliminary_chapters: - raise ValueError("缺少初步章节") + # 为每个章节生成子标题 + enhanced_chapters = [] + for chapter in preliminary_chapters: + enhanced_chapter = self._enhance_chapter_with_subs(chapter, technical_criteria) + enhanced_chapters.append(enhanced_chapter) - # 只使用AI生成，不再需要交互选择 - generation_mode = "ai" - template_file = None + logger.info(f"完成{len(enhanced_chapters)}个章节的子标题生成") - # 为每个章节生成子标题 - enhanced_chapters = [] - for chapter in preliminary_chapters: - enhanced_chapter = self._enhance_chapter_with_subs( - chapter, technical_criteria, generation_mode, template_file - ) - enhanced_chapters.append(enhanced_chapter) - - logger.info(f"完成{len(enhanced_chapters)}个章节的子标题生成") - - # 更新状态 - state["preliminary_chapters"] = enhanced_chapters - state["current_step"] = self.name - - self._log_success() - return state - - except Exception as e: - self._log_error(e) - state["error"] = str(e) - state["should_continue"] = False - raise + return self._update_state(state, preliminary_chapters=enhanced_chapters) def _enhance_chapter_with_subs(self, chapter: DocumentChapter, - technical_criteria: List[ScoringCriteria], - generation_mode: str, - template_file: Optional[str]) -> DocumentChapter: + technical_criteria: List[ScoringCriteria]) -> DocumentChapter: """为章节增强子标题 Args: chapter: 原始章节 technical_criteria: 技术评分项 - generation_mode: 生成模式 - template_file: 模板文件（仅template模式使用） Returns: 增强后的章节 """ # 找到该章节对应的评分项 - corresponding_criteria = self._find_corresponding_criteria(chapter, technical_criteria) + corresponding_criteria = CategoryManager.find_corresponding_criteria(chapter, technical_criteria) if not corresponding_criteria: logger.warning(f"章节 {chapter.title} 没有找到对应的评分项") return chapter - # 根据模式生成子标题 - if generation_mode == "ai": - sub_chapters = self._generate_ai_sub_chapters(corresponding_criteria, chapter) + # 使用AI生成子标题 + sub_chapters_data = LLMHelper.generate_sub_chapters_ai(corresponding_criteria, chapter) + + if sub_chapters_data: + chapter.children = ChapterFactory.create_chapters_from_ai_response(chapter, sub_chapters_data) + logger.info(f"章节 {chapter.title} 生成了 {len(chapter.children)} 个子标题") else: - sub_chapters = self._generate_template_sub_chapters( - corresponding_criteria[0], chapter, template_file - ) + logger.warning(f"章节 {chapter.title} AI生成子标题失败") + raise RuntimeError(f"AI生成子标题失败: {chapter.title}") - # 设置子章节 - chapter.children = sub_chapters - logger.info(f"章节 {chapter.title} 生成了 {len(sub_chapters)} 个子标题") - - return chapter - - def _find_corresponding_criteria(self, - chapter: DocumentChapter, - technical_criteria: List[ScoringCriteria]) -> List[ScoringCriteria]: - """查找章节对应的评分项 - - Args: - chapter: 章节 - technical_criteria: 技术评分项列表 - - Returns: - 对应的评分项列表 - """ - # 从章节ID提取类别 - if "_" not in chapter.id: - return [] - - parts = chapter.id.split("_") - if len(parts) < 3: - return [] - - category = "_".join(parts[2:]) - - # 找到该类别的所有评分项 - return [ - c for c in technical_criteria - if c.category.value == category - ] - - def _generate_ai_sub_chapters(self, - criteria_list: List[ScoringCriteria], - parent_chapter: DocumentChapter) -> List[DocumentChapter]: - """AI生成子标题 - - Args: - criteria_list: 对应的评分项列表 - parent_chapter: 父章节 - - Returns: - 子章节列表 - """ - try: - # 构建评分项信息 - criteria_info = [] - for criteria in criteria_list: - criteria_info.append(f"- {criteria.item_name} ({criteria.max_score}分)") - - prompt = f""" -为以下大类别生成专业的标书子标题： - -【大类别】: {parent_chapter.title} -【评分项】: -{chr(10).join(criteria_info)} - -生成要求： -1. 为每个评分项生成对应的子标题名称（不要包含编号） -2. 重要评分项可添加三级子标题（不要包含编号） -3. 只返回标题文本，编号由Word自动管理 - -返回JSON格式： -{{ - "sub_chapters": [ - {{"title": "技术架构设计", "level": 2, "score": 5, "children": []}} - ] -}} - -只返回JSON：""" - - # 使用统一的LLM服务 - response = LLMService().call(prompt) - if not response: - raise ValueError("AI生成子标题失败: API无响应") - - # 解析响应 - result_data = self._parse_ai_response(response) - sub_chapters_data = result_data.get("sub_chapters", []) - - # 构建子章节对象 - sub_chapters = [] - for i, sub_data in enumerate(sub_chapters_data, 1): - title = sub_data.get("title", f"子标题{i}") - - sub_chapter = DocumentChapter( - id=f"{parent_chapter.id}_sub_{i:02d}", - title=title, # 不添加编号 - level=sub_data.get("level", 2), - score=sub_data.get("score", 0), - template_placeholder=f"{{{{{parent_chapter.id}_sub_{i:02d}_content}}}}" - ) - - # 处理三级标题 - for j, child_data in enumerate(sub_data.get("children", []), 1): - child_title = child_data.get("title", f"三级标题{j}") - - child_chapter = DocumentChapter( - id=f"{parent_chapter.id}_sub_{i:02d}_{j:02d}", - title=child_title, # 不添加编号 - level=child_data.get("level", 3), - template_placeholder=f"{{{{{parent_chapter.id}_sub_{i:02d}_{j:02d}_content}}}}" - ) - sub_chapter.children.append(child_chapter) - - sub_chapters.append(sub_chapter) - - return sub_chapters - - except Exception as e: - logger.error(f"AI生成子标题失败: {e}") - # 立即失败，不提供后备方案 - raise - - def _parse_ai_response(self, response: str) -> Dict[str, Any]: - """解析AI响应 - - Args: - response: AI响应文本 - - Returns: - 解析后的数据 - - Raises: - ValueError: 解析失败时抛出 - """ - try: - clean_response = response.strip() - if clean_response.startswith("```json"): - clean_response = clean_response[7:] - if clean_response.endswith("```"): - clean_response = clean_response[:-3] - - return json.loads(clean_response.strip()) - - except (json.JSONDecodeError, KeyError) as e: - raise ValueError(f"解析AI响应失败: {e}") - - def _generate_template_sub_chapters(self, - criteria: ScoringCriteria, - parent_chapter: DocumentChapter, - template_file: Optional[str]) -> List[DocumentChapter]: - """基于模板生成子标题 - - Args: - criteria: 评分项（仅作参考） - parent_chapter: 父章节 - template_file: 模板文件路径 - - Returns: - 子章节列表 - """ - # 提供默认结构（不包含编号） - default_sub_chapters = [ - DocumentChapter( - id=f"{parent_chapter.id}_def_01", - title="方案概述", # 不包含编号 - level=2, - template_placeholder=f"{{{{{parent_chapter.id}_def_01_content}}}}" - ), - DocumentChapter( - id=f"{parent_chapter.id}_def_02", - title="具体实施", # 不包含编号 - level=2, - template_placeholder=f"{{{{{parent_chapter.id}_def_02_content}}}}" - ), - DocumentChapter( - id=f"{parent_chapter.id}_def_03", - title="保障措施", # 不包含编号 - level=2, - template_placeholder=f"{{{{{parent_chapter.id}_def_03_content}}}}" - ) - ] - - logger.info(f"使用默认模板为 {parent_chapter.title} 生成子标题") - return default_sub_chapters \ No newline at end of file + return chapter \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/group_criteria.py b/src/bidmaster/nodes/toc/group_criteria.py index 8017495..f0b4de2 100644 --- a/src/bidmaster/nodes/toc/group_criteria.py +++ b/src/bidmaster/nodes/toc/group_criteria.py @@ -8,6 +8,7 @@ from typing import Dict, List, Any from ..base import BaseNode, NodeContext from ...tools.parser import ScoringCriteria +from .category_manager import CategoryManager logger = logging.getLogger(__name__) @@ -25,67 +26,16 @@ class GroupCriteriaNode(BaseNode): def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: """执行分组逻辑""" - self._log_start() + technical_criteria = state.get("technical_criteria", []) - try: - technical_criteria = state.get("technical_criteria", []) + if not technical_criteria: + raise ValueError("缺少技术评分项") - if not technical_criteria: - raise ValueError("缺少技术评分项") + # 执行分组 + category_groups = CategoryManager.group_criteria_by_category(technical_criteria) - # 执行分组 - category_groups = self._group_by_category(technical_criteria) + logger.info(f"分组完成: {len(category_groups)}个类别") - logger.info(f"分组完成: {len(category_groups)}个类别") + # 更新状态 + return self._update_state(state, category_groups=category_groups) - # 更新状态 - state["category_groups"] = category_groups - state["current_step"] = self.name - - self._log_success() - return state - - except Exception as e: - self._log_error(e) - state["error"] = str(e) - state["should_continue"] = False - raise - - def _group_by_category(self, criteria: List[ScoringCriteria]) -> Dict[str, List[ScoringCriteria]]: - """按类别分组评分项 - - Args: - criteria: 评分项列表 - - Returns: - 分组后的字典，key为类别，value为该类别下的评分项列表 - """ - # 预定义类别顺序 - category_groups = { - "technical_solution": [], - "equipment_spec": [], - "implementation": [], - "quality_safety": [], - "after_sales": [], - "compliance": [] - } - - # 分组 - for criterion in criteria: - category_key = criterion.category.value - if category_key in category_groups: - category_groups[category_key].append(criterion) - else: - # 未知类别默认归入技术方案 - category_groups["technical_solution"].append(criterion) - logger.warning(f"未知类别 {category_key}，归入技术方案类别") - - # 只保留有评分项的类别 - filtered_groups = {k: v for k, v in category_groups.items() if v} - - # 记录分组统计 - for category, items in filtered_groups.items(): - total_score = sum(item.max_score for item in items) - logger.info(f"类别 {category}: {len(items)}项，总分 {total_score}") - - return filtered_groups \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/llm_helper.py b/src/bidmaster/nodes/toc/llm_helper.py new file mode 100644 index 0000000..9d84920 --- /dev/null +++ b/src/bidmaster/nodes/toc/llm_helper.py @@ -0,0 +1,202 @@ +"""LLM调用辅助类 + +统一管理LLM调用、响应解析、错误处理等逻辑。 +""" + +import json +import logging +from typing import Dict, List, Any, Optional + +from ...tools.llm import LLMService +from ...tools.parser import ScoringCriteria, DocumentChapter + +logger = logging.getLogger(__name__) + +# 避免循环导入，在需要时导入 +def _get_category_manager(): + from .category_manager import CategoryManager + return CategoryManager + + +class LLMHelper: + """LLM调用辅助类 + + 提供统一的LLM调用、响应解析方法。 + """ + + @staticmethod + def parse_ai_json_response(response: str) -> Dict[str, Any]: + """解析AI响应的JSON格式 + + Args: + response: AI响应文本 + + Returns: + 解析后的数据字典 + + Raises: + ValueError: 解析失败时抛出 + """ + try: + clean_response = response.strip() + + # 清理markdown格式 + if clean_response.startswith("```json"): + clean_response = clean_response[7:] + if clean_response.endswith("```"): + clean_response = clean_response[:-3] + + return json.loads(clean_response.strip()) + + except (json.JSONDecodeError, KeyError) as e: + raise ValueError(f"解析AI响应失败: {e}") + + @staticmethod + def call_llm_with_retry(prompt: str, max_retries: int = 2) -> Optional[str]: + """带重试的LLM调用 + + Args: + prompt: 提示词 + max_retries: 最大重试次数 + + Returns: + LLM响应，失败时返回None + """ + for attempt in range(max_retries + 1): + try: + response = LLMService().call(prompt) + if response: + return response + logger.warning(f"LLM调用第{attempt + 1}次无响应") + except Exception as e: + logger.error(f"LLM调用第{attempt + 1}次失败: {e}") + + return None + + @staticmethod + def generate_sub_chapters_ai(criteria_list: List[ScoringCriteria], + parent_chapter: DocumentChapter) -> Optional[List[Dict[str, Any]]]: + """AI生成子章节数据 + + Args: + criteria_list: 对应的评分项列表 + parent_chapter: 父章节 + + Returns: + 子章节数据列表，失败时返回None + """ + # 构建评分项信息 + criteria_info = [] + for criteria in criteria_list: + criteria_info.append(f"- {criteria.item_name} ({criteria.max_score}分)") + + prompt = f""" +为以下大类别生成专业的标书子标题： + +【大类别】: {parent_chapter.title} +【评分项】: +{chr(10).join(criteria_info)} + +生成要求： +1. 为每个评分项生成对应的子标题名称（不要包含编号） +2. 重要评分项可添加三级子标题（不要包含编号） +3. 只返回标题文本，编号由Word自动管理 + +返回JSON格式： +{{ + "sub_chapters": [ + {{"title": "技术架构设计", "level": 2, "score": 5, "children": []}} + ] +}} + +只返回JSON：""" + + try: + response = LLMHelper.call_llm_with_retry(prompt) + if not response: + return None + + result_data = LLMHelper.parse_ai_json_response(response) + return result_data.get("sub_chapters", []) + + except Exception as e: + logger.error(f"AI生成子标题失败: {e}") + return None + + @staticmethod + def review_structure_ai(technical_criteria: List[ScoringCriteria], + preliminary_chapters: List[DocumentChapter]) -> Dict[str, Any]: + """AI审查目录结构 + + Args: + technical_criteria: 技术评分项 + preliminary_chapters: 初步生成的章节 + + Returns: + 审查结果字典 + """ + # 构建审查提示词 + CategoryManager = _get_category_manager() + criteria_summary = CategoryManager.format_criteria_summary(technical_criteria) + chapters_summary = LLMHelper._format_chapters_summary(preliminary_chapters) + + review_prompt = f""" +请审查这个标书目录结构的合理性和完整性。 + +【技术评分项分布】: +{criteria_summary} + +【当前生成的章节结构】: +{chapters_summary} + +【审查要求】: +1. 是否缺少重要的标准章节？ +2. 章节顺序是否合理？ +3. 每个评分项是否都有对应章节？ + +返回JSON格式： +{{ + "overall_assessment": "总体评价", + "suggestions": [ + {{"type": "add/modify/reorder", "description": "建议内容", "priority": "high/medium/low"}} + ], + "optimization_score": 85 +}} + +只返回JSON：""" + + try: + response = LLMHelper.call_llm_with_retry(review_prompt) + if response: + return LLMHelper.parse_ai_json_response(response) + else: + return {"overall_assessment": "AI审查失败", "suggestions": []} + + except Exception as e: + logger.error(f"AI审查失败: {e}") + return {"overall_assessment": "AI审查异常", "suggestions": []} + + @staticmethod + def _format_chapters_summary(chapters: List[DocumentChapter]) -> str: + """格式化章节摘要用于显示 + + Args: + chapters: 章节列表 + + Returns: + 格式化后的字符串 + """ + lines = [] + + for chapter in chapters: + lines.append(chapter.title) + + # 显示子章节 + for sub in chapter.children: + lines.append(f" {sub.title}") + + # 显示三级章节 + for child in sub.children: + lines.append(f" {child.title}") + + return "\n".join(lines) \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/review_structure.py b/src/bidmaster/nodes/toc/review_structure.py index 28265fa..5903d1d 100644 --- a/src/bidmaster/nodes/toc/review_structure.py +++ b/src/bidmaster/nodes/toc/review_structure.py @@ -3,26 +3,15 @@ 使用AI对生成的目录结构进行合理性和完整性审查。 """ -import json import logging from typing import Dict, List, Any from ..base import BaseNode, NodeContext from ...tools.parser import ScoringCriteria, DocumentChapter -from ...tools.llm import LLMService +from .llm_helper import LLMHelper logger = logging.getLogger(__name__) -# 类别名称映射（用于审查显示） -CATEGORY_NAMES = { - "compliance": "合规响应", - "technical_solution": "技术方案", - "equipment_spec": "设备规格", - "quality_safety": "质量安全", - "after_sales": "售后服务", - "implementation": "实施方案" -} - class ReviewStructureNode(BaseNode): """AI审查目录结构的节点""" @@ -37,164 +26,21 @@ class ReviewStructureNode(BaseNode): def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: """执行AI审查""" - self._log_start() + technical_criteria = state.get("technical_criteria", []) + preliminary_chapters = state.get("preliminary_chapters", []) - try: - technical_criteria = state.get("technical_criteria", []) - preliminary_chapters = state.get("preliminary_chapters", []) + if not preliminary_chapters: + raise ValueError("缺少章节数据进行审查") - if not preliminary_chapters: - raise ValueError("缺少章节数据进行审查") + # 执行AI审查 + review_result = LLMHelper.review_structure_ai(technical_criteria, preliminary_chapters) - # 执行AI审查 - review_result = self._perform_ai_review(technical_criteria, preliminary_chapters) + # 根据审查结果添加警告 + warnings = state.setdefault("warnings", []) + if review_result.get("suggestions"): + suggestion_count = len(review_result["suggestions"]) + warnings.append(f"AI审查: {suggestion_count}条优化建议") - # 更新状态 - state["structure_review"] = review_result - state["current_step"] = self.name + logger.info(f"AI审查完成，优化评分: {review_result.get('optimization_score', 'N/A')}") - # 根据审查结果添加警告 - if review_result.get("suggestions"): - suggestion_count = len(review_result["suggestions"]) - state.setdefault("warnings", []).append(f"AI审查: {suggestion_count}条优化建议") - - logger.info(f"AI审查完成，优化评分: {review_result.get('optimization_score', 'N/A')}") - self._log_success() - return state - - except Exception as e: - self._log_error(e) - # AI审查失败不中断流程，只记录警告 - logger.warning(f"AI审查失败: {e}") - state.setdefault("warnings", []).append(f"AI审查跳过: {str(e)}") - state["structure_review"] = {} - state["current_step"] = self.name - return state - - def _perform_ai_review(self, - technical_criteria: List[ScoringCriteria], - preliminary_chapters: List[DocumentChapter]) -> Dict[str, Any]: - """执行AI审查 - - Args: - technical_criteria: 技术评分项 - preliminary_chapters: 初步生成的章节 - - Returns: - 审查结果字典 - """ - # 构建审查提示词 - criteria_summary = self._format_criteria_for_review(technical_criteria) - chapters_summary = self._format_chapters_for_review(preliminary_chapters) - - review_prompt = f""" -请审查这个标书目录结构的合理性和完整性。 - -【技术评分项分布】: -{criteria_summary} - -【当前生成的章节结构】: -{chapters_summary} - -【审查要求】: -1. 是否缺少重要的标准章节？ -2. 章节顺序是否合理？ -3. 每个评分项是否都有对应章节？ - -返回JSON格式： -{{ - "overall_assessment": "总体评价", - "suggestions": [ - {{"type": "add/modify/reorder", "description": "建议内容", "priority": "high/medium/low"}} - ], - "optimization_score": 85 -}} - -只返回JSON：""" - - try: - # 使用统一的LLM服务 - response = LLMService().call(review_prompt) - - if response: - return self._parse_review_response(response) - else: - return {"overall_assessment": "AI审查失败", "suggestions": []} - - except Exception as e: - logger.error(f"AI审查API调用失败: {e}") - return {"overall_assessment": "AI审查异常", "suggestions": []} - - def _parse_review_response(self, response: str) -> Dict[str, Any]: - """解析AI审查响应 - - Args: - response: AI响应文本 - - Returns: - 解析后的审查结果 - """ - try: - clean_response = response.strip() - if clean_response.startswith("```json"): - clean_response = clean_response[7:] - if clean_response.endswith("```"): - clean_response = clean_response[:-3] - - return json.loads(clean_response.strip()) - - except json.JSONDecodeError as e: - logger.error(f"解析AI审查响应失败: {e}") - return {"overall_assessment": "解析失败", "suggestions": []} - - def _format_criteria_for_review(self, technical_criteria: List[ScoringCriteria]) -> str: - """格式化评分项用于审查 - - Args: - technical_criteria: 技术评分项列表 - - Returns: - 格式化后的字符串 - """ - lines = [] - - # 按类别分组 - category_groups = {} - for criteria in technical_criteria: - category = criteria.category.value - if category not in category_groups: - category_groups[category] = [] - category_groups[category].append(criteria) - - # 格式化输出 - for category, items in category_groups.items(): - category_name = CATEGORY_NAMES.get(category, category) - lines.append(f"【{category_name}】({len(items)}项):") - for item in items: - lines.append(f" - {item.item_name} ({item.max_score}分)") - - return "\n".join(lines) - - def _format_chapters_for_review(self, chapters: List[DocumentChapter]) -> str: - """格式化章节用于审查 - - Args: - chapters: 章节列表 - - Returns: - 格式化后的字符串 - """ - lines = [] - - for chapter in chapters: - lines.append(chapter.title) - - # 显示子章节 - for sub in chapter.children: - lines.append(f" {sub.title}") - - # 显示三级章节 - for child in sub.children: - lines.append(f" {child.title}") - - return "\n".join(lines) \ No newline at end of file + return self._update_state(state, structure_review=review_result, warnings=warnings) \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/utils.py b/src/bidmaster/nodes/toc/utils.py new file mode 100644 index 0000000..413eb21 --- /dev/null +++ b/src/bidmaster/nodes/toc/utils.py @@ -0,0 +1,16 @@ +"""目录生成相关工具函数 + +保留少量仍需要的通用工具函数。 +大部分功能已移至专门的管理器和辅助类中。 +""" + +import logging + +logger = logging.getLogger(__name__) + +# 此文件中的函数已迁移至以下模块： +# - 类别相关功能 -> category_manager.py +# - LLM调用和解析 -> llm_helper.py +# - 章节创建 -> factories.py +# +# 如需添加新的通用工具函数，请考虑是否应该归属于上述专门模块。 \ No newline at end of file