工作流编写（YAML）

YAML 是 JamJet 工作流的主要编写格式。本页涵盖了所有节点类型、控制流选项和配置参数。

工作流头部

workflow:
  id: my-workflow          # 唯一标识符（slug 格式）
  version: 0.1.0           # 语义化版本
  state_schema:            # 类型化状态（所有键在开始时都是可选的）
    query: str
    results: list[str]
    answer: str
    score: float
  start: first-node        # 入口节点 id

Model 节点

调用 LLM 并将响应写入状态键。

nodes:
  think:
    type: model
    model: claude-haiku-4-5-20251001       # 或 gpt-4o、gemini-2.0-flash 等
    prompt: |
      You are a helpful assistant.
      Answer this question: {{ state.query }}
    output_key: answer          # 将模型响应写入此处
    next: end

    # 可选
    system: "You are concise and precise."
    temperature: 0.3
    max_tokens: 1024
    retry:
      max_attempts: 3
      backoff: exponential      # 或 linear、constant
      delay_ms: 500

支持的模型：

提供商	模型 ID
Anthropic	`claude-haiku-4-5-20251001`、`claude-sonnet-4-6`、`claude-opus-4-6`
OpenAI	`gpt-4o`、`gpt-4o-mini`、`o3-mini`
Google	`gemini-2.0-flash`、`gemini-2.5-pro`

Tool 节点（MCP）

从已连接的 MCP 服务器调用工具。

nodes:
  search:
    type: tool
    server: brave-search        # MCP 服务器名称（来自 jamjet.toml）
    tool: web_search
    arguments:
      query: "{{ state.query }}"
      count: 5
    output_key: search_results
    next: summarize

MCP 服务器在 jamjet.toml 中配置：

[[mcp.servers]]
name = "brave-search"
command = "npx"
args = ["-y", "@modelcontextprotocol/server-brave-search"]
env = { BRAVE_API_KEY = "${BRAVE_API_KEY}" }

HTTP 节点

发起 HTTP 请求并将响应体写入状态。

nodes:
  fetch-data:
    type: http
    method: GET
    url: "https://api.example.com/data/{{ state.item_id }}"
    headers:
      Authorization: "Bearer {{ env.API_KEY }}"
    output_key: raw_data
    next: process

对于 POST/PUT，添加 body 字段：

    method: POST
    body:
      query: "{{ state.query }}"
      format: json

分支节点

根据条件路由执行。

nodes:
  route:
    type: branch
    conditions:
      - if: "state.confidence >= 0.9"
        next: done
      - if: "state.confidence >= 0.5"
        next: refine
    default: escalate        # 如果没有条件匹配时的默认行为

条件按顺序评估——首个匹配的条件生效。表达式使用 Python 语法的简化子集,可访问 state.* 和 env.*。

并行节点

扇出到多个分支,等待所有分支完成后继续执行。

nodes:
  gather:
    type: parallel
    branches:
      - search
      - fetch-docs
      - check-cache
    join: synthesize         # 所有分支完成后要运行的节点

每个分支并发运行。并行分支的状态写入会被合并;如果两个分支写入同一个键,则以最后写入者为准。

等待节点

暂停执行,直到收到外部事件。

nodes:
  await-approval:
    type: wait
    event: human_approved    # 要等待的事件名称
    timeout_hours: 24
    on_timeout: escalate     # 超时时要运行的节点
    next: continue

恢复等待中的执行:

jamjet resume exec_01JM4X8NKWP2 --event human_approved --data '{"approved": true}'

评估节点

内联评估输出质量。适用于自修复循环。

nodes:
  check-quality:
    type: eval
    scorers:
      - type: llm_judge
        rubric: "答案是否准确、完整且字数少于200字?"
        min_score: 4          # 1-5 分制
        model: claude-haiku-4-5-20251001
      - type: assertion
        check: "len(output.answer) > 0"
      - type: latency
        max_ms: 5000
    on_fail: retry_with_feedback   # 或 escalate、halt、log_and_continue
    max_retries: 2
    next: end

on_fail 选项:

retry_with_feedback — 重新运行前一个模型节点,将评分器反馈注入提示词
escalate — 路由到升级节点(在工作流头部定义 escalate:)
halt — 以错误状态停止执行
log_and_continue — 记录失败但继续执行

End 节点

终止执行。每个工作流至少需要一个。

nodes:
  end:
    type: end

你可以有多个命名的 end 节点（例如 success、failure、escalated）——它们都会干净地终止执行。

重试策略

任何节点都可以有重试策略：

    retry:
      max_attempts: 3
      backoff: exponential    # linear | constant | exponential
      delay_ms: 200           # 初始延迟
      max_delay_ms: 10000     # 指数退避的上限
      retry_on:               # 可选：仅在这些错误时重试
        - model_overloaded
        - rate_limited
        - timeout

模板表达式

提示、URL、头部和参数支持 {{ }} 模板表达式：

表达式	值
`{{ state.key }}`	状态值
`{{ env.KEY }}`	环境变量
`{{ execution.id }}`	当前执行 ID
`{{ execution.workflow_id }}`	工作流 ID
`{{ node.id }}`	当前节点 ID

完整示例

workflow:
  id: research-agent
  version: 0.2.0
  state_schema:
    query: str
    search_results: list[str]
    draft: str
    answer: str
  start: search

nodes:
  search:
    type: tool
    server: brave-search
    tool: web_search
    arguments:
      query: "{{ state.query }}"
      count: 10
    output_key: search_results
    next: draft

  draft:
    type: model
    model: claude-sonnet-4-6
    prompt: |
      基于这些搜索结果：
      {{ state.search_results | join('\n') }}

      为以下问题撰写一个全面的答案：{{ state.query }}
    output_key: draft
    next: evaluate

  evaluate:
    type: eval
    scorers:
      - type: llm_judge
        rubric: "这个答案是否基于搜索结果事实准确、完整且结构良好？"
        min_score: 4
        model: claude-haiku-4-5-20251001
      - type: assertion
        check: "len(output.draft) >= 100"
    on_fail: retry_with_feedback
    max_retries: 2
    next: done

  done:
    type: end

工作流编写（YAML）

On this page