""" AI 成本变化分析服务 使用 LLM 生成成本变化原因摘要和建议追问问题 """ import json import logging from typing import Any, Dict, List, Optional from app.core.config import get_settings logger = logging.getLogger(__name__) settings = get_settings() # Prompt 模板 COST_ANALYSIS_PROMPT = """你是一个专业的财务分析师。请根据以下成本对比数据，生成简洁的成本变化分析摘要。 ## 当前月数据 - 工资成本: {curr_salary} - 社保成本: {curr_social} - 公积金成本: {curr_fund} - 总成本: {curr_total} - 员工人数: {curr_count} ## 上月数据 - 工资成本: {prev_salary} - 社保成本: {prev_social} - 公积金成本: {prev_fund} - 总成本: {prev_total} - 员工人数: {prev_count} ## 变化分析 - 新增员工: {new_count} 人，新增成本 {new_cost} - 离职员工: {left_count} 人，减少成本 {left_cost} - 薪资调整: {adjustment_count} 人，净变化 {adjustment_cost} ## 要求 1. 用2-3句话概括成本变化的主要原因 2. 包含具体金额和比例 3. 指出top变化项 4. 输出格式：纯文本，不要Markdown 请直接输出分析摘要： """ SUGGESTED_QUESTIONS_PROMPT = """基于以下成本分析数据，生成3-5个用户可能追问的问题。 ## 成本变化数据 {change_data} ## 要求 1. 问题要具体、有针对性 2. 关注关键变化项 3. 输出JSON数组格式：["问题1", "问题2", ...] 请直接输出JSON： """ class CostAnalyzerService: """AI 成本变化分析服务""" async def analyze_cost_changes( self, current_data: Dict[str, Any], previous_data: Dict[str, Any], changes: Dict[str, Any], ) -> str: """ 使用 LLM 生成成本变化原因摘要 Args: current_data: 当前月成本数据 previous_data: 上月成本数据 changes: 变化分析数据 Returns: AI 生成的分析摘要文本 """ prompt = COST_ANALYSIS_PROMPT.format( curr_salary=current_data.get("salary_cost", 0), curr_social=current_data.get("social_security_cost", 0), curr_fund=current_data.get("fund_cost", 0), curr_total=current_data.get("total_cost", 0), curr_count=current_data.get("employee_count", 0), prev_salary=previous_data.get("salary_cost", 0), prev_social=previous_data.get("social_security_cost", 0), prev_fund=previous_data.get("fund_cost", 0), prev_total=previous_data.get("total_cost", 0), prev_count=previous_data.get("employee_count", 0), new_count=len(changes.get("new_employees", [])), new_cost=changes.get("new_employee_cost", 0), left_count=len(changes.get("left_employees", [])), left_cost=changes.get("left_employee_saving", 0), adjustment_count=len(changes.get("salary_adjustments", [])), adjustment_cost=changes.get("adjustment_cost", 0), ) try: result = await self._call_llm(prompt) return result.strip() except Exception as e: logger.error(f"AI 成本分析失败: {e}") # 兜底：生成规则化摘要 return self._generate_fallback_summary(current_data, previous_data, changes) async def generate_suggested_questions( self, analysis: Dict[str, Any] ) -> List[str]: """ 基于变化分析生成可追问的问题 Args: analysis: 变化分析数据 Returns: 建议问题列表 """ change_summary = json.dumps(analysis, ensure_ascii=False, default=str) prompt = SUGGESTED_QUESTIONS_PROMPT.format(change_data=change_summary) try: result = await self._call_llm(prompt) questions = json.loads(result.strip()) if isinstance(questions, list): return questions[:5] except Exception as e: logger.error(f"生成建议问题失败: {e}") # 兜底问题 return self._generate_fallback_questions(analysis) def _generate_fallback_summary( self, current_data: Dict[str, Any], previous_data: Dict[str, Any], changes: Dict[str, Any], ) -> str: """规则兜底：生成摘要""" curr_total = current_data.get("total_cost", 0) prev_total = previous_data.get("total_cost", 0) change = curr_total - prev_total ratio = (change / prev_total * 100) if prev_total > 0 else 0 parts = [] if change > 0: parts.append(f"本月人工成本较上月增加 {change:.2f} 元（{ratio:.1f}%）") elif change < 0: parts.append(f"本月人工成本较上月减少 {abs(change):.2f} 元（{abs(ratio):.1f}%）") else: parts.append("本月人工成本与上月持平") new_count = len(changes.get("new_employees", [])) left_count = len(changes.get("left_employees", [])) if new_count > 0: parts.append(f"新增 {new_count} 名员工增加成本 {changes.get('new_employee_cost', 0):.2f} 元") if left_count > 0: parts.append(f"离职 {left_count} 名员工减少成本 {changes.get('left_employee_saving', 0):.2f} 元") adj_count = len(changes.get("salary_adjustments", [])) if adj_count > 0: parts.append(f"{adj_count} 名员工薪资调整净变化 {changes.get('adjustment_cost', 0):.2f} 元") return "，".join(parts) + "。" def _generate_fallback_questions(self, analysis: Dict[str, Any]) -> List[str]: """规则兜底：生成问题""" questions = [] new_emps = analysis.get("new_employees", []) left_emps = analysis.get("left_employees", []) adjustments = analysis.get("salary_adjustments", []) if new_emps: questions.append(f"新增的 {len(new_emps)} 名员工分布在哪些部门？") if left_emps: questions.append(f"离职的 {len(left_emps)} 名员工减少了多少成本？") if adjustments: top = max(adjustments, key=lambda x: abs(x.get("change", 0))) questions.append(f"薪资调整幅度最大的是谁？变化了多少？") questions.append("哪个部门成本变化最大？") questions.append("社保和公积金成本占比如何？") return questions[:5] async def _call_llm(self, prompt: str) -> str: """调用 LLM API""" provider = settings.ai_provider if provider == "zhipu" and settings.zhipu_api_key: return await self._call_zhipu(prompt) elif settings.openai_api_key: return await self._call_openai(prompt) else: raise ValueError("未配置 AI API Key") async def _call_openai(self, prompt: str) -> str: """调用 OpenAI API""" from openai import AsyncOpenAI client = AsyncOpenAI(api_key=settings.openai_api_key) response = await client.chat.completions.create( model=settings.openai_model, messages=[{"role": "user", "content": prompt}], temperature=0.3, max_tokens=500, ) return response.choices[0].message.content or "" async def _call_zhipu(self, prompt: str) -> str: """调用智谱 AI API""" import httpx url = "https://open.bigmodel.cn/api/paas/v4/chat/completions" headers = { "Authorization": f"Bearer {settings.zhipu_api_key}", "Content-Type": "application/json", } payload = { "model": "glm-4-flash", "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, "max_tokens": 500, } async with httpx.AsyncClient(timeout=30) as client: response = await client.post(url, json=payload, headers=headers) response.raise_for_status() data = response.json() return data["choices"][0]["message"]["content"]