# 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

```yaml
# 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

```python
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

```yaml
# 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

```python
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`:

```python
# 只允许特定 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

```yaml
# 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

```python
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

```yaml
# 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

```python
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

```yaml
# 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

```python
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:

```python
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:

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