Python으로 공부하는 OpenAI 7편 — FastAPI에서 대화 히스토리를 이어가는 채팅 API 만들기

Python으로 공부하는 OpenAI 7편 — FastAPI에서 대화 히스토리를 이어가는 채팅 API 만들기

6편에서 스트리밍까지 붙여보면, 이제 딱 이런 생각이 듭니다.
“오, 말은 잘 흘러가는데… 새로 질문할 때마다 기억을 다 잃어버리네?”

맞아요.
채팅 서비스가 “채팅 서비스처럼” 느껴지려면, 단순히 답을 잘 만드는 것만으로는 부족합니다.
이전 대화를 이어가는 구조, 그러니까 conversation state가 있어야 해요.

OpenAI 공식 문서도 이걸 꽤 분명하게 설명합니다. 텍스트 생성 요청 자체는 기본적으로 독립적이지만, 여러 턴의 대화를 만들려면 상태를 관리해야 하고, Responses API에서는 이를 위해 previous_response_id나 conversation을 사용할 수 있다고 안내합니다. 또 Responses API는 상태 관리가 더 쉬워서 OpenAI 쪽에서는 Chat Completions보다 Responses API 사용을 권장하고 있습니다. (OpenAI Platform)

이번 글에서는 이 흐름을 FastAPI에 붙여보겠습니다.

• 대화 히스토리를 왜 따로 설계해야 하는지
• previous_response_id가 정확히 어떤 역할을 하는지
• FastAPI에서 세션별 채팅 상태를 어떻게 저장할지
• 단일 서버용 메모리 저장소와, 실무용 Redis 감각
• 스트리밍 채팅에도 상태를 이어붙이는 방법

---

1. 채팅 서비스에서 “기억”은 어떻게 생기나

처음엔 다들 이렇게 생각합니다.

“이전 질문/답변 배열만 계속 들고 있으면 되는 거 아냐?”

틀린 말은 아닙니다.
실제로 OpenAI 문서도 수동으로 과거 메시지를 다음 요청에 다시 넣는 방식을 설명합니다. 하지만 Responses API 쪽에서는 더 편한 방법으로 previous_response_id를 제공하고, 별도로 Conversations API를 쓰면 아예 지속적인 conversation object를 둘 수도 있다고 안내합니다. (OpenAI Platform)

즉, 크게 보면 대화 상태를 이어가는 방법은 3가지 감각으로 볼 수 있습니다.

1. 직접 메시지 배열을 관리하는 방식
2. previous_response_id로 응답 체인을 이어가는 방식
3. conversation 객체를 써서 더 durable하게 이어가는 방식 (OpenAI Platform)

입문자 기준에서는 2번이 제일 이해하기 좋습니다.
왜냐하면 메시지 전체를 매번 다시 붙이지 않아도 되고, “이전 응답의 뒤를 잇는다”는 개념이 명확하거든요.

---

2. previous_response_id는 진짜 중요합니다

OpenAI 공식 예제를 보면 첫 번째 응답을 만든 뒤, 두 번째 요청에서 previous_response_id=response.id를 넘겨서 대화를 이어갑니다. 그러면 모델은 앞선 응답의 맥락을 활용해 자연스럽게 다음 턴을 이어갈 수 있습니다. (OpenAI Platform)

예를 들어 이런 흐름입니다.

response = client.responses.create(
    model="gpt-4o-mini",
    input="tell me a joke",
)

second_response = client.responses.create(
    model="gpt-4o-mini",
    previous_response_id=response.id,
    input=[{"role": "user", "content": "explain why this is funny."}],
)

이게 왜 좋냐면,
우리가 굳이 첫 질문과 첫 답변 전체를 다시 조립하지 않아도 “직전 대화의 연속선” 을 만들 수 있기 때문입니다. (OpenAI Platform)

다만 여기서 하나 진짜 중요한 포인트가 있습니다.
OpenAI 문서에 따르면 instructions는 현재 응답 생성 요청에만 적용됩니다. 즉, previous_response_id로 상태를 잇더라도 이전 턴의 instructions는 자동으로 계속 살아있지 않습니다. 이 부분을 놓치면, 첫 요청에 “친절한 Python 멘토처럼 답해”라고 넣어놓고 다음 턴부터 톤이 흔들리는 이유를 이해 못 하게 됩니다. (OpenAI Platform)

이거, 실제로 한 번 겪으면 좀 허무합니다.
“분명 첫 턴에 잘 세팅했는데 왜 갑자기 말투가 달라지지?”
이유가 바로 여기예요.

---

3. 오늘 만들 구조

이번 편은 6편 스트리밍 구조를 조금 확장해서,
세션별로 previous_response_id를 저장하는 방식으로 가보겠습니다.

app/
├── main.py
├── core/
│   └── config.py
├── schemas/
│   └── chat.py
├── services/
│   ├── openai_chat_service.py
│   └── chat_session_store.py
└── api/
    └── chat_router.py

역할은 단순합니다.

• chat_session_store.py
  세션별 대화 상태 저장
• openai_chat_service.py
  OpenAI 호출 + previous_response_id 연결
• chat_router.py
  요청을 받아 세션 기반 채팅 처리

이번 글에서는 단일 서버 학습용으로 메모리 저장소를 먼저 쓰겠습니다.
실무에서는 여러 인스턴스, 재시작, 장애 복구를 고려하면 Redis 같은 외부 저장소로 가는 게 자연스럽습니다. 이건 OpenAI 문서가 직접 Redis를 강제하는 건 아니지만, 상태를 앱 메모리에만 둘 경우 서버 재시작 시 사라진다는 백엔드 기본 원칙 때문에 생기는 구조적 차이입니다. 여기서는 그 사실을 전제로 설계를 나눠보겠습니다. (OpenAI Platform)

---

4. 요청 스키마부터 바꿉니다

이제는 메시지만 보내는 게 아니라,
어느 채팅 세션인지도 함께 보내야 합니다.

app/schemas/chat.py

from pydantic import BaseModel, Field


class ChatTurnRequest(BaseModel):
    session_id: str = Field(min_length=1, max_length=100)
    message: str = Field(min_length=1, max_length=2000)

여기서 session_id는
“이 사용자의 어떤 대화방인가” 정도로 보면 됩니다.

실무에서는 보통 이런 값이 됩니다.

• 로그인 사용자 ID + 채팅방 ID
• UUID 기반 대화 세션 ID
• 프로젝트 ID + 유저 ID 조합

---

5. 세션 저장소를 먼저 만듭니다

이번 글의 핵심은 사실 OpenAI보다 이쪽에 더 가깝습니다.
상태를 어디에 두느냐가 대화 이어가기의 핵심이거든요.

app/services/chat_session_store.py

from dataclasses import dataclass
from threading import Lock
from typing import Optional


@dataclass
class ChatSessionState:
    previous_response_id: Optional[str] = None


class InMemoryChatSessionStore:
    def __init__(self) -> None:
        self._store: dict[str, ChatSessionState] = {}
        self._lock = Lock()

    def get(self, session_id: str) -> ChatSessionState:
        with self._lock:
            if session_id not in self._store:
                self._store[session_id] = ChatSessionState()
            return self._store[session_id]

    def update_previous_response_id(
        self,
        session_id: str,
        previous_response_id: str,
    ) -> None:
        with self._lock:
            state = self._store.get(session_id)
            if state is None:
                state = ChatSessionState()
                self._store[session_id] = state
            state.previous_response_id = previous_response_id

    def clear(self, session_id: str) -> None:
        with self._lock:
            self._store.pop(session_id, None)

이 저장소는 아주 단순합니다.

• session_id로 상태를 찾고
• 거기에 마지막 previous_response_id를 저장하고
• 다음 요청 때 다시 꺼내 씁니다

이렇게만 해도 대화가 이어집니다.

물론 이건 프로세스 메모리 저장이라서,
서버를 재시작하면 다 사라집니다.
여기선 의도적으로 그렇게 만든 거예요.
처음엔 동작 원리를 이해하는 게 중요하니까요.

---

6. 이제 OpenAI 서비스에서 상태를 이어붙입니다

반응형

여기서 핵심은 두 가지입니다.

1. 현재 세션의 previous_response_id를 읽는다
2. 응답이 끝나면 새 response.id를 다시 저장한다

OpenAI 문서의 예제 흐름도 정확히 이 패턴입니다. 이전 응답의 id를 다음 요청의 previous_response_id로 넣어서 체인을 만드는 방식이죠. (OpenAI Platform)

app/services/openai_chat_service.py

from openai import AsyncOpenAI

from app.core.config import Settings
from app.services.chat_session_store import InMemoryChatSessionStore


class OpenAIChatService:
    def __init__(
        self,
        settings: Settings,
        session_store: InMemoryChatSessionStore,
    ) -> None:
        self.settings = settings
        self.session_store = session_store
        self.client = AsyncOpenAI(api_key=settings.openai_api_key)

    async def stream_chat_turn(
        self,
        session_id: str,
        user_message: str,
    ):
        state = self.session_store.get(session_id)

        request_kwargs = {
            "model": self.settings.openai_model,
            "instructions": (
                "당신은 주니어 개발자를 돕는 친절한 Python 멘토입니다. "
                "항상 한국어로 답하고, 초보자도 이해할 수 있게 설명하세요."
            ),
            "input": [{"role": "user", "content": user_message}],
            "stream": True,
        }

        if state.previous_response_id:
            request_kwargs["previous_response_id"] = state.previous_response_id

        stream = await self.client.responses.create(**request_kwargs)

        latest_response_id = None

        async for event in stream:
            event_type = getattr(event, "type", "")

            if hasattr(event, "response") and getattr(event.response, "id", None):
                latest_response_id = event.response.id

            if event_type == "response.created":
                response_obj = getattr(event, "response", None)
                if response_obj and getattr(response_obj, "id", None):
                    latest_response_id = response_obj.id

            elif event_type == "response.output_text.delta":
                delta = getattr(event, "delta", "")
                if delta:
                    yield delta

            elif event_type == "response.completed":
                response_obj = getattr(event, "response", None)
                if response_obj and getattr(response_obj, "id", None):
                    latest_response_id = response_obj.id
                break

        if latest_response_id:
            self.session_store.update_previous_response_id(
                session_id=session_id,
                previous_response_id=latest_response_id,
            )

여기서 중요한 감각은 이겁니다.

사용자 세션 하나당, “마지막 응답 ID” 하나를 들고 간다.
그러면 다음 요청은 그 응답의 뒤를 이어서 진행됩니다.

---

7. 왜 instructions를 매 턴 다시 넣어야 하나

이건 꼭 별도로 강조하고 싶습니다.

OpenAI 문서에 따르면 instructions는 현재 요청에만 적용됩니다. previous_response_id로 이전 응답의 문맥은 이어질 수 있지만, 이전 요청의 instructions가 자동으로 포함되지는 않습니다. (OpenAI Platform)

즉, 이런 건 매번 다시 넣는 게 좋습니다.

• “한국어로 답해”
• “주니어 개발자 눈높이로 설명해”
• “코드 예제를 포함해”
• “지나치게 장황하지 않게 써”

개인적으로 이걸 처음 알았을 때 좀 이상했어요.
“대화는 이어지는데, 개발자 지시는 왜 안 이어지지?”
근데 생각해보면 오히려 명확합니다.
대화 상태와 애플리케이션 정책은 다른 층위라는 거죠.

그래서 실무에서는 보통 instructions를 상수나 프롬프트 템플릿으로 분리해서 매 요청마다 주입합니다.

---

8. 라우터는 이렇게 단순하게 갑니다

app/api/chat_router.py

from typing import Annotated

from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse

from app.core.config import Settings, get_settings
from app.schemas.chat import ChatTurnRequest
from app.services.chat_session_store import InMemoryChatSessionStore
from app.services.openai_chat_service import OpenAIChatService

router = APIRouter(prefix="/chat", tags=["chat"])

_session_store = InMemoryChatSessionStore()


def get_session_store() -> InMemoryChatSessionStore:
    return _session_store


def get_chat_service(
    settings: Annotated[Settings, Depends(get_settings)],
    session_store: Annotated[InMemoryChatSessionStore, Depends(get_session_store)],
) -> OpenAIChatService:
    return OpenAIChatService(settings=settings, session_store=session_store)


@router.post("/stream")
async def stream_chat(
    request: ChatTurnRequest,
    service: Annotated[OpenAIChatService, Depends(get_chat_service)],
):
    async def event_generator():
        try:
            async for chunk in service.stream_chat_turn(
                session_id=request.session_id,
                user_message=request.message,
            ):
                yield f"data: {chunk}\n\n"
            yield "event: done\ndata: [DONE]\n\n"
        except Exception as e:
            yield f"event: error\ndata: {str(e)}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        },
    )

라우터가 해야 할 일은 여전히 단순합니다.

• 요청 받기
• 서비스 호출
• SSE로 내보내기

대화 상태 저장은 라우터가 직접 하지 않고,
서비스와 저장소가 맡습니다.
이게 구조가 깔끔해지는 포인트예요.

---

9. previous_response_id 방식의 장점과 한계

이 방식은 진짜 편합니다.

장점부터 보면:

• 매번 전체 대화 배열을 다시 조립하지 않아도 됨
• 체인이 단순함
• 입문자도 이해하기 쉬움
• FastAPI 서비스 계층에 넣기 좋음 (OpenAI Platform)

그런데 한계도 있습니다.

첫째, 서버 쪽에서 최소한 세션별 마지막 응답 ID는 저장해야 합니다.
둘째, OpenAI 문서에 따르면 응답 객체는 기본적으로 저장되며, 30일 기본 보존과 로그/조회 흐름이 있습니다. 저장을 원치 않으면 store: false를 설정할 수 있습니다. 셋째, previous_response_id를 쓰더라도 이전 입력 토큰은 여전히 입력 토큰으로 과금된다고 문서가 밝히고 있습니다. 즉, 체인을 쓴다고 토큰 비용이 공짜가 되는 건 아닙니다. (OpenAI Platform)

이 부분은 생각보다 중요해요.
많이들 “ID로 잇기만 하면 토큰이 안 드는 거 아닌가?” 같은 감각으로 오해하는데, 문서상 그렇지 않습니다. (OpenAI Platform)

---

10. 그럼 conversation 객체는 언제 쓰나

OpenAI 문서는 Conversations API도 함께 안내합니다. conversation object를 먼저 만들고, 이후 responses 요청에 그 conversation을 넘기면 세션·디바이스·작업 간에도 더 durable하게 상태를 유지할 수 있다고 설명합니다. (OpenAI Platform)

이 말은 곧 이런 뜻입니다.

previous_response_id가 잘 맞는 경우

• 단순한 1:1 채팅
• 직전 응답 기준으로 이어가는 흐름
• 입문/프로토타입
• 단일 세션 흐름이 분명한 경우

conversation이 더 잘 맞는 경우

• 여러 턴이 길어지고
• 세션을 장기간 유지해야 하고
• 여러 기기나 백그라운드 작업에서도 이어야 하고
• 대화 자체를 durable object처럼 다뤄야 하는 경우 (OpenAI Platform)

이번 편에서는 일부러 previous_response_id부터 다뤘습니다.
이게 훨씬 직관적이거든요.
하지만 프로젝트가 커지면 conversation object 쪽을 검토할 타이밍이 옵니다.

---

11. 프론트엔드에서는 어떻게 보내면 되나

이제 프론트에서는 session_id만 잘 유지하면 됩니다.

예를 들면 브라우저 탭 하나당 임시 UUID를 만들 수도 있고,
로그인 사용자라면 “사용자 ID + 채팅방 ID”를 조합할 수도 있습니다.

const sessionId = "chat-room-001";

const response = await fetch("/chat/stream", {
  method: "POST",
  headers: {
    "Content-Type": "application/json"
  },
  body: JSON.stringify({
    session_id: sessionId,
    message: "방금 설명한 데코레이터 예제를 다시 보여줘"
  })
});

핵심은 딱 하나예요.

같은 대화방이면 같은 session_id를 계속 보낸다.

그러면 서버가 그 session_id에 연결된 previous_response_id를 찾아서,
대화를 자연스럽게 이어붙일 수 있습니다.

---

12. 실무에서는 메모리 저장소로 끝내면 안 됩니다

이번 예제는 학습용이라 _session_store = InMemoryChatSessionStore()로 끝냈습니다.
그런데 실무에서 이걸 그대로 쓰면 바로 문제 생깁니다.

• 서버 재시작 시 대화 상태 유실
• 서버가 여러 대면 인스턴스마다 상태가 갈라짐
• 배포/스케일아웃 시 같은 사용자가 다른 서버로 갈 수 있음

그래서 실제 운영에서는 보통 Redis 같은 외부 저장소를 씁니다.
문서가 Redis를 지정해주는 건 아니지만, previous_response_id 또는 conversation ID를 프로세스 밖 durable storage에 둬야 한다는 건 멀티 인스턴스 웹 서비스 구조상 거의 필수에 가깝습니다. 이건 OpenAI conversation state 개념을 실서비스 아키텍처에 맞춰 적용하는 자연스러운 확장입니다. (OpenAI Platform)

예를 들면 저장 포맷은 대략 이렇습니다.

{
  "session_id": "chat-room-001",
  "previous_response_id": "resp_123456",
  "updated_at": "2026-03-30T10:30:00+09:00"
}

이 정도만 저장해도 1차 채팅 상태는 충분히 이어갈 수 있습니다.

---

13. 자주 하는 실수들

실수 1. session_id 없이 그냥 메시지만 보낸다

그러면 서버는 어떤 대화의 연속인지 알 수 없습니다.

실수 2. previous_response_id는 저장하면서 instructions는 매번 안 넣는다

문서상 instructions는 현재 요청에만 적용되므로, 톤이나 규칙이 흔들릴 수 있습니다. (OpenAI Platform)

실수 3. 메모리 저장소를 운영 환경에도 그대로 쓴다

단일 개발 서버에서는 되지만, 운영에선 거의 바로 깨집니다.

실수 4. 모든 대화 기록을 직접 배열로 들고 가면서 동시에 previous_response_id도 중복 사용한다

처음엔 헷갈리기 쉬운데, 어떤 방식으로 상태를 관리할지 먼저 정해야 합니다.
둘 다 쓸 수는 있지만, 설계 의도가 분명해야 합니다. (OpenAI Platform)

실수 5. previous_response_id 체인을 쓰면 비용이 크게 줄 거라고 기대한다

문서상 이전 입력 토큰은 여전히 입력 토큰으로 과금됩니다. (OpenAI Platform)

---

14. 지금 단계에서 추천하는 현실적인 설계

주니어 개발자가 처음 채팅 서비스를 만들 때는 이 정도가 가장 균형이 좋습니다.

1. FastAPI + Responses API
2. 세션별 previous_response_id 저장
3. instructions는 매 요청마다 재주입
4. 스트리밍 응답은 SSE 유지
5. 저장소는 처음엔 메모리, 운영은 Redis
6. 대화 리셋 API도 하나 만들기

예를 들면 리셋은 이렇게 단순하게 갈 수 있습니다.

@router.delete("/sessions/{session_id}")
def reset_session(
    session_id: str,
    session_store: Annotated[InMemoryChatSessionStore, Depends(get_session_store)],
):
    session_store.clear(session_id)
    return {"ok": True, "session_id": session_id}

이 기능이 있으면 “새 대화 시작” 버튼 만들기가 쉬워집니다.

---

15. 오늘 글의 핵심 요약

이번 편에서 꼭 가져가야 할 건 이것입니다.

FastAPI에서 대화를 이어가는 가장 쉬운 출발점은, 세션별로 previous_response_id를 저장하고 다음 요청에 다시 넘기는 구조다.

정리하면 흐름은 이렇습니다.

1. 첫 응답을 만든다.
2. 응답의 id를 세션 저장소에 저장한다.
3. 다음 요청에서 그 값을 previous_response_id로 넘긴다.
4. instructions는 매 요청마다 다시 넣는다.
5. 운영에선 메모리 대신 Redis 같은 외부 저장소로 옮긴다. (OpenAI Platform)

이 감각이 잡히면 이제 채팅 서비스가 진짜 “대화”처럼 느껴지기 시작합니다.

---

다음 편 예고

다음 글에서는 여기서 더 실무적으로 들어가겠습니다.

Python + FastAPI에서 채팅 히스토리를 DB에 저장하고, 최근 대화만 압축해서 보내는 방법

이 주제로,

• 전체 히스토리 저장 vs 모델 전달용 컨텍스트 분리
• 최근 N턴만 보내는 전략
• 오래된 대화 요약(compaction) 감각
• 토큰 낭비를 줄이는 구조
• Redis/DB와 역할 나누기

까지 이어가보겠습니다.

---

출처

• OpenAI Python SDK README — Python 3.9+ 지원, Responses API가 기본 인터페이스, sync/async client 제공. (GitHub)
• OpenAI Conversation state 가이드 — 대화 상태 관리 방식, previous_response_id, conversation 객체, Responses API 권장 흐름, 응답 보존 및 비용 관련 설명. (OpenAI Platform)
• OpenAI Text generation 가이드 — instructions는 현재 요청에만 적용되며, previous_response_id 체인에서 이전 instructions는 자동 포함되지 않음. (OpenAI Platform)
• OpenAI Responses create API 레퍼런스 — previous_response_id, conversation, store 등 주요 파라미터 구조 확인. (OpenAI Platform)

 

Python, OpenAI, FastAPI, Responses API, previous_response_id, Conversation State, AI 채팅 서버, OpenAI Python SDK, AsyncOpenAI, SSE, 채팅 히스토리, 세션 관리, Redis, Python 백엔드, 주니어 개발자