增加了前后文一致性,提取技术解决方案,如果没有就自己编一个

This commit is contained in:
sladro 2025-12-22 13:25:20 +08:00
parent 3d12f9d065
commit 58d628347e
8 changed files with 344 additions and 11 deletions

View File

@ -115,12 +115,15 @@ toc_prompts:
{criteria_info}
【招标上下文摘录】:
{context_snippets}
【全局技术一致性提示】:
{tech_consistency_prompt}
生成要求:
1. 为每个评分项生成对应的子标题名称(不要包含编号)
2. 重要评分项可添加三级子标题(不要包含编号)
3. 充分参考上下文摘录中的专业术语和业务背景,体现定制化判断
4. 只返回标题文本,编号由Word自动管理
4. 必须遵循“全局技术一致性提示”中的架构/技术选型/术语口径,确保全篇一致
5. 只返回标题文本,编号由Word自动管理
返回JSON格式:
{{

View File

@ -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",

View File

@ -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",

View File

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

View File

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

View File

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

View File

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

View File

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