生产级多 Agent 系统评估框架在半导体晶圆厂场景的落地实践

github：https://github.com/BumbleBee-ZDS/fab_agent_test
关键词：多 Agent 系统、LLM 评估体系、工业大模型、半导体 FAB、白盒架构、Streamlit

---

一、背景：从"能用"到"可上线"，Agent 评估的核心矛盾

在大模型应用从 PoC 走向生产的实践中，一个反复被验证的结论是：传统评估指标（准确率、BLEU、ROUGE）无法支撑多 Agent 系统的上线决策。

以半导体晶圆厂（Fabrication Plant，简称 FAB）的工艺缺陷根因分析场景为例。工艺工程师提交一个问题：

“批次 W12345 的关键尺寸（Critical Dimension, CD）超标，请分析原因。”

如果采用黑盒 Agent 框架（如 LangChain / CrewAI），系统可能在三种典型失败模式下"看似工作正常"：

1. 资源失控：反复调用设备接口，单次分析消耗数万 Token，成本不可控；
2. 异常崩溃：机台 API 超时后直接抛出未捕获异常，工程师被迫切换到人工排查；
3. 空转反思：Reflector 模块陷入"让我再想想"的循环中，20 轮输出零信息增益。

这些问题的共同特征是：结果可能是对的，但过程是危险的。

1.1 评估视角的根本转变

本文基于《生产环境多 Agent 系统评估技术框架》的核心原则，提出一套面向工业场景的评估方法论，并将其落地为一个可运行的 Streamlit MVP。核心转变如下：

评估维度	传统视角	生产环境视角
过程质量	关注最终输出是否正确	关注规划路径合理性、是否存在无效轮次、是否空转
资源成本	通常被忽略	Token 消耗、工具调用次数、端到端延迟
系统韧性	报错即视为失败	异常处理与恢复能力、降级策略有效性
业务价值	语义相似度	是否产出可执行的工艺建议（Actionable Insight）

一句话总结：评估体系需要从"任务做没做完"升级为"模块能力是否达标、协作链路是否健康、线上成本是否可控、业务价值是否兑现"。

---

二、系统设计：白盒架构与模块解耦

为避免黑盒不可观测的问题，本项目采用纯手写 Agent 逻辑 + 标准库的方案，所有模块在 core/ 包中显式定义，总代码量约 600 行，无外部 Agent 框架依赖。

2.1 目录结构

fab_agent_test/
├── app.py                       # Streamlit UI 入口（仅 UI 层）
├── core/                        # Agent 核心模块（白盒）
│   ├── __init__.py
│   ├── evaluator.py             # 评估指标收集器
│   ├── memory.py                # 记忆模块
│   ├── toolset.py               # 工具集（含不稳定接口）
│   ├── planner.py               # 规划器
│   ├── reflector.py             # 反思器
│   └── orchestrator.py          # 编排器 / 主循环
├── data/
│   └── mock_data.py             # Mock 数据层
├── gen_test_data.py             # DeepSeek 批量生成测试数据（可选）
└── fab_test_data.json           # 生成的测试数据

2.2 模块职责与接口设计

(1) Memory —— 防遗忘的短期记忆

# core/memory.py
class Memory:
    def __init__(self):
        self._store: dict = {}

    def store(self, key: str, value) -> None:
        self._store[key] = value

    def recall(self, key: str):
        return self._store.get(key)

Memory 模块在 Orchestrator 启动的第一帧被写入 Lot ID，确保后续 5 步执行中不会丢失关键上下文。

(2) Planner —— 规划与动态调度

规划器输出一个有序的子任务列表，这是过程质量的首要评估对象：

# core/planner.py
class Planner:
    def make_plan(self, question: str) -> list[str]:
        return [
            "查询批次基本信息",
            "查询机台运行日志",
            "查询工艺配方参数",
            "反思校验数据一致性",
            "生成根因分析与建议"
        ]

    def adjust_plan_skip_equipment(self, plan: list[str]) -> list[str]:
        """
        失败自愈：将"查询机台运行日志"替换为"查询批次历史"。
        仅允许调整一次，避免无限降级。
        """
        if "查询机台运行日志" in plan:
            plan[plan.index("查询机台运行日志")] = "查询批次历史（设备不可用降级）"
            return plan
        return plan

(3) ToolSet —— 工具调用与不稳定接口模拟

ToolSet 封装了 4 个工具，其中设备日志接口以 30% 的概率模拟超时，用于验证系统韧性：

# core/toolset.py
import random

class ToolSet:
    EQUIP_TIMEOUT_RATE = 0.30

    def get_equipment_log(self, chamber_id: str) -> dict | None:
        if random.random() < self.EQUIP_TIMEOUT_RATE:
            return None  # 模拟超时
        return {
            "chamber_id": chamber_id,
            "pressure_mTorr": round(random.uniform(6.0, 8.0), 2),
            "rf_power_w": round(random.uniform(1200, 1500), 0),
            "temperature_c": round(random.uniform(40, 60), 1)
        }
    # ... 其余工具省略

(4) Reflector —— 数值冲突检测

反思器不做语义理解，而是对关键数值做一致性校验。在 FAB 场景中，工艺配方设定的压力值与机台实测压力值的偏差超过阈值即视为矛盾：

# core/reflector.py
class Reflector:
    PRESSURE_DELTA_THRESHOLD = 1.0  # mTorr

    def check_conflict(self, recipe: dict, equipment_log: dict) -> dict:
        conflict = False
        if recipe and equipment_log:
            delta = abs(recipe["pressure_setpoint"] - equipment_log["pressure_mTorr"])
            if delta > self.PRESSURE_DELTA_THRESHOLD:
                conflict = True
        return {"has_conflict": conflict, "detail": f"压力差值 {delta:.2f} mTorr"}

(5) Orchestrator —— 主循环与双重熔断

编排器是整个系统的心脏，内置两道终止防线：

# core/orchestrator.py
class Orchestrator:
    MAX_STEPS = 6

    def run(self, question: str, evaluator: Evaluator) -> str:
        plan = self.planner.make_plan(question)
        last_tool_signature = None

        for idx, step in enumerate(plan):
            if evaluator.step_count >= self.MAX_STEPS:
                break  # 防线①：步数上限

            signature = self._get_step_signature(step, evaluator)
            if signature == last_tool_signature:
                evaluator.dead_loop_flag = True
                break  # 防线②：死循环检测

            # 分发到具体步骤...
            last_tool_signature = signature

        return self._generate_report(evaluator)

2.3 死循环检测机制

这是本文最值得强调的工程细节。传统超时机制只能解决"卡死"，无法解决"空转"。本方案通过工具调用签名比对实现语义级重复检测：

def _get_step_signature(self, step: str, evaluator: Evaluator) -> str:
    """生成步骤签名，用于死循环检测"""
    tool_name = step.split("（")[0].split("(")[0].strip()
    args = evaluator.memory.recall("lot_id") or "none"
    signature = f"{tool_name}|{args}"
    evaluator.tool_call_count += 1
    return signature

原理：如果连续两步的工具调用签名完全一致（同一工具 + 同一参数），判定为死循环并强制终止。这直接对应理论框架中"避免空转反思"的要求。

---

三、Evaluator：生产级评估的核心引擎

Evaluator 是整套系统的灵魂模块，它在 Agent 运行的每一个原子操作之后收集指标，对应理论中的四大评估维度。

3.1 指标清单与采集点

指标字段	数据类型	对应维度	采集位置
step_count	int	过程质量	每执行一步自增
reflection_valid	bool	过程质量	Reflector 返回冲突时置 True
tool_call_count	int	资源成本	每次工具调用后
token_cost_mock	int	资源成本	每次输出拼接后累加字符数
retry_count	int	系统韧性	工具超时重试后
dead_loop_flag	bool	系统韧性	签名重复时置 True
timeout_handled	bool	系统韧性	重试后是否恢复
resilience_score	str	系统韧性	流程结束时计算
business_value	bool	业务价值	报告生成后检查

3.2 韧性评分逻辑

韧性评分是连接技术指标与业务语义的桥梁：

def resilience_score(self) -> str:
    if self.retry_count == 0:
        return "高（未触发超时）"
    elif self.retry_count > 0 and self.timeout_handled:
        return "高（已成功自愈）"
    else:
        return "中（已重试但未恢复，已降级）"

3.3 业务价值判定

业务价值的判定遵循"可操作性优先"原则——没有明确建议的分析对工程师而言是无用的噪音：

def has_business_value(self, report: str) -> bool:
    return "建议" in report

---

四、Streamlit UI：运维视角的大屏化呈现

UI 采用双栏布局，左侧为执行流日志，右侧为实时评估面板，刻意模拟运维监控大屏的信息层级。

4.1 左栏：执行流

输入区提供 DeepSeek 生成的示例问题下拉框 + 自由文本输入，点击"开始分析"后，日志区以时间戳形式实时追加每步事件：

[12:01:03] [Planner] 生成执行计划（5 步）
[12:01:03] [Memory] 存储 Lot ID = W12345
[12:01:04] [Tool] get_lot_info(W12345) → CD 实测 52.8nm / 目标 50.0nm
[12:01:05] [Tool] get_equipment_log(ETCH-CH-007) → ⚠ 超时
[12:01:05] [Tool] 重试 get_equipment_log(ETCH-CH-007) → 仍超时
[12:01:05] [Planner] 触发降级：跳过机台日志，改用批次历史
[12:01:06] [Tool] get_lot_history(W12345) → 返回近 10 批数据
[12:01:07] [Reflect] ⚡ 发现压力设定(5.0)与历史均值(6.8)偏差 > 1.0mTorr
[12:01:08] [Report] 根因分析完成

4.2 右栏：评估面板

右栏使用 st.metric 展示核心数字卡片，配合状态标签呈现质性判断：

• 过程质量：执行步数 5/6、反思有效性 ✅ 发现矛盾
• 资源成本：工具调用 3 次、模拟 Token ~970
• 系统韧性：重试次数 0、死循环 ✅ 未触发、韧性评分 🟢 高
• 业务价值：✅ 包含可执行建议

底部以 st.json 展示完整 evaluator.to_dict()，便于后续接入自动化评测流水线。

---

五、运行结果分析

通过反复触发分析，系统呈现两条典型路径：

路径 A：常规路径（未触发超时）

批次 W12345 → CD 实测 52.8nm / 目标 50.0nm（超标 +2.8nm）
机台 ETCH-CH-007 压力 6.83mTorr vs 配方设定 5.0mTorr
Reflector → ⚡ 发现压力数据矛盾（Δ = 1.83mTorr > 1.0）
最终报告 → 根因：刻蚀压力偏差导致过刻蚀 → 3 条建议

评估：步数 5/6 | 工具 3 次 | Token ~970 | 韧性=高（未触发超时）

路径 B：超时自愈全链路（W12346 CD 偏小）

批次 W12346 → CD 实测 44.1nm / 目标 45.0nm（偏小 -0.9nm）
get_equipment_log(ETCH-CH-012) → ⚠ 超时（30% 概率命中）
  → 重试 1 次 → 仍超时 → 返回 None
  → Orchestrator 触发 Planner.adjust_plan_skip_equipment()
  → 步骤替换为：get_lot_history（降级）
  → Reflector 基于历史数据统计发现趋势性偏差
最终报告 → 根因结论（基于历史推断） + 2 条建议

评估：步数 6/6 | 工具 4 次 | 重试 1 次 | 韧性=中（已重试但未恢复）

关键观察：在路径 B 中，系统没有崩溃、没有编造机台数据、没有陷入死循环，而是以"降级+历史推断"的方式给出了一个可靠性中等但可用的结论。这种"诚实的降级"比强行给出错误精确答案更接近生产级行为。

---

六、理论到实践的映射总结

下表将本文的理论框架与代码实现一一对应，体现"评估先行"的设计哲学：

理论原则	代码实现位置	具体机制
模块级白盒评估	core/ 包 6 个类	每个模块独立可测，接口清晰
过程质量：避免空转	Orchestrator 死循环检测	签名比对，连续重复即终止
资源成本可控	Evaluator.token_cost_mock	字符级模拟，实时展示
系统韧性：失败自愈	Planner.adjust_plan_skip_equipment	30% 超时 → 重试 → 降级
业务价值兑现	Evaluator.has_business_value	报告必含"建议"字段
分层测试	单文件 MVP + 双路径覆盖	正常路径 + 异常路径均验证
全链路可观测	Streamlit 日志区	每步带时间戳的事件流

---

七、结论与展望

7.1 核心结论

本文验证了以下工程原则：

1. 评估体系应优先于 Prompt 工程：在写任何 Agent 逻辑之前，先定义 Evaluator 的采集点和阈值，所有模块设计都围绕可观测性展开。
2. 白盒架构是工业落地的底线：不引入黑盒框架，意味着你对自己系统的每一个 Token、每一次调用、每一个降级决策拥有完全的控制权。
3. "降级"不是失败，是设计：生产系统的价值不在于永不出错，而在于出错时能否给出一个诚实的、有用的、成本可控的响应。

7.2 后续演进方向

方向	具体内容
接入真实 LLM	将 _generate_report 替换为 DeepSeek / 通义千问调用，保留 Evaluator 不变，用同一套指标对比不同模型
持久化评估日志	Evaluator.to_dict() 追加写入 runs/yyyMMdd_HHmmss.json，支持 A/B 对比不同 Planner 策略
向量化记忆	使用 DashScope Embedding 对历史批次特征向量化，支持相似批次检索替代纯字符串匹配
真实接口对接	mock_data.py 与 toolset.py 保留相同的返回协议，可无缝替换为 MES / EAP / SECS-GEM 真实调用
更多 Agent 角色	新增 reporter.py（报告生成 Agent）、validator.py（建议可执行性审核 Agent）

---

附录：运行方式

# 安装依赖（仅 streamlit）
pip install streamlit>=1.57

# 启动
cd fab_agent_test
python -m streamlit run app.py

浏览器打开 http://localhost:8501，输入默认问题或选择示例问题，点击"开始分析"即可观察完整执行流与实时评估面板。约每 3 次运行会触发一次超时自愈场景，可在右侧面板观察 重试次数 与 韧性评分 的变化。