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

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}]"  # replace with real call

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:
    # In production: type: tool + MCP server (no dispatch table needed)
    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

tip: jamjet-labs/jamjet-benchmarks에서 전체 작동 예제를 확인하세요.