实现 bid_proposal 包的后半部分:投标内容生成(proposal_generator.py)、文档输出(proposal_writer.py)、流程编排(pipeline.py)、命令行入口(cli.py / __main__.py)。
先决条件:A1 工作包已完成,models.py、config.py、bid_reader.py、requirement_analyzer.py 已就位。
| 文件 | 说明 |
|---|---|
bid_proposal/proposal_generator.py |
投标内容生成 |
bid_proposal/proposal_writer.py |
投标文件 DOCX 输出 |
bid_proposal/pipeline.py |
流程编排 |
bid_proposal/cli.py |
命令行入口 |
bid_proposal/__main__.py |
python -m 支持 |
proposal_generator.py — 投标内容生成接收 A1 输出的 AnalysisResult(已分析的结构化需求),使用 OpenAI API 逐项生成对应的投标响应内容,输出 ProposalDocument。
from openai import OpenAI
from .config import Config
from .models import (
AnalysisResult, Requirement,
ProposalSection, ProposalDocument, GenerationError,
)
class ProposalGenerator:
"""投标内容生成器。"""
def __init__(self, api_key: str | None = None, model: str = "gpt-4o"):
"""初始化生成器。
Args:
api_key: OpenAI API Key,默认从环境变量读取
model: 使用的模型名称(默认 gpt-4o)
Raises:
ValueError: API Key 未提供且环境变量未设置
"""
api_key = api_key or Config.get_openai_api_key()
self.client = OpenAI(api_key=api_key, timeout=120)
self.model = model
def generate(self, analysis: AnalysisResult) -> ProposalDocument:
"""根据招标分析结果生成投标内容。
生成策略:
1. 根据 AnalysisResult 中各分类下的 Requirement 规划章节结构
2. 按分类逐项生成响应内容(每项调用一次 API)
3. 合并为 ProposalDocument
Args:
analysis: A1 输出的招标分析结果
Returns:
ProposalDocument: 完整的投标内容
Raises:
GenerationError: 内容生成失败
"""
def _build_sections(self, analysis: AnalysisResult) -> list[dict]:
"""根据分析结果规划章节结构。
返回每个章节的规划信息:
- 章节标题
- 对应哪些 Requirement
- 章节层级
"""
sections = []
# 章节 1:项目概述(固定)
sections.append({
"title": "项目概述",
"level": 1,
"requirements": [],
"description": f"对{analysis.project_name}项目的整体理解和响应承诺",
})
# 章节 2:技术方案响应(对 evaluation_criteria 的响应)
if analysis.evaluation_criteria:
sections.append({
"title": "技术方案",
"level": 1,
"requirements": analysis.evaluation_criteria,
"description": "针对评审标准的逐项响应方案",
})
# 章节 3:服务方案(对 service_requirements 的响应)
if analysis.service_requirements:
sections.append({
"title": "服务方案",
"level": 1,
"requirements": analysis.service_requirements,
"description": "针对招标服务要求的详细服务方案",
})
# 章节 4:资质证明(对 qualification_requirements 的响应)
if analysis.qualification_requirements:
sections.append({
"title": "资质与资格证明",
"level": 1,
"requirements": analysis.qualification_requirements,
"description": "资质响应及证明材料说明",
})
# 章节 5:其他响应(对 other_requirements 的响应)
if analysis.other_requirements:
sections.append({
"title": "其他承诺与说明",
"level": 1,
"requirements": analysis.other_requirements,
"description": "对其他要求的响应说明",
})
return sections
生成单个章节的 System Prompt:
SYSTEM_PROMPT = """你是一个专业的投标文件撰写专家。请根据招标要求生成对应的投标响应内容。
生成原则:
1. 内容必须紧扣招标要求,逐条响应
2. 使用正式、专业的标书语言
3. 具体而非空泛(提供可量化的承诺、具体方案)
4. 突出优势,体现对项目的深入理解
5. 对于有明确分值的评分项,确保响应内容充分覆盖评分要点
输出格式:
- 每个响应以「针对《XXX》要求」开头
- 正文使用 Markdown 格式(支持标题 ##、段落、列表 -、表格)
- 条理清晰,分点阐述"""
def _build_section_prompt(
self,
section_title: str,
section_desc: str,
requirements: list[Requirement],
project_context: str,
) -> tuple[str, str]:
"""构建章节生成的 system prompt 和 user prompt。"""
reqs_text = ""
for i, req in enumerate(requirements, 1):
reqs_text += f"\n【要求 {i}】\n"
reqs_text += f"标题:{req.title}\n"
reqs_text += f"描述:{req.description}\n"
if req.score is not None:
reqs_text += f"分值:{req.score}分\n"
if req.detail:
reqs_text += f"评分标准:{req.detail}\n"
user_message = (
f"项目背景:{project_context}\n\n"
f"请生成以下章节的投标响应内容:\n"
f"章节标题:{section_title}\n"
f"章节说明:{section_desc}\n\n"
f"需要响应的招标要求:{reqs_text}\n\n"
f"请生成完整的章节内容,包含小标题和正文。"
)
return SYSTEM_PROMPT, user_message
def generate(self, analysis: AnalysisResult) -> ProposalDocument:
sections_plan = self._build_sections(analysis)
proposal_sections = []
for plan in sections_plan:
if plan["requirements"]:
# 有具体需求项,逐项生成
sys_prompt, user_msg = self._build_section_prompt(
plan["title"],
plan["description"],
plan["requirements"],
f"项目名称:{analysis.project_name}\n招标概况:{analysis.summary}",
)
try:
section_content = self._call_api(sys_prompt, user_msg)
except Exception as e:
logger.warning(f"章节「{plan['title']}」生成失败: {e}")
section_content = "【待人工补充】\n\n该部分内容生成失败,请根据招标要求手动补充。"
else:
# 无具体需求项(如项目概述),生成概要内容
section_content = self._generate_overview(analysis)
proposal_sections.append(ProposalSection(
title=plan["title"],
level=plan["level"],
content=section_content,
requirement_ref=", ".join(
r.title for r in plan.get("requirements", [])
),
))
return ProposalDocument(
title=f"{analysis.project_name}投标文件",
project_name=analysis.project_name,
sections=proposal_sections,
summary=analysis.summary,
)
import time
import logging
logger = logging.getLogger(__name__)
def _call_api(self, system_prompt: str, user_message: str) -> str:
"""调用 OpenAI API 生成内容,带重试逻辑。"""
last_error = None
for attempt in range(3):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
temperature=0.7,
max_tokens=4096,
)
return response.choices[0].message.content
except Exception as e:
last_error = e
if attempt < 2:
wait = 2 ** attempt
logger.warning(f"API 调用失败(第{attempt+1}次),{wait}秒后重试: {e}")
time.sleep(wait)
raise GenerationError(f"内容生成失败(已重试3次): {last_error}")
def _generate_overview(self, analysis: AnalysisResult) -> str:
"""生成项目概述章节内容。"""
text = (
f"## 项目背景\n\n"
f"{analysis.summary}\n\n"
f"## 响应承诺\n\n"
f"我公司对{analysis.project_name}项目进行了深入研究和全面分析,"
f"充分理解招标人的需求和评审标准。"
f"我公司郑重承诺将严格按照招标文件要求,"
f"提供优质的{'/'.join(
r.title for r in (
analysis.service_requirements
+ analysis.evaluation_criteria
)[:3]
)}服务。\n\n"
f"## 项目优势\n\n"
f"我公司在相关领域具有丰富的经验和专业团队,"
f"能够为本项目提供有力保障。"
)
return text
temperature=0.7 在创造性写作和一致性之间取得平衡proposal_writer.py — 投标文件输出将 ProposalDocument 输出为格式化的 DOCX 投标文件。样式设计符合投标文件的正式格式。
from .models import ProposalDocument, ProposalSection, WriteError
def write_proposal_docx(
proposal: ProposalDocument,
output_path: str,
*,
page_width_cm: float = 21.0,
page_height_cm: float = 29.7,
margin_cm: float = 2.5,
company_name: str = "投标单位名称",
) -> str:
"""将投标内容写入格式化的 DOCX 文件。
生成的文档结构:
┌────────────────────────────────┐
│ 投 标 文 件 │ ← 封面(大号居中)
│ │
│ 项目名称:xxx │
│ 投标单位:xxx │
│ 日 期:xxxx-xx-xx │
├────────────────────────────────┤
│ 目 录 │ ← 目录页
├────────────────────────────────┤
│ 一、项目概述 │ ← Heading 1
│ ...... │
│ 二、技术方案 │ ← Heading 1
│ 2.1 评审项响应 │ ← Heading 2
│ ...... │
│ 三、服务方案 │ ← Heading 1
│ ...... │
│ 四、资质与资格证明 │ ← Heading 1
│ ...... │
│ 五、其他承诺与说明 │ ← Heading 1
│ ...... │
└────────────────────────────────┘
Args:
proposal: 投标内容
output_path: 输出 DOCX 路径
page_width_cm: 页面宽度(默认 A4)
page_height_cm: 页面高度(默认 A4)
margin_cm: 页边距
company_name: 投标单位名称(用于封面)
Returns:
str: 输出文件路径
Raises:
WriteError: DOCX 写入失败
"""
页面设置:
封面设计:
正文排版:
Markdown 渲染(解析 section.content 中的内容):
# / ## / ### → Heading 1 / 2 / 3**文本** → 加粗- item → 无序列表(符号:圆点)1. item → 有序列表页脚:居中的页码(第 X 页)
import logging
import os
import datetime
from docx import Document
from docx.shared import Pt, Cm, Emu
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.oxml.ns import qn, nsdecls
from docx.oxml import parse_xml
logger = logging.getLogger(__name__)
def write_proposal_docx(proposal, output_path, **kwargs):
doc = Document()
# ---- 1. 设置页面 ----
section = doc.sections[0]
section.page_width = Cm(page_width_cm)
section.page_height = Cm(page_height_cm)
section.top_margin = Cm(margin_cm)
section.bottom_margin = Cm(margin_cm)
section.left_margin = Cm(margin_cm)
section.right_margin = Cm(margin_cm)
# ---- 2. 添加封面 ----
_add_cover_page(doc, proposal, company_name)
doc.add_page_break()
# ---- 3. 添加目录 ----
_add_toc(doc)
doc.add_page_break()
# ---- 4. 添加正文章节 ----
for section_data in proposal.sections:
_add_section_to_doc(doc, section_data)
doc.add_paragraph() # 章节间距
# ---- 5. 添加页脚 ----
_add_page_number(doc)
# ---- 6. 保存 ----
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
doc.save(output_path)
logger.info(f"投标文件已保存至: {output_path}")
return output_path
def _add_cover_page(doc, proposal, company_name):
"""添加封面页。"""
# 空行占位,将标题推到页面中间
for _ in range(6):
doc.add_paragraph()
# 项目名称
p = doc.add_paragraph()
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
run = p.add_run(proposal.project_name)
run.font.size = Pt(16)
run.font.bold = True
run.font.name = "宋体"
run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
# 间距
doc.add_paragraph()
# "投标文件" 标题
p = doc.add_paragraph()
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
run = p.add_run("投 标 文 件")
run.font.size = Pt(28)
run.font.bold = True
run.font.name = "宋体"
run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
# 间距
for _ in range(4):
doc.add_paragraph()
# 投标单位
p = doc.add_paragraph()
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
run = p.add_run(f"投标单位:{company_name}")
run.font.size = Pt(14)
run.font.name = "宋体"
run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
# 日期
p = doc.add_paragraph()
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
run = p.add_run(f"日 期:{datetime.date.today().strftime('%Y年%m月%d日')}")
run.font.size = Pt(14)
run.font.name = "宋体"
run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
def _add_toc(doc):
"""添加目录页。"""
p = doc.add_paragraph()
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
run = p.add_run("目 录")
run.font.size = Pt(18)
run.font.bold = True
run.font.name = "宋体"
run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
doc.add_paragraph()
# 插入 Word TOC 超链接字段(Word 中按 Ctrl+A → F9 更新)
paragraph = doc.add_paragraph()
run = paragraph.add_run()
fld_begin = parse_xml(
f'<w:fldChar {nsdecls("w")} w:fldCharType="begin"/>'
)
run._element.append(fld_begin)
run2 = paragraph.add_run()
instr = parse_xml(
f'<w:instrText {nsdecls("w")} xml:space="preserve">'
f' TOC \\o "1-3" \\h \\z \\u '
f'</w:instrText>'
)
run2._element.append(instr)
run3 = paragraph.add_run()
fld_end = parse_xml(
f'<w:fldChar {nsdecls("w")} w:fldCharType="end"/>'
)
run3._element.append(fld_end)
def _add_section_to_doc(doc, section: ProposalSection):
"""将章节内容写入文档(支持简单 Markdown 解析)。"""
# 章节标题
heading = doc.add_heading(section.title, level=section.level)
for run in heading.runs:
run.font.name = "宋体"
run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
if not section.content:
return
# 解析 Markdown 内容并写入
lines = section.content.split("\n")
for line in lines:
stripped = line.strip()
if not stripped:
continue
# 标题解析
if stripped.startswith("### "):
h = doc.add_heading(stripped[4:], level=3)
elif stripped.startswith("## "):
h = doc.add_heading(stripped[3:], level=2)
elif stripped.startswith("# "):
h = doc.add_heading(stripped[2:], level=1)
elif stripped.startswith("- ") or stripped.startswith("* "):
# 无序列表
p = doc.add_paragraph(style="List Bullet")
run = p.add_run(stripped[2:])
_set_run_font(run)
elif stripped[0].isdigit() and ". " in stripped[:4]:
# 有序列表
p = doc.add_paragraph(style="List Number")
run = p.add_run(stripped.split(". ", 1)[1])
_set_run_font(run)
else:
# 普通正文
p = doc.add_paragraph()
run = p.add_run(stripped)
_set_run_font(run)
p.paragraph_format.first_line_indent = Cm(0.75)
def _set_run_font(run):
"""设置运行文本的字体格式。"""
run.font.size = Pt(12)
run.font.name = "宋体"
run._element.rPr.rFonts.set(qn("w:eastAsia"), "宋体")
def _add_page_number(doc):
"""添加页脚页码。"""
from docx.oxml import OxmlElement
section = doc.sections[0]
footer = section.footer
footer.is_linked_to_previous = False
p = footer.paragraphs[0]
p.alignment = WD_ALIGN_PARAGRAPH.CENTER
# "第 " 文本
run = p.add_run("第 ")
run.font.size = Pt(10)
# 页码字段
fld_char1 = OxmlElement("w:fldChar")
fld_char1.set(qn("w:fldCharType"), "begin")
run2 = p.add_run()
run2._element.append(fld_char1)
instr = OxmlElement("w:instrText")
instr.set(qn("xml:space"), "preserve")
instr.text = " PAGE "
run3 = p.add_run()
run3._element.append(instr)
fld_char2 = OxmlElement("w:fldChar")
fld_char2.set(qn("w:fldCharType"), "end")
run4 = p.add_run()
run4._element.append(fld_char2)
# " 页" 文本
run5 = p.add_run(" 页")
run5.font.size = Pt(10)
rFonts.set(qn("w:eastAsia"), "宋体"))WriteErrorpdf_table_to_docx/docx_writer.py 中边框设置、列宽等写法pipeline.py — 流程编排编排读取→分析→生成→输出的完整流程,提供一键式 API。
import logging
import os
from .bid_reader import read_bid_pdf
from .requirement_analyzer import RequirementAnalyzer
from .proposal_generator import ProposalGenerator
from .proposal_writer import write_proposal_docx
logger = logging.getLogger(__name__)
def run_pipeline(
pdf_path: str,
output_path: str | None = None,
api_key: str | None = None,
model: str = "gpt-4o",
company_name: str = "投标单位名称",
) -> str:
"""运行完整的招标→投标生成流程。
流程步骤:
1. read_bid_pdf(pdf_path) → BidDocument (招标文件读取)
2. RequirementAnalyzer.analyze(doc) → AnalysisResult (需求分析)
3. ProposalGenerator.generate(analysis) → ProposalDocument (内容生成)
4. write_proposal_docx(proposal, path) → str (DOCX输出)
每一步都有详细的日志输出,方便跟踪进度。
Args:
pdf_path: 招标文件 PDF 路径
output_path: 输出 DOCX 路径(默认与输入同目录)
api_key: OpenAI API Key(默认从环境变量读取)
model: 使用的模型名
company_name: 投标单位名称(用于封面)
Returns:
str: 输出 DOCX 文件路径
Raises:
FileNotFoundError: PDF 文件不存在
BidReadError: 招标文件读取失败
AnalysisError: 需求分析失败
GenerationError: 内容生成失败
WriteError: DOCX 输出失败
"""
# Step 1
logger.info("Step 1/4: 读取招标文件...")
bid_doc = read_bid_pdf(pdf_path)
logger.info(f" → 共 {bid_doc.total_pages} 页, {bid_doc.table_count} 个表格")
# Step 2
logger.info("Step 2/4: 分析招标要求(调用 AI)...")
analyzer = RequirementAnalyzer(api_key=api_key, model=model)
analysis = analyzer.analyze(bid_doc)
total = (len(analysis.evaluation_criteria) + len(analysis.service_requirements)
+ len(analysis.qualification_requirements) + len(analysis.other_requirements))
logger.info(f" → 识别到 {total} 项要求")
logger.info(f" 项目名称: {analysis.project_name}")
# Step 3
logger.info("Step 3/4: 生成投标内容(调用 AI)...")
generator = ProposalGenerator(api_key=api_key, model=model)
proposal = generator.generate(analysis)
logger.info(f" → 生成 {len(proposal.sections)} 个章节")
# Step 4
if output_path is None:
base = os.path.splitext(pdf_path)[0]
output_path = f"{base}_投标文件.docx"
logger.info("Step 4/4: 输出投标文件...")
result = write_proposal_docx(proposal, output_path, company_name=company_name)
logger.info(f" → 已保存: {result}")
logger.info("=== 完成 ===")
return result
{原文件名}_投标文件.docxcli.py — 命令行入口"""bid_proposal 命令行入口。"""
import argparse
import logging
import sys
from .pipeline import run_pipeline
def setup_logging(verbose: bool = False):
level = logging.DEBUG if verbose else logging.INFO
fmt = "%(asctime)s [%(levelname)s] %(message)s"
datefmt = "%H:%M:%S"
logging.basicConfig(level=level, format=fmt, datefmt=datefmt, stream=sys.stderr)
def main():
parser = argparse.ArgumentParser(
description="AI 驱动的投标文件生成工具 — 根据招标文件 PDF 自动生成投标文件 DOCX",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
使用示例:
%(prog)s 招标文件.pdf # 一键生成
%(prog)s 招标文件.pdf -o 投标文件.docx # 指定输出
%(prog)s 招标文件.pdf --model gpt-4o --verbose # 指定模型+详细日志
%(prog)s 招标文件.pdf --company "XX公司" # 指定投标单位
""",
)
parser.add_argument("pdf_path", help="输入的招标文件 PDF 路径")
parser.add_argument("-o", "--output", help="输出 DOCX 文件路径(默认自动生成)")
parser.add_argument("--model", default="gpt-4o", help="OpenAI 模型名称(默认 gpt-4o)")
parser.add_argument("--api-key", help="OpenAI API 密钥(默认从 OPENAI_API_KEY 环境变量读取)")
parser.add_argument("--company", default="投标单位名称", help="投标单位名称(用于封面)")
parser.add_argument("--verbose", "-v", action="store_true", help="输出详细日志")
args = parser.parse_args()
setup_logging(verbose=args.verbose)
logger = logging.getLogger(__name__)
try:
output_path = run_pipeline(
pdf_path=args.pdf_path,
output_path=args.output,
api_key=args.api_key,
model=args.model,
company_name=args.company,
)
print(f"\n✅ 投标文件已生成: {output_path}", file=sys.stderr)
except FileNotFoundError as e:
logger.error(f"文件未找到: {e}")
sys.exit(1)
except Exception as e:
logger.exception(f"处理失败: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
__main__.py — 模块入口"""支持 python -m bid_proposal 调用。"""
from .cli import main
main()
A2 工作包依赖 A1 提供的以下接口:
# ---- 从 bid_proposal.models ----
BidDocument # 招标文件数据结构
Requirement # 需求项
AnalysisResult # 分析结果
ProposalSection # 投标章节
ProposalDocument # 投标内容
GenerationError # 生成异常
WriteError # 写入异常
# ---- 从 bid_proposal.bid_reader ----
read_bid_pdf(pdf_path: str) -> BidDocument
# ---- 从 bid_proposal.requirement_analyzer ----
RequirementAnalyzer(api_key: str | None, model: str)
RequirementAnalyzer.analyze(doc: BidDocument) -> AnalysisResult
# ---- 从 bid_proposal.config ----
Config.get_openai_api_key() -> str
Config.get_default_model() -> str
A2 对 AnalysisResult 的使用方式:
result.evaluation_criteria → 逐项生成评审响应(技术方案章节)
result.service_requirements → 逐项生成服务方案(服务方案章节)
result.qualification_requirements → 逐项生成资质响应(资质证明章节)
result.other_requirements → 逐项生成其他响应(其他承诺章节)
result.summary → 项目概况章节内容
result.project_name → 文档标题和封面
| # | 标准 | 验证方式 |
|---|---|---|
| 1 | ProposalGenerator.generate() 能基于 AnalysisResult 生成 ProposalDocument |
assert len(proposal.sections) > 0 |
| 2 | 生成的章节结构符合预期(5个章节:概述、技术方案、服务方案、资质、其他) | 检查标题列表 |
| 3 | write_proposal_docx() 输出有效的 DOCX 文件 |
os.path.exists(path) 且文件大小 > 1KB |
| 4 | DOCX 文件包含封面页(项目名称、"投标文件"居中标题) | 手动打开验证,或解析 XML 检查 |
| 5 | run_pipeline() 端到端执行成功 |
Mock OpenAI API 的情况下模拟完整流程 |
| 6 | CLI 调用正常 python -m bid_proposal <pdf> |
子进程测试,exit code 0 |
| 7 | 单个章节生成失败不影响其他章节 | Mock 部分 API 失败,验证其他章节仍正常 |
| 8 | Mock API 测试覆盖率 >= 80% | pytest --cov |
创建 bid_proposal/tests/test_a2.py,包含以下测试:
# 1. test_generator_success
# - Mock OpenAI API,验证 ProposalDocument 结构和章节数量
# 2. test_generator_partial_failure
# - Mock 部分 API 调用失败,验证降级为占位文本
# 3. test_generator_empty_requirements
# - 所有需求列表为空时,仍能生成合理的文档结构
# 4. test_writer_creates_docx
# - 验证输出有效的 DOCX 文件
# 5. test_writer_cover_page_and_toc
# - 验证封面和目录存在
# 6. test_pipeline_success (集成测试)
# - Mock 所有外部依赖,验证完整流程
# 7. test_pipeline_file_not_found
# - PDF 不存在时抛出 FileNotFoundError
# 8. test_cli_help
# - 验证 --help 输出
bid_proposal/ARCHITECTURE.mdbid_proposal/task_a1.mdpdf_table_to_docx/docx_writer.pydata/松江区机关事务管理局物业管理服务招标文件.pdf