OpenRouter 완전 가이드 3편 — 스트리밍, 툴 콜링, 멀티모달, 구조화 출력, LangChain·LangGraph 통합

2편에서 라우팅 고급 설정을 다뤘습니다. 3편은 실전 기능입니다. 스트리밍으로 실시간 타이핑 효과, 툴 콜링으로 에이전트, 이미지·PDF 멀티모달, JSON 구조화 출력, LangChain과 LangGraph 연동까지 다룹니다.

[3편 핵심 요약]
→ 스트리밍: stream=True + for chunk in stream — 모든 모델 동일한 인터페이스
→ 스트리밍 에러: HTTP 200 반환 후 SSE 내부에 에러 포함 — 별도 처리 필요
→ 툴 콜링: OpenAI 완전 동일 — tools + tool_choice 파라미터
→ 툴 콜링 모델 필터: require_parameters: True 필수 (지원 안 하는 모델 제외)
→ 멀티모달: image_url (URL or base64), PDF (base64 or URL)
→ 구조화 출력: response_format json_schema — 지원 모델 한정
→ Instructor: Pydantic 모델로 구조화 출력 — 미지원 모델도 커버
→ LangChain: pip install langchain-openrouter → ChatOpenRouter
→ LangGraph: ChatOpenRouter를 노드 모델로 직접 사용

---

실전 1 — 스트리밍

from openai import OpenAI
import os

client = OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key=os.environ["OPENROUTER_API_KEY"],
)

# ── 기본 스트리밍 ──────────────────────────────────────
def stream_response(prompt: str, model: str = "anthropic/claude-sonnet-4-6"):
    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )

    full_text = ""
    for chunk in stream:
        # 청크에서 텍스트 추출
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)
            full_text += delta

    print()  # 줄바꿈
    return full_text

stream_response("한국 AI 스타트업 생태계를 설명해줘")

# ── 스트리밍 에러 처리 ─────────────────────────────────
# OpenRouter는 HTTP 200 반환 후 SSE 내부에 에러 포함
# → try/except 만으로는 부족, chunk 내부 확인 필요

def safe_stream(prompt: str, model: str = "anthropic/claude-sonnet-4-6") -> str:
    try:
        stream = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            stream=True,
        )

        full_text = ""
        for chunk in stream:
            # 스트리밍 에러 감지 (HTTP 200 이후 내부 에러)
            if hasattr(chunk, 'error') and chunk.error:
                raise Exception(f"스트리밍 에러: {chunk.error}")

            # finish_reason 확인
            if chunk.choices and chunk.choices[0].finish_reason:
                reason = chunk.choices[0].finish_reason
                if reason not in ("stop", "length", "tool_calls"):
                    print(f"\n⚠️ 비정상 종료: {reason}")

            delta = chunk.choices[0].delta.content if chunk.choices else None
            if delta:
                print(delta, end="", flush=True)
                full_text += delta

        return full_text

    except Exception as e:
        print(f"\n❌ 에러: {e}")
        return ""

safe_stream("긴 기술 문서를 요약해줘")

// TypeScript 스트리밍
import OpenAI from "openai";

const client = new OpenAI({
  baseURL: "https://openrouter.ai/api/v1",
  apiKey: process.env.OPENROUTER_API_KEY,
});

async function streamResponse(prompt: string): Promise {
  const stream = await client.chat.completions.create({
    model: "anthropic/claude-sonnet-4-6",
    messages: [{ role: "user", content: prompt }],
    stream: true,
  });

  let fullText = "";
  for await (const chunk of stream) {
    const delta = chunk.choices[0]?.delta?.content ?? "";
    if (delta) {
      process.stdout.write(delta);
      fullText += delta;
    }
  }
  console.log();
  return fullText;
}

// Next.js API Route에서 스트리밍
// app/api/chat/route.ts
export async function POST(req: Request) {
  const { message } = await req.json();

  const openai = new OpenAI({
    baseURL: "https://openrouter.ai/api/v1",
    apiKey: process.env.OPENROUTER_API_KEY,
  });

  const stream = await openai.chat.completions.create({
    model: "anthropic/claude-sonnet-4-6",
    messages: [{ role: "user", content: message }],
    stream: true,
  });

  // ReadableStream으로 변환해서 클라이언트에 전달
  const readableStream = new ReadableStream({
    async start(controller) {
      const encoder = new TextEncoder();
      for await (const chunk of stream) {
        const text = chunk.choices[0]?.delta?.content ?? "";
        if (text) {
          controller.enqueue(encoder.encode(`data: ${JSON.stringify({ text })}\n\n`));
        }
      }
      controller.enqueue(encoder.encode("data: [DONE]\n\n"));
      controller.close();
    },
  });

  return new Response(readableStream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
    },
  });
}

---

실전 2 — 툴 콜링 (Function Calling)

import json

# 툴 정의
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "특정 도시의 현재 날씨 조회",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "도시 이름 (예: 서울, 부산)"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "온도 단위"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "웹 검색 실행",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "검색어"}
                },
                "required": ["query"]
            }
        }
    }
]

def call_tool(tool_name: str, tool_args: dict) -> str:
    """실제 툴 실행 (예시)"""
    if tool_name == "get_weather":
        # 실제 날씨 API 호출
        return json.dumps({"city": tool_args["city"], "temp": 22, "condition": "맑음"})
    elif tool_name == "search_web":
        return f"검색 결과: {tool_args['query']}에 대한 정보..."
    return "알 수 없는 툴"

def agent_loop(user_message: str) -> str:
    """단순 에이전트 루프 — 툴 콜링 반복"""
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.chat.completions.create(
            model="anthropic/claude-sonnet-4-6",
            messages=messages,
            tools=tools,
            tool_choice="auto",
            extra_body={
                "provider": {
                    # 툴 콜링 지원 프로바이더만 선택
                    "require_parameters": True
                }
            }
        )

        message = response.choices[0].message
        finish_reason = response.choices[0].finish_reason

        # 툴 콜 없으면 최종 응답
        if finish_reason == "stop" or not message.tool_calls:
            return message.content

        # 모델이 요청한 툴 실행
        messages.append(message)

        for tool_call in message.tool_calls:
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)

            print(f"🔧 툴 실행: {tool_name}({tool_args})")
            result = call_tool(tool_name, tool_args)

            # 툴 결과를 messages에 추가
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

# 실행
result = agent_loop("서울 날씨 알려줘")
print(result)

# ── 특정 툴 강제 호출 ─────────────────────────────────
response = client.chat.completions.create(
    model="anthropic/claude-sonnet-4-6",
    messages=[{"role": "user", "content": "날씨 알려줘"}],
    tools=tools,
    # 특정 툴 강제 호출
    tool_choice={
        "type": "function",
        "function": {"name": "get_weather"}
    }
)

# ── 툴 콜링 + 스트리밍 조합 ───────────────────────────
stream = client.chat.completions.create(
    model="anthropic/claude-sonnet-4-6",
    messages=[{"role": "user", "content": "서울 날씨 알려줘"}],
    tools=tools,
    stream=True,
)

tool_calls_buffer = {}
for chunk in stream:
    delta = chunk.choices[0].delta

    # 텍스트 스트리밍
    if delta.content:
        print(delta.content, end="", flush=True)

    # 툴 콜 스트리밍 조각 수집
    if delta.tool_calls:
        for tc in delta.tool_calls:
            idx = tc.index
            if idx not in tool_calls_buffer:
                tool_calls_buffer[idx] = {"id": "", "name": "", "args": ""}

            if tc.id:
                tool_calls_buffer[idx]["id"] = tc.id
            if tc.function.name:
                tool_calls_buffer[idx]["name"] = tc.function.name
            if tc.function.arguments:
                tool_calls_buffer[idx]["args"] += tc.function.arguments

# 스트리밍 완료 후 툴 실행
for tc in tool_calls_buffer.values():
    args = json.loads(tc["args"])
    result = call_tool(tc["name"], args)
    print(f"\n툴 결과: {result}")

---

실전 3 — 멀티모달 (이미지·PDF)

import base64
from pathlib import Path

# ── 이미지 URL로 분석 ─────────────────────────────────
def analyze_image_url(image_url: str, question: str) -> str:
    response = client.chat.completions.create(
        model="anthropic/claude-sonnet-4-6",  # 비전 지원 모델
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {"url": image_url}
                    },
                    {
                        "type": "text",
                        "text": question
                    }
                ]
            }
        ]
    )
    return response.choices[0].message.content

# ── 이미지 Base64로 분석 (로컬 파일) ─────────────────
def analyze_local_image(image_path: str, question: str) -> str:
    image_data = Path(image_path).read_bytes()
    base64_image = base64.b64encode(image_data).decode("utf-8")

    # MIME 타입 감지
    suffix = Path(image_path).suffix.lower()
    mime_type = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".gif": "image/gif",
        ".webp": "image/webp"
    }.get(suffix, "image/jpeg")

    response = client.chat.completions.create(
        model="anthropic/claude-opus-4-7",  # 고해상도 지원
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:{mime_type};base64,{base64_image}"
                        }
                    },
                    {
                        "type": "text",
                        "text": question
                    }
                ]
            }
        ]
    )
    return response.choices[0].message.content

# 사용
result = analyze_image_url(
    "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg",
    "이 이미지에서 무엇이 보이나요?"
)
print(result)

# ── 여러 이미지 동시 분석 ─────────────────────────────
def compare_images(image_url1: str, image_url2: str) -> str:
    response = client.chat.completions.create(
        model="google/gemini-3-flash",  # 멀티모달 + 저렴
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "image_url", "image_url": {"url": image_url1}},
                    {"type": "image_url", "image_url": {"url": image_url2}},
                    {"type": "text", "text": "두 이미지의 차이점을 설명해줘"}
                ]
            }
        ]
    )
    return response.choices[0].message.content

# ── PDF 분석 ──────────────────────────────────────────
def analyze_pdf(pdf_path: str, question: str) -> str:
    pdf_data = Path(pdf_path).read_bytes()
    base64_pdf = base64.b64encode(pdf_data).decode("utf-8")

    response = client.chat.completions.create(
        model="google/gemini-3.1-pro-preview",  # PDF 분석 강점
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:application/pdf;base64,{base64_pdf}"
                        }
                    },
                    {
                        "type": "text",
                        "text": question
                    }
                ]
            }
        ]
    )
    return response.choices[0].message.content

# PDF URL로도 가능
def analyze_pdf_url(pdf_url: str, question: str) -> str:
    response = client.chat.completions.create(
        model="google/gemini-3.1-pro-preview",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {"url": pdf_url}
                    },
                    {"type": "text", "text": question}
                ]
            }
        ]
    )
    return response.choices[0].message.content

[멀티모달 지원 모델 선택 (2026년 5월)]

이미지 분석:
→ anthropic/claude-opus-4-7 (고해상도 2576px, 최고 품질)
→ anthropic/claude-sonnet-4-6 (균형)
→ google/gemini-3-flash (저렴 + 빠름)
→ openai/gpt-5.4 (멀티모달 강점)

PDF 분석:
→ google/gemini-3.1-pro-preview (PDF 특화 강점)
→ anthropic/claude-sonnet-4-6 (긴 문서)
→ google/gemini-3-flash (대량 처리, 저렴)

20MB 초과 파일:
→ URL 방식 권장 (base64 인코딩 크기 제한)
→ 또는 Cloud Storage에 업로드 후 URL 전달

---

실전 4 — 구조화 출력 (Structured Output)

# ── 방법 1: response_format (네이티브 JSON Schema) ────
from pydantic import BaseModel
import json

response = client.chat.completions.create(
    model="openai/gpt-5.4",  # JSON Schema 지원 모델
    messages=[
        {"role": "user", "content": "서울의 날씨 정보를 JSON으로 알려줘"}
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "weather_info",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "temperature": {"type": "number"},
                    "condition": {"type": "string"},
                    "humidity": {"type": "integer"},
                    "forecast": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "day": {"type": "string"},
                                "high": {"type": "number"},
                                "low": {"type": "number"}
                            },
                            "required": ["day", "high", "low"],
                            "additionalProperties": False
                        }
                    }
                },
                "required": ["city", "temperature", "condition", "humidity"],
                "additionalProperties": False
            }
        }
    },
    extra_body={
        "provider": {
            "require_parameters": True  # JSON Schema 지원 모델만 라우팅
        }
    }
)

weather = json.loads(response.choices[0].message.content)
print(f"도시: {weather['city']}, 기온: {weather['temperature']}°C")

# ── 방법 2: Instructor + Pydantic (더 강력, 더 넓은 모델 호환) ──
# pip install instructor
import instructor
from pydantic import BaseModel, Field
from typing import Literal

class SentimentAnalysis(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"]
    score: float = Field(ge=-1, le=1, description="-1~1 감정 점수")
    key_phrases: list[str] = Field(description="핵심 문구 목록")
    summary: str = Field(description="한 줄 요약")

# Instructor로 OpenRouter 연동
instructor_client = instructor.from_openai(
    client,
    mode=instructor.Mode.JSON  # JSON 모드 (모든 모델 호환)
)

result = instructor_client.chat.completions.create(
    model="anthropic/claude-sonnet-4-6",
    response_model=SentimentAnalysis,
    messages=[
        {
            "role": "user",
            "content": "배송이 정말 빠르고 제품도 마음에 들어요. 강력 추천합니다!"
        }
    ],
    extra_body={
        "provider": {"require_parameters": True}
    }
)

print(f"감정: {result.sentiment}")
print(f"점수: {result.score}")
print(f"핵심 문구: {result.key_phrases}")

# ── 방법 3: response_format json_object (단순 JSON) ────
response = client.chat.completions.create(
    model="anthropic/claude-sonnet-4-6",
    messages=[
        {
            "role": "system",
            "content": "항상 유효한 JSON만 응답하세요."
        },
        {
            "role": "user",
            "content": "사용자 정보: 이름은 홍길동, 나이 30세, 직업 개발자. JSON으로 반환해줘."
        }
    ],
    response_format={"type": "json_object"}
)

data = json.loads(response.choices[0].message.content)
print(data)

---

실전 5 — LangChain 통합

# 설치
pip install langchain-openrouter langchain-core

from langchain_openrouter import ChatOpenRouter
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate
import os

# ── 기본 ChatOpenRouter 초기화 ────────────────────────
model = ChatOpenRouter(
    model="anthropic/claude-sonnet-4-6",
    openrouter_api_key=os.environ["OPENROUTER_API_KEY"],
    temperature=0.7,
    # OpenRouter 라우팅 옵션
    model_kwargs={
        "provider": {
            "sort": "price",
            "require_parameters": True
        }
    }
)

# ── 기본 호출 ─────────────────────────────────────────
response = model.invoke([
    SystemMessage(content="당신은 친절한 AI 어시스턴트입니다."),
    HumanMessage(content="파이썬이란 무엇인가요?")
])
print(response.content)

# ── 스트리밍 ─────────────────────────────────────────
for chunk in model.stream([HumanMessage(content="파이썬 장점 5가지")]):
    print(chunk.content, end="", flush=True)

# ── 프롬프트 템플릿 + 체인 ───────────────────────────
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 {language} 전문가입니다."),
    ("human", "{question}")
])

chain = prompt | model

result = chain.invoke({
    "language": "Python",
    "question": "asyncio와 threading의 차이는?"
})
print(result.content)

# ── 구조화 출력 (with_structured_output) ─────────────
from pydantic import BaseModel

class CodeReview(BaseModel):
    issues: list[str]
    score: int
    summary: str

structured_model = model.with_structured_output(CodeReview)

review = structured_model.invoke(
    "이 코드를 리뷰해줘:\n\ndef add(a, b): return a+b"
)
print(f"점수: {review.score}")
print(f"이슈: {review.issues}")

# ── 폴백 체인 (LangChain 방식) ─────────────────────────
from langchain_core.runnables import RunnableWithFallbacks

primary_model = ChatOpenRouter(model="anthropic/claude-opus-4-7")
fallback_model = ChatOpenRouter(model="anthropic/claude-sonnet-4-6")

model_with_fallback = primary_model.with_fallbacks([fallback_model])
result = model_with_fallback.invoke([HumanMessage(content="안녕")])

---

실전 6 — LangGraph 통합

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openrouter import ChatOpenRouter
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from typing import TypedDict, Annotated
import operator

# ── 상태 정의 ─────────────────────────────────────────
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

# ── 툴 정의 ──────────────────────────────────────────
@tool
def get_current_time() -> str:
    """현재 시간 반환"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def calculate(expression: str) -> str:
    """수식 계산"""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"계산 오류: {e}"

tools = [get_current_time, calculate]

# ── 모델 초기화 (툴 바인딩) ───────────────────────────
model = ChatOpenRouter(
    model="anthropic/claude-sonnet-4-6",
    openrouter_api_key=os.environ["OPENROUTER_API_KEY"],
    model_kwargs={
        "provider": {"require_parameters": True}  # 툴 지원 모델만
    }
).bind_tools(tools)

# ── 노드 정의 ─────────────────────────────────────────
def call_model(state: AgentState) -> dict:
    """모델 호출 노드"""
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def call_tools(state: AgentState) -> dict:
    """툴 실행 노드"""
    last_message = state["messages"][-1]
    tool_messages = []

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        # 툴 실행
        tool_result = next(
            t for t in tools if t.name == tool_name
        ).invoke(tool_args)

        tool_messages.append(
            ToolMessage(
                content=str(tool_result),
                tool_call_id=tool_call["id"]
            )
        )

    return {"messages": tool_messages}

def should_continue(state: AgentState) -> str:
    """툴 콜 여부 확인 → 다음 노드 결정"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

# ── 그래프 구성 ───────────────────────────────────────
workflow = StateGraph(AgentState)

workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tools)

workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")

# 메모리 체크포인터 (세션 유지)
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# ── 실행 ─────────────────────────────────────────────
config = {"configurable": {"thread_id": "user-123"}}

result = app.invoke(
    {"messages": [HumanMessage(content="지금 몇 시야? 그리고 123 * 456은?")]},
    config=config
)
print(result["messages"][-1].content)

# ── 멀티 모델 LangGraph — 노드마다 다른 모델 ─────────
# 복잡한 계획 → 강력한 모델 / 실행 → 저렴한 모델

planner_model = ChatOpenRouter(
    model="anthropic/claude-opus-4-7",  # 계획: 최강 모델
    openrouter_api_key=os.environ["OPENROUTER_API_KEY"],
)

executor_model = ChatOpenRouter(
    model="google/gemini-3-flash",       # 실행: 저렴한 모델
    openrouter_api_key=os.environ["OPENROUTER_API_KEY"],
)

class MultiModelState(TypedDict):
    task: str
    plan: str
    result: str

def planning_node(state: MultiModelState) -> dict:
    """Opus로 계획 수립"""
    response = planner_model.invoke([
        HumanMessage(content=f"다음 태스크를 단계별로 계획해줘:\n{state['task']}")
    ])
    return {"plan": response.content}

def execution_node(state: MultiModelState) -> dict:
    """Gemini Flash로 계획 실행 (저렴)"""
    response = executor_model.invoke([
        HumanMessage(content=f"다음 계획을 실행해줘:\n{state['plan']}")
    ])
    return {"result": response.content}

workflow = StateGraph(MultiModelState)
workflow.add_node("planning", planning_node)
workflow.add_node("execution", execution_node)
workflow.set_entry_point("planning")
workflow.add_edge("planning", "execution")
workflow.add_edge("execution", END)

multi_model_app = workflow.compile()
result = multi_model_app.invoke({"task": "Python 크롤러 설계해줘", "plan": "", "result": ""})
print(result["result"])

---

마무리

✅ 3편에서 한 것
→ 스트리밍: stream=True + SSE 에러 처리
→ 툴 콜링: tools + agent 루프 + 스트리밍 조합
→ 멀티모달: 이미지 URL/Base64, PDF, 여러 이미지 동시 분석
→ 구조화 출력: response_format, Instructor+Pydantic, json_object
→ LangChain: ChatOpenRouter, 체인, with_structured_output, 폴백
→ LangGraph: 툴 콜링 에이전트, 멀티 모델 그래프

❌ 4편에서 다룰 것
→ 모니터링·사용량 추적 (Generations API)
→ 레이트 리밋 관리 + 재시도 전략
→ OAuth PKCE — 사용자가 자기 API 키로 쓰게 하기
→ 팀 운영 (API 키 환경별 분리, 예산 알림)
→ Zero Data Retention (ZDR) 설정

---

관련 글

• OpenRouter 완전 가이드 1편 — API 키 하나로 300개 모델 쓰는 법
• OpenRouter 완전 가이드 2편 — 폴백 라우팅, 프로바이더 제어, 비용 최적화