diff --git a/src/bidmaster/agents/__init__.py b/src/bidmaster/agents/__init__.py index 3a1a902..28d1f01 100644 --- a/src/bidmaster/agents/__init__.py +++ b/src/bidmaster/agents/__init__.py @@ -1 +1,6 @@ -# Agent层 - LangGraph编排 \ No newline at end of file +"""Agent层 - LangGraph编排""" + +from .analysis import AnalysisAgent +from .toc_generator import TocGeneratorAgent + +__all__ = ["AnalysisAgent", "TocGeneratorAgent"] \ No newline at end of file diff --git a/src/bidmaster/agents/analysis.py b/src/bidmaster/agents/analysis.py index aaf4a05..8ff3c17 100644 --- a/src/bidmaster/agents/analysis.py +++ b/src/bidmaster/agents/analysis.py @@ -236,9 +236,9 @@ def parse_content_node(state: AnalysisAgentState) -> AnalysisAgentState: return state -def generate_dynamic_structure_node(state: AnalysisAgentState) -> AnalysisAgentState: - """节点5：基于评分项类别动态生成章节结构""" - logger.info("开始动态生成章节结构...") +def generate_toc_with_agent_node(state: AnalysisAgentState) -> AnalysisAgentState: + """节点5：使用TocGeneratorAgent生成目录结构""" + logger.info("开始使用TocGeneratorAgent生成目录...") try: technical_criteria = state["technical_criteria"] @@ -246,96 +246,6 @@ def generate_dynamic_structure_node(state: AnalysisAgentState) -> AnalysisAgentS if not technical_criteria: raise ValueError("缺少技术评分项，无法生成章节结构") - # 按评分项类别分组 - category_groups = _group_criteria_by_category(technical_criteria) - - # 动态生成章节（只为有评分项的类别生成章节） - preliminary_chapters = _create_dynamic_chapters(category_groups, technical_criteria) - - if not preliminary_chapters: - raise ValueError("无法生成有效的章节结构") - - logger.info(f"初步章节生成完成: {len(preliminary_chapters)}个章节") - - state["preliminary_chapters"] = preliminary_chapters - state["current_step"] = "generate_dynamic_structure" - state["progress"] = 0.7 - - except Exception as e: - logger.error(f"动态章节生成失败: {e}") - state["error"] = str(e) - state["should_continue"] = False - - return state - - -def map_criteria_node(state: AnalysisAgentState) -> AnalysisAgentState: - """节点6：映射评分项到章节""" - logger.info("开始映射评分项到章节...") - - try: - # 创建临时BidStructure用于映射 - from ..tools.parser import BidStructure - temp_structure = BidStructure( - scoring_criteria=state["technical_criteria"], - chapters=state["preliminary_chapters"] - ) - - # 执行映射 - _map_criteria_to_dynamic_chapters(temp_structure) - - # 更新状态 - state["technical_criteria"] = temp_structure.scoring_criteria - state["current_step"] = "map_criteria" - state["progress"] = 0.75 - - logger.info("评分项映射完成") - - except Exception as e: - logger.error(f"评分项映射失败: {e}") - state["error"] = str(e) - state["should_continue"] = False - - return state - - -def ai_review_structure_node(state: AnalysisAgentState) -> AnalysisAgentState: - """节点8：AI审查章节结构的合理性""" - logger.info("开始AI审查章节结构...") - - try: - technical_criteria = state["technical_criteria"] - preliminary_chapters = state["preliminary_chapters"] - - # 调用AI审查章节结构 - review_result = _ai_review_chapter_structure(technical_criteria, preliminary_chapters) - - state["structure_review"] = review_result - state["current_step"] = "ai_review_structure" - state["progress"] = 0.85 - - # 添加审查信息到警告 - if review_result.get("suggestions"): - state["warnings"].append(f"AI结构审查: {len(review_result['suggestions'])}条优化建议") - - logger.info("AI章节结构审查完成") - - except Exception as e: - logger.error(f"AI结构审查失败: {e}") - state["error"] = str(e) - state["should_continue"] = False - - return state - - -def generate_sub_chapters_node(state: AnalysisAgentState) -> AnalysisAgentState: - """节点7：为一级章节生成子标题""" - logger.info("=== 进入子标题生成节点 ===") - - try: - preliminary_chapters = state["preliminary_chapters"] - technical_criteria = state["technical_criteria"] - # 询问用户选择生成方式 from rich.console import Console import click @@ -345,58 +255,69 @@ def generate_sub_chapters_node(state: AnalysisAgentState) -> AnalysisAgentState: console.print("1. AI智能生成子标题（推荐）- 根据评分项要求智能生成2-3级标题") console.print("2. 基于模板生成 - 使用预定义模板结构") + generation_mode = "ai" + template_file = None + try: choice = click.prompt("请输入选择", type=click.Choice(['1', '2']), show_choices=False) + if choice == '2': + generation_mode = "template" + while True: + template_path = click.prompt("请输入模板文件路径（.docx格式）", type=str) + from pathlib import Path + if Path(template_path).exists() and template_path.lower().endswith('.docx'): + template_file = template_path + console.print(f"✅ 模板文件: {template_file}", style="green") + break + else: + console.print("❌ 文件不存在或格式不正确，请重新输入", style="red") except Exception as e: logger.error(f"用户交互失败: {e}") state["error"] = f"无法获取用户输入: {e}" state["should_continue"] = False return state - template_file = None - if choice == '2': - while True: - template_path = click.prompt("请输入模板文件路径（.docx格式）", type=str) - from pathlib import Path - if Path(template_path).exists() and template_path.lower().endswith('.docx'): - template_file = template_path - console.print(f"✅ 模板文件: {template_file}", style="green") - break - else: - console.print("❌ 文件不存在或格式不正确，请重新输入", style="red") + # 调用TocGeneratorAgent + from .toc_generator import TocGeneratorAgent + toc_agent = TocGeneratorAgent() + result = toc_agent.generate_sync( + technical_criteria=technical_criteria, + generation_mode=generation_mode, + template_file=template_file + ) - # 为每个一级章节生成子标题 - enhanced_chapters = [] - for chapter in preliminary_chapters: - # 找到对应的评分项（可能有多个） - corresponding_criteria_list = [ - criteria for criteria in technical_criteria - if criteria.chapter_id == chapter.id - ] + if not result.success: + raise ValueError(f"目录生成失败: {result.error_message}") - logger.info(f"章节 {chapter.id} 匹配到 {len(corresponding_criteria_list)} 个评分项") + # 映射评分项到生成的章节 + chapters = result.chapters + for criteria in technical_criteria: + # 根据类别找到对应章节 + category = criteria.category.value + for chapter in chapters: + if "_" in chapter.id: + parts = chapter.id.split("_") + if len(parts) >= 3: + chapter_category = "_".join(parts[2:]) + if chapter_category == category: + criteria.chapter_id = chapter.id + break - if corresponding_criteria_list: - if choice == '1': - # AI智能生成子标题，传入该章节下的所有评分项 - sub_chapters = _generate_ai_sub_chapters(corresponding_criteria_list, chapter) - else: - # 基于模板生成子标题，使用第一个评分项作为参考 - sub_chapters = _generate_template_sub_chapters(corresponding_criteria_list[0], chapter, template_file) + logger.info(f"目录生成完成: {len(chapters)}个章节") - chapter.children = sub_chapters - logger.info(f"章节 {chapter.title} 基于 {len(corresponding_criteria_list)} 个评分项生成了 {len(sub_chapters)} 个子标题") + # 更新状态 + state["preliminary_chapters"] = chapters + state["technical_criteria"] = technical_criteria + state["structure_review"] = {} # TocGeneratorAgent已包含审查 + state["current_step"] = "generate_toc" + state["progress"] = 0.85 - enhanced_chapters.append(chapter) - - state["preliminary_chapters"] = enhanced_chapters - state["current_step"] = "generate_sub_chapters" - state["progress"] = 0.82 - - logger.info(f"子标题生成完成，共处理{len(enhanced_chapters)}个章节") + # 添加警告信息 + if result.warnings: + state["warnings"].extend(result.warnings) except Exception as e: - logger.error(f"子标题生成失败: {e}") + logger.error(f"目录生成失败: {e}") state["error"] = str(e) state["should_continue"] = False @@ -404,33 +325,28 @@ def generate_sub_chapters_node(state: AnalysisAgentState) -> AnalysisAgentState: def finalize_structure_node(state: AnalysisAgentState) -> AnalysisAgentState: - """节点9：根据AI审查结果最终确定章节结构（支持用户交互选择）""" - logger.info("开始最终确定章节结构...") + """节点6：最终确定标书结构""" + logger.info("开始最终确定标书结构...") try: technical_criteria = state["technical_criteria"] preliminary_chapters = state["preliminary_chapters"] - structure_review = state["structure_review"] - - # 应用AI审查建议（自动应用高优先级） - final_chapters = _apply_review_suggestions_without_interaction(preliminary_chapters, structure_review) - - # 确保核心章节存在 - final_chapters = _ensure_core_chapters(final_chapters, technical_criteria) + structure_review = state.get("structure_review", {}) # 创建最终的标书结构 bid_structure = BidStructure( project_name=f"标书项目-{Path(state['source_file']).stem}", scoring_criteria=technical_criteria, deviation_items=state["deviation_items"], - chapters=final_chapters, + chapters=preliminary_chapters, # 使用TocGeneratorAgent生成的章节 scoring_file=state["source_file"] ) - # 保存AI审查结果到bid_structure，供CLI层使用 - bid_structure.structure_review = structure_review + # 保存审查结果（如果有） + if structure_review: + bid_structure.structure_review = structure_review - logger.info(f"最终章节结构确定完成: {len(final_chapters)}个章节") + logger.info(f"标书结构确定完成: {len(preliminary_chapters)}个章节") state["bid_structure"] = bid_structure state["current_step"] = "finalize_structure" @@ -438,341 +354,14 @@ def finalize_structure_node(state: AnalysisAgentState) -> AnalysisAgentState: state["should_continue"] = False # 完成 except Exception as e: - logger.error(f"最终结构确定失败: {e}") + logger.error(f"标书结构确定失败: {e}") state["error"] = str(e) state["should_continue"] = False return state -# ========== 动态章节生成辅助函数 ========== - -def _group_criteria_by_category(technical_criteria: List[ScoringCriteria]) -> Dict[str, List[ScoringCriteria]]: - """按类别对技术评分项进行分组""" - category_groups = { - "technical_solution": [], - "equipment_spec": [], - "implementation": [], - "quality_safety": [], - "after_sales": [], - "compliance": [] - } - - for criteria in technical_criteria: - category_key = criteria.category.value - if category_key in category_groups: - category_groups[category_key].append(criteria) - else: - # 未知类别归到技术方案 - category_groups["technical_solution"].append(criteria) - - # 只返回有评分项的类别 - return {k: v for k, v in category_groups.items() if v} - - -def _create_dynamic_chapters(category_groups: Dict[str, List[ScoringCriteria]], - technical_criteria: List[ScoringCriteria]) -> List[DocumentChapter]: - """按大类别创建一级章节，每个类别一个章节""" - chapters = [] - chapter_index = 1 - - # 类别名称映射 - category_names = { - "compliance": "合规响应", - "technical_solution": "技术方案", - "equipment_spec": "设备规格", - "quality_safety": "质量安全", - "after_sales": "售后服务", - "implementation": "实施方案" - } - - # 按technical_criteria原始列表顺序确定章节顺序 - # 找到每个类别在technical_criteria中的第一个出现位置 - def get_category_first_index(category): - 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) - - # 为每个有评分项的大类别创建一级章节 - for category in sorted_categories: - criteria_list = category_groups[category] - if not criteria_list: - continue - - # 计算该类别总分值 - total_score = sum(c.max_score for c in criteria_list) - - chapter_id = f"chapter_{chapter_index:02d}_{category}" - category_name = category_names.get(category, category) - chapter_title = f"{chapter_index}. {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}}}}" - ) - - chapters.append(chapter) - chapter_index += 1 - - return chapters - - -def _ai_review_chapter_structure(technical_criteria: List[ScoringCriteria], - preliminary_chapters: List[DocumentChapter]) -> Dict[str, Any]: - """AI审查章节结构的合理性""" - try: - from ..tools.parser import BidParser - parser = BidParser() - - # 构建审查提示词 - criteria_summary = _format_criteria_for_review(technical_criteria) - chapters_summary = _format_chapters_for_review(preliminary_chapters) - - review_prompt = f""" -请审查这个标书章节结构的合理性和完整性。 - -【设计策略】: -- 每个技术评分项对应一个独立章节，确保充分展示每个评分要素 -- 章节顺序优先遵循招标文件中评分表的原始顺序 -- 只有当招标文件中顺序不明确时，才建议按照技术逻辑重新排序 - -【技术评分项分布】: -{criteria_summary} - -【当前生成的章节结构】: -{chapters_summary} - -【审查要求】: -1. 结构完整性检查： - - 是否缺少重要的标准章节（如评标索引表等）？ - - 每个评分项是否都有对应的独立章节？ - -2. 目录顺序合理性审查： - - 优先检查：当前顺序是否遵循了招标文件评分表的原始顺序？ - - 次要检查：如果评分表内顺序不够明确，是否需要按技术逻辑调整？ - - 逻辑顺序参考：合规资质 → 架构设计 → 功能实现 → 系统集成 → 质量测试 → 售后服务 - - 重要：不要随意改变招标方设定的评分顺序 - -3. 章节标题优化： - - 标题是否清晰专业？ - - 是否需要调整表述以更符合标书规范？ - -4. 标书规范性： - - 章节编号是否连续规范？ - - 整体结构是否符合招投标文件要求？ - -注意：我们采用一一对应策略，每个评分项都应该有独立章节来充分展示内容。 - -请返回JSON格式的审查结果： -{{ - "overall_assessment": "总体评价", - "missing_chapters": ["缺少的章节列表"], - "suggestions": [ - {{"type": "add", "description": "建议添加的内容", "priority": "high/medium/low"}}, - {{"type": "modify", "description": "建议修改的内容", "priority": "high/medium/low"}}, - {{"type": "reorder", "description": "建议调整的顺序", "priority": "high/medium/low"}} - ], - "optimization_score": 85 -}} - -只返回JSON，无其他文字：""" - - # 调用AI获取审查结果 - response = parser._call_llm_api(review_prompt) - - if not response: - return {"overall_assessment": "AI审查失败", "suggestions": [], "optimization_score": 0} - - # 解析AI响应 - import json - try: - clean_response = response.strip() - if clean_response.startswith("```json"): - clean_response = clean_response[7:] - if clean_response.endswith("```"): - clean_response = clean_response[:-3] - clean_response = clean_response.strip() - - review_result = json.loads(clean_response) - return review_result - - except json.JSONDecodeError: - logger.error(f"解析AI审查响应失败: {response}") - return {"overall_assessment": "响应解析失败", "suggestions": [], "optimization_score": 0} - - except Exception as e: - logger.error(f"AI结构审查异常: {e}") - return {"overall_assessment": f"审查异常: {str(e)}", "suggestions": [], "optimization_score": 0} - - -def _apply_review_suggestions_without_interaction(preliminary_chapters: List[DocumentChapter], - structure_review: Dict[str, Any]) -> List[DocumentChapter]: - """自动应用高优先级AI审查建议（无用户交互）""" - final_chapters = preliminary_chapters.copy() - suggestions = structure_review.get("suggestions", []) - - if not suggestions: - return final_chapters - - # 只自动应用高优先级建议 - applied_count = 0 - for suggestion in suggestions: - if suggestion.get("priority") == "high": - suggestion_type = suggestion.get("type", "") - if suggestion_type == "add": - _apply_add_suggestion(final_chapters, suggestion) - applied_count += 1 - elif suggestion_type == "reorder": - _apply_reorder_suggestion(final_chapters, suggestion) - applied_count += 1 - elif suggestion_type == "modify": - _apply_modify_suggestion(final_chapters, suggestion) - applied_count += 1 - - logger.info(f"自动应用了 {applied_count} 条高优先级AI审查建议") - return final_chapters - - -def _ensure_core_chapters(chapters: List[DocumentChapter], - technical_criteria: List[ScoringCriteria]) -> List[DocumentChapter]: - """确保核心章节存在""" - # 检查是否存在评标索引表 - has_index_table = any("评标索引表" in ch.title or "评分索引" in ch.title for ch in chapters) - - if not has_index_table: - # 添加评标索引表作为第一章 - index_chapter = DocumentChapter( - id="evaluation_index", - title="1. 评标索引表（技术评分完全对应）", - level=1, - template_placeholder="{{evaluation_index_content}}" - ) - chapters.insert(0, index_chapter) - - # 重新编号后续章节 - for i, chapter in enumerate(chapters[1:], 2): - chapter.title = chapter.title.replace(f"{i-1}.", f"{i}.") - - return chapters - - -def _map_criteria_to_dynamic_chapters(bid_structure: BidStructure) -> None: - """将评分项映射到对应的大类别章节""" - logger.info("开始映射评分项到章节") - - # 创建类别到章节ID的映射 - category_to_chapter = {} - logger.info(f"当前有 {len(bid_structure.chapters)} 个章节:") - for chapter in bid_structure.chapters: - logger.info(f" 章节ID: {chapter.id}, 标题: {chapter.title}") - # 从章节ID提取类别 (chapter_01_compliance -> compliance) - if "_" in chapter.id: - parts = chapter.id.split("_") - if len(parts) >= 3: - category = "_".join(parts[2:]) # 支持多段类别名 - category_to_chapter[category] = chapter.id - logger.info(f" -> 映射类别 {category} 到章节 {chapter.id}") - - logger.info(f"类别到章节映射: {category_to_chapter}") - - # 映射评分项到对应的大类别章节 - logger.info(f"开始映射 {len(bid_structure.scoring_criteria)} 个评分项:") - for criteria in bid_structure.scoring_criteria: - category = criteria.category.value - logger.info(f"评分项 '{criteria.item_name}' 类别: {category}") - if category in category_to_chapter: - old_id = criteria.chapter_id - criteria.chapter_id = category_to_chapter[category] - logger.info(f" -> 映射成功: {old_id} → {criteria.chapter_id}") - else: - logger.error(f"评分项 {criteria.item_name} 的类别 {category} 未找到对应章节") - logger.error(f"可用类别: {list(category_to_chapter.keys())}") - # 按编码规范：暴露问题，不掩盖错误 - raise ValueError(f"评分项类别 {category} 未找到对应章节") - - logger.info("评分项映射完成") - - -def _format_criteria_for_review(technical_criteria: List[ScoringCriteria]) -> str: - """格式化评分项用于AI审查""" - lines = [] - category_names = { - "technical_solution": "技术方案", - "equipment_spec": "设备规格", - "implementation": "实施方案", - "quality_safety": "质量安全", - "after_sales": "售后服务", - "compliance": "合规响应" - } - - 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}分)") - lines.append("") - - return "\n".join(lines) - - -def _format_chapters_for_review(chapters: List[DocumentChapter]) -> str: - """格式化章节结构用于AI审查""" - lines = [] - for chapter in chapters: - lines.append(f"{chapter.title}") - for sub_chapter in chapter.children: - lines.append(f" {sub_chapter.title}") - return "\n".join(lines) - - -def _apply_add_suggestion(chapters: List[DocumentChapter], suggestion: Dict[str, Any]) -> None: - """应用添加建议""" - description = suggestion.get("description", "") - if "评标索引表" in description or "索引表" in description: - # 已在 _ensure_core_chapters 中处理 - pass - - -def _apply_reorder_suggestion(chapters: List[DocumentChapter], suggestion: Dict[str, Any]) -> None: - """应用重新排序建议""" - description = suggestion.get("description", "").lower() - - # 严格遵循招标文件评分表原始顺序，不随意重排 - # 只有当明确检测到评分表顺序混乱时才进行最小调整 - if "原始顺序" in description and "混乱" in description: - logger.info("检测到评分表顺序混乱，进行最小调整") - # 这里可以添加基于评分项原始出现顺序的排序逻辑 - # 目前保持现有顺序，避免破坏招标文件原意 - pass - else: - logger.info("保持招标文件评分表原始顺序，不进行重排") - - # 章节编号将在Word生成时动态计算，这里不处理编号 - - -def _apply_modify_suggestion(chapters: List[DocumentChapter], suggestion: Dict[str, Any]) -> None: - """应用修改建议（如标题优化）""" - description = suggestion.get("description", "") - - # 简化实现：目前不做具体的标题修改 - # 在实际应用中，可以根据具体建议内容修改章节标题 - pass +# ========== 辅助函数 =========== @@ -823,10 +412,7 @@ class AnalysisAgent: workflow.add_node("extract_tables", extract_tables_node) workflow.add_node("classify_tables", classify_tables_node) workflow.add_node("parse_content", parse_content_node) - workflow.add_node("generate_dynamic_structure", generate_dynamic_structure_node) - workflow.add_node("map_criteria", map_criteria_node) - workflow.add_node("generate_sub_chapters", generate_sub_chapters_node) - workflow.add_node("ai_review_structure", ai_review_structure_node) + workflow.add_node("generate_toc", generate_toc_with_agent_node) # 使用新的目录生成节点 workflow.add_node("finalize_structure", finalize_structure_node) # 设置入口点 @@ -864,40 +450,13 @@ class AnalysisAgent: "parse_content", should_continue_processing, { - "continue": "generate_dynamic_structure", + "continue": "generate_toc", "end": END } ) workflow.add_conditional_edges( - "generate_dynamic_structure", - should_continue_processing, - { - "continue": "map_criteria", - "end": END - } - ) - - workflow.add_conditional_edges( - "map_criteria", - should_continue_processing, - { - "continue": "generate_sub_chapters", - "end": END - } - ) - - workflow.add_conditional_edges( - "generate_sub_chapters", - should_continue_processing, - { - "continue": "ai_review_structure", - "end": END - } - ) - - workflow.add_conditional_edges( - "ai_review_structure", + "generate_toc", should_continue_processing, { "continue": "finalize_structure", @@ -975,180 +534,3 @@ class AnalysisAgent: return asyncio.run(self.execute(source_file)) -# ========== 子标题生成辅助函数 ========== - -def _generate_ai_sub_chapters(criteria_list: List[ScoringCriteria], parent_chapter: DocumentChapter) -> List[DocumentChapter]: - """为大类别章节下的多个评分项生成AI子标题""" - try: - from ..tools.parser import BidParser - parser = BidParser() - - # 构建评分项信息 - criteria_info = [] - for criteria in criteria_list: - criteria_info.append(f"- {criteria.item_name} ({criteria.max_score}分): {criteria.description[:50]}...") - - prompt = f""" -根据以下大类别下的多个技术评分项，生成专业的标书章节子标题结构。 - -【大类别】: {parent_chapter.title} -【包含评分项】: -{chr(10).join(criteria_info)} - -【生成要求】: -1. 为每个评分项生成对应的二级标题（带分值） -2. 为重要评分项生成三级子标题，展开具体内容 -3. 结构要符合实际投标文档规范 -4. 二级标题格式: "X.1 评分项名称 (分值)" - -请返回JSON格式： -{{ - "sub_chapters": [ - {{"title": "2.1 供应商名称", "level": 2, "score": 0, "children": [ - {{"title": "2.1.1 企业基本信息", "level": 3}}, - {{"title": "2.1.2 资质证明材料", "level": 3}} - ]}}, - {{"title": "2.2 技术实力 (3分)", "level": 2, "score": 3, "children": []}} - ] -}} - -只返回JSON，无其他文字：""" - - response = parser._call_llm_api(prompt) - if not response: - logger.error("AI API调用失败，返回空响应") - return [] - - logger.info("AI API调用成功，开始解析响应") - - import json - try: - clean_response = response.strip() - if clean_response.startswith("```json"): - clean_response = clean_response[7:] - if clean_response.endswith("```"): - clean_response = clean_response[:-3] - clean_response = clean_response.strip() - - result_data = json.loads(clean_response) - sub_chapters_data = result_data.get("sub_chapters", []) - - logger.info(f"JSON解析成功，获得 {len(sub_chapters_data)} 个子章节数据") - - sub_chapters = [] - # 从父章节标题中提取章节号 - parent_number = parent_chapter.title.split(".")[0] if "." in parent_chapter.title else "1" - logger.info(f"父章节标题: '{parent_chapter.title}', 提取的编号: '{parent_number}'") - - for i, sub_data in enumerate(sub_chapters_data, 1): - # 生成正确的子标题编号 - original_title = sub_data.get("title", f"子标题{i}") - # 移除AI可能生成的错误编号，保留内容 - clean_title = original_title - if "." in original_title and original_title[0].isdigit(): - # 如果标题以数字开头，去掉原有编号 - parts = original_title.split(" ", 1) - if len(parts) > 1 and "." in parts[0]: - clean_title = parts[1] - - correct_title = f"{parent_number}.{i} {clean_title}" - logger.info(f"生成子标题: '{correct_title}'") - - sub_chapter = DocumentChapter( - id=f"{parent_chapter.id}_sub_{i:02d}", - title=correct_title, - level=sub_data.get("level", 2), - score=sub_data.get("score", 0), - template_placeholder=f"{{{{{parent_chapter.id}_sub_{i:02d}_content}}}}" - ) - - # 处理三级标题 - children_data = sub_data.get("children", []) - for j, child_data in enumerate(children_data, 1): - original_child_title = child_data.get("title", f"三级标题{j}") - # 清理三级标题编号 - clean_child_title = original_child_title - if "." in original_child_title and original_child_title[0].isdigit(): - parts = original_child_title.split(" ", 1) - if len(parts) > 1 and "." in parts[0]: - clean_child_title = parts[1] - - correct_child_title = f"{parent_number}.{i}.{j} {clean_child_title}" - - child_chapter = DocumentChapter( - id=f"{parent_chapter.id}_sub_{i:02d}_{j:02d}", - title=correct_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 (json.JSONDecodeError, KeyError) as e: - logger.error(f"解析AI子标题生成响应失败: {e}") - return [] - - except Exception as e: - logger.error(f"AI生成子标题失败: {e}") - return [] - - -def _generate_template_sub_chapters(criteria: ScoringCriteria, parent_chapter: DocumentChapter, template_file: str) -> List[DocumentChapter]: - """基于模板生成子标题""" - try: - from docx import Document - doc = Document(template_file) - - sub_chapters = [] - chapter_index = 1 - - # 从模板中提取标题结构作为子标题 - for paragraph in doc.paragraphs: - if paragraph.style.name.startswith('Heading'): - level = int(paragraph.style.name.split()[-1]) if paragraph.style.name.split()[-1].isdigit() else 2 - - # 限制为2-3级标题 - if level in [2, 3]: - adjusted_level = level # 保持原有层级 - - sub_chapter = DocumentChapter( - id=f"{parent_chapter.id}_tpl_{chapter_index:02d}", - title=f"{parent_chapter.title.split('.')[0]}.{chapter_index} {paragraph.text.strip()}", - level=adjusted_level, - template_placeholder=f"{{{{{parent_chapter.id}_tpl_{chapter_index:02d}_content}}}}" - ) - sub_chapters.append(sub_chapter) - chapter_index += 1 - - # 如果模板没有合适的标题，提供默认结构 - if not sub_chapters: - default_sub_chapters = [ - DocumentChapter( - id=f"{parent_chapter.id}_def_01", - title=f"{parent_chapter.title.split('.')[0]}.1 方案概述", - level=2, - template_placeholder=f"{{{{{parent_chapter.id}_def_01_content}}}}" - ), - DocumentChapter( - id=f"{parent_chapter.id}_def_02", - title=f"{parent_chapter.title.split('.')[0]}.2 具体实施", - level=2, - template_placeholder=f"{{{{{parent_chapter.id}_def_02_content}}}}" - ), - DocumentChapter( - id=f"{parent_chapter.id}_def_03", - title=f"{parent_chapter.title.split('.')[0]}.3 保障措施", - level=2, - template_placeholder=f"{{{{{parent_chapter.id}_def_03_content}}}}" - ) - ] - return default_sub_chapters - - return sub_chapters[:5] # 限制最多5个子标题 - - except Exception as e: - logger.error(f"基于模板生成子标题失败: {e}") - return [] \ No newline at end of file diff --git a/src/bidmaster/agents/toc_generator.py b/src/bidmaster/agents/toc_generator.py new file mode 100644 index 0000000..f28880e --- /dev/null +++ b/src/bidmaster/agents/toc_generator.py @@ -0,0 +1,615 @@ +"""目录生成Agent - 专门负责生成标书目录结构 + +基于LangGraph实现的目录生成Agent，负责： +1. 根据评分项类别生成一级章节 +2. AI智能生成二三级子标题 +3. 目录结构合理性审查与优化 +""" + +import asyncio +import json +import logging +from typing import List, Dict, Any, TypedDict, Optional + +from langgraph.graph import StateGraph, END +from pydantic import BaseModel, Field + +from ..tools.parser import ScoringCriteria, DocumentChapter +from ..tools.llm import llm_service +from ..config import get_settings + +logger = logging.getLogger(__name__) + +# 统一的类别名称映射 +CATEGORY_NAMES = { + "compliance": "合规响应", + "technical_solution": "技术方案", + "equipment_spec": "设备规格", + "quality_safety": "质量安全", + "after_sales": "售后服务", + "implementation": "实施方案" +} + + +class TocGeneratorState(TypedDict): + """目录生成Agent的状态定义""" + + # 输入参数 + technical_criteria: List[ScoringCriteria] + generation_mode: str # "ai" 或 "template" + template_file: Optional[str] + + # 执行状态 + current_step: str + should_continue: bool + + # 中间数据 + category_groups: Dict[str, List[ScoringCriteria]] # 按类别分组的评分项 + preliminary_chapters: List[DocumentChapter] # 初步生成的章节 + structure_review: Dict[str, Any] # AI审查结果 + + # 最终输出 + final_chapters: List[DocumentChapter] + + # 错误处理 + error: str + warnings: List[str] + + +class TocGeneratorResult(BaseModel): + """目录生成结果""" + + success: bool = Field(description="是否执行成功") + chapters: List[DocumentChapter] = Field(default_factory=list, description="生成的章节结构") + error_message: Optional[str] = Field(default=None, description="错误信息") + warnings: List[str] = Field(default_factory=list, description="警告信息") + + +# ========== LangGraph节点函数 ========== + +def group_criteria_node(state: TocGeneratorState) -> TocGeneratorState: + """节点1：按类别对评分项分组""" + logger.info("开始对评分项按类别分组...") + + try: + technical_criteria = state["technical_criteria"] + + if not technical_criteria: + raise ValueError("缺少技术评分项") + + # 按类别分组 + category_groups = { + "technical_solution": [], + "equipment_spec": [], + "implementation": [], + "quality_safety": [], + "after_sales": [], + "compliance": [] + } + + for criteria in technical_criteria: + category_key = criteria.category.value + if category_key in category_groups: + category_groups[category_key].append(criteria) + else: + category_groups["technical_solution"].append(criteria) + + # 只保留有评分项的类别 + category_groups = {k: v for k, v in category_groups.items() if v} + + logger.info(f"分组完成: {len(category_groups)}个类别") + state["category_groups"] = category_groups + state["current_step"] = "group_criteria" + + except Exception as e: + logger.error(f"评分项分组失败: {e}") + state["error"] = str(e) + state["should_continue"] = False + + return state + + +def generate_first_level_node(state: TocGeneratorState) -> TocGeneratorState: + """节点2：生成一级章节""" + logger.info("开始生成一级章节...") + + try: + category_groups = state["category_groups"] + technical_criteria = state["technical_criteria"] + + chapters = [] + chapter_index = 1 + + # 按原始顺序确定章节顺序 + def get_category_first_index(category): + 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) + + # 为每个类别创建一级章节 + for category in sorted_categories: + criteria_list = category_groups[category] + if not criteria_list: + continue + + total_score = sum(c.max_score for c in criteria_list) + chapter_id = f"chapter_{chapter_index:02d}_{category}" + category_name = CATEGORY_NAMES.get(category, category) + chapter_title = f"{chapter_index}. {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}}}}" + ) + + chapters.append(chapter) + chapter_index += 1 + + logger.info(f"生成{len(chapters)}个一级章节") + state["preliminary_chapters"] = chapters + state["current_step"] = "generate_first_level" + + except Exception as e: + logger.error(f"一级章节生成失败: {e}") + state["error"] = str(e) + state["should_continue"] = False + + return state + + +def generate_sub_chapters_node(state: TocGeneratorState) -> TocGeneratorState: + """节点3：生成二三级子标题""" + logger.info("开始生成子标题...") + + try: + preliminary_chapters = state["preliminary_chapters"] + technical_criteria = state["technical_criteria"] + generation_mode = state.get("generation_mode", "ai") + template_file = state.get("template_file") + + enhanced_chapters = [] + + for chapter in preliminary_chapters: + # 找到该章节对应的评分项 + # 从章节ID提取类别 + if "_" in chapter.id: + parts = chapter.id.split("_") + if len(parts) >= 3: + category = "_".join(parts[2:]) + + # 找到该类别的所有评分项 + corresponding_criteria = [ + c for c in technical_criteria + if c.category.value == category + ] + + if corresponding_criteria: + if generation_mode == "ai": + sub_chapters = _generate_ai_sub_chapters(corresponding_criteria, chapter) + else: + sub_chapters = _generate_template_sub_chapters( + corresponding_criteria[0], chapter, template_file + ) + + chapter.children = sub_chapters + logger.info(f"章节 {chapter.title} 生成了 {len(sub_chapters)} 个子标题") + + enhanced_chapters.append(chapter) + + state["preliminary_chapters"] = enhanced_chapters + state["current_step"] = "generate_sub_chapters" + + except Exception as e: + logger.error(f"子标题生成失败: {e}") + state["error"] = str(e) + state["should_continue"] = False + + return state + + +def review_structure_node(state: TocGeneratorState) -> TocGeneratorState: + """节点4：AI审查目录结构""" + logger.info("开始AI审查目录结构...") + + try: + technical_criteria = state["technical_criteria"] + preliminary_chapters = state["preliminary_chapters"] + + # 构建审查提示词 + criteria_summary = _format_criteria_for_review(technical_criteria) + chapters_summary = _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：""" + + # 使用统一的LLM服务 + response = llm_service.call(review_prompt) + + if response: + try: + clean_response = response.strip() + if clean_response.startswith("```json"): + clean_response = clean_response[7:] + if clean_response.endswith("```"): + clean_response = clean_response[:-3] + review_result = json.loads(clean_response.strip()) + except json.JSONDecodeError: + review_result = {"overall_assessment": "解析失败", "suggestions": []} + else: + review_result = {"overall_assessment": "AI审查失败", "suggestions": []} + + state["structure_review"] = review_result + state["current_step"] = "review_structure" + + # 根据审查结果添加警告 + if review_result.get("suggestions"): + state["warnings"].append(f"AI审查: {len(review_result['suggestions'])}条优化建议") + + except Exception as e: + logger.error(f"AI审查失败: {e}") + state["warnings"].append(f"AI审查跳过: {str(e)}") + state["structure_review"] = {} + + return state + + +def finalize_chapters_node(state: TocGeneratorState) -> TocGeneratorState: + """节点5：最终确定章节结构""" + logger.info("最终确定章节结构...") + + try: + preliminary_chapters = state["preliminary_chapters"] + structure_review = state.get("structure_review", {}) + + final_chapters = preliminary_chapters # 直接使用，不做防护编程 + + # 应用高优先级建议 + suggestions = structure_review.get("suggestions", []) + for suggestion in suggestions: + if suggestion.get("priority") == "high": + # 这里可以根据建议类型进行调整 + logger.info(f"应用高优先级建议: {suggestion.get('description')}") + + # 确保评标索引表存在（作为第一章） + has_index = any("评标索引" in ch.title for ch in final_chapters) + if not has_index: + index_chapter = DocumentChapter( + id="evaluation_index", + title="1. 评标索引表（技术评分完全对应）", + level=1, + template_placeholder="{{evaluation_index_content}}" + ) + final_chapters.insert(0, index_chapter) + + # 重新编号 + for i, chapter in enumerate(final_chapters[1:], 2): + # 安全处理章节编号 + title_parts = chapter.title.split(".") + if len(title_parts) >= 2: + old_number = title_parts[0] + chapter.title = chapter.title.replace(f"{old_number}.", f"{i}.", 1) + + state["final_chapters"] = final_chapters + state["current_step"] = "finalize" + state["should_continue"] = False + + logger.info(f"最终生成{len(final_chapters)}个章节") + + except Exception as e: + logger.error(f"章节最终确定失败: {e}") + state["error"] = str(e) + state["should_continue"] = False + + return state + + +# ========== 辅助函数 ========== + +def _generate_ai_sub_chapters(criteria_list: List[ScoringCriteria], + parent_chapter: DocumentChapter) -> List[DocumentChapter]: + """AI生成子标题""" + 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. 格式规范，符合标书要求 + +返回JSON格式： +{{ + "sub_chapters": [ + {{"title": "技术架构设计", "level": 2, "score": 5, "children": []}} + ] +}} + +只返回JSON：""" + + # 使用统一的LLM服务 + response = llm_service.call(prompt) + if not response: + raise ValueError("AI生成子标题失败: API无响应") + + try: + clean_response = response.strip() + if clean_response.startswith("```json"): + clean_response = clean_response[7:] + if clean_response.endswith("```"): + clean_response = clean_response[:-3] + + result_data = json.loads(clean_response.strip()) + sub_chapters_data = result_data.get("sub_chapters", []) + + sub_chapters = [] + # 安全处理章节号提取 + parent_parts = parent_chapter.title.split(".") + parent_number = parent_parts[0] if parent_parts else "1" + + for i, sub_data in enumerate(sub_chapters_data, 1): + # 清理标题，移除可能的错误编号 + title = sub_data.get("title", f"子标题{i}") + if "." in title and title[0].isdigit(): + parts = title.split(" ", 1) + if len(parts) > 1: + title = parts[1] + + correct_title = f"{parent_number}.{i} {title}" + + sub_chapter = DocumentChapter( + id=f"{parent_chapter.id}_sub_{i:02d}", + title=correct_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}") + if "." in child_title and child_title[0].isdigit(): + parts = child_title.split(" ", 1) + if len(parts) > 1: + child_title = parts[1] + + correct_child_title = f"{parent_number}.{i}.{j} {child_title}" + + child_chapter = DocumentChapter( + id=f"{parent_chapter.id}_sub_{i:02d}_{j:02d}", + title=correct_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 (json.JSONDecodeError, KeyError) as e: + logger.error(f"解析AI响应失败: {e}") + raise ValueError(f"解析AI响应失败: {e}") + + except Exception as e: + logger.error(f"AI生成子标题失败: {e}") + raise # 立即失败，不掩盖错误 + + +def _generate_template_sub_chapters(criteria: ScoringCriteria, + parent_chapter: DocumentChapter, + template_file: Optional[str]) -> List[DocumentChapter]: + """基于模板生成子标题""" + # 提供默认结构 + # 从父章节标题中提取章节号，处理 "2. 技术方案 (10分)" 这种格式 + parent_title = parent_chapter.title + if "." in parent_title: + parent_number = parent_title.split(".")[0].strip() + else: + parent_number = "1" + + default_sub_chapters = [ + DocumentChapter( + id=f"{parent_chapter.id}_def_01", + title=f"{parent_number}.1 方案概述", + level=2, + template_placeholder=f"{{{{{parent_chapter.id}_def_01_content}}}}" + ), + DocumentChapter( + id=f"{parent_chapter.id}_def_02", + title=f"{parent_number}.2 具体实施", + level=2, + template_placeholder=f"{{{{{parent_chapter.id}_def_02_content}}}}" + ), + DocumentChapter( + id=f"{parent_chapter.id}_def_03", + title=f"{parent_number}.3 保障措施", + level=2, + template_placeholder=f"{{{{{parent_chapter.id}_def_03_content}}}}" + ) + ] + + return default_sub_chapters + + +def _format_criteria_for_review(technical_criteria: List[ScoringCriteria]) -> str: + """格式化评分项用于审查""" + 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(chapters: List[DocumentChapter]) -> str: + """格式化章节用于审查""" + 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) + + +# ========== 条件判断 ========== + +def should_continue(state: TocGeneratorState) -> str: + """判断是否继续""" + if not state.get("should_continue", True) or state.get("error"): + return "end" + return "continue" + + +class TocGeneratorAgent: + """目录生成Agent""" + + def __init__(self): + self.settings = get_settings() + self.graph = self._build_graph() + + def _build_graph(self) -> StateGraph: + """构建工作流""" + workflow = StateGraph(TocGeneratorState) + + # 添加节点 + workflow.add_node("group_criteria", group_criteria_node) + workflow.add_node("generate_first_level", generate_first_level_node) + workflow.add_node("generate_sub_chapters", generate_sub_chapters_node) + workflow.add_node("review_structure", review_structure_node) + workflow.add_node("finalize_chapters", finalize_chapters_node) + + # 设置入口 + workflow.set_entry_point("group_criteria") + + # 添加边 + workflow.add_conditional_edges( + "group_criteria", + should_continue, + {"continue": "generate_first_level", "end": END} + ) + + workflow.add_conditional_edges( + "generate_first_level", + should_continue, + {"continue": "generate_sub_chapters", "end": END} + ) + + workflow.add_conditional_edges( + "generate_sub_chapters", + should_continue, + {"continue": "review_structure", "end": END} + ) + + workflow.add_conditional_edges( + "review_structure", + should_continue, + {"continue": "finalize_chapters", "end": END} + ) + + workflow.add_edge("finalize_chapters", END) + + return workflow.compile() + + async def generate(self, + technical_criteria: List[ScoringCriteria], + generation_mode: str = "ai", + template_file: Optional[str] = None) -> TocGeneratorResult: + """生成目录结构""" + logger.info("开始执行TocGeneratorAgent") + + # 初始化状态 + initial_state = TocGeneratorState( + technical_criteria=technical_criteria, + generation_mode=generation_mode, + template_file=template_file, + current_step="", + should_continue=True, + category_groups={}, + preliminary_chapters=[], + structure_review={}, + final_chapters=[], + error="", + warnings=[] + ) + + try: + # 执行工作流 + final_state = await self.graph.ainvoke(initial_state) + + if final_state.get("error"): + return TocGeneratorResult( + success=False, + error_message=final_state["error"], + warnings=final_state.get("warnings", []) + ) + + return TocGeneratorResult( + success=True, + chapters=final_state["final_chapters"], + warnings=final_state.get("warnings", []) + ) + + except Exception as e: + logger.error(f"TocGeneratorAgent执行异常: {e}") + return TocGeneratorResult( + success=False, + error_message=str(e) + ) + + def generate_sync(self, + technical_criteria: List[ScoringCriteria], + generation_mode: str = "ai", + template_file: Optional[str] = None) -> TocGeneratorResult: + """同步接口""" + return asyncio.run(self.generate(technical_criteria, generation_mode, template_file)) \ No newline at end of file diff --git a/src/bidmaster/tools/__init__.py b/src/bidmaster/tools/__init__.py index 2aef06a..64c8e3f 100644 --- a/src/bidmaster/tools/__init__.py +++ b/src/bidmaster/tools/__init__.py @@ -1 +1,15 @@ -# 工具层 - 原子化工具集 \ No newline at end of file +"""工具层 - 原子化工具集""" + +from .llm import llm_service +from .parser import BidParser +from .rag import RAGTool +from .table import TableGenerator +from .word import WordProcessor + +__all__ = [ + "llm_service", + "BidParser", + "RAGTool", + "TableGenerator", + "WordProcessor", +] \ No newline at end of file diff --git a/src/bidmaster/tools/llm.py b/src/bidmaster/tools/llm.py new file mode 100644 index 0000000..675b966 --- /dev/null +++ b/src/bidmaster/tools/llm.py @@ -0,0 +1,69 @@ +"""LLM服务工具 - 统一的LLM API调用接口 + +提供统一的LLM调用服务，避免重复创建客户端实例。 +""" + +import logging +from typing import Optional + +from openai import OpenAI + +from ..config import get_settings + +logger = logging.getLogger(__name__) + + +class LLMService: + """LLM服务单例""" + + _instance: Optional['LLMService'] = None + _client: Optional[OpenAI] = None + + def __new__(cls): + if cls._instance is None: + cls._instance = super().__new__(cls) + return cls._instance + + def __init__(self): + if self._client is None: + settings = get_settings() + self._client = OpenAI( + api_key=settings.api_key, + base_url=settings.base_url, + ) + + def call(self, prompt: str, temperature: float = 0.7) -> Optional[str]: + """调用LLM API + + Args: + prompt: 提示词 + temperature: 温度参数 + + Returns: + LLM响应文本，失败返回None + """ + try: + settings = get_settings() + + response = self._client.chat.completions.create( + model=settings.model_name, + messages=[ + {"role": "system", "content": "你是一个专业的招投标文档分析助手。"}, + {"role": "user", "content": prompt} + ], + temperature=temperature, + max_tokens=4000, + ) + + if response and response.choices: + return response.choices[0].message.content + + return None + + except Exception as e: + logger.error(f"LLM API调用失败: {e}") + raise # 立即失败 + + +# 全局单例 +llm_service = LLMService() \ No newline at end of file diff --git a/src/bidmaster/tools/parser.py b/src/bidmaster/tools/parser.py index 205e675..b69c0c3 100644 --- a/src/bidmaster/tools/parser.py +++ b/src/bidmaster/tools/parser.py @@ -442,6 +442,10 @@ class BidParser: logger.error(f"AI解析表格失败: {e}") return [] + def call_llm(self, prompt: str) -> str | None: + """公共方法：调用LLM API""" + return self._call_llm_api(prompt) + def _call_llm_api(self, prompt: str) -> str | None: """调用LLM API""" try: