x
This commit is contained in:
+9
-1
@@ -1119,7 +1119,15 @@ async def process_file_core(
|
||||
# 9.8 LLM 增强分析
|
||||
llm_report = None
|
||||
if analysis_result:
|
||||
llm_report = await llm_service.generate_design_report(analysis_result, detailed_cavity_json)
|
||||
side_action_ai = await llm_service.generate_side_action_analysis(
|
||||
analysis_result, detailed_cavity_json
|
||||
)
|
||||
design_report = await llm_service.generate_design_report(
|
||||
analysis_result, detailed_cavity_json
|
||||
)
|
||||
llm_report = llm_service.compose_llm_report(
|
||||
design_report, side_action_ai
|
||||
)
|
||||
|
||||
# 10. 完成处理
|
||||
await storage_service.update_stp_file_status(db_session, stp_file_id, "completed")
|
||||
|
||||
@@ -64,6 +64,53 @@ _DESIGN_REPORT_USER = """请根据以下模具分析数据生成评审报告：
|
||||
## 原始设计建议
|
||||
{recommendations}"""
|
||||
|
||||
_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 = """你是一位注塑模具分模专家。
|
||||
根据产品几何特征和多个候选分模方向的评分数据，推荐最优分模方向。
|
||||
|
||||
@@ -134,6 +181,56 @@ class LLMService:
|
||||
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[str],
|
||||
side_action_analysis: Optional[Dict[str, Any]],
|
||||
) -> Optional[str]:
|
||||
"""将结构化倒扣分析打包进既有 llm_report 字段，避免改动外部协议。"""
|
||||
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 and design_report.strip():
|
||||
sections.append(design_report.strip())
|
||||
merged = "\n\n".join(sections).strip()
|
||||
return merged or None
|
||||
|
||||
async def recommend_parting_direction(
|
||||
self,
|
||||
geometry_data: Dict[str, Any],
|
||||
@@ -199,6 +296,63 @@ class LLMService:
|
||||
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", {})
|
||||
|
||||
@@ -291,7 +291,15 @@ class ProcessingService:
|
||||
llm_report = None
|
||||
stage_started = time.perf_counter()
|
||||
if analysis_result:
|
||||
llm_report = await llm_service.generate_design_report(analysis_result, detailed_cavity_json)
|
||||
side_action_ai = await llm_service.generate_side_action_analysis(
|
||||
analysis_result, detailed_cavity_json
|
||||
)
|
||||
design_report = await llm_service.generate_design_report(
|
||||
analysis_result, detailed_cavity_json
|
||||
)
|
||||
llm_report = llm_service.compose_llm_report(
|
||||
design_report, side_action_ai
|
||||
)
|
||||
stage_timings["generate_llm_report"] = round(time.perf_counter() - stage_started, 3)
|
||||
|
||||
# 10. 完成处理
|
||||
|
||||
Reference in New Issue
Block a user