Python 에이전트 실전 10: 로컬에선 되던 서비스, 동시성에 무너지다

「치밍잉위」를 위해 만든 꼬마 小E, 앞선 아홉 편에서 쌓인 코드가 전부 demo.py 안에서 아주 잘 돌아가고 있었다. 그러다 동료가 찾아와, 첨삭을 미니프로그램에 붙여서 수강생들이 쓰게 하고 싶다고 말했다.

내 첫 반응은 "간단하지, demo만 조금 고치면 돼"였다. 그런데 막상 손을 대보니, 로컬에서 돌아가는 것과 실제로 쓸 수 있는 것 사이에는 다리 하나가 아니라 온전한 다리 하나가 놓여 있음을 알게 됐다. 미니프로그램이 어떻게 내 Python을 호출하지? 두 학생이 동시에 제출하면 세션 히스토리가 뒤섞이지 않을까? 모델 응답이 느리면 사용자는 빈 화면만 바라봐야 하나?

이번 편은 이 시리즈의 마지막 편으로, 앞선 아홉 편의 역량을 하나의 FastAPI 서비스에 담아낸다. 구조화 출력, 컨텍스트 관리, 안정성 최저선 확보를 모두 세 개의 HTTP 인터페이스로 모아 실제로 온라인에 올린다.

一. 먼저 기초 다지기: 스크립트와 서비스의 차이

1. 스크립트는 "네가 실행하는 것", 서비스는 "누구나 호출하는 것"

demo.py는 네가 더블클릭해서 실행하면 출력이 터미널로 찍힌다. 제품이 필요로 하는 것은 이것이다. 어떤 클라이언트(미니프로그램, 웹페이지, 그룹 챗봇)든 HTTP 요청을 통해 구조화된 응답을 받아낸다. 이 단계의 매개체가 바로 웹 서비스 프레임워크다. 여기서는 FastAPI를 쓴다. 가볍고, 타입 친화적이며, 대화형 API 문서가 기본 제공된다.

2. 동시성과 격리는 스크립트에는 없고 서비스에서는 반드시 처리해야 하는 문제

스크립트는 한 번에 너 한 명만 서비스한다. 서비스는 동시에 수백 수천 명의 학생을 상대해야 한다.

- 세션 격리: A의 질문이 B의 컨텍스트에 나타나면 안 된다.

- 무상태: 각 요청은 독립적이며, 기억은 서버 측이 스스로 관리해야 한다.

- 최저선 확보: 어떤 예외도 이해할 수 없는 스택을 잔뜩 프런트에 반환해서는 안 된다.

이 세 가지는 스크립트에는 하나도 없으므로, "demo만 고치면 온라인에 올릴 수 있다"는 것은 착각이다.

3. 의존성 주입: 같은 비즈니스 코드, 테스트와 프로덕션의 차이는 오직 "어떤 모델을 먹이느냐"

이것이 서비스화에서 가장 값진 결정이다. 모델을 매개변수로 넘겨받게 한다.

def create_app ( llm ) -> FastAPI: ...

테스트에는 가짜 모델을, 프로덕션에는 진짜 모델을 넘기면, 비즈니스 로직은 코드 한 벌이다. 이것이 없으면 테스트를 돌릴 때마다 실제 API에 연결하고 실제 돈을 써야 한다.

二. 엔지니어링 설계: 아홉 편의 역량을 한데 모으기

10-서비스화上线/ ├── src/ │ ├── settings.py # 설정(1편 재사용) │ ├── llm.py # 모델 팩토리(1편 재사용) │ ├── schema.py # 요청/응답의 데이터 구조 │ ├── corrector.py # 첨삭 핵심 로직(구조화 출력) │ ├── limits.py # 입력 검증 및 상한 │ ├── sessions.py # 세션 저장소(격리 + LRU 퇴출) │ ├── streaming.py # SSE 스트리밍 출력 │ ├── app.py # FastAPI 애플리케이션 │ └── server.py # uvicorn 시작 진입점 ├── tests/ # 27개 케이스, 오프라인 전부 통과 └── demo.py

표 하나로 "앞선 편들의 역량이 다 어디로 갔는지"를 한눈에 본다.

앞선 편 역량 落点(귀착점) 구조화 출력(4편) correct()가 산문이 아니라 Correction을 반환 컨텍스트 관리(3편) sessions.py 세션 격리 + 슬라이딩 윈도 히스토리 안정성 최저선 확보(8편) 실패 시 항상 파싱 가능한 ok:false를 반환

三. 먼저 계약 정하기: 데이터 구조(schema.py)

서비스가 온라인에 오르면 요청 본문과 응답 본문이 곧 프런트와 백엔드 사이의 계약이다. pydantic으로 이것을 코드로 작성한다.

class Correction ( BaseModel ): error_types: list [ErrorType] = Field(default_factory= list ) needs_rewrite: bool = False corrected_sentence: str = "" comment: str = "" class CorrectResponse ( BaseModel ): ok: bool correction: Correction | None = None error: str | None = None

CorrectResponse 바깥에 ok / error를 한 겹 감쌌고, Correction을 응답 본문으로 직접 쓰지는 않았다는 점에 주목하자. 이렇게 설계한 데에는 아주 현실적인 이유가 있다. 프런트는 언제나 먼저 ok를 보고 그다음 correction을 보며, 결코 try로 파싱할 필요가 없다. 실패와 성공이 같은 구조를 타고, 다만 ok만 뒤집힌다. 이것이 서버 측 "실패 최저선 확보"의 첫 수업이다.

四. 첨삭 로직: 두 개의 통로(corrector.py)

4편의 구조화 출력을 하나의 함수로 모은다. 두 개의 통로를 유지한다.

SYSTEM = """너는 「치밍 영어」의 AI 조교 샤오 E야. 학생이 쓴 영어 문장을 첨삭하는 역할을 맡고 있어. 요구 사항: 1. 문장별로 오류를 찾아내고, 오류 하나당 한 항목으로, 빠뜨리지 말 것 2. 문장 자체에 오류가 없을 때는 "이 문장은 문제없음"이라고 명확히 말할 것 3. JSON만 출력하고, 그 밖의 어떤 문자도 출력하지 말 것 """ PROMPT = ChatPromptTemplate.from_messages( [ ( "system" , SYSTEM), ( "human" , "{format_instructions}\n\n학생이 쓴 문장:\n{essay}" ), ] ) def correct_structured ( llm, essay ) -> Correction: """경로 1: with_structured_output, function calling을 사용, 가장 안정적.""" structured = llm.with_structured_output(Correction) return structured.invoke(PROMPT. format (essay=essay)) def correct_parse ( llm, essay ) -> Correction: """경로 2: PydanticOutputParser가 직접 parse, 모델이 도구 호출을 지원하지 않는 상황에 적합.""" parser = PydanticOutputParser(pydantic_object=Correction) prompt = PROMPT.partial(format_instructions=parser.get_format_instructions()) response = llm.invoke(prompt. format (essay=essay)) return parser.parse( getattr (response, "content" , "" ))

서비스에서는 기본적으로 correct_parse를 사용한다 — 이것은 모델의 도구 호출 능력에 의존하지 않고, parse 과정이 제어 가능하고 테스트 가능하다: 모델이 "형식대로 하지 않았을" 때 우리는 이해할 수 없는 예외가 아니라 원본 텍스트를 얻을 수 있다. format_instructions는 schema의 필드 설명, 타입, 예시를 프롬프트에 자동으로 집어넣는데, 이는 2편의 "프롬프트를 객체로 만들기"가 또 한 번 실현된 것이다.

5. 입력 방어선: 서버는 프런트엔드를 신뢰하지 않는다 (limits.py)

출시 후에는 요청이 누구에게서 오는지, 내용이 얼마나 긴지를 당신이 통제할 수 없다. 그래서 서버가 스스로 제한을 설정해야 한다:

MAX_ESSAY_CHARS = 2000 def validate_essay ( essay: str ) -> str : text = (essay or "" ).strip() if not text: raise InputError( "essay는 비어 있을 수 없습니다" ) if len (text) > MAX_ESSAY_CHARS: raise InputError( f"essay가 너무 깁니다. 최대 {MAX_ESSAY_CHARS}자까지 가능합니다" ) return text

첨삭 인터페이스는 원래 "한 문장"을 받는 것인데, 지나치게 긴 것은 보통 잘못 보낸 것이거나 악의적인 요청이다. 2000자로 설정해서, 초과하면 바로 422를 반환한다. 프런트엔드가 "이미 검증했다"는 말을 믿지 마라.

6. 세션 격리: 동시성 안전 + 정리까지 (sessions.py)

3편에서 세션 격리를 다뤘는데, 여기서는 "서버 상주 프로세스"가 가져오는 두 가지 새로운 문제 — 동시성 안전과 메모리 무한 증가 — 를 보충한다:

@dataclass class Conversation : session_id: str history: deque = field(default_factory= lambda : deque(maxlen=MAX_TURNS * 2 ))

deque(maxlen=...)는 초과 시 가장 오래된 것을 자동으로 버려서, 자연스럽게 슬라이딩 윈도를 구현한다. 저장소 자체에는 락과 LRU 제거가 딸려 있다:

class SessionStore : def get ( self, session_id: str ) -> Conversation: with self._lock: if session_id not in self._sessions: self._sessions[session_id] = Conversation(session_id=session_id) if len (self._sessions) > self._ max : # 상한 초과 시 가장 오래된 것 제거 self._sessions.popitem(last= False ) else : self._sessions.move_to_end(session_id) return self._sessions[session_id]

메모리 버전 세션 저장소는 단일 프로세스 소규모 용도에는 충분하지만, 다중 프로세스 배포 시에는 반드시 Redis로 바꿔야 한다 — 이는 출시 체크리스트의 필수 항목이며, 뒤에서 설명할 것이다.

7. 스트리밍 출력: 글자가 하나씩 튀어나온다 (streaming.py)

학생이 제출한 후 빈 화면을 보며 3초를 기다리는 것과, 글자가 하나씩 튀어나오는 것을 보는 것은 체감이 하늘과 땅 차이다. 스트리밍은 멋 부리는 기술이 아니라, "이 제품이 빠른가"라는 주관적 느낌을 직접 결정한다.

SSE(Server-Sent Events)의 형식은 딱 두 줄이다: 한 줄 `data: <json>`, 그리고 빈 줄 하나, 이렇게 반복된다:

def sse_event ( payload: dict ) -> str : return f"data: {json.dumps(payload, ensure_ascii= False )} \n\n" def sse_events ( llm, messages ): for token in stream_tokens(llm, messages): yield sse_event({ "delta" : token}) yield sse_event({ "done" : True }) # 종료 표시, 절대 빠뜨릴 수 없음

마지막의 {"done": true}는 전체 코드의 핵심이다. 이것을 보내지 않으면 프런트엔드는 이번 턴이 끝났다는 것을 알 수 없어 계속 기다리거나, 타임아웃을 종료로 잘못 판단하게 된다.

여기서 흔한 의문이 하나 있다. 왜 WebSocket을 바로 쓰지 않는가? 대화는 '문답'의 단방향 스트림이기 때문에 SSE가 더 단순하다. 일반 HTTP를 타므로 전용 연결 관리가 필요 없고, 하트비트를 처리할 필요도 없으며, 자연스럽게 로드밸런서와 로그 시스템이 이해할 수 있다. '양방향 실시간, 클라이언트도 능동적으로 보낼 수 있어야 하는' 경우(예: 다인 협업 화이트보드)에만 WebSocket을 쓴다. 줄일 수 있으면 줄이는 것, 이것이 엔지니어링에서 가장 이득이 큰 절제다.

8. 조립: 세 개의 인터페이스(app.py)

create_app(llm) 팩토리 함수가 위의 모든 모듈을 세 개의 인터페이스로 조립한다.

메서드 경로 역할 GET /health 생존 확인, 모니터링/로드밸런서용 POST /correct 동기 첨삭, 구조화된 결과 반환 POST /chat/stream 스트리밍 대화, SSE로 글자 단위 반환

첨삭은 안정적이고 구조화되어야 하고, 대화는 빠르고 스트리밍이어야 한다. 그래서 하나에 억지로 밀어넣지 않고 두 개의 인터페이스로 나눴다. 스트리밍 인터페이스의 조립을 보자.

@app.post( "/chat/stream" ) def chat_stream ( req: ChatRequest ) -> StreamingResponse: message = validate_message(req.message) session_id = req.session_id or "default" conversation = store.get(session_id) conversation.append_user(message) messages = conversation.to_messages(store.system) def generate (): parts = [] for token in stream_tokens(llm, messages): parts.append(token) yield sse_event({ "delta" : token}) conversation.append_ai( "" .join(parts)) # 답변도 기록에 남긴다 yield sse_event({ "done" : True }) return StreamingResponse(generate(), media_type= "text/event-stream" )

두 가지 세부 사항이 있다. 스트리밍 과정에서 토큰을 내보내면서 동시에 누적하고, 끝난 뒤 기록에 되써 넣는다. 그래야 다음 턴에 이번에 무엇을 이야기했는지 '기억'한다. 실패 시에는 HTTPException(422)를 던져서, 프런트엔드가 500에 스택 한 무더기를 받는 대신 명확한 에러를 받도록 한다.

9. 기동: uvicorn으로 실행(server.py)

settings = load_settings() llm = build_llm(settings, temperature= 0.3 ) # 첨삭은 안정적이어야 하므로 온도를 0.3으로 낮춘다 app = create_app(llm) import uvicorn uvicorn.run(app, host=args.host, port=args.port)

python src/server.py로 서비스를 띄우고, 브라우저에서 http://127.0.0.1:8000/docs를 열면 FastAPI에 내장된 인터랙티브 API 문서가 나온다. Postman을 설치할 필요 없이 웹페이지에서 바로 인터페이스를 시험할 수 있다.

10. 실행: 세 개의 인터페이스 한 바퀴 돌기

python demo.py는 전 과정 오프라인으로 동작한다(실제 서버를 띄우지 않고 TestClient로 직접 인터페이스를 호출한다).

1. 헬스 체크 GET /health 200 {"status": "ok" , "model" : "FakeListChatModel" } 2. 동기 첨삭 POST /correct 상태 코드: 200 ok :True 에러 유형: [ '시제' ] 재작성 필요 :True 수정안 : I went to the park yesterday. 설명 : yesterday는 과거 시간이므로 go를 went로 바꿔야 한다. 경계: 빈 essay는 차단된다(500이 아니라 422 반환) 상태 코드: 422 3. 스트리밍 대화 POST /chat/stream(SSE) content -type: text/event-stream; charset=utf- 8 data: {"delta": "이" } data: {"delta": "문" } data: {"delta": "장" } data: {"delta": "은" } ... data: {"done": true}

세 번째를 보자. 스트리밍 인터페이스가 내보내는 것은 일련의 delta이고, 마지막에 done으로 마무리된다. 이것이 프런트엔드가 글자 단위로 렌더링하는 원료다.

테스트 결과, 27개 케이스가 모두 통과했고 전 과정 오프라인이다.

tests/test_corrector.py 4개 구조화 첨삭, 두 개의 경로, 오류 없는 문장 tests/test_limits.py 6개 빈/초과 길이 입력 거부, 경계 길이 tests/test_streaming.py 4개 SSE 형식, 글자 단위 스트림, done 마무리 tests/test_sessions.py 6개 격리, 슬라이딩 윈도, LRU 제거 tests/test_app.py 7개 헬스 체크, 첨삭, 스트리밍, 기록 기억 27 passed in 1.40s

이 테스트 묶음에서 내가 가장 중시하는 것은 test_stream_keeps_history_across_turns다. 이것은 '스트리밍이 끝난 뒤 답변이 정말 다음 턴 요청에 들어갔는가'를 단언한다. 컨텍스트 기억 같은 버그는 가장 음흉한데, 에러가 나지 않고 답변이 슬그머니 멍청해지기만 하기 때문이다.

11. 배포 체크리스트: '돌아감'에서 '쓸 수 있음'까지 남은 것들

코드가 돌아간다고 해서 배포할 수 있는 것은 아니다. 이 체크리스트는 내가 직접 부딪히며 만든 것으로, 각 항목이 실제 사고 하나씩에 대응한다.

- .env에 Key를 설정하고, Git에 커밋하지 말 것 : Key가 저장소에 들어가는 것은 지갑 비밀번호를 단체 채팅방에 올리는 것과 같다. .gitignore에 이미 .env를 제외해 두었으니 습관을 들여라.

- /health 를 모니터링에 연결 : 그렇지 않으면 서비스가 죽어도 아무도 모른 채, 학부모 항의가 들어와서야 발견하게 됩니다. 그때는 이미 늦었습니다.

- 속도 제한 추가 : limits.py는 길이만 제한할 뿐 빈도는 제한하지 않아 악의적인 요청이 할당량을 태워버릴 수 있으니, 한 겹 더 빈도 제한(예: IP나 토큰별 횟수 제한)을 걸어야 합니다.

- 세션 저장소를 Redis로 교체 : sessions.py는 메모리 버전이라 다중 프로세스 배포 시 각자 따로 저장하게 되어, 사용자가 요청을 바꾸면 "기억상실"에 빠지므로 반드시 외부 저장소로 바꿔야 합니다.

- 로그와 알림 연결 : print를 쓰지 마세요. 오류가 났을 때 필요한 것은 타임스탬프가 있고 검색 가능한 로그, 그리고 한밤중에 사람을 깨울 수 있는 알림입니다.

- 구조화 출력에 캐시 추가 : 같은 문장의 "첨삭" 결과는 캐시할 수 있어 반복 요청에 돈을 두 번 쓰지 않습니다. 이는 3편의 컨텍스트 캐싱과 같은 비용 절감 사고방식입니다.

이런 것들은 "운영의 일"처럼 보이지만, 출시는 결코 코드를 작성하는 것만이 아닙니다 . 제품이 온라인에서 돌아가는 이상, 당신은 그 제품의 매 호출에 책임을 져야 합니다.

十二. 함정 목록

- "데모만 조금 고치면 출시 가능" : 스크립트에 동시성, 격리, 안전망이 없으면 그대로 출시했다가 반드시 실패합니다.

- 응답 본문에 ok / error 계층화를 하지 않음 : 프론트엔드가 어쩔 수 없이 try로 파싱해야 하고, 실패와 성공이 뒤섞이게 됩니다.

- 스트리밍에서 done을 보내지 않음 : 프론트엔드는 이번 라운드가 끝났는지 영원히 알 수 없고, 타임아웃을 종료로 착각합니다.

- 서버가 프론트엔드 검증을 믿어버림 : 입력 상한을 두지 않아 악의적인 요청이 할당량을 태워버립니다.

- 세션을 메모리에 둔 채 다중 프로세스로 실행 : 각 프로세스가 따로 저장해 사용자가 요청을 바꾸면 "기억상실"에 빠집니다.

- Key를 Git에 커밋 : 한 번 유출되면 한 번 청구서에 놀랍니다.

十三. 시리즈 마무리: 열 편, 제로에서 출시까지의 한 길

돌아보며 열 편을 하나로 꿰어보면, 사실 하나의 완전한 길입니다.

- 1-2편 : 환경을 갖추고 프롬프트를 제대로 쓴다——먼저 말을 명확히 한다.

- 3편 : 컨텍스트를 관리해 진짜 돈을 아낀다——먼저 쓸데없는 돈을 태우지 않는다.

- 4편 : 출력 구조화——프로그램이 모델의 말을 "이해"하게 한다.

- 5편 : RAG 검색——오픈북 시험을 보게 해서 함부로 지어내지 않게 한다.

- 6편 : LangGraph 상태 머신——복잡한 흐름을 되돌릴 수 있는 단계로 쪼갠다.

- 7편 : 도구 호출——실제로 손을 쓸 수 있는 "손"을 연결해준다.

- 8편 : 안정성——재시도, 타임아웃, 강등이라는 세 개의 안전장치.

- 9편 : 평가——모든 변경 사항이 점수를 받을 수 있게 한다.

- 10편 : 서비스화——이 모든 것을 출시 가능한 서비스에 담아낸다.

이것이 Python으로 에이전트 제품을 만드는 전 과정의 골격입니다. 한 번에 전부 익힐 필요는 없습니다—— 1편부터 시작해 실제로 돌아가는 데모를 하나 작성하는 것이 열 편을 다 읽는 것보다 더 유용합니다. 어떤 능력이 필요해지면 해당 편을 다시 펼쳐보세요.

마지막으로, 이 열 편에서 반복적으로 등장하고 가장 남기고 싶은 세 마디: 먼저 말을 명확히 하고(프롬프트), 다음으로 쓸데없는 돈을 태우지 않고(컨텍스트), 그다음 프로그램이 그것을 이해하게 하라(구조화 출력). 이 세 가지를 탄탄히 해두면, 이후의 검색, 오케스트레이션, 도구, 안정성, 평가, 출시는 모두 물 흐르듯 따라옵니다.

시리즈는 여기서 마무리합니다. 하지만 "치밍 영어의 꼬마 E" 이야기는 아직 끝나지 않았습니다——그것은 당신 손에 있는 다음 프로젝트의 그림자일 뿐입니다.

마지막 한마디 : 돌아가는 코드는 당신의 실력이고, 출시할 수 있는 서비스라야 남이 의존할 수 있습니다——그 차이 나는 그 길을 엔지니어링이라 부릅니다.

웨이거애프로그래밍

AI 풀스택, 공식 계정: 웨이거애프로그래밍

437

글

330k

조회수

572

팔로워