diff --git a/backend/app/engine/nodes/tools.py b/backend/app/engine/nodes/tools.py index b4d8c447..5720c775 100644 --- a/backend/app/engine/nodes/tools.py +++ b/backend/app/engine/nodes/tools.py @@ -135,6 +135,12 @@ async def run_code(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 要了硬件隔离却没拿到，必须说出来——否则用户以为自己在 VM 里跑 ctx.emit(EventType.LOG, level="warn", message=f"节点要求 strict 隔离，但 microVM 未就绪，实际用了 {result.backend}") + if result.session_reset: + # 会话工作区被清空了。同一会话的下游节点会突然读不到自己写的文件， + # 不说一声的话那个 FileNotFoundError 根本无从追查 + ctx.emit(EventType.LOG, level="warn", + message="沙箱会话已重置，此前在该会话写入的文件已丢失" + "（多为超时重建，或同会话中各节点的内存限额不一致）") ctx.emit( EventType.SANDBOX_END, ok=result.ok, diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index c9cdfcdf..5dba5e5c 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -380,6 +380,15 @@ class RunManager: elapsed = int((time.perf_counter() - started) * 1000) await self._finalize(run_id, status, error, final_state, elapsed) self._tasks.pop(run_id, None) + # 运行到终态就把沙箱会话收掉。thread_id 每个 run 都是新的， + # 不收的话每跑一次带代码节点的图就多一台常驻 microVM（几百 MB）， + # 而且没有任何自动回收路径会碰它——闲置回收只在"下次有人执行 + # 代码"时才触发。interrupted 要留着，人回来还得接着跑。 + if status in ("succeeded", "failed", "cancelled"): + with contextlib.suppress(Exception): + from app.sandbox.manager import sandbox_manager + + await sandbox_manager.cleanup(run_id) async def _handle_custom(self, run_id: str, chunk: Any) -> None: """节点通过 get_stream_writer 发来的事件。""" diff --git a/backend/app/main.py b/backend/app/main.py index 66e3a428..9f0b91f4 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -28,6 +28,9 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: await init_db() await seed_defaults() await run_manager.setup() + from app.sandbox.manager import sandbox_manager as _sbm + + await _sbm.start_reaper() logger.info("AgentLab 就绪 · 数据目录 %s", settings.data_dir) try: yield diff --git a/backend/app/memory/embeddings.py b/backend/app/memory/embeddings.py index 73d3e34a..380d5d9f 100644 --- a/backend/app/memory/embeddings.py +++ b/backend/app/memory/embeddings.py @@ -179,7 +179,11 @@ def hybrid_rank( def _norm(xs: list[float]) -> list[float]: lo, hi = min(xs), max(xs) if hi - lo < 1e-9: - return [0.0 for _ in xs] + # 全都一样分，说明这一路信号区分不出高下——那是"都一样好"， + # 不是"都一样差"。返回 0 会让候选只有一条时它的合并分恒为 0， + # 被 min_score 过滤掉，于是新建 scope 写入第一条后永远搜不到， + # 而"先写一条、下一轮再 recall"正是 agent 最常见的用法。 + return [1.0 for _ in xs] return [(x - lo) / (hi - lo) for x in xs] kw_n, vec_n = _norm(kw_scores), _norm(vec_scores) diff --git a/backend/app/memory/store.py b/backend/app/memory/store.py index 387695f5..6f5e8fe6 100644 --- a/backend/app/memory/store.py +++ b/backend/app/memory/store.py @@ -7,7 +7,11 @@ from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import MemoryItem -from app.memory.embeddings import embed_text, from_blob, hybrid_rank, to_blob +from app.memory.embeddings import cosine, embed_text, from_blob, hybrid_rank, to_blob + +# 判重阈值：余弦相似度高于它才算同一件事。比排名分数严格得多， +# 因为这里比的是真实语义距离，不是"谁排第一"。 +_DEDUPE_THRESHOLD = 0.92 async def remember( @@ -26,17 +30,33 @@ async def remember( 反复记下来，不去重的话记忆库几轮就被同义句灌满了。 """ content = content.strip() - if dedupe: - existing = await recall(session, scope=scope, query=content, limit=1, min_score=0.0) - if existing and existing[0]["score"] > 0.92: - item = await session.get(MemoryItem, existing[0]["id"]) - if item: - item.importance = max(item.importance, importance) - item.use_count += 1 - await session.commit() - return item - vec = await embed_text(content) + + if dedupe: + # 判重必须看**真实相似度**，不能用 recall 的分数。recall 走 hybrid_rank， + # 那里的分数是 min-max 排名归一化的结果：排第一的那条在两路信号上都会 + # 被归成 1.0，与它和查询像不像毫无关系。于是 scope 里只要已有两条记忆， + # 新内容随便撞上一个字排到第一，就会被判成"重复"而丢弃，接口还返回 201 + # 并把那条不相关的旧记忆当成"刚保存的"交回去——静默丢数据。 + stmt = select(MemoryItem).where(MemoryItem.scope == scope) + if kind: + stmt = stmt.where(MemoryItem.kind == kind) + candidates = list((await session.execute(stmt)).scalars()) + best, best_score = None, 0.0 + for row in candidates: + other = from_blob(row.embedding) + if other is None: + continue + score = cosine(vec, other) + if score > best_score: + best, best_score = row, score + if best is not None and best_score > _DEDUPE_THRESHOLD: + best.importance = max(best.importance, importance) + best.use_count += 1 + await session.commit() + await session.refresh(best) + return best + item = MemoryItem( scope=scope, kind=kind, diff --git a/backend/app/sandbox/base.py b/backend/app/sandbox/base.py index b60f4ce6..92741271 100644 --- a/backend/app/sandbox/base.py +++ b/backend/app/sandbox/base.py @@ -43,6 +43,11 @@ class ExecResult(BaseModel): backend: str = "" files_written: list[str] = Field(default_factory=list) error: str | None = None + # 这次执行导致会话工作区被重置了（比如 microVM 因超时或限额变更被销毁）。 + # 会话契约说"同 session_id 共享文件系统，可以先写文件下一步再读回来"， + # 所以工作区没了必须说出来——否则下一个节点只会看到 FileNotFoundError， + # 完全不知道是谁把文件弄丢的。 + session_reset: bool = False def summary(self) -> str: if self.timed_out: diff --git a/backend/app/sandbox/manager.py b/backend/app/sandbox/manager.py index e197c82e..268c28e9 100644 --- a/backend/app/sandbox/manager.py +++ b/backend/app/sandbox/manager.py @@ -56,6 +56,7 @@ class SandboxManager: self._named: dict[str, Sandbox] = {} # 按名字缓存，供节点级档位使用 self._lock = asyncio.Lock() self._resolved_from: str = "" + self._reaper: asyncio.Task[None] | None = None async def get(self, prefer: str | None = None) -> Sandbox: """取后端。prefer 指定档位时给那一档，拿不到就退回默认那台。 @@ -153,6 +154,32 @@ class SandboxManager: files=files, ) + async def start_reaper(self) -> None: + """起一个后台任务定期回收闲置会话。 + + 以前回收只挂在 run() 里，也就是说"没人再跑代码"时闲置的 microVM + 永远不会被收掉，microvm_idle_seconds 形同虚设——几百 MB 一台就这么 + 挂着。回收本来也不该占着请求路径的时间。 + """ + if self._reaper is not None: + return + + async def _loop() -> None: + interval = max(30, settings.microvm_idle_seconds // 4) + while True: + try: + await asyncio.sleep(interval) + for impl in list(self._named.values()) + ([self._backend] if self._backend else []): + reap = getattr(impl, "_reap_idle", None) + if reap is not None: + await reap() + except asyncio.CancelledError: + raise + except Exception: # noqa: BLE001 - 回收失败不该让任务停掉 + continue + + self._reaper = asyncio.create_task(_loop()) + async def health(self) -> dict[str, Any]: backend = await self.get() info = await backend.health() @@ -178,6 +205,9 @@ class SandboxManager: await impl.cleanup(session_id) async def close(self) -> None: + if self._reaper is not None: + self._reaper.cancel() + self._reaper = None for impl in {id(b): b for b in [*self._named.values(), self._backend] if b}.values(): await impl.close() diff --git a/backend/app/sandbox/microvm_sandbox.py b/backend/app/sandbox/microvm_sandbox.py index be4c54c9..b271c7a0 100644 --- a/backend/app/sandbox/microvm_sandbox.py +++ b/backend/app/sandbox/microvm_sandbox.py @@ -45,6 +45,8 @@ class _Lease: memory_mb: int cpus: int last_used: float = field(default_factory=time.monotonic) + # 这台是因为限额变更重建出来的，也就是说上一次的会话文件已经没了 + was_reset: bool = False class MicroVMSandbox(Sandbox): @@ -160,6 +162,7 @@ class MicroVMSandbox(Sandbox): cpus = max(1, int(limits.cpus)) async with self._lock: + reset = False lease = self._leases.get(session_id) if lease is not None: if (lease.network == limits.network @@ -167,8 +170,11 @@ class MicroVMSandbox(Sandbox): and lease.cpus == cpus): lease.last_used = time.monotonic() return lease - # 限额变了：旧 VM 的边界已经不对，销毁重建 + # 限额变了：VM 的内存/CPU 是启动时定死的，只能销毁重建。 + # 但重建等于把会话工作区清空——同一个 run 里两个代码节点各填了 + # 不同的 memory_mb 就会触发，而调用方对此毫无察觉。标记出来。 await self._destroy(session_id) + reset = True # network 必须是 Network 实例（传 bool 会被 SDK 拒掉） network = ms.Network.allow_all() if limits.network else ms.Network.none() @@ -189,6 +195,7 @@ class MicroVMSandbox(Sandbox): lease = _Lease(vm=vm, network=limits.network, memory_mb=limits.memory_mb, cpus=cpus) + lease.was_reset = reset self._leases[session_id] = lease return lease @@ -201,6 +208,29 @@ class MicroVMSandbox(Sandbox): except Exception: # noqa: BLE001 - 已经没了就算 pass + async def _kill_leftovers(self, session_id: str, entry: str) -> bool: + """超时后在 VM 内把跑飞的进程杀掉，保住会话工作区。 + + 返回 True 表示 VM 还能用；False 表示没收拾干净，调用方该销毁重建。 + """ + lease = self._leases.get(session_id) + if lease is None: + return False + try: + # 只杀这次的入口脚本引出来的进程——入口名带唯一后缀， + # 不会误伤同会话里别的并发执行 + await asyncio.wait_for( + lease.vm.shell(f"pkill -9 -f {entry} || true", timeout=10), + timeout=15, + ) + # 确认 VM 还答应 + out = await asyncio.wait_for( + lease.vm.exec("sh", ["-c", "echo alive"], timeout=10), timeout=15 + ) + return "alive" in str(getattr(out, "stdout_text", "") or "") + except Exception: # noqa: BLE001 - 收拾不动就让调用方销毁 + return False + async def _reap_idle(self) -> None: """回收闲置 VM：它们各自占着几百 MB 内存，不能一直挂着。""" cutoff = time.monotonic() - settings.microvm_idle_seconds @@ -281,12 +311,20 @@ class MicroVMSandbox(Sandbox): timeout=limits.timeout, ) except ms.ExecTimeoutError: - # 超时的 VM 里可能还有进程在烧 CPU，状态不可信，直接销毁 - await self._destroy(sid) + # 超时的 VM 里多半还有进程在烧 CPU，得收拾掉。但**先别销毁整台 VM**： + # 会话契约说同 session 共享文件系统（上一步写的文件下一步要读回来）， + # 销毁 VM 等于把工作区连同用户数据一起清空，而另外三个后端超时只 + # killpg、工作目录照留。先试着在 VM 里把残留进程杀干净， + # 实在不行才销毁——并且如实告诉调用方会话被重置了。 + killed = await self._kill_leftovers(sid, entry) + if not killed: + await self._destroy(sid) return ExecResult( ok=False, backend=self.name, timed_out=True, exit_code=124, duration_ms=int((time.perf_counter() - started) * 1000), - error=f"执行超过 {limits.timeout}s 被终止", + session_reset=not killed, + error=f"执行超过 {limits.timeout}s 被终止" + + ("" if killed else "；VM 已重置，该会话之前写的文件已丢失"), ) except Exception as e: # noqa: BLE001 return ExecResult( @@ -337,6 +375,7 @@ class MicroVMSandbox(Sandbox): backend=self.name, duration_ms=int((time.perf_counter() - started) * 1000), files_written=written, + session_reset=lease.was_reset, error=error, )