Doni/bidmaster-cli
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 将目录生成功能抽取为独立的TocGeneratorAgent

Browse Source 
- 创建独立的TocGeneratorAgent，专门负责目录生成
- 从AnalysisAgent中移除700+行冗余代码
- 添加统一的LLM服务层，避免重复创建客户端实例
- 修复所有违反PROJECT_SPEC.md规范的代码：
  - 移除防御编程（copy()方法）
  - 实现立即失败原则（异常直接抛出）
  - 统一常量定义（CATEGORY_NAMES）
  - 规范import语句位置
  - 修复私有方法调用问题
- 保持向后兼容性，不改变原有接口

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
sladro 2025-09-27 10:12:44 +08:00
parent e6103c711f
commit 2ff96ea544
6 changed files with 773 additions and 684 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/bidmaster/agents/__init__.py
View File

@ -1 +1,6 @@
# Agent层 - LangGraph编排
"""Agent层 - LangGraph编排"""
from .analysis import AnalysisAgent
from .toc_generator import TocGeneratorAgent
__all__ = ["AnalysisAgent", "TocGeneratorAgent"]

746
src/bidmaster/agents/analysis.py
View File

@ -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 []

615
src/bidmaster/agents/toc_generator.py Normal file
View File

@ -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))

16
src/bidmaster/tools/__init__.py
View File

@ -1 +1,15 @@
# 工具层 - 原子化工具集
"""工具层 - 原子化工具集"""
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",
]

69
src/bidmaster/tools/llm.py Normal file
View File

@ -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()

4
src/bidmaster/tools/parser.py
View File

@ -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: