Hermes CI Fix
3dddd36866
包含前后端完整代码： - 前端：Vue3+Vite+ElementPlus - 后端：FastAPI+SQLAlchemy - 模块：驾驶舱/KPI/战略地图/预警/预算/成本/预测/改善行动 - 当前版本：v1.0.0
326 lines
13 KiB
Python
326 lines
13 KiB
Python
"""AI分析引擎 — 侧边栏智能分析"""
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Request
|
|
from fastapi.responses import StreamingResponse
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func, text as sa_text
|
|
from app.database import get_db
|
|
from app.auth_middleware import require_auth, require_role
|
|
from app.models import KPIDefinition, KPIValue, KPIAlert, StrategicMap, User, ActionPlan
|
|
from app.utils.cache import get as cache_get, set as cache_set
|
|
import json, hashlib, httpx, os
|
|
from datetime import datetime
|
|
router = APIRouter(prefix="/api/cma/ai", tags=["AI分析"],
|
|
dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
|
|
)
|
|
|
|
async def _call_deepseek(prompt: str) -> str:
|
|
"""调用DeepSeek API"""
|
|
api_key = os.getenv("DEEPSEEK_API_KEY", "sk-8e24e6eb87f2475e96ea0980002dc2e8")
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
resp = await client.post(
|
|
"https://api.deepseek.com/v1/chat/completions",
|
|
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
|
|
json={
|
|
"model": "deepseek-chat",
|
|
"messages": [
|
|
{"role": "system", "content": "你是一名CMA管理会计师，擅长用数据驱动的方式分析企业经营状况，给出专业的财务分析和管理建议。回答要简洁、专业、有数据支撑。"},
|
|
{"role": "user", "content": prompt}
|
|
],
|
|
"stream": False,
|
|
"temperature": 0.3,
|
|
}
|
|
)
|
|
data = resp.json()
|
|
return data.get("choices", [{}])[0].get("message", {}).get("content", "")
|
|
|
|
@router.get("/dashboard-analysis")
|
|
async def dashboard_analysis(role: str = Query("ceo"), db: Session = Depends(get_db)):
|
|
"""AI分析驾驶舱数据"""
|
|
# 尝试缓存
|
|
cache_key = f"dashboard_analysis:{role}"
|
|
cached = cache_get("ai", cache_key)
|
|
if cached:
|
|
return cached
|
|
# 获取当前KPI数据
|
|
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
|
|
kpi_summary = []
|
|
for k in kpis:
|
|
latest = db.query(KPIValue).filter(KPIValue.kpi_id == k.id).order_by(KPIValue.period.desc()).first()
|
|
kpi_summary.append({
|
|
"name": k.kpi_name,
|
|
"code": k.kpi_code,
|
|
"dimension": k.dimension,
|
|
"target": k.target_value,
|
|
"actual": latest.actual_value if latest else None,
|
|
"period": latest.period if latest else None,
|
|
"unit": k.unit,
|
|
})
|
|
|
|
# 获取预警
|
|
alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending").count()
|
|
|
|
# 构建分析prompt
|
|
kpi_text = "\n".join([f"- {k['name']}({k['code']}): 目标={k['target']}, 实际={k['actual']}({k['period']}), 维度={k['dimension']}" for k in kpi_summary if k['actual'] is not None])
|
|
|
|
prompt = f"""我是一家公司的管理层，以下是当前管理会计系统的KPI数据和系统状态，请给出专业的分析和管理建议：
|
|
|
|
当前KPI数据：
|
|
{kpi_text}
|
|
|
|
待处理预警数：{alerts}
|
|
|
|
请从以下三个方面分析：
|
|
1. **核心发现**：当前数据反映的最关键问题是什么？
|
|
2. **深入解读**：从CMA管理会计角度，这些数据意味着什么？
|
|
3. **行动建议**：基于数据，财务和业务部门应该采取什么具体行动？
|
|
|
|
注意：角色视角为{"CEO（总经理）" if role == "ceo" else "财务部" if role == "finance" else "业务部"}。"""
|
|
|
|
try:
|
|
analysis = await _call_deepseek(prompt)
|
|
except Exception as e:
|
|
analysis = f"AI分析暂时不可用: {str(e)}"
|
|
|
|
result = {"analysis": analysis, "kpi_count": len(kpi_summary), "alert_count": alerts}
|
|
# 缓存10分钟
|
|
cache_set("ai", cache_key, result, ttl_seconds=600)
|
|
return result
|
|
|
|
|
|
@router.get("/kpi-analysis/{kpi_id}")
|
|
async def kpi_analysis(kpi_id: int, db: Session = Depends(get_db)):
|
|
"""AI分析单个KPI"""
|
|
# 尝试缓存
|
|
cache_key = f"kpi_analysis:{kpi_id}"
|
|
cached = cache_get("ai", cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
|
if not kpi:
|
|
raise HTTPException(404, "KPI不存在")
|
|
|
|
values = db.query(KPIValue).filter(KPIValue.kpi_id == kpi_id).order_by(KPIValue.period.asc()).all()
|
|
|
|
trend_data = []
|
|
for v in values:
|
|
trend_data.append({"period": v.period, "value": v.actual_value})
|
|
|
|
prompt = f"""请分析以下KPI指标：
|
|
|
|
KPI名称：{kpi.kpi_name}
|
|
维度：{kpi.dimension}
|
|
计算公式：{kpi.formula}
|
|
目标值：{kpi.target_value}
|
|
单位：{kpi.unit}
|
|
负责部门：{kpi.responsible_dept}
|
|
|
|
历史数据趋势：
|
|
{json.dumps(trend_data, ensure_ascii=False, indent=2)}
|
|
|
|
请分析：
|
|
1. 当前表现如何，是否达到目标
|
|
2. 趋势走势是否健康（上升/下降/波动）
|
|
3. 存在什么风险
|
|
4. 建议采取什么管理行动"""
|
|
|
|
try:
|
|
analysis = await _call_deepseek(prompt)
|
|
except Exception as e:
|
|
analysis = f"分析暂时不可用: {str(e)}"
|
|
|
|
result = {"kpi_name": kpi.kpi_name, "analysis": analysis}
|
|
cache_set("ai", cache_key, result, ttl_seconds=600)
|
|
return result
|
|
|
|
|
|
async def _stream_analysis(prompt: str):
|
|
"""流式调用DeepSeek并生成SSE事件"""
|
|
async with httpx.AsyncClient(timeout=60) as client:
|
|
async with client.stream(
|
|
"POST",
|
|
"https://api.deepseek.com/v1/chat/completions",
|
|
headers={
|
|
"Authorization": f"Bearer {os.getenv('DEEPSEEK_API_KEY', 'sk-8e24e6eb87f2475e96ea0980002dc2e8')}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
json={
|
|
"model": "deepseek-chat",
|
|
"messages": [
|
|
{"role": "system", "content": "你是一名CMA管理会计师，擅长用数据驱动的方式分析企业经营状况，给出专业的财务分析和管理建议。"},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
"stream": True,
|
|
"temperature": 0.3,
|
|
}
|
|
) as response:
|
|
async for line in response.aiter_lines():
|
|
if not line or line.startswith(":"):
|
|
continue
|
|
if line.startswith("data: "):
|
|
data_str = line[6:]
|
|
if data_str.strip() == "[DONE]":
|
|
break
|
|
try:
|
|
chunk = json.loads(data_str)
|
|
delta = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
|
|
if delta:
|
|
yield f"data: {json.dumps({'text': delta})}\n\n"
|
|
except json.JSONDecodeError:
|
|
continue
|
|
yield "data: {\"text\": \"[DONE]\"}\n\n"
|
|
|
|
|
|
@router.get("/dashboard-analysis-stream")
|
|
async def dashboard_analysis_stream(role: str = Query("ceo"), db: Session = Depends(get_db)):
|
|
"""AI分析驾驶舱数据 — SSE流式输出"""
|
|
cache_key = f"dashboard_analysis:{role}"
|
|
cached = cache_get("ai", cache_key)
|
|
if cached:
|
|
# 缓存存在，直接以流的形式一次性返回
|
|
full_text = cached.get("analysis", "")
|
|
async def cached_stream():
|
|
yield f"data: {json.dumps({'text': full_text})}\n\n"
|
|
yield "data: {\"text\": \"[DONE]\"}\n\n"
|
|
return StreamingResponse(cached_stream(), media_type="text/event-stream")
|
|
|
|
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
|
|
kpi_summary = []
|
|
for k in kpis:
|
|
latest = db.query(KPIValue).filter(KPIValue.kpi_id == k.id).order_by(KPIValue.period.desc()).first()
|
|
kpi_summary.append({
|
|
"name": k.kpi_name, "code": k.kpi_code, "dimension": k.dimension,
|
|
"target": k.target_value, "actual": latest.actual_value if latest else None,
|
|
"period": latest.period if latest else None, "unit": k.unit,
|
|
})
|
|
alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending").count()
|
|
kpi_text = "\n".join([f"- {k['name']}({k['code']}): 目标={k['target']}, 实际={k['actual']}({k['period']}), 维度={k['dimension']}" for k in kpi_summary if k['actual'] is not None])
|
|
role_label = {"ceo": "CEO（总经理）", "finance": "财务部", "business": "业务部"}.get(role, "管理层")
|
|
prompt = f"""我是一家公司的管理层，以下是当前管理会计系统的KPI数据和系统状态，请给出专业的分析和管理建议：
|
|
|
|
当前KPI数据：
|
|
{kpi_text}
|
|
|
|
待处理预警数：{alerts}
|
|
|
|
请从以下三个方面分析：
|
|
1. **核心发现**：当前数据反映的最关键问题是什么？
|
|
2. **深入解读**：从CMA管理会计角度，这些数据意味着什么？
|
|
3. **行动建议**：基于数据，财务和业务部门应该采取什么具体行动？
|
|
|
|
注意：角色视角为{role_label}。"""
|
|
|
|
return StreamingResponse(_stream_analysis(prompt), media_type="text/event-stream")
|
|
|
|
|
|
@router.post("/ask")
|
|
async def ask_question(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_auth),
|
|
):
|
|
"""自然语言查询 — CEO问企业经营问题"""
|
|
body = await request.json()
|
|
question = body.get("question", "").strip()
|
|
|
|
if not question:
|
|
raise HTTPException(400, "请输入问题")
|
|
|
|
# 收集系统数据作为上下文
|
|
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
|
|
kpi_context = []
|
|
for k in kpis:
|
|
latest = db.query(KPIValue).filter(KPIValue.kpi_id == k.id).order_by(KPIValue.period.desc()).first()
|
|
alert = db.query(KPIAlert).filter(KPIAlert.kpi_id == k.id, KPIAlert.status == "pending").first()
|
|
kpi_context.append(
|
|
f"{k.kpi_name}({k.kpi_code}): 当前值={latest.actual_value if latest else '无'}"
|
|
f"{' ⚠️' + alert.alert_level if alert else ''}"
|
|
)
|
|
|
|
# 获取改善计划
|
|
plans = db.query(ActionPlan).order_by(ActionPlan.created_at.desc()).limit(10).all()
|
|
plan_context = [f"- {p.title}({p.assignee}, {p.status}, {p.progress}%)" for p in plans]
|
|
|
|
# 获取预警
|
|
red_alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending", KPIAlert.alert_level == "red").count()
|
|
yellow_alerts = db.query(KPIAlert).filter(KPIAlert.status == "pending", KPIAlert.alert_level == "yellow").count()
|
|
|
|
system_context = f"""你是管理会计OS的AI助手，基于以下企业数据回答管理层问题。
|
|
|
|
时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}
|
|
当前用户：{current_user.name} ({current_user.role})
|
|
|
|
## KPI数据
|
|
{chr(10).join(kpi_context)}
|
|
|
|
## 预警概况
|
|
红色(紧急): {red_alerts}条 | 黄色(预警): {yellow_alerts}条
|
|
|
|
## 改善计划
|
|
{chr(10).join(plan_context) if plan_context else '暂无'}
|
|
|
|
请基于以上数据回答问题。如果问题需要具体数据但上下文中没有，可以根据KPI编码名称推断。回答要简洁、有数据支撑。"""
|
|
|
|
prompt = f"{system_context}\n\n用户问题：{question}"
|
|
|
|
try:
|
|
analysis = await _call_deepseek(prompt)
|
|
except Exception as e:
|
|
analysis = f"查询失败: {str(e)}"
|
|
|
|
return {"question": question, "answer": analysis, "timestamp": datetime.now().isoformat()}
|
|
|
|
|
|
@router.post("/review-plans")
|
|
async def review_plans(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_auth),
|
|
):
|
|
"""AI复盘改善行动计划执行效果"""
|
|
plans = db.query(ActionPlan).order_by(ActionPlan.created_at.asc()).all()
|
|
|
|
if not plans:
|
|
return {"analysis": "暂无改善行动计划，无法复盘"}
|
|
|
|
plan_text = []
|
|
for p in plans:
|
|
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == p.kpi_id).first()
|
|
kpi_name = kpi.kpi_name if kpi else "未知"
|
|
latest = db.query(KPIValue).filter(KPIValue.kpi_id == p.kpi_id).order_by(KPIValue.period.desc()).first()
|
|
plan_text.append(
|
|
f"- {p.title}\n"
|
|
f" 关联KPI: {kpi_name}(当前值: {latest.actual_value if latest else '无'})\n"
|
|
f" 负责人: {p.assignee} | 状态: {p.status} | 进度: {p.progress}%\n"
|
|
f" 描述: {p.description}\n"
|
|
f" 截止日: {p.due_date.strftime('%Y-%m-%d') if p.due_date else '无'}"
|
|
)
|
|
|
|
completed = sum(1 for p in plans if p.status == "completed")
|
|
in_progress = sum(1 for p in plans if p.status == "in_progress")
|
|
pending = sum(1 for p in plans if p.status == "pending")
|
|
|
|
prompt = f"""请复盘以下改善行动计划的执行情况：
|
|
|
|
## 改善计划概览
|
|
总数: {len(plans)} | 已完成: {completed} | 进行中: {in_progress} | 待开始: {pending}
|
|
|
|
## 各计划详情
|
|
{chr(10).join(plan_text)}
|
|
|
|
请分析：
|
|
1. **执行概况**：整体执行到位吗？哪些计划需要重点关注？
|
|
2. **效果评估**：已完成的计划是否真正改善了关联KPI？
|
|
3. **风险提示**：哪些计划存在延期或执行不力的风险？
|
|
4. **改进建议**：接下来应该调整或优先推进哪些计划？"""
|
|
|
|
try:
|
|
analysis = await _call_deepseek(prompt)
|
|
except Exception as e:
|
|
analysis = f"复盘失败: {str(e)}"
|
|
|
|
return {
|
|
"analysis": analysis,
|
|
"stats": {"total": len(plans), "completed": completed, "in_progress": in_progress, "pending": pending},
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|