lambdagent Runtime Spec

从 Lambda 项到真实执行：Agent 运行时规格文档

版本: 1.0 | 前置: FROM_CONFIG_SPEC.md | 理论基础: LDS ≡ Lambda Calculus

0. 为什么需要运行时

from_config() 编译器已经能把 YAML 翻译成 Lambda 项。但 Lambda 项本身不能执行——就像 Lambda 演算只定义了项的结构和规约规则，不定义"纸带在哪里"。

运行时（Runtime）就是那台执行 β-规约的机器。

编译器 (from_config)     运行时 (Runtime)
────────────────         ────────────────
YAML → Term              Term × Input → Result
定义 Lambda 项            执行 β-规约
静态的                   动态的
O(1) 时间                O(n) 时间，n = β-规约步数
不调用 LLM               调用 LLM / MCP / Shell
不产生副作用              读写 Memory、发 HTTP、写日志

Lambda 演算的类比：

Lambda 演算               lambdagent
─────────────            ─────────────
项 (Term)                 from_config() 的输出
β-规约规则                Runtime.step()
求值策略 (CBV/CBN)       Runtime 的调度策略
环境 (Γ)                  Context + Memory
Oracle 调用               MCP / CLI Tool 调用
规约终止 (Normal Form)   terminate 检测

1. 运行时架构总览

1.1 分层架构

┌─────────────────────────────────────────────────────┐
│                    用户接口层                         │
│    CLI (run/repl)  │  HTTP (serve)  │  SDK (Python)  │
├─────────────────────────────────────────────────────┤
│                    运行时核心层                       │
│  ┌─────────┐  ┌──────────┐  ┌───────────────────┐  │
│  │ Executor │  │ Scheduler│  │ TerminationOracle │  │
│  │ (β-规约) │  │ (调度)   │  │ (停止判定)         │  │
│  └────┬────┘  └────┬─────┘  └────────┬──────────┘  │
│       │            │                  │              │
│  ┌────▼────────────▼──────────────────▼──────────┐  │
│  │              ReAct Loop Engine                 │  │
│  │   think → parse → route → invoke → observe    │  │
│  └──────────────────────────────────────────────┘  │
├─────────────────────────────────────────────────────┤
│                    基础设施层                         │
│  ┌──────────────┐ ┌────────────┐ ┌────────────────┐│
│  │ LLMProvider   │ │ MCP Client │ │ Memory Backend ││
│  │ ┌ClaudeCode─┐ │ │ (HTTP/SSE/ │ │ (Local/Redis)  ││
│  │ │(--resume)  │ │ │  stdio)    │ │                ││
│  │ ├Anthropic──┤ │ └────────────┘ └────────────────┘│
│  │ │(Messages)  │ │ ┌────────────┐ ┌────────────────┐│
│  │ ├OpenAICompat┤ │ │ CLI Agent  │ │ Trace Store    ││
│  │ │(Ollama/..) │ │ │ (subprocess│ │ (JSON/SQLite)  ││
│  │ └───────────┘ │ └────────────┘ └────────────────┘│
│  └──────────────┘                                    │
├─────────────────────────────────────────────────────┤
│                    多智能体 / 协议层                   │
│  Channel  │ SharedMemory │ GroupChat │ Handoff      │
│  AsyncPar │ Skill/Registry │ A2A │ RAG │ Checkpoint │
├─────────────────────────────────────────────────────┤
│                    观测层                            │
│  Context.trace │ Metrics │ Logs │ Events            │
└─────────────────────────────────────────────────────┘

1.2 核心等式

Runtime.run(term, input) =
    let ctx = Context.new() in
    let result = Executor.reduce(term, input, ctx) in        ← β-规约
    (result, ctx.trace)                                       ← 结果 + 追踪

Executor.reduce(term, input, ctx) =
    match term with
    | ConversationLam(prov, h) → prov.chat(h ++ [input])     ← 对话感知 LLM
    | Lam(p, θ)         → LLMAdapter.call(θ, p, input)      ← 无状态 LLM (legacy)
    | Compose(f, g)     → reduce(g, reduce(f, input))        ← 链式规约
    | Loop(body, cond)  → ReActEngine.run(body, cond, input) ← Y 组合子
    | Tool(name, fn)    → ToolInvoker.call(fn, input)        ← 调工具
    | Route(cls, rs)    → ActionParser.parse → dispatch       ← CASE 分发
    | Memory(a, store)  → MemoryBackend.inject → reduce(a)   ← 环境扩展
    | Guard(a, P)       → reduce(a) then validate(P)          ← 类型约束
    | Pair(f, g)        → (reduce(f, input), reduce(g, input))← 并行
    | If(c, t, e)       → if reduce(c) then reduce(t) else reduce(e)

1.3 双引擎架构 (Phase 6.5)

Phase 6.5 引入了统一 Engine 抽象，支持 RecursiveEngine (Python 调用栈) 和 CEKEngine (Agent CEK Machine) 的运行时切换。通过 YAML 配置 runtime.engine 字段或 Python API engine_mode 参数选择。CEKEngine 提供逐步成本监控、暂停/恢复和循环检测能力。详见 docs/yaml-config.md §4。

2. Executor: β-规约引擎

2.1 职责

Executor 是运行时的核心——它接收一个 Term 和一个 Input，执行 β-规约，返回 Result。

class Executor:
    """
    β-规约引擎。

    Lambda 语义:
        Executor = 求值器 (Evaluator)
        reduce(term, input) = (term input) →β* result

    求值策略: Call-by-Value (严格求值)
        参数先求值，再传入函数体。
        原因: LLM 调用有副作用（API call），不能延迟求值。
    """

    def __init__(self, config: RuntimeConfig):
        self.llm = LLMAdapter(config.llm)
        self.mcp = MCPClient(config.mcp)
        self.memory = MemoryBackend.create(config.memory)
        self.cli_agents = CLIAgentPool(config.cli)
        self.trace = TraceStore(config.trace)
        self.termination = TerminationOracle(config.termination)

    def reduce(self, term: Term, input: Any, ctx: Context) -> Any:
        """
        对 term 执行一步或多步 β-规约。

        这是运行时的核心循环。每个 Term 类型有不同的规约规则。
        """

2.2 各 Term 类型的规约规则

def reduce(self, term: Term, input: Any, ctx: Context) -> Any:
    match type(term):

        case ConversationLam:
            # β-规约 with 对话记忆 = provider.chat(history ++ [input])
            return self._reduce_conversation_lam(term, input, ctx)

        case Lam:
            # β-规约 = LLM 前向传播 + 自回归解码 (legacy, stateless)
            return self._reduce_lam(term, input, ctx)

        case Compose:
            # f >> g = λx. g(f(x))
            # 从左到右依次规约
            result = input
            for stage in term.stages:
                result = self.reduce(stage, result, ctx)
            return result

        case Loop:
            # Y 组合子 = ReAct 循环
            return self._reduce_loop(term, input, ctx)

        case Tool:
            # 原语调用 = MCP / CLI / Python
            return self._reduce_tool(term, input, ctx)

        case Route:
            # CASE 分发 = 分类 + 路由
            return self._reduce_route(term, input, ctx)

        case Memory:
            # 环境扩展 = 注入记忆 + 执行 + 写回
            return self._reduce_memory(term, input, ctx)

        case Guard:
            # 类型约束 = 执行 + 验证 + 重试
            return self._reduce_guard(term, input, ctx)

        case Pair:
            # Church 对 = 并行执行
            a = self.reduce(term.first, input, ctx)
            b = self.reduce(term.second, input, ctx)
            return (a, b)

        case If:
            # Church 条件 = 求值条件 + 选择分支
            cond_result = self._eval_condition(term.cond, input, ctx)
            if cond_result:
                return self.reduce(term.then_, input, ctx)
            else:
                return self.reduce(term.else_, input, ctx)

2.3 Lam 规约：LLM 调用

def _reduce_lam(self, lam: Lam, input: Any, ctx: Context) -> Any:
    """
    (λ_D x) →β F_{M,D}(x)

    即: 把 input 喂给 LLM，拿到 output。
    一次 _reduce_lam = 一次 β-规约 = 一次 LLM API 调用。
    """
    t0 = time.time()

    # 构建完整 prompt
    system_prompt = lam.prompt
    user_message = str(input)

    # 注入上下文信息（如果有）
    if ctx.memory:
        user_message = self._inject_context(user_message, ctx)

    # 调用 LLM
    response = self.llm.call(
        model=lam.model,
        system=system_prompt,
        user=user_message,
        temperature=lam.temperature,
        max_tokens=lam.max_tokens,
    )

    duration = (time.time() - t0) * 1000

    # 输出解析
    result = lam.output_parser(response.text)

    # 记录 β-规约
    ctx.log(
        term_name=lam._name,
        term_id=lam._trace_id,
        inp=input,
        out=result,
        duration_ms=duration,
        model=lam.model,
        tokens=response.usage.total_tokens,
    )

    return result

2.4 ConversationLam 规约：对话感知 LLM 调用

def _reduce_conversation_lam(self, clam: ConversationLam, input: Any, ctx: Context) -> Any:
    """
    (λ_conv x) →β provider(history ++ [x])

    与无状态 Lam 的区别:
        Lam:             每次调用独立 — LLM 只看到当前输入
        ConversationLam: 每次调用累积 — LLM 看到完整对话历史

    ConversationLam.apply(input) 内部流程:
        1. self.messages.append({"role": "user", "content": input})
        2. managed = self._manage_context()  # 上下文窗口管理
        3. response = self.provider.chat(managed)
        4. self.messages.append({"role": "assistant", "content": response})
        5. return output_parser(response)

    上下文管理策略:
        - 始终保留 system message
        - 始终保留最近 N 轮 (keep_recent_turns = 20)
        - 超过 max_history_tokens 时，旧消息被压缩为 "[对话历史摘要]"
        - token 估算: 4 chars ≈ 1 token
    """
    t0 = time.time()
    result = clam.apply(input, ctx)
    duration = (time.time() - t0) * 1000
    ctx.log(clam._name, clam._trace_id, input, result, duration, clam.model)
    return result

ConversationLam 的 Lambda 语义

ConversationLam(provider, system_prompt) =
    λx. let h' = h ++ [("user", x)] in
        let managed = window(h', max_tokens) in
        let response = provider.chat(managed) in
        let h'' = h' ++ [("assistant", response)] in
        (response, h'')

其中 window(h, n) = if tokens(h) <= n then h
                     else [system] ++ summarize(old(h)) ++ recent(h)

这消除了 ReAct 循环中的"失忆"问题: LLM 在每一步都能看到完整对话，而不是只看到一个通过字符串拼接压缩的状态。

2.5 Provider 抽象层

Provider 是 LLM 调用的最底层抽象——只负责消息传输，不负责历史管理。

class LLMProvider(ABC):
    """
    统一的 LLM Provider 接口。

    Lambda 语义:
        LLMProvider = λ messages. LLM_response
        所有 provider 实现同一个 contract: messages in, text out.

    接口:
        chat(messages: list[dict]) -> str

    其中 messages = [
        {"role": "system", "content": "..."},
        {"role": "user",   "content": "..."},
        {"role": "assistant", "content": "..."},
        ...
    ]
    """

    @abstractmethod
    def chat(self, messages: List[Dict[str, str]]) -> str: ...

    @property
    def model_name(self) -> str: ...

    @property
    def context_window(self) -> int: ...

已实现的 Provider

Provider	传输方式	会话持久化	适用场景
`ClaudeCodeProvider`	subprocess (claude CLI)	`--resume <session_id>` (原生)	需要完整上下文记忆的复杂 ReAct 任务
`AnthropicProvider`	HTTP (Messages API)	由 ConversationLam 管理	标准 Anthropic API 调用
`OpenAICompatProvider`	HTTP (Chat Completions)	由 ConversationLam 管理	OpenAI / Ollama / DashScope / DeepSeek / Moonshot / Zhipu

ConversationLam 与 Provider 的职责分离

ConversationLam (对话层)          LLMProvider (传输层)
─────────────────────            ─────────────────────
管理 messages 列表                只关心 messages -> str
上下文窗口控制                    不知道历史
摘要压缩旧消息                    不知道 token 预算
记录 assistant 回复               不记录任何东西
输出解析                          不解析输出

2.6 Session 持久化机制

不同 provider 的会话持久化方式不同:

ClaudeCodeProvider: 原生 Session

第 1 次调用:
    claude -p "input" --system-prompt "..." --output-format json
    → 返回 {session_id: "abc123", result: "..."}
    → 保存 self._session_id = "abc123"

第 2+ 次调用:
    claude -p "new input" --resume abc123 --output-format text
    → Claude CLI 自动加载完整对话历史
    → 只需发送增量内容 (latest observation)

reset_session():
    self._session_id = None
    → 下次调用创建全新会话

这是最高效的模式: 后续 ReAct 步骤只发送最新的工具观察结果， Claude Code CLI 端自动保持完整上下文。

AnthropicProvider / OpenAICompatProvider: ConversationLam 管理

每次调用:
    ConversationLam 将 full messages array 传给 provider.chat()
    Provider 将 messages 转发给 HTTP API
    API 端无状态 — 每次都是独立请求

上下���管理:
    ConversationLam._manage_context() 在每次调用前检查 token 预算
    超出时进行 sliding window 压缩

session 检测在 ReAct 中的作用

# react_engine 的 _build_prompt 根据 think 的类型调整行为:
# ConversationLam 暴露 _session_id property (来自底层 provider)

if hasattr(self.think, '_session_id') and self.think._session_id:
    # Session mode: 只发送增量 (observation)
    # Provider 已有完整上下文
else:
    # Stateless mode: 发送完整状态
    # ConversationLam 负责历史管理

3. ReAct Loop Engine: Y 组合子的真实执行

3.1 为什么 ReAct 需要单独的引擎

ReAct 不是简单的 for step in range(n): body(result)。它是一个有内部结构的 Y 组合子展开，每一步包含 7 个子阶段：

Y 组合子的一次展开 (一步 β-规约) =

    think    →    parse    →    route    →    invoke    →    observe    →    update    →    check
    ─────         ─────         ─────         ──────         ───────         ──────         ─────
    LLM           提取           选择           执行            格式化          写回            终止
    推理          动作           工具           工具            观察结果        记忆            判定

3.2 ReActEngine 完整规格

class ReActEngine:
    """
    ReAct 循环引擎 = Y 组合子的运行时实现。

    Lambda 语义:
        react = Y_n(λself. λstate.
            let (thought, action) = think_and_parse(state) in
            IF (action = terminate)
                THEN extract_answer(thought)           ← base case
                ELSE let obs = invoke(action) in
                     self(state ⊕ format(thought, obs)) ← 递归
        )

    每一步展开 = 7 个子阶段:
        1. THINK:    LLM 推理 (β-规约)
        2. PARSE:    提取结构化动作
        3. ROUTE:    选择工具 (CASE)
        4. INVOKE:   执行工具 (β-规约)
        5. OBSERVE:  格式化观察结果
        6. UPDATE:   写回记忆 (Γ 更新)
        7. CHECK:    终止判定 (base case?)
    """

    def __init__(
        self,
        think: ConversationLam | Lam,    # 推理 Agent (优先 ConversationLam)
        tools: Dict[str, Tool],          # 工具集（含 terminate）
        action_parser: ActionParser,     # 动作解析器
        memory: MemoryBackend,           # 记忆后端
        termination: TerminationOracle,  # 终止判定器
        max_steps: int,                  # Y 组合子最大展开次数
        tool_timeout: int,               # 工具超时(秒)
        observation_enabled: bool,       # 是否追加观察到上下文
    ):
        ...

    def run(self, input: str, ctx: Context) -> str:
        """
        执行完整的 ReAct 循环。

        等价于: Y_n(react_body)(input)
        """
        state = input
        final_answer = None

        for step in range(self.max_steps):
            result = self._step(state, step, ctx)

            if result.terminated:
                final_answer = result.answer
                break

            state = result.next_state

        # 如果 max_steps 用尽仍未终止
        if final_answer is None:
            final_answer = self._force_terminate(state, ctx)

        return final_answer

3.3 单步展开的 7 个子阶段

@dataclass
class StepResult:
    """一步 Y 组合子展开的结果"""
    terminated: bool           # 是否到达 base case
    answer: Optional[str]      # terminated=True 时的最终答案
    next_state: Optional[str]  # terminated=False 时的下一步输入
    thought: str               # LLM 的推理文本
    action: Optional[Action]   # 解析出的动作
    observation: Optional[str] # 工具执行的观察结果
    step: int                  # 当前步数

@dataclass
class Action:
    """LLM 输出中提取的结构化动作"""
    tool: str                  # 工具名
    input: Dict[str, Any]      # 工具参数
    thought: str               # 推理过程
    raw: str                   # 原始 LLM 输出

def _step(self, state: str, step: int, ctx: Context) -> StepResult:
    """
    ReAct 的一步 = Y 组合子的一次展开。

    Phase 1-7 严格按顺序执行。
    """

    # ═══════════════════════════════════════════
    # Phase 1: THINK — LLM 推理 (β-规约)
    # ═══════════════════════════════════════════
    #
    # Lambda: let thought = think(state) in ...
    # 这是最核心的 β-规约：把当前状态喂给 LLM，拿到推理结果。

    prompt_with_context = self._build_prompt(state, step)
    thought = self.think(prompt_with_context, ctx)

    # ═══════════════════════════════════════════
    # Phase 2: PARSE — 提取结构化动作
    # ═══════════════════════════════════════════
    #
    # 从 LLM 的自由文本输出中提取结构化的 Action。
    # 支持多种格式：JSON / XML / 关键词。

    try:
        action = self.action_parser.parse(thought)
    except ParseError as e:
        # 解析失败 → 告诉 LLM 格式错误，让它重试
        observation = f"[FORMAT_ERROR] {e}. Please output a valid action."
        next_state = self._append_observation(state, thought, observation)
        ctx.log("parse_error", "", thought, observation, 0)
        return StepResult(
            terminated=False, answer=None,
            next_state=next_state,
            thought=thought, action=None, observation=observation,
            step=step,
        )

    # ═══════════════════════════════════════════
    # Phase 3: ROUTE — 选择工具 (CASE 分发)
    # ═══════════════════════════════════════════
    #
    # Lambda: CASE action.tool [
    #     ("sum",       Tool_MCP),
    #     ("improve",   Tool_MCP),
    #     ("terminate", λx.x),
    # ]

    if action.tool not in self.tools:
        observation = f"[ROUTE_ERROR] Unknown tool: {action.tool}. Available: {list(self.tools.keys())}"
        next_state = self._append_observation(state, thought, observation)
        ctx.log("route_error", "", action.tool, observation, 0)
        return StepResult(
            terminated=False, answer=None,
            next_state=next_state,
            thought=thought, action=action, observation=observation,
            step=step,
        )

    tool = self.tools[action.tool]

    # ═══════════════════════════════════════════
    # Phase 4: INVOKE — 执行工具 (β-规约)
    # ═══════════════════════════════════════════
    #
    # Lambda: let obs = tool(action.input) in ...
    #
    # 三种情况:
    # a) terminate → λx.x → 直接返回 thought (base case)
    # b) MCP Tool → HTTP 调用 MCP 端点
    # c) CLI Tool → subprocess 调用

    # 4a. Base Case: terminate = λx.x
    if action.tool == "terminate":
        answer = self._extract_final_answer(thought, action)
        ctx.log("terminate", "", state, answer, 0)
        return StepResult(
            terminated=True, answer=answer,
            next_state=None,
            thought=thought, action=action, observation=None,
            step=step,
        )

    # 4b/4c. 工具调用
    t0 = time.time()
    try:
        tool_result = self._invoke_with_timeout(
            tool, action.input, self.tool_timeout
        )
        duration = (time.time() - t0) * 1000
        ctx.log(f"Tool:{action.tool}", "", action.input, tool_result, duration)
    except TimeoutError:
        tool_result = f"[TIMEOUT] Tool '{action.tool}' exceeded {self.tool_timeout}s"
        ctx.log(f"Tool:{action.tool}", "", action.input, tool_result, 0)
    except Exception as e:
        tool_result = f"[TOOL_ERROR] {e}"
        ctx.log(f"Tool:{action.tool}", "", action.input, tool_result, 0)

    # ═══════════════════════════════════════════
    # Phase 5: OBSERVE — 格式化观察结果
    # ═══════════════════════════════════════════
    #
    # 将工具的原始输出格式化为 LLM 能理解的观察文本。
    # 截断过长的输出（防止 context overflow）。

    observation = self._format_observation(action.tool, tool_result)

    # ═══════════════════════════════════════════
    # Phase 6: UPDATE — 写回记忆 (Γ 更新)
    # ═══════════════════════════════════════════
    #
    # Lambda: Γ' = Γ ∪ {step_n: (thought, observation)}
    # 自动将本步的关键信息写入 Memory。

    self.memory.auto_save(
        key=f"step_{step}",
        thought=thought,
        action=action.tool,
        observation=observation,
    )

    # ═══════════════════════════════════════════
    # Phase 7: CHECK — 终止判定 (base case 检测)
    # ═══════════════════════════════════════════
    #
    # 除了显式调用 terminate 外，还有隐式终止条件:
    # - LLM 输出中包含 "Final Answer:" 等信号
    # - 工具返回了终止标记
    # - 达到步数上限

    if self.termination.should_stop(thought, observation, step):
        answer = self._extract_final_answer(thought, action)
        return StepResult(
            terminated=True, answer=answer,
            next_state=None,
            thought=thought, action=action, observation=observation,
            step=step,
        )

    # ═══════════════════════════════════════════
    # 准备下一步输入 (递归: self(state ⊕ obs))
    # ═══════════════════════════════════════════

    next_state = self._append_observation(state, thought, observation)

    return StepResult(
        terminated=False, answer=None,
        next_state=next_state,
        thought=thought, action=action, observation=observation,
        step=step,
    )

3.4 Prompt 构建 (`_build_prompt`)

def _build_prompt(self, state: str, step: int) -> str:
    """
    构建送给 LLM 的完整 prompt。

    结构:
        [System]         ← systemPrompt (from YAML)
        [Memory]         ← 持久记忆 (from Memory Backend)
        [Conversation]   ← 历史 think/observe 对
        [Step Info]      ← 当前步数 / 剩余步数
        [Available Tools]← 可用工具列表 + 调用格式说明
        [User Input]     ← 原始用户输入 (第一步) 或上一步状态

    Lambda 语义:
        prompt = Γ(memory) ⊕ Γ(tools) ⊕ state ⊕ step_info
    """

    parts = []

    # 记忆注入
    memory_items = self.memory.read_recent(n=self.memory_size)
    if memory_items:
        parts.append("[Memory Context]")
        for key, value, age in memory_items:
            parts.append(f"  - {key}: {value} ({age})")
        parts.append("")

    # 工具说明
    parts.append("[Available Tools]")
    parts.append("Call a tool by outputting JSON: {\"action\": \"tool_name\", \"input\": {...}}")
    parts.append("To finish, call: {\"action\": \"terminate\", \"answer\": \"your final answer\"}")
    parts.append("")
    for name, tool in self.tools.items():
        if name == "terminate":
            parts.append(f"  - terminate: Signal task completion (λx.x)")
        else:
            schema = getattr(tool, 'schema', None)
            desc = schema.get('description', '') if schema else ''
            parts.append(f"  - {name}: {desc}")
    parts.append("")

    # 步数信息
    remaining = self.max_steps - step - 1
    parts.append(f"[Step {step + 1}/{self.max_steps}, remaining: {remaining}]")
    if remaining <= 3:
        parts.append("⚠ Running low on steps. Consider wrapping up or calling terminate.")
    parts.append("")

    # 当前状态
    parts.append(state)

    return "\n".join(parts)

4. ActionParser: 从 LLM 输出提取结构化动作

4.1 职责

LLM 输出的是自由文本。ActionParser 负责从中提取结构化的 Action(tool, input, thought)。

4.2 解析优先级

优先级 1: JSON 块
    LLM 输出: "Let me calculate... ```json\n{\"action\": \"sum\", \"input\": {\"numbers\": [1,2,3]}}\n```"
    提取: Action(tool="sum", input={"numbers": [1,2,3]})

优先级 2: 内联 JSON
    LLM 输出: "I'll use the sum tool. {\"action\": \"sum\", \"input\": {\"numbers\": [1,2,3]}}"
    提取: 同上

优先级 3: XML 标签
    LLM 输出: "<action>sum</action><input>{\"numbers\": [1,2,3]}</input>"
    提取: 同上

优先级 4: 关键词 + 启发式
    LLM 输出: "I need to use the sum tool to calculate 1+2+3"
    提取: Action(tool="sum", input={"query": "calculate 1+2+3"})
    注意: 这是兜底策略，精度较低

优先级 5: 无动作 (隐式 terminate)
    LLM 输出: "The answer is 42. Task complete."
    提取: Action(tool="terminate", input={"answer": "42"})

4.3 接口规格

class ActionParser:
    """
    从 LLM 自由文本输出中提取结构化动作。

    Lambda 语义:
        parse: String → Action
        parse(thought) = (tool_name, params, reasoning)

    这对应 Lambda 演算中的"模式匹配":
        CASE thought [
            (json_pattern,    extract_json),
            (xml_pattern,     extract_xml),
            (keyword_pattern, extract_keyword),
            (_,               implicit_terminate),
        ]
    """

    def __init__(self, tool_names: List[str], tool_schemas: Dict[str, dict] = None):
        self.tool_names = tool_names
        self.tool_schemas = tool_schemas or {}

    def parse(self, llm_output: str) -> Action:
        """
        解析 LLM 输出为 Action。

        返回: Action(tool, input, thought, raw)
        抛出: ParseError（如果无法解析且不满足隐式终止条件）
        """

    def _try_json_block(self, text: str) -> Optional[Action]:
        """尝试提取 ```json ... ``` 块"""

    def _try_inline_json(self, text: str) -> Optional[Action]:
        """尝试提取行内 JSON 对象"""

    def _try_xml(self, text: str) -> Optional[Action]:
        """尝试提取 XML 标签"""

    def _try_keyword(self, text: str) -> Optional[Action]:
        """启发式关键词匹配"""

    def _try_implicit_terminate(self, text: str) -> Optional[Action]:
        """检测隐式终止信号"""

4.4 Action 校验

def validate_action(self, action: Action) -> Action:
    """
    校验 Action 的工具名和参数。

    1. 工具名必须在 tool_names 中
    2. 如果有 schema，参数必须符合 schema
    3. 参数类型必须正确

    Lambda 语义:
        validate: Action → Action | ⊥
        这是 Guard(parse, P) 的应用: 依赖类型约束
    """
    if action.tool not in self.tool_names:
        raise ParseError(
            f"Unknown tool: '{action.tool}'. "
            f"Available: {self.tool_names}"
        )

    if action.tool in self.tool_schemas:
        schema = self.tool_schemas[action.tool]
        # JSON Schema 校验
        self._validate_params(action.input, schema)

    return action

5. TerminationOracle: 何时停下

5.1 职责

判定 Y 组合子是否到达 base case。在 Lambda 演算中，base case 是"不再调用 self 的分支"。在 Agent 中，对应"不再需要新工具调用"的时刻。

5.2 终止条件层次

class TerminationOracle:
    """
    终止判定器 = Y 组合子的 base case 检测。

    Lambda 语义:
        should_stop = λ(thought, obs, step).
            IF (action = terminate) THEN TRUE           ← 显式 base case
            ELSE IF (implicit_signal(thought)) THEN TRUE ← 隐式 base case
            ELSE IF (step >= max) THEN TRUE              ← 强制截断
            ELSE FALSE                                    ← 继续递归
    """

    def should_stop(self, thought: str, observation: str, step: int) -> bool:
        """
        三层终止检测:

        Layer 1: 显式终止
            Agent 调用了 terminate 工具。
            这是 λx.x — 恒等函数 — Y 组合子的标准 base case。

        Layer 2: 隐式终止信号
            Agent 输出中包含终止信号，但没有显式调用 terminate。
            例如: "Final Answer: 42", "Task complete.", "I'm done."

        Layer 3: 强制截断
            达到 maxSteps。
            Lambda 对应: Y_n 的有界性——最多展开 n 次。
        """

        # Layer 1: 显式 (已在 Phase 4 处理，这里是备用)
        if "terminate" in thought.lower() and "action" in thought.lower():
            return True

        # Layer 2: 隐式信号
        implicit_signals = [
            "final answer:",
            "task complete",
            "task is done",
            "i have completed",
            "here is the result:",
            "in conclusion,",
        ]
        thought_lower = thought.lower()
        for signal in implicit_signals:
            if signal in thought_lower:
                return True

        # Layer 3: 在 caller 处检测 (step >= max_steps)
        # 这里不检查，由 ReActEngine.run() 的 for 循环处理

        return False

5.3 强制终止时的答案提取

def _force_terminate(self, state: str, ctx: Context) -> str:
    """
    maxSteps 用尽时的强制终止。

    Lambda: Y_n(g) 展开 n 次后截断 → 返回当前项（非 normal form）

    策略:
        1. 让 LLM 从当前状态中提取最终答案
        2. 如果 LLM 也失败，返回最后一步的 thought

    这是"有界递归"的代价——可能返回不完整的结果。
    """
    # 用一次额外的 LLM 调用提取答案
    extractor = Lam(
        "force_extract",
        "The agent ran out of steps. Extract the best answer from the conversation so far. "
        "If no clear answer, summarize what was accomplished.",
        max_tokens=512,
    )
    return extractor(state, ctx)

6. LLM Provider 层: 多模型统一接口

6.1 新架构: LLMProvider (v2)

v2 采用 LLMProvider 接口替代旧的 LLMAdapter。核心变化:

接口从 call(model, system, user) 变为 chat(messages: list[dict]) -> str
会话管理由 ConversationLam 负责，Provider 只负责传输
Provider 通过 _create_provider() 工厂函数创建，不通过注册表

class LLMProvider(ABC):
    """
    统一的 LLM Provider 接口 (v2)。

    Lambda 语义:
        LLMProvider = λ messages. response_text

    与旧 LLMAdapter 的区别:
        LLMAdapter: call(model, system, user) → LLMResponse  (model-centric)
        LLMProvider: chat(messages) → str                     (conversation-centric)

    已实现的 Provider:
        - ClaudeCodeProvider  (subprocess, session persistence)
        - AnthropicProvider   (HTTP, Anthropic Messages API)
        - OpenAICompatProvider(HTTP, OpenAI Chat Completions API)
    """

    def __init__(self, config: ProviderConfig): ...

    @abstractmethod
    def chat(self, messages: List[Dict[str, str]]) -> str:
        """
        发送 messages 给 LLM，返回 response 文本。

        Args:
            messages: [{"role": "system/user/assistant", "content": "..."}]

        Returns:
            assistant 的 response 文本。

        Raises:
            ProviderError: API/连接失败。
        """
        ...

    @property
    def model_name(self) -> str: ...

    @property
    def context_window(self) -> int: ...

@dataclass
class ProviderConfig:
    model: str = ""
    temperature: float = 0.3
    max_tokens: int = 4096
    timeout: int = 600
    context_window: int = 200000
    extra: Dict[str, Any] = field(default_factory=dict)

6.2 Provider 工厂: _create_provider

def _create_provider(model_cfg: Dict) -> (LLMProvider, bool):
    """
    工厂函数: 根据 YAML model 配置创建 Provider 实例。

    路由逻辑 (CASE dispatch):
        model.provider == "claude-code"  → ClaudeCodeProvider
        model.provider == "anthropic"    → AnthropicProvider
        model.provider ∈ {"openai", "ollama", "dashscope",
                          "deepseek", "moonshot", "zhipu"}
                                         → OpenAICompatProvider

    返回 (provider, use_conversation):
        use_conversation 由 model.conversation 字段控制 (默认 true)
        true  → 编译器将 provider 包装为 ConversationLam
        false → 编译器使用旧的 Lam (无状态)
    """

6.3 ClaudeCodeProvider: Session 持久化

class ClaudeCodeProvider(LLMProvider):
    """
    基于 Claude Code CLI 的 Provider。

    传输方式: subprocess 调用 `claude` CLI
    会话持久化: --resume <session_id>

    chat() 流程:
        第 1 次:
            1. claude -p "input" --system-prompt "..." --output-format json
            2. 解析 JSON: {session_id, result}
            3. 保存 self._session_id
            4. 返回 result

        第 2+ 次:
            1. 提取 messages 中最新的 user message
            2. claude -p "latest" --resume <session_id> --output-format text
            3. 返回 stdout

    Lambda 语义:
        chat₁ = λ msgs. let (sid, r) = claude_new(msgs) in (r, sid)
        chatₙ = λ msgs. claude_resume(sid, last(msgs))
    """

    def chat(self, messages: List[Dict[str, str]]) -> str:
        if self._session_id:
            return self._call_resume(messages)  # 只发最新消息
        else:
            return self._call_new(messages)     # 创建 session

    def reset_session(self):
        """Reset session — 下次调用创建新会话。"""
        self._session_id = None

6.4 OpenAICompatProvider: 无状态 HTTP

class OpenAICompatProvider(LLMProvider):
    """
    兼容 OpenAI Chat Completions API 的 Provider。

    支持: OpenAI / Ollama / DashScope / DeepSeek / Moonshot / Zhipu

    chat() 流程:
        1. 根据 provider_name 确定 base_url 和 api_key
        2. POST /v1/chat/completions {"model": ..., "messages": [...]}
        3. 解析 response.choices[0].message.content
        4. 返回文本

    无状态: 每次调用都是独立的 HTTP 请求，
    由 ConversationLam 在调用前组装完整 messages 数组。
    """

7. MCP Client: 工具调用的真实实现

7.1 接口

class MCPClient:
    """
    MCP 协议客户端。

    Lambda 语义:
        MCPClient = λ(server, tool, input). HTTP_POST(url, {tool, input})

    对应 YAML:
        app.mcp.custom.nodes.<server>.url
        app.mcp.custom.nodes.<server>.endpoint
        app.mcp.custom.nodes.<server>.headers
    """

    def __init__(self, nodes: Dict[str, MCPNodeConfig]):
        self.nodes = nodes   # server_name → config
        self.clients = {}    # server_name → httpx.AsyncClient

    async def invoke(
        self,
        server: str,
        tool: str,
        input: Dict[str, Any],
        timeout: int = 30,
    ) -> str:
        """
        调用 MCP 工具。

        HTTP 请求:
            POST {url}{endpoint}
            Headers: {Authorization: token, Content-Type: application/json}
            Body: {"jsonrpc": "2.0", "method": "tools/call",
                   "params": {"name": tool, "arguments": input}}

        返回: 工具执行结果（字符串）
        异常: MCPError (连接失败/超时/4xx/5xx)
        """

    async def discover(self, server: str) -> List[ToolSchema]:
        """
        发现 MCP 服务器上的可用工具。

        HTTP 请求:
            POST {url}{endpoint}
            Body: {"jsonrpc": "2.0", "method": "tools/list"}

        返回: [ToolSchema(name, description, input_schema), ...]
        """

@dataclass
class MCPNodeConfig:
    url: str
    endpoint: str
    headers: Dict[str, str]
    timeout: int = 30
    retry: int = 0

@dataclass
class ToolSchema:
    name: str
    description: str
    input_schema: Dict     # JSON Schema

7.2 连接管理

class MCPConnectionPool:
    """
    MCP 连接池。

    每个 MCP server 维护一个持久连接。
    支持:
        - HTTP (请求/响应)
        - SSE (服务端推送，用于流式工具结果)
        - WebSocket (双向，用于长时间运行的工具)

    Lambda 语义:
        Pool = Memory(connections, {server → client})
        即: 连接池是一种环境扩展——把已建立的连接存入 Γ
    """

8. Memory Backend: 环境 Γ 的持久化

8.1 接口

class MemoryBackend(ABC):
    """
    记忆后端抽象接口。

    Lambda 语义:
        MemoryBackend = Γ 的存储实现
        read()  = 查找变量绑定 (Γ.lookup)
        write() = 环境扩展 (Γ' = Γ ∪ {k: v})
        evict() = 绑定回收 (Γ' = Γ \ expired)
    """

    @abstractmethod
    def read_recent(self, n: int) -> List[Tuple[str, Any, str]]:
        """读取最近 n 条记忆。返回 [(key, value, age_str), ...]"""

    @abstractmethod
    def read(self, key: str) -> Optional[Any]:
        """读取指定 key"""

    @abstractmethod
    def write(self, key: str, value: Any) -> None:
        """写入（自动 TTL 和 LRU 淘汰）"""

    @abstractmethod
    def auto_save(self, key: str, thought: str, action: str, observation: str) -> None:
        """自动从一步执行结果中提取并保存关键信息"""

    @abstractmethod
    def clear(self) -> None:
        """清空全部记忆"""

    @staticmethod
    def create(config: MemoryConfig) -> "MemoryBackend":
        """工厂方法: 根据配置创建对应后端"""
        match config.strategy:
            case "local":  return LocalMemory(config.size, config.ttl)
            case "redis":  return RedisMemory(config.size, config.ttl, config.redis_url)
            case "sqlite": return SQLiteMemory(config.size, config.ttl, config.db_path)
            case _:        return LocalMemory(config.size, config.ttl)

8.2 LocalMemory (默认)

class LocalMemory(MemoryBackend):
    """
    进程内记忆后端。

    数据结构: OrderedDict (保持插入顺序，支持 LRU 淘汰)
    TTL: 通过时间戳检查
    Size: 通过 OrderedDict.popitem(last=False) 淘汰最旧条目

    Lambda 语义: Γ 存在 Python 进程的堆内存中。进程退出则 Γ 丢失。
    """

8.3 RedisMemory (生产)

class RedisMemory(MemoryBackend):
    """
    Redis 记忆后端。

    数据结构: Redis Sorted Set (score = timestamp，自动按时间排序)
    TTL: Redis 原生 EXPIRE
    Size: ZREMRANGEBYRANK 淘汰最旧条目
    Scope: key prefix = agent_id + session_id

    Lambda 语义: Γ 存在 Redis 中。跨进程、跨机器共享。
    """

8.4 auto_save: 自动信息提取

def auto_save(self, key: str, thought: str, action: str, observation: str):
    """
    从一步执行结果中自动提取并保存关键信息。

    策略 1 (简单): 保存整个 (thought, action, observation) 三元组
    策略 2 (摘要): 用 LLM 提取关键事实再保存
    策略 3 (结构化): 从 observation 中正则提取 key=value 对

    默认用策略 1 (最可靠)，可通过配置切换。
    """
    summary = f"[{action}] {observation[:200]}"
    self.write(key, summary)

9. Trace Store: β-规约追踪的持久化

9.1 追踪数据模型

@dataclass
class TraceRecord:
    """一条 β-规约记录 (比 TraceEntry 更完整)"""
    # 基础
    step: int
    term_name: str
    term_type: str        # "Lam" | "Tool" | "Route" | "Loop" | ...
    duration_ms: float
    timestamp: float

    # 输入输出
    input: str
    output: str
    input_tokens: int
    output_tokens: int

    # ReAct 专用
    thought: Optional[str]
    action: Optional[str]
    action_input: Optional[Dict]
    observation: Optional[str]
    terminated: bool

    # 模型
    model: str
    temperature: float

    # 错误
    error: Optional[str]

9.2 接口

class TraceStore:
    """
    β-规约追踪存储。

    Lambda 语义:
        TraceStore = List[β-reduction]
        每次 reduce 调用追加一条记录。
        最终的 trace = 完整的规约链。
    """

    def append(self, record: TraceRecord): ...
    def get_all(self) -> List[TraceRecord]: ...
    def get_step(self, n: int) -> TraceRecord: ...
    def to_json(self) -> str: ...
    def to_timeline(self) -> str: ...
    def stats(self) -> TraceStats: ...

@dataclass
class TraceStats:
    total_steps: int
    total_time_ms: float
    total_tokens: int
    tool_calls: int
    llm_calls: int
    errors: int
    terminated_by: str   # "terminate" | "implicit" | "max_steps"

10. RuntimeConfig: 运行时配置

10.1 从 YAML 构建

@dataclass
class RuntimeConfig:
    """运行时配置，从 YAML + CLI 参数 + 环境变量合并而来"""

    # LLM
    llm: LLMConfig

    # MCP
    mcp: Dict[str, MCPNodeConfig]

    # Memory
    memory: MemoryConfig

    # CLI Agents
    cli: Dict[str, CLIAgentConfig]

    # Termination
    termination: TerminationConfig

    # Trace
    trace: TraceConfig

    # ReAct
    react: ReActConfig

    @staticmethod
    def from_yaml(path: str, **overrides) -> "RuntimeConfig":
        """
        从 YAML 配置文件 + CLI 覆盖 + 环境变量构建运行时配置。

        优先级:
            CLI 参数 > 环境变量 > YAML 配置 > 默认值
        """

@dataclass
class LLMConfig:
    provider: str = "anthropic"
    model: str = "claude-sonnet-4-20250514"
    temperature: float = 0.0
    max_tokens: int = 1024
    api_key: str = ""       # 从环境变量读取
    base_url: str = ""      # 自定义端点

@dataclass
class MemoryConfig:
    enabled: bool = False
    strategy: str = "local"
    size: int = 20
    ttl: int = 3600
    redis_url: str = ""
    db_path: str = ""

@dataclass
class ReActConfig:
    max_steps: int = 10
    tool_timeout: int = 30
    observation_enabled: bool = True
    verbose: bool = False

@dataclass
class TerminationConfig:
    signals: List[str] = None   # 自定义终止信号
    implicit_detection: bool = True

11. 完整执行流程图

用户输入
   │
   ▼
┌─────────────────────┐
│  Runtime.run(input)  │
└────────┬────────────┘
         │
    ┌────▼────┐
    │ Executor │
    │ .reduce()│
    └────┬────┘
         │
    ┌────▼──────────────────────────────────────────────────┐
    │                   Match Term Type                      │
    ├───────────────┬──────────┬──────────┬─────────────────┤
    │ConversationLam│  Compose  │  Loop   │ Tool/Route/...  │
    │  (preferred)  │           │(ReAct)  │                 │
    │  Lam (legacy) │           │         │                 │
    └───────┬───────┴─────┬────┴────┬────┴─────────────────┘
            │             │         │
    ┌───────▼───────┐┌───▼───┐ ┌──▼──────────────────────┐
    │ConversationLam││ f>>g  │ │    ReActEngine.run()     │
    │   .apply()    ││ 链式  │ │                           │
    │  ┌──────────┐ ││ 规约  │ │  ┌──────────────────────┐│
    │  │ manage   │ │└───┬───┘ │  │ Step Loop (Y 展开)   ││
    │  │ context  │ │    │     │  │                        ││
    │  └────┬─────┘ │    │     │  │ 1. THINK (Conv.Lam)   ││
    │  ┌────▼─────┐ │    │     │  │ 2. PARSE (Action)     ││
    │  │ Provider  │ │    │     │  │ 3. ROUTE (CASE)       ││
    │  │  .chat()  │ │    │     │  │ 4. INVOKE (Tool)      ││
    │  └────┬─────┘ │    │     │  │    ├─ MCP (HTTP)       ││
    │       │       │    │     │  │    ├─ CLI (subproc)    ││
    │  ┌────▼──────┐│    │     │  │    └─ terminate(λx.x)  ││
    │  │LLM Call   ││    │     │  │ 5. OBSERVE (format)    ││
    │  │ Anthropic ││    │     │  │ 6. UPDATE (Memory)     ││
    │  │ ClaudeCode││    │     │  │ 7. CHECK (terminate?)  ││
    │  │ Ollama/.. ││    │     │  │                        ││
    │  └────┬──────┘│    │     │  │  ── Loop or Return ──  ││
    │       │       │    │     │  └──────────────────────┘ │
    └───────┼───────┘    │     └──────────────┬────────────┘
         │                                │
    ┌────▼────────────────────────────────▼───────┐
    │              Context / Trace                  │
    │  ctx.log(term, input, output, duration, ...)  │
    └────┬────────────────────────────────────────┘
         │
    ┌────▼─────────┐
    │ Memory Update │
    │ auto_save()   │
    └────┬─────────┘
         │
    ┌────▼────┐
    │  Result  │
    └─────────┘

12. 与 from_config 的集成

12.1 编译 + 运行时的衔接

# 目前: from_config 返回 Term，Term 自己执行自己
agent = from_config("config.yml")
result = agent("input")  # Term.__call__ → Term.apply

# 新方案: from_config 返回 Term，Runtime 执行 Term
agent = from_config("config.yml")
runtime = Runtime.from_config("config.yml")
result = runtime.run(agent, "input")

# 或者一步到位:
result = Runtime.execute("config.yml", "input")

12.2 Runtime.execute 一站式入口

class Runtime:
    """
    lambdagent 运行时。

    Lambda 语义:
        Runtime = (Executor, Γ, trace)
        Runtime.execute(config, input) =
            let term = compile(config) in
            let ctx  = Context.new() in
            Executor.reduce(term, input, ctx)
    """

    @staticmethod
    def execute(
        config_path: str,
        input: str,
        **overrides,
    ) -> RuntimeResult:
        """
        一站式执行：编译 + 初始化运行时 + β-规约。

        等价于:
            from_config(config) → term
            Runtime(config) → runtime
            runtime.run(term, input) → result

        Returns:
            RuntimeResult(result, trace, stats)
        """
        # 1. 加载配置
        config = RuntimeConfig.from_yaml(config_path, **overrides)

        # 2. 编译 Lambda 项
        term = from_config(config_path)

        # 3. 初始化运行时
        runtime = Runtime(config)

        # 4. 执行
        return runtime.run(term, input)

@dataclass
class RuntimeResult:
    result: str              # 最终结果
    trace: List[TraceRecord] # β-规约追踪
    stats: TraceStats        # 统计信息
    context: Context         # 执行上下文（含 Memory 最终状态）

13. 实现优先级

13.1 分层实现计划

P0 — 能跑起来 (MVP)
├── ActionParser (JSON 解析 + 关键词兜底)         ~150 行
├── MCPClient (同步 HTTP，单连接)                  ~200 行
├── TerminationOracle (显式 + 隐式)               ~80 行
├── ReActEngine (7 阶段完整流程)                   ~300 行
├── LocalMemory (dict + TTL + LRU)                ~100 行
└── LLMAdapter (Anthropic only)                   ~80 行
                                            合计: ~910 行

P1 — 能用起来 (Production)
├── RedisMemory                                   ~150 行
├── 多 Provider (OpenAI, DashScope)               ~200 行
├── CLI Agent 集成到 ReActEngine                  ~100 行
├── TraceStore (JSON 导出 + 统计)                 ~150 行
├── 错误重试 + 降级                               ~100 行
└── Runtime.execute 一站式入口                    ~80 行
                                            合计: ~780 行

P2 — 好用起来 (Polish)
├── MCP SSE 流式                                  ~150 行
├── 异步执行 (asyncio)                            ~200 行
├── Tool Schema 自动发现                          ~100 行
├── Memory auto_save 策略 2 (LLM 摘要)            ~80 行
├── Trace 火焰图 / 时间线可视化                    ~150 行
└── REPL 增强 (热更新/断点/回放)                  ~200 行
                                            合计: ~880 行

13.2 模块依赖图

Runtime.execute
    │
    ├── from_config (已有)
    │
    └── Runtime.run
         │
         └── Executor.reduce
              │
              ├── ConversationLam      ← 已完成
              │   └── LLMProvider      ← 已完成
              │       ├── ClaudeCodeProvider   (session persistence)
              │       ├── AnthropicProvider     (Messages API)
              │       └── OpenAICompatProvider  (Ollama/OpenAI/...)
              │
              ├── ReActEngine          ← P0
              │   ├── ActionParser     ← P0
              │   ├── TerminationOracle← P0
              │   ├── MCPClient        ← P0
              │   │   └── MCPPool      ← P2
              │   ├── CLIAgent         ← P1 (已有 shell_tool.py)
              │   └── ObservationFmt   ← P0 (内联)
              │
              ├── MemoryBackend        ← P0 (Local), P1 (Redis)
              │
              └── TraceStore           ← P1

14. 文件结构

lambdagent/
├── __init__.py              # 已有
├── __main__.py              # 已有
├── core.py                  # 已有 (Term, Context, TraceEntry)
├── primitives.py            # 已有 (Lam, Compose, Loop, Tool, ...)
├── extensions.py            # 已有 (Memory, Route, Guard, Par)
├── conversation.py          # 新增: ConversationLam (对话感知 Lambda 抽象)
├── dataset.py               # 已有
├── fromconfig/
│   ├── compiler.py          # 已有 (YAML → Term 编译器, 含 _create_provider)
│   ├── errors.py            # 已有
│   └── schema.py            # 已有
├── providers/               # 新增: LLM Provider 层
│   ├── base.py              # LLMProvider ABC + ProviderConfig + ProviderError
│   ├── claude_code_provider.py  # ClaudeCodeProvider (session persistence)
│   ├── anthropic_provider.py    # AnthropicProvider (Messages API)
│   └── openai_compat_provider.py # OpenAICompatProvider (Ollama/OpenAI/...)
├── lint.py                  # 已有
│
├── runtime/                 # ← 新增: 运行时
│   ├── __init__.py
│   ├── executor.py          # Executor (β-规约引擎)
│   ├── react_engine.py      # ReActEngine (Y 组合子展开)
│   ├── action_parser.py     # ActionParser (LLM 输出解析)
│   ├── termination.py       # TerminationOracle (停止判定)
│   ├── llm_adapter.py       # LLMAdapter (多 Provider)
│   ├── mcp_client.py        # MCPClient (MCP 协议)
│   ├── memory_backends.py   # LocalMemory, RedisMemory
│   ├── trace_store.py       # TraceStore (追踪持久化)
│   ├── config.py            # RuntimeConfig (配置合并)
│   └── runtime.py           # Runtime (一站式入口)
│
├── cli/                     # 已有
│   ├── __init__.py
│   ├── main.py              # CLI 命令
│   └── shell_tool.py        # ShellTool + CLIAgent

15. 验收标准

15.1 MVP 验收 (P0)

测试: lambdagent run agent-cofig.yml "计算 1+2+3+4+5 的和"

期望行为:
  β[0] think     (2.5s)  → "我需要计算 1+2+3+4+5。让我调用 sum 工具。"
                            {"action": "everything_get_sum", "input": {"numbers": [1,2,3,4,5]}}
  β[1] parse     (0.0s)  → Action(tool="everything_get_sum", input={...})
  β[2] invoke    (1.0s)  → MCP POST https://ai-paas.../mcp/airouting
                            → "15"
  β[3] observe   (0.0s)  → "Observation: The sum is 15."
  β[4] think     (2.0s)  → "结果是 15。任务完成。"
                            {"action": "terminate", "answer": "1+2+3+4+5 = 15"}
  β[5] terminate (0.0s)  → (base case: λx.x)

  Result: 1+2+3+4+5 = 15
  (6 β-reductions, 5.5s, ~800 tokens)

15.2 各组件验收

组件	测试	通过标准
ActionParser	10 种 LLM 输出格式	8/10 正确解析
MCPClient	调用真实 MCP 端点	返回非 stub 结果
TerminationOracle	显式 + 隐式 + 强制	3/3 正确判定
ReActEngine	完整 ReAct 循环	在 maxSteps 内完成
LocalMemory	写入 + 读取 + TTL + LRU	全部通过
LLMAdapter	调用 Anthropic API	返回有效 response
Runtime.execute	端到端执行	从 YAML 到结果，无手动步骤

15.3 端到端验收

# 测试 1: simple agent
lambdagent run simple.yml "Hello" --trace
# 期望: 1 步 β-规约，返回 LLM 输出

# 测试 2: react agent with terminate
lambdagent run react.yml "1+1=?" --trace
# 期望: 2-5 步，最终调用 terminate

# 测试 3: react agent with MCP tool
lambdagent run agent-cofig.yml "计算 1+2+...+100" --trace
# 期望: 调用 MCP 工具，返回 5050

# 测试 4: chain agent
lambdagent run chain.yml "长文本..." --trace
# 期望: N 步 β-规约，N = chain steps 数

# 测试 5: REPL with memory
lambdagent repl agent-cofig.yml
λ> 记住：我的名字是 Alice
λ> 我叫什么？
# 期望: "Alice"（Memory 跨轮次保持）

# 测试 6: CLI agent 互调
lambdagent run master.yml "研究量子计算" --tool sub="lambdagent run worker.yml --quiet -"
# 期望: master 调用 worker，worker 返回结果

16. Lambda 对应速查

运行时概念               Lambda 演算对应
──────────              ──────────────
Executor.reduce          β-规约  (λx.M)N → M[x:=N]
ReActEngine.run          Y_n(g)(x) 展开
_step (单步)             一次 β-规约
ActionParser.parse       模式匹配 (CASE)
terminate tool           λx.x (identity = base case)
TerminationOracle        base case 检测
LLMAdapter.call          (λ_D x) → F_{M,D}(x)
MCPClient.invoke         Tool(name, fn)(input)
CLIAgent.__call__        跨进程 β-规约
MemoryBackend.read       Γ.lookup(x)
MemoryBackend.write      Γ' = Γ ∪ {k: v}
Context.trace            β-规约链记录
RuntimeConfig            Γ₀ (初始环境)
Runtime.execute          eval(compile(source), Γ₀)

17. 多智能体模块 (multiagent.py)

5 个新构造，扩展 Lambda 演算至 π-演算（进程演算）级别。

17.1 Channel / Send / Receive — π-calculus 通道

Lambda + π 语义:
    Channel()      = 创建通道 ν(c)
    Send(agent, c) = λx. let v = agent(x) in c!(v); v
    Receive(c, h)  = λ_. let v = c?() in h(v)

Channel 是线程安全的消息队列 (queue.Queue)，支持：

有缓冲/无缓冲通道（capacity 参数）
阻塞/超时读写
历史记录（direction, msg, timestamp）
close() 关闭通道

Send 和 Receive 是 Term 子类，包装 Channel 操作为可组合的 Lambda 项。

17.2 SharedMemory — 共享环境

Lambda 语义:
    SharedMemory(store) = Γ_shared
    sm.wrap(agent)      = λx. agent(x) [Γ ∪ Γ_shared]
    sm.read(key)        = Γ_shared(key)
    sm.write(key, v)    = Γ_shared[key ↦ v]

线程安全（threading.RLock），支持 append_only 模式（对应 Preservation 定理 Σ' ⊇ Σ 约束 — 已有 key 不可改变类型）。

17.3 GroupChat — 群组对话

Lambda 语义:
    GroupChat([a,b,c], scheduler, n) =
        Y_n(λself.λstate.
            let speaker = scheduler(state) in
            let msg = speaker(state) in
            IF done(state') THEN state' ELSE self(state')
        )

这是 Loop + Route 的组合，不引入新 Lambda 构造。调度策略：

"round_robin": 固定轮流
"random": 随机选择
Term: LLM 分类器动态选择发言者

终止条件：包含 "CONSENSUS"/"DONE"/"TERMINATE" 等关键词，或自定义 termination 函数。

17.4 Handoff — 动态委派

Lambda 语义:
    Handoff(selector, registry) =
        λx. let target = selector(x) in
             registry[target](x)

与 Route 的区别：Route 是编译时静态路由，Handoff 是运行时动态路由。支持：

register(name, agent) / unregister(name) — 运行时动态扩展路由表
精确匹配 + 模糊匹配
fallback Agent

17.5 AsyncPar — 真并行

Lambda 语义:
    AsyncPar(f, g) = λx. let (r₁, r₂) = concurrent(f(x), g(x)) in (r₁, r₂)

使用 ThreadPoolExecutor 并发执行。与 Par 的区别：Par 是顺序执行，AsyncPar 是线程池真并发。支持 | 操作符展平。

17.6 并发实现细节

Par vs AsyncPar 对比

Par (假并行):
    def apply(self, input, ctx):
        return tuple(a.apply(input, ctx) for a in self.agents)
    # generator 顺序执行: a1 完成 → a2 完成 → a3 完成
    # 3 个 Agent 各 1s → 总共 3s

AsyncPar (真并行):
    def apply(self, input, ctx):
        with ThreadPoolExecutor(max_workers=len(self.agents)) as executor:
            futures = {executor.submit(_run, i, a): i for i, a in enumerate(self.agents)}
            for future in as_completed(futures):
                idx, result = future.result()
                results[idx] = result
        return tuple(results)
    # 线程池并发: a1, a2, a3 同时开始
    # 3 个 Agent 各 1s → 总共约 1s (3x 加速)

为什么选择线程池而非 asyncio / multiprocessing

LLM API 调用特征:
  - I/O 密集（等 HTTP 响应 2-3 秒）
  - CPU 几乎不用（不做矩阵运算）
  - Python GIL 对 I/O 等待无影响

ThreadPoolExecutor:  ✅ 最优选择，简单且高效
asyncio:             ⚠️ 更高效但所有代码需 async/await，破坏 Term.apply() 接口
multiprocessing:     ❌ 过重，跨进程序列化 Context/Term 困难

Channel 线程安全实现

class Channel:
    _queue: queue.Queue    # Python 标准库线程安全队列
    _closed: bool
    _lock: threading.Lock  # 保护 history 记录

    # capacity=0 → 无缓冲（同步模式）
    #   send() 阻塞直到 receive() 被调用
    #   对应 π-calculus 的同步通信: c!(v) 与 c?(x) 握手
    #
    # capacity>0 → 有缓冲（异步模式）
    #   send() 不阻塞（直到缓冲区满）
    #   对应 π-calculus 的异步通信: 有界邮箱

SharedMemory 线程安全实现

class SharedMemory:
    _store: Dict[str, Any]
    _lock: threading.RLock           # 可重入锁（同一线程可多次获取）
    _type_registry: Dict[str, type]  # 首次写入的类型记录

    def write(self, key, value):
        with self._lock:  # 加锁
            # append_only 模式：类型检查（Σ'⊇Σ）
            if self._append_only and key in self._type_registry:
                expected = self._type_registry[key]
                if not isinstance(value, expected):
                    raise TypeError(...)  # 类型违反，对应 Preservation 定理
            self._store[key] = value

RLock（可重入锁）而非 Lock：因为 Agent 内部可能嵌套调用 read() 和 write()，普通 Lock 会死锁。

A2AServer 并发模型

class A2AServer:
    # 基于 http.server.HTTPServer + threading
    # 每个入站 HTTP 请求在独立线程中处理
    # 每个请求 = 一次远程 β-规约

    def start(self, background=True):
        self._server = HTTPServer((host, port), Handler)
        if background:
            thread = Thread(target=self._server.serve_forever, daemon=True)
            thread.start()  # 后台运行，不阻塞主线程

并发安全总结

组件	机制	粒度	安全保证
AsyncPar	ThreadPoolExecutor	每个 Agent 一个线程	结果按原始顺序返回
Channel	queue.Queue	消息级别	FIFO，阻塞/超时
SharedMemory	threading.RLock	读写操作级别	互斥访问 + 类型安全
A2AServer	HTTPServer + threading	每个请求一个线程	无状态处理
Context.trace	list.append	追加级别	Python list.append 是原子的

18. Skill 系统 (skills.py)

18.1 核心概念

Lambda 语义:
    Skill     = let name = λx.body in ...   (命名 Lambda 项 + 元数据)
    Registry  = Γ_skills : Name → Skill     (技能注册表 = 特殊环境)
    Discovery = Route(LLM, Γ_skills)        (LLM 从注册表选择技能)

18.2 Skill

Skill 继承 Term，额外提供：

description — 自然语言描述（用于 LLM 发现）
signature — 类型签名 SkillSignature(input_type, output_type)
tags — 标签（用于搜索过滤）
examples — 使用示例（用于 few-shot）
bind(**kwargs) — 偏应用（柯里化）
>> 组合时进行类型兼容性检查
stats — 使用统计（调用次数、平均耗时）
to_dict() — 序列化

18.3 SkillRegistry

单例模式的全局注册表。支持：

register(skill) / register_pack(pack)
search(query, tags) — 文本 + 标签搜索
discover(task, classifier) — LLM 驱动的技能发现
build_route() — 构建为 Route 可用的字典
stats() — 注册表统计

18.4 SkillAgent

SkillAgent(classifier, registry) =
    λx. let skill = discover(classifier, registry, x) in skill(x)

Handoff 的技能化版本：自动从 Registry 发现并执行最佳 Skill。

18.5 @skill 装饰器

@skill("summarize", "Summarize text", tags=["writing"])
def summarize(x):
    return f"Summary: {x[:50]}..."

自动包装为 Skill 并注册到全局 Registry。

19. MCP Client (mcp_client.py)

19.1 架构

MCPServer ──→ MCPTransport (HTTP / stdio)
    │
    ├── list_tools()    → tools/list
    ├── call_tool()     → tools/call
    ├── read_resource() → resources/read
    │
    ├── to_tool(name)   → MCPTool (lambdagent Term)
    ├── to_tools()      → [MCPTool, ...]
    └── to_route_dict() → {name: MCPTool, ...}

19.2 传输层

MCPHttpTransport — Streamable HTTP (MCP 2025-11-25 spec), JSON-RPC 2.0 over POST
MCPStdioTransport — 本地子进程 stdin/stdout 通信

19.3 MCPTool

Lambda 语义:
    MCPTool(server, name) = Tool(name, λx. mcp_call(server, name, parse(x)))

输入自动解析：str → JSON 尝试 → 单参数推断 → {"input": x} fallback。支持重试。

19.4 便利函数

tools = mcp_tools("http://localhost:3000/mcp")     # 获取所有工具
search = mcp_tool("http://localhost:3000/mcp", "search")  # 获取单个工具

20. Checkpoint (checkpoint.py)

20.1 核心概念

Lambda 语义:
    Checkpoint = (Γ, trace, Γ_shared, last_input, n_steps)
    save(ctx)  = serialize(Γ, trace) → JSON 文件
    load(path) = deserialize(JSON) → Context

20.2 Checkpoint 类

包含：Context 全部状态 (bindings, trace, memory)、SharedMemory 状态、metadata、last_input、step_count。

序列化格式：JSON，带版本号 (CHECKPOINT_VERSION = "1.0.0")，向后兼容检查。

20.3 CheckpointManager

管理多个 checkpoint：

save(ctx, description) — 保存新 checkpoint（自动编号 cp_001.json, cp_002.json, ...）
list() — 列出所有 checkpoint 摘要
latest() — 获取最新
rollback() — 回退到上一个
max_checkpoints — 自动清理旧 checkpoint

20.4 序列化实现细节

JSON 格式（checkpoint 文件结构）：

{
  "version": "1.0.0",
  "timestamp": 1711454400.0,
  "description": "after research step",
  "last_input": "分析 AI Agent 市场",
  "context": {
    "bindings": {"key": "value"},
    "memory": {"session_id": "demo_001"},
    "trace": [
      {
        "term_name": "research",
        "term_id": "abc12345",
        "input": "AI Agent 市场",
        "output": "研究结果: 3 个关键发现",
        "duration_ms": 2800.5,
        "model": "claude-sonnet",
        "tokens_used": 150
      }
    ]
  },
  "shared_memories": {
    "main": {"counter": 42, "trend": "上升"}
  },
  "stats": {
    "step_count": 5,
    "total_time_ms": 12400.0
  }
}

序列化规则：

_serialize_value(v):
  str/int/float/bool/None → 直接保留
  list/tuple              → 递归序列化每个元素
  dict                    → 递归序列化每个值，key 转 str
  其他 (Term, 函数等)     → str(v)  (不可序列化的降级为字符串)

版本兼容策略：

加载时检查 version:
  主版本号不同 → CheckpointVersionError（不兼容）
  次版本号不同 → 警告但继续（向后兼容）
  修订号不同   → 静默通过

CheckpointManager 文件命名：

./checkpoints/my_agent/
  ├── cp_001.json    ← 第 1 个 checkpoint
  ├── cp_002.json    ← 第 2 个 checkpoint
  └── cp_003.json    ← 第 3 个（最新）

max_checkpoints=5 时，超出的旧文件自动删除（FIFO）
rollback() = 删除最新 + 加载倒数第二个

20.5 便利函数

save_context(ctx, "checkpoint.json", last_input="input")
ctx = load_context("checkpoint.json")

21. A2A Protocol (a2a.py)

21.1 架构

lambdagent Skill ──→ AgentCard (JSON) ──→ A2AServer (HTTP)
                                              │
                                        A2AClient (Term)
                                              │
                               远程 Agent 封装为本地 Term

21.2 AgentCard

A2A Agent 能力描述文档，对应 /.well-known/agent.json。包含 name, description, url, skills, input/output modes, authentication。扩展字段 x-lambdagent 保存 lambda_type 和 tags。

映射：Skill -> AgentCard，SkillRegistry -> AgentCard（整个注册表打包为一个 Card）。

21.3 A2AServer

Lambda 语义:
    A2AServer(agent) = HTTP 服务器
        GET  /.well-known/agent.json → AgentCard
        POST /                        → JSON-RPC (tasks/send, tasks/get)

Task 生命周期：submitted -> working -> completed | failed | canceled。

21.4 A2AClient

Lambda 语义:
    A2AClient(url) = Tool(agent_name, λx. a2a_send_task(url, x))

远程 Agent 封装为本地 Term，可参与 >> / Route / Loop 等组合。

22. RAG (rag.py)

22.1 核心概念

Lambda 语义:
    RAGTool(store, k)      = Tool("rag", λx. retrieve(store, x, k))
    AgenticRAG(agent, rag)  = λx. IF need_rag(x) THEN agent(x + rag(x)) ELSE agent(x)

RAG 不引入新 Lambda 构造——它是 Tool 的特化版本。

22.2 向量存储

两个后端，接口统一（add / add_many / search）：

SimpleVectorStore — 零依赖 TF-IDF 实现

核心算法：TF-IDF 余弦相似度

文档入库流程:
  "Lambda calculus was invented by Church"
       ↓ _tokenize()
  ["lambda", "calculus", "invented", "church"]  (小写 + 去停用词)
       ↓ _build_tfidf()
  {lambda: 0.42, calculus: 0.38, invented: 0.31, church: 0.35}  (稀疏向量)

查询流程:
  "What is lambda calculus?"
       ↓ _text_to_tfidf()
  {lambda: 0.71, calculus: 0.71}
       ↓ _cosine_similarity() 与每个文档向量
  doc_1: 0.89 ← 最相关
  doc_2: 0.12
  doc_3: 0.03
       ↓ sort + top_k
  返回 [doc_1]

三步算法详解：

Step 1: TF (词频) — 词在文档中的重要性
    TF(word, doc) = 该词出现次数 / 文档总词数
    "lambda calculus lambda" → TF("lambda") = 2/3 = 0.67

Step 2: IDF (逆文档频率) — 词在语料中的稀有度
    IDF(word) = log(总文档数 / 包含该词的文档数) + 1
    5 篇文档中 "lambda" 出现在 2 篇 → IDF = log(5/3) + 1 = 1.51
    "the" 出现在 5 篇 → IDF = log(5/6) + 1 = 0.82 (常见词权重低)

Step 3: 余弦相似度 — 两个向量的夹角
    cosine(A, B) = (A · B) / (|A| × |B|)
    = Σ(Aᵢ × Bᵢ) / (√Σ(Aᵢ²) × √Σ(Bᵢ²))
    范围: [0, 1]，1 = 完全相同，0 = 完全无关

分词实现：

def _tokenize(self, text):
    text = text.lower()
    tokens = re.findall(r'[a-z]+|[\u4e00-\u9fff]', text)  # 英文词 + 中文字
    stops = {"the", "a", "an", "is", "are", "in", "on", ...}  # 停用词
    return [t for t in tokens if t not in stops and len(t) > 1]

性能特征：

指标	SimpleVectorStore
入库复杂度	O(n × m)，n=文档数，m=平均词数
查询复杂度	O(n × v)，n=文档数，v=词汇表大小
空间复杂度	O(n × v) 稀疏矩阵（实际远小于此）
适用规模	< 1,000 文档
优势	零依赖，< 100 行核心代码
劣势	仅关键词匹配，不理解语义（"dog" ≠ "puppy"）

缓存机制：

self._tfidf_cache = None  # 首次查询时构建
# add() 后自动清除缓存
# 查询时若缓存存在则复用，避免重复计算 IDF

ChromaStore — ChromaDB 后端

# 需要: pip install chromadb
class ChromaStore:
    def __init__(self, collection_name, persist_directory=None):
        import chromadb
        if persist_directory:
            self._client = chromadb.PersistentClient(path=persist_directory)
        else:
            self._client = chromadb.Client()  # 内存模式
        self._collection = self._client.get_or_create_collection(collection_name)

与 SimpleVectorStore 对比：

维度	SimpleVectorStore	ChromaStore
向量化	TF-IDF（词频统计）	Embedding 模型（语义向量）
维度	稀疏，维度=词汇表大小	稠密，384-1536 维
语义理解	仅关键词匹配	语义相似（"dog" ≈ "puppy"）
依赖	零	chromadb + embedding model
持久化	无（内存）	SQLite + 文件系统
查询算法	暴力扫描 O(n)	ANN 近似最近邻 O(log n)
适用规模	< 1,000 文档	百万级

切换后端只需改一行：

rag = create_rag(docs, backend="simple")   # 开发
rag = create_rag(docs, backend="chroma")   # 生产

22.3 RAGTool

输出格式：[Source N, score=0.xxx] 文档内容（numbered / plain / json）。

22.4 AgenticRAG

Agent 自行决定何时检索（而非每次都检索）。decider 参数可以是 Python 函数或 LLM Agent。

22.5 便利函数

rag = create_rag(["doc1", "doc2", "doc3"], top_k=3, backend="simple")

23. 扩展后的 Lambda 对应速查

运行时概念                Lambda / π 演算对应
──────────               ──────────────────
Channel.send(v)           c!(v)       π-calculus 输出
Channel.receive()         c?(x).P     π-calculus 输入
SharedMemory.wrap(a)      a [Γ ∪ Γ_shared]  共享环境
GroupChat                 Y_n(Loop + Route)  群组对话
Handoff                   动态 CASE          运行时路由
AsyncPar                  concurrent(f(x), g(x))  并发 β-规约
Skill                     let name = term in ...   命名 Lambda 项
SkillRegistry             Γ_skills            技能环境
SkillAgent                discover >> execute  技能发现+执行
MCPTool                   Tool(name, mcp_call) MCP 工具封装
A2AClient                 Tool(name, a2a_call) 远程 Agent 封装
RAGTool                   Tool("rag", retrieve) 检索工具
AgenticRAG                IF need THEN rag+agent ELSE agent
Checkpoint.save           serialize(Γ, trace)  状态快照
Checkpoint.load           deserialize → Γ      状态恢复
SandboxedTool             λx.f(x) [f in sandbox]  隔离执行的 Tool
SecureExecutor            sandbox_all(Γ)       递归包裹所有 Tool
ResourceLimiter           RLIMIT constraints   POSIX 资源限制

24. Sandbox 运行时 (sandbox.py)

24.1 概述

Sandbox 模块提供 进程级隔离 的 Tool 执行环境。每个 SandboxedTool 在独立子进程中运行，通过 pickle IPC 通信，受 POSIX resource limits 约束。

Lambda 语义不变：SandboxedTool 仍是 Tool 的子类型，其指称语义与普通 Tool 完全相同。

⟦SandboxedTool(n, f, P)⟧ = λx.f(x)    [f 在 sandbox 中执行]

24.2 架构

调用方 (主进程)                     子进程 (sandbox)
─────────────                     ─────────────────
SandboxedTool.__call__(x)
    │
    ├── pickle.dumps(x) ──────→  stdin (子进程)
    │                              │
    │                              ├── ResourceLimiter.apply()
    │                              │     ├── RLIMIT_CPU   (timeout)
    │                              │     ├── RLIMIT_AS    (memory_mb)
    │                              │     ├── RLIMIT_NOFILE (max_fds)
    │                              │     └── RLIMIT_NPROC  (禁止 fork)
    │                              │
    │                              ├── f(x) 执行
    │                              │
    │                              └── pickle.dumps(result) ──→ stdout
    │
    ├── pickle.loads(result) ←────
    └── return result

24.3 SandboxPolicy 预设

三种安全策略预设，覆盖从严格到宽松的场景：

属性	`strict()`	`default()`	`permissive()`
`timeout`	5s	30s	300s
`memory_mb`	64	256	2048
`network`	False	False	True
`allow_subprocess`	False	False	True
`max_output_bytes`	4096	65536	1048576
`max_fds`	8	64	256

使用方式：

from lambdagent import SandboxPolicy, SandboxedTool

# 预设策略
strict = SandboxPolicy.strict()
default = SandboxPolicy.default()
permissive = SandboxPolicy.permissive()

# 自定义
custom = SandboxPolicy(timeout=10, memory_mb=128, network=False)

24.4 SandboxedTool

将任意函数包裹为隔离执行的 Tool：

from lambdagent import SandboxedTool, SandboxPolicy

tool = SandboxedTool("risky_calc", lambda x: eval(x), policy=SandboxPolicy.strict())
result = tool("2 + 3")  # 在隔离子进程中执行

@sandboxed 装饰器提供一行创建方式：

from lambdagent import sandboxed

@sandboxed(timeout=10, memory_mb=128)
def my_tool(x):
    return expensive_computation(x)

24.5 SecureExecutor

递归遍历 Term 树，将所有 Tool 节点替换为 SandboxedTool：

from lambdagent import SecureExecutor, SandboxPolicy

executor = SecureExecutor(policy=SandboxPolicy.default())
secure_term = executor.sandbox_all_tools(term_tree)
# term_tree 中所有 Tool 节点已被包裹为 SandboxedTool

24.6 ResourceLimiter

在子进程启动时，通过 POSIX resource 模块施加硬性限制：

RLIMIT 常量	对应 SandboxPolicy 属性	作用
`RLIMIT_CPU`	`timeout`	CPU 时间上限（秒）
`RLIMIT_AS`	`memory_mb`	地址空间上限
`RLIMIT_NOFILE`	`max_fds`	文件描述符上限
`RLIMIT_NPROC`	`allow_subprocess=False` → 0	禁止创建子进程

24.7 异常体系

SandboxViolation (base)
├── TimeoutViolation      # RLIMIT_CPU 或 subprocess timeout 触发
├── MemoryViolation       # RLIMIT_AS 超限
└── OutputViolation       # 输出字节数超过 max_output_bytes

24.8 升级路径 (L1 → L2 → L3)

级别	方式	适用场景
L1	`@sandboxed` 装饰器	单个 Tool 快速隔离
L2	`SandboxedTool(name, fn, policy)`	精细控制单个 Tool 的策略
L3	`SecureExecutor.sandbox_all_tools(tree)`	整棵 Term 树一键安全化

24.9 Lambda 对应

运行时概念                Lambda 对应
──────────               ──────────
SandboxedTool(n,f,P)     λx.f(x) [f in sandbox]  — Tool 类型不变
SecureExecutor           map(sandbox_wrap, Tools(Γ)) — 遍历环境
SandboxPolicy            执行策略参数 — 不影响 λ 语义
ResourceLimiter          操作语义层面的资源约束
SandboxViolation         β-规约失败（异常终止）

runtime-spec.md 80 KB Permalink History Raw