AI大模型学习（21）：Agent框架在工业自动化中的应用——从缺陷检测到自动调参的智能体闭环

┌─────────────────────────────────────────────────────────────────┐
│                        工业AI Agent 系统                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────┐         │
│  │  感知层      │───▶│  认知层      │───▶│  决策层      │         │
│  │  YOLO26检测  │    │  VLM根因分析 │    │  Agent规划   │         │
│  │  传感器采集   │    │  工艺知识库  │    │  参数优化    │         │
│  └─────────────┘    └─────────────┘    └──────┬──────┘         │
│                                                │                │
│  ┌─────────────┐    ┌─────────────┐           │                │
│  │  反馈层      │◀───│  执行层      │◀──────────┘                │
│  │  效果评估    │    │  PLC参数下发 │     安全网关                │
│  │  模型迭代    │    │  设备控制    │     (权限/限流/回滚)        │
│  └─────────────┘    └─────────────┘                            │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
from enum import Enum
import threading
import time
import logging

logger = logging.getLogger(__name__)

class ToolPermission(Enum):
    READ_ONLY = "read_only"        # 只读，无风险
    WRITE_SAFE = "write_safe"      # 可写，但有安全边界（如检测阈值）
    WRITE_CRITICAL = "write_critical"  # 关键写入，需人工确认（如温度设定）

@dataclass
class ToolParameter:
    name: str
    type: str  # string / number / integer / boolean / array
    description: str
    required: bool = True
    default: Any = None
    enum: Optional[List] = None       # 枚举白名单
    min_value: Optional[float] = None # 数值下限
    max_value: Optional[float] = None # 数值上限

@dataclass
class ToolResult:
    success: bool
    result: Any = None
    error: Optional[str] = None
    latency_ms: float = 0.0

class ToolRegistry:
    """工业级工具注册中心——单例模式"""
    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
                    cls._instance._tools = {}
                    cls._instance._disabled = set()
                    cls._instance._call_stats = {}
        return cls._instance

    @classmethod
    def reset(cls):
        """仅用于测试"""
        cls._instance = None

    def register(self, name: str, func: Callable, parameters: List[ToolParameter],
                 permission: ToolPermission = ToolPermission.READ_ONLY,
                 description: str = "", rate_limit_per_min: int = 60):
        """注册工具"""
        self._tools[name] = {
            "func": func,
            "parameters": {p.name: p for p in parameters},
            "permission": permission,
            "description": description,
            "rate_limit": rate_limit_per_min,
            "call_timestamps": [],
        }
        logger.info(f"工具注册成功: {name} (权限: {permission.value})")

    def disable(self, name: str):
        """禁用工具（设备维护时使用）"""
        self._disabled.add(name)
        logger.warning(f"工具已禁用: {name}")

    def enable(self, name: str):
        self._disabled.discard(name)

    def _validate_params(self, tool_name: str, kwargs: Dict) -> Dict:
        """参数校验——在触及设备之前把错误拦下来"""
        tool = self._tools[tool_name]
        validated = {}

        for param_name, param_def in tool["parameters"].items():
            value = kwargs.get(param_name, param_def.default)

            # 1. 必填检查
            if param_def.required and value is None:
                raise ValueError(f"参数 {param_name} 为必填项")

            if value is None:
                continue

            # 2. 类型检查+转换
            if param_def.type == "number":
                value = float(value)
            elif param_def.type == "integer":
                value = int(value)
            elif param_def.type == "boolean":
                value = bool(value)

            # 3. 枚举白名单
            if param_def.enum and value not in param_def.enum:
                raise ValueError(
                    f"参数 {param_name} 值 {value} 不在允许范围 {param_def.enum}"
                )

            # 4. 数值范围
            if param_def.min_value is not None and value < param_def.min_value:
                raise ValueError(
                    f"参数 {param_name} 值 {value} 低于下限 {param_def.min_value}"
                )
            if param_def.max_value is not None and value > param_def.max_value:
                raise ValueError(
                    f"参数 {param_name} 值 {value} 超过上限 {param_def.max_value}"
                )

            validated[param_name] = value

        return validated

    def _check_rate_limit(self, tool_name: str):
        """速率限制检查"""
        tool = self._tools[tool_name]
        now = time.time()
        # 清理1分钟前的记录
        tool["call_timestamps"] = [
            t for t in tool["call_timestamps"] if now - t < 60
        ]
        if len(tool["call_timestamps"]) >= tool["rate_limit"]:
            raise RuntimeError(f"工具 {tool_name} 触发速率限制")
        tool["call_timestamps"].append(now)

    def execute(self, name: str, **kwargs) -> ToolResult:
        """执行工具——校验顺序精心设计"""
        start = time.time()

        try:
            # 1. 工具是否已注册
            if name not in self._tools:
                return ToolResult(False, error=f"工具 {name} 未注册")

            # 2. 工具是否被禁用
            if name in self._disabled:
                return ToolResult(False, error=f"工具 {name} 已被禁用")

            # 3. 参数校验（先校验，不浪费限流令牌）
            validated = self._validate_params(name, kwargs)

            # 4. 速率限制（校验通过后才检查）
            self._check_rate_limit(name)

            # 5. 执行
            result = self._tools[name]["func"](**validated)
            latency = (time.time() - start) * 1000

            # 6. 统计
            self._call_stats.setdefault(name, {"success": 0, "fail": 0})
            self._call_stats[name]["success"] += 1

            return ToolResult(True, result=result, latency_ms=latency)

        except Exception as e:
            latency = (time.time() - start) * 1000
            self._call_stats.setdefault(name, {"success": 0, "fail": 0})
            self._call_stats[name]["fail"] += 1
            logger.error(f"工具执行失败 {name}: {e}")
            return ToolResult(False, error=str(e), latency_ms=latency)

# ========== 注册工艺参数读取工具 ==========
def read_process_params(station: str, param_names: List[str]) -> Dict:
    """通过OPC UA读取工艺参数"""
    from opcua import Client
    client = Client("opc.tcp://192.168.1.100:4840")
    client.connect()
    try:
        results = {}
        for p in param_names:
            node = client.get_node(f"ns=2;s={station}.{p}")
            results[p] = node.get_value()
        return results
    finally:
        client.disconnect()

registry = ToolRegistry()
registry.register(
    name="read_process_params",
    func=read_process_params,
    parameters=[
        ToolParameter("station", "string", "工位编号", enum=["SMT01", "SMT02", "REFLOW01"]),
        ToolParameter("param_names", "array", "要读取的参数名列表"),
    ],
    permission=ToolPermission.READ_ONLY,
    description="读取指定工位的实时工艺参数",
    rate_limit_per_min=30,
)

# ========== 注册工艺参数写入工具（安全受限）==========
def write_process_param(station: str, param_name: str, value: float) -> Dict:
    """写入工艺参数到PLC——关键参数需二次确认"""
    from opcua import Client
    client = Client("opc.tcp://192.168.1.100:4840")
    client.connect()
    try:
        node = client.get_node(f"ns=2;s={station}.{param_name}")
        old_value = node.get_value()
        node.set_value(value)
        # 验证写入成功
        time.sleep(0.1)
        new_value = node.get_value()
        return {
            "station": station,
            "param": param_name,
            "old_value": old_value,
            "new_value": new_value,
            "write_success": abs(new_value - value) < 0.001,
        }
    finally:
        client.disconnect()

registry.register(
    name="write_process_param",
    func=write_process_param,
    parameters=[
        ToolParameter("station", "string", "工位编号", enum=["SMT01", "SMT02"]),
        ToolParameter("param_name", "string", "参数名",
                      enum=["print_pressure", "print_speed", "reflow_temp_zone3"]),
        ToolParameter("value", "number", "参数值", min_value=0, max_value=100),
    ],
    permission=ToolPermission.WRITE_SAFE,
    description="写入工艺参数（自动记录旧值，支持回滚）",
    rate_limit_per_min=10,  # 写入操作严格限流
)

# ========== 注册YOLO检测工具 ==========
def run_yolo_detection(image_path: str, conf_thres: float = 0.25) -> Dict:
    """运行YOLO26缺陷检测"""
    from ultralytics import YOLO
    model = YOLO("yolo26s_int8.onnx")
    results = model(image_path, conf=conf_thres, verbose=False)
    detections = []
    for r in results:
        for box in r.boxes:
            detections.append({
                "bbox": box.xyxy[0].tolist(),
                "confidence": float(box.conf[0]),
                "class": int(box.cls[0]),
            })
    return {"detections": detections, "count": len(detections)}

registry.register(
    name="run_yolo_detection",
    func=run_yolo_detection,
    parameters=[
        ToolParameter("image_path", "string", "图像路径"),
        ToolParameter("conf_thres", "number", "置信度阈值", required=False,
                      default=0.25, min_value=0.1, max_value=0.9),
    ],
    permission=ToolPermission.READ_ONLY,
    description="运行YOLO26缺陷检测，返回检测结果",
)

from typing import TypedDict, Annotated, List, Dict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
import operator
import json

# ========== 定义共享状态 ==========
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    current_step: str
    image_path: str
    detection_result: Dict
    analysis_result: Dict
    current_params: Dict
    optimization_plan: Dict
    execution_result: Dict
    recheck_result: Dict
    safety_approved: bool
    retry_count: int
    error: str

# ========== 初始化LLM ==========
llm = ChatOpenAI(
    model="qwen2.5-72b-instruct",
    base_url="http://local-vlm:8000/v1",
    api_key="EMPTY",
    temperature=0.1,
)

# ========== 各Agent节点函数 ==========

def detection_node(state: AgentState) -> AgentState:
    """检测Agent：运行YOLO检测"""
    registry = ToolRegistry()
    result = registry.execute(
        "run_yolo_detection",
        image_path=state["image_path"],
        conf_thres=0.25,
    )

    if not result.success:
        return {**state, "error": result.error, "current_step": "error"}

    detection = result.result
    state["detection_result"] = detection

    # 判断是否有缺陷
    if detection["count"] == 0:
        return {**state, "current_step": "no_defect"}

    return {**state, "current_step": "analysis"}

def analysis_node(state: AgentState) -> AgentState:
    """分析Agent：VLM根因分析 + 知识库检索"""
    # 1. 读取当前工艺参数
    registry = ToolRegistry()
    params_result = registry.execute(
        "read_process_params",
        station="SMT01",
        param_names=["print_pressure", "print_speed", "stencil_opening_ratio"],
    )
    current_params = params_result.result if params_result.success else {}

    # 2. 调用VLM分析（简化，实际调用VLM服务）
    analysis_prompt = f"""
    检测到缺陷：{json.dumps(state['detection_result'], ensure_ascii=False)}
    当前工艺参数：{json.dumps(current_params, ensure_ascii=False)}

    请分析缺陷根因，并给出参数调整建议。输出JSON：
    {{
        "root_cause": "根本原因",
        "confidence": 0.0-1.0,
        "param_adjustments": [
            {{"param": "参数名", "current": 当前值, "suggested": 建议值, "reason": "原因"}}
        ],
        "expected_effect": "预期改善效果"
    }}
    """
    response = llm.invoke([HumanMessage(content=analysis_prompt)])
    analysis = json.loads(response.content)

    return {
        **state,
        "analysis_result": analysis,
        "current_params": current_params,
        "current_step": "optimization",
    }

def optimization_node(state: AgentState) -> AgentState:
    """优化Agent：贝叶斯优化计算最优参数"""
    analysis = state["analysis_result"]

    # 构建优化计划
    plan = {
        "adjustments": analysis["param_adjustments"],
        "method": "bayesian_optimization",
        "search_space": {
            "print_pressure": [0.1, 0.3],  # MPa
            "print_speed": [20, 80],       # mm/s
        },
        "max_trials": 5,
        "safety_constraints": {
            "max_single_adjustment_pct": 15,  # 单次调整不超过15%
            "cooldown_minutes": 10,           # 调整后等待10分钟再复检
        },
    }

    return {
        **state,
        "optimization_plan": plan,
        "current_step": "safety_check",
    }

def safety_check_node(state: AgentState) -> AgentState:
    """安全Agent：风险评估 + 权限校验"""
    plan = state["optimization_plan"]
    approved = True
    reasons = []

    for adj in plan["adjustments"]:
        current = adj["current"]
        suggested = adj["suggested"]
        change_pct = abs(suggested - current) / current * 100 if current != 0 else 100

        if change_pct > plan["safety_constraints"]["max_single_adjustment_pct"]:
            approved = False
            reasons.append(f"参数 {adj['param']} 调整幅度 {change_pct:.1f}% 超过安全限制 15%")

        # 关键参数需要人工确认
        critical_params = ["reflow_temp_zone3", "reflow_temp_zone5"]
        if adj["param"] in critical_params:
            approved = False
            reasons.append(f"参数 {adj['param']} 为关键参数，需人工确认")

    return {
        **state,
        "safety_approved": approved,
        "current_step": "execution" if approved else "human_review",
    }

def execution_node(state: AgentState) -> AgentState:
    """执行Agent：参数下发 + 复检"""
    registry = ToolRegistry()
    plan = state["optimization_plan"]
    execution_log = []

    # 1. 记录当前参数（用于回滚）
    backup = state["current_params"].copy()

    # 2. 逐个下发参数
    for adj in plan["adjustments"]:
        result = registry.execute(
            "write_process_param",
            station="SMT01",
            param_name=adj["param"],
            value=adj["suggested"],
        )
        execution_log.append({
            "param": adj["param"],
            "success": result.success,
            "old": adj["current"],
            "new": adj["suggested"],
            "error": result.error,
        })

    # 3. 等待稳定后复检（简化，实际等待10分钟）
    # time.sleep(600)
    recheck = registry.execute(
        "run_yolo_detection",
        image_path=state["image_path"],  # 实际用新拍摄的图像
    )

    return {
        **state,
        "execution_result": {"log": execution_log, "backup": backup},
        "recheck_result": recheck.result if recheck.success else {},
        "current_step": "evaluate",
    }

def evaluate_node(state: AgentState) -> AgentState:
    """评估节点：判断调参效果，决定是否回滚或继续"""
    before_count = state["detection_result"]["count"]
    after_count = state["recheck_result"].get("count", before_count)

    if after_count < before_count:
        # 改善了，记录并结束
        return {
            **state,
            "current_step": "success",
            "messages": [AIMessage(
                content=f"调参成功！缺陷数从 {before_count} 降至 {after_count}"
            )],
        }
    elif state["retry_count"] < 3:
        # 没改善，重试（换一组参数）
        return {
            **state,
            "retry_count": state["retry_count"] + 1,
            "current_step": "optimization",
        }
    else:
        # 重试3次仍失败，回滚并通知人工
        registry = ToolRegistry()
        backup = state["execution_result"]["backup"]
        for param, value in backup.items():
            registry.execute("write_process_param", station="SMT01",
                           param_name=param, value=value)
        return {
            **state,
            "current_step": "rollback",
            "messages": [AIMessage(content="调参失败，已回滚参数，需人工介入")],
        }

# ========== 构建工作流图 ==========
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("detection", detection_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("optimization", optimization_node)
workflow.add_node("safety_check", safety_check_node)
workflow.add_node("execution", execution_node)
workflow.add_node("evaluate", evaluate_node)

# 设置入口
workflow.set_entry_point("detection")

# 定义边和条件路由
workflow.add_conditional_edges(
    "detection",
    lambda state: state["current_step"],
    {
        "analysis": "analysis",
        "no_defect": END,
        "error": END,
    }
)
workflow.add_edge("analysis", "optimization")
workflow.add_edge("optimization", "safety_check")
workflow.add_conditional_edges(
    "safety_check",
    lambda state: state["current_step"],
    {
        "execution": "execution",
        "human_review": END,  # 实际接入人工审批界面
    }
)
workflow.add_edge("execution", "evaluate")
workflow.add_conditional_edges(
    "evaluate",
    lambda state: state["current_step"],
    {
        "success": END,
        "optimization": "optimization",  # 重试
        "rollback": END,
    }
)

# 编译
app = workflow.compile()

# 启动一次完整的检测→分析→调参→复检闭环
result = app.invoke({
    "messages": [HumanMessage(content="开始质检闭环")],
    "current_step": "detection",
    "image_path": "./test_images/pcb_defect_001.jpg",
    "retry_count": 0,
})

print(f"最终状态: {result['current_step']}")
print(f"检测缺陷数: {result['detection_result']['count']}")
print(f"调参后缺陷数: {result['recheck_result'].get('count', 'N/A')}")

import numpy as np
from skopt import Optimizer
from skopt.space import Real

class ProcessParameterOptimizer:
    """工艺参数贝叶斯优化器"""

    def __init__(self, param_space: Dict, baseline_params: Dict):
        """
        param_space: {"print_pressure": (0.1, 0.3), "print_speed": (20, 80)}
        baseline_params: 当前生产参数
        """
        self.param_names = list(param_space.keys())
        self.dimensions = [
            Real(low, high, name=name)
            for name, (low, high) in param_space.items()
        ]
        self.optimizer = Optimizer(
            dimensions=self.dimensions,
            base_estimator="GP",  # 高斯过程
            acq_func="EI",        # 期望提升
            random_state=42,
        )
        self.baseline = [baseline_params[name] for name in self.param_names]
        self.history = []  # (params, defect_rate)

    def suggest(self, n_suggestions: int = 1) -> List[Dict]:
        """建议下一组参数"""
        suggestions = self.optimizer.ask(n_points=n_suggestions)
        return [
            dict(zip(self.param_names, s))
            for s in suggestions
        ]

    def report(self, params: Dict, defect_rate: float):
        """报告这组参数的效果（缺陷率越低越好）"""
        x = [params[name] for name in self.param_names]
        self.optimizer.tell(x, defect_rate)
        self.history.append((params, defect_rate))

    def get_best(self) -> Dict:
        """获取历史最优参数"""
        if not self.history:
            return dict(zip(self.param_names, self.baseline))
        best = min(self.history, key=lambda x: x[1])
        return best[0]

# ========== 使用示例 ==========
optimizer = ProcessParameterOptimizer(
    param_space={
        "print_pressure": (0.1, 0.3),   # MPa
        "print_speed": (20, 80),         # mm/s
        "squeegee_angle": (45, 75),      # 度
    },
    baseline_params={
        "print_pressure": 0.2,
        "print_speed": 50,
        "squeegee_angle": 60,
    },
)

# 迭代优化
for trial in range(5):
    # 1. Agent建议参数
    suggested = optimizer.suggest()[0]
    print(f"第{trial+1}轮建议参数: {suggested}")

    # 2. 安全检查（调整幅度不超过15%）
    safe = True
    for name, value in suggested.items():
        baseline = optimizer.baseline[optimizer.param_names.index(name)]
        if abs(value - baseline) / baseline > 0.15:
            safe = False
            break

    if not safe:
        print("参数调整幅度过大，跳过")
        continue

    # 3. 下发参数到PLC（通过工具注册中心）
    # ... execution ...

    # 4. 等待稳定，复检缺陷率
    # time.sleep(600)  # 等待10分钟
    defect_rate = 0.02 + np.random.random() * 0.05  # 模拟缺陷率

    # 5. 报告效果
    optimizer.report(suggested, defect_rate)
    print(f"  缺陷率: {defect_rate:.4f}")

# 最优参数
best = optimizer.get_best()
print(f"\n最优参数: {best}")

class SafetyGuard:
    """Agent安全防护——六层刹车"""

    def __init__(self):
        self.layers = [
            self._layer1_param_validation,    # 参数校验
            self._layer2_rate_limit,          # 速率限制
            self._layer3_safety_boundary,     # 安全边界
            self._layer4_human_approval,      # 人工审批
            self._layer5_execution_verify,    # 执行验证
            self._layer6_rollback,            # 回滚机制
        ]

    def _layer1_param_validation(self, action):
        """第一层：参数格式和范围校验"""
        for param, value in action["params"].items():
            if param in action.get("constraints", {}):
                c = action["constraints"][param]
                if "min" in c and value < c["min"]:
                    raise SafetyError(f"参数 {param}={value} 低于下限 {c['min']}")
                if "max" in c and value > c["max"]:
                    raise SafetyError(f"参数 {param}={value} 超过上限 {c['max']}")
        return True

    def _layer2_rate_limit(self, action):
        """第二层：操作频率限制"""
        # 同一参数10分钟内最多调整2次
        # 写入操作每分钟不超过5次
        pass

    def _layer3_safety_boundary(self, action):
        """第三层：物理安全边界"""
        # 温度调整不超过±5℃
        # 压力调整不超过±20%
        # 关键参数（如反应釜温度）直接拒绝自动调整
        critical_params = ["reactor_temp", "high_voltage_setpoint"]
        if action["param"] in critical_params:
            raise SafetyError(f"关键参数 {action['param']} 不允许自动调整")
        return True

    def _layer4_human_approval(self, action):
        """第四层：高风险操作人工审批"""
        if action["risk_level"] == "high":
            # 发送审批请求到工程师手机
            # 等待确认，超时默认拒绝
            approved = self._request_human_approval(action)
            if not approved:
                raise SafetyError("人工审批未通过")
        return True

    def _layer5_execution_verify(self, action):
        """第五层：执行后验证"""
        # 写入后立即回读，确认参数确实改了
        # 监控设备状态，异常立即触发回滚
        pass

    def _layer6_rollback(self, action):
        """第六层：自动回滚"""
        # 调整后30分钟内缺陷率上升超过50% → 自动回滚
        # 设备报警 → 立即回滚
        # 回滚后记录审计日志
        pass

    def check(self, action) -> bool:
        for layer in self.layers:
            try:
                layer(action)
            except SafetyError as e:
                logger.warning(f"安全拦截 [{layer.__name__}]: {e}")
                return False
        return True