# 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。 ```python 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 类型的规约规则 ```python 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 调用 ```python 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 调用 ```python 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 调用的最底层抽象——只负责消息传输，不负责历史管理。 ```python 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 ` (原生) | 需要完整上下文记忆的复杂 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 中的作用 ```python # 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 完整规格 ```python 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 个子阶段 ```python @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`) ```python 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 输出: "sum{\"numbers\": [1,2,3]}" 提取: 同上 优先级 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 接口规格 ```python 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 校验 ```python 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 终止条件层次 ```python 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 强制终止时的答案提取 ```python 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()` 工厂函数创建，不通过注册表 ```python 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 ```python 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 持久化 ```python class ClaudeCodeProvider(LLMProvider): """ 基于 Claude Code CLI 的 Provider。 传输方式: subprocess 调用 `claude` CLI 会话持久化: --resume 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 --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 ```python 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 接口 ```python class MCPClient: """ MCP 协议客户端。 Lambda 语义: MCPClient = λ(server, tool, input). HTTP_POST(url, {tool, input}) 对应 YAML: app.mcp.custom.nodes..url app.mcp.custom.nodes..endpoint app.mcp.custom.nodes..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 连接管理 ```python class MCPConnectionPool: """ MCP 连接池。 每个 MCP server 维护一个持久连接。 支持: - HTTP (请求/响应) - SSE (服务端推送，用于流式工具结果) - WebSocket (双向，用于长时间运行的工具) Lambda 语义: Pool = Memory(connections, {server → client}) 即: 连接池是一种环境扩展——把已建立的连接存入 Γ """ ``` --- ## 8. Memory Backend: 环境 Γ 的持久化 ### 8.1 接口 ```python 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 (默认) ```python class LocalMemory(MemoryBackend): """ 进程内记忆后端。 数据结构: OrderedDict (保持插入顺序，支持 LRU 淘汰) TTL: 通过时间戳检查 Size: 通过 OrderedDict.popitem(last=False) 淘汰最旧条目 Lambda 语义: Γ 存在 Python 进程的堆内存中。进程退出则 Γ 丢失。 """ ``` ### 8.3 RedisMemory (生产) ```python 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: 自动信息提取 ```python 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 追踪数据模型 ```python @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 接口 ```python 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 构建 ```python @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 编译 + 运行时的衔接 ```python # 目前: 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 一站式入口 ```python 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 端到端验收 ```bash # 测试 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 线程安全实现 ```python 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 线程安全实现 ```python 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 并发模型 ```python 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 装饰器 ```python @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 便利函数 ```python 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 文件结构）：** ```json { "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 便利函数 ```python 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 = 完全无关 ``` **分词实现：** ```python 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"）| **缓存机制：** ```python self._tfidf_cache = None # 首次查询时构建 # add() 后自动清除缓存 # 查询时若缓存存在则复用，避免重复计算 IDF ``` #### ChromaStore — ChromaDB 后端 ```python # 需要: 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 文档 | 百万级 | **切换后端只需改一行：** ```python 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 便利函数 ```python 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 | 使用方式： ```python 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： ```python from lambdagent import SandboxedTool, SandboxPolicy tool = SandboxedTool("risky_calc", lambda x: eval(x), policy=SandboxPolicy.strict()) result = tool("2 + 3") # 在隔离子进程中执行 ``` `@sandboxed` 装饰器提供一行创建方式： ```python from lambdagent import sandboxed @sandboxed(timeout=10, memory_mb=128) def my_tool(x): return expensive_computation(x) ``` ### 24.5 SecureExecutor 递归遍历 Term 树，将所有 `Tool` 节点替换为 `SandboxedTool`： ```python 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 β-规约失败（异常终止） ```