欢迎光临
我们一直在努力

借鉴“代谢增长论“重写 AI Agent 记忆系统:晶圆厂场景下的三层代谢式记忆架构与 C9S 闭环调参

摘要:当前主流 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 流转
  • 长时记忆(情景/语义):皮层固化,可通过反复回忆强化
  • 遗忘:神经递质浓度随时间衰减,低浓度突触连接被"修剪"
  • 巩固:睡眠期海马体将短时记忆压缩为长时记忆

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

concentration

(

t

)

=

importance

e

λ

t

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

concentration(t)=importanceeλt

1.3 本文工作

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

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

  • 二、理论基础

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

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

    d

    C

    d

    t

    =

    λ

    C

    C

    (

    t

    )

    =

    C

    0

    e

    λ

    t

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

    dtdC=λCC(t)=C0eλt

    其中:

    • C

      0

      C_0

      C0 = 初始浓度,对应记忆的重要性分值(importance)

    • λ

      \\lambda

      λ = 衰减率,对应代谢速率(decay_rate)

    • t

      t

      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() 方法对应生物的"昼夜代谢节律",四步流程严格对应理论:

  • 重算浓度:所有工作记忆条目按当前 decay_rate 重新计算浓度
  • 识别低浓度:浓度 < threshold 的条目进入"待压缩队列"
  • 生成摘要:用"首条 + 末条 + 关键词"生成巩固摘要,写入情景记忆
  • 返回报告:本轮淘汰数、巩固数、总浓度、平均浓度
  • 亮点 3:C9S PID 闭环——首次让 Agent 记忆"自调参"

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

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

    实验数据(15 轮真实对话):

    轮次触发调参old_decaynew_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级故障,需即时响应"}

    降级链(保证工业场景可靠性):

  • LLM 调用成功且 JSON 合法 → 使用 LLM score
  • LLM 失败 / JSON 解析失败 / score 越界 → fallback 规则打分
  • 未注入 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_decaynew_decayold_thrnew_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 步处置建议:

  • 立即响应:暂停进片、确认告警详情
  • 根因排查:RF 匹配器阻抗 → MFC 漂移 → 腔体压力
  • 处置验证:PM 校准、3 片试片
  • 关联追溯:自动关联 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 记忆系统。主要贡献:

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

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

    8.3 未来工作

  • 跨会话持久化:当前工作记忆在 session 结束后丢失,未来用 SQLite/Redis 持久化
  • 多模态记忆:扩展到图像(缺陷扫描图)、时序数据(FDC 曲线)
  • 强化学习替代 PID:用 RL 学习最优 decay/threshold 调整策略
  • 代谢速率个性化:不同知识类别(告警/工艺/良率)使用不同 decay_rate
  • 分布式部署:支持多用户并发,向量库上云(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


    参考文献

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

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

    赞(0)
    未经允许不得转载:171主机测评 » 借鉴“代谢增长论“重写 AI Agent 记忆系统:晶圆厂场景下的三层代谢式记忆架构与 C9S 闭环调参
    分享到: 更多 (0)

    评论 抢沙发

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址