diff --git a/config/prompts.yaml b/config/prompts.yaml index 627b6c4..d2d2e7f 100644 --- a/config/prompts.yaml +++ b/config/prompts.yaml @@ -115,12 +115,15 @@ toc_prompts: {criteria_info} 【招标上下文摘录】: {context_snippets} + 【全局技术一致性提示】: + {tech_consistency_prompt} 生成要求: 1. 为每个评分项生成对应的子标题名称(不要包含编号) 2. 重要评分项可添加三级子标题(不要包含编号) 3. 充分参考上下文摘录中的专业术语和业务背景,体现定制化判断 - 4. 只返回标题文本,编号由Word自动管理 + 4. 必须遵循“全局技术一致性提示”中的架构/技术选型/术语口径，确保全篇一致 + 5. 只返回标题文本,编号由Word自动管理 返回JSON格式: {{ diff --git a/src/bidmaster/agents/builders/toc_builder.py b/src/bidmaster/agents/builders/toc_builder.py index a7df8d7..79570e8 100644 --- a/src/bidmaster/agents/builders/toc_builder.py +++ b/src/bidmaster/agents/builders/toc_builder.py @@ -9,6 +9,7 @@ from typing import Dict, Any, Optional from ..base import AgentBuilder, BaseAgent, BaseAgentFactory from ...nodes.toc import ( + PrepareTechConsistencyPromptNode, GroupCriteriaNode, GenerateFirstLevelNode, GenerateSubChaptersNode, @@ -44,7 +45,8 @@ class TocAgentBuilder(AgentBuilder): builder = cls(interaction_handler) # 添加所有节点 - builder.add_node(GroupCriteriaNode()) \ + builder.add_node(PrepareTechConsistencyPromptNode()) \ + .add_node(GroupCriteriaNode()) \ .add_node(GenerateFirstLevelNode()) \ .add_node(GenerateSubChaptersNode()) \ .add_node(ReviewStructureNode()) \ @@ -55,7 +57,7 @@ class TocAgentBuilder(AgentBuilder): .add_node(OptimizeWithFeedbackNode()) # 设置入口点 - builder.set_entry("group_criteria") + builder.set_entry("prepare_tech_consistency_prompt") # 配置工作流程 builder._configure_workflow() @@ -65,6 +67,12 @@ class TocAgentBuilder(AgentBuilder): def _configure_workflow(self) -> None: """配置工作流程""" + self.add_conditional_edge( + "prepare_tech_consistency_prompt", + should_continue, + {"continue": "group_criteria", "end": "END"} + ) + # 添加条件边 - 每个节点都检查是否应该继续 self.add_conditional_edge( "group_criteria", diff --git a/src/bidmaster/nodes/toc/__init__.py b/src/bidmaster/nodes/toc/__init__.py index 8453699..f693a59 100644 --- a/src/bidmaster/nodes/toc/__init__.py +++ b/src/bidmaster/nodes/toc/__init__.py @@ -13,6 +13,7 @@ from .adjust_chapters import AdjustChaptersNode from .finalize_chapters import FinalizeChaptersNode from .user_feedback import UserFeedbackNode from .optimize_with_feedback import OptimizeWithFeedbackNode +from .prepare_tech_consistency_prompt import PrepareTechConsistencyPromptNode # 辅助组件 from .factories import ChapterFactory @@ -36,6 +37,7 @@ __all__ = [ "FinalizeChaptersNode", "UserFeedbackNode", "OptimizeWithFeedbackNode", + "PrepareTechConsistencyPromptNode", # 辅助组件 "ChapterFactory", diff --git a/src/bidmaster/nodes/toc/generate_sub_chapters.py b/src/bidmaster/nodes/toc/generate_sub_chapters.py index 027a98c..23e4a61 100644 --- a/src/bidmaster/nodes/toc/generate_sub_chapters.py +++ b/src/bidmaster/nodes/toc/generate_sub_chapters.py @@ -52,6 +52,7 @@ class GenerateSubChaptersNode(BaseNode, TocNodeBase): self.log_step_info("context_searcher", f"上下文检索器初始化失败: {exc}") guidance_map = dict(state.get("chapter_guidance_map", {})) + tech_consistency_prompt = (state.get("tech_consistency_prompt") or "").strip() or None # 为每个章节生成子标题 enhanced_chapters = [] @@ -59,7 +60,8 @@ class GenerateSubChaptersNode(BaseNode, TocNodeBase): enhanced_chapter = self._enhance_chapter_with_subs( chapter, technical_criteria, - context_searcher + context_searcher, + tech_consistency_prompt, ) enhanced_chapters.append(enhanced_chapter) @@ -75,7 +77,8 @@ class GenerateSubChaptersNode(BaseNode, TocNodeBase): def _enhance_chapter_with_subs(self, chapter: DocumentChapter, technical_criteria: List[ScoringCriteria], - context_searcher: Optional[DocumentContextSearcher] = None) -> DocumentChapter: + context_searcher: Optional[DocumentContextSearcher] = None, + tech_consistency_prompt: Optional[str] = None) -> DocumentChapter: """为章节增强子标题 Args: @@ -99,7 +102,8 @@ class GenerateSubChaptersNode(BaseNode, TocNodeBase): sub_chapters_data = LLMHelper.generate_sub_chapters_ai( corresponding_criteria, chapter, - context_snippets=context_snippets + context_snippets=context_snippets, + tech_consistency_prompt=tech_consistency_prompt, ) if sub_chapters_data: diff --git a/src/bidmaster/nodes/toc/llm_helper.py b/src/bidmaster/nodes/toc/llm_helper.py index c3be050..5447941 100644 --- a/src/bidmaster/nodes/toc/llm_helper.py +++ b/src/bidmaster/nodes/toc/llm_helper.py @@ -79,15 +79,19 @@ class LLMHelper: return None @staticmethod - def generate_sub_chapters_ai(criteria_list: List[ScoringCriteria], - parent_chapter: DocumentChapter, - context_snippets: Optional[List[str]] = None) -> Optional[List[Dict[str, Any]]]: + def generate_sub_chapters_ai( + criteria_list: List[ScoringCriteria], + parent_chapter: DocumentChapter, + context_snippets: Optional[List[str]] = None, + tech_consistency_prompt: Optional[str] = None, + ) -> Optional[List[Dict[str, Any]]]: """AI生成子章节数据 Args: criteria_list: 对应的评分项列表 parent_chapter: 父章节 context_snippets: 招标文档的相关片段 + tech_consistency_prompt: 全局技术一致性提示词（用于统一术语/架构口径） Returns: 子章节数据列表，失败时返回None @@ -100,12 +104,14 @@ class LLMHelper: # 从配置获取提示词 prompt_manager = get_prompt_manager() context_text = "\n".join(context_snippets) if context_snippets else "（暂无上下文摘录）" + consistency_text = (tech_consistency_prompt or "").strip() or "（暂无全局技术一致性提示）" prompt = prompt_manager.get_toc_prompt( "generate_sub_chapters", parent_title=parent_chapter.title, criteria_info=chr(10).join(criteria_info), - context_snippets=context_text + context_snippets=context_text, + tech_consistency_prompt=consistency_text, ) try: diff --git a/src/bidmaster/nodes/toc/prepare_tech_consistency_prompt.py b/src/bidmaster/nodes/toc/prepare_tech_consistency_prompt.py new file mode 100644 index 0000000..4db0176 --- /dev/null +++ b/src/bidmaster/nodes/toc/prepare_tech_consistency_prompt.py @@ -0,0 +1,241 @@ +"""全局技术一致性提示词准备节点 + +在目录生成开始时： +1) 从评分项/招标材料中抽取“技术要求” +2) 从知识库RAG中检索“技术解决方案” +3) 若两者齐全：拼装成提示词提供给子标题撰写AI使用 +4) 否则：生成一段几百字内的“系统架构+技术选型+整体建设要求”作为统一参考 +""" + +from __future__ import annotations + +import logging +import re +import textwrap +from typing import Any, Dict, Iterable, List + +from ..base import BaseNode, NodeContext +from ...tools.parser import ScoringCriteria +from .base_mixins import TocNodeBase + +logger = logging.getLogger(__name__) + + +_REQ_KEYWORDS = ( + "须", + "应", + "必须", + "需要", + "要求", + "提供", + "包括", + "满足", + "支持", + "不少于", + "不得", +) + + +class PrepareTechConsistencyPromptNode(BaseNode, TocNodeBase): + """在目录生成开始前准备全局技术一致性提示词""" + + @property + def name(self) -> str: + return "prepare_tech_consistency_prompt" + + @property + def description(self) -> str: + return "生成全局技术一致性提示词" + + def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: + return self.safe_execute(self._do_prepare, state, "生成全局技术一致性提示词") + + def _do_prepare(self, state: Dict[str, Any]) -> Dict[str, Any]: + criteria: List[ScoringCriteria] = list(state.get("technical_criteria") or []) + + tender_requirements = self._extract_tender_requirements(criteria) + rag_solutions = self._search_rag_solutions(state, criteria, tender_requirements) + + if tender_requirements and rag_solutions: + prompt = self._build_prompt_from_sources(tender_requirements, rag_solutions) + source = "tender+rag" + else: + prompt = self._build_fallback_prompt(tender_requirements, criteria) + source = "fallback" + + logger.info( + "技术一致性提示词生成完成: source=%s, req=%s条, rag=%s条", + source, + len(tender_requirements), + len(rag_solutions), + ) + + return self._update_state( + state, + tech_consistency_prompt=prompt, + tech_consistency_source=source, + ) + + def _extract_tender_requirements(self, criteria: Iterable[ScoringCriteria]) -> List[str]: + seen: set[str] = set() + collected: List[str] = [] + + def _add(text: str) -> None: + clean = self._normalize(text) + if not clean or clean in seen: + return + seen.add(clean) + collected.append(text.strip()) + + for item in criteria: + desc = (getattr(item, "description", "") or "").strip() + name = (getattr(item, "item_name", "") or "").strip() + if not desc and not name: + continue + + if desc: + for seg in self._split_sentences(desc): + if any(key in seg for key in _REQ_KEYWORDS): + _add(seg) + elif name: + _add(name) + + return collected[:10] + + def _search_rag_solutions( + self, + state: Dict[str, Any], + criteria: List[ScoringCriteria], + tender_requirements: List[str], + ) -> List[str]: + rag_tool = state.get("rag_tool") + if rag_tool is None: + try: + from ...tools.rag import RAGTool + + rag_tool = RAGTool() + except Exception as exc: + logger.warning("RAGTool初始化失败，将走fallback: %s", exc) + return [] + + query = self._build_solution_query(criteria, tender_requirements) + try: + results = rag_tool.search(query, k=5) + except Exception as exc: + logger.warning("RAG检索失败，将走fallback: %s", exc) + return [] + + snippets: List[str] = [] + for item in results or []: + content = (item or {}).get("content") or "" + score = float((item or {}).get("score") or 0.0) + if not content.strip(): + continue + if score and score < 0.2: + continue + snippets.append(self._truncate(content, 240)) + + return snippets[:4] + + def _build_solution_query( + self, + criteria: List[ScoringCriteria], + tender_requirements: List[str], + ) -> str: + seeds: List[str] = [ + "技术解决方案", + "技术路线", + "总体架构", + "系统架构", + "技术选型", + "性能", + "安全", + "运维", + ] + + for item in criteria[:8]: + name = (item.item_name or "").strip() + if name: + seeds.append(name) + + for req in tender_requirements[:3]: + seeds.append(req) + + return " ".join(self._dedup_keep_order(seeds)) + + def _build_prompt_from_sources(self, tender_requirements: List[str], rag_solutions: List[str]) -> str: + req_block = "\n".join(f"- {self._truncate(r, 120)}" for r in tender_requirements[:8]) + sol_block = "\n".join(f"- {self._truncate(s, 160)}" for s in rag_solutions[:4]) + + return ( + "请在后续子标题生成时统一技术路线与术语，避免同一概念多种叫法。\n" + "【招标技术要求摘录】\n" + f"{req_block}\n\n" + "【知识库技术解决方案摘录】\n" + f"{sol_block}\n\n" + "生成约束：子标题应优先使用上述摘录中的专业术语/模块命名方式，并保持跨章节一致。" + ).strip() + + def _build_fallback_prompt(self, tender_requirements: List[str], criteria: List[ScoringCriteria]) -> str: + hint_text = " ".join(tender_requirements) or " ".join( + (c.item_name + " " + (c.description or "")) for c in criteria[:8] + ) + hints = self._infer_hints(hint_text) + + architecture = "前端多端+后端服务层+数据层+集成与运维安全体系" + if "云边端协同" in hints: + architecture = "云边端协同（云平台+边缘节点+终端设备）+统一数据与运维体系" + + tech_stack = "前端Web/移动端；后端服务采用可扩展框架并容器化部署；数据层采用关系型数据库+缓存+对象存储；日志/监控/告警一体化。" + if "物联网" in hints: + tech_stack = "支持物联网接入（MQTT/HTTP等）与设备管理；后端服务容器化部署；数据层关系型数据库+时序/缓存；日志/监控/告警一体化。" + + requirements = "建设要求：接口标准化与可复用；安全合规与权限审计；性能指标可量化；高可用与可扩展；交付文档齐全、便于运维。" + + return ( + "未检索到可直接复用的技术解决方案摘录，以下为统一参考（几百字内）：\n" + f"系统架构：{architecture}。\n" + f"技术选型：{tech_stack}\n" + f"{requirements}\n" + "生成约束：后续子标题命名应围绕上述架构拆分模块，并保持全篇术语一致。" + ).strip() + + @staticmethod + def _split_sentences(text: str) -> List[str]: + parts = re.split(r"[\n；;。]+", text) + return [p.strip().strip("：:·•") for p in parts if p.strip()] + + @staticmethod + def _normalize(text: str) -> str: + return re.sub(r"\s+", "", (text or "")).strip().lower() + + @staticmethod + def _truncate(text: str, limit: int) -> str: + cleaned = " ".join((text or "").split()) + return textwrap.shorten(cleaned, width=limit, placeholder="...") + + @staticmethod + def _dedup_keep_order(items: Iterable[str]) -> List[str]: + seen: set[str] = set() + result: List[str] = [] + for item in items: + key = PrepareTechConsistencyPromptNode._normalize(item) + if not key or key in seen: + continue + seen.add(key) + result.append(item) + return result + + @staticmethod + def _infer_hints(text: str) -> List[str]: + blob = text or "" + hits: List[str] = [] + if "云边" in blob or "边缘" in blob: + hits.append("云边端协同") + if "物联网" in blob or "终端" in blob or "传感" in blob: + hits.append("物联网") + if "数据中台" in blob or "数据中心" in blob or "数据治理" in blob: + hits.append("数据中台") + if "安全" in blob or "等保" in blob: + hits.append("安全合规") + return hits diff --git a/tests/unit/test_tech_consistency_prompt.py b/tests/unit/test_tech_consistency_prompt.py new file mode 100644 index 0000000..e227e7c --- /dev/null +++ b/tests/unit/test_tech_consistency_prompt.py @@ -0,0 +1,66 @@ +from bidmaster.nodes.base import NodeContext +from bidmaster.nodes.toc.prepare_tech_consistency_prompt import ( + PrepareTechConsistencyPromptNode, +) +from bidmaster.tools.parser import ScoringCriteria, TechnicalCategory + + +class DummyRagTool: + def __init__(self, results): + self._results = results + self.last_query = None + + def search(self, query: str, k: int = 5): + self.last_query = query + return list(self._results) + + +def _criteria(description: str): + return [ + ScoringCriteria( + item_name="技术方案-基本要求", + max_score=3.0, + description=description, + category=TechnicalCategory.TECHNICAL_SOLUTION, + chapter_id="chapter_01_technical_solution", + original_index=0, + ) + ] + + +def test_prepare_prompt_uses_tender_and_rag_when_available(): + rag = DummyRagTool( + results=[ + {"content": "系统采用云边端协同架构，边缘侧负责就近处理与缓存。", "score": 0.9}, + {"content": "技术选型：容器化部署+统一监控告警+权限审计。", "score": 0.8}, + ] + ) + state = { + "technical_criteria": _criteria("供应商须提供总体架构设计方案，包含技术选型与安全策略。"), + "rag_tool": rag, + } + + node = PrepareTechConsistencyPromptNode() + result = node.execute(state, NodeContext()) + + assert result["tech_consistency_source"] == "tender+rag" + prompt = result["tech_consistency_prompt"] + assert "招标技术要求摘录" in prompt + assert "知识库技术解决方案摘录" in prompt + assert "云边端协同" in prompt + + +def test_prepare_prompt_falls_back_when_rag_missing(): + rag = DummyRagTool(results=[]) + state = { + "technical_criteria": _criteria("供应商须提供总体架构设计方案，包含技术选型与安全策略。"), + "rag_tool": rag, + } + + node = PrepareTechConsistencyPromptNode() + result = node.execute(state, NodeContext()) + + assert result["tech_consistency_source"] == "fallback" + prompt = result["tech_consistency_prompt"] + assert "系统架构" in prompt + assert "技术选型" in prompt diff --git a/tests/unit/test_toc_generation.py b/tests/unit/test_toc_generation.py index ba9af99..a97f7f8 100644 --- a/tests/unit/test_toc_generation.py +++ b/tests/unit/test_toc_generation.py @@ -9,8 +9,9 @@ from bidmaster.utils.document_context import DocumentContext, DocumentContextMat def test_generate_sub_chapters_uses_context(monkeypatch): captured = {} - def fake_generate(criteria_list, parent_chapter, context_snippets=None): + def fake_generate(criteria_list, parent_chapter, context_snippets=None, tech_consistency_prompt=None): captured["context"] = context_snippets + captured["consistency"] = tech_consistency_prompt return [{"title": "示例小节", "level": 2, "score": 5, "children": []}] class DummySearcher: @@ -52,6 +53,7 @@ def test_generate_sub_chapters_uses_context(monkeypatch): "preliminary_chapters": [chapter], "technical_criteria": criteria, "document_context": DocumentContext("demo.docx", "test-model", []), + "tech_consistency_prompt": "统一术语：云边端协同/边缘节点/终端设备", } node = GenerateSubChaptersNode() @@ -59,6 +61,7 @@ def test_generate_sub_chapters_uses_context(monkeypatch): assert result["preliminary_chapters"][0].children assert captured["context"] and "项目概述" in captured["context"][0] + assert "统一术语" in (captured["consistency"] or "") def test_user_feedback_auto_triggers_optimization():