# lambdagentpaas 使用指南

> 本文档面向三类用户：
> 1. **lambdagent 用户** — 用 YAML 定义 Agent，用 CLI/API 运行
> 2. **LangChain/CrewAI/AutoGen 用户** — 不换框架，加一层安全检查
> 3. **PaaS 运维** — 部署和管理多租户 Agent 平台

---

## 一、5 分钟快速开始

### 安装

```bash
pip install lambdagent
```

### 写一个 Agent

```yaml
# agent-config.yml
agentId: my-first-agent
name: 我的第一个 Agent
type: react

model:
  provider: dashscope          # anthropic | openai | dashscope | ollama
  name: qwen3-max-2026-01-23
  temperature: 0.3

systemPrompt: |
  你是一个有用的助手。当用户提问时，使用工具搜索答案，
  然后调用 terminate 给出最终回答。

react:
  maxSteps: 10

mcp:
  localTools:
    - WebSearch
    - WebFetch
    - terminate

# 运行时引擎（可选）
runtime:
  engine: recursive            # recursive | cek | adaptive
```

### 运行

```bash
# 方式 1: CLI
lambdagent run agent-config.yml "什么是 Lambda 演算？"

# 方式 2: Python
from lambdagent.agentruntime import Runtime
result = Runtime.execute("agent-config.yml", "什么是 Lambda 演算？")
print(result.result)

# 方式 3: 一行代码
from lambdagent.fromconfig import from_config
agent = from_config("agent-config.yml")
print(agent("什么是 Lambda 演算？"))
```

---

## 二、YAML 配置详解

### Agent 类型

| type | Lambda 对应 | 用途 | 示例 |
|------|------------|------|------|
| `simple` | `Lam(name, prompt)` | 单次 LLM 调用 | 翻译、摘要 |
| `react` | `fix_n(λself.λx. case ...)` | ReAct 循环 + 工具调用 | 编程助手、研究员 |
| `chain` | `f >> g >> h` | 固定顺序管道 | 翻译 → 提取 → 分析 |
| `router` | `Route(classifier, {routes})` | 分类 → 分支处理 | 客服分流 |
| `parallel` | `Pair(f, g)` / `Par(agents)` | 并行执行 | 多角度分析 |

### 模型配置

```yaml
model:
  provider: anthropic          # 见下方 Provider 列表
  name: claude-sonnet-4-20250514
  temperature: 0.3             # 0.0-2.0
  maxTokens: 4096              # 单次回复上限
  conversation: true           # 多轮对话（默认开启）
  maxHistoryTokens: 80000      # 对话历史预算
  fallback:                    # 降级模型列表
    - dashscope/qwen-max
    - ollama/qwen2.5:7b
```

**Provider 列表**:

| provider | 说明 | 环境变量 |
|----------|------|---------|
| `anthropic` | Claude API | `ANTHROPIC_API_KEY` |
| `openai` | OpenAI API | `OPENAI_API_KEY` |
| `dashscope` | 通义千问 | `DASHSCOPE_API_KEY` |
| `deepseek` | DeepSeek | `DEEPSEEK_API_KEY` |
| `ollama` | 本地 Ollama | 无需 |
| `claude-code` | Claude Code Max Plan | 无需 |

### 运行时引擎

```yaml
runtime:
  engine: cek                  # recursive | cek | adaptive
  costBudget: 5.00             # CEK: 超 $5 自动暂停
  maxSteps: 10000              # CEK: 最大转移步数
```

| 引擎 | 适用场景 | 独有能力 |
|------|---------|---------|
| `recursive` | 简单 Agent (≤10 步) | 低开销 |
| `cek` | 复杂 Agent (>10 步, 高成本) | 成本熔断, 暂停/恢复, 循环检测, K 栈追踪 |
| `adaptive` | 不确定复杂度 | 自动分析 Term 选最优引擎 |

---

## 三、CLI 命令

### lint — 静态检查

```bash
# 检查单个配置
lambdagent lint agent-config.yml

# 检查目录下所有配置
lambdagent lint agents/

# JSON 输出（CI/CD 用）
lambdagent lint agent-config.yml --format json

# 只看 ERROR
lambdagent lint agent-config.yml --level error
```

输出示例:
```
🔍 Lint: agent-config.yml (framework: lambdagent)

[x] ERROR  L004a  No terminate tool in ReAct loop
    λA: fix₂₀ has no base case λx.x — forced truncation
[!] WARN   L013   Route missing default branch
    λA: case expression is non-exhaustive
[i] INFO   L010   systemPrompt > 2000 chars
    λA: Large LLM input may increase latency

Summary: 1 error, 1 warning, 1 info
```

### compile — 编译为 Lambda 项

```bash
lambdagent compile agent-config.yml
# → 输出: mem(fix₂₀(λs.λx. let t = (lam p θ) x in case t [...]))σ
```

### run — 编译 + 执行

```bash
lambdagent run agent-config.yml "写一个快速排序"

# 指定引擎
lambdagent run agent-config.yml "分析代码安全" --engine cek

# 流式输出
lambdagent run agent-config.yml "研究 AI 安全" --stream
```

---

## 四、Python API

### 基础用法

```python
from lambdagent.fromconfig import from_config
from lambdagent.core import Context

# 编译
agent = from_config("agent-config.yml")

# 执行
result = agent("帮我写一个 Python Web 爬虫")
print(result)

# 带 Context（追踪 + 记忆）
ctx = Context()
result = agent("帮我写一个 Python Web 爬虫", ctx)
ctx.print_trace()  # 打印每步 β-规约
```

### 编程式构建 Agent

```python
from lambdagent.primitives import Lam, Compose, Loop, Tool, If, Pair
from lambdagent.extensions import Route, Guard, Memory

# 单个 LLM Agent
translator = Lam("translator", "将中文翻译为英文")

# 管道: 翻译 → 润色
polisher = Lam("polisher", "润色英文文本，使其更地道")
pipeline = translator >> polisher  # Compose

# 工具
search = Tool("search", lambda q: web_search(q))
calc = Tool("calc", lambda expr: str(eval(expr)))

# 带验证的管道
validated = Guard(
    pipeline,
    validator=lambda x: len(x) > 10,  # 输出必须 >10 字符
    retry=3,
)

# 并行执行
analysis = Pair(
    Lam("security", "分析安全风险"),
    Lam("performance", "分析性能瓶颈"),
)

# 条件路由
router = Route(
    classifier=Lam("classify", "分类用户意图: code/doc/question"),
    routes={
        "code": Lam("coder", "编写代码"),
        "doc": Lam("writer", "编写文档"),
        "question": Lam("qa", "回答问题"),
    },
)
```

### 引擎选择

```python
from lambdagent.agentruntime import Runtime, RecursiveEngine, CEKEngine, AdaptiveEngine

# 方式 1: 通过 Runtime
result = Runtime.execute("config.yml", "input", engine_mode="cek", cost_budget=5.0)

# 方式 2: 直接使用引擎
from lambdagent.core import Context

engine = CEKEngine(cost_budget=5.0, max_steps=10000)
agent = from_config("config.yml")
result = engine.execute(agent, "input", Context())

print(f"Cost: ${result.cost.money:.4f}")
print(f"Steps: {result.steps}")
print(f"Engine: {result.engine_mode}")

# CEK 独有: 查看 K 栈
if result.final_state:
    print(f"Final K: {result.final_state.kont}")

# CEK 独有: 完整转移记录
if result.transitions:
    for t in result.transitions[:5]:
        print(f"  [{t.step}] {t.rule} {t.label}")
```

### 效果处理器（测试/生产切换）

```python
from lambdagent.handlers import TestHandler, set_current_handler, clear_handler

# 测试模式: 零 LLM 调用，零成本
handler = TestHandler()
handler.mock_llm("translator", "Hello World")
handler.mock_llm("polisher", "Hello, World!")

set_current_handler(handler)
result = pipeline("你好世界")
print(result)  # "Hello, World!" — 来自 mock，无 API 调用
clear_handler()

# 生产模式: 真实 API 调用（默认）
result = pipeline("你好世界")  # 调用真实 LLM
```

### 成本预测

```python
from lambdagent.cost_grade import estimate_cost, format_cost_estimate

agent = from_config("agent-config.yml")
grade = estimate_cost(agent)

print(f"Token 上界: {grade.tokens}")
print(f"延迟上界: {grade.latency:.1f}s")
print(f"成本上界: ${grade.money:.4f}")
print(f"成功概率: {grade.probability:.1%}")

# 格式化输出
print(format_cost_estimate(grade))
```

---

## 五、接入现有框架（不换框架）

### LangChain 用户

```python
# 原有代码不变
from langchain.agents import AgentExecutor, create_react_agent

executor = AgentExecutor(agent=agent, tools=tools, max_iterations=20)

# 加 2 行
from lambdagent_guard import guard_langchain
guarded = guard_langchain(executor, cost_budget=5.0, loop_detection=True)

# 用法不变
result = guarded.invoke({"input": "分析这个仓库"})
# 超 $5 自动暂停 | 死循环自动检测 | 类型错误编译时报警
```

### CrewAI 用户

```python
from crewai import Agent, Task, Crew
from lambdagent_guard import guard_crewai

crew = Crew(agents=[researcher, writer], tasks=[...])
guarded = guard_crewai(crew, cost_budget=10.0, parallel_safety=True)
result = guarded.kickoff()
```

### AutoGen 用户

```python
from autogen import GroupChat, GroupChatManager
from lambdagent_guard import guard_autogen

manager = GroupChatManager(groupchat=chat)
guarded = guard_autogen(manager, cost_budget=5.0,
                        empty_message_detection=True,  # 防 #108 空消息循环
                        terminate_robustness=True)      # 不依赖精确字符串匹配
proxy.initiate_chat(guarded, message="写一个排序算法")
```

### 只用静态分析（不用守卫）

```bash
# CLI: 检查任何框架的 YAML 配置
lambdagent lint crewai_agents.yml
lambdagent lint autogen_config.json

# Python: 提取配置 → 分析
from lambdagent.extractors import extract_config
from lambdagent.fromconfig.lint import lint_config

config = extract_config(my_langchain_executor)
results = lint_config(config)
for r in results:
    print(f"[{r.level}] {r.rule}: {r.message}")
```

---

## 六、AI IDE 集成 (MCP Server)

### Claude Code

```json
// .claude/settings.json
{
  "mcpServers": {
    "lambdagent": {
      "command": "python3",
      "args": ["-m", "lambdagent.mcp_server"]
    }
  }
}
```

### Cursor

```json
// .cursor/mcp.json
{
  "mcpServers": {
    "lambdagent": {
      "command": "python3",
      "args": ["-m", "lambdagent.mcp_server"]
    }
  }
}
```

配置后，在对话中直接说：
- "检查 agent-config.yml 有没有问题" → 自动调用 `lint_agent_config`
- "这个 agent 跑一次要花多少钱" → 自动调用 `estimate_agent_cost`
- "并行的 agent 安全吗" → 自动调用 `check_parallel_safety`

---

## 七、CI/CD 集成 (GitHub Action)

```yaml
# .github/workflows/agent-lint.yml
name: Agent Config Lint
on:
  pull_request:
    paths: ['**/*.yml', '**/*.yaml']

jobs:
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: lambdagent/agent-lint-action@v1
        with:
          paths: agents/
          fail-on: error
          cost-threshold: 5.00
          type-check: true
```

PR 上会显示：
```
❌ agents/scanner.yml:
   [ERROR] L004a: No terminate tool — ReAct loop may never exit
   [WARN]  COST: Estimated $12.40/run exceeds threshold $5.00

✅ agents/helper.yml: clean
```

---

## 八、PaaS 平台部署

### 启动服务

```bash
# 开发模式
cd agentpaas && python -m agentpaas serve --port 8000

# Docker
docker-compose up -d

# 环境变量
export AGENTPAAS_MASTER_KEY="your-secret-key"
export ANTHROPIC_API_KEY="sk-..."
export DASHSCOPE_API_KEY="sk-..."
```

### REST API

```bash
# 创建 Agent
curl -X POST http://localhost:8000/api/v1/agents \
  -H "Authorization: Bearer ap_your_key" \
  -d '{"name": "my-agent", "config": {...}}'

# 执行 Agent
curl -X POST http://localhost:8000/api/v1/agents/ag_xxx/run \
  -H "Authorization: Bearer ap_your_key" \
  -d '{"input": "Hello"}'

# 静态分析
curl -X POST http://localhost:8000/api/v1/analyze/full \
  -d '{"config": "type: react\nreact:\n  maxSteps: 200"}'

# 流式执行
curl -N http://localhost:8000/api/v1/agents/ag_xxx/run/stream \
  -H "Authorization: Bearer ap_your_key" \
  -d '{"input": "写代码"}'
```

---

## 九、多智能体编排

### GroupChat（群组讨论）

```yaml
agentId: research-team
type: react
subAgents:
  researcher:
    type: react
    systemPrompt: "你是研究员，负责搜索文献"
    tools: [WebSearch]
  critic:
    type: react
    systemPrompt: "你是评审员，质疑弱论点"
  writer:
    type: react
    systemPrompt: "你是作者，综合观点"
```

```python
from lambdagent.multiagent import GroupChat

chat = GroupChat(
    agents=[researcher, critic, writer],
    max_rounds=10,
    scheduler="round_robin",
)
result = chat("量子计算值得投资吗？")
```

### Channel（异步通信）

```python
from lambdagent.multiagent import Channel, Send, Receive

ch = Channel("results", capacity=5)
producer = Send(researcher, ch)    # 研究结果 → channel
consumer = Receive(ch, handler=writer)  # channel → 写作

producer("AI 安全最新进展")
result = consumer("等待研究结果")
```

### Handoff（动态委托）

```python
from lambdagent.multiagent import Handoff

handoff = Handoff(
    selector=Lam("router", "判断任务类型: code/doc/data"),
    registry={"code": coder, "doc": writer, "data": analyst},
    fallback=general_agent,
)
result = handoff("帮我写一个 REST API")  # → 自动委托给 coder
```

---

## 十、常见问题

### Agent 一直循环不停止

```yaml
# 确保有 terminate 工具
mcp:
  localTools:
    - terminate    # ← 这是 ReAct 的 base case (λx.x)
    - 其他工具...
```
或使用 CEK 引擎自动熔断：
```yaml
runtime:
  engine: cek
  costBudget: 5.00    # 超 $5 自动停止
```

### 管道中间崩溃，前面的钱白花了

```bash
# 先做类型检查
lambdagent lint pipeline.yml
# 如果有 T-COMPOSE 错误，修复后再运行
```

### 不知道跑一次要花多少钱

```python
from lambdagent.cost_grade import estimate_cost
grade = estimate_cost(agent)
print(f"最多 ${grade.money:.2f}")
```

### 并行 Agent 结果不对

```yaml
# 使用 CEK 引擎 + 并行安全检查
runtime:
  engine: cek
```
或用 CLI 检查：
```bash
lambdagent lint parallel-agents.yml
# 会报告 store independence 冲突
```

### 测试时不想调真实 API

```python
from lambdagent.handlers import TestHandler, set_current_handler

handler = TestHandler()
handler.mock_llm_default("测试响应")
set_current_handler(handler)
# 所有 Lam 调用返回 "测试响应"，零 API 调用，零成本
```