AI编程实战：我用Python+LangChain搭建了一个半导体FAB智能运维Agent

前言

2024年被称为'AI Agent元年'。从AutoGPT到LangChain，从CrewAI到OpenAI Assistants，各种Agent框架层出不穷。但作为一个在半导体行业摸爬滚打了15年的老工程师，我发现一个有趣的现象：绝大多数AI Agent的教程和案例都集中在通用场景——写代码、做客服、查资料。而真正需要AI Agent的工业现场，却几乎看不到像样的实践案例。

这篇文章，我要分享一个真实的工业项目：如何用Python+LangChain，为一家12英寸晶圆厂搭建一个智能运维Agent。它能够自动监控设备状态、诊断异常原因、推荐维修方案，甚至自动生成工单。整个项目从零到上线只用了2周时间。

一、背景：为什么半导体FAB需要AI Agent

1.1 传统运维的痛点

先说说我所在的这家晶圆厂的情况：

• 300mm晶圆厂，月产能4万片
• 设备总数：200+台（蚀刻、CVD、光刻、CMP等）
• 运维团队：15人，7×24小时轮班
• 每天平均报警数：500+条

传统运维流程是这样的：

平均一次故障处理：**62-140分钟**。其中真正修设备的时间可能只有30分钟，剩下全在查资料、找数据、填表单上。

1.2 用AI Agent能解决什么？

我的想法很简单：把步骤2-6全部交给AI Agent来做。它应该能够：

二、技术选型：为什么选LangChain

2.1 方案对比

最终选择LangChain，理由如下：

• **工具集成能力强**：LangChain有现成的SQL Database Tool、Vector Store Retriever、API Toolkit等，正好对接我们的MES数据库和知识库
• **RAG支持完善**：我们需要把几百份维修SOP、历史工单做成知识库，LangChain的RAG pipeline非常成熟
• **社区活跃**：遇到问题容易找到解决方案
• **Python原生**：我们的数据采集和分析工具都是Python写的，无缝衔接

三、系统架构设计

3.1 整体架构

系统采用经典的ReAct（Reasoning + Acting）架构：

# 智能运维Agent核心架构
# ┌─────────────────────────────────────────────┐
# │              User Interface (Web/Chat)       │
# └──────────────────────┬──────────────────────┘
#                        │
# ┌──────────────────────▼──────────────────────┐
# │            LangChain Agent Core              │
# │  ┌─────────┐  ┌─────────┐  ┌─────────────┐ │
# │  │ LLM     │  │ Memory  │  │ Tools       │ │
# │  │ (GPT-4) │  │ (RAG)   │  │ (5个工具)   │ │
# │  └────┬────┘  └────┬────┘  └──────┬──────┘ │
# │       └────────────┼──────────────┘         │
# │                    │ ReAct Loop             │
# └────────────────────┼────────────────────────┘
#                      │
#         ┌────────────┼────────────┐
#         ▼            ▼            ▼
#   ┌──────────┐ ┌──────────┐ ┌──────────┐
#   │ MES DB   │ │ Knowledge│ │ SECS/GEM │
#   │ (MySQL)  │ │ Base     │ │ Gateway  │
#   └──────────┘ └──────────┘ └──────────┘

3.2 五大工具模块

四、核心代码实现

4.1 环境准备

# requirements.txt
langchain==0.1.0
langchain-openai==0.0.2
langchain-community==0.0.2
chromadb==0.4.22
pymysql==1.1.0
influxdb-client==1.38.0
fastapi==0.109.0
uvicorn==0.27.0

4.2 定义工具函数

# tools.py – 五大工具模块
import pymysql
import json
from datetime import datetime, timedelta
from langchain.tools import tool
from typing import Dict, List

@tool
def alarm_query(equipment_id: str, hours: int = 24) -> str:
    """
    查询指定设备的报警记录。
    
    Args:
        equipment_id: 设备编号（如 ETCH-001）
        hours: 查询最近多少小时的数据，默认24小时
    
    Returns:
        报警记录JSON字符串
    """
    conn = pymysql.connect(
        host="192.168.1.100",
        user="fab_readonly",
        password="xxx",
        database="fdb",
        charset="utf8mb4"
    )
    
    start_time = datetime.now() – timedelta(hours=hours)
    
    with conn.cursor(pymysql.cursors.DictCursor) as cursor:
        sql = """
        SELECT alarm_id, equipment_id, alarm_code, alarm_message,
               severity, timestamp, status
        FROM fdc_alarm_log
        WHERE equipment_id = %s AND timestamp >= %s
        ORDER BY timestamp DESC
        LIMIT 50
        """
        cursor.execute(sql, (equipment_id, start_time))
        results = cursor.fetchall()
    
    conn.close()
    return json.dumps(results, ensure_ascii=False, default=str)

@tool
def knowledge_search(query: str, top_k: int = 3) -> str:
    """
    从维修知识库中检索相关的SOP和案例。
    
    Args:
        query: 搜索关键词或问题描述
        top_k: 返回最相关的结果数量
    
    Returns:
        相关知识文档列表
    """
    import chromadb
    from chromadb.utils import embedding_functions
    
    client = chromadb.PersistentClient(path="./knowledge_db")
    ef = embedding_functions.SentenceTransformerEmbeddingFunction(
        model_name="shibing624/text2vec-base-chinese"
    )
    collection = client.get_collection(
        name="maintenance_sop",
        embedding_function=ef
    )
    
    results = collection.query(
        query_texts=[query],
        n_results=top_k
    )
    
    output = []
    for i, doc in enumerate(results["documents"][0]):
        output.append({
            "rank": i + 1,
            "content": doc[:2000],  # 截断过长内容
            "metadata": results["metadatas"][0][i]
        })
    
    return json.dumps(output, ensure_ascii=False, indent=2)

@tool
def history_analysis(equipment_id: str, alarm_code: str) -> str:
    """
    分析指定设备和报警码的历史处理记录，
    找出最常见的根因和处理方案。
    
    Args:
        equipment_id: 设备编号
        alarm_code: 报警码
    
    Returns:
        历史分析报告
    """
    conn = pymysql.connect(
        host="192.168.1.100",
        user="fab_readonly",
        password="xxx",
        database="mes",
        charset="utf8mb4"
    )
    
    with conn.cursor(pymysql.cursors.DictCursor) as cursor:
        # 查询该设备+报警码的历史工单
        sql = """
        SELECT root_cause, solution, repair_duration_minutes,
               COUNT(*) as occurrence_count
        FROM work_order
        WHERE equipment_id = %s AND alarm_code = %s
          AND status = 'COMPLETED'
        GROUP BY root_cause, solution
        ORDER BY occurrence_count DESC
        LIMIT 10
        """
        cursor.execute(sql, (equipment_id, alarm_code))
        results = cursor.fetchall()
        
        # 统计平均修复时间
        cursor.execute("""
        SELECT AVG(repair_duration_minutes) as avg_duration,
               MIN(repair_duration_minutes) as min_duration,
               MAX(repair_duration_minutes) as max_duration
        FROM work_order
        WHERE equipment_id = %s AND alarm_code = %s AND status = 'COMPLETED'
        """, (equipment_id, alarm_code))
        stats = cursor.fetchone()
    
    conn.close()
    
    report = {
        "historical_cases": results,
        "statistics": {
            "avg_repair_time_min": round(stats["avg_duration"], 1),
            "min_repair_time_min": stats["min_duration"],
            "max_repair_time_min": stats["max_duration"],
            "total_cases": sum(r["occurrence_count"] for r in results)
        }
    }
    
    return json.dumps(report, ensure_ascii=False, indent=2)

@tool
def ticket_generator(
    equipment_id: str,
    alarm_code: str,
    root_cause: str,
    solution: str,
    priority: str = "MEDIUM"
) -> str:
    """
    自动生成维修工单并提交到MES系统。
    
    Args:
        equipment_id: 设备编号
        alarm_code: 报警码
        root_cause: 诊断出的根因
        solution: 推荐的解决方案
        priority: 优先级（HIGH/MEDIUM/LOW）
    
    Returns:
        工单号和提交结果
    """
    import requests
    
    ticket_data = {
        "equipment_id": equipment_id,
        "alarm_code": alarm_code,
        "root_cause": root_cause,
        "solution": solution,
        "priority": priority,
        "created_by": "AI_Agent_V1",
        "created_at": datetime.now().isoformat(),
        "source": "intelligent_maintenance_agent"
    }
    
    response = requests.post(
        "http://mes-api.internal:8080/api/v1/work-order",
        json=ticket_data,
        timeout=10
    )
    
    result = response.json()
    result["ai_generated"] = True
    result["confidence"] = "HIGH" if priority == "HIGH" else "MEDIUM"
    
    return json.dumps(result, ensure_ascii=False, indent=2)

@tool
def equipment_status(equipment_id: str) -> str:
    """
    获取设备当前的实时运行状态。
    
    Args:
        equipment_id: 设备编号
    
    Returns:
        设备状态信息（温度、压力、运行模式等）
    """
    # 通过OPC UA或SECS/GEM获取实时数据
    # 这里用模拟数据演示
    import random
    
    status = {
        "equipment_id": equipment_id,
        "timestamp": datetime.now().isoformat(),
        "mode": random.choice(["PROCESSING", "IDLE", "MAINTENANCE", "ALARM"]),
        "chamber_temp": round(random.uniform(20, 80), 1),
        "chamber_pressure": round(random.uniform(5, 50), 2),
        "rf_power": round(random.uniform(300, 1500), 1),
        "utilization_today": round(random.uniform(0.6, 0.95), 2),
        "last_pm_date": "2026-05-15",
        "days_since_pm": 20
    }
    
    return json.dumps(status, ensure_ascii=False, indent=2)

4.3 组装Agent

# agent.py – 组装智能运维Agent
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from tools import (
    alarm_query, knowledge_search,
    history_analysis, ticket_generator, equipment_status
)

# 初始化LLM
llm = ChatOpenAI(
    model="gpt-4-turbo-preview",
    temperature=0,
    streaming=True
)

# 工具列表
tools = [
    alarm_query,
    knowledge_search,
    history_analysis,
    ticket_generator,
    equipment_status
]

# System Prompt – 这是关键！
system_prompt = """你是一个专业的半导体晶圆厂智能运维助手。

## 你的身份
– 你服务于一家12英寸（300mm）半导体晶圆厂
– 你的职责是帮助工程师快速诊断设备故障、提供维修建议
– 你有15年半导体设备维护经验的知识储备

## 你的能力
你可以使用以下工具：
1. alarm_query: 查询设备报警记录
2. knowledge_search: 检索维修知识库（SOP、历史案例）
3. history_analysis: 分析同类故障的历史处理记录
4. ticket_generator: 生成维修工单
5. equipment_status: 获取设备实时状态

## 工作流程
当收到设备报警时，请按以下步骤操作：

Step 1: 使用 equipment_status 获取设备当前状态
Step 2: 使用 alarm_query 查询最近的报警记录
Step 3: 使用 history_analysis 分析历史同类故障
Step 4: 使用 knowledge_search 检索相关知识库
Step 5: 综合以上信息，给出诊断结论和建议方案
Step 6: 如果用户确认，使用 ticket_generator 生成工单

## 输出格式
请用中文回复，格式如下：

—
【设备】ETCH-001 | 【报警】VACUUM_LOSS
【诊断结论】真空泵密封圈老化导致真空度下降
【置信度】85%（基于3次历史相似案例）
【推荐方案】更换真空泵密封圈（Part#: VP-SEAL-200）
【预计耗时】45分钟
【历史参考】近6个月内同类故障4次，平均修复时间42分钟
—

## 注意事项
– 安全第一！如果涉及高压、有毒气体，必须提醒安全注意事项
– 不要编造数据，如果工具返回空结果，如实告知
– 建议具体到零件号（Part Number）和操作步骤
"""

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

# 创建Agent
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True,
    return_intermediate_steps=True
)

# 运行示例
if __name__ == "__main__":
    result = agent_executor.invoke({
        "input": "ETCH-001设备刚刚报了VACUUM_LOSS报警，请帮我诊断一下"
    })
    print(result["output"])

五、上线效果与数据对比

5.1 关键指标提升

5.2 一个真实案例

上线第三天的一个真实案例：

**时间**：凌晨02:17（夜班，只有1名初级工程师值班）

**事件**：CVD-003设备报 TEMPERATURE_OVERRUN 报警

**Agent处理过程**：

=== Agent 执行日志 ===

[STEP 1] 获取设备实时状态
→ CVD-003: Chamber Temp = 485°C (正常范围: 400±10°C)
→ Mode: ALARM | Utilization Today: 92%

[STEP 2] 查询最近24小时报警记录
→ 发现5条报警：3条TEMP_OVERRUN, 2条PRESSURE_LOW
→ 首次报警：01:43

[STEP 3] 历史同类故障分析
→ 近1年类似故障：18次
→ 最常见根因(67%)：加热器Thermocouple漂移
→ 平均修复时间：52分钟

[STEP 4] 知识库检索
→ 匹配到3篇相关SOP：
  [1] CVD加热器校准标准程序 (匹配度: 94%)
  [2] Thermocouple更换作业指导书 (匹配度: 89%)
  [3] CVD温度异常排查Checklist (匹配度: 86%)

[STEP 5] 综合诊断
→ 结论：Zone 3热电偶漂移，偏差约+15°C
→ 置信度：82%（基于历史18次相似案例）
→ 建议：执行热电偶校准，如偏差>10°C则更换

[STEP 6] 生成工单
→ 工单号: WO-20260604-0217
→ 优先级: HIGH
→ 预计耗时: 45分钟

=== 总耗时: 3分42秒 ===

这位初级工程师按照Agent的建议，45分钟内完成了热电偶校准，设备恢复正常。如果是以前，他可能要打电话给资深工程师，或者翻阅大量文档，至少折腾1-2个小时。

六、踩坑经验与优化建议

6.1 遇到的坑

6.2 优化方向

• **模型本地化**：部署私有化LLM（如Qwen-72B），降低数据外泄风险和API成本
• **多模态接入**：接入设备摄像头图片，让Agent能'看'设备状态
• **预测性维护**：结合时序预测模型，在故障发生前预警
• **多Agent协作**：拆分为诊断Agent、方案Agent、执行Agent，用CrewAI编排

七、总结

通过这个项目，我有几点深刻的体会：

—

**�� 关注我，持续分享半导体智能制造领域的AI实战经验！**

**�� 你在工业现场用过AI吗？欢迎评论区分享你的经验，我会一一回复！**