OpenAI SDKからの移行

手作業のエージェントループから、構造化された堅牢なJamJetワークフローへ。

なぜ移行するのか

OpenAI SDKの素のエージェントループは、デモやプロトタイプには最適です。本番環境では、必然的に以下を構築することになります:

指数バックオフを伴う手動リトライロジック
ツール呼び出しとモデル呼び出し間の状態スレッディング
ロギング（「ステップ7は実際に何を受け取ったのか？」）
実行中にプロセスがクラッシュした際の再起動ロジック
ツール追加に伴い肥大化するツールディスパッチテーブル

JamJetは、これらすべてをアプリケーションコードではなくインフラストラクチャとして処理します。

コンセプトマッピング

OpenAI SDK（素）	JamJet
`messages` リスト	`State`（Pydanticモデル — 型付き、検証済み）
`while True:` エージェントループ	ワークフローグラフ — 明示的、検査可能
手動 `tool_calls` ディスパッチ	MCPツールノード（`type: tool`）
`client.chat.completions.create(...)`	`type: model` ノード（または `@wf.step` でクライアント呼び出し）
手作りリトライ	`retry: max_attempts: 3, backoff: exponential`
`print()` デバッグ	`jamjet inspect <exec-id>` — 完全なイベントタイムライン
クラッシュ時のプロセス再起動	耐久性ランタイム — 最後に完了したステップから再開
なし	`jamjet eval run` — 全コミットでのCIリグレッション

サイドバイサイド比較

OpenAI SDK（素）

import json
from openai import OpenAI

client = OpenAI()

TOOLS = [{
    "type": "function",
    "function": {
        "name": "web_search",
        "description": "Search the web for current information",
        "parameters": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
    },
}]

def web_search(query: str) -> str:
    return f"[results for: {query}]"  # 実際の呼び出しに置き換え

def run_agent(question: str) -> str:
    messages = [
        {"role": "system", "content": "You are a helpful research assistant."},
        {"role": "user", "content": question},
    ]
    while True:
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        msg = resp.choices[0].message
        if msg.tool_calls:
            messages.append(msg)
            for tc in msg.tool_calls:
                args = json.loads(tc.function.arguments)
                result = web_search(args["query"])
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result,
                })
        else:
            return msg.content or ""

print(run_agent("Latest AI agent frameworks?"))

JamJet

from openai import OpenAI
from pydantic import BaseModel
from jamjet import Workflow

client = OpenAI()

class State(BaseModel):
    question: str
    search_results: str = ""
    answer: str = ""

wf = Workflow("research-agent")

@wf.state
class AgentState(State):
    pass

@wf.step
async def search(state: AgentState) -> AgentState:
    # 本番環境では: type: tool + MCP サーバー (ディスパッチテーブル不要)
    results = f"[results for: {state.question}]"
    return state.model_copy(update={"search_results": results})

@wf.step
async def synthesize(state: AgentState) -> AgentState:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a helpful research assistant."},
            {"role": "user", "content": (
                f"Question: {state.question}\n"
                f"Search results: {state.search_results}\n"
                "Provide a comprehensive answer."
            )},
        ],
    )
    return state.model_copy(update={"answer": resp.choices[0].message.content or ""})

result = wf.run_sync(AgentState(question="Latest AI agent frameworks?"))
print(result.state.answer)
print(f"Ran {result.steps_executed} steps in {result.total_duration_us / 1000:.1f}ms")

移行パス

状態をPydanticモデルに引き上げる。

# 移行前: 分散した変数
messages = [...]
search_results = None
final_answer = None

# 移行後: 明示的で検証された状態
class State(BaseModel):
    question: str
    search_results: str = ""
    answer: str = ""

ループを名前付きステップに分割する。

ループの各論理的な「フェーズ」が @wf.step になります。ツールディスパッチは type: tool ノードになります。
LLM呼び出しはそのまま維持する。

ステップ関数内でOpenAIクライアントを以前と全く同じように使用します。後でリトライ、コスト追跡、可観測性をランタイムに処理させたい場合は、YAMLの type: model ノードに切り替えることができます。
まずローカルで実行する。

wf.run_sync(State(...)) はサーバーなしで動作します — ループと全く同じ挙動です。
必要に応じて永続化する。
```
jamjet dev      # Rustランタイムを起動
jamjet run workflow.yaml --input '{"question": "..."}'
```
ワークフローがクラッシュセーフになり、可観測性を持ち、jamjet eval run でテスト可能になります。

追加コードなしで利用できる機能

JamJetを導入すれば、以下の機能が追加コードなしで利用可能:

try/exceptの煩雑さなしでリトライ:

nodes:
  search:
    type: tool
    server: brave-search
    tool: web_search
    arguments:
      query: "{{ state.question }}"
    retry:
      max_attempts: 3
      backoff: exponential
      delay_ms: 1000

完全な実行タイムライン:

jamjet inspect exec-abc123

# → step: search     200ms  completed

# → step: synthesize 1840ms completed

CI回帰テスト:

jamjet eval run evals/dataset.jsonl --workflow research-agent --fail-under 0.9

ヒント: 実用的なサンプルはjamjet-labs/jamjet-benchmarksで確認できます。