从 Lambda 项到真实执行:Agent 运行时规格文档
版本: 1.0 | 前置: FROM_CONFIG_SPEC.md | 理论基础: LDS ≡ Lambda Calculus
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 检测
┌─────────────────────────────────────────────────────┐
│ 用户接口层 │
│ 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 │
└─────────────────────────────────────────────────────┘
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)
Phase 6.5 引入了统一 Engine 抽象,支持 RecursiveEngine (Python 调用栈) 和 CEKEngine (Agent CEK Machine) 的运行时切换。通过 YAML 配置 runtime.engine 字段或 Python API engine_mode 参数选择。CEKEngine 提供逐步成本监控、暂停/恢复和循环检测能力。详见 docs/yaml-config.md §4。
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 类型有不同的规约规则。
"""
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)
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
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(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 在每一步都能看到完整对话, 而不是只看到一个通过字符串拼接压缩的状态。
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 | 传输方式 | 会话持久化 | 适用场景 |
|---|---|---|---|
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 (对话层) LLMProvider (传输层)
───────────────────── ─────────────────────
管理 messages 列表 只关心 messages -> str
上下文窗口控制 不知道历史
摘要压缩旧消息 不知道 token 预算
记录 assistant 回复 不记录任何东西
输出解析 不解析输出
不同 provider 的会话持久化方式不同:
第 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 端自动保持完整上下文。
每次调用:
ConversationLam 将 full messages array 传给 provider.chat()
Provider 将 messages 转发给 HTTP API
API 端无状态 — 每次都是独立请求
上下���管理:
ConversationLam._manage_context() 在每次调用前检查 token 预算
超出时进行 sliding window 压缩
# 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 负责历史管理
ReAct 不是简单的 for step in range(n): body(result)。它是一个有内部结构的 Y 组合子展开,每一步包含 7 个子阶段:
Y 组合子的一次展开 (一步 β-规约) =
think → parse → route → invoke → observe → update → check
───── ───── ───── ────── ─────── ────── ─────
LLM 提取 选择 执行 格式化 写回 终止
推理 动作 工具 工具 观察结果 记忆 判定
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
@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,
)
_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)
LLM 输出的是自由文本。ActionParser 负责从中提取结构化的 Action(tool, input, thought)。
优先级 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"})
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]:
"""检测隐式终止信号"""
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
判定 Y 组合子是否到达 base case。在 Lambda 演算中,base case 是"不再调用 self 的分支"。在 Agent 中,对应"不再需要新工具调用"的时刻。
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
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)
v2 采用 LLMProvider 接口替代旧的 LLMAdapter。核心变化:
call(model, system, user) 变为 chat(messages: list[dict]) -> strConversationLam 负责,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)
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 (无状态)
"""
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
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 数组。
"""
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
class MCPConnectionPool:
"""
MCP 连接池。
每个 MCP server 维护一个持久连接。
支持:
- HTTP (请求/响应)
- SSE (服务端推送,用于流式工具结果)
- WebSocket (双向,用于长时间运行的工具)
Lambda 语义:
Pool = Memory(connections, {server → client})
即: 连接池是一种环境扩展——把已建立的连接存入 Γ
"""
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)
class LocalMemory(MemoryBackend):
"""
进程内记忆后端。
数据结构: OrderedDict (保持插入顺序,支持 LRU 淘汰)
TTL: 通过时间戳检查
Size: 通过 OrderedDict.popitem(last=False) 淘汰最旧条目
Lambda 语义: Γ 存在 Python 进程的堆内存中。进程退出则 Γ 丢失。
"""
class RedisMemory(MemoryBackend):
"""
Redis 记忆后端。
数据结构: Redis Sorted Set (score = timestamp,自动按时间排序)
TTL: Redis 原生 EXPIRE
Size: ZREMRANGEBYRANK 淘汰最旧条目
Scope: key prefix = agent_id + session_id
Lambda 语义: Γ 存在 Redis 中。跨进程、跨机器共享。
"""
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)
@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]
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"
@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
用户输入
│
▼
┌─────────────────────┐
│ 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 │
└─────────┘
# 目前: 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")
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 最终状态)
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 行
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
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
测试: 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)
| 组件 | 测试 | 通过标准 |
|---|---|---|
| ActionParser | 10 种 LLM 输出格式 | 8/10 正确解析 |
| MCPClient | 调用真实 MCP 端点 | 返回非 stub 结果 |
| TerminationOracle | 显式 + 隐式 + 强制 | 3/3 正确判定 |
| ReActEngine | 完整 ReAct 循环 | 在 maxSteps 内完成 |
| LocalMemory | 写入 + 读取 + TTL + LRU | 全部通过 |
| LLMAdapter | 调用 Anthropic API | 返回有效 response |
| Runtime.execute | 端到端执行 | 从 YAML 到结果,无手动步骤 |
# 测试 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 返回结果
运行时概念 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), Γ₀)
5 个新构造,扩展 Lambda 演算至 π-演算(进程演算)级别。
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 参数)close() 关闭通道Send 和 Receive 是 Term 子类,包装 Channel 操作为可组合的 Lambda 项。
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 不可改变类型)。
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 函数。
Lambda 语义:
Handoff(selector, registry) =
λx. let target = selector(x) in
registry[target](x)
与 Route 的区别:Route 是编译时静态路由,Handoff 是运行时动态路由。支持:
register(name, agent) / unregister(name) — 运行时动态扩展路由表fallback AgentLambda 语义:
AsyncPar(f, g) = λx. let (r₁, r₂) = concurrent(f(x), g(x)) in (r₁, r₂)
使用 ThreadPoolExecutor 并发执行。与 Par 的区别:Par 是顺序执行,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 加速)
LLM API 调用特征:
- I/O 密集(等 HTTP 响应 2-3 秒)
- CPU 几乎不用(不做矩阵运算)
- Python GIL 对 I/O 等待无影响
ThreadPoolExecutor: ✅ 最优选择,简单且高效
asyncio: ⚠️ 更高效但所有代码需 async/await,破坏 Term.apply() 接口
multiprocessing: ❌ 过重,跨进程序列化 Context/Term 困难
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 的异步通信: 有界邮箱
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 会死锁。
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 是原子的 |
Lambda 语义:
Skill = let name = λx.body in ... (命名 Lambda 项 + 元数据)
Registry = Γ_skills : Name → Skill (技能注册表 = 特殊环境)
Discovery = Route(LLM, Γ_skills) (LLM 从注册表选择技能)
Skill 继承 Term,额外提供:
description — 自然语言描述(用于 LLM 发现)signature — 类型签名 SkillSignature(input_type, output_type)tags — 标签(用于搜索过滤)examples — 使用示例(用于 few-shot)bind(**kwargs) — 偏应用(柯里化)>> 组合时进行类型兼容性检查stats — 使用统计(调用次数、平均耗时)to_dict() — 序列化单例模式的全局注册表。支持:
register(skill) / register_pack(pack)search(query, tags) — 文本 + 标签搜索discover(task, classifier) — LLM 驱动的技能发现build_route() — 构建为 Route 可用的字典stats() — 注册表统计SkillAgent(classifier, registry) =
λx. let skill = discover(classifier, registry, x) in skill(x)
Handoff 的技能化版本:自动从 Registry 发现并执行最佳 Skill。
@skill("summarize", "Summarize text", tags=["writing"])
def summarize(x):
return f"Summary: {x[:50]}..."
自动包装为 Skill 并注册到全局 Registry。
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, ...}
MCPHttpTransport — Streamable HTTP (MCP 2025-11-25 spec), JSON-RPC 2.0 over POSTMCPStdioTransport — 本地子进程 stdin/stdout 通信Lambda 语义:
MCPTool(server, name) = Tool(name, λx. mcp_call(server, name, parse(x)))
输入自动解析:str → JSON 尝试 → 单参数推断 → {"input": x} fallback。支持重试。
tools = mcp_tools("http://localhost:3000/mcp") # 获取所有工具
search = mcp_tool("http://localhost:3000/mcp", "search") # 获取单个工具
Lambda 语义:
Checkpoint = (Γ, trace, Γ_shared, last_input, n_steps)
save(ctx) = serialize(Γ, trace) → JSON 文件
load(path) = deserialize(JSON) → Context
包含:Context 全部状态 (bindings, trace, memory)、SharedMemory 状态、metadata、last_input、step_count。
序列化格式:JSON,带版本号 (CHECKPOINT_VERSION = "1.0.0"),向后兼容检查。
管理多个 checkpoint:
save(ctx, description) — 保存新 checkpoint(自动编号 cp_001.json, cp_002.json, ...)list() — 列出所有 checkpoint 摘要latest() — 获取最新rollback() — 回退到上一个max_checkpoints — 自动清理旧 checkpointJSON 格式(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() = 删除最新 + 加载倒数第二个
save_context(ctx, "checkpoint.json", last_input="input")
ctx = load_context("checkpoint.json")
lambdagent Skill ──→ AgentCard (JSON) ──→ A2AServer (HTTP)
│
A2AClient (Term)
│
远程 Agent 封装为本地 Term
A2A Agent 能力描述文档,对应 /.well-known/agent.json。包含 name, description, url, skills, input/output modes, authentication。扩展字段 x-lambdagent 保存 lambda_type 和 tags。
映射:Skill -> AgentCard,SkillRegistry -> AgentCard(整个注册表打包为一个 Card)。
Lambda 语义:
A2AServer(agent) = HTTP 服务器
GET /.well-known/agent.json → AgentCard
POST / → JSON-RPC (tasks/send, tasks/get)
Task 生命周期:submitted -> working -> completed | failed | canceled。
Lambda 语义:
A2AClient(url) = Tool(agent_name, λx. a2a_send_task(url, x))
远程 Agent 封装为本地 Term,可参与 >> / Route / Loop 等组合。
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 的特化版本。
两个后端,接口统一(add / add_many / search):
核心算法: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
# 需要: 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") # 生产
输出格式:[Source N, score=0.xxx] 文档内容(numbered / plain / json)。
Agent 自行决定何时检索(而非每次都检索)。decider 参数可以是 Python 函数或 LLM Agent。
rag = create_rag(["doc1", "doc2", "doc3"], top_k=3, backend="simple")
运行时概念 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 资源限制
Sandbox 模块提供 进程级隔离 的 Tool 执行环境。每个 SandboxedTool 在独立子进程中运行,通过 pickle IPC 通信,受 POSIX resource limits 约束。
Lambda 语义不变:SandboxedTool 仍是 Tool 的子类型,其指称语义与普通 Tool 完全相同。
⟦SandboxedTool(n, f, P)⟧ = λx.f(x) [f 在 sandbox 中执行]
调用方 (主进程) 子进程 (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
三种安全策略预设,覆盖从严格到宽松的场景:
| 属性 | 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)
将任意函数包裹为隔离执行的 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)
递归遍历 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
在子进程启动时,通过 POSIX resource 模块施加硬性限制:
| RLIMIT 常量 | 对应 SandboxPolicy 属性 | 作用 |
|---|---|---|
RLIMIT_CPU |
timeout |
CPU 时间上限(秒) |
RLIMIT_AS |
memory_mb |
地址空间上限 |
RLIMIT_NOFILE |
max_fds |
文件描述符上限 |
RLIMIT_NPROC |
allow_subprocess=False → 0 |
禁止创建子进程 |
SandboxViolation (base)
├── TimeoutViolation # RLIMIT_CPU 或 subprocess timeout 触发
├── MemoryViolation # RLIMIT_AS 超限
└── OutputViolation # 输出字节数超过 max_output_bytes
| 级别 | 方式 | 适用场景 |
|---|---|---|
| L1 | @sandboxed 装饰器 |
单个 Tool 快速隔离 |
| L2 | SandboxedTool(name, fn, policy) |
精细控制单个 Tool 的策略 |
| L3 | SecureExecutor.sandbox_all_tools(tree) |
整棵 Term 树一键安全化 |
运行时概念 Lambda 对应
────────── ──────────
SandboxedTool(n,f,P) λx.f(x) [f in sandbox] — Tool 类型不变
SecureExecutor map(sandbox_wrap, Tools(Γ)) — 遍历环境
SandboxPolicy 执行策略参数 — 不影响 λ 语义
ResourceLimiter 操作语义层面的资源约束
SandboxViolation β-规约失败(异常终止)