""" proposal_generator.py - 投标内容生成模块 根据招标分析结果（AnalysisResult），使用OpenAI生成对应的投标响应内容。 支持按类别采用不同生成策略，可并行调用API以提高性能。 """ import json import logging import re from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Optional from openai import OpenAI from .config import get_config from .models import AnalysisResult, ProposalDocument, ProposalSection, Requirement logger = logging.getLogger(__name__) # 类别对应的生成策略提示词 _CATEGORY_PROMPTS = { "evaluation": ( "你正在撰写一份投标文件的评分响应部分。请针对以下评审因素，提供具体、可量化的响应方案。\n" "内容要求：\n" "1. 明确响应评审指标，逐条对应\n" "2. 提供具体的技术方案、组织措施或管理方法\n" "3. 如有分值权重，重点突出高分项\n" "4. 强调我方优势和经验\n" "5. 语言正式、专业，符合标书风格" ), "service": ( "你正在撰写一份投标文件的服务方案部分。请针对以下服务要求，制定详细的服务方案。\n" "内容要求：\n" "1. 详细描述服务方案、实施流程和质量标准\n" "2. 说明服务团队配置和职责分工\n" "3. 列出服务承诺和保障措施\n" "4. 包含服务响应时限和SLA指标\n" "5. 可适当引用过往成功案例" ), "qualification": ( "你正在撰写一份投标文件的资质响应部分。请针对以下资质要求，说明我方满足条件的证明材料。\n" "内容要求：\n" "1. 明确说明我方具备的相应资质和证书\n" "2. 提供资质等级、编号、有效期等关键信息\n" "3. 说明资质对应的业务范围和能力\n" "4. 附上证明材料清单\n" "5. 如暂无某项资质，说明已申请或替代方案" ), "other": ( "你正在撰写一份投标文件的响应部分。请针对以下要求，提供具体、专业的响应内容。\n" "内容要求：\n" "1. 直接回应招标要求，不回避、不跑题\n" "2. 内容具体可行，避免空泛承诺\n" "3. 语言正式、专业、规范\n" "4. 适当使用数据和事实支撑" ), } _DEFAULT_CATEGORY_PROMPT = ( "你正在撰写一份投标文件的响应部分。请针对以下要求，提供具体、专业的响应内容。\n" "内容要求：\n" "1. 直接回应招标要求，不回避、不跑题\n" "2. 内容具体可行，避免空泛承诺\n" "3. 语言正式、专业、规范\n" "4. 适当使用数据和事实支撑" ) # JSON Schema for structured output _SECTION_SCHEMA = { "name": "proposal_section", "strict": True, "schema": { "type": "object", "properties": { "title": {"type": "string"}, "content": {"type": "string"}, "requirement_ref": {"type": "string"}, "level": {"type": "number"}, }, "required": ["title", "content", "requirement_ref", "level"], "additionalProperties": False, }, } _COVER_LETTER_SYSTEM_PROMPT = ( "你是一个专业的投标文件撰写专家。请根据项目名称，撰写一份正式的投标函。\n" "投标函应包含以下要素：\n" "1. 致送单位（使用'致：招标人/采购代理机构'）\n" "2. 投标意向声明\n" "3. 报价承诺（标明'按招标文件要求报价'）\n" "4. 工期/交货期承诺\n" "5. 质量保证承诺\n" "6. 投标有效期承诺\n" "7. 法定代表人签字和公司盖章声明\n" "使用正式的公文格式，语言庄重得体。" ) _SUMMARY_SYSTEM_PROMPT = ( "你是一个专业的投标文件撰写专家。请根据招标分析结果和已生成的投标内容，" "撰写一份投标概要（编制说明）。\n" "概要应包含：\n" "1. 项目背景和理解\n" "2. 我方优势概述\n" "3. 本投标文件的编制思路和结构\n" "4. 核心承诺\n" "语言简洁有力，突出我方优势。" ) class ProposalGenerator: """投标内容生成器。 根据招标分析结果，使用OpenAI生成对应的投标响应内容。 支持结构化输出（JSON Schema），可按类别采用不同生成策略。 """ def __init__(self, api_key: Optional[str] = None, model: Optional[str] = None): config = get_config() self.client = OpenAI( api_key=api_key or config.openai_api_key, timeout=config.openai_timeout, ) self.model = model or config.openai_model self.temperature = config.openai_temperature self.max_tokens = config.openai_max_tokens def generate(self, analysis: AnalysisResult) -> ProposalDocument: """根据招标要求生成完整的投标内容。 对每个 Requirement 调用API生成响应，同时生成投标函和概要。 多个要求的生成任务并行执行以提高效率。 Args: analysis: 招标分析结果 Returns: ProposalDocument: 完整的投标内容 Raises: ValueError: 如果API密钥未配置或analysis为空 RuntimeError: 如果所有要求生成均失败 """ if not self.client.api_key: raise ValueError( "OpenAI API密钥未配置。请设置 OPENAI_API_KEY 环境变量或在调用时传入 api_key。" ) project_name = analysis.project_name or "未命名项目" sections: list[ProposalSection] = [] errors: list[str] = [] # 获取所有要求并生成响应 all_reqs = analysis.all_requirements logger.info( "开始生成投标内容：项目 '%s'，共 %d 项要求", project_name, len(all_reqs), ) # 并行生成各个要求的响应 with ThreadPoolExecutor(max_workers=min(len(all_reqs), 10)) as executor: future_map = { executor.submit(self._generate_section, req): req for req in all_reqs } for future in as_completed(future_map): req = future_map[future] try: section = future.result() sections.append(section) logger.info("生成完成：%s", section.title) except Exception as e: error_msg = f"要求 '{req.title[:40]}...' 生成失败: {e}" errors.append(error_msg) logger.error(error_msg) # 按层级排序，保持文档结构合理 sections.sort(key=lambda s: (s.level, s.title)) if not sections: raise RuntimeError( "所有要求的响应生成均失败。\n " + "\n ".join(errors) ) # 生成投标函 logger.info("开始生成投标函...") try: cover_letter = self._generate_cover_letter(project_name) logger.info("投标函生成完成") except Exception as e: cover_letter = self._fallback_cover_letter(project_name) logger.warning("投标函AI生成失败，使用模板: %s", e) # 添加投标函作为首个章节 cover_section = ProposalSection( title="投标函", level=1, content=cover_letter, requirement_ref="投标函", ) sections.insert(0, cover_section) # 构建完整的投标文档 proposal = ProposalDocument( title=project_name, project_name=project_name, sections=sections, ) # 生成投标概要 logger.info("开始生成投标概要...") try: proposal.summary = self._generate_summary(analysis, proposal) logger.info("投标概要生成完成") except Exception as e: proposal.summary = self._fallback_summary(analysis) logger.warning("投标概要AI生成失败，使用模板: %s", e) if errors: logger.warning( "部分要求生成失败（%d/%d）：\n %s", len(errors), len(all_reqs), "\n ".join(errors), ) return proposal def _build_section_prompt(self, req: Requirement) -> str: """根据要求类别构建对应的系统提示词。""" base_prompt = _CATEGORY_PROMPTS.get(req.category, _DEFAULT_CATEGORY_PROMPT) score_hint = "" if req.score is not None and req.score > 0: score_hint = f"\n【分值权重】此项评审分值为 {req.score} 分，请重点响应，确保覆盖所有得分点。" detail_hint = "" if req.detail: detail_hint = f"\n【详细说明】{req.detail}" return ( f"{base_prompt}" f"\n\n【要求类别】{req.category}" f"{score_hint}" f"{detail_hint}" ) def _generate_section(self, req: Requirement) -> ProposalSection: """为单个要求生成响应内容。 Args: req: 招标要求 Returns: ProposalSection: 生成的响应章节 Raises: Exception: API调用失败时抛出 """ system_prompt = self._build_section_prompt(req) user_prompt = ( f"请针对以下招标要求撰写响应内容：\n\n" f"【要求标题】{req.title}\n" f"【要求描述】{req.description}\n\n" f"请按 JSON Schema 格式返回，包含以下字段：\n" f"- title: 响应章节标题\n" f"- content: 响应正文（使用 markdown 格式，可包含段落、列表、表格等）\n" f"- requirement_ref: 引用原文要求（简明扼要）\n" f"- level: 章节层级（1=一级标题, 2=二级标题, 3=三级标题）" ) try: response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], response_format={ "type": "json_schema", "json_schema": _SECTION_SCHEMA, }, temperature=self.temperature, max_tokens=self.max_tokens, ) result = json.loads(response.choices[0].message.content) return ProposalSection( title=result.get("title", f"响应：{req.title[:30]}"), content=result.get("content", ""), requirement_ref=result.get("requirement_ref", req.title), level=result.get("level", 2), ) except Exception as e: logger.error( "生成要求 '%s...' 的响应时出错: %s", req.title[:50], e, ) raise def _generate_cover_letter(self, project_name: str) -> str: """生成投标函。 Args: project_name: 项目名称 Returns: str: 投标函文本（Markdown格式） """ try: response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": _COVER_LETTER_SYSTEM_PROMPT}, { "role": "user", "content": ( f"请为以下项目撰写正式投标函：\n\n" f"项目名称：{project_name}\n\n" f"使用正式的公文格式撰写，包含投标人信息（请使用'我方'代称），" f"语言庄重得体。" ), }, ], temperature=self.temperature, max_tokens=self.max_tokens, ) return response.choices[0].message.content.strip() except Exception as e: logger.error("投标函生成失败: %s", e) raise def _generate_summary( self, analysis: AnalysisResult, proposal: Optional[ProposalDocument] = None ) -> str: """生成投标概要/编制说明。 Args: analysis: 招标分析结果 proposal: 已生成的投标内容（可选，用于参考） Returns: str: 投标概要文本（Markdown格式） """ sections_summary = "" if proposal and proposal.sections: section_titles = "\n".join( f"- {s.title}" for s in proposal.sections[:10] ) sections_summary = f"\n本投标文件包含以下章节：\n{section_titles}" try: response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": _SUMMARY_SYSTEM_PROMPT}, { "role": "user", "content": ( f"项目名称：{analysis.project_name or '未命名'}\n" f"招标概要：{analysis.summary}\n\n" f"请撰写投标概要/编制说明。" ), }, ], temperature=self.temperature, max_tokens=self.max_tokens, ) return response.choices[0].message.content.strip() except Exception as e: logger.error("投标概要生成失败: %s", e) raise def _fallback_cover_letter(self, project_name: str) -> str: """当API不可用时使用的备用投标函模板。""" import datetime today = datetime.date.today().strftime("%Y年%m月%d日") return ( f"# 投标函\n\n" f"致：招标人/采购代理机构\n\n" f"我方仔细研究了《{project_name}》招标文件的全部内容，决定参与本项目的投标。\n\n" f"1. 我方愿意按照招标文件要求，承担本项目的实施工作。\n\n" f"2. 我方承诺按招标文件规定的工期和质量标准完成本项目。\n\n" f"3. 我方承诺本投标文件在投标有效期内保持有效。\n\n" f"4. 我方郑重声明，所递交的投标文件及有关资料内容完整、真实、准确。\n\n" f"5. 我方承诺将严格按照合同约定履行各项义务。\n\n" f"特此函告。\n\n" f"投标人：我方\n" f"日期：{today}" ) def _fallback_summary(self, analysis: AnalysisResult) -> str: """当API不可用时使用的备用概要模板。""" all_reqs = analysis.all_requirements categories = set(r.category for r in all_reqs) category_names = { "evaluation": "评审标准响应", "service": "服务方案", "qualification": "资质证明", "other": "其他响应", } category_strs = [category_names.get(c, c) for c in categories if c in category_names] return ( f"## 编制说明\n\n" f"### 1. 项目概述\n\n" f"本项目为「{analysis.project_name or '本次招标项目'}」，" f"我方在认真研究招标文件的基础上，编制本投标文件。\n\n" f"### 2. 编制依据\n\n" f"- 本项目招标文件及其补充文件\n" f"- 国家相关法律法规和技术标准\n" f"- 我方多年来在相关领域的项目经验\n\n" f"### 3. 投标文件结构\n\n" f"本投标文件针对以下类别的要求进行了详细响应：\n" + "\n".join(f"- {cat}" for cat in category_strs) + f"\n\n### 4. 核心承诺\n\n" f"我方郑重承诺，将严格按照招标文件和合同约定，" f"以优质的服务和专业技术，确保项目圆满完成。" )