import json import logging from dataclasses import dataclass from typing import Any import httpx from app.config import settings from app.crawler.parsers import extract_page_content logger = logging.getLogger(__name__) @dataclass class AiResult: """DeepSeek 分析结果""" is_relevant: bool = False reason: str = "" business_type: str = "" error: str | None = None content_snippet: str | None = None # 提取到的正文前 200 字,供入库参考 class AiAnalyzer: """AI 分析器 — 调用 DeepSeek 判断公告是否为中国电信可承接项目""" def __init__(self): self.api_key = settings.ai_api_key self.base_url = settings.ai_base_url.rstrip("/") self.model = settings.ai_model self.timeout = settings.ai_timeout self.prompt_template = settings.ai_prompt_template async def analyze(self, announcement: dict[str, Any]) -> AiResult: """分析单条公告""" if not self.api_key: return AiResult(error="AI_API_KEY 未配置") # 1. 获取公告正文 content_url = announcement.get("content_url", "") content = None content_snippet = None if content_url: content = await extract_page_content(content_url, self.timeout) if content: content_snippet = content[:200] else: logger.warning("无法获取公告正文: %s", content_url) # 2. 构建 prompt prompt = self.prompt_template.format( title=announcement.get("title", ""), purchase_name=announcement.get("purchase_name", ""), announcement_type=announcement.get("announcement_type", ""), content=content or "(无法获取正文,请仅根据标题和采购人信息判断)", ) # 3. 调用 DeepSeek API try: result = await self._call_deepseek(prompt) if result.error: return AiResult(error=result.error, content_snippet=content_snippet) return AiResult( is_relevant=result.is_relevant, reason=result.reason, business_type=result.business_type, content_snippet=content_snippet, ) except Exception as e: logger.exception("AI 分析异常") return AiResult(error=str(e), content_snippet=content_snippet) async def analyze_batch( self, announcements: list[dict[str, Any]], max_concurrent: int = 3 ) -> list[AiResult]: """批量分析,控制并发数""" import asyncio sem = asyncio.Semaphore(max_concurrent) async def _limited(ann: dict[str, Any]) -> AiResult: async with sem: return await self.analyze(ann) tasks = [_limited(ann) for ann in announcements] return await asyncio.gather(*tasks) async def _call_deepseek(self, prompt: str) -> AiResult: """调用 DeepSeek Chat API""" url = f"{self.base_url}/chat/completions" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } payload = { "model": self.model, "messages": [ { "role": "system", "content": "你是一个专业的政府采购项目分析师。请根据公告信息判断是否为中国电信可以承接的项目,并用 JSON 格式回答。", }, {"role": "user", "content": prompt}, ], "temperature": 0.3, # 低温度,提高判断一致性 "max_tokens": 512, } async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.post(url, headers=headers, json=payload) if response.status_code != 200: return AiResult( error=f"API 请求失败 (HTTP {response.status_code}): {response.text[:200]}" ) data = response.json() choices = data.get("choices", []) if not choices: return AiResult(error="API 返回空 choices") content = choices[0].get("message", {}).get("content", "") return self._parse_response(content) @staticmethod def _parse_response(content: str) -> AiResult: """从 LLM 回复中提取 JSON 结果""" # 清理可能的 markdown 代码块标记 content = content.strip() if content.startswith("```"): # 移除 ```json 或 ``` 包裹 lines = content.split("\n") if lines[0].strip().startswith("```"): lines = lines[1:] if lines and lines[-1].strip() == "```": lines = lines[:-1] content = "\n".join(lines).strip() try: result = json.loads(content) return AiResult( is_relevant=bool(result.get("is_relevant", False)), reason=str(result.get("reason", "")), business_type=str(result.get("business_type", "")), ) except (json.JSONDecodeError, ValueError) as e: logger.warning("JSON 解析失败: %s\n原始内容: %s", e, content[:200]) return AiResult(error=f"JSON 解析失败: {e}")