lambdagent/docs/multi-agent.md

Multi-Agent Patterns

lambdagent extends the core Lambda calculus with 5 multi-agent constructs drawn from pi-calculus (process algebra). These constructs enable agent-to-agent communication, shared state, group conversation, dynamic delegation, and true parallel execution.

Engine recommendation: Parallel agents (Par/Pair/AsyncPar) should use runtime.engine: cek or adaptive to enable Pair confluence tracking and store independence checking.

---

1. GroupChat: Multi-Agent Conversation

GroupChat orchestrates multiple agents taking turns in a structured conversation until consensus is reached or a round limit is hit.

Lambda Semantics

GroupChat([a, b, c], scheduler, n) =
    Y_n(lambda self. lambda state.
        let speaker = scheduler(state) in
        let msg     = speaker(state) in
        let state'  = state ++ msg in
        IF done(state') THEN state' ELSE self(state')
    )

GroupChat is not a new Lambda primitive -- it is a composition of Loop + Route. The Y combinator provides iteration, and the scheduler provides routing.

Scheduling Strategies

Strategy	Description	Use Case
"round_robin"	Fixed-order rotation through agent list	Predictable debate structure
"random"	Random speaker selection each round	Diverse exploration
Term (LLM classifier)	LLM decides who speaks next based on conversation state	Intelligent moderation

Configuration

# GroupChat YAML 示例: 三人研究小组
type: groupchat
agents:
  - name: researcher
    systemPrompt: "你是研究员，负责查找和分析信息。"
    model:
      provider: anthropic
      name: claude-sonnet-4-20250514
  - name: critic
    systemPrompt: "你是批评者，负责质疑论点、发现漏洞。"
    model:
      provider: anthropic
      name: claude-sonnet-4-20250514
  - name: synthesizer
    systemPrompt: "你是总结者，负责整合观点、形成共识。"
    model:
      provider: anthropic
      name: claude-sonnet-4-20250514
scheduler: round_robin        # 调度策略
maxRounds: 10                 # Y 组合子的界 n
termination:                  # 终止关键词 (base case)
  keywords: ["CONSENSUS", "DONE", "TERMINATE"]
summary: true                 # 对话结束后自动总结

Python API

from lambdagent.multiagent import GroupChat
from lambdagent.primitives import Lam

researcher = Lam("researcher", "你是研究员，查找信息。")
critic = Lam("critic", "你是批评者，质疑论点。")
synthesizer = Lam("synthesizer", "你是总结者，整合观点。")

# 基础用法: round-robin 调度
chat = GroupChat(
    agents=[researcher, critic, synthesizer],
    max_rounds=10,
    scheduler="round_robin",
    termination=lambda state, r: "CONSENSUS" in state.upper(),
)
result = chat("分析量子计算对密码学的影响")

# 高级用法: LLM 分类器调度
moderator = Lam("moderator", "根据对话状态，选择下一个发言者。返回 researcher/critic/synthesizer 之一。")
chat = GroupChat(
    agents=[researcher, critic, synthesizer],
    scheduler=moderator,  # LLM 驱动的调度
    max_rounds=15,
)

Smart Context Window

For conversations exceeding 6 messages, GroupChat automatically builds a smart context window for each speaker:

1. First-round messages (topic establishment)
2. The speaker's own previous messages (last 3)
3. Most recent 3 * len(agents) messages

This prevents context overflow while preserving conversation coherence.

---

2. Channel + Send + Receive: Pi-Calculus Communication

Channels provide typed, thread-safe message passing between agents, directly corresponding to pi-calculus name-passing communication.

Lambda + Pi Semantics

Channel:  nu(c). (P | Q)     -- P and Q communicate through private channel c
Send:     c!(v)               -- output value v to channel
Receive:  c?(x).P             -- receive from channel, bind to x, continue as P

Configuration

# Channel 通信示例: 生产者-消费者
type: parallel
agents:
  producer:
    systemPrompt: "生成研究主题列表。"
    channel_out: research_topics    # 输出到通道
  consumer:
    systemPrompt: "对每个主题撰写摘要。"
    channel_in: research_topics     # 从通道读取
channels:
  research_topics:
    capacity: 10                    # 缓冲区大小 (0=同步)
    allowed_agents:                 # 权限控制
      - producer
      - consumer

Python API

from lambdagent.multiagent import Channel, Send, Receive
from lambdagent.primitives import Lam

# 创建通道 (capacity=0 为同步通道, >0 为异步缓冲)
ch = Channel("data_pipe", capacity=5)

# 生产者: 执行 agent 后将结果发送到通道
producer = Send(
    agent=Lam("extractor", "从文本中提取关键实体。"),
    channel=ch,
)

# 消费者: 从通道接收消息，可选地传给 handler 处理
consumer = Receive(
    channel=ch,
    handler=Lam("analyzer", "分析实体之间的关系。"),
    timeout=30.0,  # 超时秒数
)

# 在不同线程中运行
import threading
threading.Thread(target=lambda: producer("长篇研究论文内容...")).start()
result = consumer(None)  # 阻塞等待通道消息

Channel Access Control

Channels support per-agent access control via allowed_agents:

# 只允许特定 Agent 访问通道
ch = Channel("secure_pipe", allowed_agents={"producer", "consumer"})
ch.send("data", agent_name="producer")     # OK
ch.send("data", agent_name="attacker")     # PermissionError

---

3. Handoff: Dynamic Agent Delegation

Handoff enables runtime-determined routing, where the target agent is selected dynamically -- potentially from agents registered after the Handoff was created.

Lambda Semantics

Handoff(selector, registry) =
    lambda x. let target = selector(x) in
              let agent  = registry[target] in
              agent(x)

Handoff vs. Route

Feature	Route (static CASE)	Handoff (dynamic CASE)
Route table	Fixed at compile time	Mutable at runtime
New agents	Requires recompilation	handoff.register(name, agent)
Selector	Classifier returns label from known set	Selector returns any string
Fallback	default branch	fallback agent

Configuration

# Handoff 示例: 客服动态路由
type: handoff
selector:
  systemPrompt: "根据用户问题类型，返回: billing / technical / general"
  model:
    provider: anthropic
    name: claude-sonnet-4-20250514
registry:
  billing:
    systemPrompt: "你是计费专家。处理账单、退款等问题。"
  technical:
    systemPrompt: "你是技术支持。解决产品使用问题。"
  general:
    systemPrompt: "你是通用客服。处理一般咨询。"
fallback:
  systemPrompt: "我是通用助手，让我尝试帮助你。"

Python API

from lambdagent.multiagent import Handoff
from lambdagent.primitives import Lam

selector = Lam("router", "分析问题类型，返回 billing/technical/general")

handoff = Handoff(
    selector=selector,
    registry={
        "billing": Lam("billing", "处理计费问题"),
        "technical": Lam("tech", "处理技术问题"),
    },
    fallback=Lam("general", "通用客服"),
)

# 运行时动态注册新 Agent
handoff.register("security", Lam("security", "处理安全问题"))

result = handoff("我的账单好像多扣了钱")  # -> billing agent

---

4. AsyncPar: True Parallel Execution

AsyncPar provides genuine concurrent execution via a thread pool, unlike Par which also uses threads but without formal safety guarantees. AsyncPar enforces store independence checking based on Paper II Proposition 30.

Lambda Semantics

AsyncPar(f, g) = lambda x. let (r1, r2) = concurrent(f(x), g(x)) in (r1, r2)

Pair Confluence (Paper II Proposition 30)

For parallel execution to be deterministic (confluent), the parallel branches must be store-independent:

writes(f) intersection writes(g) = emptyset
    => result is independent of scheduling strategy

AsyncPar checks this condition before execution via check_store_independence(). Each branch receives a forked Context (deep copy of bindings, independent trace and memory) to prevent race conditions.

Configuration

# AsyncPar 示例: 并行研究 + 批评
type: parallel
runtime:
  engine: cek               # 建议使用 CEK 引擎以获得并行安全检查
agents:
  - name: researcher
    systemPrompt: "深入研究给定主题。"
  - name: critic
    systemPrompt: "从反面论证给定主题的弱点。"
  - name: factchecker
    systemPrompt: "验证给定主题中的事实准确性。"
maxWorkers: 3                # 线程池大小
timeout: 120                 # 总超时 (秒)
checkStoreIndependence: true # 执行前检查存储独立性

Python API

from lambdagent.multiagent import AsyncPar
from lambdagent.primitives import Lam

research = Lam("research", "深入研究主题")
critique = Lam("critique", "批判性分析主题")
factcheck = Lam("factcheck", "验证事实准确性")

parallel = AsyncPar(
    research, critique, factcheck,
    max_workers=3,
    timeout=120.0,
    check_store_independence=True,  # Paper II Prop. 30
)

# 返回 (research_result, critique_result, factcheck_result)
results = parallel("量子计算的商业前景")

---

5. SharedMemory: Thread-Safe Shared State

SharedMemory provides a multi-agent shared environment (Gamma_shared), enabling agents to read and write a common state store with thread safety.

Lambda Semantics

SharedMem(store)   = create shared environment Gamma_shared
sm.wrap(agent)     = lambda x. agent(x) [Gamma union Gamma_shared]
sm.read(key)       = Gamma_shared(key)
sm.write(key, v)   = Gamma_shared[key |-> v]

Configuration

# SharedMemory 示例: 协作写作
type: groupchat
sharedMemory:
  appendOnly: true           # Sigma' supseteq Sigma 约束 (已写入的 key 不可变类型)
  initial:
    topic: "AI 安全"
    outline: ""
    draft: ""
agents:
  - name: outliner
    systemPrompt: "创建文章大纲。写入 sharedMemory.outline"
    sharedMemory: true       # 绑定到共享记忆
  - name: writer
    systemPrompt: "根据大纲撰写草稿。读取 outline，写入 draft"
    sharedMemory: true

Python API

from lambdagent.multiagent import SharedMemory
from lambdagent.primitives import Lam

# 创建共享记忆 (append_only=True 时已有 key 不可修改类型)
shared = SharedMemory(
    store={"topic": "AI Safety", "notes": ""},
    append_only=True,
)

outliner = shared.wrap(Lam("outliner", "创建大纲"))
writer = shared.wrap(Lam("writer", "根据大纲写作"))

# 两个 Agent 共享同一个 store
outliner("写一篇关于 AI 安全的文章")
shared.write("outline", "1. Introduction\n2. Risks\n3. Mitigations")
writer("根据大纲撰写正文")

# 读取共享状态
print(shared.read("outline"))
print(shared.read_all())

Type Safety

When append_only=True, SharedMemory enforces the store typing invariant Sigma' supseteq Sigma from the Preservation theorem:

shared.write("count", 42)       # OK: 首次写入，记录类型为 int
shared.write("count", 100)      # OK: 类型一致
shared.write("count", "hello")  # TypeError: key 'count' expects int, got str

---

Decision Tree: When to Use Which Pattern

Do your agents need to communicate?
|
+-- No --> Do they share state?
|          |
|          +-- No --> Are they independent?
|          |         |
|          |         +-- Yes --> AsyncPar (真并行, 最快)
|          |         +-- No  --> Compose >> (串行管道)
|          |
|          +-- Yes --> SharedMemory
|
+-- Yes --> Is it structured turn-taking?
           |
           +-- Yes --> GroupChat
           |          (round_robin / random / LLM scheduler)
           |
           +-- No  --> Is the target known at compile time?
                      |
                      +-- Yes --> Route (静态分发)
                      +-- No  --> Handoff (动态委派)
                                  |
                                  Need message passing?
                                  +-- Yes --> Channel + Send/Receive

Quick Reference

Pattern	Latency	Complexity	Store Safety	Best For
AsyncPar	Lowest (parallel)	Low	Enforced (fork)	Independent LLM calls
Pair/Par	Low (parallel)	Low	Fork-based	Simple parallel with projection
GroupChat	High (sequential rounds)	Medium	Per-round isolation	Debate, brainstorming, consensus
Handoff	Medium (2 LLM calls)	Medium	N/A	Dynamic routing, customer service
Channel	Varies	High	Manual	Producer-consumer, pipelines
SharedMemory	Medium	Medium	Lock-based	Collaborative state building

---

Integration with CEK Engine

For parallel constructs, the CEK machine (runtime.engine: cek) provides additional safety guarantees:

1. Pair Confluence Tracking: The CEK machine evaluates Pair(f, g) left-to-right via PairLK/PairRK continuation frames. The transition trace records the exact interleaving, enabling post-hoc verification of confluence.

2. Store Independence Checking: Before parallel execution, check_store_independence() analyzes each branch's effect annotations to verify writes(f) intersection writes(g) = emptyset.

3. Cost Monotonicity: The CEK machine verifies Paper II Proposition 23 at every step, ensuring parallel branches don't introduce negative cost deltas.

4. Context Forking: Each parallel branch receives a Context.fork() -- a deep copy that prevents shared mutable state. Traces are merged back to the parent context after completion.

Configure via YAML:

runtime:
  engine: cek           # 或 adaptive (小配置用 recursive, 大配置自动切换 cek)
  costBudget: 1.50      # 成本上限 (USD)
  maxSteps: 10000       # 最大 CEK 转移步数