468 lines
18 KiB
Python
468 lines
18 KiB
Python
"""
|
||
LLM 增强分析服务
|
||
|
||
提供两个核心能力：
|
||
1. generate_design_report — 将分析 JSON 转换为结构化评审报告
|
||
2. recommend_parting_direction — 基于几何 + 制造约束推荐最优分型方向
|
||
|
||
适配层：OpenAI 兼容 API（支持 OpenAI / DeepSeek / vLLM / Ollama 等）
|
||
未配置 LLM 时静默降级，不影响主流程。
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
from typing import Optional, Dict, Any, List
|
||
|
||
import httpx
|
||
|
||
from config.settings import settings
|
||
from utils.logger import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
_DESIGN_REPORT_SYSTEM = """你是一位资深注塑模具设计工程师，拥有 20 年模具 DFM 评审经验。
|
||
请根据提供的模具分析数据，生成一份专业的模具设计评审报告。
|
||
|
||
要求：
|
||
1. 使用中文
|
||
2. 按 "关键问题 → 工艺参数建议 → 改进建议" 结构组织
|
||
3. 技术术语准确（如：锁模力、投影面积、分型面、滑块、斜顶、拔模角、缩痕、熔接痕）
|
||
4. 每个问题标注优先级（high / medium / low）
|
||
5. 如果数据不足以判断某项，明确标注"数据不足，需人工确认"
|
||
|
||
严格输出 JSON，不要输出其他内容。JSON 格式：
|
||
{
|
||
"title": "模具设计评审报告",
|
||
"overview": "一段 1-2 句话的整体概述",
|
||
"sections": [
|
||
{
|
||
"heading": "关键问题",
|
||
"type": "issues",
|
||
"items": [
|
||
{"level": "high", "content": "拔模角不足，建议增加到 2° 以上"},
|
||
{"level": "medium", "content": "壁厚偏差较大，可能产生缩痕"}
|
||
]
|
||
},
|
||
{
|
||
"heading": "工艺参数建议",
|
||
"type": "params_table",
|
||
"headers": ["参数", "推荐值", "说明"],
|
||
"rows": [
|
||
["锁模力", "150 吨", "基于投影面积计算"],
|
||
["注塑温度", "230°C", "ABS 材料推荐值"]
|
||
]
|
||
},
|
||
{
|
||
"heading": "改进建议",
|
||
"type": "recommendations",
|
||
"items": [
|
||
"建议将主流道直径从 4mm 增加到 6mm",
|
||
"建议在所有垂直面增加 1-2° 拔模角"
|
||
]
|
||
}
|
||
],
|
||
"overall_score": 7.5
|
||
}"""
|
||
|
||
_DESIGN_REPORT_USER = """请根据以下模具分析数据生成评审报告：
|
||
|
||
## 产品信息
|
||
- 文件：{filename}
|
||
- 材料：{material}
|
||
- 体积：{volume}
|
||
- 表面积：{surface_area}
|
||
- 边界框：{bbox}
|
||
|
||
## 检测特征
|
||
{features}
|
||
|
||
## 质量指标
|
||
{quality_metrics}
|
||
|
||
## 分模方案
|
||
{schemes}
|
||
|
||
## 制造参数
|
||
- 推荐模具材料：{mold_material}
|
||
- 推荐模具硬度：{mold_hardness}
|
||
- 预估锁模力：{clamping_force}
|
||
- 模具尺寸（长×宽×高）：{mold_size}
|
||
- 预估成型周期：{cycle_time}
|
||
- 拔模角：{draft_angle}
|
||
- 收缩率：{shrinkage_rate}
|
||
|
||
## 原始设计建议
|
||
{recommendations}
|
||
|
||
请生成 JSON 格式评审报告。issues 部分不要超过 8 条，每条内容简洁在一行内；
|
||
params_table 至少要包含锁模力、成型周期、模仁材料、推荐型腔数 4 行；
|
||
如果某项数据标记为"自动计算"或"自动选择"，请在说明中注明"需人工确认"；
|
||
overall_score 范围 1-10。"""
|
||
|
||
_SIDE_ACTION_ANALYSIS_SYSTEM = """你是一位资深注塑模具结构工程师。
|
||
请根据提供的 STP 分析结果，判断当前产品是否需要倒扣/抽芯机构，并输出标准化结论。
|
||
|
||
输出要求：
|
||
1. 严格输出 JSON，不要输出其他内容
|
||
2. 只允许基于已给数据判断，数据不足时必须标记为 manual_review
|
||
3. 结论面向工程评审，避免坐标、面索引、底层算法术语堆砌
|
||
4. 建议必须标准化、简洁、可执行
|
||
|
||
JSON 格式：
|
||
{
|
||
"status": "required|not_required|manual_review",
|
||
"confidence": 0.0,
|
||
"conclusion": "一句中文结论",
|
||
"mechanism_recommendation": "slider|lifter|mixed|none|manual_review",
|
||
"summary": "一段 40-80 字中文摘要",
|
||
"reasons": ["原因1", "原因2"],
|
||
"standard_advice": ["建议1", "建议2"],
|
||
"manual_review_items": ["复核项1", "复核项2"]
|
||
}"""
|
||
|
||
_SIDE_ACTION_ANALYSIS_USER = """请分析当前注塑件是否需要倒扣/抽芯机构：
|
||
|
||
## 产品信息
|
||
- 文件：{filename}
|
||
- 材料：{material}
|
||
- 边界框：{bbox}
|
||
|
||
## 特征检测
|
||
{features}
|
||
|
||
## 最优方案
|
||
{best_scheme}
|
||
|
||
## 规则分析结果
|
||
{side_actions}
|
||
|
||
## DFM 风险
|
||
{dfm_violations}
|
||
|
||
判断要求：
|
||
1. 如果规则结果明确显示无倒扣，可输出 not_required
|
||
2. 如果存在外侧倒扣，优先考虑 slider
|
||
3. 如果存在内侧倒扣，优先考虑 lifter
|
||
4. 如果内外侧倒扣同时存在，可输出 mixed
|
||
5. 如果数据不够支撑明确判断，输出 manual_review"""
|
||
|
||
_PARTING_SYSTEM = """你是一位注塑模具分模专家。
|
||
根据产品几何特征和多个候选分模方向的评分数据，推荐最优分模方向。
|
||
|
||
输出要求：严格输出 JSON，不要输出其他内容。
|
||
JSON 格式：
|
||
{
|
||
"recommended_axis": "Z",
|
||
"confidence": 0.85,
|
||
"reasoning": "详细的中文推理过程...",
|
||
"risk_notes": ["风险1", "风险2"],
|
||
"rankings": [{"axis":"Z","rank":1,"score":92,"note":"..."}]
|
||
}"""
|
||
|
||
_PARTING_USER = """请评估以下候选分模方向并推荐最优方案：
|
||
|
||
产品几何：
|
||
- 边界框 (mm)：{bbox}
|
||
- 面法向分布：{normal_stats}
|
||
- 惯性矩：{inertia}
|
||
|
||
约束条件：
|
||
- 材料：{material}
|
||
- 型腔数：{cavity_count}
|
||
- 最大锁模力 (吨)：{max_clamping_force}
|
||
- 泡沫材料：{is_foam}
|
||
|
||
候选方案：
|
||
{schemes}
|
||
|
||
请综合评估制造可行性、成本和风险，给出推荐。"""
|
||
|
||
|
||
class LLMService:
|
||
"""LLM 增强分析服务（单例）"""
|
||
|
||
def __init__(self):
|
||
self._enabled = settings.LLM_ENABLED
|
||
self._api_url = settings.LLM_API_URL.rstrip("/")
|
||
self._api_key = settings.LLM_API_KEY
|
||
self._model = settings.LLM_MODEL
|
||
self._timeout = settings.LLM_TIMEOUT
|
||
self._max_tokens = settings.LLM_MAX_TOKENS
|
||
|
||
if self._enabled:
|
||
logger.info(
|
||
"LLM 增强分析已启用: model=%s endpoint=%s",
|
||
self._model, self._api_url,
|
||
)
|
||
else:
|
||
logger.info("LLM 增强分析未启用（设置 LLM_ENABLED=true 启用）")
|
||
|
||
async def generate_design_report(
|
||
self,
|
||
analysis_result: Dict[str, Any],
|
||
detailed_cavity_json: Optional[Dict[str, Any]] = None,
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""生成模具设计评审报告 (结构化 JSON)"""
|
||
if not self._enabled:
|
||
return None
|
||
|
||
try:
|
||
prompt = self._build_design_report_prompt(analysis_result, detailed_cavity_json)
|
||
response = await self._chat(
|
||
_DESIGN_REPORT_SYSTEM,
|
||
prompt,
|
||
self._max_tokens,
|
||
expect_json=True,
|
||
)
|
||
if not response:
|
||
return None
|
||
result = self._parse_json_response(response)
|
||
if result:
|
||
logger.info("LLM 设计报告生成成功 (%d sections)", len(result.get("sections", [])))
|
||
return result
|
||
except Exception as e:
|
||
logger.warning("LLM 设计报告生成失败（不影响主流程）: %s", e)
|
||
return None
|
||
|
||
async def generate_side_action_analysis(
|
||
self,
|
||
analysis_result: Dict[str, Any],
|
||
detailed_cavity_json: Optional[Dict[str, Any]] = None,
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""生成倒扣/抽芯 AI 标准化分析"""
|
||
if not self._enabled:
|
||
return None
|
||
|
||
try:
|
||
prompt = self._build_side_action_prompt(analysis_result, detailed_cavity_json)
|
||
response = await self._chat(
|
||
_SIDE_ACTION_ANALYSIS_SYSTEM,
|
||
prompt,
|
||
min(self._max_tokens, 1200),
|
||
expect_json=True,
|
||
)
|
||
if not response:
|
||
return None
|
||
result = self._parse_json_response(response)
|
||
if result:
|
||
logger.info(
|
||
"LLM 倒扣/抽芯分析生成成功: status=%s confidence=%s",
|
||
result.get("status"),
|
||
result.get("confidence"),
|
||
)
|
||
return result
|
||
except Exception as e:
|
||
logger.warning("LLM 倒扣/抽芯分析失败（不影响主流程）: %s", e)
|
||
return None
|
||
|
||
@staticmethod
|
||
def compose_llm_report(
|
||
design_report: Optional[Dict[str, Any]],
|
||
side_action_analysis: Optional[Dict[str, Any]],
|
||
) -> Optional[str]:
|
||
"""将结构化报告和倒扣分析打包进 llm_report 字段，避免改动外部协议。
|
||
|
||
设计报告以 <!--DESIGN_REPORT_BEGIN--> / <!--DESIGN_REPORT_END--> 包裹的 JSON 嵌入，
|
||
倒扣分析以 <!--SIDE_ACTION_AI_BEGIN--> / <!--SIDE_ACTION_AI_END--> 包裹的 JSON 嵌入。
|
||
"""
|
||
sections: List[str] = []
|
||
if side_action_analysis:
|
||
payload = json.dumps(side_action_analysis, ensure_ascii=False)
|
||
sections.append(
|
||
"<!--SIDE_ACTION_AI_BEGIN-->\n"
|
||
f"{payload}\n"
|
||
"<!--SIDE_ACTION_AI_END-->"
|
||
)
|
||
if design_report:
|
||
payload = json.dumps(design_report, ensure_ascii=False)
|
||
sections.append(
|
||
"<!--DESIGN_REPORT_BEGIN-->\n"
|
||
f"{payload}\n"
|
||
"<!--DESIGN_REPORT_END-->"
|
||
)
|
||
merged = "\n\n".join(sections).strip()
|
||
return merged or None
|
||
|
||
async def recommend_parting_direction(
|
||
self,
|
||
geometry_data: Dict[str, Any],
|
||
candidate_schemes: List[Dict[str, Any]],
|
||
material: Dict[str, Any],
|
||
cavity_count: int = 1,
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""推荐最优分型方向"""
|
||
if not self._enabled:
|
||
return None
|
||
|
||
try:
|
||
prompt = self._build_parting_prompt(geometry_data, candidate_schemes, material, cavity_count)
|
||
response = await self._chat(_PARTING_SYSTEM, prompt, min(self._max_tokens, 1200), expect_json=True)
|
||
if response:
|
||
result = self._parse_json_response(response)
|
||
if result:
|
||
logger.info("LLM 分型推荐: %s (%.2f)", result.get("recommended_axis", "?"), result.get("confidence", 0))
|
||
return result
|
||
return None
|
||
except Exception as e:
|
||
logger.warning("LLM 分型推荐失败（不影响主流程）: %s", e)
|
||
return None
|
||
|
||
def _build_design_report_prompt(self, analysis_result, detailed_cavity_json) -> str:
|
||
features = json.dumps(analysis_result.get("detected_features", []), ensure_ascii=False, indent=2)
|
||
if len(features) > 4000:
|
||
features = features[:4000] + "\n... (已截断)"
|
||
|
||
schemes_text = ""
|
||
if detailed_cavity_json:
|
||
schemes = detailed_cavity_json.get("candidate_schemes", [])
|
||
if schemes:
|
||
schemes_text = json.dumps([{
|
||
"scheme_id": s.get("scheme_id"), "rank": s.get("rank"), "title": s.get("title"),
|
||
"score": s.get("score"), "summary": s.get("summary"),
|
||
"parting_axis": s.get("parting", {}).get("axis"),
|
||
"mold_structure_type": s.get("mold_structure_type"),
|
||
"dfm_violations": s.get("dfm_violations", []),
|
||
} for s in schemes], ensure_ascii=False, indent=2)
|
||
|
||
best = detailed_cavity_json.get("candidate_schemes", [{}])[0] if detailed_cavity_json else {}
|
||
cd = best.get("cavity_data", {}) if isinstance(best, dict) else {}
|
||
mfg = cd.get("manufacturing_info", {})
|
||
meta = cd.get("metadata", {})
|
||
|
||
return _DESIGN_REPORT_USER.format(
|
||
filename=meta.get("file_name", "unknown.stp"),
|
||
material=meta.get("selected_material", "ABS"),
|
||
volume=f"{analysis_result.get('geometry_data', {}).get('volume', 0):.1f} mm³",
|
||
surface_area=f"{analysis_result.get('geometry_data', {}).get('surface_area', 0):.1f} mm²",
|
||
bbox=json.dumps(analysis_result.get("geometry_data", {}).get("bounding_box", {}), ensure_ascii=False),
|
||
features=features or "无特征检测数据",
|
||
quality_metrics=json.dumps(analysis_result.get("quality_metrics", {}), ensure_ascii=False, indent=2),
|
||
schemes=schemes_text or "无分模方案数据",
|
||
mold_material=mfg.get("mold_material", "自动选择"),
|
||
mold_hardness=mfg.get("mold_hardness", "自动选择"),
|
||
clamping_force=mfg.get("estimated_clamping_force", "自动计算"),
|
||
mold_size=json.dumps(mfg.get("estimated_mold_size", {}), ensure_ascii=False),
|
||
cycle_time=mfg.get("estimated_cycle_time", "自动计算"),
|
||
draft_angle=f"{meta.get('draft_angle', 2.0)}°",
|
||
shrinkage_rate="自动计算",
|
||
recommendations=json.dumps(analysis_result.get("design_recommendations", []), ensure_ascii=False, indent=2) or "无",
|
||
)
|
||
|
||
def _build_side_action_prompt(self, analysis_result, detailed_cavity_json) -> str:
|
||
features = json.dumps(
|
||
analysis_result.get("detected_features", []),
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
if len(features) > 2500:
|
||
features = features[:2500] + "\n... (已截断)"
|
||
|
||
best_scheme = {}
|
||
if detailed_cavity_json:
|
||
candidate_schemes = detailed_cavity_json.get("candidate_schemes", [])
|
||
best_scheme_id = detailed_cavity_json.get("best_scheme_id")
|
||
if candidate_schemes:
|
||
best_scheme = candidate_schemes[0]
|
||
if best_scheme_id:
|
||
for scheme in candidate_schemes:
|
||
if scheme.get("scheme_id") == best_scheme_id:
|
||
best_scheme = scheme
|
||
break
|
||
|
||
cavity_data = best_scheme.get("cavity_data", {}) if isinstance(best_scheme, dict) else {}
|
||
metadata = cavity_data.get("metadata", {}) if isinstance(cavity_data, dict) else {}
|
||
side_actions = (
|
||
best_scheme.get("side_actions")
|
||
or cavity_data.get("side_actions")
|
||
or {}
|
||
)
|
||
best_scheme_view = {
|
||
"scheme_id": best_scheme.get("scheme_id"),
|
||
"title": best_scheme.get("title"),
|
||
"score": best_scheme.get("score"),
|
||
"parting_axis": best_scheme.get("parting", {}).get("axis"),
|
||
"mold_structure_type": best_scheme.get("mold_structure_type"),
|
||
"undercut_regions_count": len(best_scheme.get("undercut_regions", []) or []),
|
||
}
|
||
side_actions_view = {
|
||
"summary": side_actions.get("summary", {}),
|
||
"recommendations": side_actions.get("recommendations", []),
|
||
"slider_count": len(side_actions.get("slider_mechanisms", []) or []),
|
||
"lifter_count": len(side_actions.get("lifter_mechanisms", []) or []),
|
||
}
|
||
dfm_violations = best_scheme.get("dfm_violations", []) if isinstance(best_scheme, dict) else []
|
||
|
||
return _SIDE_ACTION_ANALYSIS_USER.format(
|
||
filename=metadata.get("file_name", "unknown.stp"),
|
||
material=metadata.get("selected_material", "ABS"),
|
||
bbox=json.dumps(
|
||
analysis_result.get("geometry_data", {}).get("bounding_box", {}),
|
||
ensure_ascii=False,
|
||
),
|
||
features=features or "无特征检测数据",
|
||
best_scheme=json.dumps(best_scheme_view, ensure_ascii=False, indent=2),
|
||
side_actions=json.dumps(side_actions_view, ensure_ascii=False, indent=2),
|
||
dfm_violations=json.dumps(dfm_violations[:6], ensure_ascii=False, indent=2),
|
||
)
|
||
|
||
def _build_parting_prompt(self, geometry_data, candidate_schemes, material, cavity_count) -> str:
|
||
bbox = geometry_data.get("bounding_box", {})
|
||
axis_normal_stats = geometry_data.get("axis_normal_stats", {})
|
||
inertia = geometry_data.get("inertia_matrix", [])
|
||
inertia_diag = [inertia[i][i] if i < len(inertia) and i < len(inertia[i]) else 0.0 for i in range(3)]
|
||
|
||
schemes_text = json.dumps([{
|
||
"axis": s.get("parting", {}).get("axis") or s.get("axis"),
|
||
"score": s.get("score"), "summary": s.get("summary"),
|
||
"mold_structure_type": s.get("mold_structure_type"),
|
||
"core_required": s.get("core_required"),
|
||
"dfm_violations": s.get("dfm_violations", []),
|
||
"undercut_regions_count": len(s.get("undercut_regions", [])),
|
||
"score_breakdown": s.get("score_breakdown", {}),
|
||
} for s in candidate_schemes], ensure_ascii=False, indent=2)
|
||
|
||
return _PARTING_USER.format(
|
||
bbox=json.dumps(bbox, ensure_ascii=False),
|
||
normal_stats=json.dumps(axis_normal_stats, ensure_ascii=False),
|
||
inertia=json.dumps(inertia_diag, ensure_ascii=False),
|
||
material=material.get("name", "ABS"),
|
||
cavity_count=cavity_count,
|
||
max_clamping_force="3000 吨（最大）",
|
||
is_foam="是" if material.get("is_foam") else "否",
|
||
schemes=schemes_text,
|
||
)
|
||
|
||
async def _chat(self, system, user, max_tokens=2000, expect_json=False, temperature=0.3):
|
||
url = f"{self._api_url}/chat/completions"
|
||
headers = {"Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json"}
|
||
payload = {
|
||
"model": self._model,
|
||
"messages": [{"role": "system", "content": system}, {"role": "user", "content": user}],
|
||
"max_tokens": max_tokens, "temperature": temperature,
|
||
}
|
||
if expect_json:
|
||
payload["response_format"] = {"type": "json_object"}
|
||
async with httpx.AsyncClient(timeout=self._timeout) as client:
|
||
resp = await client.post(url, json=payload, headers=headers)
|
||
resp.raise_for_status()
|
||
content = resp.json()["choices"][0]["message"]["content"]
|
||
return content.strip() if content else None
|
||
|
||
@staticmethod
|
||
def _parse_json_response(raw):
|
||
try:
|
||
return json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
m = re.search(r"\{[\s\S]*\}", raw)
|
||
if m:
|
||
try:
|
||
return json.loads(m.group())
|
||
except json.JSONDecodeError:
|
||
pass
|
||
logger.warning("LLM JSON 解析失败: %s...", raw[:200])
|
||
return None
|
||
|
||
|
||
llm_service = LLMService()
|