from-config-spec.md 68 KB
from_config() 编译器需求规格文档
YAML Agent Configuration → Lambda Term Compiler
版本: 2.0 | 状态: 需求定稿 | 上游理论: LDS ≡ Lambda Calculus
1. 概述
1.1 一句话定义
from_config(path) 是一个编译器，输入 YAML 格式的 Agent 配置文件，输出一个可执行的 lambdagent.Term（Lambda 项）。
1.2 核心等式
from_config("agent-config.yml") → Term → term(input) → result
↓ ↓ ↓
YAML 解析 Lambda 项 β-规约
等价于手写：
Memory(
Loop(
think >> Route(tools) >> observe,
max_steps=20
),
store=redis(20, 7200)
)
1.3 设计原则
| 原则 | 说明 |
|---|---|
| 每个 YAML 字段有且仅有一个 Lambda 语义 | 不存在"无法翻译"的配置项 |
| 编译结果可追踪 | 每一步执行 = 一次 β-规约，记录在 Context.trace |
| 编译结果可静态分析 | lint 工具能从 Term 结构推断配置问题 |
| 编译是确定性的 | 同一 YAML 总产出同一 Term 结构 |
| 编译器不执行 Agent | from_config 只构建 Term，不调用 LLM |
2. 输入规格: YAML Schema
2.1 完整 YAML Schema
# ═══ 必填字段 ═══
agentId: string # Agent 唯一标识符
name: string # Agent 可读名称
type: enum # "react" | "chain" | "simple" | "parallel" | "router"
systemPrompt: string # Lambda body（λ 抽象的函数体）
# ═══ 模型配置 ═══
model:
provider: string # "anthropic" | "openai" | "dashscope" | "ollama"
# | "claude-code" | "deepseek" | "moonshot" | "zhipu" | "custom"
name: string # 具体模型名 (e.g., "claude-sonnet-4-20250514")
temperature: float # [0.0, 2.0]，默认 0.3
maxTokens: int # 最大输出 token 数，默认 4096
topP: float # [0.0, 1.0]，可选
stopSequences: [string] # 停止序列，可选
baseUrl: string # 自定义 API 端点，可选
conversation: bool # 是否启用 ConversationLam 对话管理，默认 true
maxHistoryTokens: int # 对话历史最大 token 数，默认 min(contextWindow/2, 80000)
contextWindow: int # 模型上下文窗口大小，默认 200000
timeout: int # 单次 LLM 调用超时(秒)，默认 600
# ═══ ReAct 配置（type=react 时必填）═══
react:
maxSteps: int # Y 组合子最大展开次数，默认 10
observationEnabled: bool # 是否将工具结果追加到上下文，默认 true
toolTimeout: int # 单个工具调用超时(秒)，默认 30
verbose: bool # 是否打印中间步骤，默认 false
earlyStop: string # 提前终止条件表达式，可选
thinkPrompt: string # 覆盖 think 步骤的 prompt，可选
# ═══ Chain 配置（type=chain 时必填）═══
chain:
steps: # 有序步骤列表
- name: string # 步骤名称
prompt: string # 该步骤的 prompt
model: object # 可选覆盖模型配置
outputParser: string # "json" | "text" | "number" | "boolean"
guard: # 可选输出验证
validator: string # Python 表达式 (e.g., "len(x) > 100")
retry: int # 验证失败重试次数
fallback: string # 兜底值
# ═══ Router 配置（type=router 时必填）═══
router:
classifier: # 分类器配置
prompt: string # 分类 prompt
model: object # 可选模型覆盖
categories: [string] # 类别列表
routes: # 类别 → Agent 映射
<category>: # 每个 category 指向一个子 agent 配置
type: string
systemPrompt: string
# ... 子 agent 的完整配置
default: object # 无法分类时的默认 Agent，可选
# ═══ Parallel 配置（type=parallel 时必填）═══
parallel:
agents: # 并行 Agent 列表
- name: string
type: string
systemPrompt: string
# ... 子 agent 完整配置
merge: string # 结果合并策略: "tuple" | "concat" | "custom"
mergePrompt: string # merge=custom 时的合并 prompt
# ═══ MCP 工具配置 ═══
mcp:
onlineTool: # 在线 MCP 工具
<server-name>: # MCP 服务器名
- string # 工具名列表
localTools: [string] # 本地工具（"terminate" 是特殊的 base case）
policy:
mode: string # "auto" | "force" | "intelligence" | "disable"
maxConcurrent: int # 最大并行工具调用数，默认 1
retryOnFail: int # 工具调用失败重试次数，默认 0
# ═══ Memory 配置 ═══
memory:
enabled: bool # 是否启用
strategy: string # "local" | "redis" | "sqlite" | "custom"
size: int # 记忆容量（最多保留多少条）
ttl: int # 过期时间(秒)，0 = 永不过期
scope: string # "session" | "global" | "user"
customStore: string # strategy=custom 时的 Python 类路径
# ═══ Runtime 配置（可选）═══
runtime:
engine: string # "recursive" | "cek" | "adaptive"，默认 "recursive"
# 执行引擎选择。详见 docs/yaml-config.md §4
# ═══ Guard 配置（可选，包裹整个 Agent）═══
guard:
validator: string # Python 表达式
retry: int # 重试次数
fallback: string # 兜底策略: "error" | "empty" | "last"
# ═══ RAG 配置（可选）═══
rag:
enabled: bool
source: string # 知识库路径或 URL
topK: int # 检索数量，默认 3
chunkSize: int # 分块大小
backend: string # "simple" | "chroma"，默认 "simple"
minScore: float # 最低相似度阈值，默认 0.0
format: string # "numbered" | "plain" | "json"，默认 "numbered"
agentic: bool # true = AgenticRAG（Agent 自行决定是否检索）
persistDirectory: string # backend=chroma 时的持久化目录
# ═══ 外部服务配置 ═══
app:
mcp:
custom:
nodes:
<server-name>:
url: string # MCP 服务端点
endpoint: string # 路径
headers:
Authorization: string
timeout: int # 超时(秒)
2.2 YAML 字段分级
| 级别 | 字段 | Lambda 语义 | 编译行为 |
|---|---|---|---|
| 必须 | type, systemPrompt |
Agent 结构 + λ body | 缺失则报编译错误 |
| 核心 | model, react, mcp |
计算单元 + Y 参数 + Oracle | 缺失则用默认值 |
| 增强 | memory, guard, chain, router, parallel |
环境扩展 + 类型约束 + 组合 | 缺失则不包裹 |
| 元数据 | agentId, name, description |
不影响 Lambda 语义 | 只用于标识和追踪 |
| 运行时 | runtime, rag, app |
引擎选择 + 外部资源 | runtime.engine: 执行引擎选择 (recursive | cek | adaptive), 默认 recursive. 详见 docs/yaml-config.md §4 |
3. 输出规格: Term 结构
3.1 编译输出的类型签名
def from_config(path: str, **overrides) -> Term:
"""
编译 YAML 配置为 Lambda 项。
参数:
path: YAML 文件路径
**overrides: 运行时覆盖项
- model: 覆盖模型名称
- temperature: 覆盖温度
- max_steps: 覆盖最大步数
- tools: Dict[str, Callable] 注入自定义工具实现
- memory_store: Dict 注入初始记忆
返回:
Term — 可执行的 Lambda 项
异常:
ConfigError — YAML 解析失败
SchemaError — 必填字段缺失
CompileError — 语义不合法 (e.g., router 无 routes)
"""
3.2 五种 type 的编译目标
type: simple — 单个 λ 抽象
输入 YAML:
type: simple
systemPrompt: "Summarize text."
model: {name: claude-sonnet}
编译目标:
Lam("agent", prompt="Summarize text.", model="claude-sonnet")
Lambda:
λx. LLM_{θ,p}(x)
type: react — Y 组合子
输入 YAML:
type: react
react: {maxSteps: 20}
mcp:
onlineTool: {server: [search, calc]}
localTools: [terminate]
编译目标:
Memory(
Loop(
Tool("react_step", fn=react_step_fn),
condition=lambda r, s: s >= 19,
max_steps=20
),
store={...}
)
其中 react_step_fn 内部结构:
think(state) ← Lam: β-规约
→ Route(thought, { ← 广义 Church 布尔
"search": Tool("search", mcp),
"calc": Tool("calc", mcp),
"terminate": Tool("terminate", λx.x) ← base case
})
→ observation ← 工具结果
→ state ⊕ observation ← 状态拼接
Lambda:
Memory(
Y₂₀(λself.λstate.
let t = think(state) in
CASE t [
(search, Tool_MCP(t)),
(calc, Tool_MCP(t)),
(terminate, λx.x) ← base case
] >>
IF is_terminate THEN t
ELSE self(state ⊕ obs)
),
Γ ∪ store
)
type: chain — 函数组合链
输入 YAML:
type: chain
chain:
steps:
- name: extract
prompt: "Extract key facts."
- name: analyze
prompt: "Analyze the facts."
guard: {validator: "len(x) > 50", retry: 2}
- name: draft
prompt: "Write a draft."
编译目标:
Lam("extract", "Extract key facts.")
>> Guard(
Lam("analyze", "Analyze the facts."),
validator=lambda x: len(x) > 50,
retry=2
)
>> Lam("draft", "Write a draft.")
Lambda:
λx. draft(Guard(analyze, P)(extract(x)))
where P(r) = len(r) > 50
type: router — 广义 Church 布尔 (CASE)
输入 YAML:
type: router
router:
classifier:
prompt: "Classify: code/math/general"
categories: [code, math, general]
routes:
code: {type: simple, systemPrompt: "You are a coder."}
math: {type: simple, systemPrompt: "You are a mathematician."}
default: {type: simple, systemPrompt: "You are a helpful assistant."}
编译目标:
Route(
classifier=Lam("classifier", "Classify: code/math/general"),
routes={
"code": Lam("code_agent", "You are a coder."),
"math": Lam("math_agent", "You are a mathematician."),
},
default=Lam("default", "You are a helpful assistant.")
)
Lambda:
λx. CASE (classifier x) [
(code, code_agent x),
(math, math_agent x),
(_, default_agent x)
]
type: parallel — Church 对 (PAIR / PAR)
输入 YAML:
type: parallel
parallel:
agents:
- {name: researcher, systemPrompt: "Research the topic."}
- {name: critic, systemPrompt: "Critique the topic."}
merge: custom
mergePrompt: "Synthesize the research and critique."
编译目标:
Par(
Lam("researcher", "Research the topic."),
Lam("critic", "Critique the topic.")
) >> Lam("merge", "Synthesize the research and critique.")
Lambda:
λx. merge(PAIR (researcher x) (critic x))
4. 编译规则（完整翻译表）
4.1 一级字段编译规则
| YAML 字段 | 编译目标 | Lambda 语义 | 编译函数 |
|---|---|---|---|
systemPrompt + model |
ConversationLam(name, provider, prompt) |
λx. provider(history + x) |
_compile_lam() → _create_provider() |
type: react |
Loop(body, cond, maxSteps) |
Y_n(λself.λs. ...) |
_compile_react() |
type: chain |
Compose(s1, s2, ..., sn) |
λx. sn(...s2(s1(x))) |
_compile_chain() |
type: router |
Route(classifier, routes) |
CASE |
_compile_router() |
type: parallel |
Par(a1, a2, ...) >> merge |
PAIR >> merge |
_compile_parallel() |
mcp.onlineTool |
Tool(name, mcp_caller) |
Tool_MCP (Oracle) |
_compile_tools() |
mcp.localTools: [terminate] |
Tool("terminate", λx.x) |
λx.x (identity = base case) |
_compile_tools() |
memory: {enabled: true} |
Memory(agent, store) |
Γ' = Γ ∪ store |
_compile_memory() |
guard |
Guard(agent, validator, retry) |
{x:T \| P(x)} |
_compile_guard() |
rag: {enabled: true} |
Tool("rag_retrieve", fn) 注入 |
外部 Oracle | _compile_rag() |
4.2 嵌套/递归编译
子 Agent 配置（router 的 routes、parallel 的 agents）递归调用 build_agent()：
def _compile_router(cfg):
classifier = _compile_lam(cfg["router"]["classifier"])
routes = {}
for category, sub_cfg in cfg["router"]["routes"].items():
routes[category] = build_agent(sub_cfg) # 递归编译
default = build_agent(cfg["router"]["default"]) if "default" in cfg["router"] else None
return Route(classifier, routes, default)
4.3 包装顺序（从内到外）
1. 核心: Lam / Compose / Loop / Route / Par ← 最内层
2. Guard: 包裹核心 Agent ← 输出验证
3. Memory: 最外层包裹 ← 环境扩展
编译伪代码:
agent = compile_core(cfg) # Step 1
if cfg.guard:
agent = Guard(agent, ...) # Step 2
if cfg.memory.enabled:
agent = Memory(agent, store) # Step 3
return agent
5. ReAct 编译器详细规格
ReAct 是最复杂的编译目标，单独详述。
5.1 ReAct 的 Lambda 语义
react_agent = Memory(
Y_n(λself. λstate.
let thought = think(state) in ← β-规约
let tool = route(thought, tools) in ← CASE
let obs = tool(thought) in ← β-规约
IF (tool = terminate)
THEN thought ← base case
ELSE self(state ⊕ format(thought, obs)) ← 递归
),
Γ ∪ store
)
5.2 react_step 函数的内部结构
def _compile_react(cfg) -> Term:
think = _compile_lam(cfg)
tools = _compile_tools(cfg)
max_steps = cfg["react"]["maxSteps"]
tool_timeout = cfg["react"].get("toolTimeout", 30)
observation_enabled = cfg["react"].get("observationEnabled", True)
def react_step(state: str) -> str:
# Phase 1: Think (β-规约)
thought = think(state)
# Phase 2: Tool Selection (Route / CASE)
selected_tool = _extract_tool_call(thought, tools)
# Phase 3: Base Case Check
if selected_tool is None or selected_tool._name == "terminate":
return thought # λx.x — 递归终止
# Phase 4: Tool Execution (β-规约)
try:
observation = _timeout_call(selected_tool, thought, tool_timeout)
except TimeoutError:
observation = f"[TIMEOUT after {tool_timeout}s]"
# Phase 5: State Update (递归准备)
if observation_enabled:
return _format_state(state, thought, selected_tool._name, observation)
else:
return thought
body = Tool(f"{cfg['name']}.react_step", react_step)
return Loop(
body=body,
condition=lambda r, s: s >= max_steps - 1,
max_steps=max_steps,
)
5.3 工具调用提取 (_extract_tool_call)
def _extract_tool_call(thought: str, tools: Dict[str, Tool]) -> Tool | None:
"""
从 LLM 输出中提取工具调用意图。
支持三种格式:
1. 结构化 JSON: {"tool": "search", "args": "query"}
2. XML 标签: <tool>search</tool><args>query</args>
3. 关键词匹配: thought 中包含工具名
优先级: JSON > XML > 关键词
返回: 匹配的 Tool 实例，或 None（表示不调用工具 → 隐式 terminate）
"""
5.4 状态格式化 (_format_state)
def _format_state(state, thought, tool_name, observation) -> str:
"""
构建下一轮 ReAct 的输入状态。
格式:
[Previous Context]
{state}
[Step N]
Thought: {thought}
Action: {tool_name}
Observation: {observation}
这个拼接操作对应 Lambda 演算中的 state ⊕ obs。
"""
5.5 Provider 创建与 ConversationLam 编译管道
_compile_lam() 不再直接构造 Lam。它首先调用 _create_provider() 创建一个 LLMProvider 实例，然后将其包装在 ConversationLam 中。ConversationLam 负责对话历史管理，LLMProvider 负责底层 LLM 调用。
def _compile_lam(cfg, name_suffix="", overrides=None) -> Term:
"""
编译管道 (v2):
1. _create_provider(model_cfg) -> (LLMProvider, use_conversation)
2. if use_conversation:
ConversationLam(name, provider, system_prompt, max_history_tokens)
else:
Lam(name, prompt, model) # legacy fallback
Lambda 语义:
旧: λx. LLM_{θ,p}(x) — 无状态
新: λx. provider(history ++ [x]) — 有状态 (对话感知)
"""
def _create_provider(model_cfg: Dict) -> (LLMProvider, bool):
"""
Provider 工厂: 根据 model.provider 字段创建对应的 LLMProvider。
返回: (provider_instance, use_conversation_flag)
路由逻辑:
"claude-code" → ClaudeCodeProvider (session persistence via --resume)
"anthropic" → AnthropicProvider (Messages API)
"openai" → OpenAICompatProvider (Chat Completions API)
"ollama" → OpenAICompatProvider (localhost:11434)
"dashscope" → OpenAICompatProvider (DashScope endpoint)
"deepseek" → OpenAICompatProvider (DeepSeek endpoint)
"moonshot" → OpenAICompatProvider (Moonshot endpoint)
"zhipu" → OpenAICompatProvider (Zhipu endpoint)
"""
Provider 与 ConversationLam 的关系
model YAML
│
▼
_create_provider()
│
├── LLMProvider (transport layer)
│ chat(messages: list[dict]) -> str
│
└── use_conversation flag
│
▼
ConversationLam (conversation layer)
│
├── manages messages: List[dict]
├── system message always first
├── appends user/assistant messages on each apply()
├── context window management (sliding window)
└── calls provider.chat(managed_messages)
model.conversation 字段
model:
provider: anthropic
name: claude-sonnet-4-20250514
conversation: true # 默认 true: 使用 ConversationLam
maxHistoryTokens: 80000 # 对话历史最大 token 数
当 conversation: true 时:
_compile_lam()返回ConversationLam- 每次
apply()将输入追加为 user message，调用 provider，记录 assistant response - 上下文管理: 超过
maxHistoryTokens时，旧消息被压缩为摘要
当 conversation: false 时:
_compile_lam()返回传统无状态Lam- 每次
apply()独立调用 LLM，不保留历史
5.6 react_step 的会话模式与无状态模式
ReAct 循环中的 think 步骤根据底层 provider 类型自动选择两种模式:
会话模式 (Session Mode) — ClaudeCodeProvider
适用于具有原生会话持久化的 provider (如 claude-code)。
Step 0: think.apply(full_input + tool_docs + step_info)
└─ ClaudeCodeProvider: 创建新 session, 捕获 session_id
└─ ConversationLam: 记录到 messages
Step 1: think.apply(observation_only)
└─ ClaudeCodeProvider: --resume <session_id> (只发新内容)
└─ 底层 Claude 保留完整上下文记忆
Step N: think.apply(observation_only)
└─ 同上, session 贯穿整个 ReAct 循环
关键优化: 后续步骤只发送最新的工具观察结果 (observation)，不重复发送完整状态。
Provider 端 (Claude Code CLI) 通过 --resume 自动保持完整上下文。
无状态模式 (Stateless Mode) — Ollama / OpenAI / Anthropic API 等
适用于 HTTP API 类型的 provider，无原生会话概念。
Step 0: think.apply(full_input)
└─ ConversationLam: messages = [system, user(full_input)]
└─ Provider: chat([system, user(full_input)]) -> response
Step 1: think.apply(observation)
└─ ConversationLam: messages = [system, user(full_input), assistant(r0), user(obs1)]
└─ Provider: chat(full_messages_array) -> response
Step N: think.apply(observation)
└─ ConversationLam: messages 持续增长, 受 maxHistoryTokens 约束
└─ Provider: chat(managed_messages) -> response
ConversationLam 的上下文管理器在每次调用前检查 token 预算:
- 始终保留 system message
- 始终保留最近 N 轮 (keep_recent_turns = 20)
- 超出预算时，将旧消息压缩为 "[对话历史摘要]"
工具输入序列化
工具参数统一序列化为 JSON 字符串传递给工具函数 (P0 fix):
tool_input = action.input if isinstance(action.input, str) else str(action.input)
工具参数文档自动生成
_generate_tool_schema_docs() 从 MCP 工具的 JSON Schema 自动生成参数说明，
注入到 systemPrompt 尾部，帮助 LLM 正确构造工具调用参数。
6. MCP 工具编译规格
6.1 MCP 调用器的构造
def _compile_mcp_caller(server_name: str, tool_name: str, app_cfg: dict) -> Callable:
"""
构造一个 MCP 工具调用函数。
输入: server 配置 (url, endpoint, headers)
输出: Callable[[str], str]
HTTP 调用流程:
POST {url}{endpoint}
Headers: Authorization: {token}, Content-Type: application/json
Body: {"tool": tool_name, "input": input_text}
Response: {"output": "..."}
错误处理:
- 网络超时 → 返回 "[MCP_TIMEOUT: {server}/{tool}]"
- HTTP 4xx/5xx → 返回 "[MCP_ERROR: {status}]"
- JSON 解析失败 → 返回原始 response text
重试策略（由 mcp.policy.retryOnFail 控制）:
- 指数退避: delay = min(2^attempt * 1s, 30s)
- 最大重试次数: retryOnFail（默认 0，即不重试）
"""
6.2 terminate 的特殊处理
# terminate 永远编译为恒等函数，不管 YAML 怎么写
if tool_name == "terminate":
return Tool("terminate", fn=lambda x: x)
# Lambda 语义: λx.x
# 这是 Y 组合子的 base case
# react_step 检测到 terminate 被选中时，停止递归
7. Memory 编译规格
7.1 Memory Store 接口
class MemoryStore(ABC):
"""Memory 的存储后端接口"""
@abstractmethod
def get(self, key: str) -> Any | None: ...
@abstractmethod
def put(self, key: str, value: Any, ttl: int = 0) -> None: ...
@abstractmethod
def list_recent(self, n: int) -> List[Tuple[str, Any]]: ...
@abstractmethod
def clear(self) -> None: ...
7.2 四种 strategy 的实现
| strategy | 实现 | 持久性 | 并发安全 |
|---|---|---|---|
local |
Python dict + LRU | 进程内 | 否 |
redis |
Redis hash | 跨进程 | 是 |
sqlite |
SQLite 表 | 磁盘 | 是（单写） |
custom |
用户提供类路径 | 取决于实现 | 取决于实现 |
7.3 Memory 注入方式
def _compile_memory(agent: Term, memory_cfg: dict) -> Term:
"""
Memory 编译 = 环境扩展 Γ' = Γ ∪ store
注入方式: 将记忆内容序列化后追加到每次 Agent 调用的输入前面
格式:
[Memory Context]
- key1: value1 (2min ago)
- key2: value2 (15min ago)
... (最近 {size} 条)
[Current Input]
{original_input}
记忆自动管理:
- 每次 Agent 输出后，自动摘要存入 memory
- 超过 size 的旧记忆 LRU 淘汰
- 超过 ttl 的记忆自动删除
"""
8. 静态分析 (lint) 规格
8.1 lint 规则表（完整版）
| ID | 级别 | 条件 | Lambda 含义 | 消息 |
|---|---|---|---|---|
| L001 | ERROR | systemPrompt 为空 |
λx.⊥ — 函数体未定义 |
"Agent 无行为定义" |
| L002 | ERROR | model 未配置 |
无 LLM 计算单元 | "无法执行 β-规约" |
| L003 | ERROR | react.maxSteps = 0 |
Y₀(g) = ⊥ |
"Agent 不会执行" |
| L004 | ERROR | type: react 且无 terminate |
Y 无 base case | "可能无限循环" |
| L005 | ERROR | type: router 且 routes 为空 |
CASE 无分支 | "路由无目标" |
| L006 | ERROR | type: chain 且 steps 为空 |
空组合链 | "Pipeline 无步骤" |
| L007 | WARN | temperature > 1.5 |
高熵 ⊕_p |
"输出可能不稳定" |
| L008 | WARN | memory.ttl = 0 且 enabled |
Γ' 绑定永不回收 | "内存可能无限增长" |
| L009 | WARN | memory.size = 0 且 enabled |
Γ' = Γ ∪ ∅ = Γ |
"Memory 无效果" |
| L010 | WARN | react.maxSteps > 50 |
Y₅₀+ |
"长运行高成本" |
| L011 | WARN | ReAct 除 terminate 外无工具 | 纯推理循环 | "Agent 只能思考" |
| L012 | WARN | guard.retry > 5 |
过多重试 | "可能死循环" |
| L013 | WARN | router 无 default |
CASE 不完全 | "未分类输入将报错" |
| L014 | INFO | temperature = 0 |
确定性 Lambda | "确定性模式" |
| L015 | INFO | Memory 已启用 | Γ' = Γ ∪ store |
"有状态 Agent" |
| L016 | INFO | RAG 已启用 | 外部知识 Oracle | "可访问知识库" |
8.2 lint 输出格式
lambdagent lint: agent-config.yml
============================================================
✗ [ERROR] mcp.localTools: 没有 terminate 工具
Lambda: Y 组合子无 base case (λx.x) → 无穷递归
⚠ [WARN] react.maxSteps: maxSteps=50 较大
Lambda: Y₅₀(g) → 最多 50 步 β-规约
ℹ [INFO] memory: 启用记忆 strategy=redis, size=20, ttl=7200s
Lambda: 环境扩展 Γ' = Γ ∪ store(redis)
ℹ [INFO] summary: Agent 'SeeCoderManus' 的 Lambda 结构:
Memory(Loop(think >> act >> observe, max_steps=20), redis)
────────────────────────────────────────────────────────
1 error(s), 1 warning(s), 2 info(s)
9. 编译器架构
9.1 模块分解
from_config.py
├── from_config(path, **overrides) → Term # 入口
├── build_agent(cfg) → Term # 递归编译核心
│ ├── _compile_lam(cfg) → ConversationLam|Lam # λ 抽象 (via _create_provider)
│ ├── _compile_react(cfg) → Loop # Y 组合子
│ ├── _compile_chain(cfg) → Compose # 函数组合
│ ├── _compile_router(cfg) → Route # CASE
│ ├── _compile_parallel(cfg) → Par # PAIR
│ ├── _compile_tools(cfg) → Dict[str, Tool] # Oracle 集
│ ├── _compile_memory(agent, cfg) → Memory # 环境扩展
│ └── _compile_guard(agent, cfg) → Guard # 类型约束
├── _create_provider(model_cfg) → (LLMProvider, bool) # Provider 工厂
├── _resolve_model(model_cfg) → str # 模型名解析 (legacy fallback)
├── _compile_mcp_caller(server, tool, app) → Callable # MCP HTTP
├── _extract_tool_call(thought, tools) → Tool | None # 工具匹配
├── _format_state(state, thought, tool, obs) → str # 状态拼接
├── describe_config(path) → str # Lambda 结构可视化
└── to_lambda_expr(path) → str # 导出纯 Lambda 表达式字符串
9.2 编译流程图
YAML 文件
│
┌────▼────┐
│ 解析 YAML │
└────┬────┘
│
┌────▼────┐
│ 校验 Schema│ ── 缺必填字段 ──→ SchemaError
└────┬────┘
│
┌────▼────┐
│ 识别 type │
└────┬────┘
│
┌────────┬───┼────────┬──────────┐
▼ ▼ ▼ ▼ ▼
simple react chain router parallel
│ │ │ │ │
_lam() _react() _chain() _router() _parallel()
│ │ │ │ │
└────────┴───┼────────┴──────────┘
│
┌────▼────┐
│ guard? │── yes ──→ Guard(agent, P)
└────┬────┘
│
┌────▼────┐
│ memory? │── yes ──→ Memory(agent, store)
└────┬────┘
│
┌────▼────┐
│ 输出 Term │
└─────────┘
10. 错误处理规格
10.1 异常层次
class CompileError(LambdagentError):
"""编译阶段错误（YAML → Term）"""
pass
class SchemaError(CompileError):
"""YAML schema 不合法"""
field: str # 出错的字段路径
expected: str # 期望的类型/值
actual: Any # 实际值
class SemanticError(CompileError):
"""语义不合法（schema 合法但逻辑矛盾）"""
rule: str # 违反的 lint 规则 ID
10.2 编译时 vs 运行时错误
| 阶段 | 示例 | 异常类型 | 应对 |
|---|---|---|---|
| 解析 | YAML 语法错误 | yaml.YAMLError |
报位置 |
| Schema | type 字段缺失 |
SchemaError |
报字段路径 |
| 语义 | router 无 routes | SemanticError |
报 lint 规则 |
| 运行时 | LLM API 超时 | LambdagentError |
记入 trace |
| 运行时 | Guard 验证失败 | ValidationError |
重试或 fallback |
| 运行时 | Route 无匹配 | RouteError |
走 default 或报错 |
11. 扩展点
11.1 自定义 Provider
from_config("agent.yml", providers={
"dashscope": DashScopeProvider(api_key="sk-..."),
"openai": OpenAIProvider(api_key="sk-..."),
})
11.2 自定义工具注入
from_config("agent.yml", tools={
"web_search": lambda q: requests.get(f"https://api.search.com?q={q}").json(),
"run_code": lambda code: exec_sandbox(code),
})
11.3 自定义 Memory Store
from_config("agent.yml", memory_store=MyRedisStore(host="localhost"))
11.4 编译后 Hook
agent = from_config("agent.yml")
agent = agent >> Tool("postprocess", my_cleanup_fn) # 在编译结果后追加步骤
12. 测试验收标准
12.1 编译正确性
| 测试 | 输入 | 期望输出 Term 类型 | 验证方式 |
|---|---|---|---|
| T01 | type: simple |
Lam |
isinstance(term, Lam) |
| T02 | type: react, maxSteps: 20 |
Loop(max_steps=20) 或 Memory(Loop(...)) |
检查 Loop.max_steps |
| T03 | type: chain, steps: [a, b, c] |
Compose(3 stages) |
len(term.stages) == 3 |
| T04 | type: router, routes: {x, y} |
Route(2 routes) |
len(term.routes) == 2 |
| T05 | type: parallel, agents: [a, b] |
Par(2 agents) >> merge |
检查 Par + Compose |
| T06 | memory.enabled: true |
最外层是 Memory |
isinstance(term, Memory) |
| T07 | guard.validator 存在 |
内含 Guard |
递归查找 Guard |
| T08 | terminate in localTools |
包含 Tool("terminate", λx.x) |
检查 fn 是恒等函数 |
12.2 端到端测试
| 测试 | 输入 YAML | 执行输入 | 验证 |
|---|---|---|---|
| E01 | 现有 agent-cofig.yml |
"1+1等于几" | 返回包含 "2" 的字符串 |
| E02 | simple agent | "Hello" | 返回非空字符串 |
| E03 | chain agent (3 steps) | 一段文本 | Context.trace 有 3 条 |
| E04 | router agent | 代码问题 | 路由到 code agent |
| E05 | react agent (有 terminate) | 简单问题 | 在 maxSteps 内终止 |
12.3 lint 测试
| 测试 | 输入配置 | 期望 lint 结果 |
|---|---|---|
| L_T01 | 无 systemPrompt | ERROR L001 |
| L_T02 | react 无 terminate | ERROR L004 |
| L_T03 | maxSteps=0 | ERROR L003 |
| L_T04 | router 无 routes | ERROR L005 |
| L_T05 | 正常配置 | 0 errors |
| L_T06 | temperature=2.0 | WARN L007 |
13. 与现有代码的差距分析
13.1 当前实现 (v1) vs 本规格 (v2)
| 能力 | v1 (当前) | v2 (本规格) | 工作量 |
|---|---|---|---|
type: simple |
✅ | ✅ | 无 |
| ConversationLam + Provider | ❌ (Lam only) | ✅ (_create_provider pipeline) | 已完成 |
provider: claude-code |
❌ | ✅ (session persistence) | 已完成 |
model.conversation 字段 |
❌ | ✅ (conversation: true/false) | 已完成 |
| Tool input JSON serialization | ❌ (broken) | ✅ (P0 fix) | 已完成 |
| Tool schema auto-gen docs | ❌ | ✅ (_generate_tool_schema_docs) | 已完成 |
type: react |
✅ (基础) | ✅ (完整工具提取) | 中 |
type: chain |
❌ | ✅ | 新增 |
type: router |
❌ | ✅ | 新增 |
type: parallel |
❌ | ✅ | 新增 |
| MCP HTTP 调用 | ❌ (占位) | ✅ (真实 HTTP) | 中 |
| 工具调用提取 | 关键词匹配 | JSON + XML + 关键词 | 中 |
| Memory store backend | dict only | local/redis/sqlite/custom | 大 |
| Guard 编译 | ❌ | ✅ | 小 |
| 嵌套子 Agent | ❌ | ✅ (递归 build_agent) | 中 |
| overrides 参数 | ❌ | ✅ | 小 |
| Schema 校验 | ❌ | ✅ | 中 |
| describe_config | ✅ | ✅ (增强) | 小 |
| to_lambda_expr | ❌ | ✅ (导出纯 Lambda) | 中 |
13.2 优先级排序
P0 (必须): chain + router + parallel 编译
P1 (重要): MCP 真实 HTTP + 工具调用提取增强
P2 (增强): Schema 校验 + 嵌套子 Agent + overrides
P3 (生态): Memory 多后端 + to_lambda_expr + 自定义 Provider
14. Lambda 对应速查卡
给后续开发者的速查表，编码时随时参考：
YAML Python (lambdagent) Lambda 演算
───────────────────────── ───────────────────────── ──────────────────
systemPrompt + model ConversationLam(name, prov, p) λx. prov(history ++ [x])
(conversation: false) Lam("name", prompt, model) λx. LLM_{θ,p}(x)
model.provider: claude-code ClaudeCodeProvider(config) session-persistent LLM
model.provider: anthropic AnthropicProvider(config) HTTP API LLM
model.provider: ollama/... OpenAICompatProvider(config) OpenAI-compat LLM
agent(input) term("hello") (f x) → β-规约
type: chain f >> g >> h λx. h(g(f(x)))
type: react + maxSteps Loop(body, cond, N) Y_N(λself.λx...)
type: router Route(cls, {k: agent}) CASE
type: parallel Par(a, b) >> merge PAIR >> merge
mcp.onlineTool Tool("name", http_fn) Oracle / primitive
terminate Tool("terminate", λx.x) λx.x (identity)
memory Memory(agent, store) Γ' = Γ ∪ store
guard Guard(agent, P, retry) {x:T | P(x)}
temperature Lam(..., temperature=T) ⊕_T (概率参数)
maxSteps Loop(..., max_steps=N) Y_N (有界 Y)
15. CLI 交互入口规格
15.1 设计哲学
CLI 是 lambdagent 的第一公民交互界面，不是附属品。理由：
- Agent 的本质是函数——函数最自然的交互方式是
f(x)，CLI 就是lambdagent run config.yml "input" - Lambda 演算是 REPL 友好的——每次输入一个表达式，求值，输出结果，循环
- 运维需要 CLI——部署前 lint、运行时 trace、生产排查全在终端完成
- MCP/工具生态是 CLI 驱动的——MCP 服务器本身就是进程间通信
15.2 命令总览
lambdagent <command> [options] [arguments]
Commands:
compile 编译 YAML → Lambda 项（不执行）
run 编译 + 执行
repl 交互式 REPL（持续对话）
lint 静态分析
trace 查看/回放 β-规约追踪
lambda 导出纯 Lambda 表达式
serve 启动 HTTP/MCP 服务
tools 管理工具（MCP 发现/测试）
version 版本信息
15.3 各命令详细规格
lambdagent compile — 编译（不执行）
# 编译并打印 Lambda 结构
lambdagent compile agent-config.yml
# 输出:
# SeeCoderManus =
# Memory(
# Y₂₀(λself.λstate.
# let t = think(state) in
# CASE t [(sum, Tool_MCP), (improve, Tool_MCP), (terminate, λx.x)]
# >> IF is_terminate THEN t ELSE self(state ⊕ obs)
# ),
# redis(size=20, ttl=7200)
# )
#
# Constructs used: Lam, Loop, Route, Tool(×3), Memory
# β-reduction bound: max 20 steps
# Base case: terminate = λx.x
# 选项:
# --format text|json|python 输出格式
# --validate 同时运行 lint
# --dry-run 验证所有工具可达（ping MCP 端点）
Lambda 语义：compile = 解析 + 翻译，不触发 β-规约。等价于"写出 Lambda 表达式但不求值"。
lambdagent run — 编译 + 执行
# 单次执行
lambdagent run agent-config.yml "帮我写一个快速排序"
# 从 stdin 读取输入
echo "帮我写一个快速排序" | lambdagent run agent-config.yml -
# 从文件读取输入
lambdagent run agent-config.yml --input task.txt
# 选项:
# --trace 打印每步 β-规约
# --trace-file trace.json β-规约追踪保存到文件
# --max-steps N 覆盖 YAML 的 maxSteps
# --temperature T 覆盖 YAML 的 temperature
# --timeout S 总超时(秒)
# --model MODEL 覆盖模型
# --tool NAME=CMD 注入 CLI 工具（见 15.4）
# --env KEY=VALUE 注入环境变量到 Memory
# --output FILE 结果输出到文件
# --format text|json 输出格式
# --quiet 只输出最终结果，不打印过程
# --verbose 打印所有中间步骤
Lambda 语义：run = compile + β-规约。run config.yml "input" 等价于 (from_config("config.yml")) ("input")。
--trace 输出格式：
β[0] think (2.8s) "帮我写快速排序" → "需要调用代码工具..."
β[1] route (0.0s) → selected: sum (Tool_MCP)
β[2] Tool:sum (1.2s) → "def quicksort(arr): ..."
β[3] think (3.1s) "代码结果..." → "代码已完成，调用 terminate"
β[4] route (0.0s) → selected: terminate (λx.x)
β[5] terminate (0.0s) → (identity: base case reached)
────────────────────────────────────────
Total: 5 β-reductions, 7.1s, ~2400 tokens
lambdagent repl — 交互式 REPL
# 启动 REPL
lambdagent repl agent-config.yml
# 输出:
# lambdagent REPL v2.0
# Agent: SeeCoderManus (react, maxSteps=20)
# Lambda: Memory(Y₂₀(think >> route >> observe), redis)
# Type :help for commands, :quit to exit
#
# λ> 帮我写一个快速排序
# [β[0] think 2.8s] 需要调用代码工具...
# [β[1] Tool:sum 1.2s] def quicksort(arr): ...
# [β[2] think 3.1s] 代码完成，调用 terminate
# [β[3] terminate 0.0s] (base case)
#
# Result: def quicksort(arr): ...
# (4 β-reductions, 7.1s)
#
# λ> 能不能加上注释？
# [Memory: injecting 1 previous exchange]
# [β[0] think 2.5s] 用户想要注释版...
# ...
# REPL 内置命令:
# :help 帮助
# :quit / :q 退出
# :trace 显示上次执行的完整 β-规约链
# :trace N 显示第 N 步的详细信息
# :memory 显示当前 Memory 内容
# :memory clear 清空 Memory
# :lambda 显示当前 Agent 的 Lambda 表达式
# :lint 重新 lint 配置
# :reload 重新加载 YAML 配置（热更新）
# :model MODEL 切换模型
# :temp T 切换温度
# :tools 列出可用工具
# :tool NAME "input" 手动调用单个工具
# :export FILE 导出完整对话 + trace 到文件
# :stats 显示会话统计（总 β-规约数、总 token、总时间）
Lambda 语义：REPL = 持续的 β-规约环境。每次输入是一个新的函数应用 (agent input_n)。Memory 使得 Γ 在多次调用间累积——这正是 REPL 与单次 run 的区别。
REPL 的 Lambda 本质：
REPL session =
let Γ₀ = {} in
let (r₁, Γ₁) = agent(input₁) [Γ₀] in
let (r₂, Γ₂) = agent(input₂) [Γ₁] in
let (r₃, Γ₃) = agent(input₃) [Γ₂] in
...
每一轮:
1. 读取用户输入 xₙ
2. β-规约: (agent xₙ) [Γₙ₋₁] → (rₙ, Γₙ)
3. 打印 rₙ
4. 更新 Γₙ₋₁ → Γₙ (Memory 写入)
5. 循环
这就是 Y 组合子在 REPL 级别的又一次体现:
REPL = Y(λself.λΓ. let x = read() in let (r, Γ') = agent(x)[Γ] in print(r); self(Γ'))
lambdagent lint — 静态分析
# Lint 单个文件
lambdagent lint agent-config.yml
# Lint 目录下所有配置
lambdagent lint configs/
# 选项:
# --level error|warn|info 最低报告级别
# --format text|json 输出格式
# --fix 自动修复可修复的问题
# --rules L001,L004 只运行指定规则
# --ignore L014 忽略指定规则
# 退出码:
# 0 = 无 error
# 1 = 有 error
# 2 = 配置文件不存在
lambdagent trace — 追踪查看/回放
# 回放上次执行的 trace
lambdagent trace trace.json
# 选项:
# --step N 只看第 N 步
# --slow 逐步回放（按回车继续）
# --filter TERM 只看特定 Term 的 β-规约
# --timeline 时间线视图
# --flamegraph 火焰图输出（SVG）
--timeline 输出：
Time ──────────────────────────────────────────→
0s 2.8s 4.0s 7.1s 7.1s
│ │ │ │ │
├─think───┤ │ │ │
├─search──┤ │ │
├─think───┤ │
├─terminate
lambdagent lambda — 导出 Lambda 表达式
# 导出为纯 Lambda 表达式
lambdagent lambda agent-config.yml
# 输出:
# Memory(
# Y₂₀(λself. λstate.
# let t = (λx. LLM_{qwen3-max, "你是SeeCoderManus..."}(x)) state in
# CASE (classify t) [
# ("sum", λx. MCP("example-mcp-server", "everything_get_sum", x)),
# ("improve", λx. MCP("example-mcp-server", "chat_improve_prompt", x)),
# ("terminate", λx. x)
# ] >> λobs.
# IF (obs = t) THEN t ELSE self(state ⊕ format(t, obs))
# ),
# Γ ∪ redis{size=20, ttl=7200}
# )
# 选项:
# --format human|formal|json 输出格式
# --desugar 展开所有语法糖到纯 λ/app/var
lambdagent serve — 启动服务
# 启动 HTTP 服务
lambdagent serve agent-config.yml --port 8080
# 启动 MCP 服务器（让其他 Agent 作为工具调用本 Agent）
lambdagent serve agent-config.yml --mcp
# 启动 WebSocket 服务（流式输出）
lambdagent serve agent-config.yml --ws --port 8080
Lambda 语义：serve = 将 Agent 包装为一个持久运行的函数，等待外部应用（application）。
serve(agent, port) = Y(λself.λΓ.
let (x, conn) = accept(port) in ← 等待请求
let (r, Γ') = agent(x) [Γ] in ← β-规约
send(conn, r); ← 返回结果
self(Γ') ← 循环（保持 Memory）
)
lambdagent tools — 工具管理
# 列出配置中的所有工具
lambdagent tools agent-config.yml
# 输出:
# Tools for SeeCoderManus:
# [MCP] everything_get_sum example-mcp-server https://ai-paas...
# [MCP] chat_improve_prompt example-mcp-server https://ai-paas...
# [Local] terminate (λx.x) base case
#
# MCP endpoints:
# example-mcp-server: https://your-mcp-endpoint.example.com/mcp/airouting
# Status: ✓ reachable (238ms)
# 测试单个工具
lambdagent tools agent-config.yml --test everything_get_sum "测试输入"
# 发现可用的 MCP 工具
lambdagent tools agent-config.yml --discover example-mcp-server
15.4 CLI 工具注入 (--tool)
CLI 工具注入是 CLI 作为交互入口的核心差异化能力——它把任意 shell 命令变成 Lambda 项中的 Tool。
# 注入 shell 命令作为工具
lambdagent run agent.yml "分析日志" \
--tool grep="grep -c ERROR /var/log/app.log" \
--tool wc="wc -l /var/log/app.log" \
--tool curl="curl -s https://api.example.com/status"
# 注入管道命令
lambdagent run agent.yml "统计代码行数" \
--tool loc="find . -name '*.py' | xargs wc -l"
# 注入交互式命令（stdin 传参）
lambdagent run agent.yml "搜索文件" \
--tool search="grep -r {} src/"
Lambda 语义：
--tool grep="grep -c ERROR /var/log/app.log"
编译为:
Tool("grep", fn=lambda x: subprocess.run(
"grep -c ERROR /var/log/app.log",
shell=True, capture_output=True
).stdout.decode())
注入到 Agent 的工具集:
tools["grep"] = Tool("grep", shell_fn)
当命令中包含 {} 占位符时，Agent 的输出会被替换进去：
--tool search="grep -r {} src/"
编译为:
Tool("search", fn=lambda x: subprocess.run(
f"grep -r {shlex.quote(x)} src/",
shell=True, capture_output=True
).stdout.decode())
安全约束：
--tool注入的命令只在用户明确指定时执行- 命令参数经过
shlex.quote()转义 - 默认超时 30 秒（可通过
--tool-timeout覆盖） - 不允许注入
rm -rf、dd、mkfs等危险命令（黑名单）
15.5 管道组合 — Unix 哲学
lambdagent CLI 遵循 Unix 管道哲学：每个命令读 stdin、写 stdout、错误写 stderr。
# 管道链: 编译检查 → 执行 → 后处理
lambdagent lint config.yml && \
lambdagent run config.yml "input" | jq '.result'
# Agent 链: 一个 Agent 的输出喂给另一个
lambdagent run summarizer.yml "长文本..." | \
lambdagent run translator.yml -
# 批量处理
cat inputs.txt | while read line; do
lambdagent run agent.yml "$line" --quiet
done > outputs.txt
# 与系统工具组合
curl -s https://api.example.com/data | \
lambdagent run analyzer.yml - | \
mail -s "分析报告" team@example.com
Lambda 语义：Unix 管道 | = 函数组合 >>。
cmd1 | cmd2 | cmd3
= cmd3(cmd2(cmd1(stdin)))
= (cmd1 >> cmd2 >> cmd3)(stdin)
= Compose(cmd1, cmd2, cmd3).apply(stdin)
这不是类比——这就是同一个数学结构。Unix shell 的管道和 lambdagent 的 >> 都是函数组合。CLI 是 Lambda 演算的自然栖息地。
15.6 环境变量接口
# API Key
export ANTHROPIC_API_KEY=sk-...
export OPENAI_API_KEY=sk-...
export DASHSCOPE_API_KEY=sk-...
# 模型覆盖
export LAMBDAGENT_MODEL=claude-sonnet-4-20250514
export LAMBDAGENT_TEMPERATURE=0.0
export LAMBDAGENT_MAX_TOKENS=4096
# 追踪
export LAMBDAGENT_TRACE=1 # 默认开启 trace
export LAMBDAGENT_TRACE_FILE=trace.json # trace 输出路径
# MCP
export LAMBDAGENT_MCP_TIMEOUT=30 # MCP 调用超时
export LAMBDAGENT_MCP_RETRY=2 # MCP 重试次数
# Memory
export LAMBDAGENT_REDIS_URL=redis://localhost:6379
15.7 退出码规范
| 码 | 含义 | 场景 |
|---|---|---|
| 0 | 成功 | Agent 正常完成，lint 无 error |
| 1 | Agent 执行失败 | LLM API 错误、工具超时、Guard 失败 |
| 2 | 编译失败 | YAML 语法错误、Schema 不合法 |
| 3 | Lint 有 ERROR | lint 发现结构性问题 |
| 4 | 配置文件不存在 | 文件路径错误 |
| 130 | 用户中断 | Ctrl+C |
15.8 输出格式
所有命令支持 --format 选项：
--format text（默认，人类可读）：
Result: def quicksort(arr): ...
(4 β-reductions, 7.1s, ~2400 tokens)
--format json（机器可读，管道友好）：
{
"result": "def quicksort(arr): ...",
"trace": [
{"step": 0, "term": "think", "duration_ms": 2800, "input": "...", "output": "..."},
{"step": 1, "term": "Tool:sum", "duration_ms": 1200, "input": "...", "output": "..."}
],
"stats": {
"total_steps": 4,
"total_time_ms": 7100,
"total_tokens": 2400
}
}
16. CLI 实现架构
16.1 模块分解
lambdagent/
├── __main__.py # python -m lambdagent 入口
├── cli/
│ ├── __init__.py
│ ├── main.py # argparse 路由
│ ├── compile_cmd.py # lambdagent compile
│ ├── run_cmd.py # lambdagent run
│ ├── repl_cmd.py # lambdagent repl
│ ├── lint_cmd.py # lambdagent lint
│ ├── trace_cmd.py # lambdagent trace
│ ├── lambda_cmd.py # lambdagent lambda
│ ├── serve_cmd.py # lambdagent serve
│ ├── tools_cmd.py # lambdagent tools
│ └── shell_tool.py # --tool 的 shell 命令包装
16.2 __main__.py 入口
"""
python -m lambdagent <command> [args]
等价于: lambdagent <command> [args]（通过 setup.py entry_points 注册）
"""
from lambdagent.cli.main import main
if __name__ == "__main__":
main()
16.3 shell_tool.py — CLI 工具注入的核心
"""
将 shell 命令包装为 lambdagent Tool。
--tool grep="grep -c ERROR /var/log/app.log"
→ Tool("grep", fn=ShellTool("grep -c ERROR /var/log/app.log"))
--tool search="grep -r {} src/"
→ Tool("search", fn=ShellTool("grep -r {} src/", parameterized=True))
"""
DANGEROUS_COMMANDS = {"rm -rf", "dd ", "mkfs", "format ", "> /dev/sd"}
class ShellTool:
def __init__(self, command: str, timeout: int = 30):
self.command = command
self.timeout = timeout
self.parameterized = "{}" in command
self._safety_check()
def _safety_check(self):
for dangerous in DANGEROUS_COMMANDS:
if dangerous in self.command:
raise ValueError(f"Dangerous command blocked: {self.command}")
def __call__(self, input_text: str) -> str:
if self.parameterized:
cmd = self.command.replace("{}", shlex.quote(str(input_text)))
else:
cmd = self.command
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True,
timeout=self.timeout
)
if result.returncode != 0 and result.stderr:
return f"[ERROR] {result.stderr.strip()}"
return result.stdout.strip()
16.4 与现有代码的集成点
CLI 命令 调用的核心 API
───────── ──────────────
compile from_config(path) → term; describe_config(path)
run from_config(path) → term; term(input, ctx)
repl from_config(path) → term; while True: term(input, ctx)
lint lint_config(path) → issues
trace ctx.print_trace() / ctx.trace → json
lambda to_lambda_expr(path) → str [新增]
serve from_config(path) → term; http.serve(term)
tools _build_tools(cfg) → dict; ping endpoints
17. Agent 间通信: CLI 作为交互协议
17.1 设计哲学
Agent 之间的通信不应只依赖 MCP。CLI（stdin/stdout）是最通用的进程间通信协议：
- 任何语言写的程序都能读写 stdin/stdout
- Unix 管道天然支持 Agent 链式调用
- 不需要 HTTP 服务器、不需要 SDK、不需要注册发现
- Lambda 语义完美对齐：管道
|= 函数组合>>
17.2 三种 Agent 间通信方式
方式 协议 Lambda 语义 适用场景
───────── ───────── ───────────── ─────────
CLI Pipe stdin/stdout f >> g (Compose) 本机 Agent 链
MCP HTTP/SSE Tool(name, mcp_fn) 远程服务调用
CLI Tool 注入 subprocess Tool(name, shell_fn) 混合生态集成
17.3 YAML 中的 CLI 通信配置
# ═══ 新增: CLI 通信配置 ═══
cli:
# 本 Agent 作为 CLI 服务（被其他 Agent 调用）
server:
enabled: true
mode: pipe # "pipe" (stdin/stdout) | "socket" (Unix socket) | "tcp"
format: json # "text" | "json" | "jsonl" (一行一条)
socketPath: /tmp/lambdagent-myagent.sock # mode=socket 时
port: 0 # mode=tcp 时，0=随机端口
# 调用其他 CLI Agent
agents:
summarizer:
command: "python -m lambdagent run summarizer.yml -"
format: json
timeout: 60
retry: 2
translator:
command: "lambdagent run translator.yml --quiet -"
format: text
timeout: 30
legacy_tool:
command: "/usr/local/bin/legacy-analyzer" # 非 lambdagent 程序也行
format: text
timeout: 10
# 远程 Agent (通过 SSH)
remote_researcher:
command: "ssh gpu-server 'lambdagent run researcher.yml --quiet -'"
format: json
timeout: 120
# 管道链预设（快捷定义常用的 Agent 组合）
pipelines:
research_and_translate:
steps: [summarizer, translator] # 等价于 summarizer >> translator
analyze_with_review:
steps: [legacy_tool, summarizer]
parallel: false # 串行执行
17.4 CLI Agent 编译规则
# cli.agents 中的每个 Agent 编译为 Tool
# 但不是普通 Tool——是 "CLI Tool"，通过 subprocess 调用
# YAML:
# cli:
# agents:
# summarizer:
# command: "lambdagent run summarizer.yml --quiet -"
# 编译为:
Tool("summarizer", fn=CLIAgent(
command="lambdagent run summarizer.yml --quiet -",
format="json",
timeout=60,
))
# Lambda 语义:
# summarizer = λx. exec("lambdagent run summarizer.yml --quiet -", x)
# 等价于: summarizer = λx. (from_config("summarizer.yml"))(x)
# 即: CLI 调用 = β-规约的跨进程版本
17.5 CLIAgent 实现
class CLIAgent:
"""
通过 CLI 调用另一个 Agent = 跨进程 β-规约
Lambda 语义:
CLIAgent(cmd) = λx. decode(exec(cmd, encode(x)))
通信协议:
1. 将 input 写入子进程的 stdin
2. 从子进程的 stdout 读取 output
3. stderr 用于日志/错误（不影响结果）
这就是 Unix 管道的语义——也是 Lambda 演算应用的语义:
(f x) 在 Lambda 中 = 把 x 喂给 f，拿到结果
cmd < input 在 Unix 中 = 把 input 喂给 cmd，拿到结果
"""
def __init__(self, command: str, format: str = "text", timeout: int = 60):
self.command = command
self.format = format
self.timeout = timeout
def __call__(self, input_text: str) -> str:
import subprocess
proc = subprocess.run(
self.command,
shell=True,
input=input_text,
capture_output=True,
text=True,
timeout=self.timeout,
)
if proc.returncode != 0:
return f"[CLI_ERROR:{proc.returncode}] {proc.stderr.strip()}"
output = proc.stdout.strip()
if self.format == "json":
import json
data = json.loads(output)
return data.get("result", output)
return output
17.6 使用场景
场景 A: Agent 链式调用（管道）
# 方法 1: Unix 管道（最简单）
echo "长文本..." | \
lambdagent run summarizer.yml --quiet - | \
lambdagent run translator.yml --quiet -
# 方法 2: YAML 配置（可复用）
# config:
# cli.agents: {summarizer: ..., translator: ...}
# type: chain
# chain.steps 引用 cli.agents
lambdagent run pipeline.yml "长文本..."
# Lambda 语义完全相同:
# 方法 1: (translator ∘ summarizer)("长文本...")
# 方法 2: (translator >> summarizer)("长文本...")
场景 B: Agent 作为工具（ReAct 中调用其他 Agent）
# master-agent.yml
type: react
react: {maxSteps: 10}
mcp:
localTools: [terminate]
cli:
agents:
code_expert:
command: "lambdagent run code-agent.yml --quiet -"
math_expert:
command: "lambdagent run math-agent.yml --quiet -"
ReAct 执行:
β[0] think → "这个问题需要代码专家"
β[1] route → selected: code_expert (CLI Agent)
β[2] cli:code_expert → subprocess: lambdagent run code-agent.yml
↳ 子进程内部: β[0] think → β[1] ... → result
β[3] think → "代码结果看起来不错，terminate"
β[4] terminate → (base case)
场景 C: 与非 lambdagent 系统集成
cli:
agents:
# Python 脚本
data_fetcher:
command: "python scripts/fetch_data.py"
# Go 程序
fast_parser:
command: "./bin/parser --format json"
# Node.js 服务
renderer:
command: "node render.js"
# 任意 REST API (通过 curl)
api_call:
command: "curl -s -X POST https://api.example.com/process -d {}"
关键洞察：CLI 通信让 lambdagent 可以编排任何能读写 stdin/stdout 的程序——不限语言、不限框架、不限部署位置（SSH 透传）。这是 MCP 做不到的通用性。
17.7 CLI vs MCP 对比
维度 CLI Pipe MCP
────── ────────── ──────
协议 stdin/stdout HTTP/SSE
发现 手动配置 command 服务注册
延迟 低 (fork) 中 (HTTP)
流式 天然支持 (pipe) SSE
跨语言 任何程序 需 MCP SDK
跨机器 SSH 透传 HTTP
有状态 每次 fork 新进程 持久连接
调试 stderr + 退出码 HTTP 状态码
Lambda 语义 f >> g (Compose) Tool(name, fn)
17.8 编译器更新
from_config() 需要新增对 cli 字段的编译:
def _compile_cli_agents(cfg: dict) -> dict[str, Tool]:
"""编译 cli.agents 为 Tool 集合"""
cli_cfg = cfg.get("cli", {})
agents_cfg = cli_cfg.get("agents", {})
tools = {}
for name, agent_cfg in agents_cfg.items():
tools[name] = Tool(
name=f"cli:{name}",
fn=CLIAgent(
command=agent_cfg["command"],
format=agent_cfg.get("format", "text"),
timeout=agent_cfg.get("timeout", 60),
),
)
return tools
CLI Agent 和 MCP Tool 合并到同一个工具集中供 ReAct/Route 使用:
all_tools = {
**_compile_tools(cfg), # MCP 工具
**_compile_cli_agents(cfg), # CLI Agent 工具
}
18. CLI 测试验收
| 测试 | 命令 | 期望 |
|---|---|---|
| C00 | lambdagent version |
打印版本信息 |
| C01 | lambdagent compile agent-cofig.yml |
打印 Lambda 结构，退出码 0 |
| C02 | lambdagent lint agent-cofig.yml |
输出 lint 结果，退出码 0（无 error） |
| C03 | lambdagent lint bad-config.yml |
退出码 3（有 error） |
| C04 | lambdagent run simple.yml "Hello" |
输出非空结果，退出码 0 |
| C05 | lambdagent run simple.yml "Hello" --trace |
输出含 β[0] 的 trace |
| C06 | lambdagent run simple.yml "Hello" --format json |
输出合法 JSON |
| C07 | echo "Hello" \| lambdagent run simple.yml - |
stdin 输入有效 |
| C08 | lambdagent run a.yml "x" --tool wc="wc -l f.txt" |
工具注入有效 |
| C09 | lambdagent lambda agent-cofig.yml |
输出 Lambda 表达式 |
| C10 | lambdagent tools agent-cofig.yml |
列出工具 + 状态 |
| C11 | lambdagent repl simple.yml 后输入 :quit |
正常退出 |
| C12 | lambdagent run a.yml "x" --tool rm="rm -rf /" |
被安全拦截 |
| C13 | echo "Hello" \| lambdagent run a.yml --quiet - \| lambdagent run b.yml - |
Agent 链管道 |
| C14 | CLI Agent 配置 cli.agents.sub |
编译为 Tool("cli:sub", CLIAgent(...)) |
| C15 | lambdagent tools config.yml 含 CLI Agent |
列出 MCP + CLI 工具 |
19. 新增 YAML 配置字段（v2 扩展）
以下字段对应 lambdagent v2 新增的模块。它们是可选的顶级配置块。
19.1 GroupChat 配置
# ═══ GroupChat 配置（type=groupchat 时使用）═══
groupchat:
agents: # 参与对话的 Agent 列表
- name: string
systemPrompt: string
model: object # 可选模型覆盖
maxRounds: int # 最大轮数（Y 组合子展开上界），默认 10
scheduler: string # "round_robin" | "random" | "llm"
schedulerPrompt: string # scheduler=llm 时的分类 prompt
termination: # 终止条件
keywords: [string] # 包含这些关键词时终止，默认 ["CONSENSUS", "DONE"]
custom: string # Python 表达式 (state, round) → bool
summaryAgent: # 可选的总结 Agent
systemPrompt: string
model: object
编译目标：
GroupChat(
agents=[Lam(a1), Lam(a2), ...],
max_rounds=10,
scheduler="round_robin" | Term(schedulerPrompt),
termination=fn,
summary_agent=Lam(summaryPrompt)
)
19.2 SharedMemory 配置
# ═══ SharedMemory 配置（多 Agent 共享状态）═══
sharedMemory:
enabled: bool # 是否启用
store: # 初始共享数据
key1: value1
key2: value2
appendOnly: bool # true = 类型安全模式（Σ' ⊇ Σ）
编译目标：
SharedMemory(store={"key1": "value1"}, append_only=True)
→ sm.wrap(agent) # 将 Agent 绑定到共享记忆
19.3 Channel 配置
# ═══ Channel 配置（Agent 间通信）═══
channels:
- name: string # 通道名称
capacity: int # 缓冲区大小，0=无缓冲（同步），默认 0
编译目标：
channels = {name: Channel(name, capacity) for ch in config.channels}
19.4 MCP Server 配置（v2 扩展）
# ═══ MCP Server 配置（v2: 结构化 MCP 连接）═══
mcpServers:
- name: string # 服务器名称
transport: string # "http" | "stdio"
url: string # transport=http 时的 URL
command: string # transport=stdio 时的命令
args: [string] # transport=stdio 时的参数
env: # transport=stdio 时的环境变量
KEY: VALUE
headers: # transport=http 时的 HTTP 头
Authorization: string
timeout: float # 超时秒数，默认 30
tools: [string] # 只暴露这些工具（空=全部），可选
编译目标：
server = MCPServer.http(url, headers) | MCPServer.stdio(command, args, env)
tools = server.to_tools() # 或 [server.to_tool(name) for name in config.tools]
19.5 Handoff 配置
# ═══ Handoff 配置（动态委派）═══
handoff:
selector: # 选择器
prompt: string # LLM 分类器 prompt
model: object # 可选模型覆盖
registry: # Agent 注册表
<name>: # 目标 Agent 名称
type: string
systemPrompt: string
# ... 子 agent 完整配置
fallback: # 兜底 Agent（可选）
systemPrompt: string
编译目标：
Handoff(
selector=Lam(selectorPrompt),
registry={name: compile(sub_agent) for ...},
fallback=Lam(fallbackPrompt)
)
19.6 Skill 配置
# ═══ Skill 配置（可复用技能）═══
skills:
- name: string # 技能名称
description: string # 自然语言描述
tags: [string] # 标签
inputType: string # 输入类型描述，默认 "Str"
outputType: string # 输出类型描述，默认 "Str"
agent: # 技能实现
type: string
systemPrompt: string
# ... 完整 agent 配置
examples: # 使用示例
- input: string
output: string
编译目标：
Skill(
name=name, term=compile(agent_config),
description=description,
signature=SkillSignature(inputType, outputType),
tags=tags, examples=[(inp, out), ...]
)
19.7 Checkpoint 配置
# ═══ Checkpoint 配置（状态持久化）═══
checkpoint:
enabled: bool # 是否启用自动 checkpoint
directory: string # checkpoint 目录路径
maxCheckpoints: int # 最多保留多少个，默认 10
autoSaveSteps: int # 每 N 步 β-规约自动保存，0=不自动保存
编译目标：在 Runtime 中注入 CheckpointManager(directory, max_checkpoints)。
19.8 A2A 配置
# ═══ A2A Protocol 配置（Agent 间协议）═══
a2a:
server: # 发布为 A2A 服务器（可选）
port: int # 监听端口，默认 8000
host: string # 监听地址，默认 "0.0.0.0"
clients: # 远程 A2A Agent（可选）
- name: string # 本地名称
url: string # 远程 A2A endpoint URL
timeout: float # 超时秒数，默认 60
编译目标：
# 远程 Agent 封装为本地 Term
remote_agents = {c.name: A2AClient(c.url, timeout=c.timeout) for c in config.a2a.clients}
# 可注入 Handoff / Route 的 registry
19.9 字段分级更新
| 级别 | 字段 | Lambda 语义 | 编译行为 |
|---|---|---|---|
| 必须 | type, systemPrompt |
Agent 结构 + λ body | 缺失则报编译错误 |
| 核心 | model, react, mcp |
计算单元 + Y 参数 + Oracle | 缺失则用默认值 |
| 增强 | memory, guard, chain, router, parallel |
环境扩展 + 类型约束 + 组合 | 缺失则不包裹 |
| 多智能体 | groupchat, sharedMemory, channels, handoff |
π-演算扩展 | 缺失则不启用 |
| 协议 | mcpServers, a2a, skills |
MCP + A2A + Skill 系统 | 缺失则不启用 |
| 持久化 | checkpoint, rag |
状态保存 + 检索增强 | 缺失则不启用 |
| 安全 | sandbox |
进程隔离 + 资源限制 | 缺失则不启用 |
| 元数据 | agentId, name, description |
不影响 Lambda 语义 | 只用于标识和追踪 |
20. Sandbox 配置字段
20.1 YAML Schema
sandbox:
enabled: true # 是否启用沙箱（默认 false）
timeout: 30 # CPU 超时（秒），对应 RLIMIT_CPU
memory_mb: 256 # 内存上限（MB），对应 RLIMIT_AS
network: false # 是否允许网络访问
allow_subprocess: false # 是否允许创建子进程（RLIMIT_NPROC）
max_output_bytes: 65536 # 最大输出字节数
max_fds: 64 # 最大文件描述符数（RLIMIT_NOFILE）
policy: default # 预设策略: strict | default | permissive
20.2 预设策略
当指定 policy 时，其余字段作为覆盖值（override）。未指定的字段使用预设默认值。
| 字段 | strict |
default |
permissive |
|---|---|---|---|
timeout |
5 | 30 | 300 |
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 |
20.3 编译目标
# sandbox.enabled = true 时:
from lambdagent import SandboxPolicy, SecureExecutor
if config.sandbox.policy:
policy = getattr(SandboxPolicy, config.sandbox.policy)()
else:
policy = SandboxPolicy(
timeout=config.sandbox.timeout,
memory_mb=config.sandbox.memory_mb,
network=config.sandbox.network,
allow_subprocess=config.sandbox.allow_subprocess,
max_output_bytes=config.sandbox.max_output_bytes,
max_fds=config.sandbox.max_fds,
)
# 字段级覆盖（policy 为基础，YAML 字段覆盖）
if config.sandbox.timeout is not None:
policy.timeout = config.sandbox.timeout
# ...
executor = SecureExecutor(policy=policy)
term = executor.sandbox_all_tools(term) # 递归包裹所有 Tool
20.4 Lambda 语义
Sandbox 不改变 Lambda 语义。编译后的 Term 树类型不变：
⟦sandbox: {enabled: true, policy: P}⟧ =
map(λt. SandboxedTool(t.name, t.fn, P), Tools(term))
所有 Tool 节点被替换为 SandboxedTool，但 SandboxedTool <: Tool，Term 树结构不变。