432 lines
16 KiB
Python
432 lines
16 KiB
Python
"""Layer 1 — Intent Formalization Engine.
|
||
|
||
Responsible for:
|
||
1. Intent decomposition - breaking natural language into structured requirements
|
||
2. Requirement typing - classifying requirements (dimensional, load, environmental, process)
|
||
3. Load case enumeration - identifying operational scenarios
|
||
"""
|
||
|
||
import re
|
||
import uuid
|
||
from typing import Any
|
||
|
||
from fusionagi.maa.schemas.intent import EngineeringIntentGraph, IntentNode, LoadCase, RequirementType
|
||
from fusionagi._logger import logger
|
||
|
||
|
||
class IntentIncompleteError(Exception):
|
||
"""Raised when intent formalization cannot be completed due to missing information."""
|
||
|
||
def __init__(self, message: str, missing_fields: list[str] | None = None):
|
||
self.missing_fields = missing_fields or []
|
||
super().__init__(message)
|
||
|
||
|
||
class IntentEngine:
|
||
"""
|
||
Intent decomposition, requirement typing, and load case enumeration.
|
||
|
||
Features:
|
||
- Pattern-based requirement extraction from natural language
|
||
- Automatic requirement type classification
|
||
- Load case identification
|
||
- Environmental bounds extraction
|
||
- LLM-assisted formalization (optional)
|
||
"""
|
||
|
||
# Patterns for dimensional requirements (measurements, tolerances)
|
||
DIMENSIONAL_PATTERNS = [
|
||
r"(\d+(?:\.\d+)?)\s*(mm|cm|m|in|inch|inches|ft|feet)\b",
|
||
r"tolerance[s]?\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"±\s*(\d+(?:\.\d+)?)",
|
||
r"(\d+(?:\.\d+)?)\s*×\s*(\d+(?:\.\d+)?)",
|
||
r"diameter\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"radius\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"thickness\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"length\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"width\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"height\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
]
|
||
|
||
# Patterns for load requirements (forces, pressures, stresses)
|
||
LOAD_PATTERNS = [
|
||
r"(\d+(?:\.\d+)?)\s*(N|kN|MN|lb|lbf|kg|kgf)\b",
|
||
r"(\d+(?:\.\d+)?)\s*(MPa|GPa|Pa|psi|ksi)\b",
|
||
r"load\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"force\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"stress\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"pressure\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"factor\s*of\s*safety\s*(?:of\s*)?(\d+(?:\.\d+)?)",
|
||
r"yield\s*strength",
|
||
r"tensile\s*strength",
|
||
r"fatigue\s*(?:life|limit|strength)",
|
||
]
|
||
|
||
# Patterns for environmental requirements
|
||
ENVIRONMENTAL_PATTERNS = [
|
||
r"(\d+(?:\.\d+)?)\s*(?:°|deg|degrees?)?\s*(C|F|K|Celsius|Fahrenheit|Kelvin)\b",
|
||
r"temperature\s*(?:range|of)?\s*(\d+)",
|
||
r"humidity\s*(?:of\s*)?(\d+)",
|
||
r"corrosion\s*resist",
|
||
r"UV\s*resist",
|
||
r"water\s*(?:proof|resist)",
|
||
r"chemical\s*resist",
|
||
r"outdoor",
|
||
r"marine",
|
||
r"aerospace",
|
||
]
|
||
|
||
# Patterns for process requirements
|
||
PROCESS_PATTERNS = [
|
||
r"CNC|machining|milling|turning|drilling",
|
||
r"3D\s*print|additive|FDM|SLA|SLS|DMLS",
|
||
r"cast|injection\s*mold|die\s*cast",
|
||
r"weld|braze|solder",
|
||
r"heat\s*treat|anneal|harden|temper",
|
||
r"surface\s*finish|polish|anodize|plate",
|
||
r"assembly|sub-assembly",
|
||
r"material:\s*(\w+)",
|
||
r"aluminum|steel|titanium|plastic|composite",
|
||
]
|
||
|
||
# Load case indicator patterns
|
||
LOAD_CASE_PATTERNS = [
|
||
r"(?:during|under|in)\s+(\w+(?:\s+\w+)?)\s+(?:conditions?|operation|mode)",
|
||
r"(\w+)\s+load\s+case",
|
||
r"(?:static|dynamic|cyclic|impact|thermal)\s+load",
|
||
r"(?:normal|extreme|emergency|failure)\s+(?:operation|conditions?|mode)",
|
||
r"operating\s+(?:at|under|in)",
|
||
]
|
||
|
||
def __init__(self, llm_adapter: Any | None = None):
|
||
"""
|
||
Initialize the IntentEngine.
|
||
|
||
Args:
|
||
llm_adapter: Optional LLM adapter for enhanced natural language processing.
|
||
"""
|
||
self._llm = llm_adapter
|
||
|
||
def formalize(
|
||
self,
|
||
intent_id: str,
|
||
natural_language: str | None = None,
|
||
file_refs: list[str] | None = None,
|
||
metadata: dict[str, Any] | None = None,
|
||
use_llm: bool = True,
|
||
) -> EngineeringIntentGraph:
|
||
"""
|
||
Formalize engineering intent from natural language and file references.
|
||
|
||
Args:
|
||
intent_id: Unique identifier for this intent.
|
||
natural_language: Natural language description of requirements.
|
||
file_refs: References to CAD files, specifications, etc.
|
||
metadata: Additional metadata.
|
||
use_llm: Whether to use LLM for enhanced processing (if available).
|
||
|
||
Returns:
|
||
EngineeringIntentGraph with extracted requirements.
|
||
|
||
Raises:
|
||
IntentIncompleteError: If required information is missing.
|
||
"""
|
||
if not intent_id:
|
||
raise IntentIncompleteError("intent_id required", ["intent_id"])
|
||
|
||
if not natural_language and not file_refs:
|
||
raise IntentIncompleteError(
|
||
"At least one of natural_language or file_refs required",
|
||
["natural_language", "file_refs"],
|
||
)
|
||
|
||
nodes: list[IntentNode] = []
|
||
load_cases: list[LoadCase] = []
|
||
environmental_bounds: dict[str, Any] = {}
|
||
|
||
# Process natural language if provided
|
||
if natural_language:
|
||
# Use LLM if available and requested
|
||
if use_llm and self._llm:
|
||
llm_result = self._formalize_with_llm(intent_id, natural_language)
|
||
if llm_result:
|
||
return llm_result
|
||
|
||
# Fall back to pattern-based extraction
|
||
extracted = self._extract_requirements(intent_id, natural_language)
|
||
nodes.extend(extracted["nodes"])
|
||
load_cases.extend(extracted["load_cases"])
|
||
environmental_bounds.update(extracted["environmental_bounds"])
|
||
|
||
# Process file references
|
||
if file_refs:
|
||
for ref in file_refs:
|
||
nodes.append(
|
||
IntentNode(
|
||
node_id=f"{intent_id}_file_{uuid.uuid4().hex[:8]}",
|
||
requirement_type=RequirementType.OTHER,
|
||
description=f"Reference: {ref}",
|
||
metadata={"file_ref": ref},
|
||
)
|
||
)
|
||
|
||
# If no nodes were extracted, create a general requirement
|
||
if not nodes and natural_language:
|
||
nodes.append(
|
||
IntentNode(
|
||
node_id=f"{intent_id}_general_0",
|
||
requirement_type=RequirementType.OTHER,
|
||
description=natural_language[:500],
|
||
)
|
||
)
|
||
|
||
logger.info(
|
||
"Intent formalized",
|
||
extra={
|
||
"intent_id": intent_id,
|
||
"num_nodes": len(nodes),
|
||
"num_load_cases": len(load_cases),
|
||
},
|
||
)
|
||
|
||
return EngineeringIntentGraph(
|
||
intent_id=intent_id,
|
||
nodes=nodes,
|
||
load_cases=load_cases,
|
||
environmental_bounds=environmental_bounds,
|
||
metadata=metadata or {},
|
||
)
|
||
|
||
def _extract_requirements(
|
||
self,
|
||
intent_id: str,
|
||
text: str,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
Extract requirements from text using pattern matching.
|
||
|
||
Returns dict with nodes, load_cases, and environmental_bounds.
|
||
"""
|
||
nodes: list[IntentNode] = []
|
||
load_cases: list[LoadCase] = []
|
||
environmental_bounds: dict[str, Any] = {}
|
||
|
||
# Split into sentences for processing
|
||
sentences = re.split(r'[.!?]+', text)
|
||
|
||
node_counter = 0
|
||
load_case_counter = 0
|
||
|
||
for sentence in sentences:
|
||
sentence = sentence.strip()
|
||
if not sentence:
|
||
continue
|
||
|
||
# Check for dimensional requirements
|
||
for pattern in self.DIMENSIONAL_PATTERNS:
|
||
if re.search(pattern, sentence, re.IGNORECASE):
|
||
nodes.append(
|
||
IntentNode(
|
||
node_id=f"{intent_id}_dim_{node_counter}",
|
||
requirement_type=RequirementType.DIMENSIONAL,
|
||
description=sentence,
|
||
metadata={"pattern": "dimensional"},
|
||
)
|
||
)
|
||
node_counter += 1
|
||
break
|
||
|
||
# Check for load requirements
|
||
for pattern in self.LOAD_PATTERNS:
|
||
if re.search(pattern, sentence, re.IGNORECASE):
|
||
nodes.append(
|
||
IntentNode(
|
||
node_id=f"{intent_id}_load_{node_counter}",
|
||
requirement_type=RequirementType.LOAD,
|
||
description=sentence,
|
||
metadata={"pattern": "load"},
|
||
)
|
||
)
|
||
node_counter += 1
|
||
break
|
||
|
||
# Check for environmental requirements
|
||
for pattern in self.ENVIRONMENTAL_PATTERNS:
|
||
match = re.search(pattern, sentence, re.IGNORECASE)
|
||
if match:
|
||
nodes.append(
|
||
IntentNode(
|
||
node_id=f"{intent_id}_env_{node_counter}",
|
||
requirement_type=RequirementType.ENVIRONMENTAL,
|
||
description=sentence,
|
||
metadata={"pattern": "environmental"},
|
||
)
|
||
)
|
||
node_counter += 1
|
||
|
||
# Extract specific bounds if possible
|
||
if "temperature" in sentence.lower():
|
||
temp_match = re.search(r"(-?\d+(?:\.\d+)?)", sentence)
|
||
if temp_match:
|
||
environmental_bounds["temperature"] = float(temp_match.group(1))
|
||
break
|
||
|
||
# Check for process requirements
|
||
for pattern in self.PROCESS_PATTERNS:
|
||
if re.search(pattern, sentence, re.IGNORECASE):
|
||
nodes.append(
|
||
IntentNode(
|
||
node_id=f"{intent_id}_proc_{node_counter}",
|
||
requirement_type=RequirementType.PROCESS,
|
||
description=sentence,
|
||
metadata={"pattern": "process"},
|
||
)
|
||
)
|
||
node_counter += 1
|
||
break
|
||
|
||
# Check for load cases
|
||
for pattern in self.LOAD_CASE_PATTERNS:
|
||
match = re.search(pattern, sentence, re.IGNORECASE)
|
||
if match:
|
||
load_case_desc = match.group(0) if match.group(0) else sentence
|
||
load_cases.append(
|
||
LoadCase(
|
||
load_case_id=f"{intent_id}_lc_{load_case_counter}",
|
||
description=load_case_desc,
|
||
metadata={"source_sentence": sentence},
|
||
)
|
||
)
|
||
load_case_counter += 1
|
||
break
|
||
|
||
return {
|
||
"nodes": nodes,
|
||
"load_cases": load_cases,
|
||
"environmental_bounds": environmental_bounds,
|
||
}
|
||
|
||
def _formalize_with_llm(
|
||
self,
|
||
intent_id: str,
|
||
natural_language: str,
|
||
) -> EngineeringIntentGraph | None:
|
||
"""
|
||
Use LLM to extract structured requirements from natural language.
|
||
|
||
Returns None if LLM processing fails (falls back to pattern matching).
|
||
"""
|
||
if not self._llm:
|
||
return None
|
||
|
||
import json
|
||
|
||
prompt = f"""Extract engineering requirements from the following text.
|
||
Return a JSON object with:
|
||
- "nodes": list of requirements, each with:
|
||
- "requirement_type": one of "dimensional", "load", "environmental", "process", "other"
|
||
- "description": the requirement text
|
||
- "load_cases": list of operational scenarios, each with:
|
||
- "description": the scenario description
|
||
- "environmental_bounds": dict of environmental limits (e.g., {{"temperature_max": 85, "humidity_max": 95}})
|
||
|
||
Text: {natural_language[:2000]}
|
||
|
||
Return only valid JSON, no markdown."""
|
||
|
||
try:
|
||
messages = [
|
||
{"role": "system", "content": "You are an engineering requirements extraction system."},
|
||
{"role": "user", "content": prompt},
|
||
]
|
||
|
||
# Try structured output if available
|
||
if hasattr(self._llm, "complete_structured"):
|
||
result = self._llm.complete_structured(messages)
|
||
if result:
|
||
return self._parse_llm_result(intent_id, result)
|
||
|
||
# Fall back to text completion
|
||
raw = self._llm.complete(messages)
|
||
if raw:
|
||
# Clean up response
|
||
if raw.startswith("```"):
|
||
raw = raw.split("```")[1]
|
||
if raw.startswith("json"):
|
||
raw = raw[4:]
|
||
result = json.loads(raw)
|
||
return self._parse_llm_result(intent_id, result)
|
||
|
||
except Exception as e:
|
||
logger.warning(f"LLM formalization failed: {e}")
|
||
|
||
return None
|
||
|
||
def _parse_llm_result(
|
||
self,
|
||
intent_id: str,
|
||
result: dict[str, Any],
|
||
) -> EngineeringIntentGraph:
|
||
"""Parse LLM result into EngineeringIntentGraph."""
|
||
nodes = []
|
||
for i, node_data in enumerate(result.get("nodes", [])):
|
||
req_type_str = node_data.get("requirement_type", "other")
|
||
try:
|
||
req_type = RequirementType(req_type_str)
|
||
except ValueError:
|
||
req_type = RequirementType.OTHER
|
||
|
||
nodes.append(
|
||
IntentNode(
|
||
node_id=f"{intent_id}_llm_{i}",
|
||
requirement_type=req_type,
|
||
description=node_data.get("description", ""),
|
||
metadata={"source": "llm"},
|
||
)
|
||
)
|
||
|
||
load_cases = []
|
||
for i, lc_data in enumerate(result.get("load_cases", [])):
|
||
load_cases.append(
|
||
LoadCase(
|
||
load_case_id=f"{intent_id}_lc_llm_{i}",
|
||
description=lc_data.get("description", ""),
|
||
metadata={"source": "llm"},
|
||
)
|
||
)
|
||
|
||
environmental_bounds = result.get("environmental_bounds", {})
|
||
|
||
return EngineeringIntentGraph(
|
||
intent_id=intent_id,
|
||
nodes=nodes,
|
||
load_cases=load_cases,
|
||
environmental_bounds=environmental_bounds,
|
||
metadata={"formalization_source": "llm"},
|
||
)
|
||
|
||
def validate_completeness(self, graph: EngineeringIntentGraph) -> tuple[bool, list[str]]:
|
||
"""
|
||
Validate that an intent graph has sufficient information.
|
||
|
||
Returns:
|
||
Tuple of (is_complete, list_of_missing_items)
|
||
"""
|
||
missing = []
|
||
|
||
if not graph.nodes:
|
||
missing.append("No requirements extracted")
|
||
|
||
# Check for at least one dimensional or load requirement for manufacturing
|
||
has_dimensional = any(n.requirement_type == RequirementType.DIMENSIONAL for n in graph.nodes)
|
||
has_load = any(n.requirement_type == RequirementType.LOAD for n in graph.nodes)
|
||
|
||
if not has_dimensional:
|
||
missing.append("No dimensional requirements specified")
|
||
|
||
# Load cases are recommended but not required
|
||
if not graph.load_cases:
|
||
logger.info("No load cases specified for intent", extra={"intent_id": graph.intent_id})
|
||
|
||
return len(missing) == 0, missing
|