""" requirement_analyzer.py - 招标要求分析模块 使用 OpenAI API 分析招标文件内容，识别并分类其中的各项要求。 """ import json import logging import time from typing import Optional from .config import get_config from .models import AnalysisResult, BidDocument, Requirement from .utils import chunk_text, table_to_markdown logger = logging.getLogger(__name__) class AnalysisError(Exception): """AI 分析失败异常""" pass class RequirementAnalyzer: """招标文件需求分析器。 使用 OpenAI 分析招标文件中的各项要求，包括评分标准、服务要求、 资质要求等，返回结构化的分析结果。 """ CATEGORY_MAP = { "evaluation": "评审标准", "service": "服务要求", "qualification": "资质要求", "other": "其他", } REVERSE_CATEGORY_MAP = {v: k for k, v in CATEGORY_MAP.items()} def __init__(self, api_key: Optional[str] = None, model: str = "gpt-4o"): """初始化分析器。 Args: api_key: OpenAI API Key，默认从环境变量读取 model: 使用的模型名称 """ try: from openai import OpenAI except ImportError: raise ImportError("请安装 openai: pip install openai") key = api_key or get_config().openai_api_key if not key: raise ValueError( "未提供 OpenAI API Key。请通过参数传入或设置 OPENAI_API_KEY 环境变量。" ) self.client = OpenAI(api_key=key) self.model = model def analyze(self, doc: BidDocument) -> AnalysisResult: """对招标文件进行 AI 分析，提取所有要求。 Args: doc: 招标文件结构化内容 Returns: AnalysisResult: 分析结果（结构化需求列表 + 摘要） Raises: AnalysisError: AI 分析失败 """ full_text = doc.full_text.strip() if not full_text: raise AnalysisError("招标文件内容为空，无法进行分析") logger.info("开始分析招标文件: %s (%d 字符)", doc.pdf_path, len(full_text)) config = get_config() max_length = config.max_text_length if len(full_text) <= max_length: prompt = self._build_prompt(doc) result = self._call_api(prompt) logger.info( "分析完成: 评审标准 %d 项, 服务要求 %d 项, 资质要求 %d 项, 其他 %d 项", len(result.evaluation_criteria), len(result.service_requirements), len(result.qualification_requirements), len(result.other_requirements), ) return result # 文本过长，分块处理 logger.info("文本过长 (%d 字符，超过 %d)，进行分块分析", len(full_text), max_length) return self._analyze_by_chunks(doc, full_text, max_length) def _analyze_by_chunks( self, doc: BidDocument, full_text: str, max_length: int ) -> AnalysisResult: """将长文本分块后分别分析，再合并结果。""" text_chunks = chunk_text(full_text, max_length) logger.info("文本已切分为 %d 个块", len(text_chunks)) all_reqs: list[Requirement] = [] summaries: list[str] = [] project_names: list[str] = [] for i, chunk in enumerate(text_chunks): logger.info("分析第 %d/%d 块...", i + 1, len(text_chunks)) partial_prompt = self._build_chunk_prompt(chunk, i + 1, len(text_chunks)) try: result = self._call_api(partial_prompt) all_reqs.extend(result.all_requirements) summaries.append(result.summary) if result.project_name: project_names.append(result.project_name) except AnalysisError as e: logger.warning("第 %d 块分析失败: %s，跳过该块", i + 1, e) # 合并结果，按类别分类 combined = AnalysisResult( summary="\n".join(summaries), project_name=project_names[0] if project_names else "", evaluation_criteria=[r for r in all_reqs if r.category == "evaluation"], service_requirements=[r for r in all_reqs if r.category == "service"], qualification_requirements=[r for r in all_reqs if r.category == "qualification"], other_requirements=[r for r in all_reqs if r.category == "other"], ) logger.info( "分块分析完成: %d 项要求", combined.total_requirements, ) return combined def _build_chunk_prompt(self, chunk: str, chunk_num: int, total_chunks: int) -> str: """为单个文本块构建分析 prompt。""" return f"""=== 招标文件内容（第 {chunk_num}/{total_chunks} 部分）=== {chunk} --- 请分析以上招标文件内容片段，提取其中包含的所有要求。""" def _build_prompt(self, doc: BidDocument) -> str: """构建包含招标文件完整内容的 prompt。 将文本和表格按页组织，方便 AI 理解内容结构。 """ sections: list[str] = ["=== 招标文件内容 ==="] # 构建表格索引：页码 → 表格索引 table_idx_by_page: dict[int, list[int]] = {} for ti, table in enumerate(doc.tables): page_num = getattr(table, 'page_num', 0) + 1 # page_num是0-indexed table_idx_by_page.setdefault(page_num, []).append(ti) for page in doc.pages: page_num = page.page_num sections.append(f"\n--- 第 {page_num} 页 ---") if page.text: sections.append(page.text) else: sections.append("（本页无文本内容）") # 添加本页中的表格 page_table_indices = table_idx_by_page.get(page_num, []) for ti in page_table_indices: if ti < len(doc.tables): table = doc.tables[ti] md_table = table_to_markdown(table) if md_table and md_table != "（空表格）": sections.append(f"\n本页中的表格：\n{md_table}") return "\n".join(sections) def _call_api(self, user_prompt: str) -> AnalysisResult: """调用 OpenAI API 进行需求分析，使用结构化输出。""" system_prompt = """你是一个专业的招标文件分析专家。请仔细阅读招标文件内容，提取并分类其中的所有要求。 请识别以下类别的要求： 1. **evaluation（评审标准）**：评审因素、评分项、分值分配、评审标准等 2. **service（服务要求）**：服务范围、服务内容、服务标准、人员配置等 3. **qualification（资质要求）**：投标人资格条件、资质证书、注册资金、业绩要求等 4. **other（其他）**：商务条款、技术规范、预算金额、付款方式、服务期限等 对于每一项要求，尽可能提取其标题(title)、详细描述(description)、分值(score)和评分标准详情(detail)。""" max_retries = 2 last_error: Optional[Exception] = None for attempt in range(max_retries + 1): try: response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=get_config().openai_temperature, max_tokens=get_config().openai_max_tokens, timeout=get_config().openai_timeout, response_format={ "type": "json_schema", "json_schema": { "name": "analysis_result", "strict": True, "schema": { "type": "object", "properties": { "summary": { "type": "string", "description": "招标文件概要总结，不超过300字", }, "project_name": { "type": "string", "description": "项目名称", }, "requirements": { "type": "array", "items": { "type": "object", "properties": { "category": { "type": "string", "enum": ["evaluation", "service", "qualification", "other"], "description": "要求类别", }, "title": { "type": "string", "description": "要求标题", }, "description": { "type": "string", "description": "详细描述", }, "score": { "type": ["number", "null"], "description": "分值（如明确有分值）", }, "detail": { "type": "string", "description": "评分标准详细说明", }, }, "required": ["category", "title", "description", "score", "detail"], "additionalProperties": False, }, }, }, "required": ["summary", "project_name", "requirements"], "additionalProperties": False, }, }, }, ) content = response.choices[0].message.content if not content: raise AnalysisError("API 返回了空响应") data = json.loads(content) # 按类别分类构建 Requirement all_reqs: list[Requirement] = [] for req in data.get("requirements", []): category = req.get("category", "other") # 校验category合法性 if category not in ("evaluation", "service", "qualification", "other"): category = "other" all_reqs.append(Requirement( category=category, title=req.get("title", ""), description=req.get("description", ""), score=req.get("score"), detail=req.get("detail", ""), )) result = AnalysisResult( summary=data.get("summary", ""), project_name=data.get("project_name", ""), evaluation_criteria=[r for r in all_reqs if r.category == "evaluation"], service_requirements=[r for r in all_reqs if r.category == "service"], qualification_requirements=[r for r in all_reqs if r.category == "qualification"], other_requirements=[r for r in all_reqs if r.category == "other"], ) logger.debug("API 分析成功: %d 项要求", result.total_requirements) return result except Exception as e: last_error = e if attempt < max_retries: wait = 2 ** (attempt + 1) logger.warning( "API 调用失败（第 %d 次），%ds 后重试: %s", attempt + 1, wait, e, ) time.sleep(wait) else: logger.error("API 调用最终失败: %s", e) raise AnalysisError(f"AI 分析失败，已重试 {max_retries} 次: {last_error}")