CodeInterpreterMiddleware 执行流程

CodeInterpreterMiddleware 用于为 Deep Agents 提供了一个可编程的工作空间，让他们可以探索数据、协调工具调用、执行 skill 脚本，并将中间工作排除在模型上下文之外。Deep Agents 编写代码表达意图，然后内存运行时执行该代码并返回相关结果。
适用场景如下表：

需求	用什么
需要循环、分支、重试或聚合结果的小程序	CodeInterpreterMiddleware
从代码中发起多个 tool 调用	CodeInterpreterMiddleware + 程序化 tool calling
跨线程可复用的辅助函数	CodeInterpreterMiddleware + Interpreter Skills

整体架构

Agent Loop
    │
    ├─ before_agent()          ← 恢复 REPL 快照
    │
    ├─ wrap_model_call()       ← 注入 system prompt + 安装 PTC 工具
    │       ↓
    │   [Model 决定调用 eval tool]
    │       ↓
    │   eval(code)             ← 执行 JS
    │
    └─ after_agent()           ← 保存快照 / 清理

---

一、初始化阶段 __init__

CodeInterpreterMiddleware(
    mode="thread",      # 持久化模式
    ptc=["web_search"], # 暴露给 JS 的 Python 工具
    skills_backend=..., # skill 文件后端
)

初始化做三件事：

1. 解析持久化模式（_resolve_persistence_flags）

mode	含义	snapshot	reset_between_calls
thread	跨 turn 持久	✅	❌
turn	仅当前 turn 内持久	❌	❌
call	每次 eval 都重置	❌	✅

2. 构建 _Registry：管理所有 thread 的 QuickJS REPL 槽位。

3. 构建 eval Tool：注册为 self.tools，LangGraph 会把它暴露给模型。

---

二、Turn 开始：before_agent

def before_agent(self, state, runtime):
    payload = state.get("_quickjs_snapshot_payload")  # 从 state 取快照
    repl = self._registry.get(thread_id)
    repl.restore_snapshot(payload)                    # 还原 JS heap

只有 mode="thread" 时才执行，把上一轮序列化的 QuickJS 堆内存还原回来，使变量、函数定义等全局状态跨 turn 存活。

---

三、每次模型调用：wrap_model_call

def wrap_model_call(self, request, handler):
    prompt = self._prepare_for_call(request)  # ← 核心
    return handler(request.override(system_message=...))

_prepare_for_call 做两件事：

① 安装 PTC 工具桥

Python tools (web_search, etc.)
        ↓ install_tools()
QuickJS globalThis.tools.webSearch(input) => Promise<string>

名称转换：web_search → tools.webSearch（snake_case → camelCase）

② 构建并缓存 system prompt addendum

base_prompt（REPL 基础说明）
    +
ptc_prompt（tools.xxx 的 TS 签名，按 exposed_names 缓存）

注入到 system message 里，让模型知道可以调用哪些工具。

---

四、模型调用 eval tool

模型生成 eval(code="...") 调用后，走到 async_eval / sync_eval：

async def async_eval(runtime, code):
    thread_id = _resolve_thread_id(fallback_id)  # 1. 定位线程
    repl = middleware._repl_for_eval(thread_id)   # 2. 取 REPL 槽
    skills = middleware._skills_for_eval(runtime) # 3. 解析可用 skills
    
    outcome = await repl.eval_async(              # 4. 执行 JS
        code,
        skills=skills,
        skills_backend=...,
        outer_runtime=runtime,
    )
    
    if reset_between_calls:
        registry.reset_repl(thread_id)            # 5. call 模式：立即重置
    
    return ToolMessage(format_outcome(outcome))   # 6. 返回结果

thread_id 解析优先级：

LangGraph config["configurable"]["thread_id"]
    ↓ 没有时
middleware 实例级 fallback id（session_xxxxxxxx）

这保证同一个中间件实例内 wrap_model_call 和 eval 看到的是同一个 REPL。

Skills 过滤逻辑（_skills_for_eval）：

for skill in state["skills_metadata"]:
    required_ptc = skill["metadata"]["required-ptc-tools"].split()
    if any(tool not in ptc_config for tool in required_ptc):
        logger.warning(...)   # 跳过，不报错（eval 时宽松）
        continue
    result[skill.name] = skill

注意：before_agent 阶段用 _validate_required_ptc_tools 严格报错，eval 阶段只 warning 跳过。

---

五、Turn 结束：after_agent

def after_agent(self, state, runtime):
    if mode == "call" or mode == "turn":
        registry.evict(thread_id)    # 直接销毁，不保存
        return None
    
    # mode == "thread"
    payload = repl.create_snapshot()         # 序列化 JS heap
    registry.evict(thread_id)               # 释放内存中的 REPL
    return {"_quickjs_snapshot_payload": payload}  # 写入 LangGraph state

快照超过 max_snapshot_bytes 时静默丢弃（_snapshot_update 里处理），下一轮从空 REPL 开始。

---

六、完整 thread 模式时序图

Turn 1                          Turn 2
──────────────────────          ──────────────────────
before_agent                    before_agent
  └─ state 无快照，跳过            └─ 还原 JS heap（x=42 还在）

wrap_model_call                 wrap_model_call
  └─ 注入 system prompt           └─ 注入 system prompt
  └─ 安装 PTC 桥                  └─ 安装 PTC 桥

eval("var x = 42")              eval("x + 1")  → 43
  └─ QuickJS 执行                  └─ 读到上轮的 x

after_agent                     after_agent
  └─ 序列化 heap                  └─ 序列化 heap
  └─ 写入 state                   └─ 写入 state
  └─ 释放内存 REPL slot            └─ 释放内存 REPL slot

---

关键设计决策小结

为什么要 evict 再 snapshot，而不是保持 REPL 常驻？

QuickJS Runtime 有 memory_limit，所有 context 共享。Turn 结束后 evict 释放堆内存，下轮从 snapshot 字节还原，避免多个并发 thread 撑爆内存限制。

为什么 PTC 桥在 wrap_model_call 而不是 before_agent 安装？

wrap_model_call 能拿到当前 request 的 live tool list（上游 middleware 可能过滤了部分工具），确保暴露给 JS 的 tools.* 和模型实际能调用的完全一致。

Skills 的代码它放到哪儿去了？

Skills 代码不写磁盘，全程在内存里流转，最终以虚拟文件系统的形式注册进 QuickJS 模块解析器。

数据流

磁盘 / 远端（BackendProtocol）
    ↓ download_files()
内存 bytes
    ↓ decode UTF-8
files dict { "index.ts": "export async function run()..." }
    ↓ ModuleScope(files)
QuickJS 模块解析器虚拟 FS
    ↓ import("@/skills/my-skill") 触发时
QuickJS 执行上下文（oxidase 实时 strip 类型后执行）

eval 时的安装流程

eval(code) 调用进入 _ThreadREPL.eval_async 后，在真正执行代码前会先做三步：

1. scan_skill_references(code)
   扫描代码里所有 "@/skills/xxx" 字符串字面量

2. aload_skill(metadata, backend)
   对每个引用的 skill：读取目录下所有 .ts/.js → 构建 files dict → 返回 LoadedSkill

3. context.install_scope("@/skills/xxx", scope)
   把 ModuleScope 注册进 QuickJS，之后 import() 就能找到

缓存机制

_ThreadREPL 内部维护一个 install cache。同一 thread 内已加载过的 skill 不会重复 download_files，后续 eval 直接复用已注册的 ModuleScope。