Core Constructs

lambdagent maps 11 agent constructs to Lambda calculus. Each construct is a subclass of Term and supports >> (composition), | (parallel), and () (application / beta-reduction).

#	Lambda Calculus	DSL	Construct
1	`lambda_D . F_{M,D}`	`Lam(name, prompt)`	LLM oracle call
2	`(f x)`	`agent(input)`	Function application
3	`lambda x. g(f(x))`	`f >> g`	Function composition
4	`IF c t e`	`If(cond, then_, else_)`	Church conditional
5	`Y combinator`	`Loop(body, cond, N)`	Bounded recursion
6	`PAIR = lambda a.lambda b.lambda f. f a b`	`Pair(f, g)`	Church pair
7	`FST / SND`	`Fst() / Snd()`	Pair projections
8	`Oracle`	`Tool(name, fn)`	External function
9	`CASE`	`Route(cls, routes)`	Multi-way dispatch
10	`{x : T \\| P(x)}`	`Guard(agent, P)`	Output validation
11	`Gamma' = Gamma union s`	`Memory(agent, store)`	Environment extension

1. Lam -- LLM Oracle Call

Lambda correspondence: lambda_D . F_{M,D} -- Lambda abstraction where the body is defined by a dataset D (prompt) and model M.

Lam is the fundamental construct. It wraps an LLM call as a Lambda abstraction. Calling a Lam performs beta-reduction, which corresponds to one round of autoregressive decoding.

Lam("summarizer", "Summarize concisely.")
  = lambda_D . F_{M,D}

(summarizer "long text...")
  = F_{M,D}("long text...")        -- beta-reduction = LLM inference
  = "A concise summary..."         -- autoregressive decoding result

Provider Examples

from lambdagent.primitives import Lam

# Anthropic Claude
claude = Lam(
    "writer",
    prompt="你是专业技术写手。",            # 系统提示 = Lambda body = 数据集 D
    model="claude-sonnet-4-20250514",
    temperature=0.3,
    max_tokens=2048,
)

# OpenAI GPT
gpt = Lam(
    "analyst",
    prompt="你是数据分析师。",
    model="gpt-4o",
    temperature=0.0,
)

# DashScope (阿里通义千问)
qwen = Lam(
    "translator",
    prompt="你是中英翻译专家。",
    model="dashscope/qwen3-max",
)

# Ollama (本地模型)
local = Lam(
    "coder",
    prompt="你是编程助手。",
    model="ollama/llama3",
)

Type and Effect

Lam : Str ->^{llm(model)} Str

# 类型: 输入字符串，输出字符串
# 效果: llm(claude-sonnet-4-20250514) -- LLM 调用效果

Output Parser

Lam supports custom output parsing to transform LLM text output into structured data:

import json

# 将 LLM 输出解析为 JSON
structured = Lam(
    "extractor",
    prompt="提取实体，返回 JSON 数组。",
    output_parser=lambda x: json.loads(x),  # str -> list
)

2. Compose (>>) -- Pipeline

Lambda correspondence: lambda x. g(f(x)) -- function composition.

Compose chains multiple agents into a sequential pipeline. Each stage's output becomes the next stage's input. Each >> step is one beta-reduction.

f >> g >> h = Compose(f, g, h)
pipeline(x) = h(g(f(x)))

3-Stage Pipeline Example

from lambdagent.primitives import Lam

# 三步研究管道: 提取 >> 分析 >> 报告
extract = Lam("extract", "从原始文本中提取关键信息和数据点。")
analyze = Lam("analyze", "分析提取的数据，识别模式和趋势。")
report  = Lam("report", "将分析结果整理为结构化报告。")

pipeline = extract >> analyze >> report

result = pipeline("...长篇研究论文...")
# 执行顺序:
#   beta[0]: extract("...长篇研究论文...")  -> "关键信息: ..."
#   beta[1]: analyze("关键信息: ...")       -> "趋势分析: ..."
#   beta[2]: report("趋势分析: ...")        -> "# 研究报告\n..."

Type Checking (T-Compose)

The >> operator checks the T-Compose typing rule at construction time:

f: A ->^e1 B,  g: B' ->^e2 C,  B <: B'
-----------------------------------------
          f >> g : A ->^{e1 . e2} C

If output_type(f) is not a subtype of input_type(g), an AgentTypeError is raised:

from lambdagent.types import T_STR, T_JSON

f = Lam("f", "...")
f.output_type = T_STR

g = Lam("g", "...")
g.input_type = T_JSON({"type": "object", "required": ["name"]})

f >> g  # AgentTypeError: Str is not a subtype of Json({name: string})

Flattening

Nested compositions are automatically flattened: (f >> g) >> h becomes Compose(f, g, h), not Compose(Compose(f, g), h).

3. If -- Conditional Branching

Lambda correspondence: IF = lambda c. lambda t. lambda e. c t e -- Church conditional.

If evaluates a condition and dispatches to one of two branches.

If(cond, then_, else_)
  = IF cond THEN then_ ELSE else_
  = lambda c. lambda t. lambda e. c t e

Example

from lambdagent.primitives import Lam, If, Tool

# 根据文本长度选择不同处理策略
short_handler = Lam("short", "简短回复即可。")
long_handler = Lam("long", "进行深度分析，给出详细报告。")

# 条件可以是 Python 函数或 Term
branch = If(
    cond=lambda x: len(str(x)) > 500,  # Python 函数条件
    then_=long_handler,                  # 长文本 -> 深度分析
    else_=short_handler,                 # 短文本 -> 简短回复
)

result = branch("Hello")           # -> short_handler("Hello")
result = branch("A" * 1000)        # -> long_handler("AAA...")

LLM-Based Condition

The condition can itself be a Term (LLM agent):

# LLM 判断是否需要深入分析
classifier = Lam("needs_analysis", "判断输入是否需要深入分析。回答 TRUE 或 FALSE。")

branch = If(
    cond=classifier,       # LLM 决定分支
    then_=long_handler,
    else_=short_handler,
)

Truthy Evaluation

String results are converted to boolean: "TRUE", "YES", "1" (case-insensitive) are truthy; everything else is falsy.

4. Loop -- Y Combinator / ReAct

Lambda correspondence: Y = lambda f. (lambda x. f(x x)) (lambda x. f(x x)) -- the Y combinator with bounded unfolding.

Loop implements bounded recursion. The body is applied repeatedly, feeding its output back as input, until the condition returns True or max_steps is reached.

Loop(body, condition, N)(x) =
    let r0 = body(x) in
    if condition(r0, 0) then r0
    else let r1 = body(r0) in
         if condition(r1, 1) then r1
         else ... (up to N iterations)

The terminate Tool = Base Case

In ReAct-style agents, the terminate tool serves as the Y combinator's base case -- the identity function lambda x. x that stops recursion:

# ReAct 配置: Loop + Route 的组合
type: react
systemPrompt: "你是研究助手。"
react:
  maxSteps: 20              # Y 组合子的界 N
mcp:
  localTools:
    - search                # 工具调用 -> 递归继续
    - calculate
    - terminate             # base case: lambda x. x -> 递归终止

Python Example

from lambdagent.primitives import Lam, Loop

# 迭代改进: 反复润色文章直到满意
refiner = Lam("refiner", "改进以下文本的表达质量。如果已经很好，原样返回并加上 DONE。")

loop = Loop(
    body=refiner,
    condition=lambda result, step: "DONE" in str(result).upper(),  # base case
    max_steps=5,            # 最多 5 次迭代
)

result = loop("这是一个粗糙的草稿...")
# beta[0]: refiner("草稿...") -> "改进版 1..."
# beta[1]: refiner("改进版 1...") -> "改进版 2..."
# beta[2]: refiner("改进版 2...") -> "DONE: 最终版本..."  <- 终止

When maxSteps > 10, the linter recommends using runtime.engine: cek for cost monitoring and loop detection.

5. Pair / Fst / Snd -- Parallel Execution + Projection

Lambda correspondence:

PAIR = lambda a. lambda b. lambda f. f a b
FST  = lambda p. p TRUE   = lambda p. p (lambda a. lambda b. a)
SND  = lambda p. p FALSE  = lambda p. p (lambda a. lambda b. b)

Pair runs two agents on the same input and returns a tuple of results. Fst and Snd extract the first and second elements.

Example

from lambdagent.primitives import Lam, Pair, Fst, Snd

researcher = Lam("researcher", "研究给定主题的最新进展。")
critic     = Lam("critic", "从反面批判给定主题的弱点。")

# Pair: 同时运行两个 Agent
both = Pair(researcher, critic)
results = both("量子计算")
# results = ("最新进展: ...", "弱点分析: ...")

# Fst/Snd: 投影取值
research_only = Pair(researcher, critic) >> Fst()   # -> "最新进展: ..."
critique_only = Pair(researcher, critic) >> Snd()   # -> "弱点分析: ..."

Pair + Compose Pattern

A common pattern is Pair followed by a merge agent:

merger = Lam("merger", "综合正反两方观点，给出平衡的结论。输入是一个包含两个视角的元组。")

# 研究 + 批评 -> 综合
balanced = Pair(researcher, critic) >> merger
result = balanced("AI 在医疗领域的应用")

Type

Pair(f, g) : A ->^{e1 || e2} (B, C)
    where f : A ->^e1 B, g : A ->^e2 C

Fst : (A, B) -> A
Snd : (A, B) -> B

Runtime Behavior

Pair and Par execute branches in a thread pool with forked contexts (Paper II Proposition 30). Each branch gets a deep copy of the context to prevent race conditions.

6. Tool -- External Function

Lambda correspondence: lambda x. oracle(x) -- an external oracle lifted into the Lambda world.

Tool wraps any Python function as a Lambda term, enabling external capabilities (search, code execution, database queries, APIs) to participate in agent pipelines.

Tool("double", lambda x: int(x) * 2)
  = lambda x. double(x)

Example

from lambdagent.primitives import Tool, Lam
import json

# 数学计算工具
calculator = Tool("calc", lambda x: str(eval(x)))  # 注意: 生产环境应使用沙箱

# 搜索工具
def search(query):
    """模拟搜索 API"""
    return f"搜索结果: 关于 '{query}' 找到 3 条相关信息..."

search_tool = Tool("search", search)

# 工具与 Agent 组合
researcher = Lam("researcher", "分析搜索结果并总结。")
pipeline = search_tool >> researcher

result = pipeline("量子计算最新论文")
# beta[0]: search("量子计算最新论文") -> "搜索结果: ..."
# beta[1]: researcher("搜索结果: ...") -> "总结: ..."

Type and Effect

Tool : A ->^io B

# 效果: io -- I/O 效果 (外部副作用)

Effect Handler Integration

When an effect handler is active, Tool calls are routed through handler.handle_tool():

from lambdagent.handlers import TestHandler, with_handler

handler = TestHandler()
handler.mock_tool("search", "Mock 搜索结果: 找到 5 篇论文")

with with_handler(handler):
    result = search_tool("量子计算")  # -> "Mock 搜索结果: 找到 5 篇论文"

7. Route -- Multi-Way Dispatch

Lambda correspondence: CASE (classifier x) [(l1, a1), (l2, a2), ...] -- generalized Church boolean.

Route is a multi-way conditional. A classifier agent determines a label, then dispatches to the corresponding branch. It generalizes Church booleans from 2 choices (TRUE/FALSE) to N choices.

Route(classifier, {"code": coder, "math": solver, "general": chatbot})
  = lambda x. CASE (classifier x)
        "code"    -> coder(x)
        "math"    -> solver(x)
        "general" -> chatbot(x)

Example

from lambdagent.primitives import Lam
from lambdagent.extensions import Route

# 分类器 Agent: 判断问题类型
classifier = Lam(
    "classifier",
    "判断用户问题的类型。只返回以下之一: code / math / general",
    temperature=0.0,
)

# 专业 Agent
coder   = Lam("coder", "你是编程专家。解答代码问题。")
solver  = Lam("solver", "你是数学专家。解答数学问题。")
chatbot = Lam("chatbot", "你是通用助手。")

# 路由分发
router = Route(
    classifier=classifier,
    routes={"code": coder, "math": solver, "general": chatbot},
    default=chatbot,  # 无匹配时的 fallback
)

result = router("如何用 Python 实现快速排序？")  # -> coder
result = router("求解方程 x^2 + 2x - 3 = 0")    # -> solver

Fuzzy Matching

Route supports fuzzy label matching: if the classifier returns "this is a code question", it will match the "code" route because "code" is a substring of the label.

Type and Effect

Route(cls, routes, default) : A ->^{e_cls . max(e_routes)} Union(B1, ..., Bn)

8. Guard -- Output Validation with Retry

Lambda correspondence: {x : T | P(x)} -- dependent type / refinement type.

Guard wraps an agent and validates its output against a predicate. If validation fails, it retries (up to retry times) or invokes a fallback.

Guard(agent, P, retry=2) =
    lambda x. let r = agent(x) in
              if P(r) then r
              else let r' = agent(x) in
                   if P(r') then r'
                   else let r'' = agent(x) in
                        if P(r'') then r''
                        else raise ValidationError

Example

from lambdagent.primitives import Lam
from lambdagent.extensions import Guard

writer = Lam("writer", "写一篇至少 200 字的文章。")

# 验证输出长度至少 200 字
validated_writer = Guard(
    agent=writer,
    validator=lambda x: len(str(x)) >= 200,  # 谓词 P
    retry=2,                                   # 最多重试 2 次 (共 3 次尝试)
    on_fail=lambda x: f"[FALLBACK] 输出太短 ({len(str(x))} 字): {x}",
)

result = validated_writer("写一篇关于 AI 安全的文章")

LLM-Based Validator

The validator can itself be a Term:

quality_checker = Lam(
    "checker",
    "评估文本质量。如果文章逻辑清晰、论据充分，返回 TRUE；否则返回 FALSE。",
)

guarded = Guard(
    agent=writer,
    validator=quality_checker,  # LLM 验证
    retry=3,
)

Type and Effect

Guard(agent, P, k) : A ->^{e^(1+k)} B
    where agent : A ->^e B

# 效果是 agent 效果的 (1 + retry) 次迭代
# 最坏情况: 所有重试都执行

9. Memory -- Persistent State Injection

Lambda correspondence: Gamma' = Gamma union s -- environment extension.

Memory wraps an agent with persistent state that is injected into the input as context. The state persists across multiple invocations.

Memory(agent, store) = lambda x. agent(x) [Gamma union store]

The store contents are serialized and prepended to the input:

[Memory]
- user: Alice
- preference: technical

[Input]
What is quantum computing?

Example

from lambdagent.extensions import Memory
from lambdagent.primitives import Lam

assistant = Lam("assistant", "你是个人助手。根据记忆中的用户信息个性化回答。")

# 创建带记忆的 Agent
smart_assistant = Memory(
    agent=assistant,
    store={
        "user_name": "Alice",
        "language": "Chinese",
        "expertise": "machine learning",
    },
)

result = smart_assistant("推荐一本好书")
# 输入被增强为:
# [Memory]
# - user_name: Alice
# - language: Chinese
# - expertise: machine learning
#
# [Input]
# 推荐一本好书

# 动态更新记忆
smart_assistant.remember("last_book", "Deep Learning by Goodfellow")
smart_assistant.forget("language")

Type and Effect

Memory(agent, store) : A ->^{state . e} B
    where agent : A ->^e B

# 效果: state 效果 (读/写存储) 串行组合 agent 的效果

YAML Configuration

All 11 constructs can be expressed in YAML via from_config(). Here is a comprehensive example combining multiple constructs:

# 完整示例: 多构造组合
agentId: research-pipeline
name: "Research Pipeline"
type: chain                           # Compose: 串行管道

stages:
  # Stage 1: Route -- 分类器选择研究方向
  - type: router
    classifier:
      systemPrompt: "判断研究主题类型: science / technology / social"
    routes:
      science:
        systemPrompt: "你是科学研究专家。"
      technology:
        systemPrompt: "你是技术分析专家。"
      social:
        systemPrompt: "你是社会科学研究员。"

  # Stage 2: Loop -- 迭代深化研究
  - type: react
    systemPrompt: "深入研究，使用工具收集信息。完成后调用 terminate。"
    react:
      maxSteps: 10                    # Y combinator bound
    mcp:
      localTools:
        - search
        - terminate                   # base case: lambda x. x

  # Stage 3: Guard -- 验证输出质量
  - type: guard
    agent:
      systemPrompt: "将研究结果整理为结构化报告。"
    validator: "len(output) > 500"    # 至少 500 字
    retry: 2

  # Stage 4: Memory -- 记录研究历史
  - type: memory
    agent:
      systemPrompt: "根据记忆中的历史研究，补充交叉引用。"
    store:
      previousResearch: []

model:
  provider: anthropic
  name: claude-sonnet-4-20250514
  temperature: 0.3

runtime:
  engine: adaptive                    # 小配置用 recursive, 复杂配置自动切换 cek

Operator Overloading Summary

All constructs support Python operator syntax:

Operator	Meaning	Example
`f >> g`	Composition	`extract >> analyze >> report`
`f \\| g`	Parallel (Par)	`researcher \\| critic`
`f(x)`	Application (beta-reduction)	`agent("hello")`

These operators enable concise, readable agent programs that mirror the underlying Lambda calculus structure.

constructs.md 17 KB História Raw

Core Constructs

1. Lam -- LLM Oracle Call

Provider Examples

Type and Effect

Output Parser

2. Compose (>>) -- Pipeline

3-Stage Pipeline Example

Type Checking (T-Compose)

Flattening

3. If -- Conditional Branching

Example

LLM-Based Condition

Truthy Evaluation

4. Loop -- Y Combinator / ReAct

The terminate Tool = Base Case

Python Example

5. Pair / Fst / Snd -- Parallel Execution + Projection

Example

Pair + Compose Pattern

Type

Runtime Behavior

6. Tool -- External Function

Example

Type and Effect

Effect Handler Integration

7. Route -- Multi-Way Dispatch

Example

Fuzzy Matching

Type and Effect

8. Guard -- Output Validation with Retry

Example

LLM-Based Validator

Type and Effect

9. Memory -- Persistent State Injection

Example

Type and Effect

YAML Configuration

Operator Overloading Summary

constructs.md 17 KB

História Raw