lambdagent/docs/usage.md

lambdagent 使用说明

环境准备

# 激活 conda 环境
eval "$(/root/anaconda3/bin/conda shell.bash hook)"
conda activate theory67

# 安装依赖
pip install pyyaml anthropic
# 可选: pip install openai redis

# 进入项目目录
cd /home/67/LDS/LLM-Dataset-System

# 设置 API Key (使用前必须)
export ANTHROPIC_API_KEY="sk-..."
# 或 OpenAI:
# export OPENAI_API_KEY="sk-..."

1. 编译 YAML 配置 -> Lambda 项 (不执行)

from lambdagent.fromconfig import from_config, describe_config, to_lambda_expr

# 编译为 Lambda 项
agent = from_config("agent-cofig.yml")
print(agent)  # 查看 Term 结构

# 查看 Lambda 结构描述
print(describe_config("agent-cofig.yml"))

# 导出纯 Lambda 表达式
print(to_lambda_expr("agent-cofig.yml"))

2. 静态分析 (Lint)

from lambdagent.fromconfig import lint_config, format_lint

results = lint_config("agent-cofig.yml")
print(format_lint(results, "agent-cofig.yml"))

# 输出示例:
# lambdagent lint: agent-cofig.yml
# ============================================================
#   [x] [ERROR] [L004] type=react but no 'terminate' in localTools
#     Lambda: Y combinator has no base case -> infinite loop
#   [i] [INFO ] [L015] Memory enabled: strategy=redis
#     Lambda: Gamma' = Gamma union store(redis)
# ------------------------------------------------------------
#   1 error(s), 0 warning(s), 1 info(s)

3. 编译并执行 Agent (单次)

from lambdagent.fromconfig import from_config

agent = from_config("agent-cofig.yml")
result = agent("帮我写一个快速排序")
print(result)

4. 使用 Runtime 执行 (带 trace)

from lambdagent.agentruntime import Runtime

# 一站式: 编译 + 运行
result = Runtime.execute("agent-cofig.yml", "1+1等于几")
print(result.result)                    # 最终结果
print(result.stats)                     # 统计: 步数/耗时/tokens
print(result.trace[0].term_name)        # 第一步的 term 名

# 或分步:
from lambdagent.agentruntime import RuntimeConfig
config = RuntimeConfig.from_yaml("agent-cofig.yml")
runtime = Runtime(config)
agent = from_config("agent-cofig.yml")
result = runtime.run(agent, "帮我分析这段代码")

5. 使用 CLI

# 编译 (查看 Lambda 结构)
python -m lambdagent compile agent-cofig.yml

# 执行
python -m lambdagent run agent-cofig.yml "帮我写快速排序"

# Lint
python -m lambdagent lint agent-cofig.yml

# 交互式 REPL
python -m lambdagent repl agent-cofig.yml

# 导出 Lambda 表达式
python -m lambdagent lambda agent-cofig.yml

6. 手写 Agent (Python DSL)

from lambdagent import Lam, Compose, Tool, Loop, Route, Guard, Memory, Par

# Simple Agent
agent = Lam("summarizer", "Summarize concisely.", model="claude-sonnet-4-20250514")
result = agent("A long article about...")

# Chain (Pipeline)
pipeline = (
    Lam("extract", "Extract key facts.") >>
    Lam("analyze", "Analyze the facts.") >>
    Lam("draft", "Write a draft report.")
)
result = pipeline("Raw data...")

# Router
router = Route(
    classifier=Lam("cls", "Classify: code/math/general. Output one word."),
    routes={
        "code": Lam("coder", "You are a coding expert."),
        "math": Lam("math", "You are a math expert."),
    },
    default=Lam("general", "You are a helpful assistant."),
)
result = router("How do I sort a list in Python?")

# Parallel + Merge
par = (
    Lam("researcher", "Research this topic.") |
    Lam("critic", "Critique this topic.")
) >> Tool("merge", lambda results: f"Research: {results[0]}\nCritique: {results[1]}")

# Guard (output validation)
safe = Guard(
    Lam("writer", "Write a 200-word essay."),
    validator=lambda x: len(x.split()) >= 150,
    retry=2,
)

# Memory (stateful agent)
stateful = Memory(
    Lam("assistant", "You are a helpful assistant."),
    store={"user_name": "Alice"},
)

# ReAct Loop (Y combinator)
def react_step(state):
    # ... think, act, observe logic
    return state

agent = Loop(
    body=Tool("step", react_step),
    condition=lambda r, s: "DONE" in str(r) or s >= 9,
    max_steps=10,
)

7. 自定义 YAML 配置示例

Simple Agent

agentId: my-agent
name: MyAgent
type: simple
systemPrompt: "You are a helpful coding assistant."
model:
  provider: anthropic
  name: claude-sonnet-4-20250514
  temperature: 0.0

Chain Agent

type: chain
name: ReportPipeline
model:
  name: claude-sonnet-4-20250514
chain:
  steps:
    - name: extract
      prompt: "Extract key facts from the input."
    - name: analyze
      prompt: "Analyze these facts and identify patterns."
      guard:
        validator: "len(x) > 100"
        retry: 2
    - name: report
      prompt: "Write a structured report."

Router Agent

type: router
name: SmartRouter
model:
  name: claude-sonnet-4-20250514
router:
  classifier:
    prompt: "Classify the input as: code, math, or general. Output one word only."
  routes:
    code:
      type: simple
      systemPrompt: "You are a coding expert."
    math:
      type: simple
      systemPrompt: "You are a mathematics expert."
  default:
    type: simple
    systemPrompt: "You are a helpful assistant."

Parallel Agent

type: parallel
name: MultiPerspective
model:
  name: claude-sonnet-4-20250514
parallel:
  agents:
    - name: optimist
      systemPrompt: "Analyze from an optimistic perspective."
    - name: pessimist
      systemPrompt: "Analyze from a pessimistic perspective."
  merge: custom
  mergePrompt: "Synthesize the optimistic and pessimistic analyses into a balanced view."

8. 核心对应关系

YAML                    Python (lambdagent)              Lambda 演算
──────────────         ─────────────────────────        ──────────────
systemPrompt + model   Lam("name", prompt, model)       lambda x. LLM(x)
agent(input)           term("hello")                    (f x) -> beta-reduce
type: chain            f >> g >> h                      lambda x. h(g(f(x)))
type: react            Loop(body, cond, N)              Y_N(lambda self.lambda x...)
type: router           Route(cls, {k: agent})           CASE
type: parallel         Par(a, b) >> merge               PAIR >> merge
mcp.onlineTool         Tool("name", http_fn)            Oracle / primitive
terminate              Tool("terminate", lambda x: x)   lambda x.x (identity)
memory                 Memory(agent, store)              Gamma' = Gamma union store
guard                  Guard(agent, P, retry)            {x:T | P(x)}