|
|
@@ -0,0 +1,1508 @@
|
|
|
+# LambdagentPaaS 工程化改善分析
|
|
|
+
|
|
|
+> 对标 Claude Code 的生产级实现,逐模块诊断 lambdagentpaas 最亟待改善的工程短板。
|
|
|
+> 按紧急程度分 P0 (必须) / P1 (应该) / P2 (建议) 三档。
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 一、总体诊断:理论优雅,工程脆弱
|
|
|
+
|
|
|
+lambdagentpaas 拥有**同类项目中最好的理论基础** (Lambda 演算 isomorphism, 60/60 Church 编码验证),但在工程鲁棒性上与 Claude Code 存在代际差距:
|
|
|
+
|
|
|
+| 维度 | Claude Code | lambdagentpaas | 差距等级 |
|
|
|
+|------|-------------|----------------|---------|
|
|
|
+| 流式架构 | 全链路 Generator streaming | 全链路同步阻塞 | **致命** |
|
|
|
+| 超时控制 | 每个工具/API 调用独立超时 | 几乎无超时 | **致命** |
|
|
|
+| 错误恢复 | 重试+降级+fallback model | 单次调用,失败即终止 | **严重** |
|
|
|
+| 并发安全 | isConcurrencySafe + AbortController 层级 | Par 是假并行,AsyncPar 有线程安全问题 | **严重** |
|
|
|
+| 上下文管理 | Auto-compact + History snip + Microcompact | Sliding window (仅 ReAct) | **严重** |
|
|
|
+| 取消机制 | AbortController 层级,WeakRef 防泄漏 | 无取消机制 | **严重** |
|
|
|
+| **文件隔离** | **Git Worktree 独立目录+分支 per agent** | **无任何文件隔离,所有 Agent 共享 CWD** | **严重** |
|
|
|
+| 内存管理 | LRU cache + 磁盘持久化 + eviction | 无上限,trace 无限增长 | **中等** |
|
|
|
+| 工具生态 | 60+ 内置工具,每个有输入校验/权限/渲染 | shell + MCP wrapper,无校验 | **中等** |
|
|
|
+| 可观测性 | GrowthBook A/B + Telemetry + 结构化日志 | 基础 print trace | **中等** |
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 二、P0 — 致命短板 (不解决无法生产使用)
|
|
|
+
|
|
|
+### P0-1: 全链路异步化 + 流式输出
|
|
|
+
|
|
|
+**现状问题**:
|
|
|
+
|
|
|
+`executor.py` 的 `reduce()` 是同步函数,LLM 调用阻塞整个线程:
|
|
|
+
|
|
|
+```python
|
|
|
+# executor.py - 当前实现
|
|
|
+def _reduce_lam(self, lam, input_val, ctx):
|
|
|
+ result = self.llm_adapter.call(...) # 阻塞 30-60 秒
|
|
|
+ return result # 必须等完整响应
|
|
|
+```
|
|
|
+
|
|
|
+**后果**:
|
|
|
+- 10 步 ReAct loop × 30 秒/步 = 5 分钟用户看不到任何输出
|
|
|
+- PaaS 服务中一个请求阻塞一个 Worker 线程
|
|
|
+- 无法实现"边生成边执行工具"
|
|
|
+
|
|
|
+**Claude Code 做法**:
|
|
|
+
|
|
|
+```typescript
|
|
|
+// query.ts - Generator-based streaming
|
|
|
+async *query(params): AsyncGenerator<Message | StreamEvent> {
|
|
|
+ for await (const event of queryWithModel(...)) {
|
|
|
+ // content_block_delta → 立即 yield 给 UI
|
|
|
+ // tool_use block 还没写完就开始准备执行
|
|
|
+ yield event
|
|
|
+ }
|
|
|
+}
|
|
|
+```
|
|
|
+
|
|
|
+**改善方案**:
|
|
|
+
|
|
|
+```python
|
|
|
+# 目标:全链路 async generator
|
|
|
+class Executor:
|
|
|
+ async def reduce(self, term, input_val, ctx) -> AsyncGenerator[Event, None]:
|
|
|
+ if isinstance(term, Lam):
|
|
|
+ async for token in self.llm_adapter.stream(model, prompt, input_text):
|
|
|
+ yield StreamEvent(type='token', data=token)
|
|
|
+ # 工具调用在流中解析,不等完整响应
|
|
|
+ elif isinstance(term, Compose):
|
|
|
+ for stage in term.stages:
|
|
|
+ async for event in self.reduce(stage, result, ctx):
|
|
|
+ yield event
|
|
|
+
|
|
|
+# LLMAdapter 添加流式接口
|
|
|
+class LLMAdapter:
|
|
|
+ async def stream(self, model, system, user, **kwargs) -> AsyncGenerator[str, None]:
|
|
|
+ if 'claude' in model:
|
|
|
+ async with client.messages.stream(...) as stream:
|
|
|
+ async for text in stream.text_stream:
|
|
|
+ yield text
|
|
|
+```
|
|
|
+
|
|
|
+**工作量**:大 (需重构 Executor + LLMAdapter + ReActEngine + 所有 Term.apply)
|
|
|
+**优先级**:P0 — 不做流式,用户体验和服务吞吐都不可接受
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### P0-2: 超时 + 重试 + 断路器
|
|
|
+
|
|
|
+**现状问题**:
|
|
|
+
|
|
|
+几乎所有 I/O 操作都没有超时和重试:
|
|
|
+
|
|
|
+```python
|
|
|
+# llm_adapter.py - 无超时
|
|
|
+response = self.client.messages.create(...) # 可以挂起无限期
|
|
|
+
|
|
|
+# mcp_client.py - 10 秒硬编码超时,无断路器
|
|
|
+req = urllib.request.Request(url, data=payload)
|
|
|
+resp = urllib.request.urlopen(req, timeout=10) # MCP 服务器慢就失败
|
|
|
+
|
|
|
+# primitives.py - Lam._call_llm() 对 DashScope/Ollama 有 120 秒硬编码
|
|
|
+urllib.request.urlopen(req, timeout=120) # 不可配置
|
|
|
+```
|
|
|
+
|
|
|
+**后果**:
|
|
|
+- 网络抖动 → 任务彻底失败
|
|
|
+- MCP 服务器慢 → 整个 Agent 卡死
|
|
|
+- 无法区分"暂时故障"和"永久故障"
|
|
|
+
|
|
|
+**Claude Code 做法**:
|
|
|
+
|
|
|
+```typescript
|
|
|
+// claude.ts
|
|
|
+categorizeRetryableAPIError(error) // 区分 transient vs permanent
|
|
|
+withRetry(fn, { maxRetries: 3, baseDelay: 1000 }) // 指数退避
|
|
|
+FallbackTriggeredError → 自动切换 fallback model
|
|
|
+checkQuotaStatus() → 速率限制
|
|
|
+```
|
|
|
+
|
|
|
+**改善方案**:
|
|
|
+
|
|
|
+```python
|
|
|
+import asyncio
|
|
|
+from dataclasses import dataclass
|
|
|
+
|
|
|
+@dataclass
|
|
|
+class RetryPolicy:
|
|
|
+ max_attempts: int = 3
|
|
|
+ base_delay: float = 1.0
|
|
|
+ max_delay: float = 30.0
|
|
|
+ jitter: bool = True
|
|
|
+ retryable_errors: tuple = (TimeoutError, ConnectionError, HTTPError)
|
|
|
+
|
|
|
+async def with_retry(fn, policy: RetryPolicy):
|
|
|
+ for attempt in range(policy.max_attempts):
|
|
|
+ try:
|
|
|
+ return await asyncio.wait_for(fn(), timeout=timeout)
|
|
|
+ except policy.retryable_errors as e:
|
|
|
+ if attempt == policy.max_attempts - 1:
|
|
|
+ raise
|
|
|
+ delay = min(policy.base_delay * (2 ** attempt), policy.max_delay)
|
|
|
+ if policy.jitter:
|
|
|
+ delay *= (0.5 + random.random())
|
|
|
+ await asyncio.sleep(delay)
|
|
|
+
|
|
|
+class CircuitBreaker:
|
|
|
+ """MCP/LLM 调用断路器"""
|
|
|
+ def __init__(self, failure_threshold=5, reset_timeout=60):
|
|
|
+ self.failures = 0
|
|
|
+ self.state = 'closed' # closed → open → half_open
|
|
|
+ self.last_failure_time = 0
|
|
|
+
|
|
|
+ async def call(self, fn):
|
|
|
+ if self.state == 'open':
|
|
|
+ if time.time() - self.last_failure_time > self.reset_timeout:
|
|
|
+ self.state = 'half_open'
|
|
|
+ else:
|
|
|
+ raise CircuitOpenError("Circuit breaker is open")
|
|
|
+ try:
|
|
|
+ result = await fn()
|
|
|
+ self._record_success()
|
|
|
+ return result
|
|
|
+ except Exception as e:
|
|
|
+ self._record_failure()
|
|
|
+ raise
|
|
|
+```
|
|
|
+
|
|
|
+**工作量**:中
|
|
|
+**优先级**:P0 — 没有重试和超时,任何网络问题都导致任务失败
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### P0-3: 取消机制 (AbortController 等价物)
|
|
|
+
|
|
|
+**现状问题**:
|
|
|
+
|
|
|
+一旦 Agent 开始执行,**无法取消**:
|
|
|
+
|
|
|
+```python
|
|
|
+# 当前:GroupChat 开始后无法中断
|
|
|
+chat = GroupChat(agents, max_rounds=100)
|
|
|
+result = chat.apply(input, ctx) # 如果每轮 30 秒,可能运行 50 分钟
|
|
|
+# 用户只能 Ctrl+C 杀进程
|
|
|
+```
|
|
|
+
|
|
|
+**Claude Code 做法**:
|
|
|
+
|
|
|
+```typescript
|
|
|
+// 层级化取消
|
|
|
+Main AbortController
|
|
|
+├── Agent1 AbortController (child, WeakRef 防泄漏)
|
|
|
+│ └── Tool AbortController (child)
|
|
|
+│ └── Bash error → siblingAbortController.abort()
|
|
|
+└── Agent2 AbortController (child)
|
|
|
+
|
|
|
+// 用户按 ESC → parent.abort() → 所有子任务优雅终止
|
|
|
+```
|
|
|
+
|
|
|
+**改善方案**:
|
|
|
+
|
|
|
+```python
|
|
|
+import asyncio
|
|
|
+
|
|
|
+class CancellationToken:
|
|
|
+ """层级化取消 token,等价于 AbortController"""
|
|
|
+ def __init__(self, parent: 'CancellationToken' = None):
|
|
|
+ self._cancelled = False
|
|
|
+ self._reason = None
|
|
|
+ self._children: list[weakref.ref] = []
|
|
|
+ self._callbacks: list[callable] = []
|
|
|
+ if parent:
|
|
|
+ parent._children.append(weakref.ref(self))
|
|
|
+
|
|
|
+ def cancel(self, reason: str = 'user_cancelled'):
|
|
|
+ if self._cancelled:
|
|
|
+ return
|
|
|
+ self._cancelled = True
|
|
|
+ self._reason = reason
|
|
|
+ for cb in self._callbacks:
|
|
|
+ cb(reason)
|
|
|
+ for child_ref in self._children:
|
|
|
+ child = child_ref()
|
|
|
+ if child:
|
|
|
+ child.cancel(reason)
|
|
|
+
|
|
|
+ @property
|
|
|
+ def is_cancelled(self) -> bool:
|
|
|
+ return self._cancelled
|
|
|
+
|
|
|
+ def check(self):
|
|
|
+ """在每个 reduction 步骤前检查"""
|
|
|
+ if self._cancelled:
|
|
|
+ raise CancelledError(self._reason)
|
|
|
+
|
|
|
+# 在 Executor 中使用
|
|
|
+class Executor:
|
|
|
+ async def reduce(self, term, input_val, ctx, cancel_token: CancellationToken):
|
|
|
+ cancel_token.check() # 每步检查
|
|
|
+ if isinstance(term, Compose):
|
|
|
+ for stage in term.stages:
|
|
|
+ cancel_token.check()
|
|
|
+ result = await self.reduce(stage, result, ctx, cancel_token)
|
|
|
+```
|
|
|
+
|
|
|
+**工作量**:中
|
|
|
+**优先级**:P0 — 长任务无法取消 = 不可控
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### P0-4: 智能体文件隔离
|
|
|
+
|
|
|
+**现状问题**:
|
|
|
+
|
|
|
+lambdagentpaas 的所有 Agent **共享同一个工作目录 (CWD)**,没有任何文件系统隔离:
|
|
|
+
|
|
|
+```python
|
|
|
+# shell_tool.py - 所有 Agent 的 shell 命令在同一目录执行
|
|
|
+def shell_tool(command: str) -> str:
|
|
|
+ result = subprocess.run(command, shell=True, capture_output=True, text=True)
|
|
|
+ return result.stdout
|
|
|
+
|
|
|
+# multiagent.py - GroupChat 中 3 个 Agent 同时写文件
|
|
|
+agent_coder = Lam("coder", "Write code to main.py...")
|
|
|
+agent_tester = Lam("tester", "Write tests to test_main.py...")
|
|
|
+agent_reviewer = Lam("reviewer", "Refactor main.py...")
|
|
|
+# coder 和 reviewer 同时修改 main.py → 文件冲突/覆盖/损坏!
|
|
|
+```
|
|
|
+
|
|
|
+**后果**:
|
|
|
+- Agent A 写入 `main.py` 的改动被 Agent B 覆盖 → 静默数据丢失
|
|
|
+- 多 Agent 并行 `pip install` 不同版本 → 依赖冲突
|
|
|
+- Agent 执行 `rm -rf temp/` 影响其他 Agent 的临时文件
|
|
|
+- 无法回滚单个 Agent 的文件修改(全在一个 git 状态中)
|
|
|
+- PaaS 多租户场景下,不同 tenant 的 Agent 可能互相读写文件
|
|
|
+
|
|
|
+**Claude Code 做法**:
|
|
|
+
|
|
|
+Claude Code 使用 **Git Worktree** 实现真正的文件隔离:
|
|
|
+
|
|
|
+```typescript
|
|
|
+// worktree.ts — 每个 Agent 获得独立目录和分支
|
|
|
+async function createAgentWorktree(slug: string) {
|
|
|
+ // 1. 校验 slug 防止路径穿越
|
|
|
+ validateWorktreeSlug(slug)
|
|
|
+
|
|
|
+ // 2. 在 .claude/worktrees/<slug>/ 创建独立目录
|
|
|
+ // 等价于完整 repo 副本
|
|
|
+ const { worktreePath, worktreeBranch } = await getOrCreateWorktree(gitRoot, slug)
|
|
|
+ // worktreeBranch = "claude-<slug>-<timestamp>" — 独立分支
|
|
|
+
|
|
|
+ // 3. 大目录 symlink 避免磁盘浪费
|
|
|
+ await symlinkDirectories(gitRoot, worktreePath, ['node_modules', '.venv'])
|
|
|
+
|
|
|
+ // 4. Agent 的 CWD 切换到 worktree 目录
|
|
|
+ return { worktreePath, worktreeBranch, headCommit, gitRoot }
|
|
|
+}
|
|
|
+
|
|
|
+// 检测 Agent 是否产生了变更
|
|
|
+async function hasWorktreeChanges(worktreePath, gitRoot): boolean
|
|
|
+
|
|
|
+// Agent 完成后清理
|
|
|
+async function removeAgentWorktree(worktreePath, worktreeBranch, gitRoot)
|
|
|
+```
|
|
|
+
|
|
|
+隔离效果:
|
|
|
+- 每个 Agent 有**独立目录**: `.claude/worktrees/<agent-slug>/`
|
|
|
+- 每个 Agent 有**独立 Git 分支**: 可独立 commit,不影响主分支
|
|
|
+- **共享大目录**: `node_modules` 等通过 symlink 共享,不浪费磁盘
|
|
|
+- **变更检测**: Agent 完成后可检查是否有文件变更
|
|
|
+- **自动清理**: Agent 结束后 `git worktree remove`
|
|
|
+
|
|
|
+**改善方案**:
|
|
|
+
|
|
|
+lambdagentpaas 需要实现**三级��件隔离**,适应不同场景:
|
|
|
+
|
|
|
+```python
|
|
|
+# lambdagent/isolation.py — 智能体文件隔离模块
|
|
|
+
|
|
|
+import os
|
|
|
+import shutil
|
|
|
+import subprocess
|
|
|
+import tempfile
|
|
|
+from dataclasses import dataclass, field
|
|
|
+from enum import Enum
|
|
|
+from pathlib import Path
|
|
|
+from typing import Optional
|
|
|
+import weakref
|
|
|
+
|
|
|
+
|
|
|
+class IsolationLevel(Enum):
|
|
|
+ """三级隔离策略"""
|
|
|
+ NONE = "none" # 共享 CWD (仅单 Agent 场景)
|
|
|
+ DIRECTORY = "directory" # 临时目录隔离 (无 Git 项目)
|
|
|
+ WORKTREE = "worktree" # Git Worktree 隔离 (Git 项目, 推荐)
|
|
|
+
|
|
|
+
|
|
|
+@dataclass
|
|
|
+class IsolatedWorkspace:
|
|
|
+ """Agent 的隔离工作空间"""
|
|
|
+ agent_id: str
|
|
|
+ workspace_path: str # Agent 的工作目录
|
|
|
+ isolation_level: IsolationLevel
|
|
|
+ branch_name: Optional[str] = None # Worktree 模式下的分支名
|
|
|
+ original_cwd: Optional[str] = None # 原始 CWD (用于回退)
|
|
|
+ git_root: Optional[str] = None
|
|
|
+ _cleanup_registered: bool = field(default=False, repr=False)
|
|
|
+
|
|
|
+ def has_changes(self) -> bool:
|
|
|
+ """检测 Agent 是否产生了文件变更"""
|
|
|
+ if self.isolation_level == IsolationLevel.WORKTREE:
|
|
|
+ result = subprocess.run(
|
|
|
+ ['git', 'status', '--porcelain'],
|
|
|
+ cwd=self.workspace_path,
|
|
|
+ capture_output=True, text=True
|
|
|
+ )
|
|
|
+ return bool(result.stdout.strip())
|
|
|
+ elif self.isolation_level == IsolationLevel.DIRECTORY:
|
|
|
+ # 对比原始目录的文件列表
|
|
|
+ return True # 临时目录默认有变更
|
|
|
+ return False
|
|
|
+
|
|
|
+ def get_diff(self) -> str:
|
|
|
+ """获取 Agent 产生的所有变更"""
|
|
|
+ if self.isolation_level == IsolationLevel.WORKTREE:
|
|
|
+ result = subprocess.run(
|
|
|
+ ['git', 'diff', 'HEAD'],
|
|
|
+ cwd=self.workspace_path,
|
|
|
+ capture_output=True, text=True
|
|
|
+ )
|
|
|
+ return result.stdout
|
|
|
+ return ""
|
|
|
+
|
|
|
+ def commit(self, message: str) -> Optional[str]:
|
|
|
+ """在隔离分支上提交变更"""
|
|
|
+ if self.isolation_level != IsolationLevel.WORKTREE:
|
|
|
+ return None
|
|
|
+ subprocess.run(['git', 'add', '-A'], cwd=self.workspace_path, check=True)
|
|
|
+ result = subprocess.run(
|
|
|
+ ['git', 'commit', '-m', message],
|
|
|
+ cwd=self.workspace_path,
|
|
|
+ capture_output=True, text=True
|
|
|
+ )
|
|
|
+ if result.returncode == 0:
|
|
|
+ # 返回 commit hash
|
|
|
+ hash_result = subprocess.run(
|
|
|
+ ['git', 'rev-parse', 'HEAD'],
|
|
|
+ cwd=self.workspace_path,
|
|
|
+ capture_output=True, text=True
|
|
|
+ )
|
|
|
+ return hash_result.stdout.strip()
|
|
|
+ return None
|
|
|
+
|
|
|
+ def merge_back(self, target_branch: str = 'main') -> bool:
|
|
|
+ """将 Agent 的变更合并回主分支"""
|
|
|
+ if self.isolation_level != IsolationLevel.WORKTREE:
|
|
|
+ return False
|
|
|
+ try:
|
|
|
+ subprocess.run(
|
|
|
+ ['git', 'checkout', target_branch],
|
|
|
+ cwd=self.git_root, check=True
|
|
|
+ )
|
|
|
+ subprocess.run(
|
|
|
+ ['git', 'merge', self.branch_name, '--no-ff',
|
|
|
+ '-m', f'Merge agent {self.agent_id} changes'],
|
|
|
+ cwd=self.git_root, check=True
|
|
|
+ )
|
|
|
+ return True
|
|
|
+ except subprocess.CalledProcessError:
|
|
|
+ return False
|
|
|
+
|
|
|
+
|
|
|
+class WorkspaceManager:
|
|
|
+ """管理所有 Agent 的隔离工作空间"""
|
|
|
+
|
|
|
+ WORKTREE_BASE = '.lambdagent/worktrees'
|
|
|
+ SYMLINK_DIRS = ['node_modules', '.venv', '__pycache__', '.tox']
|
|
|
+
|
|
|
+ def __init__(self, base_dir: str = '.'):
|
|
|
+ self.base_dir = os.path.abspath(base_dir)
|
|
|
+ self._workspaces: dict[str, IsolatedWorkspace] = {}
|
|
|
+ self._git_root = self._find_git_root()
|
|
|
+
|
|
|
+ def _find_git_root(self) -> Optional[str]:
|
|
|
+ try:
|
|
|
+ result = subprocess.run(
|
|
|
+ ['git', 'rev-parse', '--show-toplevel'],
|
|
|
+ cwd=self.base_dir,
|
|
|
+ capture_output=True, text=True
|
|
|
+ )
|
|
|
+ return result.stdout.strip() if result.returncode == 0 else None
|
|
|
+ except FileNotFoundError:
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _validate_slug(self, slug: str):
|
|
|
+ """防止路径穿越攻击"""
|
|
|
+ if '..' in slug or '/' in slug or '\\' in slug:
|
|
|
+ raise ValueError(f"Invalid workspace slug: {slug}")
|
|
|
+ if len(slug) > 64:
|
|
|
+ raise ValueError(f"Slug too long: {slug}")
|
|
|
+
|
|
|
+ def create(
|
|
|
+ self,
|
|
|
+ agent_id: str,
|
|
|
+ level: IsolationLevel = IsolationLevel.WORKTREE,
|
|
|
+ slug: Optional[str] = None,
|
|
|
+ ) -> IsolatedWorkspace:
|
|
|
+ """为 Agent 创建隔离工作空间"""
|
|
|
+ slug = slug or agent_id.replace(' ', '-').lower()[:32]
|
|
|
+ self._validate_slug(slug)
|
|
|
+
|
|
|
+ if level == IsolationLevel.WORKTREE:
|
|
|
+ return self._create_worktree(agent_id, slug)
|
|
|
+ elif level == IsolationLevel.DIRECTORY:
|
|
|
+ return self._create_temp_directory(agent_id, slug)
|
|
|
+ else:
|
|
|
+ return IsolatedWorkspace(
|
|
|
+ agent_id=agent_id,
|
|
|
+ workspace_path=self.base_dir,
|
|
|
+ isolation_level=IsolationLevel.NONE,
|
|
|
+ original_cwd=self.base_dir
|
|
|
+ )
|
|
|
+
|
|
|
+ def _create_worktree(self, agent_id: str, slug: str) -> IsolatedWorkspace:
|
|
|
+ """Git Worktree 隔离 (推荐)"""
|
|
|
+ if not self._git_root:
|
|
|
+ # 无 Git 仓库,降级为目录隔离
|
|
|
+ return self._create_temp_directory(agent_id, slug)
|
|
|
+
|
|
|
+ import time
|
|
|
+ branch_name = f"agent-{slug}-{int(time.time())}"
|
|
|
+ worktree_dir = os.path.join(self._git_root, self.WORKTREE_BASE, slug)
|
|
|
+ os.makedirs(os.path.dirname(worktree_dir), exist_ok=True)
|
|
|
+
|
|
|
+ # 创建 worktree + 新分支
|
|
|
+ subprocess.run(
|
|
|
+ ['git', 'worktree', 'add', '-b', branch_name, worktree_dir],
|
|
|
+ cwd=self._git_root,
|
|
|
+ check=True,
|
|
|
+ capture_output=True
|
|
|
+ )
|
|
|
+
|
|
|
+ # Symlink 大目录避免磁盘浪费
|
|
|
+ for dirname in self.SYMLINK_DIRS:
|
|
|
+ src = os.path.join(self._git_root, dirname)
|
|
|
+ dst = os.path.join(worktree_dir, dirname)
|
|
|
+ if os.path.exists(src) and not os.path.exists(dst):
|
|
|
+ os.symlink(src, dst)
|
|
|
+
|
|
|
+ workspace = IsolatedWorkspace(
|
|
|
+ agent_id=agent_id,
|
|
|
+ workspace_path=worktree_dir,
|
|
|
+ isolation_level=IsolationLevel.WORKTREE,
|
|
|
+ branch_name=branch_name,
|
|
|
+ original_cwd=self.base_dir,
|
|
|
+ git_root=self._git_root
|
|
|
+ )
|
|
|
+ self._workspaces[agent_id] = workspace
|
|
|
+ return workspace
|
|
|
+
|
|
|
+ def _create_temp_directory(self, agent_id: str, slug: str) -> IsolatedWorkspace:
|
|
|
+ """临时目录隔离 (无 Git 场景)"""
|
|
|
+ temp_base = os.path.join(self.base_dir, '.lambdagent', 'workspaces')
|
|
|
+ os.makedirs(temp_base, exist_ok=True)
|
|
|
+ workspace_dir = os.path.join(temp_base, slug)
|
|
|
+
|
|
|
+ if not os.path.exists(workspace_dir):
|
|
|
+ # 浅拷贝:只复制文件,不复制大目录
|
|
|
+ shutil.copytree(
|
|
|
+ self.base_dir, workspace_dir,
|
|
|
+ ignore=shutil.ignore_patterns(
|
|
|
+ 'node_modules', '.venv', '__pycache__',
|
|
|
+ '.git', '.lambdagent'
|
|
|
+ )
|
|
|
+ )
|
|
|
+ # Symlink 大目录
|
|
|
+ for dirname in self.SYMLINK_DIRS:
|
|
|
+ src = os.path.join(self.base_dir, dirname)
|
|
|
+ dst = os.path.join(workspace_dir, dirname)
|
|
|
+ if os.path.exists(src) and not os.path.exists(dst):
|
|
|
+ os.symlink(src, dst)
|
|
|
+
|
|
|
+ workspace = IsolatedWorkspace(
|
|
|
+ agent_id=agent_id,
|
|
|
+ workspace_path=workspace_dir,
|
|
|
+ isolation_level=IsolationLevel.DIRECTORY,
|
|
|
+ original_cwd=self.base_dir
|
|
|
+ )
|
|
|
+ self._workspaces[agent_id] = workspace
|
|
|
+ return workspace
|
|
|
+
|
|
|
+ def cleanup(self, agent_id: str, force: bool = False):
|
|
|
+ """清理 Agent 的工作空间"""
|
|
|
+ ws = self._workspaces.pop(agent_id, None)
|
|
|
+ if not ws:
|
|
|
+ return
|
|
|
+
|
|
|
+ if ws.isolation_level == IsolationLevel.WORKTREE:
|
|
|
+ if ws.has_changes() and not force:
|
|
|
+ raise RuntimeError(
|
|
|
+ f"Agent {agent_id} has uncommitted changes. "
|
|
|
+ f"Use force=True or commit first."
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ subprocess.run(
|
|
|
+ ['git', 'worktree', 'remove', '--force', ws.workspace_path],
|
|
|
+ cwd=ws.git_root, check=True, capture_output=True
|
|
|
+ )
|
|
|
+ # 删除临时分支 (如果没有合并)
|
|
|
+ subprocess.run(
|
|
|
+ ['git', 'branch', '-D', ws.branch_name],
|
|
|
+ cwd=ws.git_root, capture_output=True
|
|
|
+ )
|
|
|
+ except subprocess.CalledProcessError:
|
|
|
+ pass # worktree 可能已被手动删除
|
|
|
+
|
|
|
+ elif ws.isolation_level == IsolationLevel.DIRECTORY:
|
|
|
+ if os.path.exists(ws.workspace_path):
|
|
|
+ shutil.rmtree(ws.workspace_path, ignore_errors=True)
|
|
|
+
|
|
|
+ def cleanup_all(self, force: bool = False):
|
|
|
+ """清理所有工作空间"""
|
|
|
+ for agent_id in list(self._workspaces.keys()):
|
|
|
+ try:
|
|
|
+ self.cleanup(agent_id, force=force)
|
|
|
+ except RuntimeError:
|
|
|
+ pass # 跳过有未提交变更的
|
|
|
+
|
|
|
+ def get(self, agent_id: str) -> Optional[IsolatedWorkspace]:
|
|
|
+ return self._workspaces.get(agent_id)
|
|
|
+
|
|
|
+ def list_active(self) -> list[IsolatedWorkspace]:
|
|
|
+ return list(self._workspaces.values())
|
|
|
+```
|
|
|
+
|
|
|
+**与现有模块集成**:
|
|
|
+
|
|
|
+```python
|
|
|
+# 1. 在 Executor 中自动创建隔离工作空间
|
|
|
+class Executor:
|
|
|
+ def __init__(self, config, workspace_manager: WorkspaceManager = None):
|
|
|
+ self.workspace_manager = workspace_manager or WorkspaceManager()
|
|
|
+
|
|
|
+ async def reduce(self, term, input_val, ctx, cancel_token):
|
|
|
+ # 多 Agent 场景自动隔离
|
|
|
+ if isinstance(term, (Par, AsyncPar, GroupChat)):
|
|
|
+ return await self._reduce_with_isolation(term, input_val, ctx, cancel_token)
|
|
|
+ return await self._reduce_normal(term, input_val, ctx, cancel_token)
|
|
|
+
|
|
|
+ async def _reduce_with_isolation(self, term, input_val, ctx, cancel_token):
|
|
|
+ """为每个子 Agent 创建隔离工作空间"""
|
|
|
+ agents = term.agents
|
|
|
+ workspaces = []
|
|
|
+ try:
|
|
|
+ for i, agent in enumerate(agents):
|
|
|
+ agent_id = f"{getattr(agent, 'name', f'agent-{i}')}"
|
|
|
+ ws = self.workspace_manager.create(agent_id, IsolationLevel.WORKTREE)
|
|
|
+ workspaces.append(ws)
|
|
|
+
|
|
|
+ # 并行执行,每个 Agent 在自己的 workspace 中
|
|
|
+ results = await asyncio.gather(*[
|
|
|
+ self._reduce_in_workspace(agent, input_val, ctx.fork(), cancel_token.child(), ws)
|
|
|
+ for agent, ws in zip(agents, workspaces)
|
|
|
+ ])
|
|
|
+ return tuple(results)
|
|
|
+ finally:
|
|
|
+ for ws in workspaces:
|
|
|
+ self.workspace_manager.cleanup(ws.agent_id, force=True)
|
|
|
+
|
|
|
+ async def _reduce_in_workspace(self, term, input_val, ctx, cancel_token, workspace):
|
|
|
+ """在隔离工作空间中执行 Agent"""
|
|
|
+ original_cwd = os.getcwd()
|
|
|
+ try:
|
|
|
+ os.chdir(workspace.workspace_path)
|
|
|
+ ctx.bindings['__workspace__'] = workspace # 让 Agent 能感知自己的 workspace
|
|
|
+ return await self.reduce(term, input_val, ctx, cancel_token)
|
|
|
+ finally:
|
|
|
+ os.chdir(original_cwd)
|
|
|
+
|
|
|
+
|
|
|
+# 2. shell_tool 自动使用 workspace CWD
|
|
|
+class IsolatedShellTool(Term):
|
|
|
+ def apply(self, input, ctx):
|
|
|
+ workspace = ctx.bindings.get('__workspace__')
|
|
|
+ cwd = workspace.workspace_path if workspace else '.'
|
|
|
+ result = subprocess.run(
|
|
|
+ input, shell=True, capture_output=True, text=True,
|
|
|
+ cwd=cwd, timeout=30
|
|
|
+ )
|
|
|
+ return result.stdout or result.stderr
|
|
|
+
|
|
|
+
|
|
|
+# 3. GroupChat 中每个 Agent 独立 workspace
|
|
|
+class IsolatedGroupChat(GroupChat):
|
|
|
+ async def apply(self, input, ctx, cancel_token):
|
|
|
+ workspaces = {}
|
|
|
+ try:
|
|
|
+ for agent in self.agents:
|
|
|
+ name = getattr(agent, 'name', str(id(agent)))
|
|
|
+ workspaces[name] = self.workspace_manager.create(
|
|
|
+ name, IsolationLevel.WORKTREE
|
|
|
+ )
|
|
|
+ # 每轮讨论,每个 agent 在自己的 workspace 执行
|
|
|
+ # 讨论结束后,coordinator 决定哪些变更合并回主分支
|
|
|
+ ...
|
|
|
+ finally:
|
|
|
+ for ws in workspaces.values():
|
|
|
+ self.workspace_manager.cleanup(ws.agent_id, force=True)
|
|
|
+
|
|
|
+
|
|
|
+# 4. 变更合并策略
|
|
|
+class MergeStrategy(Enum):
|
|
|
+ AUTO_MERGE = "auto" # 自动合并无冲突的变更
|
|
|
+ COORDINATOR = "coordinator" # Coordinator Agent 决定合并
|
|
|
+ MANUAL = "manual" # 人工确认
|
|
|
+
|
|
|
+class ChangeConsolidator:
|
|
|
+ """合并多个 Agent 的文件变更"""
|
|
|
+ def consolidate(
|
|
|
+ self,
|
|
|
+ workspaces: list[IsolatedWorkspace],
|
|
|
+ strategy: MergeStrategy = MergeStrategy.AUTO_MERGE
|
|
|
+ ) -> dict:
|
|
|
+ results = {}
|
|
|
+ for ws in workspaces:
|
|
|
+ if not ws.has_changes():
|
|
|
+ results[ws.agent_id] = {'status': 'no_changes'}
|
|
|
+ continue
|
|
|
+
|
|
|
+ ws.commit(f"Agent {ws.agent_id} changes")
|
|
|
+
|
|
|
+ if strategy == MergeStrategy.AUTO_MERGE:
|
|
|
+ success = ws.merge_back()
|
|
|
+ results[ws.agent_id] = {
|
|
|
+ 'status': 'merged' if success else 'conflict',
|
|
|
+ 'diff': ws.get_diff()
|
|
|
+ }
|
|
|
+ return results
|
|
|
+```
|
|
|
+
|
|
|
+**YAML 配置集成**:
|
|
|
+
|
|
|
+```yaml
|
|
|
+# agent-config.yml 新增 isolation 字段
|
|
|
+agentId: multi-coder
|
|
|
+type: parallel
|
|
|
+isolation:
|
|
|
+ level: worktree # none | directory | worktree
|
|
|
+ symlink: # 共享的大目录 (symlink)
|
|
|
+ - node_modules
|
|
|
+ - .venv
|
|
|
+ merge_strategy: auto # auto | coordinator | manual
|
|
|
+ cleanup: on_complete # on_complete | manual | never
|
|
|
+agents:
|
|
|
+ - name: frontend-coder
|
|
|
+ type: react
|
|
|
+ systemPrompt: "You write frontend code..."
|
|
|
+ - name: backend-coder
|
|
|
+ type: react
|
|
|
+ systemPrompt: "You write backend code..."
|
|
|
+```
|
|
|
+
|
|
|
+**工作量**:中-大 (核心 WorkspaceManager ~300 行,集成 ~200 行,配置支持 ~100 行)
|
|
|
+**优先级**:P0 — 多 Agent 写文件场景下,无隔离 = 数据损坏
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 三、P1 — 严重短板 (影响可靠性和扩展性)
|
|
|
+
|
|
|
+### P1-1: 并发安全修复
|
|
|
+
|
|
|
+**问题 1:Par 是假并行**
|
|
|
+
|
|
|
+```python
|
|
|
+# extensions.py - Par.apply()
|
|
|
+# 这不是并行!这是顺序执行!
|
|
|
+return tuple(a.apply(input, ctx) for a in self.agents)
|
|
|
+```
|
|
|
+
|
|
|
+**问题 2:AsyncPar 有线程安全问题**
|
|
|
+
|
|
|
+```python
|
|
|
+# multiagent.py - AsyncPar
|
|
|
+# ctx 在多线程间共享,bindings dict 非线程安全
|
|
|
+with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
|
|
|
+ futures = {pool.submit(a.apply, input, ctx): a for a in self.agents}
|
|
|
+ # ^^^ 共享的 ctx!
|
|
|
+```
|
|
|
+
|
|
|
+**问题 3:SharedMemory 无原子更新**
|
|
|
+
|
|
|
+```python
|
|
|
+# multiagent.py - SharedMemory.write()
|
|
|
+def write(self, key, value):
|
|
|
+ with self._lock:
|
|
|
+ self._store[key] = value # OK: 加锁了
|
|
|
+ self._history.append(...) # OK
|
|
|
+ # 但 read + write 组合不是原子的:
|
|
|
+ # Agent A: read(x) → 10
|
|
|
+ # Agent B: read(x) → 10
|
|
|
+ # Agent A: write(x, 11)
|
|
|
+ # Agent B: write(x, 11) ← 丢失更新!
|
|
|
+```
|
|
|
+
|
|
|
+**改善方案**:
|
|
|
+
|
|
|
+```python
|
|
|
+# Par: 用 asyncio.gather 实现真并行
|
|
|
+class Par(Term):
|
|
|
+ async def apply(self, input, ctx, cancel_token):
|
|
|
+ tasks = [
|
|
|
+ self.executor.reduce(a, input, ctx.fork(), cancel_token.child())
|
|
|
+ for a in self.agents
|
|
|
+ ]
|
|
|
+ return tuple(await asyncio.gather(*tasks))
|
|
|
+
|
|
|
+# SharedMemory: 添加 compare-and-swap
|
|
|
+class SharedMemory:
|
|
|
+ def compare_and_swap(self, key, expected, new_value) -> bool:
|
|
|
+ with self._lock:
|
|
|
+ if self._store.get(key) == expected:
|
|
|
+ self._store[key] = new_value
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+```
|
|
|
+
|
|
|
+**工作量**:中
|
|
|
+**优先级**:P1
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### P1-2: 上下文窗口管理
|
|
|
+
|
|
|
+**现状问题**:
|
|
|
+
|
|
|
+```python
|
|
|
+# react_engine.py - state 每步增长,永不压缩
|
|
|
+state = state + f"\n\nObservation: {obs}" # 线性增长!
|
|
|
+# 100 步 × 2000 字符 = 200K 字符 → 超出任何模型上下文窗口
|
|
|
+```
|
|
|
+
|
|
|
+Sliding window 只在 `from_config` 编译器中实现,核心 `react_engine.py` 中**未使用**。
|
|
|
+
|
|
|
+**Claude Code 做法**:
|
|
|
+
|
|
|
+```typescript
|
|
|
+// 三层压缩机制
|
|
|
+// 1. Auto-compact: 到达上下文 80% 时自动摘要旧消息
|
|
|
+// 2. History snip: 裁剪最旧的消息,保留摘要
|
|
|
+// 3. Microcompact: API 层面的消息合并
|
|
|
+buildPostCompactMessages(messages, tokenBudget)
|
|
|
+```
|
|
|
+
|
|
|
+**改善方案**:
|
|
|
+
|
|
|
+```python
|
|
|
+class ContextManager:
|
|
|
+ def __init__(self, max_tokens: int, model: str):
|
|
|
+ self.max_tokens = max_tokens
|
|
|
+ self.model = model
|
|
|
+
|
|
|
+ def should_compact(self, state: str) -> bool:
|
|
|
+ estimated_tokens = len(state) // 4 # 粗估
|
|
|
+ return estimated_tokens > self.max_tokens * 0.8
|
|
|
+
|
|
|
+ async def compact(self, state: str, llm: LLMAdapter) -> str:
|
|
|
+ """保留最近 N 步原文,旧步骤摘要"""
|
|
|
+ steps = self._parse_steps(state)
|
|
|
+ if len(steps) <= 5:
|
|
|
+ return state
|
|
|
+ old_steps = steps[:-3]
|
|
|
+ recent_steps = steps[-3:]
|
|
|
+ summary = await llm.call(
|
|
|
+ model=self.model,
|
|
|
+ system="Summarize these agent steps concisely, preserving key findings and decisions.",
|
|
|
+ user="\n".join(old_steps),
|
|
|
+ max_tokens=500
|
|
|
+ )
|
|
|
+ return f"[Summary of steps 1-{len(old_steps)}]\n{summary.text}\n\n" + "\n".join(recent_steps)
|
|
|
+```
|
|
|
+
|
|
|
+**工作量**:中
|
|
|
+**优先级**:P1 — 长 ReAct loop 必然 OOM 或超出上下文
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### P1-3: 工具输入校验
|
|
|
+
|
|
|
+**现状问题**:
|
|
|
+
|
|
|
+```python
|
|
|
+# Tool 没有任何输入校验
|
|
|
+class Tool(Term):
|
|
|
+ def apply(self, input, ctx):
|
|
|
+ return self.fn(input) # 直接执行,无校验
|
|
|
+```
|
|
|
+
|
|
|
+LLM 可能输出任何格式的 `tool_input`,直接传给 `fn` 可能导致崩溃或安全问题。
|
|
|
+
|
|
|
+**Claude Code 做法**:
|
|
|
+
|
|
|
+```typescript
|
|
|
+// 每个工具有 Zod schema
|
|
|
+const inputSchema = z.object({
|
|
|
+ command: z.string(),
|
|
|
+ timeout: z.number().optional(),
|
|
|
+ description: z.string()
|
|
|
+})
|
|
|
+
|
|
|
+// 执行前校验
|
|
|
+const parsed = tool.inputSchema.safeParse(input)
|
|
|
+if (!parsed.success) {
|
|
|
+ return { error: `Invalid input: ${parsed.error.message}` }
|
|
|
+}
|
|
|
+```
|
|
|
+
|
|
|
+**改善方案**:
|
|
|
+
|
|
|
+```python
|
|
|
+from pydantic import BaseModel, ValidationError
|
|
|
+
|
|
|
+class ToolSchema(BaseModel):
|
|
|
+ """工具输入 schema 基类"""
|
|
|
+ class Config:
|
|
|
+ extra = 'forbid' # 拒绝未知字段
|
|
|
+
|
|
|
+class ShellToolInput(ToolSchema):
|
|
|
+ command: str
|
|
|
+ timeout: int = 30
|
|
|
+ working_dir: str = "."
|
|
|
+
|
|
|
+class ValidatedTool(Term):
|
|
|
+ def __init__(self, name, fn, schema: type[ToolSchema] = None):
|
|
|
+ self.name = name
|
|
|
+ self.fn = fn
|
|
|
+ self.schema = schema
|
|
|
+
|
|
|
+ def apply(self, input, ctx):
|
|
|
+ if self.schema:
|
|
|
+ try:
|
|
|
+ validated = self.schema.model_validate_json(input) if isinstance(input, str) else self.schema(**input)
|
|
|
+ return self.fn(validated.model_dump())
|
|
|
+ except ValidationError as e:
|
|
|
+ return f"Tool input validation failed: {e}"
|
|
|
+ return self.fn(input)
|
|
|
+```
|
|
|
+
|
|
|
+**工作量**:小
|
|
|
+**优先级**:P1
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### P1-4: GroupChat 防死锁 + 状态爆炸
|
|
|
+
|
|
|
+**现状问题**:
|
|
|
+
|
|
|
+```python
|
|
|
+# GroupChat 的 conversation history 每轮完整传给每个 Agent
|
|
|
+speaker_input = f"Discussion so far:\n{state}\n\nYour turn..."
|
|
|
+# state 包含所有前轮所有 agent 的完整输出
|
|
|
+# 10 轮 × 3 agents × 1000 token/回复 = 30K tokens/轮
|
|
|
+# 到第 10 轮,每次 API 调用消耗 30K tokens
|
|
|
+```
|
|
|
+
|
|
|
+**改善方案**:
|
|
|
+
|
|
|
+```python
|
|
|
+class GroupChat(Term):
|
|
|
+ def _build_speaker_input(self, agent_name, history, round_num):
|
|
|
+ """智能上下文窗口,避免状态爆炸"""
|
|
|
+ if len(history) <= 6:
|
|
|
+ return "\n".join(history)
|
|
|
+
|
|
|
+ # 保留:第一轮(主题) + 最近 3 轮 + 当前 agent 的所有历史发言
|
|
|
+ first_round = history[:len(self.agents)]
|
|
|
+ recent = history[-3 * len(self.agents):]
|
|
|
+ my_messages = [h for h in history if h.startswith(f"[{agent_name}]")]
|
|
|
+
|
|
|
+ return (
|
|
|
+ "[Topic]\n" + "\n".join(first_round) +
|
|
|
+ "\n\n[Your previous messages]\n" + "\n".join(my_messages[-3:]) +
|
|
|
+ "\n\n[Recent discussion]\n" + "\n".join(recent)
|
|
|
+ )
|
|
|
+```
|
|
|
+
|
|
|
+**工作量**:中
|
|
|
+**优先级**:P1
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### P1-5: Checkpoint 支持中间状态恢复
|
|
|
+
|
|
|
+**现状问题**:
|
|
|
+
|
|
|
+```python
|
|
|
+# checkpoint.py - 只能保存 Context,不能保存执行位置
|
|
|
+@dataclass
|
|
|
+class Checkpoint:
|
|
|
+ context: dict # bindings + trace + memory
|
|
|
+ shared_memories: dict
|
|
|
+ metadata: dict
|
|
|
+ last_input: str
|
|
|
+ step_count: int # 记录了步数,但 Loop/GroupChat 不用它
|
|
|
+```
|
|
|
+
|
|
|
+如果 Agent 在 Loop 第 47 步崩溃,恢复后从第 0 步重新开始。
|
|
|
+
|
|
|
+**改善方案**:
|
|
|
+
|
|
|
+```python
|
|
|
+@dataclass
|
|
|
+class ExecutionCheckpoint(Checkpoint):
|
|
|
+ """支持中间状态恢复的增强 Checkpoint"""
|
|
|
+ execution_stack: list[StackFrame] # 执行栈快照
|
|
|
+ current_term: str # 当前执行的 Term 序列化
|
|
|
+ loop_state: dict # Loop 的 step_count + last_result
|
|
|
+ groupchat_state: dict # GroupChat 的 round + history
|
|
|
+
|
|
|
+@dataclass
|
|
|
+class StackFrame:
|
|
|
+ term_type: str # 'Loop', 'Compose', 'GroupChat'
|
|
|
+ term_id: str
|
|
|
+ step_index: int # 在 Compose 中的位置 / Loop 的当前步
|
|
|
+ local_state: dict # 该 frame 的局部状态
|
|
|
+
|
|
|
+# Loop 支持恢复
|
|
|
+class Loop(Term):
|
|
|
+ async def apply(self, input, ctx, cancel_token, resume_from: int = 0):
|
|
|
+ for step in range(resume_from, self.max_steps):
|
|
|
+ cancel_token.check()
|
|
|
+ result = await self.body.apply(input, ctx)
|
|
|
+ # 每步保存 checkpoint
|
|
|
+ if step % 5 == 0: # 每 5 步
|
|
|
+ save_checkpoint(ctx, loop_state={'step': step, 'result': result})
|
|
|
+ if self.condition(result, step):
|
|
|
+ break
|
|
|
+ return result
|
|
|
+```
|
|
|
+
|
|
|
+**工作量**:大
|
|
|
+**优先级**:P1 — 长任务(100+ 步)没有中间恢复 = 浪费大量计算
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 四、P2 — 建议改善 (提升竞争力)
|
|
|
+
|
|
|
+### P2-1: 结构化可观测性
|
|
|
+
|
|
|
+**现状**:`ctx.print_trace()` 输出到 stdout,无结构化。
|
|
|
+
|
|
|
+**目标**:
|
|
|
+
|
|
|
+```python
|
|
|
+# OpenTelemetry 集成
|
|
|
+from opentelemetry import trace as otel_trace
|
|
|
+
|
|
|
+class InstrumentedExecutor(Executor):
|
|
|
+ async def reduce(self, term, input_val, ctx, cancel_token):
|
|
|
+ tracer = otel_trace.get_tracer("lambdagent")
|
|
|
+ with tracer.start_as_current_span(
|
|
|
+ f"reduce.{type(term).__name__}",
|
|
|
+ attributes={
|
|
|
+ "term.name": getattr(term, 'name', ''),
|
|
|
+ "input.length": len(str(input_val)),
|
|
|
+ }
|
|
|
+ ) as span:
|
|
|
+ result = await super().reduce(term, input_val, ctx, cancel_token)
|
|
|
+ span.set_attribute("output.length", len(str(result)))
|
|
|
+ span.set_attribute("tokens.used", ctx.trace[-1].tokens_used)
|
|
|
+ return result
|
|
|
+```
|
|
|
+
|
|
|
+### P2-2: Token 预算管理
|
|
|
+
|
|
|
+```python
|
|
|
+class TokenBudget:
|
|
|
+ def __init__(self, max_tokens: int, model: str):
|
|
|
+ self.max_tokens = max_tokens
|
|
|
+ self.used = 0
|
|
|
+ self.model = model
|
|
|
+
|
|
|
+ def estimate_cost(self, prompt: str) -> int:
|
|
|
+ return len(prompt) // 4 # 粗估,可接 tiktoken
|
|
|
+
|
|
|
+ def can_afford(self, prompt: str) -> bool:
|
|
|
+ return self.used + self.estimate_cost(prompt) < self.max_tokens
|
|
|
+
|
|
|
+ def record(self, input_tokens: int, output_tokens: int):
|
|
|
+ self.used += input_tokens + output_tokens
|
|
|
+
|
|
|
+ def remaining(self) -> int:
|
|
|
+ return self.max_tokens - self.used
|
|
|
+```
|
|
|
+
|
|
|
+### P2-3: 工具并发安全声明
|
|
|
+
|
|
|
+借鉴 Claude Code 的 `isConcurrencySafe`:
|
|
|
+
|
|
|
+```python
|
|
|
+class Tool(Term):
|
|
|
+ def __init__(self, name, fn, concurrent_safe: bool = False):
|
|
|
+ self.name = name
|
|
|
+ self.fn = fn
|
|
|
+ self.concurrent_safe = concurrent_safe
|
|
|
+
|
|
|
+# 只读工具可并发
|
|
|
+search = Tool("search", search_fn, concurrent_safe=True)
|
|
|
+read_file = Tool("read", read_fn, concurrent_safe=True)
|
|
|
+
|
|
|
+# 写工具不可并发
|
|
|
+write_file = Tool("write", write_fn, concurrent_safe=False)
|
|
|
+
|
|
|
+# Executor 根据标记决定并发策略
|
|
|
+class Executor:
|
|
|
+ async def reduce_route(self, route, input_val, ctx, cancel_token):
|
|
|
+ tools = self._resolve_tools(route)
|
|
|
+ safe = [t for t in tools if t.concurrent_safe]
|
|
|
+ unsafe = [t for t in tools if not t.concurrent_safe]
|
|
|
+ # safe 并行执行,unsafe 顺序执行
|
|
|
+```
|
|
|
+
|
|
|
+### P2-4: 速率限制器
|
|
|
+
|
|
|
+```python
|
|
|
+import asyncio
|
|
|
+import time
|
|
|
+
|
|
|
+class RateLimiter:
|
|
|
+ """令牌桶速率限制"""
|
|
|
+ def __init__(self, requests_per_minute: int):
|
|
|
+ self.rpm = requests_per_minute
|
|
|
+ self.tokens = requests_per_minute
|
|
|
+ self.last_refill = time.monotonic()
|
|
|
+ self._lock = asyncio.Lock()
|
|
|
+
|
|
|
+ async def acquire(self):
|
|
|
+ async with self._lock:
|
|
|
+ now = time.monotonic()
|
|
|
+ elapsed = now - self.last_refill
|
|
|
+ self.tokens = min(self.rpm, self.tokens + elapsed * (self.rpm / 60))
|
|
|
+ self.last_refill = now
|
|
|
+ if self.tokens < 1:
|
|
|
+ wait = (1 - self.tokens) / (self.rpm / 60)
|
|
|
+ await asyncio.sleep(wait)
|
|
|
+ self.tokens = 0
|
|
|
+ else:
|
|
|
+ self.tokens -= 1
|
|
|
+```
|
|
|
+
|
|
|
+### P2-5: MCP 连接池 + 断路器
|
|
|
+
|
|
|
+```python
|
|
|
+class ResilientMCPClient:
|
|
|
+ def __init__(self, url, pool_size=5):
|
|
|
+ self.url = url
|
|
|
+ self.circuit = CircuitBreaker(failure_threshold=3, reset_timeout=30)
|
|
|
+ self._session = aiohttp.ClientSession(
|
|
|
+ connector=aiohttp.TCPConnector(limit=pool_size)
|
|
|
+ )
|
|
|
+ self._tool_cache = {} # 工具发现缓存
|
|
|
+
|
|
|
+ async def call_tool(self, name, arguments, timeout=30):
|
|
|
+ return await self.circuit.call(
|
|
|
+ lambda: with_retry(
|
|
|
+ lambda: self._do_call(name, arguments, timeout),
|
|
|
+ RetryPolicy(max_attempts=2, retryable_errors=(TimeoutError,))
|
|
|
+ )
|
|
|
+ )
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 五、改善路线图
|
|
|
+
|
|
|
+```
|
|
|
+Phase 1 (P0): 生存基础 [6-8 周]
|
|
|
+├── P0-1: 全链路 async/await 重构
|
|
|
+├── P0-1: LLMAdapter 流式接口
|
|
|
+├── P0-2: 统一超时 + 重试 + 断路器
|
|
|
+├── P0-3: CancellationToken 层级取消
|
|
|
+└── P0-4: 智能体文件隔离 (WorkspaceManager + Git Worktree)
|
|
|
+
|
|
|
+Phase 2 (P1): 可靠运行 [4-6 周]
|
|
|
+├── P1-1: Par/AsyncPar 真并行 (asyncio.gather)
|
|
|
+├── P1-2: Context 上下文压缩 (auto-compact)
|
|
|
+├── P1-3: Pydantic 工具输入校验
|
|
|
+├── P1-4: GroupChat 防状态爆炸
|
|
|
+└── P1-5: 中间状态 Checkpoint
|
|
|
+
|
|
|
+Phase 3 (P2): 竞争力提升 [4-6 周]
|
|
|
+├── P2-1: OpenTelemetry 集成
|
|
|
+├── P2-2: Token 预算管理
|
|
|
+├── P2-3: 工具并发安全声明
|
|
|
+├── P2-4: 速率限制器
|
|
|
+└── P2-5: MCP 连接池 + 断路器
|
|
|
+```
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 六、TODO LIST
|
|
|
+
|
|
|
+完整任务清单,按执行顺序排列。每项标注预估工作量和前置依赖。
|
|
|
+
|
|
|
+### Phase 1: P0 生存基础
|
|
|
+
|
|
|
+- [ ] **T01: 实现 `lambdagent/async_core.py` — 异步 Term 基类** [2d]
|
|
|
+ - 新增 `AsyncTerm` 基类,`async def apply()` 签名
|
|
|
+ - 所有 Term 子类迁移为 async (Lam, Compose, Loop, If, Route, Par, Guard, Memory)
|
|
|
+ - 保留同步 `Term.apply()` 作为兼容层
|
|
|
+
|
|
|
+- [ ] **T02: 实现 `lambdagent/agentruntime/llm_adapter.py` 流式接口** [3d]
|
|
|
+ - 新增 `async def stream()` 方法 → `AsyncGenerator[str, None]`
|
|
|
+ - Anthropic: `client.messages.stream()` + `text_stream`
|
|
|
+ - OpenAI: `client.chat.completions.create(stream=True)`
|
|
|
+ - DashScope: SSE 流式 HTTP
|
|
|
+ - Ollama: `stream=True` 参数
|
|
|
+ - 保留同步 `call()` 作为兼容
|
|
|
+
|
|
|
+- [ ] **T03: 重构 `lambdagent/agentruntime/executor.py` 为 async** [3d]
|
|
|
+ - `reduce()` → `async def reduce()`
|
|
|
+ - 所有 `_reduce_*` 方法异步化
|
|
|
+ - 依赖: T01, T02
|
|
|
+
|
|
|
+- [ ] **T04: 重构 `lambdagent/agentruntime/react_engine.py` 为 async + streaming** [3d]
|
|
|
+ - 7-phase loop 异步化
|
|
|
+ - 工具调用超时: `asyncio.wait_for(tool(input), timeout=tool_timeout)`
|
|
|
+ - 依赖: T03
|
|
|
+
|
|
|
+- [ ] **T05: 实现 `lambdagent/retry.py` — 统一重试/超时/断路器** [2d]
|
|
|
+ - `RetryPolicy` dataclass (max_attempts, base_delay, jitter, retryable_errors)
|
|
|
+ - `async def with_retry(fn, policy)` — 指数退避 + jitter
|
|
|
+ - `CircuitBreaker` class (closed/open/half_open 三状态)
|
|
|
+ - 集成到 LLMAdapter 和 MCPClient
|
|
|
+
|
|
|
+- [ ] **T06: 为所有 I/O 调用添加可配置超时** [1d]
|
|
|
+ - `LLMAdapter.call()`: `timeout` 参数 (默认 120s)
|
|
|
+ - `MCPClient.call_tool()`: `timeout` 参数 (默认 30s)
|
|
|
+ - `Lam._call_llm()`: 移除硬编码 120s,使用 RuntimeConfig.timeout
|
|
|
+ - `shell_tool()`: subprocess.run timeout (默认 30s)
|
|
|
+ - 依赖: T05
|
|
|
+
|
|
|
+- [ ] **T07: 实现 `lambdagent/cancellation.py` — CancellationToken** [2d]
|
|
|
+ - `CancellationToken(parent=None)`: 层级化取消
|
|
|
+ - `cancel(reason)`: 递归取消子 token (WeakRef 防泄漏)
|
|
|
+ - `check()`: 检查并抛出 `CancelledError`
|
|
|
+ - `child()`: 创建子 token
|
|
|
+ - 集成到 Executor.reduce(), Loop, GroupChat, AsyncPar
|
|
|
+
|
|
|
+- [ ] **T08: 实现 `lambdagent/isolation.py` — 智能体文件隔离** [4d]
|
|
|
+ - `IsolationLevel` enum: NONE / DIRECTORY / WORKTREE
|
|
|
+ - `IsolatedWorkspace` dataclass: workspace_path, branch, has_changes(), get_diff(), commit(), merge_back()
|
|
|
+ - `WorkspaceManager`: create(), cleanup(), cleanup_all()
|
|
|
+ - Git Worktree 创建 + symlink 大目录 + 路径穿越校验
|
|
|
+ - 临时目录隔离 (无 Git 场景降级)
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **T09: WorkspaceManager 集成到 Executor** [2d]
|
|
|
+ - Par/AsyncPar/GroupChat 自动为每个 Agent 创建 workspace
|
|
|
+ - shell_tool 自动使用 workspace CWD
|
|
|
+ - Context 注入 `__workspace__` binding
|
|
|
+ - 依赖: T08, T03
|
|
|
+
|
|
|
+- [ ] **T10: 变更合并策略 (`ChangeConsolidator`)** [2d]
|
|
|
+ - `MergeStrategy`: AUTO_MERGE / COORDINATOR / MANUAL
|
|
|
+ - 自动合并无冲突变更
|
|
|
+ - 冲突检测 + 报告
|
|
|
+ - 集成到 GroupChat/Par 的 finally 清理流程
|
|
|
+ - 依赖: T08, T09
|
|
|
+
|
|
|
+- [ ] **T11: from_config 编译器支持 `isolation` 字段** [1d]
|
|
|
+ - YAML schema 新增 `isolation.level`, `isolation.symlink`, `isolation.merge_strategy`
|
|
|
+ - 编译器将 isolation 配置传递给 Executor
|
|
|
+ - ���赖: T08
|
|
|
+
|
|
|
+- [ ] **T12: 集成测试 — Phase 1 全部功能** [2d]
|
|
|
+ - 异步 executor 端到端测试
|
|
|
+ - 流式输出验证
|
|
|
+ - 超时/重试场景测试
|
|
|
+ - 取消传播测试
|
|
|
+ - 文件隔离: 多 Agent 同时写文件不冲突
|
|
|
+ - Worktree 创建/清理/合并测试
|
|
|
+
|
|
|
+### Phase 2: P1 可靠运行
|
|
|
+
|
|
|
+- [ ] **T13: Par/AsyncPar 真并行** [2d]
|
|
|
+ - Par: `asyncio.gather()` 替代顺序 generator comprehension
|
|
|
+ - AsyncPar: 改用 asyncio (而非 ThreadPoolExecutor),解决 ctx 线程安全问题
|
|
|
+ - 每个并行 Agent 获得 `ctx.fork()` 独立上下文
|
|
|
+ - 依���: T01, T08
|
|
|
+
|
|
|
+- [ ] **T14: Context 上下文压缩 (`ContextManager`)** [3d]
|
|
|
+ - `should_compact(state)`: token 估算超过 80% 阈值
|
|
|
+ - `compact(state, llm)`: 旧步骤摘要 + 保留最近 N 步
|
|
|
+ - 集成到 ReActEngine 的 UPDATE 阶段
|
|
|
+ - 依赖: T02, T04
|
|
|
+
|
|
|
+- [ ] **T15: Pydantic 工具输入校验** [2d]
|
|
|
+ - `ToolSchema(BaseModel)` 基类
|
|
|
+ - `ValidatedTool(name, fn, schema)`: apply 时自动校验
|
|
|
+ - 内置 shell_tool, terminate 等添加 schema
|
|
|
+ - 校验失败返回错误提示,不崩溃
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **T16: GroupChat 防状态爆炸** [2d]
|
|
|
+ - `_build_speaker_input()`: 智能上下文窗口
|
|
|
+ - 保留 topic + 自己的发言 + 最近 N 轮
|
|
|
+ - 旧轮摘要压缩
|
|
|
+ - 依赖: T14
|
|
|
+
|
|
|
+- [ ] **T17: 中间状态 Checkpoint** [3d]
|
|
|
+ - `ExecutionCheckpoint`: execution_stack + loop_state + groupchat_state
|
|
|
+ - `StackFrame`: term_type, step_index, local_state
|
|
|
+ - Loop/GroupChat 每 N 步自动保存
|
|
|
+ - 恢复时从 checkpoint.step_index 继续,不从头
|
|
|
+ - 依赖: T03, T04
|
|
|
+
|
|
|
+### Phase 3: P2 竞争力提升 + 安全加固
|
|
|
+
|
|
|
+- [ ] **T18: OpenTelemetry 集成** [2d]
|
|
|
+ - `InstrumentedExecutor`: 每个 reduce 步骤生成 span
|
|
|
+ - Attributes: term.name, input.length, tokens.used, duration_ms
|
|
|
+ - 依赖: T03
|
|
|
+
|
|
|
+- [ ] **T19: Token 预算管理 (`TokenBudget`)** [1d]
|
|
|
+ - estimate_cost(), can_afford(), record(), remaining()
|
|
|
+ - 集成到 Executor: 预算耗尽时优雅停止
|
|
|
+ - 依赖: T02
|
|
|
+
|
|
|
+- [ ] **T20: 工具并发安全声明** [1d]
|
|
|
+ - `Tool(name, fn, concurrent_safe=False)` 参数
|
|
|
+ - Executor Route 按安全标记分组: safe 并行, unsafe 顺序
|
|
|
+ - 依赖: T13
|
|
|
+
|
|
|
+- [ ] **T21: 速率限制器 (`RateLimiter`)** [1d]
|
|
|
+ - 令牌桶算法
|
|
|
+ - 集成到 LLMAdapter: 每次调用前 acquire()
|
|
|
+ - 依赖: T02
|
|
|
+
|
|
|
+- [ ] **T22: MCP 连接池 + 断路器** [2d]
|
|
|
+ - `ResilientMCPClient`: aiohttp 连接池 + CircuitBreaker
|
|
|
+ - 工具发现缓存 (避免重复 list_tools)
|
|
|
+ - 依赖: T05
|
|
|
+
|
|
|
+### Phase 4: 安全加固 (来源: lambdagent/SECURITYSPEC.md + agentpaas/SECURITYSPEC.md)
|
|
|
+
|
|
|
+> 以下安全项来源于两份 SECURITYSPEC.md 中尚未完成的 TODO,按优先级排列。
|
|
|
+> 标注 `LSEC-xx` 来自 lambdagent,`SEC-xx` 来自 agentpaas。
|
|
|
+
|
|
|
+**P1 — 高优先级安全 (阻塞生产部署)**
|
|
|
+
|
|
|
+- [ ] **S01: 安全 lint 规则 (S001-S003)** [1d] (LSEC-04)
|
|
|
+ - `fromconfig/lint.py` 新增: S001 无界 maxSteps, S002 无 Guard 的工具 Agent, S003 过宽 SandboxPolicy
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S02: maxSteps 上限校验** [0.5d] (LSEC-05)
|
|
|
+ - 编译器强制 `maxSteps <= 1000`,超出报错
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S03: 工具进程级超时执行** [1d] (LSEC-06)
|
|
|
+ - 所有 Tool 调用可配置 timeout (默认 30s),进程级而非仅 asyncio
|
|
|
+ - 依赖: T06
|
|
|
+
|
|
|
+- [ ] **S04: 工具引用白名单** [1d] (LSEC-07)
|
|
|
+ - `from_config()` 编译时工具引用必须来自预定义白名单,禁止任意 import
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S05: MCP policy.mode 运行时执行** [1d] (LSEC-23)
|
|
|
+ - `mcp.policy.mode` (auto/force/intelligence/disable) 当前声明但未读取,需真正执行
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S06: PaaS 速率限制中间件** [2d] (SEC-06)
|
|
|
+ - 基于 `TenantContext.rate_limit` 的请求限流,每 API key 独立 (令牌桶)
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S07: Secrets 注入方式修复** [1d] (SEC-07)
|
|
|
+ - 当前 `inject.py` 写入 `os.environ` (全局可见),改为 Context dict 注入,仅当前 Agent 可见
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S08: API 输入大小校验** [1d] (SEC-08 + SEC-12)
|
|
|
+ - `RunRequest.input` 添加 `max_length` (默认 100KB)
|
|
|
+ - Agent config 字段大小限制
|
|
|
+ - SQL LIKE 查询转义通配符 `%`, `_`
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S09: 移除加密 fallback** [0.5d] (SEC-10)
|
|
|
+ - 当前加密失败 fallback 到 base64 (等于无加密),生产模式强制 `cryptography`
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S10: HTTPS 强制中间件** [0.5d] (SEC-11)
|
|
|
+ - 生产模式下强制 HTTPS,拒绝 HTTP 请求
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S11: ToolGateway 审计日志接入 PaaS** [1d] (SEC-29)
|
|
|
+ - `gateway.audit.entries` 暴露到 `/api/v1/traces/{run_id}` 端点
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S12: HIGH-risk 确认回调 webhook** [2d] (SEC-30)
|
|
|
+ - `highRiskConfirmation: true` 时通过 API/webhook 通知 tenant,暂停等待确认
|
|
|
+ - 依赖: S06
|
|
|
+
|
|
|
+**P2 — 中优先级安全 (纵深防御)**
|
|
|
+
|
|
|
+- [ ] **S13: Checkpoint 加密** [1d] (LSEC-09)
|
|
|
+ - Checkpoint JSON 使用 AES-256-GCM 加密,文件权限 0600
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S14: Memory 按 Agent/Tenant 隔离** [2d] (LSEC-10)
|
|
|
+ - 添加 `scope_key = f"{tenant_id}:{agent_id}"` 前缀,Redis/SQLite 表隔离
|
|
|
+ - 依赖: T08
|
|
|
+
|
|
|
+- [ ] **S15: Channel 访问控制** [1d] (LSEC-11)
|
|
|
+ - Agent 只能访问显式声明的 Channel,创建时绑定 agent_id 列表
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S16: Prompt 注入防御** [1d] (LSEC-12)
|
|
|
+ - 默认 system prompt 加入注入抵抗指令,工具输出标记为 untrusted data
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S17: Token 用量硬限制** [1d] (LSEC-13)
|
|
|
+ - 每次执行上下文 token 总量硬上限,超限优雅终止
|
|
|
+ - 依赖: T19
|
|
|
+
|
|
|
+- [ ] **S18: ToolGateway 路径 ACL** [2d] (LSEC-24)
|
|
|
+ - 执行 SandboxPolicy 的 `allowed_read_paths` / `allowed_write_paths`
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S19: ToolGateway 网络 ACL** [1d] (LSEC-25)
|
|
|
+ - `SandboxPolicy.network=False` 时阻止工具出站网络
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S20: API Key 哈希升级 PBKDF2** [1d] (SEC-13)
|
|
|
+ - 从裸 SHA-256 升级为 PBKDF2 + salt
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S21: 安全审计日志** [2d] (SEC-14)
|
|
|
+ - 记录: key 创建/吊销, tenant 创建, agent 部署, secret 访问, 认证失败
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S22: WeChat 凭证加密** [0.5d] (SEC-15)
|
|
|
+ - `~/.agentpaas/wechat/credentials.json` 用 master key 加密
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S23: 安全响应头** [0.5d] (SEC-16)
|
|
|
+ - HSTS, X-Frame-Options, Content-Security-Policy, X-Content-Type-Options
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S24: 请求 ID 追踪** [0.5d] (SEC-17)
|
|
|
+ - 每请求分配唯一 ID,贯穿日志链路
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S25: Per-Tenant Gateway Policy 覆盖** [1d] (SEC-32)
|
|
|
+ - Admin 可强制全 tenant 使用 strict() 策略
|
|
|
+ - 依赖: S11
|
|
|
+
|
|
|
+- [ ] **S26: ToolGateway 审计接入 OpenTelemetry** [1d] (LSEC-26 + SEC-34)
|
|
|
+ - 审计日志作为 OTel span 发出
|
|
|
+ - 依赖: T18
|
|
|
+
|
|
|
+**P3 — 低优先级安全 (持续改进)**
|
|
|
+
|
|
|
+- [ ] **S27: L2 容器沙箱** [5d] (LSEC-14)
|
|
|
+ - Docker/cgroup 完全隔离不信任 Agent
|
|
|
+ - 依赖: T08
|
|
|
+
|
|
|
+- [ ] **S28: YAML 解析器 Fuzz 测试** [2d] (LSEC-16)
|
|
|
+ - `from_config()` 模糊测试
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S29: 多 Agent 安全审计** [2d] (LSEC-17)
|
|
|
+ - 审计 Channel/GroupChat/AsyncPar 安全边界 (共享 ctx 线程安全, Channel 泄漏, Handoff 劫持)
|
|
|
+ - 依赖: T13
|
|
|
+
|
|
|
+- [ ] **S30: macOS sandbox-exec(1) 内核级隔离** [3d] (LSEC-28)
|
|
|
+ - 替代 Python monkey-patch fallback
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S31: 依赖漏洞扫描** [0.5d] (SEC-21)
|
|
|
+ - 集成 `pip-audit` / `safety` 到 CI
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S32: Per-Agent 工具速率限制** [1d] (LSEC-27 + SEC-33)
|
|
|
+ - 每分钟每 Agent 工具调用上限
|
|
|
+ - 依赖: S06
|
|
|
+
|
|
|
+- [ ] **S33: API Key 格式升级** [0.5d] (SEC-23)
|
|
|
+ - `secrets.token_hex(16)` → `secrets.token_urlsafe(32)`
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+- [ ] **S34: WeChat token 轮换策略** [1d] (SEC-20)
|
|
|
+ - 自动刷新 WeChat bot token
|
|
|
+ - 依赖: 无
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### TODO 统计
|
|
|
+
|
|
|
+| 类别 | 数量 | 预估总工作量 |
|
|
|
+|------|------|-------------|
|
|
|
+| 工程改善 (T01-T22) | 22 项 | ~42 人天 |
|
|
|
+| 安全加固 (S01-S34) | 34 项 | ~38 人天 |
|
|
|
+| **总计** | **56 项** | **~80 人天** |
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+### TODO 依赖图
|
|
|
+
|
|
|
+```
|
|
|
+T01 (async Term)
|
|
|
+ ├── T02 (LLM streaming)
|
|
|
+ │ ├── T03 (async Executor)
|
|
|
+ │ │ ├── T04 (async ReAct)
|
|
|
+ │ │ │ ├── T12 (集成测试)
|
|
|
+ │ │ │ ├── T14 (Context 压缩)
|
|
|
+ │ │ │ └── T17 (中间 Checkpoint)
|
|
|
+ │ │ ├── T09 (Workspace 集成)
|
|
|
+ │ │ │ └── T10 (变更合并)
|
|
|
+ │ │ ├── T13 (真并行)
|
|
|
+ │ │ └── T18 (OpenTelemetry)
|
|
|
+ │ ├── T19 (Token 预算)
|
|
|
+ │ └── T21 (速率限制)
|
|
|
+ ├── T05 (重试/断路器)
|
|
|
+ │ ├��─ T06 (统一超时)
|
|
|
+ │ └── T22 (MCP 连接池)
|
|
|
+ └── T07 (CancellationToken)
|
|
|
+
|
|
|
+T08 (文件隔离) ← 独立,可与 T01-T07 并行开发
|
|
|
+ ├── T09 (集成 Executor)
|
|
|
+ ├── T10 (变更合并)
|
|
|
+ └── T11 (YAML 配置)
|
|
|
+
|
|
|
+T15 (工具校验) ← 独立
|
|
|
+T16 (GroupChat 优化) ← 依赖 T14
|
|
|
+T20 (并发声明) ← 依赖 T13
|
|
|
+```
|
|
|
+
|
|
|
+**建议并行路线**:
|
|
|
+- **路线 A** (核心引擎): T01 → T02 → T03 → T04 → T05 → T06 → T07
|
|
|
+- **路线 B** (文件隔离): T08 → T09 → T10 → T11 (可与路线 A 并行)
|
|
|
+- **路线 C** (独立优化): T15 (可随时开始)
|
|
|
+- **路线 D** (安全加固): S01-S12 大部分无工程依赖,可立即与 A/B/C 并行启动
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 七、改善路线图 (含安全)
|
|
|
+
|
|
|
+```
|
|
|
+Phase 1 (P0): 生存基础 [6-8 周]
|
|
|
+├── 路线 A: 全链路 async + 流式 + 超时重试 + 取消
|
|
|
+├── 路线 B: 智能体文件隔离 (WorkspaceManager)
|
|
|
+└── 路线 D 前半: S01-S10 安全独立项 (并行)
|
|
|
+
|
|
|
+Phase 2 (P1): 可靠运行 [4-6 周]
|
|
|
+├── T13-T17: 真并行 + 上下文压缩 + 校验 + Checkpoint
|
|
|
+└── S11-S12, S13-S25: 安全纵深防御
|
|
|
+
|
|
|
+Phase 3 (P2+P3): 竞争力 + 持续安全 [4-6 周]
|
|
|
+├── T18-T22: OTel + Token预算 + 速率限制 + MCP连接池
|
|
|
+└── S26-S34: 容器沙箱 + Fuzz + 漏洞扫描 + 审计
|
|
|
+```
|
|
|
+
|
|
|
+**总计: 56 项 TODO (22 工程 + 34 安全), 预估 ~80 人天**
|
|
|
+
|
|
|
+---
|
|
|
+
|
|
|
+## 八、一句话总结
|
|
|
+
|
|
|
+**lambdagentpaas 的理论层是同类最优,但执行层和安全层都停留在"能跑通 demo"阶段。最核心的五个改善是:async 流式化、超时重试机制、取消传播链、智能体文件隔离、ToolGateway 全面落地。工程和安全可 4 路并行推进,做完 Phase 1 就从"研究原型"变成"可用工具"。**
|