JamJet

コアコンセプト

JamJetにおけるエージェント、ノード、状態、耐久性について理解します。

コアコンセプト

JamJetはAIエージェント向けの耐久性ランタイムです。核心となる考え方はシンプルです。ワークフローは一時的な関数呼び出しではなく、復旧可能な状態遷移として実行されます。これにより、リプレイ、クラッシュリカバリー、承認待機、正確なトレースが可能になります。このページでは主要なプリミティブを解説します。

オーサリング手法の選択

JamJetには3つのエージェント構築方法があります。3つとも同じIRにコンパイルされ、同じRustランタイムで実行されます。違いは、どの場面で各手法を選ぶかです。

判断フローチャート

エージェントは推論戦略を用いてLLMを呼び出しますか?
(react、plan-and-execute、critic)

├─ はい → Agent + Workflow SDKを使用
│        (from jamjet import Agent, Workflow, tool)

└─ いいえ → フローは主にツールのオーケストレーション、ルーティング、
            またはコードデプロイなしで変更したい設定ですか?

            ├─ はい → YAMLワークフローを使用

            └─ いいえ → Pythonデコレータ APIを使用
                       (@workflow、@node)

3つの手法

手法最適な用途状態
Agent + Workflow SDK推論を伴うマルチエージェントシステムPydantic BaseModel(型付き、検証済み)claims-processingwealth-management
YAMLワークフローツールパイプライン、設定駆動フローYAMLのスキーマ(型付き)rag-assistant
Pythonデコレータ APIカスタムロジック、Pythonライブラリアクセス辞書ベースのStatePython SDKドキュメント

Agent + Workflow SDK

エージェントが思考する必要がある場合に使用します。ツールを選択し、結果を推論し、出力を洗練させます。各エージェントはタスクに適した推論戦略を持ちます。

from jamjet import Agent, Workflow, tool
from pydantic import BaseModel

@tool
def search_web(query: str) -> dict:
    """ウェブで情報を検索する。"""
    ...

researcher = Agent(
    name="researcher",
    model="claude-sonnet-4-6",
    tools=[search_web],
    instructions="指定されたトピックについて正確な情報を見つける。",
    strategy="react",        # 密なツール使用ループ
    max_iterations=5,
)

writer = Agent(
    name="writer",
    model="claude-sonnet-4-6",
    instructions="リサーチから明確な要約を書く。",
    strategy="critic",       # 下書き → 評価 → 改良
    max_iterations=3,
)

workflow = Workflow("research-writer", version="0.1.0")

@workflow.state
class ResearchState(BaseModel):
    topic: str
    research: str | None = None
    summary: str | None = None

@workflow.step
async def research(state: ResearchState) -> ResearchState:
    result = await researcher.run(f"Research: {state.topic}")
    return state.model_copy(update={"research": result.output})

@workflow.step
async def write(state: ResearchState) -> ResearchState:
    result = await writer.run(f"Summarize:\n{state.research}")
    return state.model_copy(update={"summary": result.output})

この手法を選ぶ場合: エージェントに戦略(react、plan-and-execute、critic、consensus、debate)が必要な場合、型付きPydantic状態が必要な場合、ヒューマン・イン・ザ・ループ(human_approval=True)が必要な場合、または複数の専門エージェントが協働する場合。

YAMLワークフロー

フローが決定論的なオーケストレーションである場合に使用します — このツールを呼び出し、次にそのモデルを呼び出し、条件に基づいてルーティングします。エージェントの推論は不要です。

id: summarize-url
version: "0.1.0"

nodes:
  fetch:
    type: tool
    server: web-fetcher
    tool: fetch_page
    arguments:
      url: "{{ state.url }}"
    output_key: page_content
    next: summarize

  summarize:
    type: model
    model: claude-haiku-4-5-20251001
    prompt: "Summarize: {{ state.page_content }}"
    output_key: summary

選択すべき場合: フローがツールとモデルの固定パイプラインである場合、開発者以外がフローを編集する必要がある場合、またはコードを再デプロイせずにルーティングを変更する必要がある場合。

Python デコレーターAPI

ノード内でカスタムPythonロジックが必要な場合に使用します — データ変換、ライブラリ呼び出し、YAMLでは複雑すぎる条件分岐など。

from jamjet import workflow, node, State

@workflow(id="data-processor", version="0.1.0")
class DataProcessor:
    @node(start=True)
    async def process(self, state: State) -> State:
        import pandas as pd
        df = pd.read_csv(state["file_path"])
        summary = df.describe().to_dict()
        return {"analysis": summary}

    @node
    async def report(self, state: State) -> State:
        response = await self.model(
            model="claude-sonnet-4-6",
            prompt=f"Write a report from this analysis:\n{state['analysis']}",
        )
        return {"report": response.text}

選択すべき場合: ノード内でPythonライブラリが必要な場合、ロジックがYAMLには複雑すぎるがエージェントの推論は不要な場合、またはクラスベースのワークフローを好む場合。

混在させることはできますか?

はい。3つすべて同じIRにコンパイルされます。YAMLワークフローはエージェントをツールとして呼び出すことができます。Pythonワークフローのステップは、YAMLで定義されたサブワークフローを呼び出すことができます。ランタイムは、どのオーサリング手段がIRを生成したかを気にしません。


ワークフロー

ワークフローはノードの有向グラフです。以下を持ちます:

  • 一意のidversion
  • state_schema — グラフを流れるデータの型付けされた構造
  • 実行が開始されるstartノード
  • 1つ以上のendノード
workflow:
  id: my-agent
  version: 0.1.0
  state_schema:
    query: str
    answer: str
    confidence: float
  start: think

ワークフローは実行前に**IR(中間表現)**グラフにコンパイルされます。IRはRustスケジューラが実際に実行するものです — YAMLとPythonは単にオーサリング手段にすぎません。

ノード

ノードはワークフロー内の計算単位です。各ノードには、その動作を決定するtypeがあります:

ノードタイプ動作
modelLLM(Claude、GPT-4、Geminiなど)を呼び出す
toolMCP経由で外部ツールを呼び出す
httpHTTPリクエストを実行する
branch条件に基づいて実行をルーティングする
parallel複数のブランチに同時に分岐する
wait外部イベントまで一時停止する
eval出力品質を評価する(ルーブリック、アサーション、レイテンシ)
endワークフローを終了する

すべてのノードはステートから読み取り、書き込みを行います。

ステート

ステートはワークフロー実行のための共有データストアです。ノード間および再起動後も永続化されます。

state_schema:
  query: str        # ユーザーからの入力
  search_results: list[str]  # 中間データ
  answer: str       # 最終出力

ステートは型付きです — スキーマはコンパイル時に検証されます。実行時、各ノードは任意のステートキーを読み取り、そのoutput_keyに書き込むことができます。

tip: ステートはメモリではなくデータベースに保存されます。実行中にランタイムがクラッシュしても、ステートは完全に復元され、最後のチェックポイントから実行が再開されます。

実行

実行とは、特定の入力を使用したワークフローの単一実行です。各実行には一意のID(例:exec_01JM4X8NKWP2)が付与されます。

実行の特徴:

  • 永続的 — データベースに保存され、再起動後も存続
  • 可観測性 — すべてのステート遷移がイベントとして記録される
  • 検査可能jamjet inspectで完全なステート、イベントタイムライン、トークン使用量を表示

永続性

永続性はJamJetの中核的な保証です。ランタイムがクラッシュしても、実行は必ず完了します

これはイベントソーシングによって実現されています。

  1. 各ノードの実行前に、node_startedイベントがデータベースに書き込まれます
  2. 各ノードの完了後に、node_completedイベントが状態パッチと共に書き込まれます
  3. 再起動時、スケジューラはイベントログを再生し、実行が停止した正確な位置を再構築します
  4. 未完了の最初のノードから実行が再開されます

作業が失われることはありません。ノードが二重に実行されることもありません。

注記: これは「最低1回」配信とは異なります。JamJetのスケジューラは分散ロックを使用し、複数のワーカープロセスがある場合でも各ノードが厳密に1回だけ実行されることを保証します。

エージェント

エージェントとは、以下が可能なワークフローです。

  • 他のエージェントから検出され、呼び出される(エージェントカード経由)
  • 他のエージェントにタスクを委任する(A2Aプロトコル経由)
  • 複数のユーザーインタラクションにわたって長期実行状態を維持する

すべてのエージェントにはエージェントカードがあります。これは、その機能、エンドポイント、入出力スキーマの機械可読な記述です。これがA2Aプロトコルの基盤となります。

スケジューラ

JamJetスケジューラはRustで書かれており、jamjet dev(ローカル)またはホストされたランタイム(本番環境)の一部として実行されます。

その動作:

  1. 保留中の作業について実行キューをポーリングします
  2. 重複実行を防ぐため、実行のロックを取得します
  3. ワーカースレッドにノードをディスパッチします
  4. 各ノードの後にチェックポイントを書き込みます

スケジューラこそが、JamJetワークフローがデフォルトで永続的である理由です。実行を決して忘れることはありません。

ローカル vs. 本番環境

機能jamjet dev(ローカル)ホスト型 / セルフホスト型
ストレージSQLitePostgreSQL
ワーカーシングルプロセス分散
MCPサーバーローカルstdioリモートSSE/HTTP
認証なしmTLS / APIキー

プログラミングモデルは同一です。同じYAMLまたはPythonコードが両方の環境で変更なしに動作します。


次のステップ

On this page