배경

Docker로 서비스를 구성하는데 있어서 항상 하나의 이미지나 컨테이너로 작업을 할 수 있는 것은 또 아니다. 그리고 LLM 을 활용한 서비스를 사용한다고 가정했을 때, 다양한 이유로 오류가 발생해 컨테이너가 떨어질 수 도 있다. 따라서 여러개의 컨테이너를 사용해 작업을 진행해보고, 컨테이너들 여러 개가 중단되지 않도록 하는 연습이 필요하다.

이에 따라 Langgraph 매니저 에이전트(요즘 사용하는 AI 에이전트는 아니다)와 FastAPI 기반의 API 서비스, Redis 메시지 버스를 오케스트레이션 하는 실습을 진행한다. 또 하는 김에 컨테이너 교체시 무중단 배포 연습을 하려고 한다. 기본적으로 docker-compose를 사용하는 것을 기본으로 한다.

기본적으로 디렉터리 구조는 다음과 같다.

fastapi-langgraph-demo/  
├── api/  
│   ├── Dockerfile  
│   ├── main.py  
│   └── requirements.txt  
├── agent/  
│   ├── Dockerfile  
│   ├── agent.py  
│   └── requirements.txt  
├── message-bus/          # (선택) RabbitMQ 설정용 디렉터리  
├── .env                  # 공통 환경변수  
└── docker-compose.yml  

---

api (FastAPI 부분)

main.py

from fastapi import FastAPI
import redis.asyncio as aioredis
import os

app = FastAPI()
REDIS_URL = os.getenv('REDIS_URL', 'redis://redis:6379')

@app.on_event('startup')
async def startup_event():
    app.state.redis = await aioredis.from_url(REDIS_URL)

@app.get('/health')
async def health():
    return {'status': 'ok'}

@app.post('/tasks')
async def create_task(payload: str):
    await app.state.redis.lpush('tasks_stream', payload)
    return {'status': 'queued', 'task': payload}

참고 할 지점은 해당 FastAPI에서는 REDIS를 바탕으로 메시지큐를 받아 입력 받는 문자열을 쌓는 것이다.

Dockerfile

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

requirements.txt

fastapi==0.116.1             
uvicorn==0.30.3           
redis>=4.2.0        
python-dotenv==1.0.0       

---

LangGraph 부분

agent.py

import os, asyncio
from typing_extensions import TypedDict
import redis.asyncio as aioredis

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, LLMNode
from langgraph_core.tools import tool


class TaskState(TypedDict, total=False):
    text: str
    summary: str

@tool
def echo_tool(text: str) -> str:
    """입력된 텍스트를 그대로 반환하는 함수"""
    return text


def summarize_node(state: TaskState) -> dict:
    """
    LCEL을 사용한 요약 노드 구현 예시.
    ChatPromptTemplate | ChatOpenAI 파이프 연산자 활용.
    """
    # 1) 프롬프트 템플릿 정의
    prompt = ChatPromptTemplate.from_template(
        "Summarize this text:\n\n{text}"
    )
    # 2) LLM 인스턴스 정의
    llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.0)
    # 3) LCEL 체인 구성: prompt → llm 순으로 파이프
    chain = prompt | llm
    # 4) invoke로 실행 (동기/비동기 모두 지원)
    summary = chain.invoke({"text": state["text"]})  
    return {"summary": summary}


async def main():
    redis = await aioredis.from_url(os.getenv("REDIS_URL"))
    STREAM_KEY = "tasks_stream"
    last_id = "0-0"

    builder = StateGraph(TaskState)
    builder.add_node("summarize", LLMNode(fn=summarize_node))
    builder.add_node("echo", ToolNode([echo_tool]))

    builder.add_edge(START, "summarize")
    builder.add_edge("summarize", "echo")
    graph = builder.compile()
    
    while True:
        entries = await redis.xread({STREAM_KEY: last_id}, block=5000, count=5)
        if not entries:
            continue
        _, tasks = entries[0]
        for task_id, data in tasks:
            payload = {k.decode(): v.decode() for k, v in data.items()}
            result = graph.invoke(payload)
            print(f"[{task_id}] result", result)
            last_id = task_id

if __name__ == "__main__":
    asyncio.run(main())

Langgraph의 경우 데이터를 입력 받으면, 요약하는 summarize 노드로 갔다가 echo 노드로 간다. echo 노드의 경우 결과를 입력 받은 값을 그대로 반환한다.
while 문을 이용해서 큐에 들어온 값들을 Langgrapgh로 전달한다는 것만 확인하면 된다.

Dockerfile

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY agent.py .
CMD ["python", "agent.py"]

requirements.txt

langgraph==0.5.4           
langchain-openai==0.3.28      
langchain
langgraph-prebuilt==0.5.2
redis>=4.2.0          
typing-extensions>=4.11.0
python-dotenv==1.0.0
langsmith>=0.3.45

docker-compose

.env

OPENAI_API_KEY=API_KEY
REDIS_URL=redis://redis:6379

docker-compose

version: '3.8'
services:
  redis:
    image: redis:7
    networks: [backend]

  api:
    build: ./api
    env_file: .env
    depends_on: [ redis ]
    networks: [backend]
    ports: [8000:8000]

  agent:
    build: ./agent
    env_file: .env
    depends_on: [ redis ]
    networks: [backend]

networks:
  backend:
    driver: bridge

네트워크의 경우 굳이 만들지 않아도, docker-compose를 이용하면 자동적으로 bridge를 만들어 사용하지만, 네트워크 이름을 명시하여 사용했을 때 통신격리와 네임스페이스 관리에 이점이 있을 수 있음

실행

 $ docker-compose build
 $ docker-compose up -d

이미지를 빌드하고, compose up 하면 컨테이너도 실행된다.

Zero-Downtime

Docker Compose를 기반으로 하면 무중단(Zero-Downtime) 배포를 가능하게 할 수 있다. 수정된 서비스 이미지만 순차 교체하고 나머지는 중단하지 않는 것을 의미하는 것을 말하는 것이 무중단 배포다.

명령은 다음과 같다.

**docker-compose up -d --no-deps --build <서비스명>**

옵션을 설명하면 다음과 같다.

• -d, --detach => Detached 모드로 실행해 터미널을 차단하지 않고, 백그라운드에서 컨테이너를 띄운다.
• --no-deps => 지정한 서비스의 의존 서비스, 선행 서비스는 재시작 하지 않는 것을 의미한다.
• --build => 컨테이너를 실행하기 전에 이미지를 자동으로 빌드하는 것을 의미한다. (소스코드가 변경되었으니까)

헬스 체크 추가

api:
  build: ./api
  ports:
    - "8000:8000"
  healthcheck:
    test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
    interval: 10s
    timeout: 2s
    retries: 3

무중단 배포의 핵심은 "새버전이 준비되었을 때만" 트래픽을 전환하는 것이다. 이를 위해 준비되었는지 아닌지 확인하기 위해서는 헬스 체크를 추가할 필요가 있다.

롤링 업데이트

1. 새로운 이미지 빌드

docker-compose build api

2. 서비스만 교체

docker-compose up -d --no-deps --build api

위에 작업해 둔것을 예시로 삼는다면 redis랑 agent는 중단없이 api만 업데이트 할 수 있다.