摘要：当前主流 LLM Agent 的"记忆"多以向量数据库 + Top-K 检索为主，存在"信息堆积、过期记忆无法淘汰、参数静态不可调"三大痛点。本文借鉴陈平先生《代谢增长论》中"代谢物浓度随时间衰减"的核心思想，结合神经科学三层记忆模型（工作/情景/语义）与控制论闭环（C9S PID），设计并实现了一个面向半导体晶圆厂场景的代谢式 AI Agent 记忆系统。系统在 Streamlit 上完整落地，包含浓度梯度衰减、回报率淘汰、巩固摘要、LLM 语义重要性评分、三级降级向量库、C9S 自适应调参六大核心机制。本文将系统阐述理论基础、创新亮点、架构设计与代码实现，并给出 15 轮真实对话的 C9S 调参实验数据。
github:https://github.com/BumbleBee-ZDS/dissipative_structure_theory_agent

---

一、引言：为什么 AI Agent 需要一套"代谢系统"？

1.1 现有 Agent 记忆系统的三大痛点

在 LangChain、AutoGPT、MemGPT 等主流框架中，Agent 的记忆实现普遍采用 “向量库 + 检索 + 拼接 Prompt” 范式。这种范式在短对话场景下表现尚可，但在长周期、高密度交互的工业场景（如半导体晶圆厂工程师助手）中暴露三大问题：

痛点	表现	根因
信息堆积	历史对话越积越多，Prompt 越来越长，Token 成本爆炸	缺乏"遗忘"机制
过期记忆干扰	三个月前的设备故障记录干扰当前判断	缺乏"时间衰减"机制
参数静态	decay_rate、threshold 写死，无法随业务负载自适应	缺乏"闭环反馈"机制

1.2 生物学给出的启示

活体生物的神经系统并非"全量记忆硬盘"，而是一套代谢系统：

• 短时记忆（工作记忆）：海马体临时缓存，容量有限，FIFO 流转
• 长时记忆（情景/语义）：皮层固化，可通过反复回忆强化
• 遗忘：神经递质浓度随时间衰减，低浓度突触连接被"修剪"
• 巩固：睡眠期海马体将短时记忆压缩为长时记忆

这套机制的本质是 “代谢物浓度随时间衰减，低浓度被淘汰，高浓度被巩固”——这正是陈平老师《代谢增长论》的核心公式：

$$\text{concentration}(t)=\text{importance}\cdot e^{-\lambda \cdot t}$$

1.3 本文工作

我们以此为理论基石，结合 PID 控制论，构建了一个代谢式 AI Agent 记忆系统，部署于晶圆厂工程师助手场景，主要贡献如下：

1. 理论迁移：首次将《代谢增长论》的浓度梯度公式系统化引入 LLM Agent 记忆管理
2. 三层架构：Working / Episodic / Semantic 三层记忆，对应神经科学已知的三类记忆系统
3. 代谢流闭环：用户输入 → 重要性评分 → 写入工作记忆 → 检索 → 生成回复 → 回复写回记忆 → tick 代谢
4. C9S 闭环调参：用 PID 思想动态调整 decay_rate / threshold，解决"参数静态"痛点
5. 三级降级：ChromaDB → OpenAI Embedding → TF-IDF 三级向量库降级，保证工业场景可用性
6. LLM 语义评分：用 LLM 输出 JSON {score, reason} 替代规则打分，失败自动 fallback

---

二、理论基础

2.1 代谢增长论与浓度梯度公式

陈平先生的代谢增长论将生物代谢过程抽象为：

$$\frac{dC}{dt}=-\lambda C\Rightarrow C(t)=C_0\cdot e^{-\lambda t}$$

其中：

• $C_0$ = 初始浓度，对应记忆的重要性分值（importance）
• $\lambda$ = 衰减率，对应代谢速率（decay_rate）
• $t$ = 时间，对应记忆年龄（age_minutes）

关键洞察：

• 高重要性记忆（如"Lot-A1234 良率掉了 13%"）即使时间流逝，浓度仍高于阈值，被保留
• 低重要性记忆（如"今天天气不错"）浓度迅速衰减到阈值以下，被淘汰
• 这与人类记忆的"情感锚定"现象高度一致

2.2 神经科学三层记忆模型

记忆类型	神经基础	容量	持续时间	系统对应
工作记忆	前额叶皮层	7±2 项	秒~分钟	WorkingMemory（滑动窗口 FIFO）
情景记忆	海马体	有限	小时~天	EpisodicMemory（会话级摘要）
语义记忆	新皮层	近乎无限	永久	SemanticMemory（知识向量库）

记忆巩固（Consolidation）：海马体在睡眠期将工作记忆压缩为情景摘要，再固化为语义知识。我们用 “窗口满 → 生成摘要 → 写入情景记忆 → 清空工作记忆” 模拟这一过程。

2.3 控制论与 C9S 闭环

经典控制论的闭环结构：感知（Observe）→ 决策（Compute）→ 执行（Apply）→ 反馈（Feedback）。

我们将这一思想用于代谢参数的自适应调整，称为 C9S 控制器（C9S = Cybernetic 9-Step Cycle，借鉴 9 步控制循环）：

对话轮次累积 → 每 5 轮触发
   ↓
Observe: 读取滚动 20 轮性能快照（平均浓度/淘汰率/巩固率/命中率）
   ↓
Compute: PID 计算
   - 平均浓度 > 上限 0.30 → 提升 decay_rate（加速淘汰）
   - 平均浓度 < 下限 0.15 → 降低 decay_rate（保留信息）
   - 淘汰率 > 上限 0.40 → 提升 threshold（避免误淘汰）
   - 淘汰率 < 下限 0.10 且浓度高 → 降低 threshold（释放空间）
   ↓
Apply: 钳位后写入 metabolism 引擎
   - decay_rate ∈ [0.01, 0.30]
   - threshold ∈ [0.05, 0.60]

这是把"工业 PID 控制"思想首次引入 LLM Agent 记忆超参数调节的工作之一。

2.4 工业场景：半导体晶圆厂

晶圆厂工程师助手是一个典型的高密度知识场景：

• 15 条核心知识：FDC 告警分级、ETCH/CVD/CMP/LITHO 工艺参数、Lot 追溯、良率规范等
• 三类典型查询：
  • FDC 告警处置（高重要性，含机台 ID + 告警词）
  • 机台借机评估（中重要性，含机台 ID + 借机词）
  • 良率异常排查（高重要性，含 Lot 号 + 百分比）
• 关键实体：机台编号（ETCH-03）、Lot 批次（Lot-A1234）、告警关键词（FDC/超限）

这个场景天然适合代谢式记忆：告警类信息高重要性、需长时保留；闲聊类低重要性、应快速淘汰。

---

三、创新亮点

亮点 1：浓度梯度公式的工程化落地

将抽象的代谢公式 C(t) = importance · exp(-λt) 落地为 Python MemoryItem.concentration() 方法，每条记忆自带"年龄"和"浓度"，让记忆有了"生命"：

def concentration(self, decay_rate: float) -> float:
    return self.importance * exp(-decay_rate * self.age_minutes())

亮点 2：代谢周期四步 tick

每轮对话后触发的 tick() 方法对应生物的"昼夜代谢节律"，四步流程严格对应理论：

1. 重算浓度：所有工作记忆条目按当前 decay_rate 重新计算浓度
2. 识别低浓度：浓度 < threshold 的条目进入"待压缩队列"
3. 生成摘要：用"首条 + 末条 + 关键词"生成巩固摘要，写入情景记忆
4. 返回报告：本轮淘汰数、巩固数、总浓度、平均浓度

亮点 3：C9S PID 闭环——首次让 Agent 记忆"自调参"

传统 Agent 的 decay_rate、threshold 是写死的超参数，依赖人工调优。C9S 控制器让系统具备"自我调节"能力：

• 感知层：PerformanceTracker 滚动 20 轮记录命中率/淘汰率/巩固率/平均浓度
• 决策层：C9SController.compute_adjustment() 基于死区 + P 增益计算调整量
• 执行层：钳位后写入 MetabolismEngine，避免极端值
• 反馈层：下一轮 tick 自然反映调参效果，形成闭环

实验数据（15 轮真实对话）：

轮次	触发调参	old_decay	new_decay	原因
5	是	0.0500	0.0551	avg_conc=0.556>上限
10	是	0.0551	0.0598	avg_conc=0.534>上限
15	是	0.0598	0.0659	avg_conc=0.604>上限

系统在 15 轮内自主将 decay_rate 从 0.05 提升到 0.0659（+31.8%），有效缓解了浓度堆积。

亮点 4：LLM 语义重要性评分 + 规则兜底

V1 用正则规则打分（机台 ID +0.3、告警词 +0.25、Lot 号 +0.2、百分比 +0.15），快但语义粗糙。V2 引入 LLM 评分器，输出严格 JSON：

{"score": 0.85, "reason": "含机台ID ETCH-03 + FDC告警关键词，属L2级故障，需即时响应"}

降级链（保证工业场景可靠性）：

1. LLM 调用成功且 JSON 合法 → 使用 LLM score
2. LLM 失败 / JSON 解析失败 / score 越界 → fallback 规则打分
3. 未注入 provider 或 use_llm_scoring=False → 直接规则打分

亮点 5：三级降级向量库

工业部署中，向量库依赖（chromadb、sentence-transformers）可能因网络/磁盘/权限问题不可用。我们设计了三级降级链：

ChromaDB + paraphrase-multilingual-MiniLM-L12-v2
    ↓ (失败)
OpenAI text-embedding-3-small（兼容 DeepSeek Key）
    ↓ (失败/无 OPENAI_API_KEY)
sklearn TF-IDF + cosine_similarity（V1 实现，永远可用）

每级失败自动降级到下一级，全失败时仍可运行（只是检索质量下降），保证工业场景的"永不宕机"。

亮点 6：全链路状态可视化

Streamlit 仪表盘 5 个 tab：

• 工作记忆：表格 + 浓度柱状图（st.progress）
• 衰减曲线：每条记忆的浓度轨迹 + threshold 水平参考线
• 情景记忆：摘要历史 + 二次巩固标记
• 代谢报告：历轮 tick 报告表
• 效能趋势（V2 新增）：性能指标折线 + C9S 调参轨迹双图

---

四、系统架构

4.1 整体架构

┌─────────────────────────────────────────────────────────────┐
│                    Streamlit UI 层                          │
│   ┌─────────────┐  ┌─────────────────────────────────┐    │
│   │  侧边栏     │  │     主区域（双栏布局）           │    │
│   │  - 参数滑杆  │  │  ┌──────────┐ ┌──────────────┐ │    │
│   │  - V2 配置区 │  │  │ 对话面板 │ │ 仪表盘 5 tab │ │    │
│   │  - 操作按钮  │  │  └──────────┘ └──────────────┘ │    │
│   │  - 实时统计  │  │                                 │    │
│   └─────────────┘  └─────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────┐
│              MetabolicMemoryManager（门面）                 │
│   ┌──────────────────────────────────────────────────┐     │
│   │  ingest() → recall() → tick()  代谢流闭环         │     │
│   └──────────────────────────────────────────────────┘     │
└─────────────────────────────────────────────────────────────┘
        │            │              │              │
        ▼            ▼              ▼              ▼
┌────────────┐ ┌──────────┐ ┌─────────────┐ ┌──────────────┐
│ LLM 评分器  │ │ 三层记忆  │ │ 代谢引擎    │ │ C9S 控制层   │
│ (V2)       │ │          │ │             │ │ (V2)         │
│ - 规则兜底  │ │ Working  │ │ tick 四步   │ │ - Tracker    │
│ - JSON 解析 │ │ Episodic │ │ - 重算浓度  │ │ - Controller │
│ - few-shot │ │ Semantic │ │ - 识别低浓  │ │ - PID 调参   │
└────────────┘ └──────────┘ │ - 摘要巩固  │ └──────────────┘
                              │ - 报告返回  │
                              └─────────────┘

4.2 代谢流闭环

用户输入
   │
   ▼
[ingest] 重要性评分（LLM/规则）→ 写入 WorkingMemory
   │
   ▼
[recall] 三路检索融合
   ├── WorkingMemory（按浓度降序）
   ├── EpisodicMemory（top2 摘要）
   └── SemanticMemory（top3 知识，V2 走向量库）
   │
   ▼
LLM 生成回复
   │
   ▼
[ingest] 回复写回 WorkingMemory（代谢流闭环）
   │
   ▼
[tick] 代谢周期
   ├── 重算浓度
   ├── 低浓度 → 摘要巩固到 Episodic
   ├── PerformanceTracker.observe()
   └── C9S 每 5 轮 apply（V2）

4.3 项目文件结构

wafer_metabolic_mvp/
├── app.py                          # Streamlit 入口（V2 配置区）
├── config.py                       # MetabolicConfig（含 V2 5 开关）
├── requirements.txt                # 含 chromadb / sentence-transformers
├── .env                            # DEEPSEEK_API_KEY
│
├── domain/                         # 业务领域层
│   ├── knowledge.py                # MockKnowledgeBase（15 条晶圆厂知识）
│   └── importance_rules.py         # V1 规则打分（正则）
│
├── llm/
│   ├── provider.py                 # LLMProvider 抽象 + MockLLM + OpenAIProvider
│   └── llm_importance_scorer.py    # ⭐ V2: LLM 重要性评分器
│
├── memory/                         # 记忆层
│   ├── schema.py                   # MemoryItem + MemoryType
│   ├── working_memory.py           # 工作记忆（deque FIFO）
│   ├── episodic_memory.py          # 情景记忆（摘要 + 二次巩固）
│   ├── semantic_memory.py          # 语义记忆（V2 注入向量库）
│   ├── metabolism.py               # 代谢引擎（tick 四步）
│   ├── manager.py                  # 门面（V2 接入 scorer/c9s/vector_store）
│   └── vector_store.py             # ⭐ V2: 三级降级向量库
│
├── control/                        # ⭐ V2 控制层
│   ├── __init__.py
│   ├── c9s_controller.py           # C9S PID 控制器
│   └── performance_tracker.py      # 滚动 20 轮性能追踪
│
├── ui/
│   ├── chat_panel.py               # 聊天面板
│   └── metabolism_dashboard.py     # 仪表盘（V2 增"效能趋势"tab）
│
└── tests/
    └── test_v2_degradation.py      # ⭐ V2 6 套降级测试

---

五、核心实现

5.1 浓度梯度公式落地

memory/schema.py 中的 MemoryItem：

@dataclass
class MemoryItem:
    content: str
    importance: float
    memory_type: MemoryType
    created_at: datetime = field(default_factory=datetime.now)
    role: str = "user"
    metadata: dict = field(default_factory=dict)
    summary: Optional[str] = None

    def age_minutes(self) -> float:
        """记忆年龄（分钟）。"""
        return (datetime.now() - self.created_at).total_seconds() / 60.0

    def concentration(self, decay_rate: float) -> float:
        """浓度梯度公式：concentration = importance * exp(-decay_rate * age)"""
        return self.importance * exp(-decay_rate * self.age_minutes())

5.2 代谢周期四步 tick

memory/metabolism.py 中的 MetabolismEngine.tick()：

def tick(self) -> MetabolismReport:
    report = MetabolismReport()
    report.pre_tick_working = len(self.working)  # V2: 供 C9S 计算淘汰率

    # 步骤1：重算浓度，识别低浓度条目
    to_compress, survivors = [], []
    for item in list(self.working.items):
        conc = item.concentration(self.decay_rate)
        if conc < self.threshold:        # 步骤2：浓度 < threshold → 待压缩
            to_compress.append(item)
        else:
            survivors.append(item)

    # 步骤3：待压缩队列非空 → 生成摘要 → 写入 EpisodicMemory
    if to_compress:
        max_imp = max(it.importance for it in to_compress)
        eliminated_summary = EpisodicMemory.generate_summary(to_compress)
        self.episodic.add_summary(
            eliminated_summary,
            importance=max_imp,
            metadata={"source": "metabolism_tick", "count": len(to_compress)},
        )
        for item in to_compress:
            self.working.remove(item)

    # 步骤4：返回本轮代谢报告
    report.eliminated_count = len(to_compress)
    report.consolidated_count = 1 if to_compress else 0
    report.total_concentration = self.working.total_concentration(self.decay_rate)
    n_remaining = len(self.working)
    report.avg_concentration = (
        report.total_concentration / n_remaining if n_remaining > 0 else 0.0
    )
    return report

5.3 LLM 重要性评分器（V2）

llm/llm_importance_scorer.py 核心逻辑：

class LLMImportanceScorer:
    def __init__(self, provider=None, enabled=False):
        self.provider = provider
        self.enabled = enabled and provider is not None
        # MockLLM 无意义，强制走规则
        if self.enabled and getattr(provider, "name", "") == "MockLLM":
            self.enabled = False

    def score(self, text: str) -> ImportanceResult:
        # 路径1：未启用 LLM → 直接规则
        if not self.enabled:
            return ImportanceResult(
                score=score_importance(text),
                reason="[rules] V1规则打分", source="rules"
            )
        # 路径2：LLM 评分
        try:
            raw = self._call_llm(text)
            result = self._parse_json(raw)
            if result is None:
                raise ValueError(f"JSON解析失败: {raw[:80]}")
            score = float(result.get("score", -1))
            if not (0.0 <= score <= 1.0):
                raise ValueError(f"score越界: {score}")
            return ImportanceResult(
                score=round(score, 4),
                reason=f"[llm] {result.get('reason', '')[:200]}",
                source="llm",
            )
        except Exception as e:
            # 路径3：降级规则
            return ImportanceResult(
                score=score_importance(text),
                reason=f"[rules-fallback] LLM失败({type(e).__name__})",
                source="rules",
            )

    @staticmethod
    def _parse_json(raw: str) -> Optional[dict]:
        """兼容纯JSON / markdown代码块 / 前后多余文本。"""
        if not raw:
            return None
        try:
            return json.loads(raw)
        except json.JSONDecodeError:
            pass
        # 提取 ```json ... ```代码块
        m = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", raw, re.DOTALL)
        if m:
            try:
                return json.loads(m.group(1))
            except json.JSONDecodeError:
                pass
        # 提取首个 {...} 块
        m = re.search(r"\{[^{}]*\"score\"[^{}]*\}", raw, re.DOTALL)
        if m:
            try:
                return json.loads(m.group(0))
            except json.JSONDecodeError:
                pass
        return None

few-shot 示例（晶圆厂场景）：

FEW_SHOT_EXAMPLES = """\
示例1：
输入："ETCH-03 的 RF 功率今天 FDC 超限告警了，怎么办？"
输出：{"score": 0.85, "reason": "含机台ID ETCH-03 + FDC告警关键词，属L2级故障，需即时响应"}

示例2：
输入："今天天气怎么样？"
输出：{"score": 0.10, "reason": "与晶圆厂业务无关的闲聊，重要性最低"}

示例3：
输入："Lot-A1234 这批良率掉了 13%，帮我看看。"
输出：{"score": 0.95, "reason": "含Lot号+良率跌幅>10%，需升级PIE+YE联合分析，高优先级"}
"""

5.4 C9S PID 控制器（V2）

control/c9s_controller.py：

# 安全钳位区间
DECAY_MIN, DECAY_MAX = 0.01, 0.30
THRESHOLD_MIN, THRESHOLD_MAX = 0.05, 0.60

# 健康区间
TARGET_CONC_RANGE = (0.15, 0.30)
TARGET_ELIM_RANGE = (0.10, 0.40)

class C9SController:
    def compute_adjustment(self) -> C9SAdjustment:
        snap = self.observe()
        avg_conc = snap.get("avg_concentration", 0.0)
        avg_elim = snap.get("avg_eliminate_rate", 0.0)
        n = snap.get("n", 0)

        new_decay = self.decay_rate
        new_threshold = self.threshold
        reasons = []

        if n >= 2:  # 至少 2 轮数据才调
            conc_lo, conc_hi = TARGET_CONC_RANGE
            if avg_conc > conc_hi + DEAD_ZONE_CONC:
                delta = KP_DECAY * (avg_conc - conc_hi)
                new_decay = self._clamp_decay(self.decay_rate + delta)
                if new_decay != self.decay_rate:
                    reasons.append(f"avg_conc={avg_conc:.3f}>上限,decay+{delta:.4f}")
            elif avg_conc < conc_lo - DEAD_ZONE_CONC:
                delta = KP_DECAY * (conc_lo - avg_conc)
                new_decay = self._clamp_decay(self.decay_rate - delta)
                if new_decay != self.decay_rate:
                    reasons.append(f"avg_conc={avg_conc:.3f}<下限,decay-{delta:.4f}")

            elim_lo, elim_hi = TARGET_ELIM_RANGE
            if avg_elim > elim_hi + DEAD_ZONE_ELIM:
                delta = KP_THRESHOLD * (avg_elim - elim_hi)
                new_threshold = self._clamp_threshold(self.threshold + delta)
                if new_threshold != self.threshold:
                    reasons.append(f"elim_rate={avg_elim:.3f}>上限,thr+{delta:.4f}")
            elif avg_elim < elim_lo - DEAD_ZONE_ELIM and avg_conc > conc_hi:
                delta = KP_THRESHOLD * (elim_lo - avg_elim)
                new_threshold = self._clamp_threshold(self.threshold - delta)
                if new_threshold != self.threshold:
                    reasons.append(f"elim_rate={avg_elim:.3f}<下限且conc高,thr-{delta:.4f}")

        return C9SAdjustment(
            tick_idx=self.tick_count,
            old_decay=self.decay_rate, new_decay=new_decay,
            old_threshold=self.threshold, new_threshold=new_threshold,
            reason="; ".join(reasons) if reasons else "指标在健康区间，无需调参",
            applied=False,
        )

    def tick(self) -> Optional[C9SAdjustment]:
        """每轮对话后调用：累计计数，到间隔则计算+应用。"""
        self.tick_count += 1
        if not self.should_adjust():  # 每 5 轮
            return None
        adj = self.compute_adjustment()
        return self.apply(adj)

5.5 三级降级向量库（V2）

memory/vector_store.py：

class VectorStore:
    def __init__(self, persist_path="./.chroma_cache",
                 collection_name="wafer_semantic",
                 openai_api_key=None, openai_base_url=None):
        self.backend = "disabled"
        # 优先级1：ChromaDB + sentence-transformers
        if self._init_chroma(openai_api_key, openai_base_url):
            return
        # 优先级2：OpenAI text-embedding-3-small
        if self._init_openai(openai_api_key, openai_base_url):
            return
        # 优先级3：TF-IDF
        if self._init_tfidf():
            return
        self.backend = "disabled"

    def _init_chroma(self, ...):
        try:
            import chromadb
            from sentence_transformers import SentenceTransformer
            self._chroma_client = chromadb.PersistentClient(
                path=self.persist_path,
                settings=Settings(anonymized_telemetry=False, allow_reset=True),
            )
            self._collection = self._chroma_client.get_or_create_collection(
                name=self.collection_name, metadata={"hnsw:space": "cosine"}
            )
            # 多语言模型（首次下载约 470MB）
            self._embed_fn = SentenceTransformer(
                "paraphrase-multilingual-MiniLM-L12-v2"
            )
            self.backend = "chroma"
            return True
        except Exception as e:
            self._last_error = f"chroma_init_failed: {type(e).__name__}"
            return False

SemanticMemory.search() 的回退逻辑：

def search(self, query: str, top_k: int = 3):
    if not self.documents:
        return []
    # V2 优先走向量库
    if self.vector_store is not None and self.vector_store.backend != "disabled":
        try:
            vs_results = self.vector_store.search(query, top_k=top_k)
            if vs_results:
                return [(r.doc, r.score) for r in vs_results]
        except Exception:
            pass  # 降级 TF-IDF
    # V1 TF-IDF 路径
    return self._search_tfidf(query, top_k)

5.6 性能追踪器（V2）

control/performance_tracker.py：

@dataclass
class RoundMetrics:
    round_idx: int
    recall_hit_rate: float = 0.0      # 检索命中率
    eliminate_rate: float = 0.0       # 淘汰率
    consolidate_rate: float = 0.0     # 巩固率
    avg_concentration: float = 0.0    # 平均浓度
    total_concentration: float = 0.0  # 总浓度
    scorer_source: str = "rules"      # 评分来源

class PerformanceTracker:
    def __init__(self, window: int = 20):
        self._buf: deque[RoundMetrics] = deque(maxlen=window)

    def snapshot(self) -> dict:
        if not self._buf:
            return {"n": 0, "avg_recall_hit": 0.0, ...}
        n = len(self._buf)
        return {
            "n": n,
            "avg_recall_hit": sum(m.recall_hit_rate for m in self._buf) / n,
            "avg_eliminate_rate": sum(m.eliminate_rate for m in self._buf) / n,
            "avg_consolidate_rate": sum(m.consolidate_rate for m in self._buf) / n,
            "avg_concentration": sum(m.avg_concentration for m in self._buf) / n,
            "llm_score_ratio": sum(1 for m in self._buf if m.scorer_source == "llm") / n,
        }

---

六、实验与验证

6.1 V2 降级测试套件

我们编写了 6 套降级测试，覆盖所有失败路径：

测试	覆盖场景	结果
1. LLM 评分器降级链	未启用 / MockLLM / LLM 失败 / 非 JSON / 合法 JSON / score 越界	6/6 通过
2. VectorStore 三级降级	chromadb → openai → tfidf → disabled	通过
3. SemanticMemory 回退	vector_store disabled → TF-IDF	通过
4. C9S 钳位 + 调参	初始钳位 / 调参触发 / 越界钳位	通过
5. V1 等价冒烟	V2 开关全关，行为等价 V1	通过
6. C9S 15 轮对话序列	15 轮触发 3 次调参（轮 5/10/15）	通过

6.2 C9S 15 轮真实调参轨迹

输入序列（混合高/低重要性）：

1. ETCH-03 FDC 超限告警          (高)
2. 今天天气不错                  (低)
3. Lot-A1234 良率掉了 13%        (高)
4. CVD-07 能借给 PE 吗           (中)
5. hello                        (极低)
6. CMP-05 终点漂移告警           (高)
7. 午饭吃什么                    (低)
8. LITHO-01 Overlay 超 3σ       (高)
9. WET-01 颗粒数异常             (中)
10. thank you                   (极低)
11. PVD-02 靶材寿命 90%          (中)
12. Lot-B5678 良率 88%           (高)
13. ETCH-03 又告警了             (高)
14. 下班啦                       (低)
15. SPC 控制限超 3σ 需停机        (高)

调参结果：

轮	old_decay	new_decay	old_thr	new_thr	原因
5	0.0500	0.0551	0.1500	0.1500	avg_conc=0.556>上限,decay+0.0051
10	0.0551	0.0598	0.1500	0.1500	avg_conc=0.534>上限,decay+0.0047
15	0.0598	0.0659	0.1500	0.1500	avg_conc=0.604>上限,decay+0.0061

性能快照（滚动 15 轮）：

指标	数值
平均检索命中率	0.60
平均淘汰率	0.13
平均巩固率	0.33
平均浓度	0.6035
LLM 评分占比	0.0（MockLLM 测试）

6.3 真实 DeepSeek LLM 调用验证

启用 DeepSeek API（DEEPSEEK_API_KEY）后，对话"ETCH-03 的 RF 功率今天 FDC 超限告警了，怎么办？"返回结构化 4 步处置建议：

1. 立即响应：暂停进片、确认告警详情
2. 根因排查：RF 匹配器阻抗 → MFC 漂移 → 腔体压力
3. 处置验证：PM 校准、3 片试片
4. 关联追溯：自动关联 Lot-A1234 的历史告警记忆

LLM 主动调用了知识库中的 KB001（ETCH-03 SOP）和 KB009（Lot-A1234 追溯记录），证明三路检索融合有效。

---

七、运行与部署

7.1 依赖安装

pip install -r requirements.txt

requirements.txt：

streamlit>=1.30
scikit-learn>=1.3
numpy>=1.24
pandas>=2.0
openai>=1.10
python-dotenv>=1.0
# V2 新增（可选，缺失时自动降级到 V1 TF-IDF / 规则评分）
chromadb>=0.4.22
sentence-transformers>=2.3.0

7.2 配置 API Key

在项目根目录创建 .env：

DEEPSEEK_API_KEY=sk-your-deepseek-key
# 或
OPENAI_API_KEY=sk-your-openai-key

未配置时自动降级 MockLLM（规则+模板+检索拼接），仍可完整运行。

7.3 启动

cd wafer_metabolic_mvp
streamlit run app.py --server.port 8513

浏览器访问 http://localhost:8513/

7.4 V2 开关说明

侧边栏"V2 高级配置"折叠区提供 3 个开关：

开关	默认	说明
启用 LLM 重要性评分	关	用 LLM 语义评分替代规则，失败自动降级
启用 C9S 闭环调参	关	每 5 轮 PID 调整 decay/threshold
启用向量库	关	chromadb/openai/tfidf 三级降级（需重启生效）

关闭所有开关时，系统行为完全等价于 V1，保证向后兼容。

界面展示：

---

八、总结与展望

8.1 工作总结

本文将代谢增长论的浓度梯度公式首次系统化引入 LLM Agent 记忆管理，结合神经科学三层记忆模型与控制论 C9S 闭环，设计并实现了一个面向晶圆厂场景的代谢式 AI Agent 记忆系统。主要贡献：

1. 理论迁移：浓度梯度公式 → MemoryItem.concentration()，让记忆"有生命"
2. 架构创新：三层记忆 + 代谢流闭环 + 巩固摘要，对应神经科学已知机制
3. 自适应调参：C9S PID 控制器，首次让 Agent 记忆超参数"自调参"
4. 工程鲁棒：LLM 评分 + 向量库 + 规则三层降级链，工业场景"永不宕机"
5. 可视化：5 tab 仪表盘 + 双图效能趋势，全链路状态可观测

8.2 创新点对照表

创新点	传统 Agent	本系统
记忆淘汰	无（全量保留）	浓度衰减 + 阈值淘汰
时间敏感性	无	浓度梯度公式 exp(-λt)
参数调节	人工静态	C9S PID 自适应
重要性评分	规则或无	LLM 语义 + 规则兜底
向量库	单一依赖	三级降级链
记忆巩固	无	窗口满 → 摘要 → 情景记忆

8.3 未来工作

1. 跨会话持久化：当前工作记忆在 session 结束后丢失，未来用 SQLite/Redis 持久化
2. 多模态记忆：扩展到图像（缺陷扫描图）、时序数据（FDC 曲线）
3. 强化学习替代 PID：用 RL 学习最优 decay/threshold 调整策略
4. 代谢速率个性化：不同知识类别（告警/工艺/良率）使用不同 decay_rate
5. 分布式部署：支持多用户并发，向量库上云（Pinecone/Weaviate）

8.4 开源与复现

本项目完整代码已开源（含 6 套降级测试），运行步骤：

git clone <repo>
cd wafer_metabolic_mvp
pip install -r requirements.txt
echo "DEEPSEEK_API_KEY=sk-xxx" > .env
streamlit run app.py

---

参考文献

1. 陈平. 《代谢增长论：复杂经济学的基本框架》. 经济科学出版社, 2019.
2. Tulving E. Episodic and semantic memory. Organization of Memory, 1972.
3. Baddeley A. Working memory. Science, 1992.
4. Wiener N. Cybernetics: Or Control and Communication in the Animal and the Machine. MIT Press, 1948.
5. Lewis P et al. Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks. NeurIPS, 2020.
6. Park J et al. Generative Agents: Interactive Simulacra of Human Behavior. UIST, 2023.

---

如果你觉得这篇文章对你有启发，欢迎点赞、收藏、关注一键三连！项目代码已开源，欢迎 Star 和 Issue 交流。