lambdagentpaas 使用指南

本文档面向三类用户：

lambdagent 用户 — 用 YAML 定义 Agent，用 CLI/API 运行

LangChain/CrewAI/AutoGen 用户 — 不换框架，加一层安全检查

PaaS 运维 — 部署和管理多租户 Agent 平台

一、5 分钟快速开始

安装

pip install lambdagent

写一个 Agent

# 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

运行

# 方式 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)`	并行执行	多角度分析

模型配置

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	无需

运行时引擎

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 — 静态检查

# 检查单个配置
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 项

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

run — 编译 + 执行

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

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

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

四、Python API

基础用法

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

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", "回答问题"),
    },
)

引擎选择

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}")

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

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

成本预测

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 用户

# 原有代码不变
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 用户

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 用户

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="写一个排序算法")

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

# 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

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

Cursor

// .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)

# .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 平台部署

启动服务

# 开发模式
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

# 创建 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（群组讨论）

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

from lambdagent.multiagent import GroupChat

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

Channel（异步通信）

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（动态委托）

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 一直循环不停止

# 确保有 terminate 工具
mcp:
  localTools:
    - terminate    # ← 这是 ReAct 的 base case (λx.x)
    - 其他工具...

或使用 CEK 引擎自动熔断：

runtime:
  engine: cek
  costBudget: 5.00    # 超 $5 自动停止

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

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

不知道跑一次要花多少钱

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

并行 Agent 结果不对

# 使用 CEK 引擎 + 并行安全检查
runtime:
  engine: cek

或用 CLI 检查：

lambdagent lint parallel-agents.yml
# 会报告 store independence 冲突

测试时不想调真实 API

from lambdagent.handlers import TestHandler, set_current_handler

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

usage-guide.md 14 KB Lien permanent Historique Raw

lambdagentpaas 使用指南

一、5 分钟快速开始

安装

写一个 Agent

运行

二、YAML 配置详解

Agent 类型

模型配置

运行时引擎

三、CLI 命令

lint — 静态检查

compile — 编译为 Lambda 项

run — 编译 + 执行

四、Python API

基础用法

编程式构建 Agent

引擎选择

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

成本预测

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

LangChain 用户

CrewAI 用户

AutoGen 用户

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

六、AI IDE 集成 (MCP Server)

Claude Code

Cursor

七、CI/CD 集成 (GitHub Action)

八、PaaS 平台部署

启动服务

REST API

九、多智能体编排

GroupChat（群组讨论）

Channel（异步通信）

Handoff（动态委托）

十、常见问题

Agent 一直循环不停止

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

不知道跑一次要花多少钱

并行 Agent 结果不对

测试时不想调真实 API

usage-guide.md 14 KB

Lien permanent Historique Raw